【AI应用开发】 Agent篇(五):多 Agent 协作与生产落地
·
【AI应用开发】 Agent篇(五):多 Agent 协作与生产落地
前言
前四篇我们构建了一个功能完整的单 Agent。但现实世界的复杂任务往往需要多个专业 Agent 分工协作——就像公司里有工程师、设计师、产品经理,各司其职又互相配合。
同时,从 Demo 到生产上线有巨大的鸿沟:部署、监控、评估、成本控制、安全防护……每一个环节处理不好,Agent 就无法真正服务用户。
本文串联多 Agent 协作架构和生产落地关键要点,给整个系列画上完整的工程闭环。
目录
- 为什么需要多 Agent
- 多 Agent 协作模式
- 2.1 顺序流水线(Pipeline)
- 2.2 层级调度(Supervisor-Worker)
- 2.3 自由协作(Collaborative)
- 2.4 辩论模式(Debate)
- CrewAI 多 Agent 实战
- LangGraph 多 Agent 实战
- 生产部署架构
- 监控与评估
- 成本优化
- 安全防护
- 学习路线与展望
1. 为什么需要多 Agent
单 Agent 的天然限制:
单 Agent 困境:
* 一个 Prompt 要覆盖所有能力 → Prompt 越来越臃肿
* 工具列表越来越长 → LLM 选择困难,准确率下降
* 复杂任务容易"迷路" → 缺乏分解和分工
* 缺乏自我校验 → 错误一路走到黑
多 Agent 的核心优势:
┌─────────────┐
│ Supervisor │ ← 管理者:分配任务、仲裁
│ Agent │
└──┬───┬───┬──┘
│ │ │
┌────────────────┘ │ └────────────────┐
▼ ▼ ▼
┌──────────┐ ┌──────────┐ ┌──────────┐
│ 研究员 │ │ 写手 │ │ 审核员 │
│Research │───────→│ Writer │───────→│ Reviewer │
│ Agent │ 资料 │ Agent │ 草稿 │ Agent │
└──────────┘ └──────────┘ └──────────┘
擅长:搜索分析 擅长:写作表达 擅长:质量把控
分工带来专业化:每个 Agent 只需要做好一件事,Prompt 简洁、工具少、准确率高。
2. 多 Agent 协作模式

2.1 顺序流水线(Pipeline)
最直接的模式——A 做完给 B,B 做完给 C:
# 顺序流水线示例
def pipeline_agents(user_input: str) -> str:
"""研究 → 写作 → 审核 三阶段流水线"""
# Agent 1:研究员搜集资料
research_result = research_agent.run(user_input)
# Agent 2:写手基于研究结果写作
draft = writer_agent.run(
f"基于以下资料写一篇报告:\n{research_result}"
)
# Agent 3:审核员检查质量
review_result = reviewer_agent.run(
f"审核以下报告,修改错误并优化表达:\n{draft}"
)
return review_result
# 适用场景:步骤明确、依赖关系清晰的任务
# 优点:简单、可预测
# 缺点:缺乏反馈循环,前面错了后面无法纠正
2.2 层级调度(Supervisor-Worker)
一个 Supervisor 管理多个 Worker,动态分配任务:
# Supervisor-Worker 模式
class SupervisorAgent:
"""管理多个 Worker Agent,动态分配任务"""
def __init__(self, workers: dict):
self.workers = workers # {"研究员": Agent, "写手": Agent, ...}
def run(self, user_input: str) -> str:
context = {"task": user_input, "results": {}, "history": []}
while True:
# Supervisor 分析当前状态,决定下一步
decision = self.llm.decide(
f"""
当前任务:{user_input}
已完成:{context['results']}
可用工人:{list(self.workers.keys())}
请决定:
1. 是否需要继续?如果已完成,输出 FINISH
2. 如果需要继续,下一步派给哪个工人?任务是什么?
"""
)
if decision.get("action") == "FINISH":
break
worker_name = decision["worker"]
task = decision["task"]
# 分配任务给 Worker
result = self.workers[worker_name].run(task)
context["results"][worker_name] = result
context["history"].append({
"worker": worker_name,
"task": task,
"result": result
})
# 汇总所有结果
return self.summarize(context)
# 适用场景:复杂任务,需要动态分配和协调
# 优点:灵活,能根据中间结果调整策略
# 缺点:Supervisor 本身可能出错,单点风险
2.3 自由协作(Collaborative)
Agent 之间自由对话、自主分工——像人类小组讨论:
# 自由协作模式(AutoGen 风格)
def collaborative_agents(user_input: str) -> str:
"""Agent 之间自由对话协作"""
messages = [{"role": "user", "content": user_input}]
agents = [research_agent, coder_agent, reviewer_agent]
for round_num in range(10):
# 每个 Agent 都可以发言
for agent in agents:
should_speak = agent.should_respond(messages)
if should_speak:
response = agent.respond(messages)
messages.append({
"role": "agent",
"name": agent.name,
"content": response
})
print(f"🤖 [{agent.name}]: {response[:80]}...")
# 检查是否达成共识
if agent.consensus_reached(messages):
break
return messages[-1]["content"]
# 适用场景:创造型任务(头脑风暴、方案设计)
# 优点:最灵活,可能出现意外的好方案
# 缺点:难以控制,可能跑偏
2.4 辩论模式(Debate)
两个 Agent 扮演对立角色,通过辩论找到更好的答案:
辩论模式流程:
正方 Agent:"方案A更好,因为..."
反方 Agent:"方案A有问题:1... 2... 我提议方案B..."
正方 Agent:"反驳:方案B的缺陷是... 但方案A可以改进为..."
裁判 Agent:"综合双方观点,最终建议..."
适用场景:需要多角度分析的决策任务
优点:减少偏见,发现盲点
缺点:成本高(多次 LLM 调用),可能陷入死循环
3. CrewAI 多 Agent 实战
CrewAI 是目前最流行的多 Agent 框架之一,用"角色扮演"的方式定义 Agent:
"""
完整的 CrewAI 多 Agent 示例:
建立一个内容创作团队——研究员、写手、审核员协作完成一篇技术文章
"""
from crewai import Agent, Task, Crew, Process
# ============================================================
# 定义 Agent —— 像写角色卡一样
# ============================================================
researcher = Agent(
role="技术研究员",
goal="深度调研指定技术主题,搜集最新、最权威的资料",
backstory="你是资深技术研究员,擅长从海量信息中提取关键见解。搜索资料时注重权威性和时效性。",
tools=[web_search_tool, arxiv_tool],
verbose=True
)
writer = Agent(
role="技术写手",
goal="将研究资料转化为易读、有深度的技术文章",
backstory="你是技术专栏作家,擅长用通俗语言解释复杂概念。文章结构清晰、举例生动。",
verbose=True
)
reviewer = Agent(
role="技术审核",
goal="审核文章的技术准确性、可读性和完整性",
backstory="你是资深技术编辑,对技术文章的准确性有近乎苛刻的要求。发现错误绝不放水。",
verbose=True
)
# ============================================================
# 定义任务 —— 每个 Agent 分到什么活
# ============================================================
research_task = Task(
description="""调查以下主题,搜集最新资料:
1. AI Agent 在 2025-2026 年的最新进展
2. 主流 Agent 框架对比(LangGraph, CrewAI, AutoGen, OpenAI Agents SDK)
3. 企业级 Agent 落地的典型案例
输出一份结构化的研究报告。""",
expected_output="一份包含摘要、关键发现、框架对比表、案例列表的研究报告",
agent=researcher
)
writing_task = Task(
description="""基于研究员提供的报告,撰写一篇面向开发者的技术博文:
- 标题吸引人
- 开头有场景化引入
- 正文包含至少 3 个实用代码示例
- 结尾有总结和选型建议
- 总字数 3000-4000 字""",
expected_output="一篇完整的技术博文 Markdown 格式",
agent=writer
)
review_task = Task(
description="""审核写手完成的文章:
- 技术内容是否准确?有无法事实性错误?
- 代码是否能运行?示例是否合理?
- 行文逻辑是否清晰?可读性如何?
- 逐条列出需要修改的地方,并给出修改后的版本""",
expected_output="审核意见(逐条修改建议)+ 修改后的最终版本",
agent=reviewer
)
# ============================================================
# 组建 Crew —— 团队开工
# ============================================================
crew = Crew(
agents=[researcher, writer, reviewer],
tasks=[research_task, writing_task, review_task],
process=Process.sequential, # 顺序执行
verbose=True
)
# 启动!
result = crew.kickoff()
print(result)
CrewAI 的关键设计理念:
- Role Playing:每个 Agent 有明确的角色、目标和背景故事
- Task Decomposition:将大任务拆解为多个子任务
- Process:sequential(顺序)、hierarchical(层级调度)
4. LangGraph 多 Agent 实战
LangGraph 用图结构实现更灵活的多 Agent 编排:
"""
LangGraph 多 Agent 示例:
Supervisor-Worker 模式,带条件路由和人工审核
"""
from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
# ============================================================
# 共享状态
# ============================================================
class MultiAgentState(TypedDict):
messages: Annotated[list, add_messages]
next_agent: str # 下一步由哪个 Agent 处理
research_result: str # 研究员结果
draft: str # 写手草稿
review_passed: bool # 审核是否通过
# ============================================================
# 各 Agent 节点
# ============================================================
def supervisor_node(state: MultiAgentState):
"""Supervisor:分析当前状态,决定路由到哪个 Agent"""
last_msg = state["messages"][-1].content
if not state.get("research_result"):
return {"next_agent": "researcher"}
elif not state.get("draft"):
return {"next_agent": "writer"}
elif not state.get("review_passed"):
return {"next_agent": "reviewer"}
else:
return {"next_agent": "FINISH"}
def researcher_node(state: MultiAgentState):
"""研究员 Agent"""
result = research_agent.invoke(state["messages"])
return {
"research_result": result,
"messages": [{"role": "assistant", "content": f"[研究完成] {result}"}]
}
def writer_node(state: MultiAgentState):
"""写手 Agent"""
draft = writer_agent.invoke(
f"基于研究资料写文章:{state['research_result']}"
)
return {
"draft": draft,
"messages": [{"role": "assistant", "content": f"[文章草稿完成] {draft[:200]}..."}]
}
def reviewer_node(state: MultiAgentState):
"""审核员 Agent"""
review = reviewer_agent.invoke(f"审核文章:{state['draft']}")
passed = "通过" in review
return {
"review_passed": passed,
"messages": [{"role": "assistant", "content": review}]
}
def human_review_node(state: MultiAgentState):
"""人工审核节点——暂停等待人类确认"""
# 实际场景中,这里可以发送通知并等待回调
print(f"\n🔔 请审核以下内容:\n{state['draft'][:500]}...")
user_input = input("是否通过?(y/n): ")
return {"review_passed": user_input.lower() == "y"}
# ============================================================
# 路由函数
# ============================================================
def route_agent(state: MultiAgentState) -> Literal["researcher", "writer", "reviewer", "human_review", END]:
return state["next_agent"] if state["next_agent"] != "FINISH" else END
# ============================================================
# 构建图
# ============================================================
builder = StateGraph(MultiAgentState)
builder.add_node("supervisor", supervisor_node)
builder.add_node("researcher", researcher_node)
builder.add_node("writer", writer_node)
builder.add_node("reviewer", reviewer_node)
builder.add_node("human_review", human_review_node)
builder.set_entry_point("supervisor")
# 所有 Agent 执行完后回到 Supervisor 判断下一步
for agent in ["researcher", "writer", "reviewer", "human_review"]:
builder.add_edge(agent, "supervisor")
builder.add_conditional_edges("supervisor", route_agent)
app = builder.compile()
# 运行
result = app.invoke({
"messages": [("user", "写一篇关于 AI Agent 发展趋势的技术博文")]
})
LangGraph 多 Agent 的优势:
- 图结构天然支持复杂的路由逻辑(条件分支、循环、并行)
- 状态在图中显式定义和流转,可观测性好
- Checkpointer 支持中断和恢复
- 支持人机协作(Human-in-the-Loop)
5. 生产部署架构
┌─────────────────────────────────────────────────────┐
│ 用户层 │
│ Web / App / API / 企业微信 │
└──────────────────────┬──────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────┐
│ Nginx(负载均衡 / 限流) │
└──────────────────────┬──────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────┐
│ FastAPI 应用层(无状态) │
│ ┌──────────┐ ┌──────────┐ ┌──────────────────┐ │
│ │ Auth │ │ 会话管理 │ │ Agent 编排 │ │
│ │ 认证鉴权 │ │ Session │ │ 调用 LangGraph │ │
│ └──────────┘ └──────────┘ └──────────────────┘ │
└──────────────────────┬──────────────────────────────┘
│
┌────────────┼────────────┐
▼ ▼ ▼
┌──────────┐ ┌──────────┐ ┌──────────────┐
│ Redis │ │PostgreSQL│ │ Milvus/Qdrant│
│ 缓存 │ │ 持久化 │ │ 向量检索 │
│ 会话/Semaphore│ 用户/日志 │ │ 记忆检索 │
└──────────┘ └──────────┘ └──────────────┘
│ │ │
└────────────┼────────────┘
▼
┌────────────────┐
│ LLM API │
│ OpenAI / 私有 │
│ (带重试+降级) │
└────────────────┘
关键组件说明:
| 组件 | 作用 | Agent 特有关注点 |
|---|---|---|
| 任务队列(Celery/RQ) | 异步执行长时间 Agent 任务 | Agent 可能跑几分钟,必须异步 |
| Redis 信号量 | 限制并发 Agent 数量 | LLM API 有速率限制,并发多了会被限 |
| 会话恢复 | Agent 中断后能继续 | LangGraph Checkpointer 天然支持 |
| 流式输出(SSE) | 实时推送 Agent 进度 | 用户不想盯着空白页面等 2 分钟 |
| 超时控制 | 防止 Agent 无限循环 | max_iterations + 总超时时间 |
# 生产级 Agent 服务示例
import asyncio
from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from langgraph.checkpoint.sqlite import SqliteSaver
import signal
app = FastAPI()
# 全局 Agent 实例(编译好,复用)
agent_app = build_agent_graph().compile(checkpointer=SqliteSaver.from_conn_string("checkpoints.db"))
# 并发控制
semaphore = asyncio.Semaphore(10) # 最多 10 个 Agent 并发
@app.post("/agent/chat")
async def agent_chat(request: ChatRequest):
async with semaphore:
try:
config = {"configurable": {"thread_id": request.session_id}}
result = await asyncio.wait_for(
agent_app.ainvoke(
{"messages": [("user", request.message)]},
config
),
timeout=120 # 2分钟总超时
)
return {"response": result["messages"][-1].content}
except asyncio.TimeoutError:
raise HTTPException(408, "Agent 处理超时,请简化您的请求")
except Exception as e:
raise HTTPException(500, f"Agent 处理异常:{str(e)}")
@app.get("/agent/stream")
async def agent_stream(request: ChatRequest):
"""流式 SSE 输出 Agent 进度"""
async def event_stream():
async for event in agent_app.astream_events(
{"messages": [("user", request.message)]},
config={"configurable": {"thread_id": request.session_id}},
version="v2"
):
if event["event"] == "on_tool_start":
yield f"data: {{\"type\": \"tool_start\", \"tool\": \"{event['name']}\"}}\n\n"
elif event["event"] == "on_tool_end":
yield f"data: {{\"type\": \"tool_end\", \"tool\": \"{event['name']}\"}}\n\n"
elif event["event"] == "on_chat_model_stream":
yield f"data: {{\"type\": \"token\", \"content\": \"{event['data']['chunk'].content}\"}}\n\n"
yield "data: {\"type\": \"done\"}\n\n"
return StreamingResponse(event_stream(), media_type="text/event-stream")
6. 监控与评估
6.1 监控指标
# Agent 专属监控指标
METRICS = {
# 性能指标
"agent_latency_p50": "Agent 响应 P50 延迟",
"agent_latency_p99": "Agent 响应 P99 延迟",
"tool_call_count": "平均每个请求的工具调用次数",
"llm_call_count": "平均每个请求的 LLM 调用次数",
# 质量指标
"task_success_rate": "任务完成率",
"tool_error_rate": "工具调用错误率",
"hallucination_rate": "幻觉率(基于人工标注)",
"user_satisfaction": "用户满意度评分",
# 成本指标
"tokens_per_request": "每个请求消耗的 Token 数",
"cost_per_request": "每个请求的成本(元)",
"daily_cost": "每日总成本",
}
6.2 Agent 评估方法
评估维度:
1. 任务完成度 —— Agent 是否达成了用户的目标?
评估方法:LLM-as-Judge(用另一个 LLM 评判结果)
2. 工具使用准确性 —— 是否选了正确的工具和参数?
评估方法:对比标准答案的工具调用序列
3. 推理合理性 —— 中间推理步骤是否合理?
评估方法:人工审查 Thought 日志
4. 效率 —— 用了多少步/多少 Token 完成任务?
评估方法:统计指标
# LLM-as-Judge 评估示例
def evaluate_agent_output(user_query: str, agent_output: str) -> dict:
"""用 LLM 评判 Agent 的输出质量"""
eval_prompt = f"""
请评估以下 Agent 对用户请求的响应质量。
用户请求:{user_query}
Agent 响应:{agent_output}
请打分(1-10)并说明理由:
1. 任务完成度:是否完全回答了用户问题?
2. 准确性:信息是否准确?有无事实错误?
3. 完整性:是否遗漏了关键信息?
4. 可用性:格式是否清晰?是否可直接使用?
返回 JSON:
{{"task_completion": 8, "accuracy": 9, "completeness": 7, "usability": 8, "comment": "..."}}
"""
response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": eval_prompt}],
response_format={"type": "json_object"}
)
return json.loads(response.choices[0].message.content)
7. 成本优化
Agent 的成本大头在 LLM 调用:
| 优化策略 | 预期节省 | 实现难度 |
|---|---|---|
| 小模型做简单任务 | 30-50% | 中 |
| 缓存重复调用 | 10-30% | 低 |
| 精简上下文(摘要压缩) | 20-40% | 中 |
| 限制最大迭代次数 | 10-20% | 低 |
| 批量处理 | 10-20% | 中 |
| 使用本地 Embedding 模型 | 100% | 低 |
# 成本优化的 Agent 框架
class CostOptimizedAgent:
"""智能选择模型以优化成本"""
# 模型分层
MODELS = {
"cheap": "gpt-4o-mini", # 简单推理、摘要
"standard": "gpt-4o", # 常规任务
"premium": "gpt-4o", # 复杂推理(可换成 o1 等)
}
# 带缓存的 LLM 调用
def __init__(self):
self.cache = {}
def cached_llm_call(self, messages: list, model: str) -> str:
"""带缓存的 LLM 调用"""
cache_key = hash(str(messages) + model)
if cache_key in self.cache:
print(" 💰 命中缓存,节省一次 LLM 调用")
return self.cache[cache_key]
response = client.chat.completions.create(
model=model, messages=messages
)
result = response.choices[0].message.content
self.cache[cache_key] = result
return result
def smart_model_choice(self, task: str) -> str:
"""根据任务复杂度选择模型"""
simple_tasks = ["摘要", "提取", "分类", "翻译"]
if any(t in task for t in simple_tasks):
return self.MODELS["cheap"]
return self.MODELS["standard"]
8. 安全防护
Agent 比普通 LLM 应用面临更多安全风险——因为它能实际操作外部系统。
Agent 安全五重防线:
第一层:输入过滤
- Prompt Injection 检测和清洗
- 敏感词过滤
- 输入长度限制
第二层:工具权限控制
- 白名单机制:工具只能调用预定义的 API
- 参数校验:SQL 注入防护、路径遍历防护
- 权限分级:不同用户不同工具访问权限
第三层:执行沙箱
- 代码执行在隔离环境中(Docker)
- 文件访问限制在白名单目录
- 网络访问限制
第四层:输出审查
- 敏感信息脱敏(手机号、身份证、密钥)
- 输出内容合规检查
- 注入检测:输出中不应包含工具调用指令
第五层:审计日志
- 记录所有 Agent 决策和工具调用
- 异常行为告警
- 可追溯、可回滚
# 工具调用安全检查
class SecureToolExecutor:
"""带安全检测的工具执行器"""
DANGEROUS_PATTERNS = [
r"rm\s+-rf", # 危险删除命令
r"DROP\s+TABLE", # SQL 删表
r"\.\./", # 路径遍历
r"sudo\s+", # 提权
]
def execute(self, tool_name: str, args: dict) -> str:
# 1. 工具白名单检查
if tool_name not in self.allowed_tools:
return json.dumps({"error": True, "message": f"工具 {tool_name} 不在白名单中"})
# 2. 参数安全检查
for key, value in args.items():
if isinstance(value, str):
for pattern in self.DANGEROUS_PATTERNS:
if re.search(pattern, value):
self.alert_security(f"检测到危险参数:{pattern}")
return json.dumps({"error": True, "message": "参数包含危险内容"})
# 3. 执行并记录
result = self._do_execute(tool_name, args)
self.audit_log(tool_name, args, result)
return result
9. 学习路线与展望
完整学习路径
阶段一:理解原理(1-2 周)
├── 精读本系列五篇文章
├── 手写 Agent Loop(不用框架)
├── 实现 ReAct Agent
└── 理解 Function Calling 本质
阶段二:动手实践(2-4 周)
├── 用 LangGraph 构建完整 Agent
├── 接入真实工具(搜索、数据库、邮件)
├── 添加记忆系统
└── 搭一个可用的 Demo
阶段三:多 Agent(1-2 周)
├── 用 CrewAI 构建多 Agent 团队
├── 实现 Supervisor-Worker 模式
└── 对比单 Agent vs 多 Agent 效果
阶段四:生产化(2-4 周)
├── 部署 FastAPI + Redis + 数据库
├── 添加监控和告警
├── 成本优化和缓存
├── 安全防护和权限控制
└── 灰度发布和 A/B 测试
阶段五:持续优化(长期)
├── 用户反馈收集和分析
├── 工具库扩展
├── 效果评估和迭代
└── 探索前沿(MCP 协议、Agent 框架演进)
前沿方向
| 方向 | 说明 | 值得关注 |
|---|---|---|
| MCP 协议 | Anthropic 推出的 Agent-工具标准协议 | ⭐⭐⭐⭐⭐ |
| Agent Computer Interface | Agent 操作电脑界面(GUI Agent) | ⭐⭐⭐⭐ |
| A2A 协议 | Google 的 Agent-to-Agent 通信协议 | ⭐⭐⭐⭐ |
| Self-Improving Agent | Agent 从经验中自我改进 | ⭐⭐⭐ |
| Multi-Modal Agent | 能理解图片、视频、音频的 Agent | ⭐⭐⭐⭐⭐ |
| Code Agent | 专门写代码、调 Bug 的 Agent(Devin, Cursor 等) | ⭐⭐⭐⭐⭐ |
常见陷阱
❌ 工具定义太模糊 → LLM 选错工具
❌ Prompt 里塞太多指令 → LLM 遵循度下降
❌ 没有 max_iterations → Agent 无限循环烧钱
❌ 长期记忆不清理 → 上下文越来越臃肿
❌ 没有监控 → 上线了都不知道 Agent 在瞎搞
❌ 过度设计多 Agent → 3 个 Agent 能做的事用了 8 个
总结
五篇系列至此完结。回顾全程:
- 概述:Agent = LLM + 感知 + 规划 + 行动 + 记忆
- 工具调用:Agent 的手,让 LLM 能操作外部世界
- 规划推理:Agent 的脑,ReAct/Plan-Execute/ReWOO 三种范式
- 记忆管理:Agent 的心,三层记忆让 Agent 持续进化
- 多 Agent 与生产:从玩具到产品的最后一公里
Agent 开发的核心心法:先理解原理(手写),再用框架(提效);先简单(单 Agent),再复杂(多 Agent);先 Demo(验证想法),再生产(打磨细节)。
希望这个系列能帮你真正掌握 Agent 开发,从 0 到 1 构建自己的智能体应用。
更多推荐


所有评论(0)