openai-agents-python 结果接口完全指南:从 RunResult 到流式生命周期与运行恢复
【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python
调用Runner.run系列方法后,SDK 会返回包含最终输出、运行项、原始模型响应与恢复快照的结果对象。本文以 docs/zh/results.md 为核心,结合 src/agents/result.py、src/agents/run_state.py 等源码实现,系统讲解RunResult/RunResultStreaming的每个结果面、如何选择合适的接口、如何续接或恢复对话,以及流式运行的生命周期与诊断信息,读完即可在真实多智能体应用中正确消费运行结果。
结果类型体系:两种结果、一个共享基类
调用Runner.run方法时,你会收到以下两种结果类型之一:
- 来自
Runner.run(...)或Runner.run_sync(...)的RunResult - 来自
Runner.run_streamed(...)的RunResultStreaming
两者都继承自RunResultBase,后者公开了共享的结果接口,例如final_output、new_items、last_agent、raw_responses和to_state()。在 src/agents/result.py 中可以看到,RunResultBase是一个抽象基类,声明了input、new_items、raw_responses、final_output、四组安全防护措施结果数组和context_wrapper等字段,并把last_agent定义为抽象属性,由具体子类分别实现。
RunResultStreaming增加了流式传输专用的控制项:stream_events()、current_agent、is_complete和cancel(...)。从源码看,它内部维护了_event_queue(asyncio.Queue[StreamEvent | QueueCompleteSentinel])、后台run_loop_task以及多个安全防护措施任务队列,这些是流式事件逐条产出机制的基础(src/agents/result.py)。
合适的结果接口:按需求选属性
大多数应用只需要少数几个结果属性或辅助方法。下表直接对应官方文档给出的选型矩阵:
| 如果你需要…… | 使用 |
|---|---|
| 向用户显示的最终答案 | final_output |
| 包含完整本地对话记录、可供重放的下一轮输入列表 | to_input_list() |
| 包含智能体、工具、任务转移和审批元数据的丰富运行项 | new_items |
| 通常应处理下一轮用户输入的智能体 | last_agent |
使用previous_response_id的 OpenAI Responses API 链式调用 | last_response_id |
| 待处理的审批和可恢复的快照 | interruptions和to_state() |
当前嵌套Agent.as_tool()调用的元数据 | agent_tool_invocation |
| 原始模型调用或安全防护措施诊断信息 | raw_responses和安全防护措施结果数组 |
从源码实现看,last_response_id本质上是一个便捷属性:它直接返回raw_responses列表中最后一个ModelResponse的response_id(src/agents/result.py)。而agent_tool_invocation则检查context_wrapper是否为ToolContext实例,若是则构造一个不可变的AgentToolInvocation(含tool_name、tool_call_id、tool_arguments),普通顶层运行的该属性为None(src/agents/result.py)。
最终输出:final_output
final_output属性包含最后运行的智能体所生成的最终输出。它可能是:
- 如果最后一个智能体未定义
output_type,则为str - 如果最后一个智能体定义了输出类型,则为
last_agent.output_type类型的对象 - 如果运行在生成最终输出之前停止,则为
None,例如因审批中断而暂停
注意:
final_output的类型标注为Any。任务转移可能会改变完成运行的智能体,因此 SDK 无法静态确定所有可能的输出类型。
在流式传输模式下,final_output会一直保持为None,直到流处理完成。有关逐事件流程,请参阅 流式传输指南。
源码中还提供了一个配套的便捷方法final_output_as(cls, raise_if_incorrect_type=False):默认仅做类型检查器层面的转换,当raise_if_incorrect_type=True且实际类型不符时抛出TypeError(src/agents/result.py),适合在结构化输出场景下安全读取最终结果。
输入、下一轮历史记录和新项目
这些接口分别回答不同的问题:
| 属性或辅助方法 | 包含的内容 | 最适合 |
|---|---|---|
input | 此运行片段的基础输入。如果任务转移输入过滤器重写了历史记录,这里会反映运行继续使用的已过滤输入。 | 审核此运行实际使用的输入 |
to_input_list() | 运行的输入项视图。默认的mode="preserve_all"会保留来自new_items的转换后历史记录,但不会再次追加已移入 SDK 默认嵌套任务转移历史记录中的同一会话项;当任务转移过滤重写模型历史记录时,mode="normalized"会优先采用规范的延续输入。 | 手动聊天循环、由客户端管理的对话状态,以及普通项目形式的历史记录检查 |
new_items | 包含智能体、工具、任务转移和审批元数据的丰富RunItem包装器。 | 日志、UI、审核和调试 |
raw_responses | 运行中每次模型调用产生的原始ModelResponse对象。 | 提供商级别的诊断或原始响应检查 |
实际使用时:
- 如果需要运行的普通输入项视图,请使用
to_input_list()。 - 如果在任务转移过滤或嵌套任务转移历史记录重写后,需要用于下一次
Runner.run(..., input=...)调用的规范本地输入,请使用to_input_list(mode="normalized")。 - 如果希望 SDK 为你加载和保存历史记录,请使用
session=...(参见 会话文档)。 - 如果正在使用通过
conversation_id或previous_response_id实现的 OpenAI 服务器托管状态,通常只需传递新的用户输入并复用已存储的 ID,而不是重新发送to_input_list()。 - 如果日志、UI 或审核需要完整的转换后历史记录,请使用默认的
to_input_list()模式或new_items。
to_input_list()的实现细节在 src/agents/result.py:它会将公共输入通过ItemHelpers.input_to_new_input_list规范化,再叠加由new_items转换而来的重放项目,最终返回原始输入 + 重放历史的完整输入列表。mode="normalized"的实现位于_input_items_for_result(src/agents/result.py):只有运行器显式标记了_replay_from_model_input_items分歧(例如任务转移过滤重写了模型历史)时,才会改用_model_input_items作为规范延续输入;大多数普通运行下它与preserve_all结果一致。
当 SDK 默认的嵌套任务转移历史记录逐字保留某个消息项时,Sessions、RunState和to_input_list()会追踪准确的自有项实例,而不是按内容去重。分别出现的相同消息仍会保持分离;只会避免再次追加已经归属其中的项实例。这一点对应源码中的NestedHistoryOwnedItemRef机制(src/agents/result.py):SDK 通过_nested_history_owned_session_item_refs记录已归属的历史项引用,并用摘要(digest)与索引坐标校验所有权,防止同一消息项被重复追加。
与 JavaScript SDK 不同,Python 不会公开单独的output属性来仅包含运行期间新生成的模型格式项目。需要 SDK 元数据时,请使用new_items;需要原始模型载荷时,请检查raw_responses。
将计算机工具项目作为对话输入重新提交时,会使用原始 Responses 载荷结构。预览模型的computer_call项目会保留单个action,而gpt-5.5计算机调用可以保留批量的actions[]。to_input_list()和RunState会保留模型生成的结构,因此,在将这些项目手动重新提交为对话输入时,暂停/恢复流程和已存储的对话记录都能继续兼容预览版和 GA 版计算机工具调用。本地执行结果仍会在new_items中显示为computer_call_output项目。
新项目:new_items 的常见类型
new_items提供运行过程中所发生事件的最丰富视图。常见项目类型包括(各类的完整定义参见 src/agents/items.py):
InputItem,表示在恢复后的模型调用之前立即从RunState.pending_input接纳的输入MessageOutputItem,表示助手消息ReasoningItem,表示推理项目ToolSearchCallItem和ToolSearchOutputItem,表示 Responses 工具搜索请求和已加载的工具搜索结果ToolCallItem和ToolCallOutputItem,表示工具调用及其结果ToolApprovalItem,表示因等待审批而暂停的工具调用MCPApprovalRequestItem、MCPApprovalResponseItem和MCPListToolsItem,表示托管 MCP 的审批和工具目录HandoffCallItem和HandoffOutputItem,表示任务转移请求和已完成的转移
只要需要智能体关联信息、工具输出、任务转移边界或审批边界,就应选择new_items,而不是to_input_list()。
使用托管工具搜索时,请检查ToolSearchCallItem.raw_item以查看模型发出的搜索请求,并检查ToolSearchOutputItem.raw_item以查看该轮加载了哪些命名空间、函数或托管 MCP 服务器。
使用程序化工具调用时,生成的program是一个ToolCallItem,该程序拥有的普通子工具调用也是ToolCallItem条目,而对应的program_output是一个ToolCallOutputItem。程序拥有的托管 MCPmcp_approval_request和mcp_list_tools项目属于例外:它们会成为MCPApprovalRequestItem和MCPListToolsItem条目。
原始项目可以是有类型的 Responses 对象或映射。特别是,程序拥有的 shell 和 apply-patch 调用使用映射。请使用映射安全的检查模式:
from collections.abc import Mapping def raw_field(item, name): raw_item = item.raw_item if isinstance(raw_item, Mapping): return raw_item.get(name) return getattr(raw_item, name, None) raw_type = raw_field(item, "type") caller = raw_field(item, "caller") caller_id = ( caller.get("caller_id") if isinstance(caller, Mapping) else getattr(caller, "caller_id", None) )对于程序拥有的子调用,caller的type字段为program,而caller_id用于标识父程序调用。
对话的继续或恢复
下一轮智能体:last_agent
last_agent包含最后运行的智能体。任务转移后,它通常是下一轮用户输入最适合复用的智能体。
在流式传输模式下,RunResultStreaming.current_agent会随着运行进展而更新,因此你可以在流结束前观察任务转移。
中断和运行状态:interruptions 与 to_state()
如果某个工具需要审批,待处理的审批会公开在RunResult.interruptions或RunResultStreaming.interruptions中。其中可能包括直接工具、任务转移后调用的工具,或嵌套Agent.as_tool()运行所触发的审批。
调用to_state()以捕获可恢复的RunState,批准或拒绝待处理项目,然后使用Runner.run(...)或Runner.run_streamed(...)恢复运行。在源码中,RunResult.to_state()会基于当前结果构造新的RunState(保留原始输入、起始智能体、max_turns、当前轮次、已处理响应、会话持久化计数和工具使用追踪快照),并把中断步骤写入_current_step(src/agents/result.py)。
当ToolCallOutputItem的输出是 Pydantic 模型或数据类时,RunState会将该输出序列化为结构化数据。RunState还会遍历字典、列表和元组,并转换在这些容器中遇到的 Pydantic 模型或数据类;经过 JSON 往返转换后,元组会还原为列表。其他与 JSON 不兼容的值可能会回退为其字符串表示形式,因此,如果某个自定义类型必须在序列化后保持精确,请返回明确与 JSON 兼容的数据。
from agents import Agent, Runner agent = Agent(name="Assistant", instructions="Use tools when needed.") result = await Runner.run(agent, "Delete temp files that are no longer needed.") if result.interruptions: state = result.to_state() for interruption in result.interruptions: state.approve(interruption) result = await Runner.run(agent, state)RunState.approve()与RunState.reject()的实现支持嵌套场景:approve()会先通过_find_nested_approval_state判断该审批是否属于嵌套Agent.as_tool()运行,若是则递归到嵌套状态处理;否则把审批解析到当前状态的权威待审批项,再调用context.approve_tool(...)(src/agents/run_state.py)。reject()还支持rejection_message参数,将精确文本回传给模型(src/agents/run_state.py)。
恢复前添加输入:RunState.add_input()
如果运行在暂停后,或在完成一轮后停止,但尚未执行未完成运行中的下一次模型调用时有新的用户输入到达,请使用RunState.add_input()。字符串会成为一条用户消息,多次调用会保留插入顺序。暂存输入是已序列化RunState的一部分,因此在to_json()/from_json()和to_string()/from_string()往返转换后仍会保留。
state = result.to_state() state.add_input("Also keep the generated report in the project folder.") for interruption in state.get_interruptions(): state.approve(interruption) result = await Runner.run(agent, state)恢复时,运行器仅对暂存输入应用当前智能体的输入安全防护措施,以及RunConfig中的输入安全防护措施。配置由客户端管理的Session后,运行器会将已接受的暂存输入转换为持久化的InputItem,等待会话写入完成,然后才发出模型请求。如果没有由客户端管理的会话或服务器托管的对话,运行器会在发出模型请求前,将已接受的暂存输入转换为InputItem。对于服务器托管的对话,输入会保持待处理状态,直到服务器请求接受它。在序列化、恢复和可安全重放的重试过程中,SDK 会保留一个持久化的InputItem实例。此 SDK 实例保证并不代表提供商交付保证:如果请求可能已到达提供商后,重试策略返回RetryDecision(approve_unsafe_replay=True),运行器可能会重新发送暂存输入,提供商侧的工作也可能重复执行。成功接纳的输入会在new_items中显示为InputItem。读取RunState.pending_input可获取一个分离副本(源码中通过copy.deepcopy返回,src/agents/run_state.py),或调用RunState.clear_pending_input()在恢复前丢弃所有暂存输入(src/agents/run_state.py)。
RunState.add_input()会拒绝以下状态(对应 src/agents/run_state.py 中的前置校验逻辑):
- 终止状态(
_current_step不是NextStepInterruption或NextStepRunAgain) - 没有剩余模型轮次的状态(
_current_turn >= _max_turns) - 已接受的模型响应正在等待本地处理的状态(
response_accepted为真) - 待处理工具结果可能在下一次模型调用前结束运行的中断状态(如
stop_on_first_tool或stop_at_tool_names命中、callable工具行为)
在这些情况下,应完成当前运行,然后开始新的用户轮次。
对于流式传输运行,请先完成对stream_events()的消费,然后检查result.interruptions,并从result.to_state()恢复。有关完整审批流程,请参阅 人在回路指南。
服务器托管的延续:last_response_id
last_response_id是运行中最新的模型响应 ID。如果希望在下一轮继续 OpenAI Responses API 链,请将其作为previous_response_id传回。
如果已通过to_input_list()、session或conversation_id继续对话,通常不需要last_response_id。如果需要多步骤运行中的每个模型响应,请改为检查raw_responses。
智能体作为工具的元数据
当结果来自嵌套的Agent.as_tool()运行时,agent_tool_invocation会公开有关外层Agent.as_tool()调用的不可变元数据:
tool_nametool_call_idtool_arguments
对于普通的顶层运行,agent_tool_invocation为None。对应的AgentToolInvocation是一个 frozen dataclass(src/agents/result.py),保证元数据不可变。
这在custom_output_extractor中尤其有用,因为在对嵌套结果进行后处理时,你可能需要外层Agent.as_tool()调用的工具名称、调用 ID 或原始参数。有关相关的Agent.as_tool()模式,请参阅 工具指南。
如果还需要该嵌套运行的已解析结构化输入,请读取context_wrapper.tool_input。这是RunState为嵌套工具输入进行通用序列化的字段,而agent_tool_invocation会直接在结果中公开当前嵌套调用的元数据。
流式传输生命周期和诊断
RunResultStreaming继承了上述相同的结果接口,但增加了流式传输专用的控制项:
stream_events(),用于消费语义流事件current_agent,用于在运行过程中追踪活动智能体is_complete,用于查看流式传输运行是否已完全结束cancel(...),用于立即停止运行或在当前轮次结束后停止运行
持续消费stream_events(),直到异步迭代器结束。只有该迭代器结束后,流式传输运行才算完成;在最后一个可见 token 到达后,final_output、interruptions、raw_responses等汇总属性以及会话持久化副作用可能仍在收尾。从源码看,stream_events()内部会等待QueueCompleteSentinel、清理后台任务、等待输入安全防护措施任务收尾,并在迭代器结束前通过_check_errors()检查是否存储了异常(如MaxTurnsExceeded或安全防护措施 tripwire),最后统一抛出(src/agents/result.py)。
如果调用cancel(),请继续消费stream_events(),以便正确完成取消和清理。cancel(mode=...)支持两种模式(src/agents/result.py):
"immediate"(默认):立即停止,取消所有任务并清空队列"after_turn":优雅地完成当前轮次后再停止,允许 LLM 响应完成、执行待处理的工具调用、正确保存会话状态并准确记录用量,然后在下一轮开始前停止
Python 不会公开单独的流式completedpromise 或error属性。导致运行终止的流式传输失败会由stream_events()抛出,而is_complete会反映运行是否已达到终止状态。
原始响应:raw_responses
raw_responses包含运行期间收集的原始模型响应。多步骤运行可能会生成多个响应,例如在任务转移期间或重复的模型/工具/模型循环中。
last_response_id只是raw_responses中最后一个条目的 ID。
每个ModelResponse还会公开两项适用于单次模型调用的诊断信息(字段定义见 src/agents/items.py):
request_id是模型适配器和传输层传播请求 ID 时的传输请求 ID。内置的OpenAIResponsesModel和OpenAIChatCompletionsModel会在其 HTTP 和 SSE 传输路径中传播可用的、由服务器生成的x-request-id。当配置的端点为 OpenAI API 时,请在生产环境中记录非None值,以便将故障与 OpenAI 支持关联起来;对于与 OpenAI 兼容的提供商或代理,请改用相应服务的支持渠道。OpenAIResponsesWSModel当前会将request_id保持为None。第三方适配器不保证会传播请求 ID。AnyLLM Chat Completions 适配器和LitellmModel当前会将request_id保持为None。当 Agents SDK AnyLLM Responses 适配器在规范化提供商响应时未保留传输请求 ID,它也可能会将request_id保持为None。raw_usage是可选启用的、与 JSON 兼容的提供商用量载荷快照,捕获时机是在 Agents SDK 规范化该载荷之前。使用ModelSettings(preserve_raw_usage=True)启用raw_usage;该参数定义于 src/agents/model_settings.py,具体语义请参阅 保留提供商用量载荷。
ModelResponse.request_id和ModelResponse.raw_usage都可能是None,因此应将这些值视为可选诊断信息,而不是对话状态。
安全防护措施结果
智能体级安全防护措施分别通过input_guardrail_results和output_guardrail_results公开。
工具安全防护措施则分别通过tool_input_guardrail_results和tool_output_guardrail_results公开。
这些数组会在整个运行期间持续累积,因此可用于记录决策、存储额外的安全防护措施元数据,或调试运行被阻止的原因。
当智能体级输出安全防护措施阻止由终止函数工具直接生成的最终输出时,会应用一条脱敏规则。对于当前被阻止的响应,output_guardrail_results会替换被拒绝的智能体输出,并清除包含载荷的输出元数据,而tool_output_guardrail_results会替换包含载荷的工具元数据。此前已接受的结果保持不变。经过净化的输出安全防护措施结果会在OutputGuardrailTripwireTriggered上公开为guardrail_result。经过净化的输出安全防护措施和工具输出安全防护措施结果也会通过流式传输结果状态和RunState公开;请参阅 输出安全防护措施。
上下文和用量
context_wrapper会公开你的应用上下文,以及由 SDK 管理的运行时元数据,例如审批、用量和嵌套的tool_input。
用量会在context_wrapper.usage上追踪。对于流式传输运行,用量总计可能会滞后,直到处理完流的最后几个数据块。有关完整的包装器结构和持久化注意事项,请参阅 上下文管理指南。
小结
RunResult与RunResultStreaming是消费智能体运行结果的统一入口:普通运行用final_output、new_items、last_agent、raw_responses完成展示、审计与续接;审批中断用interruptions+to_state()+RunState实现可序列化、可恢复的人机协作闭环;流式运行则通过stream_events()的完整消费来驱动生命周期结束,并借助request_id与raw_usage做生产级诊断。理解 src/agents/result.py 与 src/agents/run_state.py 的底层实现,能帮助你在多智能体、任务转移与审批恢复等复杂场景中精准选用正确的接口。
【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考