LangGraph多Agent协作系统实战

LangGraph多Agent协作系统实战：构建生产级多智能体应用

从单Agent到多Agent演进，手把手教你用LangGraph构建可扩展的协作系统

前言

在AI Agent开发领域，单个Agent的能力终究有限。当你需要构建一个能处理复杂任务的系统时——比如一个能同时进行研究、写作、代码审查的助手——多Agent协作就成了必经之路。

LangGraph作为LangChain团队推出的图编排框架，专为多Agent场景设计。它把工作流抽象成状态图（StateGraph），让多个Agent像接力赛一样协同工作。本文基于我在实际项目中的经验，带你从零搭建一个多Agent协作系统。

你将学到：

• LangGraph核心概念与状态图设计
• 多Agent通信与状态共享机制
• 条件路由与动态任务分配
• 错误处理与重试策略
• 生产环境部署注意事项

一、为什么需要多Agent？

1.1 单Agent的局限性

我在开发第一个AI助手时，把所有功能都塞进一个Agent：

# ❌ 错误做法：一个Agent干所有事system_prompt=""" 你是一个全能助手，需要： 1. 搜索网络获取信息 2. 分析数据生成报告 3. 编写和审查代码 4. 处理用户对话 5. 管理任务和日程 """

问题很快就暴露了：

• 上下文窗口爆炸：工具定义+系统提示+对话历史，轻松超过token限制
• 职责混乱：Agent经常在"搜索模式"和"代码模式"间混乱切换
• 错误传播：一个环节出错，整个流程崩溃
• 难以调试：出了问题不知道是哪个环节的锅
  

1.2 多Agent的优势

拆分成多个专用Agent后：

# ✅ 正确做法：每个Agent专注一件事research_agent=create_agent(tools=[search,browse],role="研究员")coder_agent=create_agent(tools=[execute,lint],role="程序员")reviewer_agent=create_agent(tools=[analyze],role="审查员")

关键收益：

• 每个Agent的prompt更短、更聚焦
• 可以独立测试和优化每个Agent
• 失败隔离，一个Agent挂了不影响其他
• 可以并行执行，提升效率

二、LangGraph核心概念

2.1 状态图（StateGraph）

LangGraph的核心是状态图。把它想象成一个流程图：

fromlanggraph.graphimportStateGraph,START,ENDfromtypingimportTypedDict,Annotatedimportoperator# 定义状态结构classAgentState(TypedDict):messages:Annotated[list,operator.add]# 消息列表，自动累加current_agent:str# 当前活跃的Agenttask_status:str# 任务状态results:dict# 各Agent的结果

关键点：

• TypedDict定义状态结构，强类型更安全
• Annotated[list, operator.add]表示这个字段是"可累加的"，多个节点可以往里追加内容
• 状态在图中流转，每个节点都能读取和修改

2.2 节点（Node）

节点就是处理函数，接收当前状态，返回更新后的状态：

defresearch_node(state:AgentState)->dict:"""研究员节点：搜索并整理信息"""messages=state["messages"]# 调用LLM进行研究response=research_llm.invoke(messages)return{"messages":[response],# 追加到消息列表"results":{"research":response.content}}defcode_node(state:AgentState)->dict:"""程序员节点：根据研究结果写代码"""research_result=state["results"].get("research","")prompt=f"根据以下研究结果编写代码：\n{research_result}"response=coder_llm.invoke([HumanMessage(content=prompt)])return{"messages":[response],"results":{**state["results"],"code":response.content}}

2.3 边（Edge）

边定义节点之间的连接关系：

# 普通边：A完成后直接到Bgraph.add_edge("research","code")# 条件边：根据状态决定去哪defshould_continue(state:AgentState)->str:ifstate["task_status"]=="done":returnENDelifstate["task_status"]=="need_review":return"reviewer"else:return"coder"graph.add_conditional_edges("router",should_continue,{END:END,"reviewer":"reviewer_node","coder":"code_node"})

三、实战：构建研究写作系统

3.1 系统设计

我们要构建一个研究写作系统，包含三个Agent：

1. 研究员（Researcher）：搜索网络，收集资料
2. 写手（Writer）：根据资料撰写文章
3. 编辑（Editor）：审查文章，提出修改意见

[用户输入] → [研究员] → [写手] → [编辑] → [完成/修改] ↑ ↓ └──────────┘

3.2 完整代码实现

fromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.prebuiltimportToolNodefromlangchain_openaiimportChatOpenAIfromlangchain_core.messagesimportHumanMessage,SystemMessagefromtypingimportTypedDict,Annotated,Literalimportoperator# ========== 1. 定义状态 ==========classWritingState(TypedDict):topic:str# 写作主题research_results:str# 研究结果draft:str# 文章草稿review_feedback:str# 审查反馈revision_count:int# 修改次数messages:Annotated[list,operator.add]# 对话历史status:str# 当前状态# ========== 2. 初始化Agent ==========llm=ChatOpenAI(model="gpt-4",temperature=0)# 研究员：带搜索工具fromlangchain_community.toolsimportTavilySearchResults search_tool=TavilySearchResults(max_results=5)researcher=llm.bind_tools([search_tool])# 写手：纯文本生成writer=llm# 编辑：审查和反馈editor=llm# ========== 3. 定义节点函数 ==========defresearcher_node(state:WritingState)->dict:"""研究员节点：搜索并整理资料"""topic=state["topic"]messages=[SystemMessage(content="你是一个专业的研究员。搜索并整理关于给定主题的资料，提供关键事实、数据和观点。输出结构化的研究结果。"),HumanMessage(content=f"请研究以下主题并整理资料：{topic}")]response=researcher.invoke(messages)# 如果有工具调用，执行搜索ifresponse.tool_calls:tool_results=[]fortool_callinresponse.tool_calls:result=search_tool.invoke(tool_call["args"])tool_results.append(result)research_content="\n".join([str(r)forrintool_results])else:research_content=response.contentreturn{"research_results":research_content,"messages":[HumanMessage(content=f"[研究完成] 已收集关于'{topic}'的资料")]}defwr