跳到主要内容

单 Agent 不够用?多 Agent 协作的三大模式和五个坑

一个 Agent 承担所有角色——分析需求、写代码、测试——结果什么都做得一般。拆成多个专责 Agent(分析师、设计师、程序员、测试员)后,代码质量提升了一个档次。

多 Agent 协作的真实难点和解决方案。

核心问题:为什么单 Agent 做不好?

我做过一个实验,让单 Agent 和多 Agent 完成同样的任务:"实现一个用户登录功能,要求有验证码、记住我、密码加密"。

单 Agent 的输出

# 它把所有逻辑塞在一个文件里,没有测试
def login(username, password):
if check_password(username, password):
return "success"
return "failed"

问题:

  • 没有需求分析,漏掉了"验证码"和"记住我"
  • 没有设计文档,代码结构混乱
  • 没有测试,安全性没考虑

多 Agent 的输出

Analyst Agent: 需求拆解为 5 个功能点
Designer Agent: 设计了 3 层架构 + 时序图
Coder Agent: 分 3 个文件实现,有类型注解
Tester Agent: 写了 8 个测试用例,覆盖率 92%

原因:每个 Agent 都有自己的"专业领域",专注一件事做得更好

三大协作模式

模式一:顺序流水线(最简单)

适用场景:流程固定,不需要回退。

需求 → 分析师 → 设计师 → 程序员 → 测试员 → 交付

实现:

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END

class DevState(TypedDict):
requirement: str
analysis: str
design: str
code: str
test_result: str
iteration: int

def analyst_node(state: DevState) -> dict:
"""需求分析师:拆解需求"""
requirement = state["requirement"]

prompt = f"""你是需求分析师。分析以下需求,拆解为功能点:

{requirement}

输出格式:
1. 功能点列表
2. 技术要求
3. 风险点"""

analysis = llm.invoke(prompt)
return {"analysis": analysis}

def designer_node(state: DevState) -> dict:
"""架构师:设计方案"""
prompt = f"""根据需求分析设计技术方案:

{state['analysis']}

输出:
1. 模块划分
2. 数据结构
3. 关键代码框架"""

design = llm.invoke(prompt)
return {"design": design}

def coder_node(state: DevState) -> dict:
"""程序员:实现代码"""
prompt = f"""根据设计实现代码:

{state['design']}

要求:
- 完整可运行
- 有类型注解
- 有异常处理"""

code = llm.invoke(prompt)
return {"code": code}

def tester_node(state: DevState) -> dict:
"""测试员:验证代码"""
prompt = f"""测试以下代码:

{state['code']}

输出:
1. 测试用例
2. 发现的问题
3. 修复建议"""

result = llm.invoke(prompt)

# 如果有问题,返回"需要修复"
if "问题" in result and "无问题" not in result:
return {"test_result": result, "needs_fix": True}

return {"test_result": result, "needs_fix": False}

# 构建流程
workflow = StateGraph(DevState)
workflow.add_node("analyst", analyst_node)
workflow.add_node("designer", designer_node)
workflow.add_node("coder", coder_node)
workflow.add_node("tester", tester_node)

workflow.set_entry_point("analyst")
workflow.add_edge("analyst", "designer")
workflow.add_edge("designer", "coder")
workflow.add_edge("coder", "tester")

# 简单的路由:有问题就修,没问题就结束
def route_after_test(state: DevState) -> str:
if state.get("needs_fix") and state.get("iteration", 0) < 3:
return "coder" # 回到程序员修复
return END

workflow.add_conditional_edges("tester", route_after_test, {"coder": "coder", END: END})

模式二:反馈循环(最实用)

适用场景:需要迭代改进,质量要求高。

需求 → 分析 → 设计 → 编码 → 测试
↓ 失败
反馈 → 编码 → 测试

真实案例:jojo-code 的代码生成流程。

class CodeGenState(TypedDict):
requirement: str
code: str
test_result: TestResult
feedback: list[str]
iteration: int

@dataclass
class TestResult:
passed: bool
failures: list[str]
coverage: float

def tester_with_feedback(state: CodeGenState) -> dict:
"""测试并生成具体反馈"""
code = state["code"]

# 1. 运行代码
exec_result = execute_code_safely(code)

# 2. 运行测试
test_result = run_tests(code)

# 3. 如果失败,生成具体反馈
if not test_result.passed:
feedback = analyze_failures(
code=code,
failures=test_result.failures
)
return {
"test_result": test_result,
"feedback": feedback,
}

return {"test_result": test_result}

def analyze_failures(code: str, failures: list[str]) -> list[str]:
"""分析失败原因,给出具体修改建议"""
prompt = f"""代码:
{code}

测试失败:
{failures}

请给出具体的修改建议:
1. 哪一行有问题
2. 应该怎么改
3. 为什么"""

return llm.invoke(prompt)

关键改进:反馈不是笼统的"有问题",而是"第 15 行应该改成..."。

模式三:层级协作(最复杂)

适用场景:大项目,需要"经理"协调。

项目经理 Agent
├── 分析师 Agent
├── 开发组
│ ├── 前端 Agent
│ └── 后端 Agent
└── 测试 Agent

实现:

class ProjectState(TypedDict):
requirement: str
tasks: list[Task] # 拆分的任务
results: dict[str, str] # 各 Agent 的结果
status: str

def manager_agent(state: ProjectState) -> dict:
"""项目经理:拆分任务、分配、验收"""

# 1. 拆分任务
if not state.get("tasks"):
tasks = decompose_requirement(state["requirement"])
return {"tasks": tasks}

# 2. 验收结果
all_done = all(
task.id in state["results"]
for task in state["tasks"]
)

if all_done:
# 合并结果
final = merge_results(state["results"])
return {"status": "completed", "final_result": final}

return {"status": "in_progress"}

def decompose_requirement(requirement: str) -> list[Task]:
"""拆分需求为独立任务"""
prompt = f"""将以下需求拆分为独立任务:

{requirement}

每个任务应该:
- 有明确的输入输出
- 可以独立完成
- 任务之间依赖关系清晰"""

tasks = llm.invoke(prompt)
return parse_tasks(tasks)

实际效果:用户输入"做一个用户系统",经理拆分为:

  • 任务 1:数据库设计(后端 Agent)
  • 任务 2:API 开发(后端 Agent)
  • 任务 3:前端页面(前端 Agent)
  • 任务 4:测试用例(测试 Agent)

我踩过的五个坑

坑一:状态污染

问题:Agent A 写了很多中间变量到状态,Agent B 看晕了。

# 错误示范
def analyst_agent(state):
return {
"analysis": "...",
"temp_1": "...", # 临时变量
"temp_2": "...", # 也存进去了
"debug_info": "...", # 调试信息
}

解决:每个 Agent 只返回自己的核心输出。

def analyst_agent(state):
analysis = do_analysis(state["requirement"])
# 临时变量留在函数内部,不污染状态
return {"analysis": analysis} # 只有核心输出

坑二:无限循环

问题:测试失败返回给程序员,程序员改完又失败,无限循环。

# 错误:没有退出条件
def route_after_test(state):
if state["test_result"].passed:
return END
return "coder" # 可能无限循环

解决:加迭代上限。

MAX_ITERATIONS = 3

def route_after_test(state):
if state["test_result"].passed:
return END
if state.get("iteration", 0) >= MAX_ITERATIONS:
return END # 强制结束
return "coder"

坑三:角色重叠

问题:两个 Agent 职责模糊,互相覆盖。

Coder Agent: 我来写登录逻辑
Designer Agent: 我也写了一份登录逻辑
结果:两份代码冲突

解决:明确边界。

CODER_PROMPT = """你是程序员,只负责:
- 根据设计实现代码
- 不修改设计
- 有问题反馈给设计师"""

DESIGNER_PROMPT = """你是设计师,只负责:
- 设计架构和数据结构
- 不写实现代码
- 设计完成后交给程序员"""

坑四:上下文丢失

问题:Agent A 做了决策,Agent B 不知道。

Analyst: 用户要求用 PostgreSQL
Coder: [用 MySQL 实现] # 不知道分析师的决策

解决:关键决策写入状态。

def analyst_agent(state):
analysis = analyze(state["requirement"])

# 关键决策单独存储
decisions = {
"database": "PostgreSQL",
"framework": "FastAPI",
"test_framework": "pytest",
}

return {
"analysis": analysis,
"decisions": decisions, # 后续 Agent 都能看到
}

def coder_agent(state):
# 读取决策
db = state["decisions"]["database"] # PostgreSQL
# 按决策实现...

坑五:工具冲突

问题:多个 Agent 同时写同一个文件。

Coder A: 写入 main.py
Coder B: 也写入 main.py
结果:后者覆盖前者

解决:加锁或分工。

# 方案一:文件锁
import fcntl

def safe_write(path: str, content: str):
with open(path, 'w') as f:
fcntl.flock(f, fcntl.LOCK_EX) # 加锁
f.write(content)
fcntl.flock(f, fcntl.LOCK_UN) # 解锁

# 方案二:明确分工
CODER_A_FILES = ["user.py", "auth.py"]
CODER_B_FILES = ["product.py", "order.py"]

真实案例:jojo-code 的多 Agent 架构

jojo-code 实际用了一个简化版:

class JojoCodeState(TypedDict):
messages: Annotated[list, merge_lists]
tool_calls: list
tool_results: list
is_complete: bool
iteration: int

# 实际只有两个"角色"
# 1. thinking_node: 决策(相当于分析师+设计师)
# 2. execute_node: 执行(相当于程序员)

def thinking_node(state):
"""思考节点:决策下一步做什么"""
# 读取历史
messages = state["messages"]

# 决策
response = llm_with_tools.invoke(messages)

# 如果有工具调用,交给执行节点
if response.tool_calls:
return {
"tool_calls": response.tool_calls,
"is_complete": False,
}

# 否则完成任务
return {"is_complete": True}

def execute_node(state):
"""执行节点:运行工具"""
results = []
for tc in state["tool_calls"]:
result = execute_tool(tc["name"], tc["args"])
results.append(result)

return {"tool_results": results, "tool_calls": []}

为什么简化?因为 jojo-code 是 CLI 工具,不需要太复杂的分工。够用就好,不要过度设计

什么时候该用多 Agent?

场景单 Agent多 Agent
任务简单(问答、查询)❌ 杀鸡用牛刀
质量要求高(生产代码)✅ 每个环节专人把关
流程固定(生成报告)⚠️ 可选
需要迭代(代码审查)✅ 反馈循环改进
团队协作(大项目)✅ 经理协调分工

我的建议:先单 Agent 跑通,质量不够再拆多 Agent。

下一步行动

  1. 评估你的任务:有明显的阶段划分吗?(分析→设计→编码→测试)
  2. 从顺序模式开始:先跑通最简单的流水线
  3. 逐步加反馈:质量不够再加反馈循环
  4. 避免过度设计:够用就好,不要一上来就搞层级架构

多 Agent 不是为了炫技,是为了让每个环节都专业。但专业是有代价的——更多的 Agent、更多的状态管理、更多的调试。


一个 Agent 做所有事,就像一个人又当厨师又当服务员又当收银员,效率和 quality 都有限。多 Agent 就是分工协作,每个人专注自己的领域。但分工的前提是:任务复杂到值得分工。