多 Agent 编排高频踩坑
来自中文开发者社区的真实反馈(2026年4-5月)
从「加三个 Agent 并发就能解决问题吗?」这个真实痛点切入,系统拆解 OpenAI Agents SDK 的两种多 Agent 编排范式——LLM 驱动(Orchestrator 自主决策)与代码驱动(确定性流水线)。深入讲解 `as_tool()` 与 Handoff 的本质区别、`asyncio.gather` 并发的适用边界与代码示例、多 Agent 间 Context 的流转机制,并结合中文社区的三个高频踩坑(上下文丢失、依赖并发错误、max_turns 失控),给出 3 条立即可落地的实践建议。
研究速览
await Runner.run() 调这个 Agent,拿到结构化输出,用 if 判断要不要继续,然后再 await Runner.run() 调下一个。确定性流水线,人说了算。| 维度 | LLM 驱动 | 代码驱动 |
|---|---|---|
| 灵活性 | 高,可处理开放式任务 | 低,流程预先定义 |
| 可预测性 | 低,依赖模型推理 | 高,行为完全确定 |
| 调试难度 | 高,「黑盒」决策 | 低,逻辑清晰可追踪 |
| 成本 | 高,多次 LLM 调用 | 低,减少不必要推理 |
| 适用场景 | 开放式问题、动态路由 | 多步工程任务、批量处理 |
as_tool():把 Agent 变成工具
as_tool(),到底什么区别?as_tool():任务委托。主 Agent 调用子 Agent 就像调普通函数,子 Agent 只拿到本次调用生成的输入,完成任务后把结果返回给主 Agent,控制权始终在主 Agent 手里4。as_tool() = 让别人帮你做一个子任务,结果交回来你继续开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)as_tool() 暴露给 orchestrator,orchestrator 决定调用时机与顺序,结果链式传给 synthesizer。整个控制权始终在 orchestrator 手里——这正是 Manager 模式的精髓5。asyncio.gather 真正提速的姿势
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}")asyncio.gather 就能把总耗时从「串行之和」压缩到「最慢那个单任务的耗时」1。# 代码驱动串行示例:大纲 → 质量评估 → 按条件决定是否写故事
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_outputif 判断由 Python 执行,不是 LLM 在推理6。这意味着行为百分之百可预期,也百分之百可测试。RunContextWrapper 的双轨设计——本地 Context 对 LLM 不可见,只在工具函数内流转。多 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 用——上下文全丢as_tool()。开发者发现 pricing agent「不知道用户说了什么」,排查半天才明白——as_tool() 的子 Agent 只拿到工具调用时构造的输入,不是完整对话历史。as_tool()。判断标准是「子 Agent 需要知道之前发生了什么吗?」asyncio.gatherasyncio.gather 并发,B 会拿到空数据或者 None。结果不是报错,而是 B 「安静地」输出了一个毫无意义的结果——这种 bug 很难发现。gather。max_turnsmax_turns 是 109。生产环境里千万别依赖默认值。根据任务的实际步骤上限,显式设置一个合理的 max_turns,否则一个出问题的 orchestrator 会不停消耗 token 直到被外部超时机制终止。as_tool() 给独立子任务,Handoff 给需要上下文继承的场景。这条规则基本能避免 90% 的上下文问题。判断方法就一个:「子任务完成后,控制权需要回到主 Agent 吗?」需要回来就用 as_tool(),不需要就用 Handoff。gather。花 5 分钟在白板上画出各 Agent 之间的数据流,然后对着图写代码,比事后排查「为什么并发结果不对」省出几倍时间。Vercel 工程团队砍掉 80% 工具数量后可靠性反而大幅提升10——越少的节点,越清晰的拓扑,越不容易出问题。RunResult 对象里到底装了什么?final_output、new_items、to_input_list() 之间是什么关系?为什么有时候取到空值?
围绕这条内容继续补充观点或上下文。