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的核心是状态图。把它想象成一个流程图：
from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated
import operator
# 定义状态结构
class AgentState(TypedDict):
messages: Annotated[list, operator.add] # 消息列表，自动累加
current_agent: str # 当前活跃的Agent
task_status: str # 任务状态
results: dict # 各Agent的结果
关键点：
- TypedDict定义状态结构，强类型更安全
- Annotated[list, operator.add]表示这个字段是\”可累加的\”，多个节点可以往里追加内容
- 状态在图中流转，每个节点都能读取和修改
2.2 节点（Node）
节点就是处理函数，接收当前状态，返回更新后的状态：
def research_node(state: AgentState) –> dict:
\”\”\”研究员节点：搜索并整理信息\”\”\”
messages = state[\”messages\”]
# 调用LLM进行研究
response = research_llm.invoke(messages)
return {
\”messages\”: [response], # 追加到消息列表
\”results\”: {
\”research\”: response.content}
}
def code_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完成后直接到B
graph.add_edge(\”research\”, \”code\”)
# 条件边：根据状态决定去哪
def should_continue(state: AgentState) –> str:
if state[\”task_status\”] == \”done\”:
return END
elif state[\”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：
[用户输入] → [研究员] → [写手] → [编辑] → [完成/修改]
↑ ↓
└──────────┘
3.2 完整代码实现
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from typing import TypedDict, Annotated, Literal
import operator
# ========== 1. 定义状态 ==========
class WritingState(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)
# 研究员：带搜索工具
from langchain_community.tools import TavilySearchResults
search_tool = TavilySearchResults(max_results=5)
researcher = llm.bind_tools([search_tool])
# 写手：纯文本生成
writer = llm
# 编辑：审查和反馈
editor = llm
# ========== 3. 定义节点函数 ==========
def researcher_node(state: WritingState) –> dict:
\”\”\”研究员节点：搜索并整理资料\”\”\”
topic = state[\”topic\”]
messages = [
SystemMessage(content=\”你是一个专业的研究员。搜索并整理关于给定主题的资料，提供关键事实、数据和观点。输出结构化的研究结果。\”),
HumanMessage(content=f\”请研究以下主题并整理资料：{
topic}\”)
]
response = researcher.invoke(messages)
# 如果有工具调用，执行搜索
if response.tool_calls:
tool_results = []
for tool_call in response.tool_calls:
result = search_tool.invoke(tool_call[\”args\”])
tool_results.append(result)
research_content = \”\\n\”.join([str(r) for r in tool_results])
else:
research_content = response.content
return {
\”research_results\”: research_content,
\”messages\”: [HumanMessage(content=f\”[研究完成] 已收集关于\’{
topic}\’的资料\”)]
}
def wr
