LangGraph实战指南：从零到一构建生产级多智能体系统

引言：为什么你的AI应用需要一个“大脑”？

想象一下这个场景：你用LangChain写了个智能客服，把用户问题扔进去，AI生成回答，结束。你觉得一切都很完美——直到用户问了一个需要多步推理的问题：“我想查一下我的账户余额，再看看最近有没有优惠活动，如果两样都有的话，帮我买个最合适的套餐。”

单次LLM调用显然搞不定。你需要分支判断、条件循环、状态记忆，甚至需要多个智能体协同工作。

这就是LangGraph出场的时刻。

LangGraph是LangChain生态中专门构建有状态、多步骤AI工作流的框架。它把复杂任务建模为有向图（Graph） ，通过节点（Node） 和边（Edge） 来编排执行流程。自2024年首次发布以来，经过多个版本迭代，LangGraph已成长为生产级智能体开发的基石工具。

本文将带你从核心概念走到实战落地，全程干货，可以直接复制代码跟着跑。

一、核心概念：用图来思考AI工作流

LangGraph的核心思想其实很简单：一个图由节点和边组成，节点负责做事，边负责决定下一步去哪里。

StateGraph：状态的共享黑板

LangGraph最核心的抽象是StateGraph——一个拥有全局状态的图。所有节点通过读写同一个State对象来通信。就像办公室里的共享白板：每个人可以在白板上写东西，其他人看到后继续执行自己的任务。

State定义示例：

fromtypingimportTypedDict,Annotated,List,LiteralimportoperatorclassAgentState(TypedDict):question:str# 用户原始问题next_action:Literal["calc","wiki","respond","end"]# 路由决策tool_input:str# 工具参数tool_output:str# 工具返回结果scratchpad:Annotated[List[str],operator.add]# 思考日志（自动合并）final_answer:str# 最终输出

其中scratchpad使用Annotated声明合并规则，框架会自动将多个节点追加的日志合并成一个列表——省去手动拼接的麻烦。

节点：业务逻辑的封装单元

每个节点就是一个函数，接收当前State，返回状态的增量更新（partial state）：

defcalculator_node(state:AgentState)->dict:"""数学计算节点示例"""ifstate["next_action"]!="calc":return{}# 不匹配则跳过try:result=eval(state["tool_input"])return{"tool_output":str(result),"scratchpad":[f"计算结果:{result}"]}exceptExceptionase:return{"tool_output":f"ERROR:{str(e)}"}

节点遵循纯函数原则：相同输入必然产生相同输出，无副作用，易于测试和组合。

边：动态决策的交通枢纽

LangGraph提供两种边：

· 确定性边：固定指向下一个节点
· 条件边：根据State内容动态选择下一个节点

fromlanggraph.graphimportStateGraph,ENDdefrouter(state:AgentState)->str:ifstate["tool_output"].startswith("ERROR"):return"error_handler"elifstate["next_action"]=="respond":return"final_node"else:return"continue"# 条件边配置graph.add_conditional_edges("calculator_node",# 源节点router,# 路由函数{"error_handler":"error_node","final_node":END,"continue":"next_node"})

构建与执行：三部曲

使用StateGraph分三步：定义 → 添加节点和边 → 编译。

fromlanggraph.graphimportStateGraph# 1. 初始化图workflow=StateGraph(AgentState)# 2. 添加节点workflow.add_node("calculator",calculator_node)workflow.add_node("formatter",formatter_node)workflow.add_node("error_handler",error_node)# 3. 设置入口和边workflow.set_entry_point("calculator")workflow.add_edge("calculator","formatter")workflow.add_conditional_edges("formatter",check_errors)# 4. 编译后执行app=workflow.compile()result=app.invoke({"question":"计算 25 * 4","next_action":"calc","tool_input":"25 * 4"})

编译后，等待执行，就得到最终结果。

二、实战案例：从零搭建ReAct智能体

纸上谈兵不如动手写代码。我们从最经典的ReAct模式开始——这个模式的本质是循环执行“思考→行动→观察”，直到任务完成。

什么是ReAct？ 它是Reasoning + Acting的组合，让AI在思考（Reason）和行动（Act）之间来回切换。例如用户问“明天北京天气怎么样”，AI先思考“需要查天气API”，然后执行天气查询的行动，最后观察结果并给出回答——就像人类的“先想后做、边做边想”。

完整代码实现：

fromtypingimportTypedDict,Annotated,List,Literalimportoperatorfromlanggraph.graphimportStateGraph,ENDfromlangchain_openaiimportChatOpenAIfromlangchain_core.toolsimporttool# 1. 定义状态结构classReActState(TypedDict):messages:Annotated[List[dict],operator.add]# 对话历史，自动合并next_action:Literal["think","act","respond","end"]tool_name:strtool_input:dicttool_result:strfinal_answer:str# 2. 定义工具@tooldefsearch(query:str)->str:"""模拟搜索功能"""returnf"关于'{query}'的搜索结果：找到3条相关信息..."@tooldefcalculate(expression:str)->str:"""计算表达式"""returnstr(eval(expression))tools_by_name={"search":search,"calculate":calculate}# 3. 定义节点函数defthink_node(state:ReActState)->dict:"""思考节点：LLM决定下一步做什么"""model=ChatOpenAI(model="gpt-4",temperature=0)prompt=f""" 你是一个智能助手。当前对话:{state['messages']}请判断下一步操作，输出格式: THINK: 推理过程 ACTION: 工具名 ACTION_INPUT: 工具输入 """response=model.invoke(prompt)# 简化示例：根据用户输入判断last_msg=state["messages"][-1]["content"]ifstate["messages"]else""if"计算"inlast_msg:# 提取表达式expr=last_msg.replace("计算","").strip()return{"tool_name":"calculate","tool_input":{"expression":expr},"next_action":"act","messages":[{"role":"assistant","content":f"我需要计算:{expr}"}]}elif"搜索"inlast_msg:query=last_msg.replace("搜索","").strip()return{"tool_name":"search","tool_input":{"query":query},"next_action":"act","messages":[{"role":"assistant","content":f"正在搜索:{query}"}]}else:return{"next_action":"respond","messages":[{"role":"assistant","content":"我直接回答您的问题"}]}defact_node(state:ReActState)->dict:"""执行节点：调用工具执行具体操作"""tool=tools_by_name.get(state["tool_name"])ifnottool:return{"tool_result":f"错误：找不到工具{state['tool_name']}","next_action":"think"}try:# 调用工具result=tool.invoke(state["tool_input"])return{"tool_result":str(result),"next_action":"think","messages":[{"role":"assistant","content":f"执行结果:{result}"}]}exceptExceptionase:return{"tool_result":f"执行出错:{str(e)}","next_action":"think"}defrespond_node(state:ReActState)->dict:"""回答节点：生成最终回复"""model=ChatOpenAI(model="gpt-4",temperature=0.7)prompt=f"根据上下文，回答用户：{state['messages']}\n工具执行结果：{state.get('tool_result','无')}"response=model.invoke(prompt)return{"final_answer":response.content,"next_action":"end"}# 4. 构建路由函数defroute_after_think(state:ReActState)->str:ifstate["next_action"]=="act":return"act_node"elifstate["next_action"]=="respond":return"respond_node"else:returnENDdefroute_after_act(state:ReActState)->str:return"think_node"# 5. 构建状态图workflow=StateGraph(ReActState)# 添加节点workflow.add_node("think",think_node)workflow.add_node("act",act_node)workflow.add_node("respond",respond_node)# 设置入口workflow.set_entry_point("think")# 添加边workflow.add_conditional_edges("think",route_after_think)workflow.add_edge("act","think_node")# 循环：行动后回到思考workflow.add_edge("respond",END)# 6. 编译并运行app=workflow.compile()result=app.invoke({"messages":[{"role":"user","content":"计算 123 * 456"}],"next_action":"","tool_name":"","tool_input":{},"tool_result":"","final_answer":""})print(f"最终回答:{result['final_answer']}")

这个例子展示了一个完整的ReAct循环：思考→行动→记录结果→再思考→最终回答。整个流程清晰可追踪。

你可以在此基础上拓展：

扩展方向 实现方式
添加更多工具 用@tool装饰器封装API调用、数据库查询等
增加重试机制 在act_node中添加异常处理和max_retries参数
实现状态持久化 集成LangGraph内置的MemorySaver或Redis存储

三、多智能体协作：从独行侠到团队作战

单智能体的天花板很明显：工具选择困难、上下文爆炸、角色功能混杂等问题会拖垮系统性能。

LangGraph天然支持多智能体架构，将复杂任务拆解给多个专业化Agent协同处理。某电商客服系统案例显示，采用多智能体方案后，工单处理效率提升了80%以上。

三种主流协作架构

1. 主管模式（Supervisor Pattern）

主管Agent统一接收请求，分析后分配给专业子Agent。适用于结构清晰、需要集中管控的场景。

fromlanggraph.graphimportStateGraph,ENDfromtypingimportTypedDict,LiteralclassSupervisorState(TypedDict):query:strcurrent_agent:Literal["billing","technical","sales","supervisor"]result:strdefsupervisor_node(state:SupervisorState):"""主管：分析请求并分发给专门的Agent"""query=state["query"]if"账单"inqueryor"支付"inquery:return{"current_agent":"billing"}elif"技术"inqueryor"报错"inquery:return{"current_agent":"technical"}elif"购买"inqueryor"价格"inquery:return{"current_agent":"sales"}else:return{"current_agent":"supervisor"}# 自己处理defbilling_agent(state:SupervisorState):"""账单处理智能体"""return{"result":f"已为您查询账单:{state['query']}"}deftechnical_agent(state:SupervisorState):"""技术问题处理智能体"""return{"result":f"技术团队已收到:{state['query']}，预计24小时内回复"}defsales_agent(state:SupervisorState):"""销售处理智能体"""return{"result":f"销售团队已关注:{state['query']}，即刻为您报价"}defrouter_after_supervisor(state:SupervisorState)->str:returnstate["current_agent"]# 构建图workflow=StateGraph(SupervisorState)workflow.add_node("supervisor",supervisor_node)workflow.add_node("billing",billing_agent)workflow.add_node("technical",technical_agent)workflow.add_node("sales",sales_agent)workflow.set_entry_point("supervisor")workflow.add_conditional_edges("supervisor",router_after_supervisor)workflow.add_edge("billing",END)workflow.add_edge("technical",END)workflow.add_edge("sales",END)

电信巨头Fastweb + Vodafone正是采用这种架构构建了服务近950万用户的AI客服系统，主管Agent负责识别用户意图并路由到专业用例模块。

2. 网络模式（Network Pattern）

Agent之间自由通信，形成网状拓扑。灵活性极高，但需要处理通信复杂度和冲突仲裁。

3. 分层模式（Hierarchical Pattern）

引入多层管理结构，协调层负责跨Agent通信，执行层专注任务完成。适合超大型企业系统，如ServiceNow用其构建覆盖客户全旅程的多智能体系统。

四、进阶玩法：让LangGraph更好用

1. Context API：告别配置地狱

LangGraph v0.6引入了革命性的Context API，解决了传统开发中上下文传递脆弱、类型不安全、扩展受限三大痛点。

旧模式（痛点） ：

defprocess_request(config:dict):user_id=config.get("user",{}).get("id")# 可能返回Nonedb_conn=config["database"]["connection"]# 潜在KeyError

新模式（类型安全） ：

fromdataclassesimportdataclass@dataclassclassAppContext:user_id:strsession_token:strmax_retries:int=3allowed_tools:list[str]=Nonedefhandle_request(runtime:Runtime[AppContext]):ifruntime.context.user_id=="admin":returnspecial_admin_handler(runtime)

通过dataclass定义上下文，IDE自动补全，编译时检查类型错误，整体开发效率提升60%以上。

2. 状态持久化与检查点

生产环境中，工作流可能执行几分钟甚至几小时。LangGraph内置的检查点机制可以随时保存状态，支持中断恢复：

fromlanggraph.checkpointimportMemorySaver# 使用内存存储（也可以换成RedisSaver、SqliteSaver等）memory=MemorySaver()app=workflow.compile(checkpointer=memory)# 带thread_id的执行，可中断后恢复config={"configurable":{"thread_id":"session_001"}}result=app.invoke({"query":"我的订单"+"..."*1000},config)

执行中断后，用相同thread_id调用invoke即可从中断点继续执行。

3. 生产环境部署最佳实践

从原型到生产，需要关注以下关键点：

依赖管理：使用Conda虚拟环境隔离，通过requirements-prod.txt固化生产版本。

name:langgraph-proddependencies:-python=3.9.12-langgraph=1.0.3

可观测性：集成ELK日志栈实现全链路追踪，用Prometheus+Grafana监控智能体响应延迟、状态同步成功率等关键指标。

弹性扩展：基于Kubernetes部署，支持API服务器与队列分离、独立扩缩容。某个采用LangGraph重构的智能客服系统，并发处理能力提升400%，故障恢复时间从小时级缩短至秒级。

五、LangGraph vs 传统方案：一张表看懂优势

对比维度 传统LangChain Chain LangGraph
控制流 线性管道，分支需要嵌套 完整图模型：分支、循环、并发
状态管理 隐式传递，难以追踪 显式State对象，全局可观察
可调试性 断点困难，像黑盒 每个节点都是独立单元，易测试
长时间任务 不支持中断恢复 检查点机制，天然支持持久化
多智能体 需手写协调器 内置多智能体模式

在复杂业务场景中，LangGraph的任务完成率比传统方案提升37%。

结语：LangGraph的无限可能

LangGraph远不止本文介绍的这些。从基础的ReAct模式到分布式多智能体系统，从单一工作流编排到千级智能体协同，LangGraph正在重新定义AI应用开发的边界。

正如Fastweb + Vodafone用它支撑近950万用户对话，ServiceNow用它构建企业级多智能体系统——LangGraph已经证明了自己在生产环境中的可靠性。

如果你还没有尝试过LangGraph，现在就是最好的开始时机。别把复杂的业务逻辑硬塞进线性Chain里了——用图来思考，用LangGraph来构建。