OpenAI Agents SDK生产环境实战：从执行循环到多Agent编排

先说明一点：这个系列走到第四篇，我不打算再花篇幅重复“什么是 Agent”“怎么装 SDK”这些基础内容了。前三篇已经覆盖了从环境搭建、单个 Agent 的定义、工具调用，到基本的多 Agent 协作，如果你还没读过前面那些，建议先回去翻一下，不然后面很多东西你会觉得我在讲天书。

这一篇我想聊点真正能让你的项目从“demo 能跑”变成“生产环境能用”的东西：Agent 的内部执行循环到底是怎么转的、多个 Agent 之间怎么编排才不会乱、调试的时候怎么定位问题、上生产之后怎么保证它不会半夜挂掉。这些都是我在跑了大量真实场景之后踩出来的经验，不是官方文档里直接写着的那种“Hello World”级别的知识。

1. Agent 内部执行循环的真相：turns 与 lifecycle 拆解

1.1 不是“调用一次模型”，而是一个 while 循环

很多人刚开始写 Agents SDK 的时候，会把Runner.run(agent)理解成“调一次 LLM 拿个结果就结束”。这是最大的误解。实际上，runner.run()内部是一个循环：LLM 返回一个响应，SDK 会检查响应里有没有tool_calls或handoffs，有就继续执行工具、把结果发回模型，再让模型下一次输出，直到模型返回一个没有工具调用的最终消息，或者达到了max_turns上限。

这个机制的学名叫“agent 循环（agent loop）”，也叫turn（轮次）。一次 turn 包含：模型输出一次 + 执行其中的工具调用（可以有多个工具并行执行）+ 把工具结果送回模型。整条链路跑完一轮，叫一个run。

我见过很多新手在这个地方翻车：写了一个需要循环调用工具的任务，结果发现 Agent 只执行了一次工具就停了。原因是他们没有理解“模型要看到工具的结果之后，才会决定下一步做什么”，也没有显式告诉 Agent“不要急着结束，继续查”。这时候就涉及到一个很关键的概念——turn 控制。

from agents import Agent, Runner agent = Agent( name="ResearchAgent", instructions="你是研究助理，负责汇总信息。", ) result = Runner.run_sync( agent, input="帮我把2024年各大云厂商的定价模式整理出来", max_turns=10, )

max_turns是整个循环的上限。注意，它不是“模型调用次数”，而是“模型输出 + 工具执行”的完整轮次数。如果你的任务需要多轮工具调用链，这个值给太小会导致 Agent 在中间被硬生生截断，然后给你一个半成品答案。我自己的经验是：普通工具类任务给 5~8，涉及深入调研、需要多次思考的任务给 15~20。但同时，这个值也是防死循环的保命绳，给太大反而危险，具体怎么权衡后面专门讲。

1.2 模型输出、工具调用与 output guardrails 的执行顺序

Agent 循环里还有一个容易被忽略的细节：output guardrails（输出护栏）也不是只在最后一步生效的。准确说，它在每一次 turn 的模型输出之后都会跑一遍。也就是说，如果你写了 output guardrails，模型每生成一条消息，SDK 都会先拿这条消息去跑护栏，出了问题就直接打断循环，不再往下走。

这个机制实际用起来有两面性。好的一面是，你可以在 Agent 中途跑偏的时候尽早拦截，不必等它把工具调用链全跑完；坏的一面是，如果你在 output guardrails 里写了耗时的校验逻辑，比如调用外部接口做敏感信息检测，那每个 turn 都会付出一次额外延迟，整个 run 的时间会被拉长好几倍。

我在一个金融问答项目里踩过这个坑：当时给 Agent 配了一个“输出是否含个人手机号”的 guardrail，规则本身没问题，但没意识到它每次 turn 都会触发。最后线上一个 6 turn 的任务平均耗时从前一天的 8 秒涨到了 32 秒。后来做了两个改动才解决：一是把 guardrail 的规则从“调用外部接口”换成“本地正则匹配”，耗时从几百毫秒降到几毫秒；二是明确判断哪些 turn 需要严格校验、哪些可以跳过。输出护栏适合拦截“绝对红线”，不适合做高频次的质量过滤，这个定位要想清楚。

关于 guardrails 本身，还有个我认为很重要的点：input guardrails 和 output guardrails 的执行时机完全不同。input guardrails 在 turn 循环开始前的第一条消息进入时就执行一次；output guardrails 在每一轮模型输出后都会执行。理解了这一点，你才会明白为什么 input guardrail 挂了不影响后续工具调用、而 output guardrail 一旦触发会直接中断整个 run。

1.3 Agent 之间的交接不是“发消息”，是“转交控制权”

前三篇里我提过 handoffs，但当时只是说了“A 可以把任务交给 B”。这一篇我要把话说透：handoffs 本质上是一个特殊的 tool call，不是进程间的消息传递，也不是把对话历史拷贝过去。它的底层逻辑是——当前 Agent 决定自己搞不定了，于是调用一个内置工具，告诉 Runner：“接下来的循环控制权给 Agent B，我给你一个理由，并附带结构化数据。”

from agents import Agent, Runner from agents.extensions.handoff_prompt import RECOMMENDED_PROMPT_PREFIX billing_agent = Agent( name="BillingAgent", instructions="你负责账单和订单问题。", handoff_description="专门处理账单、发票、订单支付问题", ) triage_agent = Agent( name="TriageAgent", instructions=f"{RECOMMENDED_PROMPT_PREFIX}你是前置客服，先判断用户问题归属，再交给对应专员。", handoffs=[billing_agent], )

这里有一个非常关键的参数：handoff_description（交接描述）。这串描述在 SDK 内部会被转成工具 schema 里的 description 字段，也就是模型在决定“要不要交接、交给谁”时看到的唯一参考。很多人写 handoffs 不填这个字段，或者写得很随意，比如“负责其他事情”，后果就是模型完全不知道该什么场景下触发交接，结果就是把规则问题交给了账单 Agent，把账单问题留在了自己手里。

我见过一份写得特别好的 handoff_description，原文是：“负责处理退货退款，包括但不限于质量问题、七天无理由、物流丢件理赔，只有当用户明确要退货时才交接，咨询类不要转过来。”这一句把触发条件、范围边界、排除项全说清楚了。写交接描述的三个要素是：管什么、什么时候转、什么时候千万别转。缺一个，模型就会在边界场景里给你做错误决策。

另外，handoffs 传递的是控制权，不是对话历史副本，但上下文窗口仍然会包含之前的消息（在同一个 run 内）。这一点也很容易被误解——有人以为换 Agent 就换了个干净 slate，其实严格来说，上下文里还是会有前面 Agent 说的话的。如果你希望 B 在接手时不看到 A 的思考过程，那就要考虑通过 session 隔离或信息摘要的方式来切分上下文，这个我在后面第五节展开。

2. 多 Agent 协作里的执行调度：串行、并行与资源水位

2.1 先搞清楚你的编排是“路由”还是“流水线”

当 Agent 数量超过 3 个之后，架构混乱是必然的。我发现最容易犯的一个错误是把编排方式理解成单一的“谁先谁后”。实际上，多 Agent 协作的拓扑结构基本只有三种：

• 路由型：一个入口 Agent 判断任务类型，分发给下游不同专家 Agent。典型场景是智能客服分流、工单分类。
• 顺序型：A 的输出是 B 的输入，B 的输出是 C 的输入，形成一条流水线。典型场景是“信息提取 → 分析 → 报告生成”。
• 并行型：多个 Agent 各干各的，最后汇总。典型场景是同时调研多个竞品，最后统一整理。

这三种形态不是互斥的，真实系统往往是它们的组合：入口路由 → 某一步并行调研 → 汇总 Agent 收口。问题在于，很多人在代码层面没有体现出这样的结构，全部用线性代码串起来，导致明明可以并行的动作被白白等成了串行，白白丢掉了好几倍性能。

import asyncio from agents import Agent, Runner async def run_market_research(): agents = [ Agent(name="ResearcherA", instructions="调研厂商A的定价"), Agent(name="ResearcherB", instructions="调研厂商B的定价"), Agent(name="ResearcherC", instructions="调研厂商C的定价"), ] async def run_one(agent, topic): result = await Runner.run(agent, topic) return result.final_output # 并行跑三个调研 Agent outputs = await asyncio.gather( run_one(agents[0], "A厂商"), run_one(agents[1], "B厂商"), run_one(agents[2], "C厂商"), ) return outputs

2.2 Runner.run 与多 Agent 并行时的真实代价

并行编排最容易被忽略的是资源水位。OpenAI Agents SDK 在同一个进程里可以轻松并发跑多个 Runner.run，但每个 run 背后都是一个完整的 LLM 调用流。也就是说，你开 10 路并行，就是同时向 API 发起 10 个独立请求，token 消耗、QPS 配额、上下文容量都会成倍增长。很多人做完并行化之后发现 API 开始频繁返回 429 限流错误，就是这个原因。

我自己做数据收集型项目时，会给并行度设一个上限，比如同一时刻最多 3~4 个 Agent 在跑，剩下的排队等待。实现方式很粗暴但有效：用信号量（Semaphore）控制并发数量。

import asyncio from agents import Agent, Runner semaphore = asyncio.Semaphore(3) async def limited_run(agent, task): async with semaphore: result = await Runner.run(agent, task) return result.final_output async def main(): tasks = [limited_run(Agent(name="Agent", instructions=...), f"任务{i}") for i in range(10)] return await asyncio.gather(*tasks)

这个“重试 + 限流 + 并发控制”的组合，在我跑过的数据采集项目里稳定度提升了非常多。别小看这层控制，很多线上故障不是模型问题，而是你自己的进程资源被某个失控的 Agent 批量请求打满了。调度层做得好的系统，底下的 Agent 随便怎么波动都是稳的。

另外说一句，Runner 的 API 设计是支持异步的（Runner.run 是 async 版本，Runner.run_sync 是同步版）。如果你在 FastAPI 这类异步 Web 框架里用了run_sync，并且并发量一高，线程池会被占满，整个服务就卡死了。不要问我是怎么知道的。我的建议很简单：Web 服务里一律用async模式，把 run 丢进事件循环，别用同步版本。

3. 追问题别靠瞎猜：Tracing 与可观测性建置

3.1 SDK 自带 Tracing 能让你看到每一层干了什么

OpenAI Agents SDK 最被低估的能力其实是内置的 tracing（追踪）系统。它不是普通的日志打印，而是把一次 Agent run 完整拆成了 trace → span 的树状结构，从一次 run 开始，到每一轮模型调用、每一个工具执行、每一条 guardrail 触发，全都被记录下来。你可以清楚地看到：这次 run 为什么花了 15 秒？瓶颈在哪一步？那个工具有没有卡住？模型是在哪一步产生了错误的 handoff 决定？

官方推荐的做法是把 trace 数据导到外部平台看，比如 Logfire 或者你自己配置的处理器。但如果你不想额外依赖任何 SaaS，SDK 也允许你通过 processor 接口把 traces 写到本地文件，或者转发到自己的日志中心。

from agents import set_trace_processors, trace_processor @trace_processor def my_processor(trace): # trace 对象里有 spans，每一个 span 是模型调用/工具执行的切片 with open("traces.jsonl", "a") as f: f.write(trace.to_json() + "\n") return trace set_trace_processors([my_processor])

这段代码会把每次 run 的完整轨迹追加写入traces.jsonl，字段里包含模型名称、耗时、token 数、工具调用参数和结果。我强烈建议每折腾一个复杂一点的 Agent 场景，就打开这个开关跑一轮，然后去看 JSON 里的 span 树。你会发现一些平时根本意识不到的问题，比如某个工具被调用了两次（模型第一次拿到的结果不满意，又重复调用）、某个 guardrail 耗时占比异常地高、某次手写代码传入的参数比预期大得多。这些都是调整系统的重要依据。

3.2 如何从 trace 数据里定位“答非所问”的根因

说一个我实际处理过的 case。有一次线上一个客服 Agent 出现了“用户问 A，Agent 却答 B”的问题。我当时没有急着改 prompt，而是先导出了那段时间的 trace 数据。结果发现，这个 Agent 在前两轮明明已经正确判断出用户需要退款，但在第三轮却做了一个多余的工具调用，把订单状态查成了另一个订单号，然后模型基于这个错误状态给出了牛头不对马嘴的回复。

看完 trace 我才知道，根因不在提示词，而在于工具返回的数据格式不明确——查询工具返回的是一个没有字段名的大 JSON，模型在上下文里误读了数字含义。于是我把工具返回改成了结构化文本，加上“这是最近一单的订单号”“当前状态为已发货”这样的显式描述。改完之后，同样的问题再也没出现过。

这个案例我觉得特别典型，因为它说明一个道理：调试 Agent 别靠“猜 prompt 写得不好然后一顿乱改”。先看 trace，让数据告诉你问题出在哪一层——是模型决策、工具输出、guardrail 拦截还是外部接口异常，然后你才知道应该动哪里。

4. 生产级可靠性策略：max_turns、重试与错误分类

4.1 无限循环是失控的源头，防死循环要同时做三层限制

我把这个标题起名叫“防死循环的三层限制”，是因为只靠max_turns一个参数根本挡不住实际问题。max_turns挡不住的是：模型在循环里干的事太耗时（比如每次调用外部 API 就要 5 秒），或者工具不停地在产生新副作用（比如每次循环都写入一条数据库记录）——即使然后对齐max_turns，副作用已经造成了。

真正的三层限制应该是：第一层，max_turns限制轮数；第二层，给每个工具加超时；第三层，在业务逻辑层面做任务级别的时间预算。我自己的实现是，每个 throw 的任务都套一个总超时，比如任务最多执行 30 秒，超时就取消整个 run，返回一条兜底消息给用户。

import asyncio from agents import Agent, Runner async def run_with_budget(agent, user_input, budget_seconds=30): try: return await asyncio.wait_for( Runner.run(agent, user_input), timeout=budget_seconds, ) except asyncio.TimeoutError: return "抱歉，这个请求处理超时了，请稍后再试。"

这里用asyncio.wait_for做的任务级超时，其实是兜底中的兜底，但它非常关键。LLM 本身没有“响应时间”的概念，工具调用链也可能因为外部接口变慢而不合理地拉长。预算超时能保证用户侧的体验不会因为后端某个 Agent 的失控而无限恶化。

4.2 异常分类：哪些错误值得重试，哪些重试也没用

运行 Agent 应用，你会遇到四类典型异常：限流（429）、服务端过载（5xx）、鉴权失败（401/403）、依赖的外部工具异常。我的经验是：不要对所有这些异常做同一套重试逻辑，要分类处理。

把鉴权错误和限流错误混在一起处理，是我见过的最多的错误。做过那套逻辑之后你会发现，429 重试是有意义的，因为过两秒配额可能就恢复了；但 401 哪怕重试一百次也一样是 401，纯粹是浪费时间和钱。

4.3 上下文失控：会话状态的瘦身与摘要

还有一个生产环境必然遇到的坑：长时间运行的 Agent 上下文会越长越大，先是警告提示 token 超限，接着就是 API 报错。这时候你会面临两个选择：把消息历史硬截断，或者摘要化。硬截断简单但会丢掉重要信息，比如用户三天前提过的偏好设置；摘要化更聪明，但由于摘要本身就是模型生成的，也可能有信息失真。

我用的策略是“分层压缩”：保留最近 10 轮完整对话，更早的消息交给一个压缩 Agent，让它提取关键事实、用户偏好、未完成事项，压缩成 300 字以内的摘要，在进入新的 run 时把摘要作为 system 附加内容。这个方案的成本是每次压缩要调一次模型，但换来的是上下文可控、成本可控。官方 SDK 里有专门的 session 对象来管理这类持久状态，但我实际用下来发现，最重要的不是 SDK 提供了什么，而是你自己定义好“哪些信息必须保留、哪些可以丢”。没有这个取舍标准，任何压缩算法都会丢错东西。

5. 我踩过的几个反模式：为什么你的 Agent 总在关键时刻掉链子

5.1 反模式一：把 Agent 当 API 包装器用

这是最多人犯的错误：创建一个 Agent，instructions 里写了“你是 API，用户的输入直接转发给后端接口，接口返回什么你就输出什么”。这种 Agent 完全没用到模型的能力，反而因为 LLM 的随机性给你加了戏——明明让你原样转发，它非要润色一下、加个问候语、把 JSON 格式改了。

正确的做法是，这种场景就不要用 Agent，直接用原生 API 调用。Agent 的价值在于“模型能基于工具反馈进行多步决策”，如果你的任务不需要决策，绑定 Agent、指定工具、走 handoffs 反而增加延迟和失败概率。我曾经把一组简单的“查天气接口然后返回”的工具封装成 Agent，结果发现每次调用有 20% 的概率模型在返回结果前面加一句“好的，这是您所在城市的天气”。后来我把它换成了普通函数调用，延迟从 2 秒降到 500 毫秒，再也没出过格式问题。

5.2 反模式二：Agent 链越深越好

我在早期做多 Agent 系统的时候，也特别迷信“分层越细越聪明”的理论。路由 Agent → 子路由 Agent → 执行 Agent → 汇总 Agent，整整四层，结果每层都在消耗 token、都在引入延迟和随机性，最后一环出了问题，前面三层全都白白跑了一遍。

后来我做了一次减法：把四层砍到两层——入口 Agent 直接路由到专家 Agent，专家 Agent 完成后直接返回结果。准确率没有下降，但平均延迟减少了 40%，成本也低了很多。这里有个判断标准我觉得很实用：如果下一层 Agent 能做的事，用一个工具调用就可以搞定，那就不要用 Agent。Agent 只在需要模型“自己决定下一步做什么”的时候才值得存在。

5.3 反模式三：忽略 Agent 的随机性，没有做输出稳定性设计

很多人写应用的时候，把 Agent 的返回结果直接当结构化数据用，比如让 Agent 输出一段 JSON，然后代码里json.loads(result.final_output)。这在 demo 里没问题，上生产就等着半夜被报警吵醒吧——模型生成 JSON 不可能保证每次都合法。我发现的最稳定方案是：不要依赖模型直接输出 JSON 再用代码解析，而是让 Agent 调用一个“格式化工具”，把信息写进工具参数里，由代码拿工具参数做结构化处理。这样，结构化数据走的是工具调用通道而不是文本通道，可靠性会提高一个量级。

import json from agents import Agent, Runner def report_formatter(content: str, category: str, confidence: float): # 由系统生成的工具调用参数，天然是结构化数据 return json.dumps({"content": content, "category": category, "confidence": confidence}) agent = Agent( name="StructuredAgent", instructions="把分析结果通过 report_formatter 工具输出，不要直接输出 JSON。", tools=[report_formatter], )

这个技巧的适用面其实非常广。凡是“模型需要产出一段固定结构数据”的场景，都值得改为“模型调用工具传入结构化参数”，你的下游代码从解析文本改成了直接读工具参数，省掉了文本解析的脆弱性。

5.4 反模式四：sessions 和 handoffs 被混用

最后说一个比较容易混淆的设计问题。sessions 是用来保存多轮对话上下文状态的，handoffs 是用来切换职责的，两个机制负责的事情完全不同。但现实里我看到不少项目把 handoffs 当上下文清理机制用——“切到 B 就相当于重新开始，上下文干净了”——这个理解是有代价的。正如前面说的，在同一个 run 内部，handoffs 之后模型上下文里还是能看到前一个 Agent 的输出的，所以如果你真的想隔离上下文，要靠 session 管理，而不是 handoff。

用一句话总结：handoffs 是任务的重新分配，sessions 是上下文的存续边界。把两者的边界划清楚了，多 Agent 系统的架构才会干净，排查问题也不会一头雾水。

6. 一些最终的碎碎念

做 Agent 系统跟做普通 API 服务很不一样，最大的不一样在于：你没法百分之百控制系统行为，模型给出的结果本质上是一个概率分布上的采样。所以这个领域的“工程”其实就是围绕着“降低方差”展开的。工具的 schema 要写得尽可能清晰、guardrails 的规则要遵守单一职责、tracing 数据要落地、并发水位要有上限、异常要分类处理，这些单看每一项都不难，难的是把它们组合在一起形成一个稳定系统。

我个人在写完这套体系的 Agent 系统之后，最大的体感变化是：线上故障从“不知道发生了什么”变成了“看一眼 trace 就能定位到具体一步”，工作日晚上被电话吵醒的频次明显下降。如果你也正在用 OpenAI Agents SDK 构建自己的应用，我的建议是别急着加功能，先把 trace、超时、max_turns、错误分类这四件事做好。这四件做完之前加再多花活，后面都会变成你熬夜排查的素材。

一点点经验，不保证每一条都适合所有人，但希望至少能帮你少走几个弯。