一个Agent干活慢？让十个Agent一起干
上个月我遇到了一个让我头疼的任务：帮客户分析100家竞品公司的产品信息，生成一份详细的竞品分析报告。
如果让一个Agent来做，它需要逐个搜索每家公司的信息、提取关键数据、对比分析。我试了一下，跑了将近3个小时，中间还崩了两次。
后来我换了个思路：让10个Agent并行工作，每个Agent负责10家公司，最后由一个”总编Agent”汇总。结果：25分钟搞定，质量还比一个Agent做的好。
这就是多智能体协作的威力。
这篇文章我会从零开始教你搭建一个多Agent协作系统。不需要任何分布式系统的知识，LangChain + Python就够了。
一、什么是多智能体协作
多智能体协作（Multi-Agent Collaboration）说白了就是：多个Agent分工合作，共同完成一个任务。
这跟人类团队是一样的道理。一个软件项目，你不可能让一个人同时做产品、设计、前端、后端、测试。你需要一个团队，每个人负责一个角色，协同工作。
多Agent系统也是一样。常见的角色分工：
- 管理者（Manager）：负责拆解任务、分配给其他Agent、汇总结果
- 研究员（Researcher）：负责搜索和收集信息
- 写手（Writer）：负责撰写内容
- 审核员（Reviewer）：负责检查质量、提出修改意见
- 程序员（Coder）：负责写代码、执行代码
- 测试员（Tester）：负责测试代码、报告bug
二、方案选择：三种多Agent架构
在LangChain生态里，搭建多Agent系统主要有三种方式：
方案一：LangGraph（推荐）
LangGraph是LangChain团队推出的专门用于构建多Agent系统的框架。它用”图”（Graph）的方式来定义Agent之间的协作流程。
优点：官方支持、功能完善、可视化好
缺点：学习曲线稍陡、版本迭代快
方案二：AgentExecutor + 手动编排
用LangChain的AgentExecutor，手动在代码中定义Agent之间的调用关系。
优点：简单直接、容易理解
缺点：复杂流程难以维护
方案三：CrewAI
CrewAI是专门做多Agent协作的第三方框架，设计理念是”把Agent当成团队成员来管理”。
优点：角色定义清晰、API简洁
缺点：社区不如LangChain大、高级功能有限
这篇文章我用LangGraph来演示，因为它是最成熟、最灵活的方案。
三、环境准备
pip install langchain==0.3.7 \\\\ langchain-openai==0.2.8 \\\\ langgraph==0.2.45 \\\\ langchain-community==0.3.7 \\\\ duckduckgo-search==6.3.0 \\\\ python-dotenv==1.0.1四、实战：搭建一个”写作团队”
我们来搭建一个由三个Agent组成的写作团队：
- 研究员：搜索资料、收集数据
- 写手：根据资料写文章
- 编辑：审核文章、提出修改意见
这三个Agent协作流程：研究员搜索 -> 写手写作 -> 编辑审核 -> 如果不合格，退回写手修改 -> 最终输出。
第一步：定义状态
from typing import TypedDict, Annotated, Sequenceimport operatorfrom langchain_core.messages import BaseMessageclass AgentState(TypedDict): messages: Annotated[Sequence[BaseMessage], operator.add] research_data: str draft: str review_feedback: str final_output: str iteration_count: int状态（State）是Agent之间共享的数据。每个Agent可以读取和修改状态。
第二步：创建Agent
import osfrom dotenv import load_dotenvfrom langchain_openai import ChatOpenAIfrom langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholderfrom langchain_core.messages import HumanMessage, SystemMessagefrom langchain_community.tools import DuckDuckGoSearchRunfrom langchain.agents import create_tool_calling_agent, AgentExecutorload_dotenv()llm = ChatOpenAI( model="gpt-4o-mini", temperature=0.7, api_key=os.getenv("OPENAI_API_KEY"))# ===== 研究员Agent =====search_tool = DuckDuckGoSearchRun()researcher_prompt = ChatPromptTemplate.from_messages([ ("system", """你是一个专业的研究员。你的任务是搜索和收集与主题相关的信息。【工作方式】1. 使用搜索工具进行多角度搜索2. 收集至少5个不同来源的信息3. 整理成结构化的研究报告4. 标注信息来源【输出格式】## 核心发现（3-5个最重要的发现）## 详细数据（具体的数据、事实、引用）## 趋势分析（当前趋势和未来展望）## 信息来源（列出所有引用的来源）"""), MessagesPlaceholder(variable_name="chat_history", optional=True), ("human", "{input}"), MessagesPlaceholder(variable_name="agent_scratchpad"),])researcher_agent = create_tool_calling_agent( llm=llm, tools=[search_tool], prompt=researcher_prompt)researcher_executor = AgentExecutor( agent=researcher_agent, tools=[search_tool], verbose=False, max_iterations=5)def researcher_node(state: AgentState): """研究员节点""" topic = state["messages"][-1].content result = researcher_executor.invoke({"input": f"请深入研究以下主题：{topic}"}) return { "research_data": result["output"], "messages": [HumanMessage(content=f"研究完成：{result['output'][:200]}...")] }# ===== 写手Agent =====writer_prompt = ChatPromptTemplate.from_messages([ ("system", """你是一个专业的技术写手。你的任务是根据研究资料撰写高质量的文章。【写作要求】1. 文章结构清晰：引言、正文（分3-5个小节）、结尾2. 语言生动接地气，避免AI腔3. 适当使用类比和故事4. 数据引用要准确5. 字数1500-2000字【写作风格】- 像朋友聊天一样自然- 可以有个人观点和感受- 适当用第一人称- 避免"首先、其次、最后"这种结构"""), ("human", "研究资料：\\{research_data}\\\\请根据以上资料写一篇文章。")])def writer_node(state: AgentState): """写手节点""" research_data = state["research_data"] chain = writer_prompt | llm result = chain.invoke({"research_data": research_data}) return { "draft": result.content, "messages": [HumanMessage(content=f"初稿完成，约{len(result.content)}字")] }# ===== 编辑Agent =====reviewer_prompt = ChatPromptTemplate.from_messages([ ("system", """你是一个严格的编辑。你的任务是审核文章质量。【审核标准】1. 内容准确性（是否有错误信息）2. 结构清晰度（逻辑是否通顺）3. 语言质量（是否有AI腔、套话）4. 可读性（是否有趣、吸引人）【输出格式】## 评分（每项10分）- 内容准确性：X/10- 结构清晰度：X/10- 语言质量：X/10- 可读性：X/10## 主要问题（列出2-3个最需要改进的问题）## 修改建议（具体、可操作的修改建议）## 总分XX/40## 结论APPROVED（总分>=30且每项>=6） 或 NEEDS_REVISION（需要修改）"""), ("human", "请审核以下文章：\\{draft}")])def reviewer_node(state: AgentState): """编辑节点""" draft = state["draft"] chain = reviewer_prompt | llm result = chain.invoke({"draft": draft}) return { "review_feedback": result.content, "messages": [HumanMessage(content=f"审核完成：{result.content[:200]}...")] }第三步：定义流程和条件
from langgraph.graph import StateGraph, ENDdef should_revise(state: AgentState): """判断是否需要修改""" iteration = state.get("iteration_count", 0) # 最多修改3次 if iteration >= 3: return "finalize" feedback = state.get("review_feedback", "") if "APPROVED" in feedback.upper(): return "finalize" else: return "revise"def revise_node(state: AgentState): """修改节点：根据编辑反馈修改文章""" draft = state["draft"] feedback = state["review_feedback"] revise_prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个专业写手。请根据编辑反馈修改文章。保持原有风格，只修改被指出的问题。"), ("human", "原文：\\{draft}\\\\编辑反馈：\\{feedback}\\\\请修改文章。") ]) chain = revise_prompt | llm result = chain.invoke({"draft": draft, "feedback": feedback}) return { "draft": result.content, "iteration_count": state.get("iteration_count", 0) + 1, "messages": [HumanMessage(content=f"第{state.get('iteration_count', 0) + 1}次修改完成")] }def finalize_node(state: AgentState): """最终输出节点""" return { "final_output": state["draft"], "messages": [HumanMessage(content="文章最终完成！")] }第四步：组装成图
# 创建图workflow = StateGraph(AgentState)# 添加节点workflow.add_node("researcher", researcher_node)workflow.add_node("writer", writer_node)workflow.add_node("reviewer", reviewer_node)workflow.add_node("revise", revise_node)workflow.add_node("finalize", finalize_node)# 设置入口workflow.set_entry_point("researcher")# 添加边workflow.add_edge("researcher", "writer")workflow.add_edge("writer", "reviewer")# 条件边：根据审核结果决定下一步workflow.add_conditional_edges( "reviewer", should_revise, { "revise": "revise", # 需要修改 "finalize": "finalize" # 审核通过 })workflow.add_edge("revise", "reviewer") # 修改后再审核workflow.add_edge("finalize", END) # 完成# 编译app = workflow.compile()第五步：运行！
# 运行多Agent写作系统result = app.invoke({ "messages": [HumanMessage(content="2026年AI Agent的发展趋势")], "research_data": "", "draft": "", "review_feedback": "", "final_output": "", "iteration_count": 0})# 输出最终文章print("=" * 50)print("最终文章：")print("=" * 50)print(result["final_output"])# 查看迭代次数print(f"\\迭代次数: {result['iteration_count']}")运行后，你会看到三个Agent自动协作：研究员搜索资料 -> 写手根据资料写初稿 -> 编辑审核 -> 如果有问题退回修改 -> 最终输出。整个过程完全自动。
五、进阶：并行Agent
上面的例子是串行的（一个接一个）。但很多时候，Agent可以并行工作，大幅提升速度。
比如，研究员可以同时搜索多个主题：
from langgraph.graph import StateGraph, ENDimport asyncio# 并行研究多个子主题async def parallel_research(state): topics = state["sub_topics"] # ["市场规模", "技术趋势", "应用场景"] async def research_one_topic(topic): result = researcher_executor.invoke({"input": topic}) return result["output"] # 并行执行 results = await asyncio.gather( *[research_one_topic(t) for t in topics] ) return {"research_data": "\\\\".join(results)}# 在LangGraph中，只要从同一个节点出发有多条边指向不同节点，# 这些节点就会并行执行workflow.add_node("parallel_research", parallel_research)workflow.add_edge("parallel_research", "writer")我用并行方式处理那个100家竞品分析的任务，10个Agent同时搜索，速度提升了将近10倍。
六、调试技巧
多Agent系统比单Agent复杂得多，调试是个挑战。几个实用技巧：
1. 给每个Agent的输出加日志
def researcher_node(state): print(f"[研究员] 开始研究...") result = researcher_executor.invoke(...) print(f"[研究员] 完成，输出{len(result['output'])}字") return {"research_data": result["output"]}2. 限制迭代次数
多Agent系统最容易出的问题就是无限循环。一定要设置最大迭代次数。
3. 可视化流程
LangGraph支持可视化：
# 生成流程图from langchain_core.runnables.graph import MermaidDrawMethodgraph_image = app.get_graph().draw_mermaid_png( draw_method=MermaidDrawMethod.API)with open("workflow.png", "wb") as f: f.write(graph_image)七、什么时候该用多Agent
多Agent不是万能的。以下情况适合用：
- 任务可以拆解：大任务能分成多个独立的小任务
- 需要不同专长：不同Agent可以有不同的工具和知识
- 追求速度：并行执行能大幅缩短时间
- 需要质量把控：有专门的审核Agent
以下情况没必要用：
- 任务简单，一个Agent就能搞定
- Agent之间有复杂的依赖关系，难以并行
- 预算有限（多Agent消耗更多token）
八、真实效果对比
我用同一个任务（写一篇2000字的AI趋势分析文章）对比了单Agent和多Agent：
| 指标 | 单Agent | 多Agent（串行） | 多Agent（并行） |
|---|---|---|---|
| 完成时间 | 约5分钟 | 约12分钟 | 约8分钟 |
| 内容质量 | ★★★☆☆ | ★★★★☆ | ★★★★☆ |
| 数据引用 | 3-5个 | 8-12个 | 8-12个 |
| Token消耗 | 约8000 | 约15000 | 约15000 |
| 迭代次数 | 无 | 1-2次 | 1-2次 |
多Agent的优势在于质量，不在于速度（除非用并行）。Token消耗大约是单Agent的2倍，但质量提升明显。
如果你已经熟悉了单Agent的搭建，多Agent是自然的下一步。从两个Agent开始（比如研究员 + 写手），跑通了再加第三个。别一上来就搞5个Agent，调试起来会让你怀疑人生。
更多关于Agent搭建的基础知识，可以参考 LangChain搭建AI Agent详细步骤。如果你想看Agent在实际业务中的应用，可以看看 AI Agent多场景落地案例详解。
