ByteFisher AI 编程实战(二十一):多Agent协作与综合实战

这是本系列的最后一篇。经过 20 篇的学习,你已经掌握了从概念到工具、从编程应用到高级 Agent 的全链路 AI 编程技能。本篇将前面所有知识串联起来,构建一个多 Agent 协作系统。

一、为什么需要多 Agent

一个 Agent 的能力再强,也有边界。多 Agent 系统通过专业分工和协作,突破单 Agent 的局限:

场景 单 Agent 多 Agent
代码审查 生成一段代码 一个写代码,一个审查,一个测试
项目开发 逐步生成 架构师规划→工程师实现→QA 测试
Bug 修复 分析并修复 定位→分析根因→修复→验证
知识问答 基于训练数据回答 检索 Agent + 分析 Agent + 验证 Agent

专业分工的核心逻辑:每个 Agent 有自己独立的角色设定和上下文,不会被其他任务的上下文污染。

二、多 Agent 架构模式

2.1 主管-工人模式(Manager-Worker)

1
2
3
4
5
6
        主管 Agent ← 任务分解、进度协调、质量把控
/ | \
工人A 工人B 工人C ← 各自负责专业领域

适用场景:自动化开发流水线、项目管理
优势:分工明确,可横向扩展工人数量

2.2 流水线模式(Pipeline)

1
2
3
4
需求 → 分析师 → 架构师 → 开发者 → QA → 部署

适用场景:有固定流程的任务
优势:每个阶段产出清晰,便于人工介入审核

2.3 辩论模式(Debate)

1
2
3
4
Agent A(正方) ↔ Agent B(反方) → 裁判 Agent → 综合结论

适用场景:技术选型、架构评审、方案评估
优势:通过观点碰撞减少偏差

三、LangGraph 实战——有状态工作流

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
from langgraph.graph import StateGraph, END
from typing import TypedDict

class WorkflowState(TypedDict):
requirements: str
architecture: str
code: str
tests: str
iteration: int

# 架构师 Agent
def architect_node(state: WorkflowState):
response = llm.invoke(
f"你是一名软件架构师。基于以下需求设计系统架构:\n{state['requirements']}\n"
f"输出:技术栈、模块划分、数据流"
)
return {"architecture": response.content}

# 开发者 Agent
def developer_node(state: WorkflowState):
response = llm.invoke(
f"你是一名高级开发者。基于以下架构实现代码:\n{state['architecture']}\n"
f"需求:{state['requirements']}"
)
return {"code": response.content, "iteration": state.get("iteration", 0) + 1}

# QA Agent
def qa_node(state: WorkflowState):
response = llm.invoke(
f"你是一名 QA 工程师。审查以下代码:\n{state['code']}\n"
f"检查是否有 bug、安全漏洞、性能问题。"
f"如果发现问题,以 'FAIL:' 开头描述。"
)
return {"tests": response.content}

# 条件判断:QA 不通过则退回开发者
def should_continue(state: WorkflowState):
if "FAIL" in state["tests"] and state["iteration"] < 3:
return "developer" # 退回修改(最多重试 3 次)
return END

# 构建工作流图
workflow = StateGraph(WorkflowState)
workflow.add_node("architect", architect_node)
workflow.add_node("developer", developer_node)
workflow.add_node("qa", qa_node)

workflow.set_entry_point("architect")
workflow.add_edge("architect", "developer")
workflow.add_edge("developer", "qa")
workflow.add_conditional_edges("qa", should_continue, {
"developer": "developer",
END: END
})

app = workflow.compile()
result = app.invoke({"requirements": "创建用户注册登录 API,使用 Express + TypeScript"})

四、CrewAI 实战——角色化协作

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
from crewai import Agent, Task, Crew, Process

# 定义角色
architect = Agent(
role="软件架构师",
goal="设计最佳系统架构",
backstory="15 年经验的软件架构师,擅长微服务和 DDD",
verbose=True
)

developer = Agent(
role="高级开发者",
goal="实现高质量、可维护的代码",
backstory="全栈开发者,精通 C# 和 Python",
verbose=True
)

qa = Agent(
role="QA 工程师",
goal="确保代码质量和测试覆盖",
backstory="专注于自动化测试和代码审查",
verbose=True
)

# 定义任务
design_task = Task(
description="设计博客系统架构,包含用户、文章、评论模块",
agent=architect,
expected_output="技术栈和模块划分文档"
)

implement_task = Task(
description="实现博客系统的核心 CRUD API",
agent=developer,
expected_output="完整的 API 代码",
context=[design_task] # 依赖架构设计
)

test_task = Task(
description="为所有 API 端点编写集成测试",
agent=qa,
expected_output="测试代码和执行报告",
context=[implement_task]
)

# 组建团队并执行
crew = Crew(
agents=[architect, developer, qa],
tasks=[design_task, implement_task, test_task],
process=Process.sequential # 按顺序执行
)

result = crew.kickoff()
print("最终输出:", result)

五、综合实战:自动化开发流水线

5.1 流水线架构

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
用户输入(自然语言描述)

PM Agent(需求分析 → 输出 PRD 文档)

架构师 Agent(技术选型 → 输出架构设计文档)

前端 Agent(界面实现 → 输出组件代码)

后端 Agent(API 实现 → 输出路由/服务/模型代码)

QA Agent(测试验证 → 输出测试报告)

部署 Agent(CI/CD 配置 → 输出部署脚本)

输出:完整可部署项目

5.2 关键设计:Agent 间通信协议

多 Agent 系统中,通信协议比各个 Agent 的能力更重要:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
{
"agent": "architect",
"output": {
"tech_stack": {
"frontend": "React 18 + TypeScript + Tailwind",
"backend": "Node.js + Express + Prisma",
"database": "PostgreSQL"
},
"modules": [
{"name": "User", "routes": ["POST /register", "POST /login"]},
{"name": "Post", "routes": ["GET /posts", "POST /posts"]}
],
"data_flow": "Client → API Gateway → Service → Database"
}
}

结构化输出确保:

  1. 下游 Agent 能可靠解析上游输出
  2. 每步可独立验证
  3. 支持人类在中间介入修改

六、多 Agent 的成本考量

多 Agent 虽然功能强大,但成本是实际部署时必须考虑的因素:

成本类型 单 Agent 多 Agent(3 个)
每次任务的 Token 消耗 ~5K ~15K-30K
响应延迟 ~3 秒 ~10-30 秒
API 费用(DeepSeek) ~0.005 元 ~0.02 元
调试复杂度
错误率 中(但调试困难)

6.1 成本优化策略

  1. 先用单 Agent 试:80% 的任务单 Agent 就能完成
  2. 按需增加 Agent:只在需要专业分工的场景引入多 Agent
  3. 减少不必要的中转:Agent 间通信也消耗 Token,减少非必要的信息传递
  4. 设置退出条件:QA 最多重试 3 次,避免死循环
  5. 监控 Token 消耗:在关键节点记录累计 Token 用量

七、多 Agent 系统最佳实践

实践 说明
明确职责 每个 Agent 只做一件事,角色边界清晰
结构化通信 Agent 间输出使用 JSON Schema 定义格式
验证环节 每个 Agent 输出都要验证后再传给下一个
日志追踪 完整记录每个 Agent 的输入输出,便于调试
错误恢复 Agent 失败时可回退到上一步,不至于全盘重来
预算控制 多 Agent 调用成本远高于单次调用,设置上限
人类介入 关键决策点留有人工确认环节

回顾本系列

1
2
3
4
5
6
7
8
9
10
第一阶段(01-04):概念基础——理解 AI 编程的本质
→ Token、Transformer、Prompt 五要素、开发环境
第二阶段(05-07):环境工具——掌握三大主流工具
→ Copilot、Cursor、Claude Code
第三阶段(08-13):编程应用——全场景实战
→ 前端、后端、Unity、代码审查、调试
第四阶段(14-17):扩展应用——写作、数据、自动化、RAG
→ 技术写作、数据分析、自动化工作流、知识库
第五阶段(18-21):高级进阶——Agent、MCP、Skill、多Agent
→ 自主 Agent、标准化工具接口、可复用 Skill、团队协作

AI 编程不是让你不再需要编程,而是让你编程的方式彻底改变——从手写每一行代码,到用自然语言调度 AI 完成更复杂的工作。工具在变,但思考能力、架构能力、审美能力始终是开发者的核心竞争力。

感谢你读到这里。现在,打开你的编辑器,开始实践吧。

ByteFisher
分享编程技术 · 记录钓鱼乐趣
扫码关注
▸ 扫码关注 ◂
分享: