看到标题里的“吊打付费”四个字先别急着点收藏。LangGraph 这个框架真正硬核的地方不在于它有多“高级”而在于它把 AI Agent 从“单轮问答玩具”推进到了“可编排、可控制、可恢复的工程化工作流”。我结合最近给多个业务方落地多智能体系统的经验把 LangGraph 的核心组件、多智能体架构、条件路由、子图、并行分支、状态记忆逐一拆开配一份能直接跑起来的实战代码。不管你是刚接触 AI 大模型应用开发还是已经在用 LangChain 写链式调用这篇文章都能让你少走很多弯路。1. 背景与核心概念1.1 LangGraph 到底是什么LangGraph 是 LangChain 社区推出的一个专门用于构建有状态 Agent 工作流的编排框架。它的名字里有两个关键词Lang 代表它和 LangChain 生态天然打通Graph 则揭示了它的底层模型——把整个智能体任务看成一张有向图Directed Graph。在上面的图里每个节点Node代表一个可执行单元比如“调用大模型”“读取数据库”“调用外部 API”“执行 Python 函数”每条边Edge代表控制流也就是从一个节点运行完之后下一步该去哪里。和传统 Chain 那种“一条线走到黑”的固定管道不同LangGraph 允许中间节点根据当前状态动态决定下一步走向这就是“条件路由Conditional Edge”的核心能力。用一个通俗的比喻来理解LangChain Chain像一条传送带物料从入口放上去沿着固定路线依次经过加工台最后从出口出来。LangGraph像一间工厂的中央调度室每个工位都有自己的职责调度员会根据当前物料的状态、质检结果、订单类型动态决定这个物料下一步去哪条产线、要不要返工、要不要几个工位并行处理。对 AI 大模型应用开发者来说这意味着你不再只能写“用户提问 - 调用模型 - 返回答案”这种固定流程而是可以搭建真正的多智能体系统一个 Agent 负责意图识别一个 Agent 负责工具调用一个 Agent 负责结果校验它们之间通过状态共享和条件路由协同工作。1.2 LangGraph 解决的真实痛点在实际项目里单次大模型调用往往不能满足业务需求。以智能客服系统为例用户先问“我的订单什么时候发货”客服 Agent 需要查订单系统发现订单已发货但用户又追问“那运单号是多少”Agent 需要记住这是同一个用户、同一个订单上下文再去查物流系统。如果用户对答案不满意Agent 需要自动转入人工投诉流程。这个过程涉及多轮对话状态维护、工具调用、条件判断、异常降级、人机切换如果纯粹靠 if-else 胶水代码去拼很快就会失控。LangGraph 通过状态对象State在不同节点之间传递数据通过 Checkpointer 实现状态持久化和断点恢复让这类复杂流程变得可编排、可观测、可恢复。1.3 LangGraph 与 LangChain 的区别这是很多初学者最容易混淆的地方。我直接用一张表说明对比维度LangChainLangGraph核心抽象Chain链StateGraph状态图控制流线性、固定分支、循环、并行、动态路由状态管理弱通常靠外部变量内置 State跨节点共享持久化不支持Checkpointer支持断点续跑适合场景简单 RAG、提示词模板、单 Agent 工具调用多智能体协作、复杂工作流、需要人工审批的长流程学习曲线较低中等简单说LangChain 解决的是“怎么让模型调用工具更简单”LangGraph 解决的是“怎么把多个模型调用和工具调用编排成一套健壮的分布式流程”。新项目中如果涉及多 Agent 协作我建议直接基于 LangGraph 起步而不是先搭好 LangChain 再迁移。1.4 多智能体的常见交互模式在正式写代码之前有必要了解一下多智能体系统中常见的四种交互模式这能帮助你判断自己的业务场景该用哪种图结构顺序协作模式SequentialAgent A 的输出作为 Agent B 的输入适合流水线式处理。比如先做意图分类再做实体抽取最后生成回复。路由分发模式Router一个主导 Agent 根据输入内容把任务分发给不同的专用 Agent。这是客服、工单系统最常见的模式。并行协作模式Parallel多个 Agent 同时处理同一任务的不同子任务最后汇总。比如同时生成多方言版本的回复或者同时检查代码和检查安全风险。层级监督模式Hierarchical有一个“主管 Agent”负责任务拆解和结果验收多个“执行 Agent”负责具体执行。这是复杂项目中最接近真实团队协作的模式但实现成本也最高。了解这些交互模式后你会发现它们几乎都可以用 LangGraph 的节点 条件边 子图 并行分支组合出来。接下来我们直接进入实操环节。2. 环境准备与版本说明2.1 运行环境说明本文的示例代码基于以下环境版本可以根据你的项目实际情况调整重点演示的是配置和实现思路操作系统Windows 10/11、macOS、Linux 均可本文以 Linux 终端命令为例Python 版本3.10 或更高建议 3.11包管理工具pip 或 uv核心依赖langgraph、langchain-core、langchain-openai或你本地部署模型的 SDK可选langgraph-cli、langsmith需要提醒的是LangGraph 的 API 在 0.1.x、0.2.x、0.3.x 之间有少量调整。本文示例以 0.2.x 和 0.3.x 通用写法为主如果遇到兼容性问题优先查看官方文档和 changelog。2.2 安装 LangGraph建议创建一个干净的虚拟环境python -m venv .venv source .venv/bin/activate # Windows 下使用 .venv\Scripts\activate pip install --upgrade pip pip install langgraph langchain-core langchain-openai如果你同时需要 LangGraph 的开发调试工具可以安装 CLIpip install langgraph-cli安装完成后可以用下面的命令验证版本python -c import langgraph; print(langgraph.__version__)我这里演示时安装的是0.3.x版本的 LangGraph。如果你后续引入了langgraph dev相关命令它本质上会启动一个带持久化存储的后端服务而直接uvicorn app:app启动普通的 FastAPI 应用则不会自动附加 LangGraph 的持久化和热重载能力。生产环境部署时根据是否需要状态管理来选择启动方式。2.3 示例项目结构为了后续实战代码更清晰推荐按下面的目录结构组织项目langgraph_agent_demo/ ├── agent_system/ │ ├── __init__.py │ ├── state.py # 定义全局 State │ ├── tools.py # 工具函数模拟订单/物流查询 │ ├── nodes.py # 各个节点函数 │ ├── subgraphs.py # 子图定义 │ └── graph.py # 主图组装 ├── .env # 存放模型 API Key └── main.py # 程序入口这样拆分的好处是每一个节点函数都可以独立测试后续扩展新 Agent 时不需要改主图逻辑只需要加节点、加边。3. LangGraph 核心组件深度拆解3.1 State全局状态State 是 LangGraph 的灵魂。它决定了节点之间如何传递数据。在实际工程中State 通常是一个 TypedDict 或者 dataclass。下面是一个典型的定义# 文件路径agent_system/state.py from typing import TypedDict, Annotated, List from langgraph.graph.message import add_messages def merge_lists(left: list, right: list) - list: 自定义状态合并函数用于多个节点并行写入同一字段时进行合并。 return left right class AgentState(TypedDict): # 用户原始输入 user_input: str # 意图识别结果 intent: str # 订单号 order_no: str # 节点间的中间消息列表 messages: Annotated[List[str], add_messages] # 最终回复 reply: str # 当前重试次数 attempts: int # 并行任务结果汇总 parallel_results: Annotated[List[str], merge_lists]这里的关键点在于Annotated[List[str], add_messages]这种写法。add_messages是 LangGraph 内置的状态归约器Reducer当多个节点试图写入同一个messages字段时会用追加的方式合并而不是直接覆盖。同理自定义的merge_lists函数也能控制并行分支回填状态的方式。为什么需要 Reducer因为在并行分支场景下多个节点会同时更新同一个状态字段如果没有合并策略后写入的会覆盖先写入的导致数据丢失。实际项目中凡是会被多个节点写入的列表字段都应该定义合适的 Reducer。3.2 Node工作流节点Node 就是普通的 Python 函数签名统一为(state: AgentState) - dict[str, Any]返回值会被合并到全局状态中。下面是一个最小节点示例# 文件路径agent_system/nodes.py from .state import AgentState def receive_input(state: AgentState) - dict: 接收用户输入并初始化相关字段。 print(f收到用户问题{state[user_input]}) return { messages: [f用户输入{state[user_input]}], attempts: 0, } def classify_intent(state: AgentState) - dict: 意图识别节点。 真实项目中这里应该调用大模型本示例用简单规则代替方便演示。 user_input state.get(user_input, ) if 订单 in user_input or 发货 in user_input: intent after_sales elif 投诉 in user_input or 不满 in user_input: intent complaint else: intent consult print(f意图识别结果{intent}) return {intent: intent, messages: [f意图{intent}]}注意Node 函数不直接修改外部全局变量而是返回一个字典告诉 LangGraph 哪些状态字段需要更新。这种设计保证了每个节点都是纯函数式的便于测试、回放和并行执行。3.3 Edge连接与路由Edge 用来定义节点之间的连接关系。基础用法很简单from langgraph.graph import StateGraph, START, END # 创建状态图 graph StateGraph(AgentState) # 添加节点 graph.add_node(receive_input, receive_input) graph.add_node(classify_intent, classify_intent) # 添加边 graph.add_edge(START, receive_input) graph.add_edge(receive_input, classify_intent) graph.add_edge(classify_intent, END)START是图入口END是图出口。这种写法形成了一条“开始 - 接收输入 - 识别意图 - 结束”的线性链路。如果所有流程都这么线性LangGraph 和普通 Chain 就没区别了。它的威力体现在add_conditional_edges上。3.4 Conditional Edge条件路由条件路由允许你根据当前状态动态决定下一步走向。这是实现 Router 模式、多智能体分工的基础。用法如下from typing import Literal from .state import AgentState def route_after_classify(state: AgentState) - Literal[consult, after_sales, complaint]: 根据意图将任务路由到不同 Agent。 intent state.get(intent, consult) # 这里返回的是下一步节点的名称 if intent after_sales: return after_sales_agent elif intent complaint: return complaint_agent return consult_agent # 在主图中添加条件路由 graph.add_conditional_edges( classify_intent, route_after_classify, { consult: consult_agent, after_sales: after_sales_agent, complaint: complaint_agent, } )关键点第一个参数是源节点名也就是执行完哪个节点后调用这个路由函数。第二个参数是路由函数它接收当前 State返回一个目标节点标识。第三个参数字典是可选的它把路由函数的返回值映射到实际的节点名。如果路由函数返回的值和节点名一致可以省略这个字典。在实际项目中路由函数通常由大模型驱动。比如在意图识别节点里模型返回 JSON{intent: refund}然后路由函数读取这个字段决定走退款流程还是人工客服。3.5 Checkpointer记忆与断点Checkpointer 是 LangGraph 相对较难理解但价值极高的组件。它的核心作用是保存图的运行状态快照从而支持三类能力多轮对话记忆同一线程thread_id的多轮交互共用同一份状态。断点恢复比如人工审批节点执行完毕后可以从断点继续往下走。时间旅行可以回放历史状态调试复杂工作流。LangGraph 提供了多种持久化后端常用的有MemorySaver内存存储适合开发测试。SqliteSaverSQLite 文件存储适合单机生产。PostgresSaverPostgreSQL 存储适合多实例部署。用法上只需要在编译时传入一个 Checkpointer 实例from langgraph.checkpoint.memory import MemorySaver memory MemorySaver() # 编译图时传入 checkpointer app graph.compile(checkpointermemory) # 运行时传入 thread_id用于区分不同会话 config {configurable: {thread_id: user-001}} result app.invoke({user_input: 我想查一下订单发货进度}, configconfig)这里需要特别说明一个热词InMemorySaver 中的内容如何参与大模型上下文。调用app.invoke()时State 里的messages字段会自动带上历史消息。假设你将messages设计为消息列表那么后续节点可以直接把state[messages]拼接成提示词发送给大模型。步骤如下def build_prompt(state: AgentState) - str: # 从 Checkpointer 恢复的消息历史 history_messages state.get(messages, []) # 把历史消息格式化成大模型可用的对话上下文 context \n.join([f{msg} for msg in history_messages]) prompt f以下是历史对话记录\n{context}\n\n当前用户问题{state[user_input]} return prompt换句话说Checkpointer 负责保存状态而状态里哪些字段需要作为大模型上下文完全由你决定。3.6 编译与执行图搭建完成后调用compile()得到一个可调用的CompiledStateGraph对象app graph.compile(checkpointermemory) # 同步调用 result app.invoke({user_input: 我的订单怎么还没发货}, configconfig) # 异步调用 # result await app.ainvoke({user_input: ...}, configconfig) # 流式输出 # for event in app.stream({user_input: ...}, configconfig): # print(event)compile()底层会做一次拓扑校验如果图中有不可达节点、环上缺少条件边、或节点函数签名不符合要求会在编译阶段报错。这也是 LangGraph 适合工程化的原因——很多结构性问题在运行前就能暴露。4. 完整实战案例智能客服工单多智能体系统4.1 需求分析我们构建一个“智能客服工单多智能体系统”完整覆盖以下几种真实场景咨询类问题用户询问退换货政策。系统调用咨询 Agent 直接回答。售后类问题用户询问订单发货进度。系统进入售后子图该子图尝试查询订单和物流信息如果查询失败则自动重试最多 3 次失败后转入人工处理。投诉类问题用户表达严重不满。系统同时启动“安抚话术生成”和“补偿方案生成”两个并行节点汇总后生成最终回复。所有对话记录通过 Checkpointer 持久化支持多轮追问联系上下文。整个系统的主图结构如下START | v receive_input | v classify_intent | v route_after_classify (条件路由) |--- consult_agent ---------- merge_result --- END |--- after_sales_subgraph --- merge_result --- END |--- complaint_agent --------- merge_result --- END其中after_sales_subgraph是一个子图complaint_agent内部有并行分支。4.2 创建项目结构mkdir -p langgraph_agent_demo/agent_system cd langgraph_agent_demo touch agent_system/__init__.py后续代码都放在agent_system目录下。4.3 定义状态先完善state.py# 文件路径langgraph_agent_demo/agent_system/state.py from typing import TypedDict, Annotated, List from langgraph.graph.message import add_messages def merge_lists(left: list, right: list) - list: 并行节点结果合并函数。 return left right class AgentState(TypedDict): user_input: str intent: str order_no: str attempts: int messages: Annotated[List[str], add_messages] parallel_results: Annotated[List[str], merge_lists] reply: str4.4 编写基础节点下面编写nodes.py包含主图的几个基础节点# 文件路径langgraph_agent_demo/agent_system/nodes.py from .state import AgentState def receive_input(state: AgentState) - dict: 接收用户输入。 print(f[receive_input] {state[user_input]}) return { messages: [f用户{state[user_input]}], attempts: 0, } def classify_intent(state: AgentState) - dict: 意图识别节点。 简化版用关键词判断真实项目可替换为 LLM 调用。 text state.get(user_input, ) if 投诉 in text or 不满 in text or 差评 in text: intent complaint elif 订单 in text or 发货 in text or 物流 in text or 退款 in text: intent after_sales else: intent consult print(f[classify_intent] 意图{intent}) return {intent: intent, messages: [f系统识别到意图 - {intent}]} def consult_agent(state: AgentState) - dict: 咨询 Agent直接基于知识库规则回答。 reply 我们的退换货政策是签收后 7 天内支持无理由退换货。请问还需要了解其他内容吗 print(f[consult_agent] 咨询 Agent 回复) return {reply: reply, messages: [f咨询Agent{reply}]} def complaint_agent(state: AgentState) - dict: 投诉 Agent 的入口节点。 print([complaint_agent] 启动投诉处理流程) # 这里只写入一个占位消息真正的并行逻辑在子图节点中处理 return {messages: [系统已进入投诉处理流程]} def merge_result(state: AgentState) - dict: 汇总节点把多个节点的输出合并成最终回复。 results state.get(parallel_results, []) if results: final_reply .join(results) else: final_reply state.get(reply, 抱歉暂时无法处理已转接人工客服。) print(f[merge_result] 最终回复{final_reply}) return {reply: final_reply, messages: [f系统{final_reply}]}为了让代码更接近真实场景我们在tools.py中准备几个模拟工具函数用于模拟订单查询、物流查询等外部系统# 文件路径langgraph_agent_demo/agent_system/tools.py import random def query_order(order_no: str) - dict: 模拟查询订单信息。 print(f调用订单系统order_no{order_no}) status_list [已支付, 已发货, 运输中, 已签收] return { order_no: order_no, status: random.choice(status_list), logistics_company: 顺丰速运, tracking_no: fSF{order_no}123456, } def simulate_success(probability0.6) - bool: 模拟外部接口调用成功率用于演示重试机制。 return random.random() probability4.5 编写售后子图子图Subgraph是 LangGraph 实现模块化的重要手段。你可以把一个负责“售后进度查询 重试机制”的流程封装成子图然后在主图中当作一个普通节点来使用。下面是子图的完整实现# 文件路径langgraph_agent_demo/agent_system/subgraphs.py from typing import TypedDict from langgraph.graph import StateGraph, START, END from .state import AgentState from .tools import query_order, simulate_success class AfterSalesState(TypedDict): order_no: str query_result: str attempts: int status: str # success / failed / max_retry def query_order_node(state: AfterSalesState) - dict: 查询订单与物流信息模拟可能失败。 order_no state.get(order_no, UNKNOWN) attempts state.get(attempts, 0) 1 if not simulate_success(probability0.5): print(f[售后子图] 第 {attempts} 次查询订单失败) return { query_result: f第 {attempts} 次查询失败, attempts: attempts, status: failed, } result query_order(order_no) text f订单 {order_no} 当前状态{result[status]}物流公司{result[logistics_company]}运单号{result[tracking_no]} print(f[售后子图] 查询成功{text}) return { query_result: text, attempts: attempts, status: success, } def check_retry(state: AfterSalesState) - str: 判断是否需要重试。 if state.get(status) success: return success if state.get(attempts, 0) 3: return max_retry return retry def build_after_sales_subgraph(): 构建售后处理子图。 subgraph StateGraph(AfterSalesState) # 添加节点 subgraph.add_node(query_order, query_order_node) # 添加入口和出口 subgraph.add_edge(START, query_order) # 条件路由支持循环重试 subgraph.add_conditional_edges( query_order, check_retry, { success: END, retry: query_order, # 回到自身节点形成循环 max_retry: END, }, ) return subgraph.compile()这个子图体现了 LangGraph 的循环检测能力。由于check_retry是一个条件边LangGraph 能识别出“从 query_order 到 query_order”的循环边只要循环存在出口本例中的success和max_retry图就能正常编译和运行不会进入死循环。4.6 编写并行分支节点在投诉处理场景中我们希望同时生成“安抚话术”和“补偿方案”然后在汇合节点统一汇总。LangGraph 天然支持这种扇出fan-out汇聚fan-in结构# 文件路径langgraph_agent_demo/agent_system/parallel_nodes.py from .state import AgentState def generate_comfort_message(state: AgentState) - dict: 并行分支1生成安抚话术。 comfort 非常抱歉给您带来了不好的体验我们已经高度重视您的问题。 print(f[并行分支1] 安抚话术生成完毕) return { parallel_results: [f安抚话术{comfort}], messages: [系统安抚话术生成完毕], } def generate_compensation(state: AgentState) - dict: 并行分支2生成补偿方案。 compensation 针对本次问题我们为您提供 20 元无门槛优惠券作为补偿。 print(f[并行分支2] 补偿方案生成完毕) return { parallel_results: [f补偿方案{compensation}], messages: [系统补偿方案生成完毕], }这里的parallel_results字段配置了merge_lists归约器所以两个并行节点返回的列表会被合并而不是后者覆盖前者。这是 LangGraph 并行分支中最需要注意的细节。4.7 组装主图现在把所有部分组装成完整的主图graph.py# 文件路径langgraph_agent_demo/agent_system/graph.py from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver from typing import Literal from .state import AgentState from .nodes import ( receive_input, classify_intent, consult_agent, complaint_agent, merge_result, ) from .parallel_nodes import generate_comfort_message, generate_compensation from .subgraphs import build_after_sales_subgraph # 构建售后子图 after_sales_subgraph_app build_after_sales_subgraph() def route_after_classify(state: AgentState) - Literal[consult_agent, after_sales_subgraph, complaint_agent]: 根据意图分发到不同 Agent。 intent state.get(intent, consult) if intent after_sales: return after_sales_subgraph elif intent complaint: return complaint_agent return consult_agent def build_main_graph(): graph StateGraph(AgentState) # 添加主图节点 graph.add_node(receive_input, receive_input) graph.add_node(classify_intent, classify_intent) graph.add_node(consult_agent, consult_agent) graph.add_node(complaint_agent, complaint_agent) # 子图作为节点加入主图 graph.add_node(after_sales_subgraph, after_sales_subgraph_app) # 并行分支节点 graph.add_node(generate_comfort_message, generate_comfort_message) graph.add_node(generate_compensation, generate_compensation) # 汇总节点 graph.add_node(merge_result, merge_result) # 入口 - 接收输入 - 意图识别 graph.add_edge(START, receive_input) graph.add_edge(receive_input, classify_intent) # 条件路由分发到不同 Agent graph.add_conditional_edges( classify_intent, route_after_classify, { consult_agent: consult_agent, after_sales_subgraph: after_sales_subgraph, complaint_agent: complaint_agent, }, ) # 咨询 Agent 和售后子图完成后都进入汇总节点 graph.add_edge(consult_agent, merge_result) graph.add_edge(after_sales_subgraph, merge_result) # 投诉 Agent启动并行分支 graph.add_edge(complaint_agent, generate_comfort_message) graph.add_edge(complaint_agent, generate_compensation) # 并行分支汇合到汇总节点 graph.add_edge(generate_comfort_message, merge_result) graph.add_edge(generate_compensation, merge_result) # 汇总节点输出 graph.add_edge(merge_result, END) return graph.compile(checkpointerMemorySaver()) main_app build_main_graph()这里其实有一个小的补充如果希望售后子图的输出直接回填到主图的reply字段需要把主图节点函数和子图 State 的结构对齐。在上面的示例中子图内部只维护了query_result字段而主图状态中的reply没有被更新因此汇总节点的兜底逻辑会生效“抱歉暂时无法处理”。在实际项目中建议在子图内部返回一个与主图 State 对应的字段比如reply或者通过封装节点函数做字段映射。4.8 运行与验证在项目根目录创建main.py# 文件路径langgraph_agent_demo/main.py from agent_system.graph import main_app def run_demo(): # 模拟三个用户的会话每个会话都有独立的 thread_id状态隔离 sessions [ {thread_id: user-001, user_input: 我想了解一下退换货政策}, {thread_id: user-002, user_input: 我的订单怎么还没发货订单号是10086}, {thread_id: user-003, user_input: 我要投诉你们服务太差了}, ] for session in sessions: thread_id session[thread_id] user_input session[user_input] config {configurable: {thread_id: thread_id}} print(f\n 新会话 {thread_id} ) result main_app.invoke( {user_input: user_input}, configconfig, ) print(f\n最终回复{result[reply]}) # 第二轮回访测试 Checkpointer 记忆恢复 if thread_id user-002: print(\n--- 用户追问 ---) result2 main_app.invoke( {user_input: 那什么时候能送到}, configconfig, ) print(f追问回复{result2[reply]}) if __name__ __main__: run_demo()运行结果示意如下 新会话 user-001 [receive_input] 我想了解一下退换货政策 [classify_intent] 意图consult [consult_agent] 咨询 Agent 回复 最终回复我们的退换货政策是签收后 7 天内支持无理由退换货。请问还需要了解其他内容吗 新会话 user-002 [receive_input] 我的订单怎么还没发货订单号是10086 [classify_intent] 意图after_sales [售后子图] 第 1 次查询订单失败 [售后子图] 第 2 次查询订单失败 [售后子图] 第 3 次查询订单失败 最终回复抱歉暂时无法处理已转接人工客服。 新会话 user-003 [receive_input] 我要投诉你们服务太差了 [classify_intent] 意图complaint [complaint_agent] 启动投诉处理流程 [并行分支1] 安抚话术生成完毕 [并行分支2] 补偿方案生成完毕 最终回复安抚话术非常抱歉给您带来了不好的体验我们已经高度重视您的问题。补偿方案针对本次问题我们为您提供 20 元无门槛优惠券作为补偿。从运行结果可以看出三点条件路由正确、售后子图循环重试机制生效、投诉场景并行分支正确合并。如果你第二次运行 user-002 的会话会发现子图状态从 Checkpointer 恢复attempts会被重置还是保留取决于子图状态初始化逻辑这一点需要开发者在设计子图时明确状态初始化时机。5. 常见问题与排查思路LangGraph 上手虽然快但实际开发中还是有不少容易踩的坑。我把高频问题整理成一张排查表方便你对照处理问题现象常见原因解决思路编译报错InvalidUpdateError某个字段被多个节点同时写入但没有定义 Reducer在 State 中使用Annotated[list, add_messages]或自定义合并函数图进入死循环条件边把状态送回上游节点但缺少终止条件检查路由函数的返回值务必保证存在能通向END的分支多轮对话上下文丢失没有传thread_id或没有配置 Checkpointer编译图时传入checkpointer调用invoke时传config{configurable: {thread_id: ...}}子图无法访问父图状态子图使用了自己的 State 类型与父图字段名不一致设计子图 State 时保持关键字段命名一致或写转换函数并行分支结果被覆盖并行节点写入同一个列表字段但没有使用 Reducer为列表字段配置合并策略模型调用超时导致整个图失败未对 Node 做超时和异常处理在节点函数内部使用try/except并设置合理的模型调用超时时间条件路由返回的节点名不存在路由函数返回值与add_node的节点名不一致检查路由分支字典的映射保持名称完全一致langgraph dev与普通 FastAPI 启动行为不同开发服务和普通 Web 服务对状态持久化的处理不同开发环境用uv run langgraph dev生产环境明确是否启用持久化存储这里单独说一下InvalidUpdateError这个问题。很多初学者搭建第一个高并发 Agent 时会遇到典型的错误提示是langgraph.errors.InvalidUpdateError: Expected list, got str这个报错的本质是State 中某个字段被定义为列表但某个节点返回了字符串。比如class State(TypedDict): chat_history: Annotated[list, add_messages] def node_a(state: State): return {chat_history: hello} # 错误应该返回 [hello]只要保持返回值和 State 类型一致这类问题就能避免。6. 最佳实践与工程建议6.1 状态设计拒绝“万能 State”很多开发者习惯把所有中间变量都塞进 State最终导致状态臃肿、难以追踪。我的建议是State 只保存需要跨节点传递的、会影响路由或最终输出的数据。临时中间结果比如一批原始 JSON尽量在节点函数内部消化不要全部塞进 State。列表字段必须设计 Reducer避免并行写入覆盖。6.2 子图拆分以业务边界为准不要为了“炫技”把图拆得过于复杂。子图拆分的合理依据是业务边界比如“售后处理”是一个完整的子流程内部有重试逻辑“投诉处理”是另一个完整子流程内部有并行计算。一个合格的多智能体系统主图的节点数量通常控制在 5~8 个更多细节应该沉淀到子图中。6.3 外部系统调用重试与降级在真实项目中Agent 的节点函数往往要调用订单系统、物流系统、支付系统等外部服务。这些服务不可能 100% 可用所以每个外部调用节点都应该做三件事超时控制给 HTTP 客户端设置合理的连接超时和读超时。重试策略利用 LangGraph 的条件边实现有限次重试而不是在节点内部写死 while 循环。降级方案重试失败后必须给出降级回复不能直接抛异常导致整个图崩溃。本文的售后子图就是一个很好的参考模板。你可以把simulate_success替换成真实的 HTTP 调用并在query_order_node里捕获外部异常。6.4 Checkpointer 的正确用法Checkpointer 在开发调试阶段用MemorySaver最方便但生产环境要注意单实例部署可以用SqliteSaver保证服务重启后状态不丢失。多实例部署必须使用PostgresSaver或 Redis 这类共享存储否则不同实例之间看不到彼此的会话状态。涉及用户隐私的会话数据建议在 Checkpointer 层做加密或脱敏。另外一个容易被忽略的点Checkpointer 保存的 State 会包含模型响应消息、工具调用中间结果等敏感信息。在把 State 写入日志或传给前端时需要显式过滤。6.5 安全边界与大模型风险控制只要涉及大模型调用就不能忽略安全问题。结合 LangGraph 的实际工程特性我给出以下清单提示词注入防护用户在输入中提到“忽略上述指令”时节点应拒绝执行敏感操作。权限最小化子图中的工具调用节点应使用独立的 API Key访问范围最小化避免一个 Agent 拥有全量系统权限。输出校验模型生成的回复在返回用户之前建议经过内容安全过滤和关键词校验。人工审批涉及退款、删除操作时加入“人工确认”节点。可以用interrupt_before实现挂起等待人工审批后恢复。# 编译时挂起示例 app graph.compile( checkpointermemory, interrupt_before[execute_refund], # 到达退款节点前挂起 )6.6 日志与可观测性LangGraph 生态支持 LangSmith 的完整链路追踪但如果团队没有接入第三方平台也可以通过自定义 Node 的日志实现基础的可观测性。建议每个节点统一打印结构化日志import json import time def log_node_entry(node_name: str, state: dict): print(json.dumps({ event: node_entry, node: node_name, timestamp: time.time(), intent: state.get(intent), attempts: state.get(attempts), }, ensure_asciiFalse))6.7 测试策略多智能体系统的测试和普通单测不太一样。除了单元测试节点函数还应该做图结构测试编译时无异常、节点覆盖率达到预期。路由矩阵测试构造不同输入验证意图识别和条件路由结果符合预期。故障注入测试模拟外部服务异常验证重试和降级逻辑。状态恢复测试同一个thread_id下验证多轮对话上下文正确恢复。建议用pytest组织这些测试。LangGraph 的纯函数式节点设计让单测变得非常简单不需要 mock 复杂的内部状态。7. 总结与学习路线写完这套智能客服多智能体系统你应该已经掌握了 LangGraph 最核心的思维方式和编码套路State 管理节点间数据、普通边保证线性流程、条件边实现动态路由和循环、子图承载独立业务模块、并行分支提高处理效率、Checkpointer 提供会话记忆和断点恢复。对比纯 LangChain 的链式调用这种基于图结构的编排方式在面对复杂业务时优势非常明显尤其是路由、重试、人工审批、多 Agent 协作这些场景代码的可维护性和可观测性都上了一个台阶。接下来你可以按这个顺序继续深入把示例中的关键词意图识别替换成真实的大模型调用感受 LLM 驱动路由的实际效果。将MemorySaver换成SqliteSaver部署一个带持久化的本地服务。研究 LangGraph 的interrupt_before/interrupt_after实现人工审批中断恢复。尝试写一个层级监督模式的多智能体系统主管 Agent 拆解任务多个执行 Agent 并行处理主管统一验收。如果本文对你有帮助可以收藏备用。实际动手时如果遇到新的报错场景欢迎在评论区把日志贴出来我们一起排查。技术迭代很快但 LangGraph 这种“图心态”一旦建立你在面对任何复杂多智能体应用时都能游刃有余。 LangGraph多智能体实战：条件路由、子图与状态编排解析
看到标题里的“吊打付费”四个字先别急着点收藏。LangGraph 这个框架真正硬核的地方不在于它有多“高级”而在于它把 AI Agent 从“单轮问答玩具”推进到了“可编排、可控制、可恢复的工程化工作流”。我结合最近给多个业务方落地多智能体系统的经验把 LangGraph 的核心组件、多智能体架构、条件路由、子图、并行分支、状态记忆逐一拆开配一份能直接跑起来的实战代码。不管你是刚接触 AI 大模型应用开发还是已经在用 LangChain 写链式调用这篇文章都能让你少走很多弯路。1. 背景与核心概念1.1 LangGraph 到底是什么LangGraph 是 LangChain 社区推出的一个专门用于构建有状态 Agent 工作流的编排框架。它的名字里有两个关键词Lang 代表它和 LangChain 生态天然打通Graph 则揭示了它的底层模型——把整个智能体任务看成一张有向图Directed Graph。在上面的图里每个节点Node代表一个可执行单元比如“调用大模型”“读取数据库”“调用外部 API”“执行 Python 函数”每条边Edge代表控制流也就是从一个节点运行完之后下一步该去哪里。和传统 Chain 那种“一条线走到黑”的固定管道不同LangGraph 允许中间节点根据当前状态动态决定下一步走向这就是“条件路由Conditional Edge”的核心能力。用一个通俗的比喻来理解LangChain Chain像一条传送带物料从入口放上去沿着固定路线依次经过加工台最后从出口出来。LangGraph像一间工厂的中央调度室每个工位都有自己的职责调度员会根据当前物料的状态、质检结果、订单类型动态决定这个物料下一步去哪条产线、要不要返工、要不要几个工位并行处理。对 AI 大模型应用开发者来说这意味着你不再只能写“用户提问 - 调用模型 - 返回答案”这种固定流程而是可以搭建真正的多智能体系统一个 Agent 负责意图识别一个 Agent 负责工具调用一个 Agent 负责结果校验它们之间通过状态共享和条件路由协同工作。1.2 LangGraph 解决的真实痛点在实际项目里单次大模型调用往往不能满足业务需求。以智能客服系统为例用户先问“我的订单什么时候发货”客服 Agent 需要查订单系统发现订单已发货但用户又追问“那运单号是多少”Agent 需要记住这是同一个用户、同一个订单上下文再去查物流系统。如果用户对答案不满意Agent 需要自动转入人工投诉流程。这个过程涉及多轮对话状态维护、工具调用、条件判断、异常降级、人机切换如果纯粹靠 if-else 胶水代码去拼很快就会失控。LangGraph 通过状态对象State在不同节点之间传递数据通过 Checkpointer 实现状态持久化和断点恢复让这类复杂流程变得可编排、可观测、可恢复。1.3 LangGraph 与 LangChain 的区别这是很多初学者最容易混淆的地方。我直接用一张表说明对比维度LangChainLangGraph核心抽象Chain链StateGraph状态图控制流线性、固定分支、循环、并行、动态路由状态管理弱通常靠外部变量内置 State跨节点共享持久化不支持Checkpointer支持断点续跑适合场景简单 RAG、提示词模板、单 Agent 工具调用多智能体协作、复杂工作流、需要人工审批的长流程学习曲线较低中等简单说LangChain 解决的是“怎么让模型调用工具更简单”LangGraph 解决的是“怎么把多个模型调用和工具调用编排成一套健壮的分布式流程”。新项目中如果涉及多 Agent 协作我建议直接基于 LangGraph 起步而不是先搭好 LangChain 再迁移。1.4 多智能体的常见交互模式在正式写代码之前有必要了解一下多智能体系统中常见的四种交互模式这能帮助你判断自己的业务场景该用哪种图结构顺序协作模式SequentialAgent A 的输出作为 Agent B 的输入适合流水线式处理。比如先做意图分类再做实体抽取最后生成回复。路由分发模式Router一个主导 Agent 根据输入内容把任务分发给不同的专用 Agent。这是客服、工单系统最常见的模式。并行协作模式Parallel多个 Agent 同时处理同一任务的不同子任务最后汇总。比如同时生成多方言版本的回复或者同时检查代码和检查安全风险。层级监督模式Hierarchical有一个“主管 Agent”负责任务拆解和结果验收多个“执行 Agent”负责具体执行。这是复杂项目中最接近真实团队协作的模式但实现成本也最高。了解这些交互模式后你会发现它们几乎都可以用 LangGraph 的节点 条件边 子图 并行分支组合出来。接下来我们直接进入实操环节。2. 环境准备与版本说明2.1 运行环境说明本文的示例代码基于以下环境版本可以根据你的项目实际情况调整重点演示的是配置和实现思路操作系统Windows 10/11、macOS、Linux 均可本文以 Linux 终端命令为例Python 版本3.10 或更高建议 3.11包管理工具pip 或 uv核心依赖langgraph、langchain-core、langchain-openai或你本地部署模型的 SDK可选langgraph-cli、langsmith需要提醒的是LangGraph 的 API 在 0.1.x、0.2.x、0.3.x 之间有少量调整。本文示例以 0.2.x 和 0.3.x 通用写法为主如果遇到兼容性问题优先查看官方文档和 changelog。2.2 安装 LangGraph建议创建一个干净的虚拟环境python -m venv .venv source .venv/bin/activate # Windows 下使用 .venv\Scripts\activate pip install --upgrade pip pip install langgraph langchain-core langchain-openai如果你同时需要 LangGraph 的开发调试工具可以安装 CLIpip install langgraph-cli安装完成后可以用下面的命令验证版本python -c import langgraph; print(langgraph.__version__)我这里演示时安装的是0.3.x版本的 LangGraph。如果你后续引入了langgraph dev相关命令它本质上会启动一个带持久化存储的后端服务而直接uvicorn app:app启动普通的 FastAPI 应用则不会自动附加 LangGraph 的持久化和热重载能力。生产环境部署时根据是否需要状态管理来选择启动方式。2.3 示例项目结构为了后续实战代码更清晰推荐按下面的目录结构组织项目langgraph_agent_demo/ ├── agent_system/ │ ├── __init__.py │ ├── state.py # 定义全局 State │ ├── tools.py # 工具函数模拟订单/物流查询 │ ├── nodes.py # 各个节点函数 │ ├── subgraphs.py # 子图定义 │ └── graph.py # 主图组装 ├── .env # 存放模型 API Key └── main.py # 程序入口这样拆分的好处是每一个节点函数都可以独立测试后续扩展新 Agent 时不需要改主图逻辑只需要加节点、加边。3. LangGraph 核心组件深度拆解3.1 State全局状态State 是 LangGraph 的灵魂。它决定了节点之间如何传递数据。在实际工程中State 通常是一个 TypedDict 或者 dataclass。下面是一个典型的定义# 文件路径agent_system/state.py from typing import TypedDict, Annotated, List from langgraph.graph.message import add_messages def merge_lists(left: list, right: list) - list: 自定义状态合并函数用于多个节点并行写入同一字段时进行合并。 return left right class AgentState(TypedDict): # 用户原始输入 user_input: str # 意图识别结果 intent: str # 订单号 order_no: str # 节点间的中间消息列表 messages: Annotated[List[str], add_messages] # 最终回复 reply: str # 当前重试次数 attempts: int # 并行任务结果汇总 parallel_results: Annotated[List[str], merge_lists]这里的关键点在于Annotated[List[str], add_messages]这种写法。add_messages是 LangGraph 内置的状态归约器Reducer当多个节点试图写入同一个messages字段时会用追加的方式合并而不是直接覆盖。同理自定义的merge_lists函数也能控制并行分支回填状态的方式。为什么需要 Reducer因为在并行分支场景下多个节点会同时更新同一个状态字段如果没有合并策略后写入的会覆盖先写入的导致数据丢失。实际项目中凡是会被多个节点写入的列表字段都应该定义合适的 Reducer。3.2 Node工作流节点Node 就是普通的 Python 函数签名统一为(state: AgentState) - dict[str, Any]返回值会被合并到全局状态中。下面是一个最小节点示例# 文件路径agent_system/nodes.py from .state import AgentState def receive_input(state: AgentState) - dict: 接收用户输入并初始化相关字段。 print(f收到用户问题{state[user_input]}) return { messages: [f用户输入{state[user_input]}], attempts: 0, } def classify_intent(state: AgentState) - dict: 意图识别节点。 真实项目中这里应该调用大模型本示例用简单规则代替方便演示。 user_input state.get(user_input, ) if 订单 in user_input or 发货 in user_input: intent after_sales elif 投诉 in user_input or 不满 in user_input: intent complaint else: intent consult print(f意图识别结果{intent}) return {intent: intent, messages: [f意图{intent}]}注意Node 函数不直接修改外部全局变量而是返回一个字典告诉 LangGraph 哪些状态字段需要更新。这种设计保证了每个节点都是纯函数式的便于测试、回放和并行执行。3.3 Edge连接与路由Edge 用来定义节点之间的连接关系。基础用法很简单from langgraph.graph import StateGraph, START, END # 创建状态图 graph StateGraph(AgentState) # 添加节点 graph.add_node(receive_input, receive_input) graph.add_node(classify_intent, classify_intent) # 添加边 graph.add_edge(START, receive_input) graph.add_edge(receive_input, classify_intent) graph.add_edge(classify_intent, END)START是图入口END是图出口。这种写法形成了一条“开始 - 接收输入 - 识别意图 - 结束”的线性链路。如果所有流程都这么线性LangGraph 和普通 Chain 就没区别了。它的威力体现在add_conditional_edges上。3.4 Conditional Edge条件路由条件路由允许你根据当前状态动态决定下一步走向。这是实现 Router 模式、多智能体分工的基础。用法如下from typing import Literal from .state import AgentState def route_after_classify(state: AgentState) - Literal[consult, after_sales, complaint]: 根据意图将任务路由到不同 Agent。 intent state.get(intent, consult) # 这里返回的是下一步节点的名称 if intent after_sales: return after_sales_agent elif intent complaint: return complaint_agent return consult_agent # 在主图中添加条件路由 graph.add_conditional_edges( classify_intent, route_after_classify, { consult: consult_agent, after_sales: after_sales_agent, complaint: complaint_agent, } )关键点第一个参数是源节点名也就是执行完哪个节点后调用这个路由函数。第二个参数是路由函数它接收当前 State返回一个目标节点标识。第三个参数字典是可选的它把路由函数的返回值映射到实际的节点名。如果路由函数返回的值和节点名一致可以省略这个字典。在实际项目中路由函数通常由大模型驱动。比如在意图识别节点里模型返回 JSON{intent: refund}然后路由函数读取这个字段决定走退款流程还是人工客服。3.5 Checkpointer记忆与断点Checkpointer 是 LangGraph 相对较难理解但价值极高的组件。它的核心作用是保存图的运行状态快照从而支持三类能力多轮对话记忆同一线程thread_id的多轮交互共用同一份状态。断点恢复比如人工审批节点执行完毕后可以从断点继续往下走。时间旅行可以回放历史状态调试复杂工作流。LangGraph 提供了多种持久化后端常用的有MemorySaver内存存储适合开发测试。SqliteSaverSQLite 文件存储适合单机生产。PostgresSaverPostgreSQL 存储适合多实例部署。用法上只需要在编译时传入一个 Checkpointer 实例from langgraph.checkpoint.memory import MemorySaver memory MemorySaver() # 编译图时传入 checkpointer app graph.compile(checkpointermemory) # 运行时传入 thread_id用于区分不同会话 config {configurable: {thread_id: user-001}} result app.invoke({user_input: 我想查一下订单发货进度}, configconfig)这里需要特别说明一个热词InMemorySaver 中的内容如何参与大模型上下文。调用app.invoke()时State 里的messages字段会自动带上历史消息。假设你将messages设计为消息列表那么后续节点可以直接把state[messages]拼接成提示词发送给大模型。步骤如下def build_prompt(state: AgentState) - str: # 从 Checkpointer 恢复的消息历史 history_messages state.get(messages, []) # 把历史消息格式化成大模型可用的对话上下文 context \n.join([f{msg} for msg in history_messages]) prompt f以下是历史对话记录\n{context}\n\n当前用户问题{state[user_input]} return prompt换句话说Checkpointer 负责保存状态而状态里哪些字段需要作为大模型上下文完全由你决定。3.6 编译与执行图搭建完成后调用compile()得到一个可调用的CompiledStateGraph对象app graph.compile(checkpointermemory) # 同步调用 result app.invoke({user_input: 我的订单怎么还没发货}, configconfig) # 异步调用 # result await app.ainvoke({user_input: ...}, configconfig) # 流式输出 # for event in app.stream({user_input: ...}, configconfig): # print(event)compile()底层会做一次拓扑校验如果图中有不可达节点、环上缺少条件边、或节点函数签名不符合要求会在编译阶段报错。这也是 LangGraph 适合工程化的原因——很多结构性问题在运行前就能暴露。4. 完整实战案例智能客服工单多智能体系统4.1 需求分析我们构建一个“智能客服工单多智能体系统”完整覆盖以下几种真实场景咨询类问题用户询问退换货政策。系统调用咨询 Agent 直接回答。售后类问题用户询问订单发货进度。系统进入售后子图该子图尝试查询订单和物流信息如果查询失败则自动重试最多 3 次失败后转入人工处理。投诉类问题用户表达严重不满。系统同时启动“安抚话术生成”和“补偿方案生成”两个并行节点汇总后生成最终回复。所有对话记录通过 Checkpointer 持久化支持多轮追问联系上下文。整个系统的主图结构如下START | v receive_input | v classify_intent | v route_after_classify (条件路由) |--- consult_agent ---------- merge_result --- END |--- after_sales_subgraph --- merge_result --- END |--- complaint_agent --------- merge_result --- END其中after_sales_subgraph是一个子图complaint_agent内部有并行分支。4.2 创建项目结构mkdir -p langgraph_agent_demo/agent_system cd langgraph_agent_demo touch agent_system/__init__.py后续代码都放在agent_system目录下。4.3 定义状态先完善state.py# 文件路径langgraph_agent_demo/agent_system/state.py from typing import TypedDict, Annotated, List from langgraph.graph.message import add_messages def merge_lists(left: list, right: list) - list: 并行节点结果合并函数。 return left right class AgentState(TypedDict): user_input: str intent: str order_no: str attempts: int messages: Annotated[List[str], add_messages] parallel_results: Annotated[List[str], merge_lists] reply: str4.4 编写基础节点下面编写nodes.py包含主图的几个基础节点# 文件路径langgraph_agent_demo/agent_system/nodes.py from .state import AgentState def receive_input(state: AgentState) - dict: 接收用户输入。 print(f[receive_input] {state[user_input]}) return { messages: [f用户{state[user_input]}], attempts: 0, } def classify_intent(state: AgentState) - dict: 意图识别节点。 简化版用关键词判断真实项目可替换为 LLM 调用。 text state.get(user_input, ) if 投诉 in text or 不满 in text or 差评 in text: intent complaint elif 订单 in text or 发货 in text or 物流 in text or 退款 in text: intent after_sales else: intent consult print(f[classify_intent] 意图{intent}) return {intent: intent, messages: [f系统识别到意图 - {intent}]} def consult_agent(state: AgentState) - dict: 咨询 Agent直接基于知识库规则回答。 reply 我们的退换货政策是签收后 7 天内支持无理由退换货。请问还需要了解其他内容吗 print(f[consult_agent] 咨询 Agent 回复) return {reply: reply, messages: [f咨询Agent{reply}]} def complaint_agent(state: AgentState) - dict: 投诉 Agent 的入口节点。 print([complaint_agent] 启动投诉处理流程) # 这里只写入一个占位消息真正的并行逻辑在子图节点中处理 return {messages: [系统已进入投诉处理流程]} def merge_result(state: AgentState) - dict: 汇总节点把多个节点的输出合并成最终回复。 results state.get(parallel_results, []) if results: final_reply .join(results) else: final_reply state.get(reply, 抱歉暂时无法处理已转接人工客服。) print(f[merge_result] 最终回复{final_reply}) return {reply: final_reply, messages: [f系统{final_reply}]}为了让代码更接近真实场景我们在tools.py中准备几个模拟工具函数用于模拟订单查询、物流查询等外部系统# 文件路径langgraph_agent_demo/agent_system/tools.py import random def query_order(order_no: str) - dict: 模拟查询订单信息。 print(f调用订单系统order_no{order_no}) status_list [已支付, 已发货, 运输中, 已签收] return { order_no: order_no, status: random.choice(status_list), logistics_company: 顺丰速运, tracking_no: fSF{order_no}123456, } def simulate_success(probability0.6) - bool: 模拟外部接口调用成功率用于演示重试机制。 return random.random() probability4.5 编写售后子图子图Subgraph是 LangGraph 实现模块化的重要手段。你可以把一个负责“售后进度查询 重试机制”的流程封装成子图然后在主图中当作一个普通节点来使用。下面是子图的完整实现# 文件路径langgraph_agent_demo/agent_system/subgraphs.py from typing import TypedDict from langgraph.graph import StateGraph, START, END from .state import AgentState from .tools import query_order, simulate_success class AfterSalesState(TypedDict): order_no: str query_result: str attempts: int status: str # success / failed / max_retry def query_order_node(state: AfterSalesState) - dict: 查询订单与物流信息模拟可能失败。 order_no state.get(order_no, UNKNOWN) attempts state.get(attempts, 0) 1 if not simulate_success(probability0.5): print(f[售后子图] 第 {attempts} 次查询订单失败) return { query_result: f第 {attempts} 次查询失败, attempts: attempts, status: failed, } result query_order(order_no) text f订单 {order_no} 当前状态{result[status]}物流公司{result[logistics_company]}运单号{result[tracking_no]} print(f[售后子图] 查询成功{text}) return { query_result: text, attempts: attempts, status: success, } def check_retry(state: AfterSalesState) - str: 判断是否需要重试。 if state.get(status) success: return success if state.get(attempts, 0) 3: return max_retry return retry def build_after_sales_subgraph(): 构建售后处理子图。 subgraph StateGraph(AfterSalesState) # 添加节点 subgraph.add_node(query_order, query_order_node) # 添加入口和出口 subgraph.add_edge(START, query_order) # 条件路由支持循环重试 subgraph.add_conditional_edges( query_order, check_retry, { success: END, retry: query_order, # 回到自身节点形成循环 max_retry: END, }, ) return subgraph.compile()这个子图体现了 LangGraph 的循环检测能力。由于check_retry是一个条件边LangGraph 能识别出“从 query_order 到 query_order”的循环边只要循环存在出口本例中的success和max_retry图就能正常编译和运行不会进入死循环。4.6 编写并行分支节点在投诉处理场景中我们希望同时生成“安抚话术”和“补偿方案”然后在汇合节点统一汇总。LangGraph 天然支持这种扇出fan-out汇聚fan-in结构# 文件路径langgraph_agent_demo/agent_system/parallel_nodes.py from .state import AgentState def generate_comfort_message(state: AgentState) - dict: 并行分支1生成安抚话术。 comfort 非常抱歉给您带来了不好的体验我们已经高度重视您的问题。 print(f[并行分支1] 安抚话术生成完毕) return { parallel_results: [f安抚话术{comfort}], messages: [系统安抚话术生成完毕], } def generate_compensation(state: AgentState) - dict: 并行分支2生成补偿方案。 compensation 针对本次问题我们为您提供 20 元无门槛优惠券作为补偿。 print(f[并行分支2] 补偿方案生成完毕) return { parallel_results: [f补偿方案{compensation}], messages: [系统补偿方案生成完毕], }这里的parallel_results字段配置了merge_lists归约器所以两个并行节点返回的列表会被合并而不是后者覆盖前者。这是 LangGraph 并行分支中最需要注意的细节。4.7 组装主图现在把所有部分组装成完整的主图graph.py# 文件路径langgraph_agent_demo/agent_system/graph.py from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver from typing import Literal from .state import AgentState from .nodes import ( receive_input, classify_intent, consult_agent, complaint_agent, merge_result, ) from .parallel_nodes import generate_comfort_message, generate_compensation from .subgraphs import build_after_sales_subgraph # 构建售后子图 after_sales_subgraph_app build_after_sales_subgraph() def route_after_classify(state: AgentState) - Literal[consult_agent, after_sales_subgraph, complaint_agent]: 根据意图分发到不同 Agent。 intent state.get(intent, consult) if intent after_sales: return after_sales_subgraph elif intent complaint: return complaint_agent return consult_agent def build_main_graph(): graph StateGraph(AgentState) # 添加主图节点 graph.add_node(receive_input, receive_input) graph.add_node(classify_intent, classify_intent) graph.add_node(consult_agent, consult_agent) graph.add_node(complaint_agent, complaint_agent) # 子图作为节点加入主图 graph.add_node(after_sales_subgraph, after_sales_subgraph_app) # 并行分支节点 graph.add_node(generate_comfort_message, generate_comfort_message) graph.add_node(generate_compensation, generate_compensation) # 汇总节点 graph.add_node(merge_result, merge_result) # 入口 - 接收输入 - 意图识别 graph.add_edge(START, receive_input) graph.add_edge(receive_input, classify_intent) # 条件路由分发到不同 Agent graph.add_conditional_edges( classify_intent, route_after_classify, { consult_agent: consult_agent, after_sales_subgraph: after_sales_subgraph, complaint_agent: complaint_agent, }, ) # 咨询 Agent 和售后子图完成后都进入汇总节点 graph.add_edge(consult_agent, merge_result) graph.add_edge(after_sales_subgraph, merge_result) # 投诉 Agent启动并行分支 graph.add_edge(complaint_agent, generate_comfort_message) graph.add_edge(complaint_agent, generate_compensation) # 并行分支汇合到汇总节点 graph.add_edge(generate_comfort_message, merge_result) graph.add_edge(generate_compensation, merge_result) # 汇总节点输出 graph.add_edge(merge_result, END) return graph.compile(checkpointerMemorySaver()) main_app build_main_graph()这里其实有一个小的补充如果希望售后子图的输出直接回填到主图的reply字段需要把主图节点函数和子图 State 的结构对齐。在上面的示例中子图内部只维护了query_result字段而主图状态中的reply没有被更新因此汇总节点的兜底逻辑会生效“抱歉暂时无法处理”。在实际项目中建议在子图内部返回一个与主图 State 对应的字段比如reply或者通过封装节点函数做字段映射。4.8 运行与验证在项目根目录创建main.py# 文件路径langgraph_agent_demo/main.py from agent_system.graph import main_app def run_demo(): # 模拟三个用户的会话每个会话都有独立的 thread_id状态隔离 sessions [ {thread_id: user-001, user_input: 我想了解一下退换货政策}, {thread_id: user-002, user_input: 我的订单怎么还没发货订单号是10086}, {thread_id: user-003, user_input: 我要投诉你们服务太差了}, ] for session in sessions: thread_id session[thread_id] user_input session[user_input] config {configurable: {thread_id: thread_id}} print(f\n 新会话 {thread_id} ) result main_app.invoke( {user_input: user_input}, configconfig, ) print(f\n最终回复{result[reply]}) # 第二轮回访测试 Checkpointer 记忆恢复 if thread_id user-002: print(\n--- 用户追问 ---) result2 main_app.invoke( {user_input: 那什么时候能送到}, configconfig, ) print(f追问回复{result2[reply]}) if __name__ __main__: run_demo()运行结果示意如下 新会话 user-001 [receive_input] 我想了解一下退换货政策 [classify_intent] 意图consult [consult_agent] 咨询 Agent 回复 最终回复我们的退换货政策是签收后 7 天内支持无理由退换货。请问还需要了解其他内容吗 新会话 user-002 [receive_input] 我的订单怎么还没发货订单号是10086 [classify_intent] 意图after_sales [售后子图] 第 1 次查询订单失败 [售后子图] 第 2 次查询订单失败 [售后子图] 第 3 次查询订单失败 最终回复抱歉暂时无法处理已转接人工客服。 新会话 user-003 [receive_input] 我要投诉你们服务太差了 [classify_intent] 意图complaint [complaint_agent] 启动投诉处理流程 [并行分支1] 安抚话术生成完毕 [并行分支2] 补偿方案生成完毕 最终回复安抚话术非常抱歉给您带来了不好的体验我们已经高度重视您的问题。补偿方案针对本次问题我们为您提供 20 元无门槛优惠券作为补偿。从运行结果可以看出三点条件路由正确、售后子图循环重试机制生效、投诉场景并行分支正确合并。如果你第二次运行 user-002 的会话会发现子图状态从 Checkpointer 恢复attempts会被重置还是保留取决于子图状态初始化逻辑这一点需要开发者在设计子图时明确状态初始化时机。5. 常见问题与排查思路LangGraph 上手虽然快但实际开发中还是有不少容易踩的坑。我把高频问题整理成一张排查表方便你对照处理问题现象常见原因解决思路编译报错InvalidUpdateError某个字段被多个节点同时写入但没有定义 Reducer在 State 中使用Annotated[list, add_messages]或自定义合并函数图进入死循环条件边把状态送回上游节点但缺少终止条件检查路由函数的返回值务必保证存在能通向END的分支多轮对话上下文丢失没有传thread_id或没有配置 Checkpointer编译图时传入checkpointer调用invoke时传config{configurable: {thread_id: ...}}子图无法访问父图状态子图使用了自己的 State 类型与父图字段名不一致设计子图 State 时保持关键字段命名一致或写转换函数并行分支结果被覆盖并行节点写入同一个列表字段但没有使用 Reducer为列表字段配置合并策略模型调用超时导致整个图失败未对 Node 做超时和异常处理在节点函数内部使用try/except并设置合理的模型调用超时时间条件路由返回的节点名不存在路由函数返回值与add_node的节点名不一致检查路由分支字典的映射保持名称完全一致langgraph dev与普通 FastAPI 启动行为不同开发服务和普通 Web 服务对状态持久化的处理不同开发环境用uv run langgraph dev生产环境明确是否启用持久化存储这里单独说一下InvalidUpdateError这个问题。很多初学者搭建第一个高并发 Agent 时会遇到典型的错误提示是langgraph.errors.InvalidUpdateError: Expected list, got str这个报错的本质是State 中某个字段被定义为列表但某个节点返回了字符串。比如class State(TypedDict): chat_history: Annotated[list, add_messages] def node_a(state: State): return {chat_history: hello} # 错误应该返回 [hello]只要保持返回值和 State 类型一致这类问题就能避免。6. 最佳实践与工程建议6.1 状态设计拒绝“万能 State”很多开发者习惯把所有中间变量都塞进 State最终导致状态臃肿、难以追踪。我的建议是State 只保存需要跨节点传递的、会影响路由或最终输出的数据。临时中间结果比如一批原始 JSON尽量在节点函数内部消化不要全部塞进 State。列表字段必须设计 Reducer避免并行写入覆盖。6.2 子图拆分以业务边界为准不要为了“炫技”把图拆得过于复杂。子图拆分的合理依据是业务边界比如“售后处理”是一个完整的子流程内部有重试逻辑“投诉处理”是另一个完整子流程内部有并行计算。一个合格的多智能体系统主图的节点数量通常控制在 5~8 个更多细节应该沉淀到子图中。6.3 外部系统调用重试与降级在真实项目中Agent 的节点函数往往要调用订单系统、物流系统、支付系统等外部服务。这些服务不可能 100% 可用所以每个外部调用节点都应该做三件事超时控制给 HTTP 客户端设置合理的连接超时和读超时。重试策略利用 LangGraph 的条件边实现有限次重试而不是在节点内部写死 while 循环。降级方案重试失败后必须给出降级回复不能直接抛异常导致整个图崩溃。本文的售后子图就是一个很好的参考模板。你可以把simulate_success替换成真实的 HTTP 调用并在query_order_node里捕获外部异常。6.4 Checkpointer 的正确用法Checkpointer 在开发调试阶段用MemorySaver最方便但生产环境要注意单实例部署可以用SqliteSaver保证服务重启后状态不丢失。多实例部署必须使用PostgresSaver或 Redis 这类共享存储否则不同实例之间看不到彼此的会话状态。涉及用户隐私的会话数据建议在 Checkpointer 层做加密或脱敏。另外一个容易被忽略的点Checkpointer 保存的 State 会包含模型响应消息、工具调用中间结果等敏感信息。在把 State 写入日志或传给前端时需要显式过滤。6.5 安全边界与大模型风险控制只要涉及大模型调用就不能忽略安全问题。结合 LangGraph 的实际工程特性我给出以下清单提示词注入防护用户在输入中提到“忽略上述指令”时节点应拒绝执行敏感操作。权限最小化子图中的工具调用节点应使用独立的 API Key访问范围最小化避免一个 Agent 拥有全量系统权限。输出校验模型生成的回复在返回用户之前建议经过内容安全过滤和关键词校验。人工审批涉及退款、删除操作时加入“人工确认”节点。可以用interrupt_before实现挂起等待人工审批后恢复。# 编译时挂起示例 app graph.compile( checkpointermemory, interrupt_before[execute_refund], # 到达退款节点前挂起 )6.6 日志与可观测性LangGraph 生态支持 LangSmith 的完整链路追踪但如果团队没有接入第三方平台也可以通过自定义 Node 的日志实现基础的可观测性。建议每个节点统一打印结构化日志import json import time def log_node_entry(node_name: str, state: dict): print(json.dumps({ event: node_entry, node: node_name, timestamp: time.time(), intent: state.get(intent), attempts: state.get(attempts), }, ensure_asciiFalse))6.7 测试策略多智能体系统的测试和普通单测不太一样。除了单元测试节点函数还应该做图结构测试编译时无异常、节点覆盖率达到预期。路由矩阵测试构造不同输入验证意图识别和条件路由结果符合预期。故障注入测试模拟外部服务异常验证重试和降级逻辑。状态恢复测试同一个thread_id下验证多轮对话上下文正确恢复。建议用pytest组织这些测试。LangGraph 的纯函数式节点设计让单测变得非常简单不需要 mock 复杂的内部状态。7. 总结与学习路线写完这套智能客服多智能体系统你应该已经掌握了 LangGraph 最核心的思维方式和编码套路State 管理节点间数据、普通边保证线性流程、条件边实现动态路由和循环、子图承载独立业务模块、并行分支提高处理效率、Checkpointer 提供会话记忆和断点恢复。对比纯 LangChain 的链式调用这种基于图结构的编排方式在面对复杂业务时优势非常明显尤其是路由、重试、人工审批、多 Agent 协作这些场景代码的可维护性和可观测性都上了一个台阶。接下来你可以按这个顺序继续深入把示例中的关键词意图识别替换成真实的大模型调用感受 LLM 驱动路由的实际效果。将MemorySaver换成SqliteSaver部署一个带持久化的本地服务。研究 LangGraph 的interrupt_before/interrupt_after实现人工审批中断恢复。尝试写一个层级监督模式的多智能体系统主管 Agent 拆解任务多个执行 Agent 并行处理主管统一验收。如果本文对你有帮助可以收藏备用。实际动手时如果遇到新的报错场景欢迎在评论区把日志贴出来我们一起排查。技术迭代很快但 LangGraph 这种“图心态”一旦建立你在面对任何复杂多智能体应用时都能游刃有余。