先说明一点这个系列走到第四篇我不打算再花篇幅重复“什么是 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( nameResearchAgent, instructions你是研究助理负责汇总信息。, ) result Runner.run_sync( agent, input帮我把2024年各大云厂商的定价模式整理出来, max_turns10, )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( nameBillingAgent, instructions你负责账单和订单问题。, handoff_description专门处理账单、发票、订单支付问题, ) triage_agent Agent( nameTriageAgent, instructionsf{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(nameResearcherA, instructions调研厂商A的定价), Agent(nameResearcherB, instructions调研厂商B的定价), Agent(nameResearcherC, 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 outputs2.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(nameAgent, 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 或者你自己配置的处理器。但如果你不想额外依赖任何 SaaSSDK 也允许你通过 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 出现了“用户问 AAgent 却答 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_seconds30): try: return await asyncio.wait_for( Runner.run(agent, user_input), timeoutbudget_seconds, ) except asyncio.TimeoutError: return 抱歉这个请求处理超时了请稍后再试。这里用asyncio.wait_for做的任务级超时其实是兜底中的兜底但它非常关键。LLM 本身没有“响应时间”的概念工具调用链也可能因为外部接口变慢而不合理地拉长。预算超时能保证用户侧的体验不会因为后端某个 Agent 的失控而无限恶化。4.2 异常分类哪些错误值得重试哪些重试也没用运行 Agent 应用你会遇到四类典型异常限流429、服务端过载5xx、鉴权失败401/403、依赖的外部工具异常。我的经验是不要对所有这些异常做同一套重试逻辑要分类处理。异常类型是否值得重试推荐策略限流 429 / 5xx值得但要退避指数退避初始 1 秒上限 30 秒重试最多 3 次鉴权 401/403不值得直接报错检查 API key 和权限配置工具外部接口异常看接口语义只读接口可重试写操作接口不能盲目重试防重复提交上下文过长报错不值得重试压缩上下文或切换模型代码层面做 chunking把鉴权错误和限流错误混在一起处理是我见过的最多的错误。做过那套逻辑之后你会发现429 重试是有意义的因为过两秒配额可能就恢复了但 401 哪怕重试一百次也一样是 401纯粹是浪费时间和钱。4.3 上下文失控会话状态的瘦身与摘要还有一个生产环境必然遇到的坑长时间运行的 Agent 上下文会越长越大先是警告提示 token 超限接着就是 API 报错。这时候你会面临两个选择把消息历史硬截断或者摘要化。硬截断简单但会丢掉重要信息比如用户三天前提过的偏好设置摘要化更聪明但由于摘要本身就是模型生成的也可能有信息失真。我用的策略是“分层压缩”保留最近 10 轮完整对话更早的消息交给一个压缩 Agent让它提取关键事实、用户偏好、未完成事项压缩成 300 字以内的摘要在进入新的 run 时把摘要作为 system 附加内容。这个方案的成本是每次压缩要调一次模型但换来的是上下文可控、成本可控。官方 SDK 里有专门的 session 对象来管理这类持久状态但我实际用下来发现最重要的不是 SDK 提供了什么而是你自己定义好“哪些信息必须保留、哪些可以丢”。没有这个取舍标准任何压缩算法都会丢错东西。5. 我踩过的几个反模式为什么你的 Agent 总在关键时刻掉链子5.1 反模式一把 Agent 当 API 包装器用这是最多人犯的错误创建一个 Agentinstructions 里写了“你是 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( nameStructuredAgent, 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、错误分类这四件事做好。这四件做完之前加再多花活后面都会变成你熬夜排查的素材。一点点经验不保证每一条都适合所有人但希望至少能帮你少走几个弯。