OpenAI Agents SDK #15:一个 Agent 搞不定——加一个真的够吗?

从「加三个 Agent 并发就能解决问题吗?」这个真实痛点切入,系统拆解 OpenAI Agents SDK 的两种多 Agent 编排范式——LLM 驱动(Orchestrator 自主决策)与代码驱动(确定性流水线)。深入讲解 `as_tool()` 与 Handoff 的本质区别、`asyncio.gather` 并发的适用边界与代码示例、多 Agent 间 Context 的流转机制,并结合中文社区的三个高频踩坑(上下文丢失、依赖并发错误、max_turns 失控),给出 3 条立即可落地的实践建议。

研究速览

多 Agent 编排的两种路,你走哪条?

两种编排范式,两种世界观

SDK 文档里清清楚楚地写着两种模式1
LLM 驱动编排:一个 Orchestrator Agent 掌舵,它读懂 instructions,决定调用哪个工具、什么时候 handoff 给哪个子 Agent,整个流程靠模型的推理来驱动。
代码驱动编排:流程写死在 Python 里。await Runner.run() 调这个 Agent,拿到结构化输出,用 if 判断要不要继续,然后再 await Runner.run() 调下一个。确定性流水线,人说了算。
乍看起来,LLM 驱动更「智能」,代码驱动更「笨」。但这个判断完全搞反了。
维度LLM 驱动代码驱动
灵活性高,可处理开放式任务低,流程预先定义
可预测性低,依赖模型推理高,行为完全确定
调试难度高,「黑盒」决策低,逻辑清晰可追踪
成本高,多次 LLM 调用低,减少不必要推理
适用场景开放式问题、动态路由多步工程任务、批量处理
现实里,大多数「坚持用 LLM 驱动」的团队,后来都在某个角落偷偷补了一堆代码驱动的流程——因为 LLM 在第三次 handoff 时突然「脑洞大开」跑偏了,而他们没有办法在代码层面精确控制2

as_tool():把 Agent 变成工具

代码编辑器中的深色主题编程界面
代码编辑器中的深色主题编程界面
这是多 Agent 协作里最容易被忽视的机制,也是最有用的一个3

Handoff vs as_tool(),到底什么区别?

先说 Handoff:控制权转移。主 Agent 调用 handoff 后,目标 Agent 接管整条对话历史,相当于「接班」——下一个 Agent 知道之前发生了什么,从那个历史状态继续。
再说 as_tool():任务委托。主 Agent 调用子 Agent 就像调普通函数,子 Agent 只拿到本次调用生成的输入,完成任务后把结果返回给主 Agent,控制权始终在主 Agent 手里4
一句话总结:
  • Handoff = 把方向盘交给别人
  • as_tool() = 让别人帮你做一个子任务,结果交回来你继续开

完整代码示例:Manager 模式(三语翻译)

这段代码解决什么问题:orchestrator 决定调用顺序,三个语言 Agent 并行承接任务,synthesizer 负责最终合并校验。
import asyncio
from agents import Agent, Runner, trace
from agents.extensions.handoff_prompt import RECOMMENDED_PROMPT_PREFIX

# ── Step 1:定义三个专业翻译 Agent ──────────────────────────────
spanish_agent = Agent(
    name="西班牙语翻译",
    instructions="你是一名专业的英译西翻译,只输出翻译结果,不加任何解释。",
    handoff_description="将英文文本翻译为西班牙语",
)

french_agent = Agent(
    name="法语翻译",
    instructions="你是一名专业的英译法翻译,只输出翻译结果,不加任何解释。",
    handoff_description="将英文文本翻译为法语",
)

italian_agent = Agent(
    name="意大利语翻译",
    instructions="你是一名专业的英译意翻译,只输出翻译结果,不加任何解释。",
    handoff_description="将英文文本翻译为意大利语",
)

# ── Step 2:构建 Orchestrator,将三个 Agent 转换为工具 ──────────
orchestrator_agent = Agent(
    name="翻译编排器",
    instructions=(
        f"{RECOMMENDED_PROMPT_PREFIX}\n"
        "你是一个多语言翻译编排器。收到翻译请求后,"
        "调用西班牙语、法语、意大利语三个翻译工具,"
        "按顺序逐一调用,收集全部翻译结果。"
    ),
    tools=[
        # as_tool() 将 Agent 包装成普通工具
        # 子 Agent 只拿到 tool 调用时生成的输入,不接收主 Agent 的完整历史
        spanish_agent.as_tool(
            tool_name="translate_to_spanish",
            tool_description="将英文翻译为西班牙语",
        ),
        french_agent.as_tool(
            tool_name="translate_to_french",
            tool_description="将英文翻译为法语",
        ),
        italian_agent.as_tool(
            tool_name="translate_to_italian",
            tool_description="将英文翻译为意大利语",
        ),
    ],
)

# ── Step 3:Synthesizer 合并校验结果 ───────────────────────────
synthesizer_agent = Agent(
    name="结果合成器",
    instructions=(
        "你收到了三种语言的翻译结果,请按以下格式输出最终版本:\n"
        "🇪🇸 西班牙语:...\n🇫🇷 法语:...\n🇮🇹 意大利语:..."
    ),
)

# ── Step 4:完整流程 ────────────────────────────────────────────
async def translate_all(text: str) -> str:
    with trace("多语言翻译工作流"):  # trace 包装,方便 Tracing 面板观察
        # 第一步:orchestrator 编排三个翻译工具
        orchestrator_result = await Runner.run(
            orchestrator_agent,
            f"请将以下文本翻译成三种语言:{text}",
        )
        # 第二步:synthesizer 合并输出
        # to_input_list() 把上一步的结果链接到下一步的输入
        final_result = await Runner.run(
            synthesizer_agent,
            orchestrator_result.to_input_list(),
        )
        return final_result.final_output

if __name__ == "__main__":
    result = asyncio.run(translate_all("The quick brown fox jumps over the lazy dog."))
    print(result)
关键点:子 Agent 通过 as_tool() 暴露给 orchestrator,orchestrator 决定调用时机与顺序,结果链式传给 synthesizer。整个控制权始终在 orchestrator 手里——这正是 Manager 模式的精髓5

并发编排:asyncio.gather 真正提速的姿势

指挥中心多屏控制台,象征多 Agent 并发协作
指挥中心多屏控制台,象征多 Agent 并发协作
上面的翻译 orchestrator 是串行调用三个工具。如果你的任务之间没有依赖关系,并发才是正确做法。

这段代码解决什么问题:同一批任务独立运行,互不等待

import asyncio
from agents import Agent, Runner

# 定义三个独立任务的 Agent
summarize_agent = Agent(
    name="摘要 Agent",
    instructions="用三句话总结以下内容。",
)

sentiment_agent = Agent(
    name="情感分析 Agent",
    instructions="分析以下文本的情感倾向,输出:正面/负面/中性,附一句理由。",
)

keyword_agent = Agent(
    name="关键词提取 Agent",
    instructions="提取以下文本的 5 个核心关键词,用逗号分隔输出。",
)

async def analyze_text(text: str):
    """
    三个 Agent 并发执行,总耗时 ≈ 最慢的那个,而不是三者之和
    适用条件:各 Agent 之间没有数据依赖
    """
    summary_task = Runner.run(summarize_agent, text)
    sentiment_task = Runner.run(sentiment_agent, text)
    keyword_task = Runner.run(keyword_agent, text)

# gather 等待所有任务完成,结果顺序与输入顺序对应
    summary_result, sentiment_result, keyword_result = await asyncio.gather(
        summary_task,
        sentiment_task,
        keyword_task,
    )

return {
        "summary": summary_result.final_output,
        "sentiment": sentiment_result.final_output,
        "keywords": keyword_result.final_output,
    }

if __name__ == "__main__":
    sample_text = """
    OpenAI Agents SDK 在 2026 年持续迭代,多 Agent 编排能力进一步成熟。
    v0.15.2 修复了交接过滤器的 bug,测试覆盖率也在持续扩展。
    """
    result = asyncio.run(analyze_text(sample_text))
    for k, v in result.items():
        print(f"[{k}] {v}")
并发适用条件很简单:各 Agent 的输入不依赖彼此的输出。只要满足这一点,asyncio.gather 就能把总耗时从「串行之和」压缩到「最慢那个单任务的耗时」1

什么时候不能并发?

如果任务 B 需要任务 A 的输出,就必须串行。代码驱动的链式调用正是为这种场景而生6
# 代码驱动串行示例:大纲 → 质量评估 → 按条件决定是否写故事
from pydantic import BaseModel

class OutlineCheckerOutput(BaseModel):
    good_quality: bool   # 大纲质量是否合格
    is_scifi: bool       # 是否属于科幻主题

story_outline_agent = Agent(name="大纲生成", instructions="生成一个科幻故事大纲。")
outline_checker_agent = Agent(
    name="大纲评估",
    instructions="评估大纲质量和题材,输出结构化结果。",
    output_type=OutlineCheckerOutput,  # 强制结构化输出,方便代码层判断
)
story_agent = Agent(name="故事写作", instructions="根据大纲写一个完整的科幻短故事。")

async def generate_story(prompt: str) -> str:
    # Step 1:生成大纲
    outline_result = await Runner.run(story_outline_agent, prompt)

# Step 2:评估大纲——结构化输出保证 .final_output 是 OutlineCheckerOutput 实例
    check_result = await Runner.run(
        outline_checker_agent,
        outline_result.final_output,
    )
    evaluation: OutlineCheckerOutput = check_result.final_output

# Step 3:代码层判断,不是 LLM 在猜——这就是「代码驱动」的关键
    if not evaluation.good_quality or not evaluation.is_scifi:
        print("大纲不符合要求,流程终止。")
        return ""

# Step 4:写故事
    story_result = await Runner.run(story_agent, outline_result.final_output)
    return story_result.final_output
这个示例展示了代码驱动编排的核心:第三步的 if 判断由 Python 执行,不是 LLM 在推理6。这意味着行为百分之百可预期,也百分之百可测试。

多 Agent 间的 Context 流转

上一篇 #10 讲了 RunContextWrapper 的双轨设计——本地 Context 对 LLM 不可见,只在工具函数内流转。多 Agent 场景下,这个机制有一个关键特性需要牢记。
Handoff 场景:Context 对象自动传递给下一个 Agent,原来的 run_context 实例在整条链路里是同一个,不会复制7
as_tool() 场景:子 Agent 在独立的 Runner 内运行,默认情况下没有父 Agent 的 Context。如果子 Agent 的工具函数需要用到父级上下文数据(比如用户 ID、数据库连接),你需要在调用时显式传入。
from dataclasses import dataclass
from agents import Agent, Runner, RunConfig

@dataclass
class AppContext:
    user_id: str
    db_connection: str  # 实际场景里是数据库连接对象

context = AppContext(user_id="u_12345", db_connection="postgresql://...")

# as_tool() 子 Agent 通过 run_config 传入 context
child_agent = Agent(name="子任务 Agent", instructions="完成具体的数据处理任务。")

parent_agent = Agent(
    name="主编排 Agent",
    instructions="根据任务类型调用子任务工具。",
    tools=[
        child_agent.as_tool(
            tool_name="process_data",
            tool_description="处理用户数据",
        )
    ],
)

# 在主 Agent 运行时传入 context,子 Agent 通过 run_config 继承
result = await Runner.run(
    parent_agent,
    "处理用户数据",
    context=context,  # context 在整条链路中自动流转
)
as_tool() 内部,SDK 会把父 Agent 运行时的 context 传递给子 Agent 的执行环境,所以工具函数里的 ctx.context 访问的仍然是同一个 context 实例4

踩坑经验:中文社区的三个高频错误

正在加载统计卡片...

as_tool() 当 Handoff 用——上下文全丢

知乎的讨论里有个典型案例8:triage agent 通过 handoff 委托给 refund agent,refund agent 再调用 pricing agent 的 as_tool()。开发者发现 pricing agent「不知道用户说了什么」,排查半天才明白——as_tool() 的子 Agent 只拿到工具调用时构造的输入,不是完整对话历史
解决方案:如果子 Agent 需要完整上下文,用 Handoff;只需要完成一个独立子任务,用 as_tool()。判断标准是「子 Agent 需要知道之前发生了什么吗?」

把有依赖的任务丢进 asyncio.gather

「A 的输出是 B 的输入」这种情况,如果用 asyncio.gather 并发,B 会拿到空数据或者 None。结果不是报错,而是 B 「安静地」输出了一个毫无意义的结果——这种 bug 很难发现。
解决方案:画出任务依赖图,有箭头的地方就是串行,没有箭头的地方才是并发。这两种结构经常混合出现,不要偷懒全部丢进 gather

LLM 驱动编排忘了设 max_turns

LLM 驱动编排的 orchestrator 在开放任务上特别容易「转圈」:调工具 A 拿结果 → 觉得不够好再调工具 A → 觉得还要改 → 调工具 B 再来一遍……
SDK 默认的 max_turns 是 109。生产环境里千万别依赖默认值。根据任务的实际步骤上限,显式设置一个合理的 max_turns,否则一个出问题的 orchestrator 会不停消耗 token 直到被外部超时机制终止。

实践建议:3 条立即可落地的

先用代码驱动,再考虑 LLM 驱动。大多数业务流程——提取、转换、分类、生成——都可以拆成确定性步骤,没必要引入 LLM 来做路由决策。代码驱动比 LLM 驱动快、便宜、好调试。只有当任务本身是「开放式、需要动态规划」时,再考虑用 LLM 做 orchestrator2
as_tool() 给独立子任务,Handoff 给需要上下文继承的场景。这条规则基本能避免 90% 的上下文问题。判断方法就一个:「子任务完成后,控制权需要回到主 Agent 吗?」需要回来就用 as_tool(),不需要就用 Handoff。
并发之前先画依赖图,确认没有箭头才丢进 gather。花 5 分钟在白板上画出各 Agent 之间的数据流,然后对着图写代码,比事后排查「为什么并发结果不对」省出几倍时间。Vercel 工程团队砍掉 80% 工具数量后可靠性反而大幅提升10——越少的节点,越清晰的拓扑,越不容易出问题。

下一篇预告:#16 Results 解析

理解了多 Agent 的编排逻辑,下一个问题自然出现:Runner 跑完之后,你拿到的 RunResult 对象里到底装了什么?final_outputnew_itemsto_input_list() 之间是什么关系?为什么有时候取到空值?
#16 我们系统拆解 Results——把 RunResult / RunResultStreaming 的完整结构摸透,帮你写出能正确处理各种返回情况的生产代码。

封面图来自 Pexels · Tara Winstead

围绕这条内容继续补充观点或上下文。

  • 登录后可发表评论。