Results

RunResultBase dataclass

Bases: ABC

Source code in agents/result.py

@dataclass
class RunResultBase(abc.ABC):
    input: str | list[TResponseInputItem]
    """原始输入项，即调用 run() 之前的输入项。如果有 handoff 输入过滤器对输入进行了变更，这里可能是变更后的版本。"""

    new_items: list[RunItem]
    """在 agent 运行过程中生成的新项。这些包括新消息、工具调用及其输出等。"""

    raw_responses: list[ModelResponse]
    """在 agent 运行过程中由模型生成的原始 LLM 响应。"""

    final_output: Any
    """最后一个 agent 的输出。"""

    input_guardrail_results: list[InputGuardrailResult]
    """输入消息的 guardrail（护栏）检测结果。"""

    output_guardrail_results: list[OutputGuardrailResult]
    """agent 最终输出的 guardrail（护栏）检测结果。"""

    context_wrapper: RunContextWrapper[Any]
    """agent 运行的上下文包装器。"""

    @property
    @abc.abstractmethod
    def last_agent(self) -> Agent[Any]:
        """最后运行的 agent。"""

    def final_output_as(self, cls: type[T], raise_if_incorrect_type: bool = False) -> T:
        """一个便捷方法，将 final_output 转换为指定类型。默认情况下，这个转换只对类型检查器有效。
        如果将 `raise_if_incorrect_type` 设置为 True，当 final_output 不是指定类型时会抛出 TypeError。

        参数:
            cls: 要将 final_output 转换成的类型。
            raise_if_incorrect_type: 如果为 True，当 final_output 不是指定类型时会抛出 TypeError。

        返回:
            转换为指定类型的 final_output。
        """
        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]:
        """创建一个新的输入列表，将原始输入与所有新生成的项合并。"""
        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:
        """获取最后一个模型响应的 response_id 的便捷方法。"""
        if not self.raw_responses:
            return None

        return self.raw_responses[-1].response_id

input instance-attribute

input: str | list[TResponseInputItem]

原始输入项，即调用 run() 之前的输入项。如果有 handoff 输入过滤器对输入进行了变更，这里可能是变更后的版本。

new_items instance-attribute

new_items: list[RunItem]

在 agent 运行过程中生成的新项。这些包括新消息、工具调用及其输出等。

raw_responses instance-attribute

raw_responses: list[ModelResponse]

在 agent 运行过程中由模型生成的原始 LLM 响应。

final_output instance-attribute

final_output: Any

最后一个 agent 的输出。

input_guardrail_results instance-attribute

input_guardrail_results: list[InputGuardrailResult]

输入消息的 guardrail（护栏）检测结果。

output_guardrail_results instance-attribute

output_guardrail_results: list[OutputGuardrailResult]

agent 最终输出的 guardrail（护栏）检测结果。

context_wrapper instance-attribute

context_wrapper: RunContextWrapper[Any]

agent 运行的上下文包装器。

last_agent abstractmethod property

last_agent: Agent[Any]

最后运行的 agent。

last_response_id property

last_response_id: str | None

获取最后一个模型响应的 response_id 的便捷方法。

final_output_as

final_output_as(cls: type[T], raise_if_incorrect_type: bool = False) -> T

一个便捷方法，将 final_output 转换为指定类型。默认情况下，这个转换只对类型检查器有效。 如果将 raise_if_incorrect_type 设置为 True，当 final_output 不是指定类型时会抛出 TypeError。

参数

cls: 要将 final_output 转换成的类型。 raise_if_incorrect_type: 如果为 True，当 final_output 不是指定类型时会抛出 TypeError。

返回

转换为指定类型的 final_output。

Source code in agents/result.py

def final_output_as(self, cls: type[T], raise_if_incorrect_type: bool = False) -> T:
    """一个便捷方法，将 final_output 转换为指定类型。默认情况下，这个转换只对类型检查器有效。
    如果将 `raise_if_incorrect_type` 设置为 True，当 final_output 不是指定类型时会抛出 TypeError。

    参数:
        cls: 要将 final_output 转换成的类型。
        raise_if_incorrect_type: 如果为 True，当 final_output 不是指定类型时会抛出 TypeError。

    返回:
        转换为指定类型的 final_output。
    """
    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)

to_input_list

to_input_list() -> list[TResponseInputItem]

创建一个新的输入列表，将原始输入与所有新生成的项合并。

Source code in agents/result.py

def to_input_list(self) -> list[TResponseInputItem]:
    """创建一个新的输入列表，将原始输入与所有新生成的项合并。"""
    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

RunResult dataclass

Bases: RunResultBase

Source code in agents/result.py

@dataclass
class RunResult(RunResultBase):
    _last_agent: Agent[Any]

    @property
    def last_agent(self) -> Agent[Any]:
        """最后运行的 agent。"""
        return self._last_agent

    def __str__(self) -> str:
        return pretty_print_result(self)

last_agent property

last_agent: Agent[Any]

最后运行的 agent。

input instance-attribute

input: str | list[TResponseInputItem]

原始输入项，即调用 run() 之前的输入项。如果有 handoff 输入过滤器对输入进行了变更，这里可能是变更后的版本。

new_items instance-attribute

new_items: list[RunItem]

在 agent 运行过程中生成的新项。这些包括新消息、工具调用及其输出等。

raw_responses instance-attribute

raw_responses: list[ModelResponse]

在 agent 运行过程中由模型生成的原始 LLM 响应。

final_output instance-attribute

final_output: Any

最后一个 agent 的输出。

input_guardrail_results instance-attribute

input_guardrail_results: list[InputGuardrailResult]

输入消息的 guardrail（护栏）检测结果。

output_guardrail_results instance-attribute

output_guardrail_results: list[OutputGuardrailResult]

agent 最终输出的 guardrail（护栏）检测结果。

context_wrapper instance-attribute

context_wrapper: RunContextWrapper[Any]

agent 运行的上下文包装器。

last_response_id property

last_response_id: str | None

获取最后一个模型响应的 response_id 的便捷方法。

final_output_as

final_output_as(cls: type[T], raise_if_incorrect_type: bool = False) -> T

一个便捷方法，将 final_output 转换为指定类型。默认情况下，这个转换只对类型检查器有效。 如果将 raise_if_incorrect_type 设置为 True，当 final_output 不是指定类型时会抛出 TypeError。

参数

cls: 要将 final_output 转换成的类型。 raise_if_incorrect_type: 如果为 True，当 final_output 不是指定类型时会抛出 TypeError。

返回

转换为指定类型的 final_output。

Source code in agents/result.py

def final_output_as(self, cls: type[T], raise_if_incorrect_type: bool = False) -> T:
    """一个便捷方法，将 final_output 转换为指定类型。默认情况下，这个转换只对类型检查器有效。
    如果将 `raise_if_incorrect_type` 设置为 True，当 final_output 不是指定类型时会抛出 TypeError。

    参数:
        cls: 要将 final_output 转换成的类型。
        raise_if_incorrect_type: 如果为 True，当 final_output 不是指定类型时会抛出 TypeError。

    返回:
        转换为指定类型的 final_output。
    """
    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)

to_input_list

to_input_list() -> list[TResponseInputItem]

创建一个新的输入列表，将原始输入与所有新生成的项合并。

Source code in agents/result.py

def to_input_list(self) -> list[TResponseInputItem]:
    """创建一个新的输入列表，将原始输入与所有新生成的项合并。"""
    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

RunResultStreaming dataclass

Bases: RunResultBase

以流式模式运行 agent 的结果。你可以使用 stream_events 方法实时接收语义事件。

流式方法会抛出： - 如果 agent 超过 max_turns 限制，会抛出 MaxTurnsExceeded 异常。 - 如果 guardrail 被触发，会抛出 GuardrailTripwireTriggered 异常。

Source code in agents/result.py

@dataclass
class RunResultStreaming(RunResultBase):
    """以流式模式运行 agent 的结果。你可以使用 `stream_events` 方法实时接收语义事件。

    流式方法会抛出：
    - 如果 agent 超过 max_turns 限制，会抛出 MaxTurnsExceeded 异常。
    - 如果 guardrail 被触发，会抛出 GuardrailTripwireTriggered 异常。
    """

    current_agent: Agent[Any]
    """当前正在运行的 agent。"""

    current_turn: int
    """当前轮次编号。"""

    max_turns: int
    """agent 允许运行的最大轮次数。"""

    final_output: Any
    """agent 的最终输出。在 agent 完成运行前为 None。"""

    _current_agent_output_schema: AgentOutputSchemaBase | None = field(repr=False)

    trace: Trace | None = field(repr=False)

    is_complete: bool = False
    """agent 是否已经完成运行。"""

    # 后台 run_loop 写入的队列
    _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
    )

    # 存储正在等待的 asyncio 任务
    _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]:
        """最后运行的 agent。随着 agent 运行进展会更新，只有在 agent 运行完成后才是真正的最后一个 agent。"""
        return self.current_agent

    def cancel(self) -> None:
        """取消流式运行，停止所有后台任务并将运行标记为已完成。"""
        self._cleanup_tasks()  # 取消所有正在运行的任务
        self.is_complete = True  # 标记运行已完成，停止事件流

        # 可选：清空事件队列，防止处理过期事件
        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]:
        """在新项生成时流式传递增量。我们使用 OpenAI Responses API 的类型，这些是语义事件：
        每个事件有一个 `type` 字段描述事件类型，以及该事件的数据。

        该方法会抛出：
        - 如果 agent 超过 max_turns 限制，会抛出 MaxTurnsExceeded 异常。
        - 如果 guardrail 被触发，会抛出 GuardrailTripwireTriggered 异常。
        """
        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()
                # 检查错误，以防队列因异常而完成
                self._check_errors()
                break

            yield item
            self._event_queue.task_done()

        self._cleanup_tasks()

        if self._stored_exception:
            raise self._stored_exception

    def _check_errors(self):
        if self.current_turn > self.max_turns:
            self._stored_exception = MaxTurnsExceeded(f"Max turns ({self.max_turns}) exceeded")

        # 从队列中获取所有已完成的 guardrail 结果，如有需要则抛出异常
        while not self._input_guardrail_queue.empty():
            guardrail_result = self._input_guardrail_queue.get_nowait()
            if guardrail_result.output.tripwire_triggered:
                self._stored_exception = InputGuardrailTripwireTriggered(guardrail_result)

        # 检查任务是否有异常
        if self._run_impl_task and self._run_impl_task.done():
            exc = self._run_impl_task.exception()
            if exc and isinstance(exc, Exception):
                self._stored_exception = exc

        if self._input_guardrails_task and self._input_guardrails_task.done():
            exc = self._input_guardrails_task.exception()
            if exc and isinstance(exc, Exception):
                self._stored_exception = exc

        if self._output_guardrails_task and self._output_guardrails_task.done():
            exc = self._output_guardrails_task.exception()
            if exc and isinstance(exc, Exception):
                self._stored_exception = 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)

current_agent instance-attribute

current_agent: Agent[Any]

当前正在运行的 agent。

current_turn instance-attribute

current_turn: int

当前轮次编号。

max_turns instance-attribute

max_turns: int

agent 允许运行的最大轮次数。

final_output instance-attribute

final_output: Any

agent 的最终输出。在 agent 完成运行前为 None。

is_complete class-attribute instance-attribute

is_complete: bool = False

agent 是否已经完成运行。

last_agent property

last_agent: Agent[Any]

最后运行的 agent。随着 agent 运行进展会更新，只有在 agent 运行完成后才是真正的最后一个 agent。

input instance-attribute

input: str | list[TResponseInputItem]

原始输入项，即调用 run() 之前的输入项。如果有 handoff 输入过滤器对输入进行了变更，这里可能是变更后的版本。

new_items instance-attribute

new_items: list[RunItem]

在 agent 运行过程中生成的新项。这些包括新消息、工具调用及其输出等。

raw_responses instance-attribute

raw_responses: list[ModelResponse]

在 agent 运行过程中由模型生成的原始 LLM 响应。

input_guardrail_results instance-attribute

input_guardrail_results: list[InputGuardrailResult]

输入消息的 guardrail（护栏）检测结果。

output_guardrail_results instance-attribute

output_guardrail_results: list[OutputGuardrailResult]

agent 最终输出的 guardrail（护栏）检测结果。

context_wrapper instance-attribute

context_wrapper: RunContextWrapper[Any]

agent 运行的上下文包装器。

last_response_id property

last_response_id: str | None

获取最后一个模型响应的 response_id 的便捷方法。

cancel

cancel() -> None

取消流式运行，停止所有后台任务并将运行标记为已完成。

Source code in agents/result.py

def cancel(self) -> None:
    """取消流式运行，停止所有后台任务并将运行标记为已完成。"""
    self._cleanup_tasks()  # 取消所有正在运行的任务
    self.is_complete = True  # 标记运行已完成，停止事件流

    # 可选：清空事件队列，防止处理过期事件
    while not self._event_queue.empty():
        self._event_queue.get_nowait()
    while not self._input_guardrail_queue.empty():
        self._input_guardrail_queue.get_nowait()

stream_events async

stream_events() -> AsyncIterator[StreamEvent]

在新项生成时流式传递增量。我们使用 OpenAI Responses API 的类型，这些是语义事件： 每个事件有一个 type 字段描述事件类型，以及该事件的数据。

该方法会抛出： - 如果 agent 超过 max_turns 限制，会抛出 MaxTurnsExceeded 异常。 - 如果 guardrail 被触发，会抛出 GuardrailTripwireTriggered 异常。

Source code in agents/result.py

async def stream_events(self) -> AsyncIterator[StreamEvent]:
    """在新项生成时流式传递增量。我们使用 OpenAI Responses API 的类型，这些是语义事件：
    每个事件有一个 `type` 字段描述事件类型，以及该事件的数据。

    该方法会抛出：
    - 如果 agent 超过 max_turns 限制，会抛出 MaxTurnsExceeded 异常。
    - 如果 guardrail 被触发，会抛出 GuardrailTripwireTriggered 异常。
    """
    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()
            # 检查错误，以防队列因异常而完成
            self._check_errors()
            break

        yield item
        self._event_queue.task_done()

    self._cleanup_tasks()

    if self._stored_exception:
        raise self._stored_exception

final_output_as

final_output_as(cls: type[T], raise_if_incorrect_type: bool = False) -> T

一个便捷方法，将 final_output 转换为指定类型。默认情况下，这个转换只对类型检查器有效。 如果将 raise_if_incorrect_type 设置为 True，当 final_output 不是指定类型时会抛出 TypeError。

参数

cls: 要将 final_output 转换成的类型。 raise_if_incorrect_type: 如果为 True，当 final_output 不是指定类型时会抛出 TypeError。

返回

转换为指定类型的 final_output。

Source code in agents/result.py

def final_output_as(self, cls: type[T], raise_if_incorrect_type: bool = False) -> T:
    """一个便捷方法，将 final_output 转换为指定类型。默认情况下，这个转换只对类型检查器有效。
    如果将 `raise_if_incorrect_type` 设置为 True，当 final_output 不是指定类型时会抛出 TypeError。

    参数:
        cls: 要将 final_output 转换成的类型。
        raise_if_incorrect_type: 如果为 True，当 final_output 不是指定类型时会抛出 TypeError。

    返回:
        转换为指定类型的 final_output。
    """
    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)

to_input_list

to_input_list() -> list[TResponseInputItem]

创建一个新的输入列表，将原始输入与所有新生成的项合并。

Source code in agents/result.py

def to_input_list(self) -> list[TResponseInputItem]:
    """创建一个新的输入列表，将原始输入与所有新生成的项合并。"""
    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