企业官网建设流程全解析

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类上。它提供三种调用方式，分别覆盖异步、同步与流式三种场景：

1. Runner.run()：异步执行，返回RunResult。
2. Runner.run_sync()：同步方法，内部只是简单包装并运行.run()。
3. Runner.run_streamed()：异步执行，返回RunResultStreaming；它以流式模式调用 LLM，并在事件到达时即刻推送给调用方。

最基本的用法如下：

from agents import Agent, Runner async def main(): agent = Agent(name="Assistant", instructions="You 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 loop's dance

运行结束后，RunResult上携带的final_output、new_items、last_response_id、usage等字段共同构成了本次运行的完整结果视图，详细字段说明可参考 结果指南。

执行器生命周期：代理循环（The Agent Loop）

三种输入形态

调用上述任一Runner方法时，需要传入起始 Agent 和输入。从run()的签名可以看到，输入可以是：

• 字符串：被当作一条用户消息处理；
• OpenAI Responses API 格式的输入项列表（list[TResponseInputItem]）；
• RunState：用于恢复一个被暂停的运行，或恢复被cancel(mode="after_turn")中断的运行；该状态还可以携带为下次恢复的模型调用准备好的待定输入。

循环的判定逻辑

执行器随后进入循环。在源码层面，这一循环由 run_loop.py 中的run_single_turn驱动，每一步产生的"下一步动作"由 run_steps.py 中的NextStepHandoff/NextStepFinalOutput/NextStepRunAgain/NextStepInterruption四种类型表达，与下述流程一一对应：

1. 用当前输入调用当前 Agent 的 LLM。
2. LLM 产生输出后，执行器进行三分支判定：
   • 判定为最终输出→ 循环结束，返回结果；
   • 请求握手（handoff）→ 更新当前 Agent 与输入，重新进入循环；
   • 产生工具调用→ 执行这些工具调用、追加结果，重新进入循环。
3. 若超过传入的max_turns，则抛出MaxTurnsExceeded异常；传入max_turns=None可禁用该轮次上限。

注意：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(name="Assistant", instructions="Be 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(name="Assistant", instructions="Be 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_id=first.last_response_id, ) async for _event in second.stream_events(): pass asyncio.run(main())

几点重要的工程约束：

• 务必在退出上下文前消费完流式结果。若 WebSocket 请求仍在进行中就退出上下文，可能强制关闭共享连接。
• 服务在每个 WebSocket 连接上一次只处理一个响应，并将连接时长限制为 60 分钟。辅助器复用了连接，但并不会消除这些限制。重连之后，store=False与 ZDR 流程无法恢复未被缓存的previous_response_id；应使用完整输入上下文开启新链，或从本地管理的会话状态重建。完整的恢复行为参见 Responses WebSocket 传输注意事项。
• 若长推理轮次触发 WebSocket keepalive 超时，可调大ping_timeout，或将其设为ping_timeout=None以禁用心跳超时。当可靠性比 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_history=True)，或对特定握手设置handoff(..., nest_handoff_history=True)。内置 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_name；group_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(name="Assistant", tools=[...]) result = await Runner.run( agent, "Run the required tool calls.", run_config=RunConfig( tool_execution=ToolExecutionConfig( max_function_tool_concurrency=2, pre_approval_tool_input_guardrails=True, ), ), )

• max_function_tool_concurrency=None保持默认行为：模型在一轮中发出多个函数工具调用时，SDK 会启动全部本地函数工具调用；设置整数值则限制同时运行的本地函数工具调用数量。源码层面（run_config.py）还会校验该值必须 ≥ 1，否则抛出ValueError。
• 这与提供者侧的ModelSettings.parallel_tool_calls是两回事：parallel_tool_calls控制模型是否被允许在单个响应中发出多个工具调用；tool_execution.max_function_tool_concurrency则控制模型发出调用后，SDK 如何执行本地函数工具调用。
• pre_approval_tool_input_guardrails=False保持默认审批流程：函数工具需要审批时，先暂停运行，工具输入护栏仅在审批通过后、执行前一刻运行。设为True可在发出待处理审批中断（interruption）之前先运行函数工具输入护栏。通过该审批前检查的调用，在审批后仍会再次运行相同的输入护栏，因此时间敏感的检查会在执行前被重新验证。

tool_not_found_behavior：让运行保持可恢复

默认情况下，若模型发出的函数工具调用与当前 Agent 可用工具均不匹配，执行器抛出ModelBehaviorError。希望运行保持可恢复时，设置tool_not_found_behavior="return_error_to_model"。在该模式下，SDK 会为无法解析的工具调用追加一个function_call_output并重新运行模型，使模型可以选用可用工具，或在不使用该工具的情况下作答：

from agents import Agent, RunConfig, Runner agent = Agent(name="Assistant", tools=[...]) result = await Runner.run( agent, "Handle this request with the available tools.", run_config=RunConfig(tool_not_found_behavior="return_error_to_model"), )

目前该选项仅适用于工具名查找失败的函数工具调用；其他无效的工具载荷仍沿用既有错误行为。

tool_error_formatter：自定义模型可见的工具错误消息

当 SDK 创建模型可见的工具错误输出时，用tool_error_formatter自定义返回给模型的消息。格式化器接收包含以下字段的ToolErrorFormatterArgs：

• kind：错误类别，如"approval_rejected"或"tool_not_found"；
• tool_type：工具运行时（"function"、"computer"、"shell"、"apply_patch"或"custom"）；
• tool_name：工具名；
• call_id：工具调用 ID；
• default_message：SDK 默认的模型可见消息；
• 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 ( f"Tool 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 f"Tool '{args.tool_name}' is not available. Choose one of the listed tools." return None agent = Agent(name="Assistant") result = Runner.run_sync( agent, "Please delete the production database.", run_config=RunConfig(tool_error_formatter=format_rejection), )

reasoning_item_id_policy：推理项 ID 保留策略

reasoning_item_id_policy控制执行器将历史带入下一轮时（例如使用RunResult.to_input_list()或基于 session 的运行）如何把推理项转换为下一轮模型输入：

• None或"preserve"（默认）：保留推理项 ID；
• "omit"：从生成的下一轮输入中移除推理项 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_policy="omit"会保留推理内容但剥离推理项id，从而避免在 SDK 生成的后续输入中触发该 API 不变式。

作用域说明：

• 仅改变 SDK 构建后续输入时生成/转发的推理项；
• 不重写用户提供的初始输入项；
• 应用该策略后，call_model_input_filter仍可有意识地重新引入推理 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 调用），但在聊天对话中它只代表一个逻辑轮次。例如：

1. 用户轮次：用户输入文本；
2. 执行器运行：第一个 Agent 调用 LLM、运行工具、握手到第二个 Agent，第二个 Agent 再运行更多工具并产生输出。

Agent 运行结束时，你可以选择向用户展示什么——例如展示 Agent 生成的全部新项，或只展示最终输出。无论哪种方式，用户随后可能提出追问，此时再次调用运行方法即可。

手动对话管理

可以使用RunResultBase.to_input_list()获取下一轮输入，手动管理对话历史：

from agents import Agent, Runner, trace async def main(): agent = Agent(name="Assistant", instructions="Reply very concisely.") thread_id = "thread_123" # Example thread ID with trace(workflow_name="Conversation", group_id=thread_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(name="Assistant", instructions="Reply very concisely.") # Create session instance session = SQLiteSession("conversation_123") thread_id = "thread_123" # Example thread ID with trace(workflow_name="Conversation", group_id=thread_id): # First turn result = await Runner.run(agent, "What city is the Golden Gate Bridge in?", session=session) print(result.final_output) # San Francisco # Second turn - agent automatically remembers previous context result = await Runner.run(agent, "What state is it in?", session=session) print(result.final_output) # California

Sessions 会自动完成三件事：

• 每次运行前取回对话历史；
• 每次运行后保存新消息；
• 对不同 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 创建对话，然后在后续每次调用中复用其 ID：

from agents import Agent, Runner from openai import AsyncOpenAI client = AsyncOpenAI() async def main(): agent = Agent(name="Assistant", instructions="Reply 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_id=conv_id) print(f"Assistant: {result.final_output}")

方式 2：使用previous_response_id（响应链式续接）

另一种选项是响应链式续接（response chaining），每一轮显式链接到上一轮响应的 ID：

from agents import Agent, Runner async def main(): agent = Agent(name="Assistant", instructions="Reply very concisely.") previous_response_id = None while True: user_input = input("You: ") # Setting auto_previous_response_id=True enables response chaining automatically # for the first turn, even when there's no actual previous response ID yet. result = await Runner.run( agent, user_input, previous_response_id=previous_response_id, auto_previous_response_id=True, ) previous_response_id = result.last_response_id print(f"Assistant: {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字段为必填且必须是输入项列表；返回其他形状会抛出UserError：

from 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(input=trimmed, instructions=data.model_data.instructions) agent = Agent(name="Assistant", instructions="Answer concisely.") result = Runner.run_sync( agent, "Explain quines", run_config=RunConfig(call_model_input_filter=drop_old_messages), )

执行器会把准备好的输入列表的副本传给钩子，因此你可以在不原地修改调用方原始列表的情况下裁剪、替换或重排其中的项。

• 若正在使用 session，call_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(name="Assistant", instructions="Be concise.") def on_max_turns(_data: RunErrorHandlerInput[None]) -> RunErrorHandlerResult: return RunErrorHandlerResult( final_output="I couldn't finish within the turn limit. Please narrow the request.", include_in_history=False, ) result = Runner.run_sync( agent, "Analyze this long transcript", max_turns=3, 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_output=True) agent = Agent( name="Recipe assistant", instructions="Return a structured recipe.", output_type=Recipe, ) result = Runner.run_sync( agent, "Plan tonight's 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_history=False。

"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_reason=data.error.refusal) agent = Agent( name="Recipe assistant", instructions="Return a structured recipe.", output_type=Recipe, ) result = Runner.run_sync( agent, "Make me something unsafe.", error_handlers={"model_refusal": on_model_refusal}, ) print(result.final_output)

持久化执行集成与人类在环（HITL）

工具审批的暂停/恢复模式请先阅读专门的 人类在环指南。以下集成适用于运行可能跨越长时间等待、重试或进程重启的持久化编排场景：

• Dapr：Agents SDK 的 Dapr（Diagrid）集成可运行具备故障自动恢复能力、并支持人类在环工作流的持久化长时运行 Agent。Dapr 是供应商中立的 CNCF 工作流编排器。仓库中提供了对应的容器级集成测试（test_dapr_redis.py）与示例（dapr_session_example.py）。
• Temporal：Agents SDK 的 Temporal 集成可运行包括人类在环任务在内的持久化长时工作流。
• Restate：Agents SDK 的 Restate 集成可构建轻量级持久化 Agent，覆盖人工审批、握手与会话管理。该集成以 Restate 的单二进制运行时为依赖，支持以进程/容器或 serverless 函数方式运行 Agent。
• DBOS：Agents SDK 的 DBOS 集成可运行在故障与重启后仍保留进度的可靠 Agent，支持长时运行 Agent、人类在环工作流与握手，同步与异步方法均可，仅需 SQLite 或 Postgres 数据库。

异常体系一览

SDK 在特定情况下会抛出异常，完整清单位于agents.exceptions。概览如下：

• AgentsException：SDK 抛出的所有异常的基类，是其他具体异常的通用父类型。
• MaxTurnsExceeded：Agent 运行超过传给Runner.run、Runner.run_sync或Runner.run_streamed方法的max_turns上限时抛出，表示 Agent 未能在指定数量的代理循环轮次（LLM 调用）内完成任务。设置max_turns=None可禁用该上限。
• ModelTimeoutError：单次模型调用尝试超过ModelSettings.timeout时抛出。作用范围与重试行为参见 模型调用超时。
• ModelBehaviorError：底层模型（LLM）产生意外或无效输出时抛出，可能包括：
  • 畸形 JSON：模型在工具调用或直接输出中给出畸形 JSON 结构，尤其是定义了特定output_type时；
  • 意外工具相关失败：模型未按预期方式使用工具；
  • 失败或不完整的非流式 Responses 调用：OpenAIResponsesModel与AnyLLMModel的 Responses 路径在返回的响应终止状态为failed或incomplete时抛出，异常会标识终止状态并包含响应中可用的错误或未完成详情。
• ToolTimeoutError：函数工具调用超过其配置的超时时间，且工具使用timeout_behavior="raise_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