openai-agents-python 运行指南:掌握 Runner 执行、RunConfig 配置与多轮会话状态管理
【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python
本文基于 docs/ja/running_agents.md 编写,围绕 openai-agents-python 框架的
Runner执行体系展开,系统讲解三种执行入口、Agent 循环的工作原理、RunConfig全部配置类别、四类多轮记忆策略、错误处理器以及异常体系。读者完成后将能够熟练编排单 Agent 与多 Agent 工作流,按需启用流式输出、Responses WebSocket 传输、Sessions 持久化与会话级恢复,并正确处置运行中的各类异常。
一、三种执行入口:run / run_sync / run_streamed
在 openai-agents-python 中,Agent 的一切执行都经由Runner类完成。它提供了三个等价的入口方法,分别面向异步、同步与流式场景:
| 方法 | 执行方式 | 返回值 | 适用场景 |
|---|---|---|---|
Runner.run() | 异步协程 | RunResult | 常规 async 应用、FastAPI、Jupyter 等已有事件循环的环境 |
Runner.run_sync() | 同步阻塞 | RunResult | 脚本、简单命令行工具;内部只是对.run()的封装 |
Runner.run_streamed() | 异步 + 流式 | RunResultStreaming | 需要边生成边消费事件(token 级体验、进度反馈) |
最小可运行示例:
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 asyncio.run(main())关于三个入口方法的完整签名(context、max_turns、hooks、run_config、error_handlers、previous_response_id、auto_previous_response_id、conversation_id、session等参数),可直接参考 src/agents/run.py 中的Runner类定义。
需要注意run_sync()的约束:由于它内部直接调用.run(),因此不能在已存在事件循环的上下文中使用(例如 async 函数内部、Jupyter Notebook、FastAPI 请求处理中),这些场景请改用run()。运行结果对象的具体字段(final_output、last_agent、to_input_list()、last_response_id等)可参考 执行结果指南。
二、Runner 的生命周期与配置
2.1 Agent 循环(The Agent Loop)
调用上述任一入口时,需要传入起始 Agent(starting agent)与输入。输入支持三种形式:
- 字符串:作为一条用户消息;
- OpenAI Responses API 格式的输入条目列表(
list[TResponseInputItem]); RunState:用于恢复一次被暂停的执行,或恢复通过cancel(mode="after_turn")停止的执行。状态中可以携带为下次恢复准备好的输入条目(详见 results.md 的"恢复前追加输入")。
随后 Runner 执行如下循环:
- 使用当前输入调用当前 Agent 的 LLM;
- LLM 产生输出后分三种情况处理:
- 判定为最终输出:终止循环并返回运行结果;
- 产生handoff(转交):更新当前 Agent 与输入,重新进入循环;
- 产生工具调用:执行这些工具调用、把结果追加回输入,再重新进入循环;
- 若超过传入的
max_turns,抛出MaxTurnsExceeded异常;传max_turns=None可禁用该限制。
注意:LLM 输出被判定为"最终输出"的条件是——生成了目标类型的文本输出且没有工具调用。另外从源码看,只有起始 Agent 的输入护栏(input guardrails)会被执行(见 run.py 的 docstring)。
max_turns的默认值为DEFAULT_MAX_TURNS = 10,定义在 src/agents/run_config.py。所谓"一个 turn"定义为一次 LLM 调用(含该轮可能发生的所有工具调用)。
2.2 流式执行(Streaming)
流式模式让你在 LLM 执行过程中实时收到语义事件;流结束后,RunResultStreaming中仍保存包含全部新输出的完整运行信息。通过.stream_events()消费事件,事件类型复用 OpenAI Responses API 的语义事件。详见 流式指南。
from agents import Agent, Runner async def main(): 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())2.3 Responses WebSocket 传输(可选辅助)
启用 OpenAI Responses 的 WebSocket 传输后,普通RunnerAPI 依然照常可用;responses_websocket_session()会话辅助器只是推荐用于连接复用的可选增强,并非强制。注意这是经 WebSocket 传输的 Responses API,不是Realtime API(Realtime 见 docs/realtime/guide.md)。传输选择规则及具体模型对象 / 自定义 provider 的注意事项见 模型文档。
模式 1:不使用会话辅助器(可用但每次可能重连)
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 实例,每次执行都可能重新建立连接。该全局设置函数定义于 src/agents/_config.py 与 src/agents/init.py,仅接受"http"与"websocket"两个取值。
模式 2:使用responses_websocket_session()(多轮复用推荐)
当需要在多次执行间共享 WebSocket 能力的 provider 与RunConfig时,使用responses_websocket_session()。它也会作用于继承了同一run_config的嵌套调用(如把 Agent 当作工具使用)。
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 请求处理中退出 context,可能强制关闭共享连接;
- 每个 WebSocket 连接同时只处理一个响应,且连接时长限制为 60 分钟;辅助器只负责复用连接,并不能解除这些限制;
- 重连后,在
store=False与 ZDR(zero-data-retention)流程中,未缓存的previous_response_id无法恢复;此时应携带完整输入上下文开启新链,或从本地管理的会话状态重建; - 长推理轮次若触发 WebSocket keepalive 超时,可增大
ping_timeout,或设ping_timeout=None禁用心跳超时;当可靠性比延迟更重要时,改用 HTTP/SSE 传输。
底层实现上,responses_websocket_session()构造了一个使用openai_use_responses_websocket=True的MultiProvider,并把共享RunConfig注入其ResponsesWebSocketSession包装对象(见 src/agents/responses_websocket_session.py);keepalive 参数ping_interval/ping_timeout由OpenAIResponsesWebSocketOptions定义并透传给底层websockets库(见 src/agents/models/openai_responses.py)。
2.4 运行配置(RunConfig)
run_config参数用于配置一次 Agent 运行的全局设置:在不修改任何 Agent 定义的前提下,仅覆盖单次执行的行为。其完整字段定义见 src/agents/run_config.py。按类别归纳如下。
模型、Provider 与会话默认值
model:设置全局 LLM 模型,会覆盖每个 Agent 自带的model(传入的model_provider必须能解析该模型名);model_provider:解析模型名所用的 provider,默认 OpenAI;model_settings:覆盖 Agent 级设置,例如设置全局temperature或top_p(接受ModelSettings实例或字段字典);session_settings:覆盖执行时取历史所用的会话级默认值,例如SessionSettings(limit=...);session_input_callback:使用 Sessions 时,定制每次Runner执行前新用户输入与会话历史的合并方式,支持同步或异步回调。
护栏、Handoff 与模型输入整形
input_guardrails、output_guardrails:应用到所有执行的输入/输出护栏列表;handoff_input_filter:当某 handoff 未自带输入过滤器时,作为全局过滤器应用,可编辑传给新 Agent 的输入(参考Handoff.input_filter);nest_handoff_history:opt-in 的 beta 功能。开启后会把可总结的历史压缩为有序的 assistant 摘要段,同时把无损消息条目保留在原始位置;默认关闭。设True开启,保持False则原样传递原始 transcript。Sessions、RunState、RunResult.to_input_list()不会对已保留的同一消息重复追加,但会保留独立存在的相同消息;单个 handoff 可用Handoff.nest_handoff_history覆盖该设置;handoff_history_mapper:在nest_handoff_history开启时生效的可选 callable,接收规范化 transcript(历史 + handoff 条目),返回应传给下一个 Agent 的精确输入条目列表,用于替换内置的有序摘要段;call_model_input_filter:模型调用前一刻编辑完整模型输入(instructions 与输入条目)的钩子,常用于裁剪历史或插入系统提示;reasoning_item_id_policy:控制 Runner 把上一轮输出转换为下一轮模型输入时,推理条目 ID 是保留还是省略。
追踪与可观测性
tracing_disabled:禁用整个执行的追踪;tracing:传入TracingConfig覆盖导出设置(如每次执行的追踪 API key);trace_include_sensitive_data:是否把 LLM 与工具调用的输入/输出等敏感数据写入追踪(默认值由环境变量OPENAI_AGENTS_TRACE_INCLUDE_SENSITIVE_DATA控制,见 src/agents/run_config.py);workflow_name、trace_id、group_id:设置执行的追踪工作流名、trace ID 与分组 ID(建议至少设置workflow_name;group_id用于把多次执行的 trace 关联到同一会话);trace_metadata:附加到所有 trace 的元数据字典。
工具执行、审批与错误行为
tool_execution:配置 SDK 侧本地工具调用的执行行为,例如限制并发数;tool_not_found_behavior:当模型生成的函数工具调用名在当前 Agent 中找不到时,Runner 的处理策略,默认抛出ModelBehaviorError,也可改为向模型返回可感知的错误输出;tool_name_collision_policy:无命名空间的函数工具名与 handoff 名冲突时的策略。默认"warn"(记录可操作的告警,只暴露当前选中的分发目标);"error"则在调用模型前抛出UserError。带命名空间与延迟加载工具的严格校验不受影响;tool_error_formatter:定制返回给模型的可感知工具错误消息(如审批拒绝、opt-in 的未找到工具输出)。
另外,开启RunConfig(nest_handoff_history=True)后,如需修改内置摘要段的包装文本而不手写自定义 mapper,可调用set_conversation_history_wrappers(恢复默认用reset_conversation_history_wrappers)。
2.5 RunConfig 关键配置项详解
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 会并发启动全部本地调用;设置整数可限制同时执行的本地函数工具调用数(最小为 1,传 0 或负数会在__post_init__中抛ValueError,见 src/agents/run_config.py)。- 它与 provider 侧的
ModelSettings.parallel_tool_calls是两个不同维度:parallel_tool_calls控制模型能否在一个响应中生成多个工具调用;max_function_tool_concurrency控制模型生成调用之后、SDK 如何执行这些本地函数工具调用。 pre_approval_tool_input_guardrails=False保持默认审批流程:函数工具需要审批时,先暂停执行,工具输入护栏在审批通过后、执行前才运行;设True则在生成待审批中断前先运行工具输入护栏。注意即便通过审批前检查,审批后同一输入护栏仍会再执行一次,因此时间敏感检查会在执行前被重新验证。
tool_not_found_behavior:模型调用不存在的工具
默认情况下,模型生成的函数工具调用名与当前 Agent 的任何函数工具都不匹配时,Runner 抛出ModelBehaviorError。若希望运行保持可恢复状态,设"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"), )当前该选项仅适用于工具名查找失败的函数工具调用;其他无效工具载荷仍沿用既有错误行为。类型别名定义见 src/agents/run_config.py。
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 的保留与省略
该策略控制 Runner 在继承历史时(例如使用RunResult.to_input_list()或基于 session 的执行)如何把推理条目转换为下一轮的模型输入:
None或"preserve"(默认):保留推理条目 ID;"omit":从生成的下一轮输入中移除推理条目 ID。
"omit"主要作为对一类 Responses API 400 错误的 opt-in 缓解:当推理条目携带id发送、却缺少其必需的后继条目(例如报错Item 'rs_...' of type 'reasoning' was provided without its required following item.)时触发。这在 SDK 从上一轮输出构造后续输入的多轮 Agent 执行中可能发生,涉及 session 持久化、服务端管理的会话差分、流式与非流式的后续轮次以及恢复路径。
设"omit"后,推理内容仍然保留,只是去掉推理条目的id,从而避免违反该 API 不变量。适用范围注意:
- 仅修改 SDK 构造后续输入时生成或转发的推理条目;
- 不会重写用户显式提供的初始输入条目;
call_model_input_filter在该策略生效后仍可有意地重新引入推理 ID。
类型定义ReasoningItemIdPolicy = Literal["preserve", "omit"]见 src/agents/run_config.py。
三、状态与会话管理
3.1 选择记忆策略:四种方案对比
把状态传递到下一轮,常见的有四种方式:
| 策略 | 状态存放位置 | 最佳用途 | 下一轮需要传入的内容 |
|---|---|---|---|
result.to_input_list() | 应用内存 | 小型聊天循环、完全手动控制、任意 provider | to_input_list()返回的列表 + 下一条用户消息 |
session | 存储层 + SDK | 持久化聊天状态、可恢复执行、自定义存储 | 同一个session实例,或引用同一存储的另一个实例 |
conversation_id | OpenAI Conversations API | 在 worker / 服务间共享的具名服务端会话 | 同一个conversation_id+ 新用户回合 |
previous_response_id | OpenAI Responses API | 不创建会话资源的轻量服务端续接 | result.last_response_id+ 新用户回合 |
to_input_list()与session属于客户端管理;conversation_id与previous_response_id属于OpenAI 管理,仅在走 OpenAI Responses API 时适用。大多数应用请为每个会话选择一种持久化策略;除非有意协调两层,否则混用客户端历史与服务端状态可能造成上下文重复。
注意:同一次执行中不能同时使用 session 持久化与服务端会话设置(
conversation_id、previous_response_id或auto_previous_response_id),每次调用只能选择其中一种方式。
3.2 会话与聊天线程
一次执行方法调用可能驱动一个或多个 Agent、引发一次或多次 LLM 调用,但在聊天会话层面它只代表一个逻辑回合,例如:
- 用户回合:用户输入文本;
- Runner 执行:首个 Agent 调用 LLM、执行工具,然后 handoff 给第二个 Agent;第二个 Agent 继续执行工具并生成输出。
执行结束后,你可以自行选择展示给用户的内容:展示 Agent 生成的全部新条目,或仅展示最终输出。之后用户可能继续提问,此时再次调用执行方法即可。
手动管理会话: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) # Californiato_input_list()定义于RunResultBase,支持mode参数(默认preserve_all)控制返回条目的保真程度。
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) # CaliforniaSessions 会自动完成三件事:
- 每次执行前取回会话历史;
- 每次执行后保存新消息;
- 按 session ID 维护相互独立的会话。
服务端管理的会话(Server-Managed Conversations)
除了本地处理,还可以用 OpenAI 的会话状态能力在服务端维护会话,从而无需手动重发全部历史消息。两种服务端方式都在每个请求中只传新回合输入、复用保存的 ID。
方式 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(响应链式续接)——把每个回合显式关联到上一回合的响应 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}")last_response_id属性定义于 src/agents/result.py,返回最后一次模型响应的 ID。
补充要点:
- 当执行因等待审批暂停并从
RunState恢复时,SDK 会保留保存的conversation_id/previous_response_id/auto_previous_response_id设置,使恢复后的回合继续在同一服务端会话中; conversation_id与previous_response_id不可同时使用:需要跨系统共享的具名会话资源用conversation_id;需要回合间最轻量的 Responses API 续接基元用previous_response_id;- SDK 会对
conversation_locked错误进行带退避的自动重试:服务端会话执行在重试前会回滚内部会话跟踪器的输入,从而可安全重发相同已准备条目;本地 session 执行也会尽力回滚最近持久化的输入条目,减少重试后的历史重复。该兼容性重试即使未设置ModelSettings.retry也会执行;更广泛的模型请求 opt-in 重试行为见 模型文档。
四、钩子与定制:call_model_input_filter
call_model_input_filter在模型调用前一刻编辑模型输入。钩子接收当前 Agent、上下文与合并后的输入条目(含 session 历史,如有),返回新的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), )Runner 会把准备输入列表的副本交给钩子,因此可以在不原地修改调用方原列表的情况下完成裁剪、替换、重排。
执行时机语义:
- 使用 Sessions 时,
call_model_input_filter在会话历史已加载并与当前回合合并之后运行;想定制更早的合并过程本身,改用session_input_callback; - 使用
conversation_id/previous_response_id/auto_previous_response_id服务端会话时,钩子作用于为下一次 Responses API 调用准备的载荷;该载荷可能已只包含新回合的差分而不再重发全部历史,返回的条目会被记录为已发送。
典型用途包括敏感数据脱敏、长历史裁剪、插入额外系统指令——均通过run_config按执行粒度配置。
五、错误与恢复
5.1 错误处理器(error_handlers)
所有Runner入口都接受error_handlers:一个以错误种类为键的字典,支持的键为"max_turns"、"model_refusal"、"invalid_final_output"(类型定义见 src/agents/run_error_handlers.py)。当对应错误发生时,不再以异常终止执行,而是返回受控的最终输出。
处理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 的 structuredoutput_type校验、或模型未返回结构化最终消息时使用。处理器可返回应用特定的回退值,SDK 会针对同一output_type校验它;SDK不会重试模型调用或重新执行工具副作用。返回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)处理model_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)RunErrorHandlerResult.include_in_history默认为True:在 max_turns 处理器中,合成的回退输出会被加入会话历史并持久化到已配置的 session;若只想把回退返回给调用方而不写入历史或存储,设include_in_history=False。
六、持久化执行集成与人在回路
工具审批的暂停/恢复模式,请从专门的人在回路指南开始。以下集成面向执行可能跨越长时间等待、重试与进程重启的持久化编排场景:
- Dapr:借助 Agents SDK 的 Dapr(Diagrid)集成,可运行自动从故障恢复、支持人在回路工作流的持久化长时 Agent。Dapr 是厂商中立的 CNCF 工作流编排器;
- Temporal:借助 Temporal 集成,可运行包含人在回路任务的持久化长时工作流;
- Restate:借助 Restate 集成,可实现包含人工审批、handoff 与会话管理的轻量持久化 Agent;该集成把 Restate 单二进制运行时作为依赖,Agent 可运行在进程、容器或 serverless 函数中;
- DBOS:借助 DBOS 集成,可运行在故障与重启后仍保留进度的可靠 Agent,支持长时 Agent、人在回路工作流与 handoff,同步/异步方法均支持,仅需 SQLite 或 Postgres 数据库。
七、异常体系
SDK 在特定情况下抛出异常,完整列表见agents.exceptions。要点如下:
AgentsException:SDK 所有异常的基类,其余专属异常均派生自该通用类型;MaxTurnsExceeded:Agent 执行超过Runner.run/run_sync/run_streamed传入的max_turns时抛出,表示 Agent 未能在指定回合数(LLM 调用次数)内完成任务;max_turns=None可禁用限制;ModelTimeoutError:模型调用尝试超过ModelSettings.timeout时抛出,适用范围与重试行为见模型调用超时;ModelBehaviorError:底层模型产生意外或无效输出时抛出,包括:- 畸形 JSON(模型在工具调用或直接输出中返回非法 JSON,尤其定义了
output_type时); - 意外的工具相关失败(模型未能按要求使用工具);
- 失败或未完成的非流式 Responses 调用(最终状态为
failed或incomplete时,OpenAIResponsesModel与AnyLLMModel的 Responses 路径会抛出,异常携带最终状态及响应中可取得的错误/未完成详情);
- 畸形 JSON(模型在工具调用或直接输出中返回非法 JSON,尤其定义了
ModelRefusalError:模型拒绝产生所请求的输出时抛出,携带refusal文本;ToolTimeoutError:函数工具调用超过配置超时且工具使用timeout_behavior="raise_exception"时抛出;UserError:SDK 使用方写代码时的失误,通常源于实现错误、无效配置或 API 误用;InputGuardrailTripwireTriggered/OutputGuardrailTripwireTriggered:输入护栏条件满足时抛前者,输出护栏条件满足时抛后者;输入护栏在交付前检查入站消息,输出护栏在交付前检查 Agent 最终响应。
八、快速上手建议
- 一次性脚本:用
Runner.run_sync()即可;已在 async 环境(FastAPI、Notebook)中请改用Runner.run()。 - 需要流式体验:用
Runner.run_streamed()+.stream_events();多轮复用 WebSocket 连接时包一层responses_websocket_session()。 - 多轮对话:轻量用
result.to_input_list()手动续接;需要持久化用 Sessions(如 SQLiteSession 示例);需要跨服务共享会话用conversation_id;需要最简续接用previous_response_id。 - 生产可靠性:为
max_turns/invalid_final_output/model_refusal配置error_handlers,并善用tool_not_found_behavior="return_error_to_model"与tool_error_formatter保持运行可恢复。 - 可观测性:至少设置
RunConfig(workflow_name=...),并按需配置group_id、trace_metadata与trace_include_sensitive_data。
以上配置项与 API 均可在仓库的 src/agents/run.py、src/agents/run_config.py、src/agents/exceptions.py 中直接查阅,相关行为也有对应单元测试与 API 契约(如 tests/fixtures/released_api_contract.json)可以佐证。
【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考