52 KiB
52 KiB
我把一些底层库的代码贴在这里来帮助我查找报错原因：
/.local/lib/python3.10/site-packages/agents/result.py
from __future__ import annotations
import abc
import asyncio
from collections.abc import AsyncIterator
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any, cast
from typing_extensions import TypeVar
from ._run_impl import QueueCompleteSentinel
from .agent import Agent
from .agent_output import AgentOutputSchemaBase
from .exceptions import (
AgentsException,
InputGuardrailTripwireTriggered,
MaxTurnsExceeded,
RunErrorDetails,
)
from .guardrail import InputGuardrailResult, OutputGuardrailResult
from .items import ItemHelpers, ModelResponse, RunItem, TResponseInputItem
from .logger import logger
from .run_context import RunContextWrapper
from .stream_events import StreamEvent
from .tracing import Trace
from .util._pretty_print import (
pretty_print_result,
pretty_print_run_result_streaming,
)
if TYPE_CHECKING:
from ._run_impl import QueueCompleteSentinel
from .agent import Agent
T = TypeVar("T")
@dataclass
class RunResultBase(abc.ABC):
input: str | list[TResponseInputItem]
"""The original input items i.e. the items before run() was called. This may be a mutated
version of the input, if there are handoff input filters that mutate the input.
"""
new_items: list[RunItem]
"""The new items generated during the agent run. These include things like new messages, tool
calls and their outputs, etc.
"""
raw_responses: list[ModelResponse]
"""The raw LLM responses generated by the model during the agent run."""
final_output: Any
"""The output of the last agent."""
input_guardrail_results: list[InputGuardrailResult]
"""Guardrail results for the input messages."""
output_guardrail_results: list[OutputGuardrailResult]
"""Guardrail results for the final output of the agent."""
context_wrapper: RunContextWrapper[Any]
"""The context wrapper for the agent run."""
@property
@abc.abstractmethod
def last_agent(self) -> Agent[Any]:
"""The last agent that was run."""
def final_output_as(self, cls: type[T], raise_if_incorrect_type: bool = False) -> T:
"""A convenience method to cast the final output to a specific type. By default, the cast
is only for the typechecker. If you set `raise_if_incorrect_type` to True, we'll raise a
TypeError if the final output is not of the given type.
Args:
cls: The type to cast the final output to.
raise_if_incorrect_type: If True, we'll raise a TypeError if the final output is not of
the given type.
Returns:
The final output casted to the given type.
"""
if raise_if_incorrect_type and not isinstance(self.final_output, cls):
raise TypeError(f"Final output is not of type {cls.__name__}")
return cast(T, self.final_output)
def to_input_list(self) -> list[TResponseInputItem]:
"""Creates a new input list, merging the original input with all the new items generated."""
original_items: list[TResponseInputItem] = ItemHelpers.input_to_new_input_list(self.input)
new_items = [item.to_input_item() for item in self.new_items]
return original_items + new_items
@property
def last_response_id(self) -> str | None:
"""Convenience method to get the response ID of the last model response."""
if not self.raw_responses:
return None
return self.raw_responses[-1].response_id
@dataclass
class RunResult(RunResultBase):
_last_agent: Agent[Any]
@property
def last_agent(self) -> Agent[Any]:
"""The last agent that was run."""
return self._last_agent
def __str__(self) -> str:
return pretty_print_result(self)
@dataclass
class RunResultStreaming(RunResultBase):
"""The result of an agent run in streaming mode. You can use the `stream_events` method to
receive semantic events as they are generated.
The streaming method will raise:
- A MaxTurnsExceeded exception if the agent exceeds the max_turns limit.
- A GuardrailTripwireTriggered exception if a guardrail is tripped.
"""
current_agent: Agent[Any]
"""The current agent that is running."""
current_turn: int
"""The current turn number."""
max_turns: int
"""The maximum number of turns the agent can run for."""
final_output: Any
"""The final output of the agent. This is None until the agent has finished running."""
_current_agent_output_schema: AgentOutputSchemaBase | None = field(repr=False)
trace: Trace | None = field(repr=False)
is_complete: bool = False
"""Whether the agent has finished running."""
# Queues that the background run_loop writes to
_event_queue: asyncio.Queue[StreamEvent | QueueCompleteSentinel] = field(
default_factory=asyncio.Queue, repr=False
)
_input_guardrail_queue: asyncio.Queue[InputGuardrailResult] = field(
default_factory=asyncio.Queue, repr=False
)
# Store the asyncio tasks that we're waiting on
_run_impl_task: asyncio.Task[Any] | None = field(default=None, repr=False)
_input_guardrails_task: asyncio.Task[Any] | None = field(default=None, repr=False)
_output_guardrails_task: asyncio.Task[Any] | None = field(default=None, repr=False)
_stored_exception: Exception | None = field(default=None, repr=False)
@property
def last_agent(self) -> Agent[Any]:
"""The last agent that was run. Updates as the agent run progresses, so the true last agent
is only available after the agent run is complete.
"""
return self.current_agent
def cancel(self) -> None:
"""Cancels the streaming run, stopping all background tasks and marking the run as
complete."""
self._cleanup_tasks() # Cancel all running tasks
self.is_complete = True # Mark the run as complete to stop event streaming
# Optionally, clear the event queue to prevent processing stale events
while not self._event_queue.empty():
self._event_queue.get_nowait()
while not self._input_guardrail_queue.empty():
self._input_guardrail_queue.get_nowait()
async def stream_events(self) -> AsyncIterator[StreamEvent]:
"""Stream deltas for new items as they are generated. We're using the types from the
OpenAI Responses API, so these are semantic events: each event has a `type` field that
describes the type of the event, along with the data for that event.
This will raise:
- A MaxTurnsExceeded exception if the agent exceeds the max_turns limit.
- A GuardrailTripwireTriggered exception if a guardrail is tripped.
"""
while True:
self._check_errors()
if self._stored_exception:
logger.debug("Breaking due to stored exception")
self.is_complete = True
break
if self.is_complete and self._event_queue.empty():
break
try:
item = await self._event_queue.get()
except asyncio.CancelledError:
break
if isinstance(item, QueueCompleteSentinel):
self._event_queue.task_done()
# Check for errors, in case the queue was completed due to an exception
self._check_errors()
break
yield item
self._event_queue.task_done()
self._cleanup_tasks()
if self._stored_exception:
raise self._stored_exception
def _create_error_details(self) -> RunErrorDetails:
"""Return a `RunErrorDetails` object considering the current attributes of the class."""
return RunErrorDetails(
input=self.input,
new_items=self.new_items,
raw_responses=self.raw_responses,
last_agent=self.current_agent,
context_wrapper=self.context_wrapper,
input_guardrail_results=self.input_guardrail_results,
output_guardrail_results=self.output_guardrail_results,
)
def _check_errors(self):
if self.current_turn > self.max_turns:
max_turns_exc = MaxTurnsExceeded(f"Max turns ({self.max_turns}) exceeded")
max_turns_exc.run_data = self._create_error_details()
self._stored_exception = max_turns_exc
# Fetch all the completed guardrail results from the queue and raise if needed
while not self._input_guardrail_queue.empty():
guardrail_result = self._input_guardrail_queue.get_nowait()
if guardrail_result.output.tripwire_triggered:
tripwire_exc = InputGuardrailTripwireTriggered(guardrail_result)
tripwire_exc.run_data = self._create_error_details()
self._stored_exception = tripwire_exc
# Check the tasks for any exceptions
if self._run_impl_task and self._run_impl_task.done():
run_impl_exc = self._run_impl_task.exception()
if run_impl_exc and isinstance(run_impl_exc, Exception):
if isinstance(run_impl_exc, AgentsException) and run_impl_exc.run_data is None:
run_impl_exc.run_data = self._create_error_details()
self._stored_exception = run_impl_exc
if self._input_guardrails_task and self._input_guardrails_task.done():
in_guard_exc = self._input_guardrails_task.exception()
if in_guard_exc and isinstance(in_guard_exc, Exception):
if isinstance(in_guard_exc, AgentsException) and in_guard_exc.run_data is None:
in_guard_exc.run_data = self._create_error_details()
self._stored_exception = in_guard_exc
if self._output_guardrails_task and self._output_guardrails_task.done():
out_guard_exc = self._output_guardrails_task.exception()
if out_guard_exc and isinstance(out_guard_exc, Exception):
if isinstance(out_guard_exc, AgentsException) and out_guard_exc.run_data is None:
out_guard_exc.run_data = self._create_error_details()
self._stored_exception = out_guard_exc
def _cleanup_tasks(self):
if self._run_impl_task and not self._run_impl_task.done():
self._run_impl_task.cancel()
if self._input_guardrails_task and not self._input_guardrails_task.done():
self._input_guardrails_task.cancel()
if self._output_guardrails_task and not self._output_guardrails_task.done():
self._output_guardrails_task.cancel()
def __str__(self) -> str:
return pretty_print_run_result_streaming(self)
/.local/lib/python3.10/site-packages/agents/run.py
from __future__ import annotations
import asyncio
import copy
from dataclasses import dataclass, field
from typing import Any, cast
from openai.types.responses import ResponseCompletedEvent
from ._run_impl import (
AgentToolUseTracker,
NextStepFinalOutput,
NextStepHandoff,
NextStepRunAgain,
QueueCompleteSentinel,
RunImpl,
SingleStepResult,
TraceCtxManager,
get_model_tracing_impl,
)
from .agent import Agent
from .agent_output import AgentOutputSchema, AgentOutputSchemaBase
from .exceptions import (
AgentsException,
InputGuardrailTripwireTriggered,
MaxTurnsExceeded,
ModelBehaviorError,
OutputGuardrailTripwireTriggered,
RunErrorDetails,
)
from .guardrail import InputGuardrail, InputGuardrailResult, OutputGuardrail, OutputGuardrailResult
from .handoffs import Handoff, HandoffInputFilter, handoff
from .items import ItemHelpers, ModelResponse, RunItem, TResponseInputItem
from .lifecycle import RunHooks
from .logger import logger
from .model_settings import ModelSettings
from .models.interface import Model, ModelProvider
from .models.multi_provider import MultiProvider
from .result import RunResult, RunResultStreaming
from .run_context import RunContextWrapper, TContext
from .stream_events import AgentUpdatedStreamEvent, RawResponsesStreamEvent
from .tool import Tool
from .tracing import Span, SpanError, agent_span, get_current_trace, trace
from .tracing.span_data import AgentSpanData
from .usage import Usage
from .util import _coro, _error_tracing
DEFAULT_MAX_TURNS = 10
@dataclass
class RunConfig:
"""Configures settings for the entire agent run."""
model: str | Model | None = None
"""The model to use for the entire agent run. If set, will override the model set on every
agent. The model_provider passed in below must be able to resolve this model name.
"""
model_provider: ModelProvider = field(default_factory=MultiProvider)
"""The model provider to use when looking up string model names. Defaults to OpenAI."""
model_settings: ModelSettings | None = None
"""Configure global model settings. Any non-null values will override the agent-specific model
settings.
"""
handoff_input_filter: HandoffInputFilter | None = None
"""A global input filter to apply to all handoffs. If `Handoff.input_filter` is set, then that
will take precedence. The input filter allows you to edit the inputs that are sent to the new
agent. See the documentation in `Handoff.input_filter` for more details.
"""
input_guardrails: list[InputGuardrail[Any]] | None = None
"""A list of input guardrails to run on the initial run input."""
output_guardrails: list[OutputGuardrail[Any]] | None = None
"""A list of output guardrails to run on the final output of the run."""
tracing_disabled: bool = False
"""Whether tracing is disabled for the agent run. If disabled, we will not trace the agent run.
"""
trace_include_sensitive_data: bool = True
"""Whether we include potentially sensitive data (for example: inputs/outputs of tool calls or
LLM generations) in traces. If False, we'll still create spans for these events, but the
sensitive data will not be included.
"""
workflow_name: str = "Agent workflow"
"""The name of the run, used for tracing. Should be a logical name for the run, like
"Code generation workflow" or "Customer support agent".
"""
trace_id: str | None = None
"""A custom trace ID to use for tracing. If not provided, we will generate a new trace ID."""
group_id: str | None = None
"""
A grouping identifier to use for tracing, to link multiple traces from the same conversation
or process. For example, you might use a chat thread ID.
"""
trace_metadata: dict[str, Any] | None = None
"""
An optional dictionary of additional metadata to include with the trace.
"""
class Runner:
@classmethod
async def run(
cls,
starting_agent: Agent[TContext],
input: str | list[TResponseInputItem],
*,
context: TContext | None = None,
max_turns: int = DEFAULT_MAX_TURNS,
hooks: RunHooks[TContext] | None = None,
run_config: RunConfig | None = None,
previous_response_id: str | None = None,
) -> RunResult:
"""Run a workflow starting at the given agent. The agent will run in a loop until a final
output is generated. The loop runs like so:
1. The agent is invoked with the given input.
2. If there is a final output (i.e. the agent produces something of type
`agent.output_type`, the loop terminates.
3. If there's a handoff, we run the loop again, with the new agent.
4. Else, we run tool calls (if any), and re-run the loop.
In two cases, the agent may raise an exception:
1. If the max_turns is exceeded, a MaxTurnsExceeded exception is raised.
2. If a guardrail tripwire is triggered, a GuardrailTripwireTriggered exception is raised.
Note that only the first agent's input guardrails are run.
Args:
starting_agent: The starting agent to run.
input: The initial input to the agent. You can pass a single string for a user message,
or a list of input items.
context: The context to run the agent with.
max_turns: The maximum number of turns to run the agent for. A turn is defined as one
AI invocation (including any tool calls that might occur).
hooks: An object that receives callbacks on various lifecycle events.
run_config: Global settings for the entire agent run.
previous_response_id: The ID of the previous response, if using OpenAI models via the
Responses API, this allows you to skip passing in input from the previous turn.
Returns:
A run result containing all the inputs, guardrail results and the output of the last
agent. Agents may perform handoffs, so we don't know the specific type of the output.
"""
if hooks is None:
hooks = RunHooks[Any]()
if run_config is None:
run_config = RunConfig()
tool_use_tracker = AgentToolUseTracker()
with TraceCtxManager(
workflow_name=run_config.workflow_name,
trace_id=run_config.trace_id,
group_id=run_config.group_id,
metadata=run_config.trace_metadata,
disabled=run_config.tracing_disabled,
):
current_turn = 0
original_input: str | list[TResponseInputItem] = copy.deepcopy(input)
generated_items: list[RunItem] = []
model_responses: list[ModelResponse] = []
context_wrapper: RunContextWrapper[TContext] = RunContextWrapper(
context=context, # type: ignore
)
input_guardrail_results: list[InputGuardrailResult] = []
current_span: Span[AgentSpanData] | None = None
current_agent = starting_agent
should_run_agent_start_hooks = True
try:
while True:
all_tools = await cls._get_all_tools(current_agent, context_wrapper)
# Start an agent span if we don't have one. This span is ended if the current
# agent changes, or if the agent loop ends.
if current_span is None:
handoff_names = [h.agent_name for h in cls._get_handoffs(current_agent)]
if output_schema := cls._get_output_schema(current_agent):
output_type_name = output_schema.name()
else:
output_type_name = "str"
current_span = agent_span(
name=current_agent.name,
handoffs=handoff_names,
output_type=output_type_name,
)
current_span.start(mark_as_current=True)
current_span.span_data.tools = [t.name for t in all_tools]
current_turn += 1
if current_turn > max_turns:
_error_tracing.attach_error_to_span(
current_span,
SpanError(
message="Max turns exceeded",
data={"max_turns": max_turns},
),
)
raise MaxTurnsExceeded(f"Max turns ({max_turns}) exceeded")
logger.debug(
f"Running agent {current_agent.name} (turn {current_turn})",
)
if current_turn == 1:
input_guardrail_results, turn_result = await asyncio.gather(
cls._run_input_guardrails(
starting_agent,
starting_agent.input_guardrails
+ (run_config.input_guardrails or []),
copy.deepcopy(input),
context_wrapper,
),
cls._run_single_turn(
agent=current_agent,
all_tools=all_tools,
original_input=original_input,
generated_items=generated_items,
hooks=hooks,
context_wrapper=context_wrapper,
run_config=run_config,
should_run_agent_start_hooks=should_run_agent_start_hooks,
tool_use_tracker=tool_use_tracker,
previous_response_id=previous_response_id,
),
)
else:
turn_result = await cls._run_single_turn(
agent=current_agent,
all_tools=all_tools,
original_input=original_input,
generated_items=generated_items,
hooks=hooks,
context_wrapper=context_wrapper,
run_config=run_config,
should_run_agent_start_hooks=should_run_agent_start_hooks,
tool_use_tracker=tool_use_tracker,
previous_response_id=previous_response_id,
)
should_run_agent_start_hooks = False
model_responses.append(turn_result.model_response)
original_input = turn_result.original_input
generated_items = turn_result.generated_items
if isinstance(turn_result.next_step, NextStepFinalOutput):
output_guardrail_results = await cls._run_output_guardrails(
current_agent.output_guardrails + (run_config.output_guardrails or []),
current_agent,
turn_result.next_step.output,
context_wrapper,
)
return RunResult(
input=original_input,
new_items=generated_items,
raw_responses=model_responses,
final_output=turn_result.next_step.output,
_last_agent=current_agent,
input_guardrail_results=input_guardrail_results,
output_guardrail_results=output_guardrail_results,
context_wrapper=context_wrapper,
)
elif isinstance(turn_result.next_step, NextStepHandoff):
current_agent = cast(Agent[TContext], turn_result.next_step.new_agent)
current_span.finish(reset_current=True)
current_span = None
should_run_agent_start_hooks = True
elif isinstance(turn_result.next_step, NextStepRunAgain):
pass
else:
raise AgentsException(
f"Unknown next step type: {type(turn_result.next_step)}"
)
except AgentsException as exc:
exc.run_data = RunErrorDetails(
input=original_input,
new_items=generated_items,
raw_responses=model_responses,
last_agent=current_agent,
context_wrapper=context_wrapper,
input_guardrail_results=input_guardrail_results,
output_guardrail_results=[],
)
raise
finally:
if current_span:
current_span.finish(reset_current=True)
@classmethod
def run_sync(
cls,
starting_agent: Agent[TContext],
input: str | list[TResponseInputItem],
*,
context: TContext | None = None,
max_turns: int = DEFAULT_MAX_TURNS,
hooks: RunHooks[TContext] | None = None,
run_config: RunConfig | None = None,
previous_response_id: str | None = None,
) -> RunResult:
"""Run a workflow synchronously, starting at the given agent. Note that this just wraps the
`run` method, so it will not work if there's already an event loop (e.g. inside an async
function, or in a Jupyter notebook or async context like FastAPI). For those cases, use
the `run` method instead.
The agent will run in a loop until a final output is generated. The loop runs like so:
1. The agent is invoked with the given input.
2. If there is a final output (i.e. the agent produces something of type
`agent.output_type`, the loop terminates.
3. If there's a handoff, we run the loop again, with the new agent.
4. Else, we run tool calls (if any), and re-run the loop.
In two cases, the agent may raise an exception:
1. If the max_turns is exceeded, a MaxTurnsExceeded exception is raised.
2. If a guardrail tripwire is triggered, a GuardrailTripwireTriggered exception is raised.
Note that only the first agent's input guardrails are run.
Args:
starting_agent: The starting agent to run.
input: The initial input to the agent. You can pass a single string for a user message,
or a list of input items.
context: The context to run the agent with.
max_turns: The maximum number of turns to run the agent for. A turn is defined as one
AI invocation (including any tool calls that might occur).
hooks: An object that receives callbacks on various lifecycle events.
run_config: Global settings for the entire agent run.
previous_response_id: The ID of the previous response, if using OpenAI models via the
Responses API, this allows you to skip passing in input from the previous turn.
Returns:
A run result containing all the inputs, guardrail results and the output of the last
agent. Agents may perform handoffs, so we don't know the specific type of the output.
"""
return asyncio.get_event_loop().run_until_complete(
cls.run(
starting_agent,
input,
context=context,
max_turns=max_turns,
hooks=hooks,
run_config=run_config,
previous_response_id=previous_response_id,
)
)
@classmethod
def run_streamed(
cls,
starting_agent: Agent[TContext],
input: str | list[TResponseInputItem],
context: TContext | None = None,
max_turns: int = DEFAULT_MAX_TURNS,
hooks: RunHooks[TContext] | None = None,
run_config: RunConfig | None = None,
previous_response_id: str | None = None,
) -> RunResultStreaming:
"""Run a workflow starting at the given agent in streaming mode. The returned result object
contains a method you can use to stream semantic events as they are generated.
The agent will run in a loop until a final output is generated. The loop runs like so:
1. The agent is invoked with the given input.
2. If there is a final output (i.e. the agent produces something of type
`agent.output_type`, the loop terminates.
3. If there's a handoff, we run the loop again, with the new agent.
4. Else, we run tool calls (if any), and re-run the loop.
In two cases, the agent may raise an exception:
1. If the max_turns is exceeded, a MaxTurnsExceeded exception is raised.
2. If a guardrail tripwire is triggered, a GuardrailTripwireTriggered exception is raised.
Note that only the first agent's input guardrails are run.
Args:
starting_agent: The starting agent to run.
input: The initial input to the agent. You can pass a single string for a user message,
or a list of input items.
context: The context to run the agent with.
max_turns: The maximum number of turns to run the agent for. A turn is defined as one
AI invocation (including any tool calls that might occur).
hooks: An object that receives callbacks on various lifecycle events.
run_config: Global settings for the entire agent run.
previous_response_id: The ID of the previous response, if using OpenAI models via the
Responses API, this allows you to skip passing in input from the previous turn.
Returns:
A result object that contains data about the run, as well as a method to stream events.
"""
if hooks is None:
hooks = RunHooks[Any]()
if run_config is None:
run_config = RunConfig()
# If there's already a trace, we don't create a new one. In addition, we can't end the
# trace here, because the actual work is done in `stream_events` and this method ends
# before that.
new_trace = (
None
if get_current_trace()
else trace(
workflow_name=run_config.workflow_name,
trace_id=run_config.trace_id,
group_id=run_config.group_id,
metadata=run_config.trace_metadata,
disabled=run_config.tracing_disabled,
)
)
output_schema = cls._get_output_schema(starting_agent)
context_wrapper: RunContextWrapper[TContext] = RunContextWrapper(
context=context # type: ignore
)
streamed_result = RunResultStreaming(
input=copy.deepcopy(input),
new_items=[],
current_agent=starting_agent,
raw_responses=[],
final_output=None,
is_complete=False,
current_turn=0,
max_turns=max_turns,
input_guardrail_results=[],
output_guardrail_results=[],
_current_agent_output_schema=output_schema,
trace=new_trace,
context_wrapper=context_wrapper,
)
# Kick off the actual agent loop in the background and return the streamed result object.
streamed_result._run_impl_task = asyncio.create_task(
cls._run_streamed_impl(
starting_input=input,
streamed_result=streamed_result,
starting_agent=starting_agent,
max_turns=max_turns,
hooks=hooks,
context_wrapper=context_wrapper,
run_config=run_config,
previous_response_id=previous_response_id,
)
)
return streamed_result
@classmethod
async def _run_input_guardrails_with_queue(
cls,
agent: Agent[Any],
guardrails: list[InputGuardrail[TContext]],
input: str | list[TResponseInputItem],
context: RunContextWrapper[TContext],
streamed_result: RunResultStreaming,
parent_span: Span[Any],
):
queue = streamed_result._input_guardrail_queue
# We'll run the guardrails and push them onto the queue as they complete
guardrail_tasks = [
asyncio.create_task(
RunImpl.run_single_input_guardrail(agent, guardrail, input, context)
)
for guardrail in guardrails
]
guardrail_results = []
try:
for done in asyncio.as_completed(guardrail_tasks):
result = await done
if result.output.tripwire_triggered:
_error_tracing.attach_error_to_span(
parent_span,
SpanError(
message="Guardrail tripwire triggered",
data={
"guardrail": result.guardrail.get_name(),
"type": "input_guardrail",
},
),
)
queue.put_nowait(result)
guardrail_results.append(result)
except Exception:
for t in guardrail_tasks:
t.cancel()
raise
streamed_result.input_guardrail_results = guardrail_results
@classmethod
async def _run_streamed_impl(
cls,
starting_input: str | list[TResponseInputItem],
streamed_result: RunResultStreaming,
starting_agent: Agent[TContext],
max_turns: int,
hooks: RunHooks[TContext],
context_wrapper: RunContextWrapper[TContext],
run_config: RunConfig,
previous_response_id: str | None,
):
if streamed_result.trace:
streamed_result.trace.start(mark_as_current=True)
current_span: Span[AgentSpanData] | None = None
current_agent = starting_agent
current_turn = 0
should_run_agent_start_hooks = True
tool_use_tracker = AgentToolUseTracker()
streamed_result._event_queue.put_nowait(AgentUpdatedStreamEvent(new_agent=current_agent))
try:
while True:
if streamed_result.is_complete:
break
all_tools = await cls._get_all_tools(current_agent, context_wrapper)
# Start an agent span if we don't have one. This span is ended if the current
# agent changes, or if the agent loop ends.
if current_span is None:
handoff_names = [h.agent_name for h in cls._get_handoffs(current_agent)]
if output_schema := cls._get_output_schema(current_agent):
output_type_name = output_schema.name()
else:
output_type_name = "str"
current_span = agent_span(
name=current_agent.name,
handoffs=handoff_names,
output_type=output_type_name,
)
current_span.start(mark_as_current=True)
tool_names = [t.name for t in all_tools]
current_span.span_data.tools = tool_names
current_turn += 1
streamed_result.current_turn = current_turn
if current_turn > max_turns:
_error_tracing.attach_error_to_span(
current_span,
SpanError(
message="Max turns exceeded",
data={"max_turns": max_turns},
),
)
streamed_result._event_queue.put_nowait(QueueCompleteSentinel())
break
if current_turn == 1:
# Run the input guardrails in the background and put the results on the queue
streamed_result._input_guardrails_task = asyncio.create_task(
cls._run_input_guardrails_with_queue(
starting_agent,
starting_agent.input_guardrails + (run_config.input_guardrails or []),
copy.deepcopy(ItemHelpers.input_to_new_input_list(starting_input)),
context_wrapper,
streamed_result,
current_span,
)
)
try:
turn_result = await cls._run_single_turn_streamed(
streamed_result,
current_agent,
hooks,
context_wrapper,
run_config,
should_run_agent_start_hooks,
tool_use_tracker,
all_tools,
previous_response_id,
)
should_run_agent_start_hooks = False
streamed_result.raw_responses = streamed_result.raw_responses + [
turn_result.model_response
]
streamed_result.input = turn_result.original_input
streamed_result.new_items = turn_result.generated_items
if isinstance(turn_result.next_step, NextStepHandoff):
current_agent = turn_result.next_step.new_agent
current_span.finish(reset_current=True)
current_span = None
should_run_agent_start_hooks = True
streamed_result._event_queue.put_nowait(
AgentUpdatedStreamEvent(new_agent=current_agent)
)
elif isinstance(turn_result.next_step, NextStepFinalOutput):
streamed_result._output_guardrails_task = asyncio.create_task(
cls._run_output_guardrails(
current_agent.output_guardrails
+ (run_config.output_guardrails or []),
current_agent,
turn_result.next_step.output,
context_wrapper,
)
)
try:
output_guardrail_results = await streamed_result._output_guardrails_task
except Exception:
# Exceptions will be checked in the stream_events loop
output_guardrail_results = []
streamed_result.output_guardrail_results = output_guardrail_results
streamed_result.final_output = turn_result.next_step.output
streamed_result.is_complete = True
streamed_result._event_queue.put_nowait(QueueCompleteSentinel())
elif isinstance(turn_result.next_step, NextStepRunAgain):
pass
except AgentsException as exc:
streamed_result.is_complete = True
streamed_result._event_queue.put_nowait(QueueCompleteSentinel())
exc.run_data = RunErrorDetails(
input=streamed_result.input,
new_items=streamed_result.new_items,
raw_responses=streamed_result.raw_responses,
last_agent=current_agent,
context_wrapper=context_wrapper,
input_guardrail_results=streamed_result.input_guardrail_results,
output_guardrail_results=streamed_result.output_guardrail_results,
)
raise
except Exception as e:
if current_span:
_error_tracing.attach_error_to_span(
current_span,
SpanError(
message="Error in agent run",
data={"error": str(e)},
),
)
streamed_result.is_complete = True
streamed_result._event_queue.put_nowait(QueueCompleteSentinel())
raise
streamed_result.is_complete = True
finally:
if current_span:
current_span.finish(reset_current=True)
if streamed_result.trace:
streamed_result.trace.finish(reset_current=True)
@classmethod
async def _run_single_turn_streamed(
cls,
streamed_result: RunResultStreaming,
agent: Agent[TContext],
hooks: RunHooks[TContext],
context_wrapper: RunContextWrapper[TContext],
run_config: RunConfig,
should_run_agent_start_hooks: bool,
tool_use_tracker: AgentToolUseTracker,
all_tools: list[Tool],
previous_response_id: str | None,
) -> SingleStepResult:
if should_run_agent_start_hooks:
await asyncio.gather(
hooks.on_agent_start(context_wrapper, agent),
(
agent.hooks.on_start(context_wrapper, agent)
if agent.hooks
else _coro.noop_coroutine()
),
)
output_schema = cls._get_output_schema(agent)
streamed_result.current_agent = agent
streamed_result._current_agent_output_schema = output_schema
system_prompt = await agent.get_system_prompt(context_wrapper)
handoffs = cls._get_handoffs(agent)
model = cls._get_model(agent, run_config)
model_settings = agent.model_settings.resolve(run_config.model_settings)
model_settings = RunImpl.maybe_reset_tool_choice(agent, tool_use_tracker, model_settings)
final_response: ModelResponse | None = None
input = ItemHelpers.input_to_new_input_list(streamed_result.input)
input.extend([item.to_input_item() for item in streamed_result.new_items])
# 1. Stream the output events
async for event in model.stream_response(
system_prompt,
input,
model_settings,
all_tools,
output_schema,
handoffs,
get_model_tracing_impl(
run_config.tracing_disabled, run_config.trace_include_sensitive_data
),
previous_response_id=previous_response_id,
):
if isinstance(event, ResponseCompletedEvent):
usage = (
Usage(
requests=1,
input_tokens=event.response.usage.input_tokens,
output_tokens=event.response.usage.output_tokens,
total_tokens=event.response.usage.total_tokens,
input_tokens_details=event.response.usage.input_tokens_details,
output_tokens_details=event.response.usage.output_tokens_details,
)
if event.response.usage
else Usage()
)
final_response = ModelResponse(
output=event.response.output,
usage=usage,
response_id=event.response.id,
)
context_wrapper.usage.add(usage)
streamed_result._event_queue.put_nowait(RawResponsesStreamEvent(data=event))
# 2. At this point, the streaming is complete for this turn of the agent loop.
if not final_response:
raise ModelBehaviorError("Model did not produce a final response!")
# 3. Now, we can process the turn as we do in the non-streaming case
single_step_result = await cls._get_single_step_result_from_response(
agent=agent,
original_input=streamed_result.input,
pre_step_items=streamed_result.new_items,
new_response=final_response,
output_schema=output_schema,
all_tools=all_tools,
handoffs=handoffs,
hooks=hooks,
context_wrapper=context_wrapper,
run_config=run_config,
tool_use_tracker=tool_use_tracker,
)
RunImpl.stream_step_result_to_queue(single_step_result, streamed_result._event_queue)
return single_step_result
@classmethod
async def _run_single_turn(
cls,
*,
agent: Agent[TContext],
all_tools: list[Tool],
original_input: str | list[TResponseInputItem],
generated_items: list[RunItem],
hooks: RunHooks[TContext],
context_wrapper: RunContextWrapper[TContext],
run_config: RunConfig,
should_run_agent_start_hooks: bool,
tool_use_tracker: AgentToolUseTracker,
previous_response_id: str | None,
) -> SingleStepResult:
# Ensure we run the hooks before anything else
if should_run_agent_start_hooks:
await asyncio.gather(
hooks.on_agent_start(context_wrapper, agent),
(
agent.hooks.on_start(context_wrapper, agent)
if agent.hooks
else _coro.noop_coroutine()
),
)
system_prompt = await agent.get_system_prompt(context_wrapper)
output_schema = cls._get_output_schema(agent)
handoffs = cls._get_handoffs(agent)
input = ItemHelpers.input_to_new_input_list(original_input)
input.extend([generated_item.to_input_item() for generated_item in generated_items])
new_response = await cls._get_new_response(
agent,
system_prompt,
input,
output_schema,
all_tools,
handoffs,
context_wrapper,
run_config,
tool_use_tracker,
previous_response_id,
)
return await cls._get_single_step_result_from_response(
agent=agent,
original_input=original_input,
pre_step_items=generated_items,
new_response=new_response,
output_schema=output_schema,
all_tools=all_tools,
handoffs=handoffs,
hooks=hooks,
context_wrapper=context_wrapper,
run_config=run_config,
tool_use_tracker=tool_use_tracker,
)
@classmethod
async def _get_single_step_result_from_response(
cls,
*,
agent: Agent[TContext],
all_tools: list[Tool],
original_input: str | list[TResponseInputItem],
pre_step_items: list[RunItem],
new_response: ModelResponse,
output_schema: AgentOutputSchemaBase | None,
handoffs: list[Handoff],
hooks: RunHooks[TContext],
context_wrapper: RunContextWrapper[TContext],
run_config: RunConfig,
tool_use_tracker: AgentToolUseTracker,
) -> SingleStepResult:
processed_response = RunImpl.process_model_response(
agent=agent,
all_tools=all_tools,
response=new_response,
output_schema=output_schema,
handoffs=handoffs,
)
tool_use_tracker.add_tool_use(agent, processed_response.tools_used)
return await RunImpl.execute_tools_and_side_effects(
agent=agent,
original_input=original_input,
pre_step_items=pre_step_items,
new_response=new_response,
processed_response=processed_response,
output_schema=output_schema,
hooks=hooks,
context_wrapper=context_wrapper,
run_config=run_config,
)
@classmethod
async def _run_input_guardrails(
cls,
agent: Agent[Any],
guardrails: list[InputGuardrail[TContext]],
input: str | list[TResponseInputItem],
context: RunContextWrapper[TContext],
) -> list[InputGuardrailResult]:
if not guardrails:
return []
guardrail_tasks = [
asyncio.create_task(
RunImpl.run_single_input_guardrail(agent, guardrail, input, context)
)
for guardrail in guardrails
]
guardrail_results = []
for done in asyncio.as_completed(guardrail_tasks):
result = await done
if result.output.tripwire_triggered:
# Cancel all guardrail tasks if a tripwire is triggered.
for t in guardrail_tasks:
t.cancel()
_error_tracing.attach_error_to_current_span(
SpanError(
message="Guardrail tripwire triggered",
data={"guardrail": result.guardrail.get_name()},
)
)
raise InputGuardrailTripwireTriggered(result)
else:
guardrail_results.append(result)
return guardrail_results
@classmethod
async def _run_output_guardrails(
cls,
guardrails: list[OutputGuardrail[TContext]],
agent: Agent[TContext],
agent_output: Any,
context: RunContextWrapper[TContext],
) -> list[OutputGuardrailResult]:
if not guardrails:
return []
guardrail_tasks = [
asyncio.create_task(
RunImpl.run_single_output_guardrail(guardrail, agent, agent_output, context)
)
for guardrail in guardrails
]
guardrail_results = []
for done in asyncio.as_completed(guardrail_tasks):
result = await done
if result.output.tripwire_triggered:
# Cancel all guardrail tasks if a tripwire is triggered.
for t in guardrail_tasks:
t.cancel()
_error_tracing.attach_error_to_current_span(
SpanError(
message="Guardrail tripwire triggered",
data={"guardrail": result.guardrail.get_name()},
)
)
raise OutputGuardrailTripwireTriggered(result)
else:
guardrail_results.append(result)
return guardrail_results
@classmethod
async def _get_new_response(
cls,
agent: Agent[TContext],
system_prompt: str | None,
input: list[TResponseInputItem],
output_schema: AgentOutputSchemaBase | None,
all_tools: list[Tool],
handoffs: list[Handoff],
context_wrapper: RunContextWrapper[TContext],
run_config: RunConfig,
tool_use_tracker: AgentToolUseTracker,
previous_response_id: str | None,
) -> ModelResponse:
model = cls._get_model(agent, run_config)
model_settings = agent.model_settings.resolve(run_config.model_settings)
model_settings = RunImpl.maybe_reset_tool_choice(agent, tool_use_tracker, model_settings)
new_response = await model.get_response(
system_instructions=system_prompt,
input=input,
model_settings=model_settings,
tools=all_tools,
output_schema=output_schema,
handoffs=handoffs,
tracing=get_model_tracing_impl(
run_config.tracing_disabled, run_config.trace_include_sensitive_data
),
previous_response_id=previous_response_id,
)
context_wrapper.usage.add(new_response.usage)
return new_response
@classmethod
def _get_output_schema(cls, agent: Agent[Any]) -> AgentOutputSchemaBase | None:
if agent.output_type is None or agent.output_type is str:
return None
elif isinstance(agent.output_type, AgentOutputSchemaBase):
return agent.output_type
return AgentOutputSchema(agent.output_type)
@classmethod
def _get_handoffs(cls, agent: Agent[Any]) -> list[Handoff]:
handoffs = []
for handoff_item in agent.handoffs:
if isinstance(handoff_item, Handoff):
handoffs.append(handoff_item)
elif isinstance(handoff_item, Agent):
handoffs.append(handoff(handoff_item))
return handoffs
@classmethod
async def _get_all_tools(
cls, agent: Agent[Any], context_wrapper: RunContextWrapper[Any]
) -> list[Tool]:
return await agent.get_all_tools(context_wrapper)
@classmethod
def _get_model(cls, agent: Agent[Any], run_config: RunConfig) -> Model:
if isinstance(run_config.model, Model):
return run_config.model
elif isinstance(run_config.model, str):
return run_config.model_provider.get_model(run_config.model)
elif isinstance(agent.model, Model):
return agent.model
return run_config.model_provider.get_model(agent.model)