10000 Modularizing Event Loop (#106) · marcilio/sdk-python@a331e63 · GitHub
[go: up one dir, main page]

Skip to content

Commit a331e63

Browse files
authored
Modularizing Event Loop (strands-agents#106)
1 parent f2d2cb6 commit a331e63

File tree

1 file changed

+134
-77
lines changed

1 file changed

+134
-77
lines changed

src/strands/event_loop/event_loop.py

Lines changed: 134 additions & 77 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,10 @@
2828

2929
logger = logging.getLogger(__name__)
3030

31+
MAX_ATTEMPTS = 6
32+
INITIAL_DELAY = 4
33+
MAX_DELAY = 240 # 4 minutes
34+
3135

3236
def initialize_state(**kwargs: Any) -> Any:
3337
"""Initialize the request state if not present.
@@ -51,7 +55,7 @@ def event_loop_cycle(
5155
system_prompt: Optional[str],
5256
messages: Messages,
5357
tool_config: Optional[ToolConfig],
54-
callback_handler: Any,
58+
callback_handler: Callable[..., Any],
5559
tool_handler: Optional[ToolHandler],
5660
tool_execution_handler: Optional[ParallelToolExecutorInterface] = None,
5761
**kwargs: Any,
@@ -130,13 +134,9 @@ def event_loop_cycle(
130134
stop_reason: StopReason
131135
usage: Any
132136
metrics: Metrics
133-
max_attempts = 6
134-
initial_delay = 4
135-
max_delay = 240 # 4 minutes
136-
current_delay = initial_delay
137137

138138
# Retry loop for handling throttling exceptions
139-
for attempt in range(max_attempts):
139+
for attempt in range(MAX_ATTEMPTS):
140140
model_id = model.config.get("model_id") if hasattr(model, "config") else None
141141
model_invoke_span = tracer.start_model_invoke_span(
142142
parent_span=cycle_span,
@@ -177,7 +177,7 @@ def event_loop_cycle(
177177

178178
# Handle throttling errors with exponential backoff
179179
should_retry, current_delay = handle_throttling_error(
180-
e, attempt, max_attempts, current_delay, max_delay, callback_handler, kwargs
180+
e, attempt, MAX_ATTEMPTS, INITIAL_DELAY, MAX_DELAY, callback_handler, kwargs
181181
)
182182
if should_retry:
183183
continue
@@ -204,80 +204,35 @@ def event_loop_cycle(
204204

205205
# If the model is requesting to use tools
206206
if stop_reason == "tool_use":
207-
tool_uses: List[ToolUse] = []
208-
tool_results: List[ToolResult] = []
209-
invalid_tool_use_ids: List[str] = []
210-
211-
# Extract and validate tools
212-
validate_and_prepare_tools(message, tool_uses, tool_results, invalid_tool_use_ids)
213-
214-
# Check if tools are available for execution
215-
if tool_uses:
216-
if tool_handler is None:
217-
raise ValueError("toolUse present but tool handler not set")
218-
if tool_config is None:
219-
raise ValueError("toolUse present but tool config not set")
220-
221-
# Create the tool handler process callable
222-
tool_handler_process: Callable[[ToolUse], ToolResult] = partial(
223-
tool_handler.process,
224-
messages=messages,
225-
model=model,
226-
system_prompt=system_prompt,
227-
tool_config=tool_config,
228-
callback_handler=callback_handler,
229-
**kwargs,
207+
if not tool_handler:
208+
raise EventLoopException(
209+
Exception("Model requested tool use but no tool handler provided"),
210+
kwargs["request_state"],
230211
)
231212

232-
# Execute tools (parallel or sequential)
233-
run_tools(
234-
handler=tool_handler_process,
235-
tool_uses=tool_uses,
236-
event_loop_metrics=event_loop_metrics,
237-
request_state=cast(Any, kwargs["request_state"]),
238-
invalid_tool_use_ids=invalid_tool_use_ids,
239-
tool_results=tool_results,
240-
cycle_trace=cycle_trace,
241-
parent_span=cycle_span,
242-
parallel_tool_executor=tool_execution_handler,
213+
if tool_config is None:
214+
raise EventLoopException(
215+
Exception("Model requested tool use but no tool config provided"),
216+
kwargs["request_state"],
243217
)
244218

245-
# Update state for the next cycle
246-
kwargs = prepare_next_cycle(kwargs, event_loop_metrics)
247-
248-
# Create the tool result message
249-
tool_result_message: Message = {
250-
"role": "user",
251-
"content": [{"toolResult": result} for result in tool_results],
252-
}
253-
messages.append(tool_result_message)
254-
callback_handler(message=tool_result_message)
255-
256-
if cycle_span:
257-
tracer.end_event_loop_cycle_span(
258-
span=cycle_span, message=message, tool_result_message=tool_result_message
259-
)
260-
261-
# Check if we should stop the event loop
262-
if kwargs["request_state"].get("stop_event_loop"):
263-
event_loop_metrics.end_cycle(cycle_start_time, cycle_trace)
264-
return (
265-
stop_reason,
266-
message,
267-
event_loop_metrics,
268-
kwargs["request_state"],
269-
)
270-
271-
# Recursive call to continue the conversation
272-
return recurse_event_loop(
273-
model=model,
274-
system_prompt=system_prompt,
275-
messages=messages,
276-
tool_config=tool_config,
277-
callback_handler=callback_handler,
278-
tool_handler=tool_handler,
279-
**kwargs,
280-
)
219+
# Handle tool execution
220+
return _handle_tool_execution(
221+
stop_reason,
222+
message,
223+
model,
224+
system_prompt,
225+
messages,
226+
tool_config,
227+
tool_handler,
228+
callback_handler,
229+
tool_execution_handler,
230+
event_loop_metrics,
231+
cycle_trace,
232+
cycle_span,
233+
cycle_start_time,
234+
kwargs,
235+
)
281236

282237
# End the cycle and return results
283238
event_loop_metrics.end_cycle(cycle_start_time, cycle_trace)
@@ -377,3 +332,105 @@ def prepare_next_cycle(kwargs: Dict[str, Any], event_loop_metrics: EventLoopMetr
377332
kwargs["event_loop_pare 10000 nt_cycle_id"] = kwargs["event_loop_cycle_id"]
378333

379334
return kwargs
335+
336+
337+
def _handle_tool_execution(
338+
stop_reason: StopReason,
339+
message: Message,
340+
model: Model,
341+
system_prompt: Optional[str],
342+
messages: Messages,
343+
tool_config: ToolConfig,
344+
tool_handler: ToolHandler,
345+
callback_handler: Callable[..., Any],
346+
tool_execution_handler: Optional[ParallelToolExecutorInterface],
347+
event_loop_metrics: EventLoopMetrics,
348+
cycle_trace: Trace,
349+
cycle_span: Any,
350+
cycle_start_time: float,
351+
kwargs: Dict[str, Any],
352+
) -> Tuple[StopReason, Message, EventLoopMetrics, Dict[str, Any]]:
353+
tool_uses: List[ToolUse] = []
354+
tool_results: List[ToolResult] = []
355+
invalid_tool_use_ids: List[str] = []
356+
357+
"""
358+
Handles the execution of tools requested by the model during an event loop cycle.
359+
360+
Args:
361+
stop_reason (StopReason): The reason the model stopped generating.
362+
message (Message): The message from the model that may contain tool use requests.
363+
model (Model): The model provider instance.
364+
system_prompt (Optional[str]): The system prompt instructions for the model.
365+
messages (Messages): The conversation history messages.
366+
tool_config (ToolConfig): Configuration for available tools.
367+
tool_handler (ToolHandler): Handler for tool execution.
368+
callback_handler (Callable[..., Any]): Callback for processing events as they happen.
369+
tool_execution_handler (Optional[ParallelToolExecutorInterface]): Optional handler for parallel tool execution.
370+
event_loop_metrics (EventLoopMetrics): Metrics tracking object for the event loop.
371+
cycle_trace (Trace): Trace object for the current event loop cycle.
372+
cycle_span (Any): Span object for tracing the cycle (type may vary).
373+
cycle_start_time (float): Start time of the current cycle.
374+
kwargs (Dict[str, Any]): Additional keyword arguments, including request state.
375+
376+
Returns:
377+
Tuple[StopReason, Message, EventLoopMetrics, Dict[str, Any]]:
378+
- The stop reason,
379+
- The updated message,
380+
- The updated event loop metrics,
381+
- The updated request state.
382+
"""
383+
validate_and_prepare_tools(message, tool_uses, tool_results, invalid_tool_use_ids)
384+
385+
if not tool_uses:
386+
return stop_reason, message, event_loop_metrics, kwargs["request_state"]
387+
388+
tool_handler_process = partial(
389+
tool_handler.process,
390+
messages=messages,
391+
model=model,
392+
system_prompt=system_prompt,
393+
tool_config=tool_config,
394+
callback_handler=callback_handler,
395+
**kwargs,
396+
)
397+
398+
run_tools(
399+
handler=tool_handler_process,
400+
tool_uses=tool_uses,
401+
event_loop_metrics=event_loop_metrics,
402+
request_state=cast(Any, kwargs["request_state"]),
403+
invalid_tool_use_ids=invalid_tool_use_ids,
404+
tool_results=tool_results,
405+
cycle_trace=cycle_trace,
406+
parent_span=cycle_span,
407+
parallel_tool_executor=tool_execution_handler,
408+
)
409+
410+
kwargs = prepare_next_cycle(kwargs, event_loop_metrics)
411+
412+
tool_result_message: Message = {
413+
"role": "user",
414+
"content": [{"toolResult": result} for result in tool_results],
415+
}
416+
417+
messages.append(tool_result_message)
418+
callback_handler(message=tool_result_message)
419+
420+
if cycle_span:
421+ tracer = get_tracer()
422+
tracer.end_event_loop_cycle_span(span=cycle_span, message=message, tool_result_message=tool_result_message)
423+
424+
if kwargs["request_state"].get("stop_event_loop", False):
425+
event_loop_metrics.end_cycle(cycle_start_time, cycle_trace)
426+
return stop_reason, message, event_loop_metrics, kwargs["request_state"]
427+
428+
return recurse_event_loop(
429+
model=model,
430+
system_prompt=system_prompt,
431+
messages=messages,
432+
tool_config=tool_config,
433+
callback_handler=callback_handler,
434+
tool_handler=tool_handler,
435+
**kwargs,
436+
)

0 commit comments

Comments
 (0)
0