本文是《从零到上线:Agent 应用开发完整路线图》系列第四篇。前三篇分别讲了整体路线、工具调用设计、记忆与 RAG,这一篇进入多 Agent 编排——也是这个系列里最容易被滥用的一站。
先说结论:多 Agent 不是”更强的 Agent”,而是用额外的工程复杂度,换取任务拆解的清晰度。如果你的任务一个 Agent 配几个工具就能搞定,引入多 Agent 架构通常只会让系统更难调试、更贵、更慢。这篇文章讲清楚什么时候真的需要它,以及需要时怎么搭。
一、先问自己:真的需要多 Agent 吗
判断标准很具体,不是”任务复杂”就需要多 Agent,而是看这个任务有没有以下特征:
- 子任务之间关注点完全不同,且不需要共享细节上下文(比如”检索资料”和”控制输出格式”是两件互不相关的事)
- 子任务可以并行执行,串行做明显浪费时间
- 不同子任务需要不同的工具集、不同的 system prompt 风格,混在一个 Agent 里会互相干扰(比如一个要写诗,一个要写精确的 SQL,语气和约束完全不同)
如果你的任务达不到这几条,先回到单 Agent + 工具调用的方案——这是第一篇文章反复强调的事,这里再强调一次是因为多 Agent 框架的文档往往把它讲得很诱人,容易让人觉得”看起来更专业”,但额外的复杂度是要还的:更多的中间状态要追踪,出错时定位问题的链路更长。
二、Supervisor 模式:最常见、最该先学的模式
Supervisor 模式的结构很直白:一个主 Agent(Supervisor)负责理解整体任务、拆解成子任务、调度合适的子 Agent 去执行,再把结果汇总。
┌─────────────┐
用户任务 ──> │ Supervisor │
└──────┬──────┘
│ 拆解任务,决定调用谁
┌──────────────┼──────────────┐
▼ ▼ ▼
┌──────────┐ ┌──────────┐ ┌──────────┐
│ 研究 Agent│ │ 写作 Agent│ │ 审核 Agent│
└──────────┘ └──────────┘ └──────────┘
│ │ │
└──────────────┴──────────────┘
│ 汇总结果
▼
最终输出
实现上,子 Agent 本质上就是 Supervisor 手里的”工具”——区别只是这个工具内部不是一个简单函数,而是另一个完整的 LLM 调用循环。这个理解能帮你少踩很多坑:你不需要学一套全新的心智模型,把第一篇文章里那个最小 Agent 循环,套上”子 Agent 作为工具”这一层就够了。
import anthropic
client = anthropic.Anthropic()
# ---- 子 Agent 1:研究员,只负责检索信息,不关心最终格式 ----
def research_agent(topic: str) -> str:
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=800,
system="你是研究助理,只负责收集和整理事实信息,不需要考虑最终呈现格式。",
messages=[{"role": "user", "content": f"收集关于'{topic}'的关键信息要点"}]
)
return response.content[0].text
# ---- 子 Agent 2:写作员,只负责把素材写成指定风格的文章 ----
def writing_agent(research_notes: str, style: str) -> str:
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1000,
system=f"你是文案撰写者,根据提供的素材写成{style}风格的内容,不需要自己查资料。",
messages=[{"role": "user", "content": f"根据以下素材撰写文章:\n{research_notes}"}]
)
return response.content[0].text
# ---- Supervisor:把子 Agent 当工具调用 ----
supervisor_tools = [
{
"name": "research",
"description": "派研究员收集某个主题的关键信息,适用于需要事实素材的场景",
"input_schema": {
"type": "object",
"properties": {"topic": {"type": "string"}},
"required": ["topic"]
}
},
{
"name": "write",
"description": "派写作员根据已有素材撰写指定风格的文章,必须先有素材才能调用",
"input_schema": {
"type": "object",
"properties": {
"research_notes": {"type": "string"},
"style": {"type": "string", "description": "如:轻松幽默、严肃正式、技术教程"}
},
"required": ["research_notes", "style"]
}
}
]
def execute_supervisor_tool(name: str, tool_input: dict) -> str:
if name == "research":
return research_agent(tool_input["topic"])
elif name == "write":
return writing_agent(tool_input["research_notes"], tool_input["style"])
return f"未知工具:{name}"
def run_supervisor(task: str, max_turns: int = 6) -> str:
messages = [{"role": "user", "content": task}]
system_prompt = (
"你是任务调度员。先派研究员收集素材,再派写作员根据素材撰写最终内容。"
"不要自己直接撰写,必须通过工具调度子 Agent 完成。"
)
for _ in range(max_turns):
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
system=system_prompt,
tools=supervisor_tools,
messages=messages
)
if response.stop_reason != "tool_use":
return response.content[0].text
messages.append({"role": "assistant", "content": response.content})
tool_results = []
for block in response.content:
if block.type == "tool_use":
result = execute_supervisor_tool(block.name, block.input)
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": result
})
messages.append({"role": "user", "content": tool_results})
return "达到最大轮次仍未完成"
if __name__ == "__main__":
result = run_supervisor("写一篇关于'远程办公效率工具'的轻松幽默风格短文")
print(result)
这个例子刻意控制了子 Agent 的边界:研究员不知道最终要写成什么风格,写作员不需要自己查资料。这是 Supervisor 模式真正的价值——子任务之间互不污染上下文,各自的 system prompt 可以写得很纯粹,不用照顾”既要又要”的复杂指令。
三、并行子任务:别让独立的工作互相等待
如果几个子任务彼此没有依赖关系,串行调用纯粹是浪费时间。比如”调研三个互不相关的竞品”,完全可以同时发起。
import concurrent.futures
def research_multiple_topics(topics: list[str]) -> dict[str, str]:
"""并行执行多个互不依赖的研究任务"""
results = {}
with concurrent.futures.ThreadPoolExecutor(max_workers=len(topics)) as executor:
future_to_topic = {
executor.submit(research_agent, topic): topic
for topic in topics
}
for future in concurrent.futures.as_completed(future_to_topic):
topic = future_to_topic[future]
try:
results[topic] = future.result()
except Exception as e:
results[topic] = f"研究失败:{str(e)}"
return results
if __name__ == "__main__":
topics = ["竞品A的定价策略", "竞品B的定价策略", "竞品C的定价策略"]
results = research_multiple_topics(topics)
for topic, notes in results.items():
print(f"=== {topic} ===\n{notes}\n")
注意这里每个任务的失败被单独捕获,不会因为一个子任务出错就让整批任务全部失败——这是并行设计里容易被忽略的细节,生产环境里任何一次外部调用都可能超时或出错,必须假设它会失败。
判断该不该并行很简单:画一下任务依赖图,没有箭头连接的节点,就是可以并行的候选。
四、Human-in-the-loop:关键节点暂停等确认
涉及不可逆操作(发邮件、改数据库、花钱)的场景,不该让 Agent 自主决定执行,而要在执行前暂停,等人工确认。
PENDING_CONFIRMATION = "需要用户确认后才能执行的操作"
def execute_with_confirmation(action_name: str, action_detail: str, auto_confirm: bool = False) -> str:
"""
高风险操作执行前的拦截层。
实际项目里 auto_confirm=False 时,应该把这个状态持久化,
等下一次用户消息里包含确认意图时再真正执行。
"""
if not auto_confirm:
return (
f"{PENDING_CONFIRMATION}:{action_name}\n"
f"详情:{action_detail}\n"
f"请回复'确认'以执行,或说明需要调整的地方。"
)
# 真正执行的逻辑放这里
return f"已执行:{action_name}"
这一层拦截看起来简单,但在多 Agent 系统里特别容易被遗漏——因为子 Agent 的输出往往直接被 Supervisor 拿去调用下一个工具,如果没有显式的”高风险操作”标记,很容易在某个子 Agent 的输出里夹带了一个本该被确认、却被自动执行的操作。建议的做法是:在工具描述层面就把”是否需要确认”作为一个显式字段标出来,而不是依赖 Supervisor 临时判断。
五、可观测性:没有它,多 Agent 系统等于黑盒
单 Agent 出错时,看一遍对话历史通常就能定位问题。多 Agent 系统不行——Supervisor 调度了几个子 Agent、每个子 Agent 内部又跑了几轮工具调用,出错时如果没有结构化日志,你只能对着最终输出干瞪眼。
最简单的可观测性方案:给每一层调用打日志,记录输入、输出、耗时。
import time
import json
def logged_call(agent_name: str, func, *args, **kwargs):
start = time.time()
try:
result = func(*args, **kwargs)
status = "success"
except Exception as e:
result = str(e)
status = "error"
duration = time.time() - start
log_entry = {
"agent": agent_name,
"status": status,
"duration_seconds": round(duration, 2),
"input_preview": str(args)[:200],
"output_preview": str(result)[:200],
}
print(json.dumps(log_entry, ensure_ascii=False)) # 实际项目落到文件或日志系统
if status == "error":
raise Exception(result)
return result
# 用法:把每个子 Agent 调用包一层
# result = logged_call("research_agent", research_agent, topic="远程办公工具")
生产项目里这类日志通常会接到专门的追踪系统(比如 LangSmith、Langfuse,或者自建的简单数据库表),但哪怕只是这种最简陋的 print 日志,在排查”为什么 Supervisor 调度错了子 Agent”这类问题时,也比没有强得多。
六、一个常见误区:把多 Agent 当成解决”模型不够聪明”的手段
容易踩的一个坑是:发现单 Agent 在某个复杂任务上表现不好,第一反应是”那就拆成多个 Agent 分工试试”。但很多时候,表现不好的根本原因不是任务该不该拆,而是第二篇文章讲的工具调用设计问题——工具描述不清晰、边界没划好、错误处理缺失。这些问题不会因为换成多 Agent 架构而消失,反而会在更多层级上重复出现,排查起来更费劲。
正确的排查顺序应该是:
- 先确认单 Agent + 清晰工具设计的方案,是不是已经做到位了(参考第二篇的排查清单)
- 真的是任务结构性地需要拆分(关注点不同、可并行、需要不同工具集),才引入多 Agent
- 引入之后,从第一天就配好日志,不要等出问题了才补
七、小结:一张决策表
| 信号 | 该用什么 |
|---|---|
| 单一目标,工具调用清晰,模型能一次性想清楚怎么做 | 单 Agent + 工具调用 |
| 子任务关注点完全不同,各自需要独立的 system prompt | Supervisor + 子 Agent |
| 子任务之间没有依赖关系 | 并行执行 |
| 涉及不可逆/高风险操作 | Human-in-the-loop 拦截 |
| 调度链路超过 2 层,或排查问题靠”猜” | 必须先补可观测性,再继续加复杂度 |
多 Agent 架构解决的是”任务结构”问题,不是”模型能力”问题。先把任务结构想清楚,再决定要不要为它多付出这一层工程复杂度。
下一篇进入系列最后一站——生产化实践:怎么加护栏防止 Agent 做出格的操作、怎么控制和监控成本、以及如何搭一个评估集(eval set),让你每次改 prompt 或换模型之后,能用数据说话而不是凭感觉判断”是不是变好了”。
发表回复