尧图网站建设 
openai-agents-python 实战Runner 执行机制、RunConfig 调优与多轮会话管理完整指南【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python本篇技术指南以 openai-agents-python 框架的Runner执行体系为核心系统讲解如何运行 Agent、理解代理循环agent loop的底层判定逻辑、通过RunConfig精确调控单次运行的模型/追踪/工具行为、在四种内存策略中选择合适的多轮会话方案以及利用错误处理器与异常体系构建健壮的恢复路径。读完本文你将能够从会调 API进阶到能掌控运行全生命周期独立设计可生产化的多 Agent 工作流。Runner 的三种运行方式在 openai-agents-python 中Agent 的启动入口统一收敛在Runner类上。它提供三种调用方式分别覆盖异步、同步与流式三种场景Runner.run()异步执行返回RunResult。Runner.run_sync()同步方法内部只是简单包装并运行.run()。Runner.run_streamed()异步执行返回RunResultStreaming它以流式模式调用 LLM并在事件到达时即刻推送给调用方。最基本的用法如下from agents import Agent, Runner async def main(): agent Agent(nameAssistant, instructionsYou are a helpful assistant) result await Runner.run(agent, Write a haiku about recursion in programming.) print(result.final_output) # Code within the code, # Functions calling themselves, # Infinite loops dance运行结束后RunResult上携带的final_output、new_items、last_response_id、usage等字段共同构成了本次运行的完整结果视图详细字段说明可参考 结果指南。执行器生命周期代理循环The Agent Loop三种输入形态调用上述任一Runner方法时需要传入起始 Agent 和输入。从run()的签名可以看到输入可以是字符串被当作一条用户消息处理OpenAI Responses API 格式的输入项列表list[TResponseInputItem]RunState用于恢复一个被暂停的运行或恢复被cancel(modeafter_turn)中断的运行该状态还可以携带为下次恢复的模型调用准备好的待定输入。循环的判定逻辑执行器随后进入循环。在源码层面这一循环由 run_loop.py 中的run_single_turn驱动每一步产生的下一步动作由 run_steps.py 中的NextStepHandoff/NextStepFinalOutput/NextStepRunAgain/NextStepInterruption四种类型表达与下述流程一一对应用当前输入调用当前 Agent 的 LLM。LLM 产生输出后执行器进行三分支判定判定为最终输出→ 循环结束返回结果请求握手handoff→ 更新当前 Agent 与输入重新进入循环产生工具调用→ 执行这些工具调用、追加结果重新进入循环。若超过传入的max_turns则抛出MaxTurnsExceeded异常传入max_turnsNone可禁用该轮次上限。注意LLM 输出被判定为最终输出的规则是——它生成了期望类型的文本输出且没有任何工具调用。这一点在ProcessedResponse.has_tools_or_approvals_to_run()run_steps.py中有直接体现只要本轮响应中还存在 handoffs、函数调用、computer 动作、shell 调用、apply_patch 调用或待审批的 MCP 请求运行就尚未终结。另外max_turns的默认值定义在 run_config.py 的DEFAULT_MAX_TURNS 10即不显式传参时默认允许 10 轮 LLM 调用。流式执行与 Responses WebSocket 传输流式事件流式模式允许在 LLM 运行期间持续接收流式事件。流结束后RunResultStreaming会包含本次运行的完整信息包括所有新产生的输出通过.stream_events()遍历流式事件。更完整的流式消费模式参见 流式指南。Responses WebSocket 传输可选辅助器启用 OpenAI Responses WebSocket 传输后依然可以沿用普通的RunnerAPI。为复用连接SDK 推荐使用 WebSocket 会话辅助器但并非强制。需要强调的是这是基于 WebSocket 传输的 Responses API而非 Realtime API。关于具体模型对象或自定义提供者相关的传输选择规则与注意事项参见 模型文档。模式 1不使用会话辅助器可用当只需要 WebSocket 传输、不需要 SDK 代为管理共享提供者/会话时使用import asyncio from agents import Agent, Runner, set_default_openai_responses_transport async def main(): set_default_openai_responses_transport(websocket) agent Agent(nameAssistant, instructionsBe concise.) result Runner.run_streamed(agent, Summarize recursion in one sentence.) async for event in result.stream_events(): if event.type raw_response_event: continue print(event.type) asyncio.run(main())这种模式适合单次运行。如果反复调用Runner.run()/Runner.run_streamed()除非手动复用同一个RunConfig/ provider 实例否则每次运行都可能重新建立连接。模式 2使用responses_websocket_session()多轮复用推荐当需要在多次运行间共享支持 WebSocket 的 provider 与RunConfig包括继承同一run_config的嵌套 agent-as-tool 调用时使用responses_websocket_session()import asyncio from agents import Agent, responses_websocket_session async def main(): agent Agent(nameAssistant, instructionsBe concise.) async with responses_websocket_session( responses_websocket_options{ping_interval: 20.0, ping_timeout: 60.0}, ) as ws: first ws.run_streamed(agent, Say hello in one short sentence.) async for _event in first.stream_events(): pass second ws.run_streamed( agent, Now say goodbye., previous_response_idfirst.last_response_id, ) async for _event in second.stream_events(): pass asyncio.run(main())几点重要的工程约束务必在退出上下文前消费完流式结果。若 WebSocket 请求仍在进行中就退出上下文可能强制关闭共享连接。服务在每个 WebSocket 连接上一次只处理一个响应并将连接时长限制为 60 分钟。辅助器复用了连接但并不会消除这些限制。重连之后storeFalse与 ZDR 流程无法恢复未被缓存的previous_response_id应使用完整输入上下文开启新链或从本地管理的会话状态重建。完整的恢复行为参见 Responses WebSocket 传输注意事项。若长推理轮次触发 WebSocket keepalive 超时可调大ping_timeout或将其设为ping_timeoutNone以禁用心跳超时。当可靠性比 WebSocket 延迟更重要时改用 HTTP/SSE 传输。RunConfig单次运行的全局配置中枢run_config参数允许在不修改每个 Agent 定义的前提下覆盖单次运行的部分全局设置。RunConfig的完整字段定义位于 run_config.py它属于 dataclass 配置类型支持直接传字典进行强制转换。下面按类别逐一说明。模型、提供者与会话默认值model设置全局 LLM 模型无论每个 Agent 自身配置了什么model都生效。model_provider负责按模型名查找模型的提供者默认为 OpenAI。model_settings覆盖 Agent 级设置例如可全局设置temperature或top_p。session_settings运行中取回历史记录时覆盖会话级默认值如SessionSettings(limit...)。session_input_callback使用 Sessions 时自定义每次Runner运行前新用户输入与会话历史的合并方式回调支持同步或异步。护栏、握手与模型输入塑形input_guardrails、output_guardrails要在所有运行中包含的输入/输出护栏列表。handoff_input_filter应用于所有握手的全局输入过滤器若该握手本身未定义过滤器。输入过滤器允许编辑发送给新 Agent 的输入详见Handoff.input_filter的文档。nest_handoff_history选择加入的 beta 功能在调用下一个 Agent 前将可摘要的历史压缩为有序的 assistant 摘要片段同时把无损失的 message 项保留在原始位置。该功能默认关闭设为True开启保持False则原样透传原始转录。Sessions、RunState与RunResult.to_input_list()在 SDK 默认嵌套历史已持有某条消息实例时不会重复追加同时保留彼此独立但相同的消息。所有 Runner 方法 在未传入RunConfig时会自动创建一个因此快速上手示例默认保持关闭显式的Handoff.input_filter回调仍可覆盖该设置。单个握手可通过Handoff.nest_handoff_history覆盖此设置。handoff_history_mapper可选 callable在选择nest_handoff_history时接收归一化后的转录历史 握手项必须返回要转发给下一个 Agent 的确切输入项列表用于在不编写完整握手过滤器的情况下替换内置的有序摘要片段。call_model_input_filter在模型调用前一刻编辑完全准备好的模型输入instructions 与输入项的钩子例如裁剪历史或注入系统提示。reasoning_item_id_policy控制执行器将先前输出转换为下一轮模型输入时是否保留推理项的 ID。嵌套握手以 opt-in beta 形式提供。要启用有序转录压缩可传RunConfig(nest_handoff_historyTrue)或对特定握手设置handoff(..., nest_handoff_historyTrue)。内置 mapper 会在无损失 message 项周围放置生成的 assistant 摘要片段而不是把整个转录折叠成一条消息。若偏好保持原始转录默认则不设置该标志或提供按需原样转发对话的handoff_input_filter或handoff_history_mapper。若想在不用自定义 mapper 的情况下修改生成摘要片段所用的包装文本可调用set_conversation_history_wrappers恢复默认值则调用reset_conversation_history_wrappers。追踪与可观测性tracing_disabled为整个运行禁用追踪。tracing传入TracingConfig以覆盖追踪导出设置如单次运行的追踪 API Key。trace_include_sensitive_data配置追踪是否包含潜在敏感数据如 LLM 与工具调用的输入/输出。其默认值由环境变量OPENAI_AGENTS_TRACE_INCLUDE_SENSITIVE_DATA决定见 run_config.py。workflow_name、trace_id、group_id设置运行的工作流名称、追踪 ID 与追踪组 ID。建议至少设置workflow_namegroup_id为可选字段可将多次运行的追踪串接起来。trace_metadata要附加到所有追踪上的元数据。工具执行、审批与工具错误行为tool_execution配置本地工具调用的 SDK 侧执行行为例如限制同时执行的本地函数工具调用数量。tool_not_found_behavior配置执行器如何处理模型发出的函数工具调用名与当前 Agent 可用工具均不匹配的情况。默认抛出ModelBehaviorError可 opt-in 改为返回模型可见的错误输出。tool_name_collision_policy配置执行器如何处理无命名空间的函数工具与握手名称冲突。默认值warn会记录可操作警告并仅暴露当前分发的胜出者error会在调用模型前抛出UserError。对带命名空间与延迟加载工具的严格校验保持不变。tool_error_formatter自定义模型可见的工具错误消息例如审批拒绝与 opt-in 的 tool-not-found 输出。RunConfig 细节深入tool_execution本地函数工具并发与审批前护栏当需要配置本地函数工具的 SDK 侧执行行为时使用tool_execution例如限制单次运行的本地函数工具并发度from agents import Agent, RunConfig, Runner, ToolExecutionConfig agent Agent(nameAssistant, tools[...]) result await Runner.run( agent, Run the required tool calls., run_configRunConfig( tool_executionToolExecutionConfig( max_function_tool_concurrency2, pre_approval_tool_input_guardrailsTrue, ), ), )max_function_tool_concurrencyNone保持默认行为模型在一轮中发出多个函数工具调用时SDK 会启动全部本地函数工具调用设置整数值则限制同时运行的本地函数工具调用数量。源码层面run_config.py还会校验该值必须 ≥ 1否则抛出ValueError。这与提供者侧的ModelSettings.parallel_tool_calls是两回事parallel_tool_calls控制模型是否被允许在单个响应中发出多个工具调用tool_execution.max_function_tool_concurrency则控制模型发出调用后SDK 如何执行本地函数工具调用。pre_approval_tool_input_guardrailsFalse保持默认审批流程函数工具需要审批时先暂停运行工具输入护栏仅在审批通过后、执行前一刻运行。设为True可在发出待处理审批中断interruption之前先运行函数工具输入护栏。通过该审批前检查的调用在审批后仍会再次运行相同的输入护栏因此时间敏感的检查会在执行前被重新验证。tool_not_found_behavior让运行保持可恢复默认情况下若模型发出的函数工具调用与当前 Agent 可用工具均不匹配执行器抛出ModelBehaviorError。希望运行保持可恢复时设置tool_not_found_behaviorreturn_error_to_model。在该模式下SDK 会为无法解析的工具调用追加一个function_call_output并重新运行模型使模型可以选用可用工具或在不使用该工具的情况下作答from agents import Agent, RunConfig, Runner agent Agent(nameAssistant, tools[...]) result await Runner.run( agent, Handle this request with the available tools., run_configRunConfig(tool_not_found_behaviorreturn_error_to_model), )目前该选项仅适用于工具名查找失败的函数工具调用其他无效的工具载荷仍沿用既有错误行为。tool_error_formatter自定义模型可见的工具错误消息当 SDK 创建模型可见的工具错误输出时用tool_error_formatter自定义返回给模型的消息。格式化器接收包含以下字段的ToolErrorFormatterArgskind错误类别如approval_rejected或tool_not_foundtool_type工具运行时function、computer、shell、apply_patch或customtool_name工具名call_id工具调用 IDdefault_messageSDK 默认的模型可见消息run_context当前运行的上下文包装器。返回字符串以替换消息返回None则使用 SDK 默认值from agents import Agent, RunConfig, Runner, ToolErrorFormatterArgs def format_rejection(args: ToolErrorFormatterArgs[None]) - str | None: if args.kind approval_rejected: return ( fTool call {args.tool_name} was rejected by a human reviewer. Ask for confirmation or propose a safer alternative. ) if args.kind tool_not_found: return fTool {args.tool_name} is not available. Choose one of the listed tools. return None agent Agent(nameAssistant) result Runner.run_sync( agent, Please delete the production database., run_configRunConfig(tool_error_formatterformat_rejection), )reasoning_item_id_policy推理项 ID 保留策略reasoning_item_id_policy控制执行器将历史带入下一轮时例如使用RunResult.to_input_list()或基于 session 的运行如何把推理项转换为下一轮模型输入None或preserve默认保留推理项 IDomit从生成的下一轮输入中移除推理项 ID。omit主要用于规避一类 Responses API 400 错误——推理项携带id却没有必需的后继项例如Item rs_... of type reasoning was provided without its required following item.。当 SDK 从先前输出构建后续输入包括会话持久化、服务端管理的对话增量、流式/非流式后续轮次与恢复路径且保留了推理项 ID、而提供者要求该 ID 必须与其后继项配对时多轮 Agent 运行就可能触发此类问题。设置reasoning_item_id_policyomit会保留推理内容但剥离推理项id从而避免在 SDK 生成的后续输入中触发该 API 不变式。作用域说明仅改变 SDK 构建后续输入时生成/转发的推理项不重写用户提供的初始输入项应用该策略后call_model_input_filter仍可有意识地重新引入推理 ID。状态与会话管理选择内存策略将状态带入下一轮的常见方式有四种下表对比了各自的适用场景策略状态存放位置适用场景下一轮要传递的内容result.to_input_list()应用内存小型聊天循环、完全手动控制、任意提供者result.to_input_list()的列表 下一条用户消息session你的存储 SDK持久聊天状态、可恢复运行、自定义存储同一个session实例或指向同一存储的另一个实例conversation_idOpenAI Conversations API想在多个 worker 或服务间共享的具名服务端会话同一个conversation_id 仅新的用户轮次previous_response_idOpenAI Responses API不创建会话资源的轻量服务端托管续接result.last_response_id 仅新的用户轮次result.to_input_list()与session由客户端管理conversation_id与previous_response_id由 OpenAI 管理且仅在使用 OpenAI Responses API 时适用。大多数应用应为每个对话选择一种持久化策略除非刻意协调两层数据否则将客户端管理的历史与 OpenAI 托管状态混用会导致上下文重复。注意同一运行中会话持久化不能与服务端托管的会话设置conversation_id、previous_response_id或auto_previous_response_id组合使用每次调用请选择一种方式。对话/聊天线程调用任一运行方法可能驱动一个或多个 Agent进而发生一次或多次 LLM 调用但在聊天对话中它只代表一个逻辑轮次。例如用户轮次用户输入文本执行器运行第一个 Agent 调用 LLM、运行工具、握手到第二个 Agent第二个 Agent 再运行更多工具并产生输出。Agent 运行结束时你可以选择向用户展示什么——例如展示 Agent 生成的全部新项或只展示最终输出。无论哪种方式用户随后可能提出追问此时再次调用运行方法即可。手动对话管理可以使用RunResultBase.to_input_list()获取下一轮输入手动管理对话历史from agents import Agent, Runner, trace async def main(): agent Agent(nameAssistant, instructionsReply very concisely.) thread_id thread_123 # Example thread ID with trace(workflow_nameConversation, group_idthread_id): # First turn result await Runner.run(agent, What city is the Golden Gate Bridge in?) print(result.final_output) # San Francisco # Second turn new_input result.to_input_list() [{role: user, content: What state is it in?}] result await Runner.run(agent, new_input) print(result.final_output) # California使用 Sessions 自动管理对话更简单的方式是使用 Sessions无需手动调用.to_input_list()即可自动处理对话历史from agents import Agent, Runner, SQLiteSession, trace async def main(): agent Agent(nameAssistant, instructionsReply very concisely.) # Create session instance session SQLiteSession(conversation_123) thread_id thread_123 # Example thread ID with trace(workflow_nameConversation, group_idthread_id): # First turn result await Runner.run(agent, What city is the Golden Gate Bridge in?, sessionsession) print(result.final_output) # San Francisco # Second turn - agent automatically remembers previous context result await Runner.run(agent, What state is it in?, sessionsession) print(result.final_output) # CaliforniaSessions 会自动完成三件事每次运行前取回对话历史每次运行后保存新消息对不同 session ID 维护相互独立的会话。SQLiteSession的实现位于 sqlite_session.py支持自定义db_path、sessions_table与messages_table等参数。更多细节参见 Sessions 文档。服务端托管的对话也可以让 OpenAI 的对话状态功能在服务端管理对话状态而不是用to_input_list()或Sessions在本地处理。这样可以保留对话历史无需手动重发所有历史消息。使用下面任一种服务端托管方式时每次请求只传新一轮的输入并复用保存的 ID。OpenAI 提供两种跨轮次跟踪状态的方式。方式 1使用conversation_id先用 OpenAI Conversations API 创建对话然后在后续每次调用中复用其 IDfrom agents import Agent, Runner from openai import AsyncOpenAI client AsyncOpenAI() async def main(): agent Agent(nameAssistant, instructionsReply very concisely.) # Create a server-managed conversation conversation await client.conversations.create() conv_id conversation.id while True: user_input input(You: ) result await Runner.run(agent, user_input, conversation_idconv_id) print(fAssistant: {result.final_output})方式 2使用previous_response_id响应链式续接另一种选项是响应链式续接response chaining每一轮显式链接到上一轮响应的 IDfrom agents import Agent, Runner async def main(): agent Agent(nameAssistant, instructionsReply very concisely.) previous_response_id None while True: user_input input(You: ) # Setting auto_previous_response_idTrue enables response chaining automatically # for the first turn, even when theres no actual previous response ID yet. result await Runner.run( agent, user_input, previous_response_idprevious_response_id, auto_previous_response_idTrue, ) previous_response_id result.last_response_id print(fAssistant: {result.final_output})如果运行因审批而暂停并从RunState恢复SDK 会保留保存的conversation_id/previous_response_id/auto_previous_response_id设置使恢复后的轮次继续处于同一个服务端托管的对话中。conversation_id与previous_response_id互斥需要可在系统间共享的具名对话资源时用conversation_id需要轮次之间最轻量的 Responses API 续接原语时用previous_response_id。注意SDK 会以退避backoff方式自动重试conversation_locked错误。在服务端托管的对话运行中重试前会回卷内部对话追踪器输入使相同准备项能被干净地重新发送。在本地基于 session 的运行中不能与conversation_id、previous_response_id或auto_previous_response_id组合SDK 也会对最近持久化的输入项做尽力而为的回滚以减少重试后产生的重复历史条目。即使未配置ModelSettings.retry该兼容性重试也会发生。关于模型请求更广泛的 opt-in 重试行为参见 Runner 托管重试。钩子与自定义call_model_input_filter使用call_model_input_filter可在模型调用前一刻编辑模型输入。钩子接收当前 Agent、上下文以及合并后的输入项存在会话历史时包含在内并返回新的ModelInputData。返回值必须是ModelInputData对象其input字段为必填且必须是输入项列表返回其他形状会抛出UserErrorfrom agents import Agent, Runner, RunConfig from agents.run import CallModelData, ModelInputData def drop_old_messages(data: CallModelData[None]) - ModelInputData: # Keep only the last 5 items and preserve existing instructions. trimmed data.model_data.input[-5:] return ModelInputData(inputtrimmed, instructionsdata.model_data.instructions) agent Agent(nameAssistant, instructionsAnswer concisely.) result Runner.run_sync( agent, Explain quines, run_configRunConfig(call_model_input_filterdrop_old_messages), )执行器会把准备好的输入列表的副本传给钩子因此你可以在不原地修改调用方原始列表的情况下裁剪、替换或重排其中的项。若正在使用 sessioncall_model_input_filter会在会话历史已加载并与当前轮次合并之后运行。若想自定义更早的合并步骤本身请使用session_input_callback。若正通过conversation_id、previous_response_id或auto_previous_response_id使用 OpenAI 服务端托管对话状态钩子会在为下一次 Responses API 调用准备的载荷上运行。该载荷可能只体现新轮次的增量而非先前历史的完整回放只有你返回的项会被标记为在该服务端托管续接中发送。通过run_config按运行设置钩子可用于脱敏敏感数据、裁剪过长历史或注入额外的系统指引。错误与恢复错误处理器Error Handlers所有Runner入口都接受error_handlers——一个以错误种类为键的字典。支持的键为max_turns、model_refusal与invalid_final_output。当希望返回受控的最终输出而不是以相应错误终止运行时使用它们。max_turns示例——超过轮次上限时返回友好的兜底文案from agents import ( Agent, RunErrorHandlerInput, RunErrorHandlerResult, Runner, ) agent Agent(nameAssistant, instructionsBe concise.) def on_max_turns(_data: RunErrorHandlerInput[None]) - RunErrorHandlerResult: return RunErrorHandlerResult( final_outputI couldnt finish within the turn limit. Please narrow the request., include_in_historyFalse, ) result Runner.run_sync( agent, Analyze this long transcript, max_turns3, error_handlers{max_turns: on_max_turns}, ) print(result.final_output)invalid_final_output用于模型消息无法通过 Agent 结构化output_type校验、或模型未返回结构化最终消息的场景。处理器可返回应用特定的兜底值SDK 会用同一个output_type对其进行校验。它不会重试模型调用也不会重放任何工具副作用。返回None表示拒绝恢复若无兜底值非空的校验失败会继续抛出ModelBehaviorError空的结构化响应则保持既有的下一轮行为from pydantic import BaseModel from agents import Agent, ModelBehaviorError, RunErrorHandlerInput, Runner class Recipe(BaseModel): ingredients: list[str] recovered_from_invalid_output: bool False def on_invalid_final_output(data: RunErrorHandlerInput[None]) - Recipe: assert isinstance(data.error, ModelBehaviorError) return Recipe(ingredients[], recovered_from_invalid_outputTrue) agent Agent( nameRecipe assistant, instructionsReturn a structured recipe., output_typeRecipe, ) result Runner.run_sync( agent, Plan tonights dinner., error_handlers{invalid_final_output: on_invalid_final_output}, ) print(result.final_output)RunErrorHandlerResult.include_in_history默认为True。对 max-turns 处理器而言这会把合成的兜底输出追加到对话历史并持久化到已配置的 session若希望把兜底值返回给调用方但不加入结果历史或 session 存储请设置include_in_historyFalse。model_refusal用于让模型拒绝refusal产生应用特定的兜底值而不是以ModelRefusalError终止运行from pydantic import BaseModel from agents import Agent, ModelRefusalError, RunErrorHandlerInput, Runner class Recipe(BaseModel): ingredients: list[str] refusal_reason: str | None None def on_model_refusal(data: RunErrorHandlerInput[None]) - Recipe: assert isinstance(data.error, ModelRefusalError) return Recipe(ingredients[], refusal_reasondata.error.refusal) agent Agent( nameRecipe assistant, instructionsReturn a structured recipe., output_typeRecipe, ) result Runner.run_sync( agent, Make me something unsafe., error_handlers{model_refusal: on_model_refusal}, ) print(result.final_output)持久化执行集成与人类在环HITL工具审批的暂停/恢复模式请先阅读专门的 人类在环指南。以下集成适用于运行可能跨越长时间等待、重试或进程重启的持久化编排场景DaprAgents SDK 的 DaprDiagrid集成可运行具备故障自动恢复能力、并支持人类在环工作流的持久化长时运行 Agent。Dapr 是供应商中立的 CNCF 工作流编排器。仓库中提供了对应的容器级集成测试test_dapr_redis.py与示例dapr_session_example.py。TemporalAgents SDK 的 Temporal 集成可运行包括人类在环任务在内的持久化长时工作流。RestateAgents SDK 的 Restate 集成可构建轻量级持久化 Agent覆盖人工审批、握手与会话管理。该集成以 Restate 的单二进制运行时为依赖支持以进程/容器或 serverless 函数方式运行 Agent。DBOSAgents SDK 的 DBOS 集成可运行在故障与重启后仍保留进度的可靠 Agent支持长时运行 Agent、人类在环工作流与握手同步与异步方法均可仅需 SQLite 或 Postgres 数据库。异常体系一览SDK 在特定情况下会抛出异常完整清单位于agents.exceptions。概览如下AgentsExceptionSDK 抛出的所有异常的基类是其他具体异常的通用父类型。MaxTurnsExceededAgent 运行超过传给Runner.run、Runner.run_sync或Runner.run_streamed方法的max_turns上限时抛出表示 Agent 未能在指定数量的代理循环轮次LLM 调用内完成任务。设置max_turnsNone可禁用该上限。ModelTimeoutError单次模型调用尝试超过ModelSettings.timeout时抛出。作用范围与重试行为参见 模型调用超时。ModelBehaviorError底层模型LLM产生意外或无效输出时抛出可能包括畸形 JSON模型在工具调用或直接输出中给出畸形 JSON 结构尤其是定义了特定output_type时意外工具相关失败模型未按预期方式使用工具失败或不完整的非流式 Responses 调用OpenAIResponsesModel与AnyLLMModel的 Responses 路径在返回的响应终止状态为failed或incomplete时抛出异常会标识终止状态并包含响应中可用的错误或未完成详情。ToolTimeoutError函数工具调用超过其配置的超时时间且工具使用timeout_behaviorraise_exception时抛出。UserError使用 SDK 编写代码的人在使用过程中出错时抛出通常源于错误的代码实现、无效配置或 SDK API 误用。InputGuardrailTripwireTriggered、OutputGuardrailTripwireTriggered输入护栏条件满足时抛出前者输出护栏条件满足时抛出后者。输入护栏在处理前检查入站消息输出护栏在交付前检查 Agent 的最终响应。小结Runner是 openai-agents-python 一切执行的起点但真正的工程能力体现在对运行生命周期的掌控上理解代理循环的最终输出 / 握手 / 工具调用三分支判定run_steps.py用RunConfig统一注入模型、护栏、追踪与工具错误策略run_config.py在to_input_list()/session/conversation_id/previous_response_id四种内存策略间做出正确取舍最后用error_handlers与异常体系为运行兜底。把这些能力组合起来即可在真实项目中构建健壮、可观测、可恢复的多 Agent 工作流。【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 openai-agents-python 实战：Runner 执行机制、RunConfig 调优与多轮会话管理完整指南
openai-agents-python 实战Runner 执行机制、RunConfig 调优与多轮会话管理完整指南【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python本篇技术指南以 openai-agents-python 框架的Runner执行体系为核心系统讲解如何运行 Agent、理解代理循环agent loop的底层判定逻辑、通过RunConfig精确调控单次运行的模型/追踪/工具行为、在四种内存策略中选择合适的多轮会话方案以及利用错误处理器与异常体系构建健壮的恢复路径。读完本文你将能够从会调 API进阶到能掌控运行全生命周期独立设计可生产化的多 Agent 工作流。Runner 的三种运行方式在 openai-agents-python 中Agent 的启动入口统一收敛在Runner类上。它提供三种调用方式分别覆盖异步、同步与流式三种场景Runner.run()异步执行返回RunResult。Runner.run_sync()同步方法内部只是简单包装并运行.run()。Runner.run_streamed()异步执行返回RunResultStreaming它以流式模式调用 LLM并在事件到达时即刻推送给调用方。最基本的用法如下from agents import Agent, Runner async def main(): agent Agent(nameAssistant, instructionsYou are a helpful assistant) result await Runner.run(agent, Write a haiku about recursion in programming.) print(result.final_output) # Code within the code, # Functions calling themselves, # Infinite loops dance运行结束后RunResult上携带的final_output、new_items、last_response_id、usage等字段共同构成了本次运行的完整结果视图详细字段说明可参考 结果指南。执行器生命周期代理循环The Agent Loop三种输入形态调用上述任一Runner方法时需要传入起始 Agent 和输入。从run()的签名可以看到输入可以是字符串被当作一条用户消息处理OpenAI Responses API 格式的输入项列表list[TResponseInputItem]RunState用于恢复一个被暂停的运行或恢复被cancel(modeafter_turn)中断的运行该状态还可以携带为下次恢复的模型调用准备好的待定输入。循环的判定逻辑执行器随后进入循环。在源码层面这一循环由 run_loop.py 中的run_single_turn驱动每一步产生的下一步动作由 run_steps.py 中的NextStepHandoff/NextStepFinalOutput/NextStepRunAgain/NextStepInterruption四种类型表达与下述流程一一对应用当前输入调用当前 Agent 的 LLM。LLM 产生输出后执行器进行三分支判定判定为最终输出→ 循环结束返回结果请求握手handoff→ 更新当前 Agent 与输入重新进入循环产生工具调用→ 执行这些工具调用、追加结果重新进入循环。若超过传入的max_turns则抛出MaxTurnsExceeded异常传入max_turnsNone可禁用该轮次上限。注意LLM 输出被判定为最终输出的规则是——它生成了期望类型的文本输出且没有任何工具调用。这一点在ProcessedResponse.has_tools_or_approvals_to_run()run_steps.py中有直接体现只要本轮响应中还存在 handoffs、函数调用、computer 动作、shell 调用、apply_patch 调用或待审批的 MCP 请求运行就尚未终结。另外max_turns的默认值定义在 run_config.py 的DEFAULT_MAX_TURNS 10即不显式传参时默认允许 10 轮 LLM 调用。流式执行与 Responses WebSocket 传输流式事件流式模式允许在 LLM 运行期间持续接收流式事件。流结束后RunResultStreaming会包含本次运行的完整信息包括所有新产生的输出通过.stream_events()遍历流式事件。更完整的流式消费模式参见 流式指南。Responses WebSocket 传输可选辅助器启用 OpenAI Responses WebSocket 传输后依然可以沿用普通的RunnerAPI。为复用连接SDK 推荐使用 WebSocket 会话辅助器但并非强制。需要强调的是这是基于 WebSocket 传输的 Responses API而非 Realtime API。关于具体模型对象或自定义提供者相关的传输选择规则与注意事项参见 模型文档。模式 1不使用会话辅助器可用当只需要 WebSocket 传输、不需要 SDK 代为管理共享提供者/会话时使用import asyncio from agents import Agent, Runner, set_default_openai_responses_transport async def main(): set_default_openai_responses_transport(websocket) agent Agent(nameAssistant, instructionsBe concise.) result Runner.run_streamed(agent, Summarize recursion in one sentence.) async for event in result.stream_events(): if event.type raw_response_event: continue print(event.type) asyncio.run(main())这种模式适合单次运行。如果反复调用Runner.run()/Runner.run_streamed()除非手动复用同一个RunConfig/ provider 实例否则每次运行都可能重新建立连接。模式 2使用responses_websocket_session()多轮复用推荐当需要在多次运行间共享支持 WebSocket 的 provider 与RunConfig包括继承同一run_config的嵌套 agent-as-tool 调用时使用responses_websocket_session()import asyncio from agents import Agent, responses_websocket_session async def main(): agent Agent(nameAssistant, instructionsBe concise.) async with responses_websocket_session( responses_websocket_options{ping_interval: 20.0, ping_timeout: 60.0}, ) as ws: first ws.run_streamed(agent, Say hello in one short sentence.) async for _event in first.stream_events(): pass second ws.run_streamed( agent, Now say goodbye., previous_response_idfirst.last_response_id, ) async for _event in second.stream_events(): pass asyncio.run(main())几点重要的工程约束务必在退出上下文前消费完流式结果。若 WebSocket 请求仍在进行中就退出上下文可能强制关闭共享连接。服务在每个 WebSocket 连接上一次只处理一个响应并将连接时长限制为 60 分钟。辅助器复用了连接但并不会消除这些限制。重连之后storeFalse与 ZDR 流程无法恢复未被缓存的previous_response_id应使用完整输入上下文开启新链或从本地管理的会话状态重建。完整的恢复行为参见 Responses WebSocket 传输注意事项。若长推理轮次触发 WebSocket keepalive 超时可调大ping_timeout或将其设为ping_timeoutNone以禁用心跳超时。当可靠性比 WebSocket 延迟更重要时改用 HTTP/SSE 传输。RunConfig单次运行的全局配置中枢run_config参数允许在不修改每个 Agent 定义的前提下覆盖单次运行的部分全局设置。RunConfig的完整字段定义位于 run_config.py它属于 dataclass 配置类型支持直接传字典进行强制转换。下面按类别逐一说明。模型、提供者与会话默认值model设置全局 LLM 模型无论每个 Agent 自身配置了什么model都生效。model_provider负责按模型名查找模型的提供者默认为 OpenAI。model_settings覆盖 Agent 级设置例如可全局设置temperature或top_p。session_settings运行中取回历史记录时覆盖会话级默认值如SessionSettings(limit...)。session_input_callback使用 Sessions 时自定义每次Runner运行前新用户输入与会话历史的合并方式回调支持同步或异步。护栏、握手与模型输入塑形input_guardrails、output_guardrails要在所有运行中包含的输入/输出护栏列表。handoff_input_filter应用于所有握手的全局输入过滤器若该握手本身未定义过滤器。输入过滤器允许编辑发送给新 Agent 的输入详见Handoff.input_filter的文档。nest_handoff_history选择加入的 beta 功能在调用下一个 Agent 前将可摘要的历史压缩为有序的 assistant 摘要片段同时把无损失的 message 项保留在原始位置。该功能默认关闭设为True开启保持False则原样透传原始转录。Sessions、RunState与RunResult.to_input_list()在 SDK 默认嵌套历史已持有某条消息实例时不会重复追加同时保留彼此独立但相同的消息。所有 Runner 方法 在未传入RunConfig时会自动创建一个因此快速上手示例默认保持关闭显式的Handoff.input_filter回调仍可覆盖该设置。单个握手可通过Handoff.nest_handoff_history覆盖此设置。handoff_history_mapper可选 callable在选择nest_handoff_history时接收归一化后的转录历史 握手项必须返回要转发给下一个 Agent 的确切输入项列表用于在不编写完整握手过滤器的情况下替换内置的有序摘要片段。call_model_input_filter在模型调用前一刻编辑完全准备好的模型输入instructions 与输入项的钩子例如裁剪历史或注入系统提示。reasoning_item_id_policy控制执行器将先前输出转换为下一轮模型输入时是否保留推理项的 ID。嵌套握手以 opt-in beta 形式提供。要启用有序转录压缩可传RunConfig(nest_handoff_historyTrue)或对特定握手设置handoff(..., nest_handoff_historyTrue)。内置 mapper 会在无损失 message 项周围放置生成的 assistant 摘要片段而不是把整个转录折叠成一条消息。若偏好保持原始转录默认则不设置该标志或提供按需原样转发对话的handoff_input_filter或handoff_history_mapper。若想在不用自定义 mapper 的情况下修改生成摘要片段所用的包装文本可调用set_conversation_history_wrappers恢复默认值则调用reset_conversation_history_wrappers。追踪与可观测性tracing_disabled为整个运行禁用追踪。tracing传入TracingConfig以覆盖追踪导出设置如单次运行的追踪 API Key。trace_include_sensitive_data配置追踪是否包含潜在敏感数据如 LLM 与工具调用的输入/输出。其默认值由环境变量OPENAI_AGENTS_TRACE_INCLUDE_SENSITIVE_DATA决定见 run_config.py。workflow_name、trace_id、group_id设置运行的工作流名称、追踪 ID 与追踪组 ID。建议至少设置workflow_namegroup_id为可选字段可将多次运行的追踪串接起来。trace_metadata要附加到所有追踪上的元数据。工具执行、审批与工具错误行为tool_execution配置本地工具调用的 SDK 侧执行行为例如限制同时执行的本地函数工具调用数量。tool_not_found_behavior配置执行器如何处理模型发出的函数工具调用名与当前 Agent 可用工具均不匹配的情况。默认抛出ModelBehaviorError可 opt-in 改为返回模型可见的错误输出。tool_name_collision_policy配置执行器如何处理无命名空间的函数工具与握手名称冲突。默认值warn会记录可操作警告并仅暴露当前分发的胜出者error会在调用模型前抛出UserError。对带命名空间与延迟加载工具的严格校验保持不变。tool_error_formatter自定义模型可见的工具错误消息例如审批拒绝与 opt-in 的 tool-not-found 输出。RunConfig 细节深入tool_execution本地函数工具并发与审批前护栏当需要配置本地函数工具的 SDK 侧执行行为时使用tool_execution例如限制单次运行的本地函数工具并发度from agents import Agent, RunConfig, Runner, ToolExecutionConfig agent Agent(nameAssistant, tools[...]) result await Runner.run( agent, Run the required tool calls., run_configRunConfig( tool_executionToolExecutionConfig( max_function_tool_concurrency2, pre_approval_tool_input_guardrailsTrue, ), ), )max_function_tool_concurrencyNone保持默认行为模型在一轮中发出多个函数工具调用时SDK 会启动全部本地函数工具调用设置整数值则限制同时运行的本地函数工具调用数量。源码层面run_config.py还会校验该值必须 ≥ 1否则抛出ValueError。这与提供者侧的ModelSettings.parallel_tool_calls是两回事parallel_tool_calls控制模型是否被允许在单个响应中发出多个工具调用tool_execution.max_function_tool_concurrency则控制模型发出调用后SDK 如何执行本地函数工具调用。pre_approval_tool_input_guardrailsFalse保持默认审批流程函数工具需要审批时先暂停运行工具输入护栏仅在审批通过后、执行前一刻运行。设为True可在发出待处理审批中断interruption之前先运行函数工具输入护栏。通过该审批前检查的调用在审批后仍会再次运行相同的输入护栏因此时间敏感的检查会在执行前被重新验证。tool_not_found_behavior让运行保持可恢复默认情况下若模型发出的函数工具调用与当前 Agent 可用工具均不匹配执行器抛出ModelBehaviorError。希望运行保持可恢复时设置tool_not_found_behaviorreturn_error_to_model。在该模式下SDK 会为无法解析的工具调用追加一个function_call_output并重新运行模型使模型可以选用可用工具或在不使用该工具的情况下作答from agents import Agent, RunConfig, Runner agent Agent(nameAssistant, tools[...]) result await Runner.run( agent, Handle this request with the available tools., run_configRunConfig(tool_not_found_behaviorreturn_error_to_model), )目前该选项仅适用于工具名查找失败的函数工具调用其他无效的工具载荷仍沿用既有错误行为。tool_error_formatter自定义模型可见的工具错误消息当 SDK 创建模型可见的工具错误输出时用tool_error_formatter自定义返回给模型的消息。格式化器接收包含以下字段的ToolErrorFormatterArgskind错误类别如approval_rejected或tool_not_foundtool_type工具运行时function、computer、shell、apply_patch或customtool_name工具名call_id工具调用 IDdefault_messageSDK 默认的模型可见消息run_context当前运行的上下文包装器。返回字符串以替换消息返回None则使用 SDK 默认值from agents import Agent, RunConfig, Runner, ToolErrorFormatterArgs def format_rejection(args: ToolErrorFormatterArgs[None]) - str | None: if args.kind approval_rejected: return ( fTool call {args.tool_name} was rejected by a human reviewer. Ask for confirmation or propose a safer alternative. ) if args.kind tool_not_found: return fTool {args.tool_name} is not available. Choose one of the listed tools. return None agent Agent(nameAssistant) result Runner.run_sync( agent, Please delete the production database., run_configRunConfig(tool_error_formatterformat_rejection), )reasoning_item_id_policy推理项 ID 保留策略reasoning_item_id_policy控制执行器将历史带入下一轮时例如使用RunResult.to_input_list()或基于 session 的运行如何把推理项转换为下一轮模型输入None或preserve默认保留推理项 IDomit从生成的下一轮输入中移除推理项 ID。omit主要用于规避一类 Responses API 400 错误——推理项携带id却没有必需的后继项例如Item rs_... of type reasoning was provided without its required following item.。当 SDK 从先前输出构建后续输入包括会话持久化、服务端管理的对话增量、流式/非流式后续轮次与恢复路径且保留了推理项 ID、而提供者要求该 ID 必须与其后继项配对时多轮 Agent 运行就可能触发此类问题。设置reasoning_item_id_policyomit会保留推理内容但剥离推理项id从而避免在 SDK 生成的后续输入中触发该 API 不变式。作用域说明仅改变 SDK 构建后续输入时生成/转发的推理项不重写用户提供的初始输入项应用该策略后call_model_input_filter仍可有意识地重新引入推理 ID。状态与会话管理选择内存策略将状态带入下一轮的常见方式有四种下表对比了各自的适用场景策略状态存放位置适用场景下一轮要传递的内容result.to_input_list()应用内存小型聊天循环、完全手动控制、任意提供者result.to_input_list()的列表 下一条用户消息session你的存储 SDK持久聊天状态、可恢复运行、自定义存储同一个session实例或指向同一存储的另一个实例conversation_idOpenAI Conversations API想在多个 worker 或服务间共享的具名服务端会话同一个conversation_id 仅新的用户轮次previous_response_idOpenAI Responses API不创建会话资源的轻量服务端托管续接result.last_response_id 仅新的用户轮次result.to_input_list()与session由客户端管理conversation_id与previous_response_id由 OpenAI 管理且仅在使用 OpenAI Responses API 时适用。大多数应用应为每个对话选择一种持久化策略除非刻意协调两层数据否则将客户端管理的历史与 OpenAI 托管状态混用会导致上下文重复。注意同一运行中会话持久化不能与服务端托管的会话设置conversation_id、previous_response_id或auto_previous_response_id组合使用每次调用请选择一种方式。对话/聊天线程调用任一运行方法可能驱动一个或多个 Agent进而发生一次或多次 LLM 调用但在聊天对话中它只代表一个逻辑轮次。例如用户轮次用户输入文本执行器运行第一个 Agent 调用 LLM、运行工具、握手到第二个 Agent第二个 Agent 再运行更多工具并产生输出。Agent 运行结束时你可以选择向用户展示什么——例如展示 Agent 生成的全部新项或只展示最终输出。无论哪种方式用户随后可能提出追问此时再次调用运行方法即可。手动对话管理可以使用RunResultBase.to_input_list()获取下一轮输入手动管理对话历史from agents import Agent, Runner, trace async def main(): agent Agent(nameAssistant, instructionsReply very concisely.) thread_id thread_123 # Example thread ID with trace(workflow_nameConversation, group_idthread_id): # First turn result await Runner.run(agent, What city is the Golden Gate Bridge in?) print(result.final_output) # San Francisco # Second turn new_input result.to_input_list() [{role: user, content: What state is it in?}] result await Runner.run(agent, new_input) print(result.final_output) # California使用 Sessions 自动管理对话更简单的方式是使用 Sessions无需手动调用.to_input_list()即可自动处理对话历史from agents import Agent, Runner, SQLiteSession, trace async def main(): agent Agent(nameAssistant, instructionsReply very concisely.) # Create session instance session SQLiteSession(conversation_123) thread_id thread_123 # Example thread ID with trace(workflow_nameConversation, group_idthread_id): # First turn result await Runner.run(agent, What city is the Golden Gate Bridge in?, sessionsession) print(result.final_output) # San Francisco # Second turn - agent automatically remembers previous context result await Runner.run(agent, What state is it in?, sessionsession) print(result.final_output) # CaliforniaSessions 会自动完成三件事每次运行前取回对话历史每次运行后保存新消息对不同 session ID 维护相互独立的会话。SQLiteSession的实现位于 sqlite_session.py支持自定义db_path、sessions_table与messages_table等参数。更多细节参见 Sessions 文档。服务端托管的对话也可以让 OpenAI 的对话状态功能在服务端管理对话状态而不是用to_input_list()或Sessions在本地处理。这样可以保留对话历史无需手动重发所有历史消息。使用下面任一种服务端托管方式时每次请求只传新一轮的输入并复用保存的 ID。OpenAI 提供两种跨轮次跟踪状态的方式。方式 1使用conversation_id先用 OpenAI Conversations API 创建对话然后在后续每次调用中复用其 IDfrom agents import Agent, Runner from openai import AsyncOpenAI client AsyncOpenAI() async def main(): agent Agent(nameAssistant, instructionsReply very concisely.) # Create a server-managed conversation conversation await client.conversations.create() conv_id conversation.id while True: user_input input(You: ) result await Runner.run(agent, user_input, conversation_idconv_id) print(fAssistant: {result.final_output})方式 2使用previous_response_id响应链式续接另一种选项是响应链式续接response chaining每一轮显式链接到上一轮响应的 IDfrom agents import Agent, Runner async def main(): agent Agent(nameAssistant, instructionsReply very concisely.) previous_response_id None while True: user_input input(You: ) # Setting auto_previous_response_idTrue enables response chaining automatically # for the first turn, even when theres no actual previous response ID yet. result await Runner.run( agent, user_input, previous_response_idprevious_response_id, auto_previous_response_idTrue, ) previous_response_id result.last_response_id print(fAssistant: {result.final_output})如果运行因审批而暂停并从RunState恢复SDK 会保留保存的conversation_id/previous_response_id/auto_previous_response_id设置使恢复后的轮次继续处于同一个服务端托管的对话中。conversation_id与previous_response_id互斥需要可在系统间共享的具名对话资源时用conversation_id需要轮次之间最轻量的 Responses API 续接原语时用previous_response_id。注意SDK 会以退避backoff方式自动重试conversation_locked错误。在服务端托管的对话运行中重试前会回卷内部对话追踪器输入使相同准备项能被干净地重新发送。在本地基于 session 的运行中不能与conversation_id、previous_response_id或auto_previous_response_id组合SDK 也会对最近持久化的输入项做尽力而为的回滚以减少重试后产生的重复历史条目。即使未配置ModelSettings.retry该兼容性重试也会发生。关于模型请求更广泛的 opt-in 重试行为参见 Runner 托管重试。钩子与自定义call_model_input_filter使用call_model_input_filter可在模型调用前一刻编辑模型输入。钩子接收当前 Agent、上下文以及合并后的输入项存在会话历史时包含在内并返回新的ModelInputData。返回值必须是ModelInputData对象其input字段为必填且必须是输入项列表返回其他形状会抛出UserErrorfrom agents import Agent, Runner, RunConfig from agents.run import CallModelData, ModelInputData def drop_old_messages(data: CallModelData[None]) - ModelInputData: # Keep only the last 5 items and preserve existing instructions. trimmed data.model_data.input[-5:] return ModelInputData(inputtrimmed, instructionsdata.model_data.instructions) agent Agent(nameAssistant, instructionsAnswer concisely.) result Runner.run_sync( agent, Explain quines, run_configRunConfig(call_model_input_filterdrop_old_messages), )执行器会把准备好的输入列表的副本传给钩子因此你可以在不原地修改调用方原始列表的情况下裁剪、替换或重排其中的项。若正在使用 sessioncall_model_input_filter会在会话历史已加载并与当前轮次合并之后运行。若想自定义更早的合并步骤本身请使用session_input_callback。若正通过conversation_id、previous_response_id或auto_previous_response_id使用 OpenAI 服务端托管对话状态钩子会在为下一次 Responses API 调用准备的载荷上运行。该载荷可能只体现新轮次的增量而非先前历史的完整回放只有你返回的项会被标记为在该服务端托管续接中发送。通过run_config按运行设置钩子可用于脱敏敏感数据、裁剪过长历史或注入额外的系统指引。错误与恢复错误处理器Error Handlers所有Runner入口都接受error_handlers——一个以错误种类为键的字典。支持的键为max_turns、model_refusal与invalid_final_output。当希望返回受控的最终输出而不是以相应错误终止运行时使用它们。max_turns示例——超过轮次上限时返回友好的兜底文案from agents import ( Agent, RunErrorHandlerInput, RunErrorHandlerResult, Runner, ) agent Agent(nameAssistant, instructionsBe concise.) def on_max_turns(_data: RunErrorHandlerInput[None]) - RunErrorHandlerResult: return RunErrorHandlerResult( final_outputI couldnt finish within the turn limit. Please narrow the request., include_in_historyFalse, ) result Runner.run_sync( agent, Analyze this long transcript, max_turns3, error_handlers{max_turns: on_max_turns}, ) print(result.final_output)invalid_final_output用于模型消息无法通过 Agent 结构化output_type校验、或模型未返回结构化最终消息的场景。处理器可返回应用特定的兜底值SDK 会用同一个output_type对其进行校验。它不会重试模型调用也不会重放任何工具副作用。返回None表示拒绝恢复若无兜底值非空的校验失败会继续抛出ModelBehaviorError空的结构化响应则保持既有的下一轮行为from pydantic import BaseModel from agents import Agent, ModelBehaviorError, RunErrorHandlerInput, Runner class Recipe(BaseModel): ingredients: list[str] recovered_from_invalid_output: bool False def on_invalid_final_output(data: RunErrorHandlerInput[None]) - Recipe: assert isinstance(data.error, ModelBehaviorError) return Recipe(ingredients[], recovered_from_invalid_outputTrue) agent Agent( nameRecipe assistant, instructionsReturn a structured recipe., output_typeRecipe, ) result Runner.run_sync( agent, Plan tonights dinner., error_handlers{invalid_final_output: on_invalid_final_output}, ) print(result.final_output)RunErrorHandlerResult.include_in_history默认为True。对 max-turns 处理器而言这会把合成的兜底输出追加到对话历史并持久化到已配置的 session若希望把兜底值返回给调用方但不加入结果历史或 session 存储请设置include_in_historyFalse。model_refusal用于让模型拒绝refusal产生应用特定的兜底值而不是以ModelRefusalError终止运行from pydantic import BaseModel from agents import Agent, ModelRefusalError, RunErrorHandlerInput, Runner class Recipe(BaseModel): ingredients: list[str] refusal_reason: str | None None def on_model_refusal(data: RunErrorHandlerInput[None]) - Recipe: assert isinstance(data.error, ModelRefusalError) return Recipe(ingredients[], refusal_reasondata.error.refusal) agent Agent( nameRecipe assistant, instructionsReturn a structured recipe., output_typeRecipe, ) result Runner.run_sync( agent, Make me something unsafe., error_handlers{model_refusal: on_model_refusal}, ) print(result.final_output)持久化执行集成与人类在环HITL工具审批的暂停/恢复模式请先阅读专门的 人类在环指南。以下集成适用于运行可能跨越长时间等待、重试或进程重启的持久化编排场景DaprAgents SDK 的 DaprDiagrid集成可运行具备故障自动恢复能力、并支持人类在环工作流的持久化长时运行 Agent。Dapr 是供应商中立的 CNCF 工作流编排器。仓库中提供了对应的容器级集成测试test_dapr_redis.py与示例dapr_session_example.py。TemporalAgents SDK 的 Temporal 集成可运行包括人类在环任务在内的持久化长时工作流。RestateAgents SDK 的 Restate 集成可构建轻量级持久化 Agent覆盖人工审批、握手与会话管理。该集成以 Restate 的单二进制运行时为依赖支持以进程/容器或 serverless 函数方式运行 Agent。DBOSAgents SDK 的 DBOS 集成可运行在故障与重启后仍保留进度的可靠 Agent支持长时运行 Agent、人类在环工作流与握手同步与异步方法均可仅需 SQLite 或 Postgres 数据库。异常体系一览SDK 在特定情况下会抛出异常完整清单位于agents.exceptions。概览如下AgentsExceptionSDK 抛出的所有异常的基类是其他具体异常的通用父类型。MaxTurnsExceededAgent 运行超过传给Runner.run、Runner.run_sync或Runner.run_streamed方法的max_turns上限时抛出表示 Agent 未能在指定数量的代理循环轮次LLM 调用内完成任务。设置max_turnsNone可禁用该上限。ModelTimeoutError单次模型调用尝试超过ModelSettings.timeout时抛出。作用范围与重试行为参见 模型调用超时。ModelBehaviorError底层模型LLM产生意外或无效输出时抛出可能包括畸形 JSON模型在工具调用或直接输出中给出畸形 JSON 结构尤其是定义了特定output_type时意外工具相关失败模型未按预期方式使用工具失败或不完整的非流式 Responses 调用OpenAIResponsesModel与AnyLLMModel的 Responses 路径在返回的响应终止状态为failed或incomplete时抛出异常会标识终止状态并包含响应中可用的错误或未完成详情。ToolTimeoutError函数工具调用超过其配置的超时时间且工具使用timeout_behaviorraise_exception时抛出。UserError使用 SDK 编写代码的人在使用过程中出错时抛出通常源于错误的代码实现、无效配置或 SDK API 误用。InputGuardrailTripwireTriggered、OutputGuardrailTripwireTriggered输入护栏条件满足时抛出前者输出护栏条件满足时抛出后者。输入护栏在处理前检查入站消息输出护栏在交付前检查 Agent 的最终响应。小结Runner是 openai-agents-python 一切执行的起点但真正的工程能力体现在对运行生命周期的掌控上理解代理循环的最终输出 / 握手 / 工具调用三分支判定run_steps.py用RunConfig统一注入模型、护栏、追踪与工具错误策略run_config.py在to_input_list()/session/conversation_id/previous_response_id四种内存策略间做出正确取舍最后用error_handlers与异常体系为运行兜底。把这些能力组合起来即可在真实项目中构建健壮、可观测、可恢复的多 Agent 工作流。【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考