【学习笔记】多 Agent 编排实战——从单兵到团队作战-11/15
前面两篇解剖了 Claude Code 和 OpenAI Codex 的 Harness 架构。它们是单 Agent 的极限——用一个 Agent 把事情做到最好。
这篇说的是另一个方向:用多个 Agent 完成一个 Agent做不到的事。
一、单 Agent 的结构性上限
先搞清楚为什么需要多 Agent。不是功能不够,是结构性限制:
1.1 上下文窗口:再大的窗口也有上限。一个处理 100 万行代码库的任务,不可能把所有相关代码一次性塞进去。
1.2 线性执行:单 Agent 一次只做一件事。代码生成、测试生成、安全审查——这三件事逻辑上可以并行,但单 Agent 只能顺序来。
1.3 专业化边界:一个通用 Agent 在所有领域都是「够用」,但在某些专业任务上,一个专门调校的 Agent 会显著更好——写代码的、审查代码的、写测试的,训练数据和系统提示可以各自优化。
1.4 故障范围:单 Agent 出错,整个任务失败。多 Agent 可以做故障隔离——某个子任务失败,不影响其他并行进行的子任务。
这是结构性问题,不是调参能解决的。多 Agent 是应对这类问题的架构选择。
二、三大编排模式
当前生产中主要有三种多 Agent 编排模式,各自有适用场景。
2.1 模式一:图编排(LangGraph)
LangGraph 的核心抽象是显式状态机:你定义节点(Agent 或工具)、边(跳转条件)、和全局状态对象。整个工作流是一张有向图,每次执行都经过确定性的路径。
核心特点:
- 状态持久化:每个节点执行完,状态自动写入检查点。任务中断后可以从检查点恢复,不用从头来。
- 条件分支:可以根据上一步的输出决定下一步走哪条路径——测试通过走 A,测试失败走 B 进入修复循环。
- 循环支持:原生支持循环结构,适合「执行 → 验证 → 修复 → 再执行」这类迭代工作流。
- 确定性可调试:因为是显式状态机,执行路径可预期、可追踪、可重放。
适用场景:需要精确控制流程、有复杂分支逻辑、需要跨任务状态持久化的工作流。典型案例:多步骤代码生成流水线、带人工审批节点的自动化流程。
学习曲线:中等。状态机概念不难理解,但图结构的设计需要一定经验。
2.2 模式二:角色编排(CrewAI)
CrewAI 的核心抽象是角色和任务:你定义一组 Agent,每个 Agent 有角色(Role)、目标(Goal)、背景故事(Backstory);然后定义一系列任务,分配给合适的 Agent,最后以 Crew 的形式运行。
from crewai import Agent, Task, Crew
# 定义 Agent 角色
code_writer = Agent(
role="高级 Python 工程师",
goal="写出高质量、可测试的 Python 代码",
backstory="10 年 Python 经验,擅长 FastAPI 和异步编程"
)
test_engineer = Agent(
role="QA 工程师",
goal="设计覆盖完整边界条件的测试套件",
backstory="专注于测试驱动开发和集成测试"
)
security_reviewer = Agent(
role="安全工程师",
goal="发现代码中的安全漏洞和风险",
backstory="专注于 OWASP Top 10 和供应链安全"
)
# 定义任务
write_code = Task(
description="实现用户认证模块,包括 JWT 生成和验证",
agent=code_writer,
expected_output="完整的 Python 模块,包含所有接口"
)
write_tests = Task(
description="为认证模块编写完整的测试套件",
agent=test_engineer,
context=[write_code] # 依赖上一个任务的输出
)
security_review = Task(
description="审查认证模块的安全性,关注 JWT 密钥管理和注入风险",
agent=security_reviewer,
context=[write_code] # 与测试并行
)
# 组建 Crew 并运行
crew = Crew(
agents=[code_writer, test_engineer, security_reviewer],
tasks=[write_code, write_tests, security_review],
verbose=True
)
result = crew.kickoff()
核心特点:
- 角色化系统提示:每个 Agent 的 Role/Goal/Backstory 自动构成系统提示,专业化程度高。
- 任务依赖管理:通过
context参数声明任务间依赖,自动处理执行顺序和数据传递。 - Human-in-the-loop:可以在任意任务节点插入人工审批。
- 工具共享:Agent 可以共享工具(文件读写、搜索、代码执行),也可以各自独立。
生产数据:CrewAI 目前是最广泛部署的多 Agent 框架——60%+ 财富 500 强在用,2025 年 Q3 记录了超过 11 亿次 Agent 操作。生产就绪度最高。
适用场景:需要明确角色分工的创意或专业工作流——写作、研究、代码审查、产品分析。
学习曲线:低。角色和任务的概念直觉易懂,上手快。
2.3 模式三:会话编排(AutoGen/AG2)
AutoGen(现在由社区以 AG2 继续维护,Microsoft 内部演进为 MAF——将 AutoGen 和 Semantic Kernel 合并)的核心抽象是多轮对话:多个 Agent 像人类团队开会一样,通过对话协作解决问题。
核心机制是 GroupChat——多个 Agent 在一个共享对话里轮流发言,由一个 GroupChatManager 决定下一个发言的 Agent:
import autogen
config_list = [{"model": "claude-opus-4-7", "api_key": "..."}]
# 定义 Agent
coder = autogen.AssistantAgent(
name="Coder",
system_message="你是一个 Python 专家。写出简洁、高效的代码。",
llm_config={"config_list": config_list}
)
reviewer = autogen.AssistantAgent(
name="CodeReviewer",
system_message="你是代码审查专家。从可读性、性能、安全三个维度审查代码,给出具体修改建议。",
llm_config={"config_list": config_list}
)
user_proxy = autogen.UserProxyAgent(
name="UserProxy",
human_input_mode="NEVER", # 自动执行代码
code_execution_config={"work_dir": "./workspace"}
)
# 启动对话
groupchat = autogen.GroupChat(
agents=[user_proxy, coder, reviewer],
messages=[],
max_round=10
)
manager = autogen.GroupChatManager(groupchat=groupchat)
user_proxy.initiate_chat(
manager,
message="实现一个异步 Redis 连接池,支持自动重连和连接健康检查"
)
核心特点:
- 动态协作:Agent 可以根据对话内容动态决定下一步——不是预先定义好的流程,而是涌现出来的协作。
- 代码执行集成:UserProxy Agent 可以直接执行代码,结果返回到对话,形成「写代码 → 执行 → 看结果 → 修改」的自然循环。
- 灵活终止:可以配置对话在满足条件时自动终止(比如代码测试通过)。
适用场景:探索性任务、需要动态决策的复杂问题、研究型任务。不适合需要严格控制流程的生产工作流。
学习曲线:中等偏高。会话式协作的不确定性比图状态机和角色编排更难调试。
三、三种模式的横向对比
| 维度 | LangGraph | CrewAI | AutoGen/AG2 |
|---|---|---|---|
| 控制精度 | 最高(状态机) | 中(任务依赖) | 最低(对话涌现) |
| 上手难度 | 中 | 低 | 中高 |
| 适合任务 | 流程确定、需要断点续传 | 角色分工清晰 | 探索性、动态协作 |
| 生产就绪度 | 高 | 最高 | 中 |
| 调试难度 | 低(可追踪状态) | 中 | 高(非确定性) |
| 成本控制 | 精确 | 中等 | 较难 |

3.1 生产混合模式:CrewAI 外层 + AutoGen 推理子任务
真实的生产工作流往往不是纯粹用一种模式。一个常见的混合方案:
CrewAI 做外层编排——定义整体工作流的角色分工和任务流转,结构清晰、便于维护。
AutoGen 做推理子任务——当某个任务需要动态的多轮推理(比如调试一个复杂 bug,需要多个专家视角来回讨论),在那个节点启动一个 AutoGen GroupChat,完成后把结果返回给 CrewAI 流程。
这个组合的逻辑:外层用最稳定的结构,内层在需要的地方引入灵活性。不是为了混合而混合,是在合适的地方用合适的工具。
LangGraph 在这个组合里也可以充当外层——它的检查点机制让整个流程更健壮,适合需要跨会话持久化的长程任务。
3.2 Harness 在多 Agent 中的特殊挑战
多Agent 不是把几个单 Agent 拼在一起这么简单。Harness 层面有几个单 Agent 不存在的挑战:
3.2.1 挑战一:Agent 间通信协议
单Agent 里没有「通信」这个问题——只有一个 Agent。多 Agent 里,Agent A 的输出是 Agent B 的输入,格式不一致就会出问题。
实践建议:定义清晰的消息格式——通常是结构化的 JSON 或 Pydantic 模型。明确每个 Agent 的输入输出 schema,不要依赖自然语言描述来传递结构化数据。
from pydantic import BaseModel
from typing import Optional
class CodeReviewResult(BaseModel):
passed: bool
issues: list[str]
severity: str # "low" | "medium" | "high" | "critical"
suggested_fixes: Optional[list[str]] = None
# Reviewer Agent 的输出必须符合这个 schema
# Downstream Agent 可以可靠地处理
3.2.2 挑战二:共享状态管理
多个Agent 同时写入同一个状态存储,会出现竞争条件。特别是在并行执行的场景——两个 Agent 同时更新同一个任务状态,最后写入的覆盖了前面的。
LangGraph 通过 State Schema 和 Reducer 函数解决这个问题:每个状态字段可以定义合并规则(追加列表而不是覆盖、取最大值等)。
from langgraph.graph import StateGraph
from typing import Annotated
import operator
class AgentState(TypedDict):
# 用 operator.add 作为 reducer,追加而不是覆盖
messages: Annotated[list[str], operator.add]
review_results: Annotated[list[dict], operator.add]
is_complete: bool
3.2.3 挑战三:故障隔离与级联防护
Agent B失败时,不应该让整个流程停下来,更不应该让 Agent A 和 Agent C 因此崩溃。
基本原则:每个 Agent 在独立的执行环境里运行(独立的上下文,独立的工具访问),失败只影响自己的输出,上游处理者决定是重试、跳过还是报错。
LangGraph 的检查点机制在这里很有价值:某个节点失败,可以从最近的检查点重试,不用从头开始。
3.2.4 挑战四:成本放大
这是多 Agent 最容易被忽视的问题。
单Agent 完成某个任务花了 5000 tokens。三个 Agent 并行处理同一个任务的不同部分,理论上每个花 2000 tokens,总共 6000——比单 Agent 还多一点。
但现实中往往不是这样。每个 Agent 都有完整的系统提示(假设各 1000 tokens),都需要在上下文里包含共享背景(假设 1500 tokens),各自的任务上下文再加 1000 tokens——三个 Agent 的成本很快就是单 Agent 的 3-5 倍,而不是期望的 0.7 倍。
应对策略:
- 共享系统提示:把所有 Agent 共用的背景信息提取出来,通过 Prompt Caching 只计算一次
- 轻量子 Agent:子任务 Agent 尽量用小模型(Claude Haiku、gpt-4o-mini),只有关键节点用大模型
- 限制 Agent 数量:不是越多越好,3-5 个专注的 Agent 往往比 10 个泛化 Agent 效果更好且成本更低
------------------------------

------------------------------
四、Anthropic 的子 Agent 模式
Anthropic 在自己的 Harness 文档里推荐的模式相对保守,但可靠性高:
4.1 独立上下文:每个子 Agent 在自己干净的上下文窗口里运行,不共享主 Agent 的历史对话。这避免了上下文污染,也让每个子 Agent 的成本可预期。
4.2 轻量派遣:主 Agent 给子 Agent 的指令应该是清晰、完整的——子 Agent 不应该需要回来问主 Agent「你的意思是……?」。主 Agent 在派遣时就把子任务所需的全部信息打包好。
4.3 结果汇总:子 Agent 完成后,结果写回共享的文件系统(不是直接传回主 Agent 的上下文),主 Agent 按需读取需要的部分。这个模式避免了大量结果直接涌入主 Agent 上下文导致的上下文膨胀。
主 Agent 上下文
↓ 派遣(完整指令 + 所需上下文)
子 Agent 1 | 子 Agent 2 | 子 Agent 3
(独立上下文)(独立上下文)(独立上下文)
↓ 写入文件系统
主 Agent 按需读取 → 汇总决策
这个模式没有 CrewAI 或 LangGraph 那么「炫」,但在生产中非常稳定。Claude Code 内部的 Task 工具就是这个模式的实现。
五、实战:三 Agent 代码审查流水线
把前面讲的东西落地——用 CrewAI 构建一个「代码生成 → 测试生成 → 安全审查」三 Agent 流水线。
5.1 场景设定
输入:一个功能需求描述(「实现带速率限制的用户登录接口」)
输出:完整代码 + 测试套件 + 安全审查报告
三个 Agent 角色:
- CodeWriter:根据需求写实现代码
- TestEngineer:根据代码写测试套件
- SecurityReviewer:审查代码的安全性
5.2 关键设计决策
任务依赖:TestEngineer 和 SecurityReviewer 都依赖 CodeWriter 的输出,但两者之间没有依赖,可以并行。
输出格式标准化:每个 Agent 的输出指定明确的格式,方便下游处理:
write_code_task = Task(
description="""
实现用户登录接口,要求:
1. 支持邮箱 + 密码认证
2. 速率限制:每个 IP 每分钟最多 5 次失败尝试
3. 成功返回 JWT token(有效期 24h)
4. 失败锁定逻辑:连续 10 次失败锁定 30 分钟
输出格式:
- 完整的 Python 代码(FastAPI 路由)
- 所有依赖的数据结构定义
- 简短的实现说明(不超过 200 字)
""",
agent=code_writer,
expected_output="Python 代码 + 数据结构 + 实现说明"
)
write_tests_task = Task(
description="""
为上一步的登录实现编写完整测试套件。
必须覆盖:
- 正常登录成功路径
- 密码错误的失败路径
- 速率限制触发和重置
- 账户锁定和解锁
- JWT token 格式验证
使用 pytest,mock 外部依赖(Redis、数据库)。
""",
agent=test_engineer,
context=[write_code_task]
)
security_review_task = Task(
description="""
从安全角度审查登录接口实现。
重点关注:
- JWT 密钥管理(硬编码风险?)
- 速率限制绕过可能性
- SQL 注入风险
- 时序攻击风险(密码比较是否使用常数时间比较?)
- 日志中是否泄露敏感信息
输出格式:
风险等级(Low/Medium/High/Critical)
发现的问题列表(每条包含:问题描述 + 代码位置 + 修复建议)
""",
agent=security_reviewer,
context=[write_code_task] # 不依赖 test_engineer,可并行
)
5.3 Harness 层的关键配置
5.3.1 工具沙箱:测试 Agent 需要执行代码,要配置安全的执行环境——不能让 Agent 随便执行系统命令或访问生产数据库:
# 测试执行工具的沙箱配置
code_execution_config = {
"work_dir": "./sandbox", # 独立工作目录
"use_docker": True, # Docker 容器隔离
"timeout": 60, # 最大执行时间
"allowed_imports": ["pytest", "fastapi", "pydantic", "unittest.mock"]
}
5.3.2 成本控制:三个 Agent 用不同规模的模型:
code_writer = Agent(
role="高级 Python 工程师",
llm=LLM(model="claude-sonnet-4-6") # 代码生成用 Sonnet
)
test_engineer = Agent(
role="QA 工程师",
llm=LLM(model="claude-haiku-4-5-20251001") # 测试生成用 Haiku(更便宜)
)
security_reviewer = Agent(
role="安全工程师",
llm=LLM(model="claude-opus-4-7") # 安全审查用 Opus(最关键)
)
5.3.3 故障处理:加上基本的重试逻辑和失败后降级:
crew = Crew(
agents=[code_writer, test_engineer, security_reviewer],
tasks=[write_code_task, write_tests_task, security_review_task],
process=Process.sequential, # 先写代码,再并行跑测试和审查
max_retries=2, # 失败最多重试 2 次
verbose=True
)

六、什么时候不需要多 Agent
最后说一个反向思考:多 Agent 不总是正确答案。
6.1 任务简单时:如果一个任务 5 分钟内可以用单 Agent 完成,搭一套多 Agent 编排框架的工程成本是不合算的。
6.2 顺序依赖强时:如果任务 B 严重依赖任务 A 的完整输出才能开始,并行化没有收益,反而增加协调成本。
6.3 调试成本高时:多 Agent 系统的调试比单 Agent 复杂得多——状态传递、时序依赖、非确定性问题都会增加排查难度。如果团队还没有成熟的 Agent 可观测性基础设施(下一篇的主题),引入多 Agent 可能带来更多麻烦。
6.4 成本是硬约束时:多 Agent 的成本放大效应不能忽视。先算清楚多 Agent 架构下每次运行的预期成本,再决定是否值得。
实用原则:先用单 Agent 把问题解决,遇到了明确的结构性瓶颈(上下文溢出、强并行需求、明显的专业化收益)再引入多 Agent。不要为了「看起来更高级」而做架构升级。
下一篇:Agent 可观测性——追踪、指标、日志,以及当你的多 Agent 系统开始在生产中出奇怪的问题时,怎么看见它在想什么。
参考文献:
更多推荐



所有评论(0)