openai-agents-python 结果接口完全指南:从 RunResult 到流式生命周期与运行恢复 📅 发布时间:2026/9/11 19:34:44 👁 浏览次数: 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_queueasyncio.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_idsrc/agents/result.py。而agent_tool_invocation则检查context_wrapper是否为ToolContext实例若是则构造一个不可变的AgentToolInvocation含tool_name、tool_call_id、tool_arguments普通顶层运行的该属性为Nonesrc/agents/result.py。最终输出final_outputfinal_output属性包含最后运行的智能体所生成的最终输出。它可能是如果最后一个智能体未定义output_type则为str如果最后一个智能体定义了输出类型则为last_agent.output_type类型的对象如果运行在生成最终输出之前停止则为None例如因审批中断而暂停注意final_output的类型标注为Any。任务转移可能会改变完成运行的智能体因此 SDK 无法静态确定所有可能的输出类型。在流式传输模式下final_output会一直保持为None直到流处理完成。有关逐事件流程请参阅 流式传输指南。源码中还提供了一个配套的便捷方法final_output_as(cls, raise_if_incorrect_typeFalse)默认仅做类型检查器层面的转换当raise_if_incorrect_typeTrue且实际类型不符时抛出TypeErrorsrc/agents/result.py适合在结构化输出场景下安全读取最终结果。输入、下一轮历史记录和新项目这些接口分别回答不同的问题属性或辅助方法包含的内容最适合input此运行片段的基础输入。如果任务转移输入过滤器重写了历史记录这里会反映运行继续使用的已过滤输入。审核此运行实际使用的输入to_input_list()运行的输入项视图。默认的modepreserve_all会保留来自new_items的转换后历史记录但不会再次追加已移入 SDK 默认嵌套任务转移历史记录中的同一会话项当任务转移过滤重写模型历史记录时modenormalized会优先采用规范的延续输入。手动聊天循环、由客户端管理的对话状态以及普通项目形式的历史记录检查new_items包含智能体、工具、任务转移和审批元数据的丰富RunItem包装器。日志、UI、审核和调试raw_responses运行中每次模型调用产生的原始ModelResponse对象。提供商级别的诊断或原始响应检查实际使用时如果需要运行的普通输入项视图请使用to_input_list()。如果在任务转移过滤或嵌套任务转移历史记录重写后需要用于下一次Runner.run(..., input...)调用的规范本地输入请使用to_input_list(modenormalized)。如果希望 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转换而来的重放项目最终返回原始输入 重放历史的完整输入列表。modenormalized的实现位于_input_items_for_resultsrc/agents/result.py只有运行器显式标记了_replay_from_model_input_items分歧例如任务转移过滤重写了模型历史时才会改用_model_input_items作为规范延续输入大多数普通运行下它与preserve_all结果一致。当 SDK 默认的嵌套任务转移历史记录逐字保留某个消息项时Sessions、RunState和to_input_list()会追踪准确的自有项实例而不是按内容去重。分别出现的相同消息仍会保持分离只会避免再次追加已经归属其中的项实例。这一点对应源码中的NestedHistoryOwnedItemRef机制src/agents/result.pySDK 通过_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.pyInputItem表示在恢复后的模型调用之前立即从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_agentlast_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_stepsrc/agents/result.py。当ToolCallOutputItem的输出是 Pydantic 模型或数据类时RunState会将该输出序列化为结构化数据。RunState还会遍历字典、列表和元组并转换在这些容器中遇到的 Pydantic 模型或数据类经过 JSON 往返转换后元组会还原为列表。其他与 JSON 不兼容的值可能会回退为其字符串表示形式因此如果某个自定义类型必须在序列化后保持精确请返回明确与 JSON 兼容的数据。from agents import Agent, Runner agent Agent(nameAssistant, instructionsUse 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_replayTrue)运行器可能会重新发送暂存输入提供商侧的工作也可能重复执行。成功接纳的输入会在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_idlast_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 dataclasssrc/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.pyimmediate默认立即停止取消所有任务并清空队列after_turn优雅地完成当前轮次后再停止允许 LLM 响应完成、执行待处理的工具调用、正确保存会话状态并准确记录用量然后在下一轮开始前停止Python 不会公开单独的流式completedpromise 或error属性。导致运行终止的流式传输失败会由stream_events()抛出而is_complete会反映运行是否已达到终止状态。原始响应raw_responsesraw_responses包含运行期间收集的原始模型响应。多步骤运行可能会生成多个响应例如在任务转移期间或重复的模型/工具/模型循环中。last_response_id只是raw_responses中最后一个条目的 ID。每个ModelResponse还会公开两项适用于单次模型调用的诊断信息字段定义见 src/agents/items.pyrequest_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_usageTrue)启用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完成展示、审计与续接审批中断用interruptionsto_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),仅供参考