尧图网站建设 
LangGraph 是 LangChain 生态中用于构建有状态、可编排智能体工作流的核心框架。很多初学者从简单的链式调用开始学习 LangChain等真正进入 LangGraph 后才发现状态如何传递、条件分支如何路由、循环如何终止、子图如何与父图交互才是实际项目里最关键的难点。这篇文章不从概念堆砌开始而是沿着一条可复现的路线先理解图的执行模型再动手写一个带条件路由、循环控制、子图和状态更新的完整工作流最后给出调试路径和常见问题排查方法。读完以后你可以把 LangGraph 用于意图路由、多步骤任务编排、Agent 状态管理等实际场景而不再是只会在示例里复制粘贴。如果原始材料没有给出明确版本落地前要先确认依赖版本。文中所有代码都基于 Python 环境代码内容用于说明 LangGraph 的核心用法实际项目需要结合自己的包名、目录结构和依赖版本调整。1. 先把 LangGraph 的执行模型讲清楚节点、状态和边的运行顺序LangGraph 不是一个“调用一次就返回结果”的普通工具它把智能体流程建模成一张图。图里有节点node、边edge、状态state和条件边conditional edge执行时由运行时按照边的关系调度节点。理解这张图的运行机制是后续写条件路由、子图和并行分支的基础。1.1 LangGraph 解决什么问题传统 LangChain 的Chain是固定的顺序调用A 完成后执行 BB 完成后执行 C。但真实业务里流程往往不是线性的。用户问一个问题系统要先判断意图再决定是查数据库、调用外部 API 还是让模型直接回答回答之前可能还要检查历史会话、重试两次、人工介入。这种带分支、有状态、可能循环的流程用普通 Chain 写会非常别扭。LangGraph 把流程建模为状态机。节点函数负责处理状态并返回更新边决定下一个节点是谁状态对象在整张图执行期间持续存在。这样做的好处是流程分支不再通过if/else散落在代码里而是显式表达在图结构中。循环可以建模成边不需要递归地调用自身。状态集中管理调试时可以查看每一步之后状态发生了什么变化。1.2 StateGraph 中的 State 是唯一数据源在 LangGraph 里状态是整个图的“唯一数据源”。每一个节点函数的输入都是一个状态对象返回值会被合并回状态然后传给下一个节点。状态通常使用 TypedDict 定义。例如from typing import TypedDict class MyState(TypedDict): user_input: str intent: str messages: list这里的user_input只存字符串messages存历史消息。每次节点执行时LangGraph 会读取当前状态传给节点函数节点函数返回一个字典LangGraph 把它合并到当前状态里。这里有一个容易误解的地方节点返回的字典不是替换整个状态而是类似更新操作。默认情况下同名字段会被覆盖列表字段也默认被覆盖如果你希望messages字段是追加而不是覆盖需要给字段声明 reducer这会在后面详细介绍。1.3 节点函数和边的关系节点函数是普通 Python 函数通常长这样def analyze_input(state: MyState) - dict: text state[user_input] return {intent: classify(text)}它接收当前状态返回要更新的字段。LangGraph 不会要求节点函数必须修改状态如果你返回空字典也不会报错只是本轮没有状态更新。边则是节点之间的连接关系。普通边表示“执行完 A 后无条件执行 B”。条件边表示“执行完 A 后根据状态或函数返回值动态决定下一个节点是哪一个”。1.4 图编译与运行compile 和 invokeLangGraph 的图不是定义完就能直接调用需要先编译from langgraph.graph import StateGraph, START, END graph StateGraph(MyState) graph.add_node(analyze, analyze_input) graph.add_node(respond, respond) graph.add_edge(START, analyze) graph.add_edge(analyze, respond) graph.add_edge(respond, END) app graph.compile() result app.invoke({user_input: 你好})compile()会返回一个可执行的CompiledStateGraph。invoke()是同步执行接口适合常规业务调用异步场景可以使用ainvoke()。执行过程可以理解为传入初始状态。从START节点出发。进入analyze节点执行函数合并返回值。根据边关系进入respond。到达END返回最终状态。你不需要手动维护“当前走到哪一步”的变量LangGraph 运行时已经替你做了这件事。2. 环境准备安装 LangGraph 并确认 Python 版本在写第一个图之前先把环境准备好。小项目在 Jupyter Notebook 或本地 Python 虚拟环境里跑都可以但生产项目建议使用虚拟环境和明确的依赖锁定文件。2.1 安装前需要确认什么LangGraph 基于 Python 3.9 以上版本。如果你的机器上已经安装了 Anaconda建议创建一个独立环境不要直接装在系统 Python 里否则后续升级依赖时容易出现冲突。检查当前 Python 版本python --version建议使用 3.10 或 3.11。这两个版本对 LangGraph 的支持比较成熟而且许多 LangChain 生态包都默认兼容。2.2 安装 langgraph使用 pip 安装即可pip install langgraph如果需要使用消息列表、工具调用等能力通常会再安装 LangChain 核心包pip install langchain-core如果你的环境需要调用 OpenAI、Anthropic 等模型的接口还需要安装对应的 SDK。这里不绑定任何厂商只说明常见的依赖关系包名作用是否需要langgraph核心图执行引擎必须langchain-core提供消息、提示词、工具等基础抽象强烈建议langchain-openaiOpenAI 模型封装按需langchain-anthropicAnthropic 模型封装按需langgraph-checkpoint持久化状态和断点恢复生产环境建议2.3 验证安装是否成功安装完成后在 Python 里检查导入和版本信息import langgraph print(langgraph.__version__)如果输出版本号说明安装成功。如果提示模块不存在说明当前解释器不是 pip 安装时使用的那个环境需要检查虚拟环境是否激活。2.4 学习环境和生产环境的差异学习环境的重点是快速跑通一台普通笔记本、一个虚拟环境、一个 Jupyter Notebook 就够。生产环境还需要考虑状态是否要持久化。如果任务执行到一半进程重启如何恢复是否需要重试和超时控制。节点函数里的外部调用如何管理连接池。日志和监控是否能覆盖每个节点的执行时间。模型 API 的限流和配额。这些不是一开始就要全部实现但至少要在设计图结构时预留接口。后面专门有一节讲生产化建议。3. 编写第一个可运行工作流状态定义、节点和边的最简组合这一节的目标是写一个能真正跑起来的最简例子。不要急着加入复杂路由先把最基础的节点、状态、边的组合跑通。3.1 定义状态类型下面示例中State 包含两个字段question和answer。这个例子足够小能清楚看到状态如何传递。from typing import TypedDict class QnAState(TypedDict): question: str answer: str3.2 定义节点函数第一个节点负责记录进入时间或处理问题第二个节点负责回答。这里的回答用简单字符串拼接便于观察。from datetime import datetime def log_question(state: QnAState) - dict: return {answer: f收到问题{state[question]}处理时间{datetime.now()}} def generate_answer(state: QnAState) - dict: # 实际项目中这里可能调用 LLM 或查询知识库 return {answer: f答案你问的是『{state[question]}』}注意log_question和generate_answer都返回了answer字段。因为默认情况下后一个返回值会覆盖前一个所以最终结果来自generate_answer。如果你不确定“覆盖”行为运行后打印结果就能看到。3.3 创建图并连接边LangGraph 的执行顺序由边决定。下面的代码把流程定义为START - log_question - generate_answer - END。from langgraph.graph import StateGraph, START, END graph StateGraph(QnAState) graph.add_node(log, log_question) graph.add_node(answer, generate_answer) graph.add_edge(START, log) graph.add_edge(log, answer) graph.add_edge(answer, END) app graph.compile()3.4 运行并观察结果调用invoke并传入初始状态result app.invoke({question: LangGraph 是什么}) print(result)预期输出大致是{question: LangGraph 是什么, answer: 答案你问的是『LangGraph 是什么』}如果你在generate_answer前打印状态会发现第二个节点拿到的是已经被第一个节点更新过的状态。这就证明状态确实在节点之间传递。3.5 这段代码的执行流程拆解执行流程可以拆成五步调用方传入{question: ...}LangGraph 把这个字典作为初始状态。找到START顺着边进入log节点。log_question返回{answer: 收到问题...}运行时更新状态。顺着边进入answer节点generate_answer拿到的是更新后的状态。返回最终状态。这里最关键的理解是节点函数的输入是“当前状态”输出是“要更新的字段”而不是“下一个节点”。下一个节点由边决定不在节点函数内部决定。如果你想根据内容动态决定下一步就需要使用条件边这是下一节的内容。4. 条件路由与分支控制conditional_edges 深度解析条件边是 LangGraph 里最常用的能力之一也是从入门走向实战必须掌握的分水岭。很多复杂功能比如意图识别后分流、循环重试、并行分支都可以通过条件边实现。4.1 什么时候需要条件边如果你遇到以下场景普通边就不够用了用户输入先做意图分类分类结果决定走“查询订单”还是“转人工”。大模型回答前先检查是否有记忆有记忆走带上下文的节点没有记忆走普通回答。某个任务失败后根据重试次数决定是继续重试还是返回错误。多个节点之间需要形成循环直到某个条件满足才退出。条件边的作用是在一条有向边上挂一个路由函数执行完当前节点后调用路由函数根据返回值选择下一个节点。4.2 conditional_edges 的签名与返回值规则在 LangGraph 中条件边使用add_conditional_edges添加。典型形式如下graph.add_conditional_edges( analyze, route_by_intent, { order: handle_order, chat: chat_respond, human: transfer_human, } )这段代码的意思是analyze节点执行完成后调用route_by_intent(state)然后根据函数返回值在映射表里找到下一个节点。route_by_intent的返回值必须和映射表的键一致。如果返回order就走handle_order如果返回表里没有的键比如unknowLangGraph 默认会报错。所以路由函数里通常要给一个兜底值。4.3 条件路由示例意图分类后分流假设工作流分为三步识别意图、根据意图执行、生成最终回复。状态定义from typing import TypedDict, Literal class RouteState(TypedDict): user_input: str intent: str response: str路由函数def route_by_intent(state: RouteState) - Literal[order, chat, human]: if state[intent] 查订单: return order if state[intent] 闲聊: return chat return human节点函数def detect_intent(state: RouteState) - dict: # 生产环境会用 LLM 或分类模型这里直接示例 text state[user_input] if 订单 in text: return {intent: 查订单} return {intent: 闲聊} def handle_order(state: RouteState) - dict: return {response: 你有一个待发货订单。} def chat_respond(state: RouteState) - dict: return {response: 今天天气不错有什么可以帮你} def transfer_human(state: RouteState) - dict: return {response: 正在为你转接人工客服。}图结构from langgraph.graph import StateGraph, START, END graph StateGraph(RouteState) graph.add_node(detect, detect_intent) graph.add_node(order, handle_order) graph.add_node(chat, chat_respond) graph.add_node(human, transfer_human) graph.add_edge(START, detect) graph.add_conditional_edges( detect, route_by_intent, { order: order, chat: chat, human: human, } ) graph.add_edge(order, END) graph.add_edge(chat, END) graph.add_edge(human, END) app graph.compile()运行result app.invoke({user_input: 我想查订单}) print(result)这里要注意add_conditional_edges之后还要为每个目标节点添加到END的边否则图会缺少终止路径。如果后面还要接同一个汇总节点也可以统一连到汇总节点。4.4 循环检测用条件边实现安全终止条件边的另一个重要用途是构建循环。LangGraph 本身不做“最多执行几次”的限制如果条件边一直返回同一个节点图就会一直循环。一个典型场景是“检查回答质量不合格就重写最多重试三次”。实现方式是在图里加一个validator节点它判断结果是否合格同时记录重试次数。状态from typing import TypedDict class LoopState(TypedDict): input_text: str draft: str retry_count: int finished: bool节点def generate_draft(state: LoopState) - dict: return {draft: f草稿第{state[retry_count]}次} def validate_draft(state: LoopState) - dict: # 如果 retry_count 小于 2继续重写如果大于等于 2强制结束 return {finished: state[retry_count] 2, retry_count: state[retry_count] 1}路由def route_after_validate(state: LoopState) - str: if state[finished]: return end return generate图graph StateGraph(LoopState) graph.add_node(generate, generate_draft) graph.add_node(validate, validate_draft) graph.add_edge(START, generate) graph.add_edge(generate, validate) graph.add_conditional_edges( validate, route_after_validate, { generate: generate, end: END, } ) app graph.compile()执行时validate每次会把retry_count加 1第三次执行validate后finished变为 True路由函数返回end图终止。这里的核心是循环出口必须由状态数据决定而且状态里每次都要有变化否则就是死循环。4.5 并行分支Fan-out 的两种实现方式并行分支不是指多线程执行而是指 LangGraph 可以同时从一个节点扩展到多个目标节点。常见方式有两种一种是通过add_edge一对多连接另一种是通过SendAPI 在运行时批量生成目标节点。简单的一对多并列graph.add_edge(source_node, parallel_a) graph.add_edge(source_node, parallel_b) graph.add_edge(source_node, parallel_c)三个节点都能从source_node拿到同一份状态。它们的返回值会在后续合并时决定如何更新状态。如果三个节点都修改同一个字段LangGraph 需要注意reducer的合并策略。使用SendAPI 可以从一个节点动态创建多个任务适合对列表中的每一项分别处理。示例from langgraph.types import Send def dispatch_nodes(state: LoopState) - list[Send]: return [Send(process_item, {item: item}) for item in state[input_text]]dispatch_nodes是一个节点它返回一组Send对象。每个Send包含目标节点名称和要传给该节点的状态。这种写法适合数据并行场景比如把一个文档列表拆成多个子任务每个子任务独立处理。并行分支的常见坑是多个子节点同时修改同一个状态字段时最后一个写入方的结果可能覆盖前面的结果。如果需要保留所有结果状态字段应该设计成列表并使用 reducer 合并。5. 子图Subgraph把复杂工作流拆分到可管理单元当一张图包含十几个节点时全部平铺在一个StateGraph里会很难维护。子图允许你把一部分节点封装成一个独立的图再作为另一个图的节点嵌入。5.1 子图解决什么问题子图的主要作用是拆解复杂度。你可以在一个文件里定义“订单处理子图”在另一个文件里定义“售后子图”然后在主图里将它们作为节点挂载。这样每个子图可以独立测试主图只负责编排。另外当你需要复用一套流程时子图也比复制粘贴更合适。例如不同的入口节点都可能走“意图识别 - 槽位填充 - 执行动作”的流程就可以把这一段封装成子图。5.2 定义一个最简单的子图先定义一个内部子图from langgraph.graph import StateGraph, START, END class SubState(TypedDict): task: str sub_result: str def sub_node(state: SubState) - dict: return {sub_result: f子图处理{state[task]}} sub_graph StateGraph(SubState) sub_graph.add_node(sub_node, sub_node) sub_graph.add_edge(START, sub_node) sub_graph.add_edge(sub_node, END) sub_app sub_graph.compile()这里sub_app是一个CompiledStateGraph。5.3 在主图中嵌入子图主图的状态需要包含子图需要的数据且节点可以是一个CompiledStateGraph。class MainState(TypedDict): task: str sub_result: str final_result: str def after_sub(state: MainState) - dict: return {final_result: f最终结果{state[sub_result]}} main_graph StateGraph(MainState) main_graph.add_node(sub, sub_app) main_graph.add_node(after, after_sub) main_graph.add_edge(START, sub) main_graph.add_edge(sub, after) main_graph.add_edge(after, END) main_app main_graph.compile() result main_app.invoke({task: 分析用户情绪}) print(result)运行时主图进入sub节点相当于把当前状态交给子图执行子图执行完返回的状态会与主图状态合并。这里要特别注意子图的输入状态只能包含子图状态里定义的字段。如果子图里没有定义final_result即使主图有该字段子图也无法直接访问或修改。也就是说主图和子图之间是“按状态字段”解耦的而不是自动共享全部字段。5.4 子图与普通节点的区别从主图视角看子图就是一个节点。节点函数可以自定义输入输出而子图有自己完整的状态定义和执行流程。普通节点适合做轻量计算子图适合承载一套完整业务流程。使用子图时要注意状态匹配问题。如果主图状态字段名和子图状态字段名不一致要在传参时做转换。例如主图字段叫user_text子图字段叫task可以写一个转换节点def prepare_sub_state(state: MainState) - dict: return {task: state[user_text]}这样能更清晰地管理数据流而不会让字段名不一致导致运行时缺字段。6. 在节点函数中修改 State 的几种姿势很多初学者在写 LangGraph 节点函数时最大的困惑是“到底怎么修改状态”。 LangGraph 不允许你在节点函数内部直接对整个状态对象赋值它要求节点返回一个字典由运行时来合并。但这里的合并规则并不是每种字段都一样。6.1 最直接的方式返回字段字典当节点需要更新一个普通字段时直接返回包含该字段的字典即可def set_answer(state: MyState) - dict: return {answer: 新的答案}如果字段名不存在LangGraph 默认会新增一个字段。如果你不希望节点意外创建新字段可以在状态类型里明确声明所有字段。6.2 用 Annotated 声明 Reducer实现列表追加如果状态里有消息列表直接返回列表会覆盖而不是追加。比如class MessageState(TypedDict): messages: list某个节点返回{messages: [你好]}下一轮另一个节点也返回{messages: [有什么需要帮助吗]}最终状态里只会保留最后一个列表。要解决这个问题需要使用消息追加的 reducer。LangGraph 的常见写法是使用Annotated和operator.addfrom typing import TypedDict, Annotated import operator class MessageState(TypedDict): messages: Annotated[list, operator.add]此时每个节点返回的messages列表都会被合并到原列表后面。例如[hello] [world]最终是[hello, world]。如果你想自己控制合并逻辑也可以写一个自定义 reducer 函数。reducer 接收两个值旧值和新值返回合并后的结果。例如def merge_messages(old: list, new: list) - list: return old new然后用Annotated[list, merge_messages]声明即可。6.3 节点函数内部更新局部变量但不直接改 State节点函数里可以读取整个 state但不建议直接执行state[xxx] yyy这类语句。一个原因是 LangGraph 的执行环境可能对状态做并发控制直接修改外部字典对象容易产生不可预期的副作用另一个原因是调试时难以追踪状态变化。正确写法是def node_func(state: MyState) - dict: local_var state[user_input] 已处理 return {processed_input: local_var}如果你确实需要修改很多字段可以构建一个新字典统一返回def node_func(state: MyState) - dict: return { field_a: 1, field_b: 2, field_c: state[old] x, }6.4 常见错误与正确写法对比错误现象错误写法正确写法节点返回后状态没有变化函数内直接修改state对象不返回内容返回字典由运行时合并messages一直覆盖而不是追加返回{messages: [新消息]}且字段没有 reducer字段用Annotated[list, operator.add]子图返回后主图缺少字段子图状态没有声明该字段在子图状态类型中补充该字段循环里状态不更新每次返回相同的值在返回字典中包含自增次数或变化结果在调试状态更新问题时最有效的方法是打印每个节点的输入和输出状态。LangGraph 也支持在invoke时传入debug参数或者自定义回调来观察每个节点的状态变更。7. 一个完整实战构建带条件路由和子图的多轮问答工作流把前面的知识点串起来写一个更完整的示例。这个工作流模拟“智能客服”用户输入问题后先判断意图如果是订单相关问题进入订单子图查询如果要转人工直接转人工如果只是闲聊直接回复。7.1 需求拆解这个实战包含四个步骤输入用户问题。判断意图。根据意图分流订单子图、闲聊回复、人工转接。收集结果生成最终回复。用例可以通过但为了学习 LangGraph重点放在状态设计、条件路由和子图使用上。7.2 状态设计from typing import TypedDict, Annotated, Literal import operator class OrderState(TypedDict): order_id: str order_info: str class CustomerServiceState(TypedDict): user_input: str intent: Literal[order, chat, human] messages: Annotated[list, operator.add] order_result: str final_response: str这里messages使用追加 reducer便于记录整个对话过程。7.3 意图识别节点实际项目中可以用 LLM 分类这里用关键词模拟def detect_intent(state: CustomerServiceState) - dict: text state[user_input] if 订单 in text or 物流 in text: return {intent: order, messages: [识别为订单意图]} if 人工 in text or 客服 in text: return {intent: human, messages: [识别为人工意图]} return {intent: chat, messages: [识别为闲聊意图]}7.4 条件路由def route_after_detect(state: CustomerServiceState) - Literal[order, chat, human]: return state[intent]然后把三个分支节点都定义好def chat_reply(state: CustomerServiceState) - dict: return {final_response: 我是智能助手可以陪你聊聊天气、编程或日常问题。, messages: [执行闲聊回复]} def human_transfer(state: CustomerServiceState) - dict: return {final_response: 正在为你转接人工客服请稍候。, messages: [执行人工转接]}订单子图单独定义def query_order(state: OrderState) - dict: return {order_info: f订单 {state[order_id]} 已发货} order_graph StateGraph(OrderState) order_graph.add_node(query, query_order) order_graph.add_edge(START, query) order_graph.add_edge(query, END) order_app order_graph.compile()主图里接入子图前需要写一个适配节点把主图状态转换成子图需要的字段def order_subgraph_entry(state: CustomerServiceState) - dict: # 简化从输入中获取订单号 order_id SO123456 sub_result order_app.invoke({order_id: order_id}) return {order_result: sub_result[order_info], messages: [订单子图执行完成]}完整主图构造from langgraph.graph import StateGraph, START, END main_graph StateGraph(CustomerServiceState) main_graph.add_node(detect, detect_intent) main_graph.add_node(chat, chat_reply) main_graph.add_node(human, human_transfer) main_graph.add_node(order, order_subgraph_entry) main_graph.add_edge(START, detect) main_graph.add_conditional_edges( detect, route_after_detect, { order: order, chat: chat, human: human, } ) # 分支统一结束 main_graph.add_edge(order, END) main_graph.add_edge(chat, END) main_graph.add_edge(human, END) main_app main_graph.compile()这里子图不是完全黑盒主图通过order_subgraph_entry调用了order_app.invoke()相当于在一个普通节点里手动调用子图。这种方式的优点是控制精确缺点是该节点会阻塞等待子图完成。如果希望子图作为独立节点直接嵌入LangGraph 也支持把CompiledStateGraph作为节点加入主图但需要保证主图状态字段和子图状态字段兼容。7.5 运行验证运行三个测试输入test_cases [ 我想查一下订单号 SO123456 的物流, 你叫什么名字, 转人工客服 ] for case in test_cases: result main_app.invoke({user_input: case}) print(输入:, case) print(意图:, result[intent]) print(最终回复:, result[final_response]) print(消息记录:, result[messages]) print(---)预期结果大致是输入包含“订单”时走order分支返回订单子图结果。输入“你叫什么名字”时走chat分支。输入“转人工”时走human分支。如果某个分支没有按预期执行优先检查条件路由函数返回值是否与add_conditional_edges的映射表键一致。7.6 学习环境与生产环境的差异上面的代码已经把 LangGraph 的核心功能串起来了。但生产使用还需要补充以下能力能力学习环境生产环境状态存储内存进程重启即丢失使用 checkpointer 持久化到数据库模型调用本地或测试 key统一 API 网关、限流、重试日志print 临时查看结构化日志、链路追踪分支异常直接抛错兜底分支、人工介入、告警子图复用单文件示例独立模块化、单元测试检查点checkpointer是 LangGraph 的一个重要能力。它可以把图的状态保存下来支持断点恢复、时间旅行调试和人工审批流程。生产环境建议一开始就接入数据库存储比如使用SqliteSaver或生产级数据库实现。8. 常见问题排查与调试清单实际使用 LangGraph 时问题往往不在语法而在执行语义。这里整理几个高频问题和排查路径。8.1 节点执行了但状态里没有看到字段更新现象节点函数明明返回了字典但invoke结果里没有对应字段。排查步骤检查节点函数是否真的被调用了。可以在函数入口打印日志。检查返回的字典是否拼写错误。如果状态类型是 TypedDict字段名不一致不会报错。检查是否有其他节点在后续覆盖了同一个字段。检查是否使用了Annotated字段且 reducer 合并逻辑不符合预期。检查边是否跳过了该节点。如果节点连接到END后续节点自然拿不到它的返回值。8.2 条件边不执行或总是走默认分支现象条件路由后目标节点不是预期节点。排查步骤在路由函数里打印返回值确认返回的是字符串。检查返回值是不是包含空格、引号或类型不一致。比如返回Literal[order]但实际值是order 。检查add_conditional_edges的映射表里是否有该键。如果路由函数返回None也会导致映射失败需要给所有路径提供兜底返回值。确认使用的是add_conditional_edges而不是add_edge。8.3 图陷入死循环现象程序一直执行不返回。排查步骤查看循环节点的日志确认状态是否在每次循环中发生变化。检查循环出口条件。比如重试次数是否真的在递增。检查路由函数是否存在永远不会返回出口键的分支。为循环节点添加最大重试次数保护这是最稳妥的兜底方案。例如可以在循环节点内增加if state[retry_count] 5: return {finished: True}这样即使路由逻辑写错也能在有限次循环后强制退出。8.4 子图返回结果与父图状态不兼容现象子图执行完主图后续节点读不到子图返回的内容。排查步骤检查子图状态类型是否包含该字段。检查子图是否通过END正常返回。检查主图是否在子图后继续使用主图状态对象而不是子图返回的局部状态字典。如果通过节点函数调用子图需要把子图返回结果手动映射回主图字段。8.5 并行分支共享状态污染现象多个并行节点都修改messages结果顺序不确定或字段被覆盖。排查步骤为需要追加的字段添加 reducer比如Annotated[list, operator.add]。如果并行节点各自返回相同字段确认期望是覆盖还是合并。必要时为每个并行分支定义独立字段避免竞态。8.6 排查顺序清单当一张图运行结果和预期不一致时按下面顺序检查输入状态是否正确。节点函数是否都被调用。每个节点返回的字典是否正确。普通边和条件边的目标节点是否符合预期。状态字段是否被 reducer 正确合并。子图是否独立运行正常。最终返回状态里是否有预期字段。9. 最佳实践与进阶方向LangGraph 从示例到可用中间差的是工程化思维。以下实践建议同样适用于 LangChain 生态的其他工作流项目。9.1 状态设计先于节点设计很多初学者先写节点函数再思考状态类型导致状态字段混乱。建议先画一张表列出整个流程中哪些数据需要在节点之间共享哪些只是局部变量哪些需要追加哪些需要覆盖。可以参考这个状态字段设计清单字段名类型更新策略用途user_inputstr覆盖保存用户原始输入messageslist追加保存对话历史intentstr覆盖保存分类结果retry_countint覆盖控制循环次数tool_resultstr覆盖工具调用结果状态字段越少图越容易理解。能用局部变量解决的问题不要都放到全局状态里。9.2 节点函数尽量保持“纯函数”一个节点只做一件事输入和输出尽量可预测。不要在节点里写复杂的业务逻辑更不要让节点之间通过全局变量交换数据。推荐做法每个节点函数只读取自己需要的字段。每个节点函数返回明确的数据结构。对可能失败的调用增加 try/except再返回错误状态。这样既方便单元测试也方便后面接入日志和监控。9.3 设置最大执行步数和循环保护LangGraph 虽然提供了灵活的执行机制但生产环境必须防止失控循环。可以在编译后的图上设置递归限制或者在状态里增加步数计数器。具体限制值和业务相关但至少要有兜底。9.4 先本地调试再接入模型不要一开始就把所有节点都接上大模型。先用字符串拼接、规则匹配把流程跑通确认状态传递、条件路由、子图都正确再替换为真实模型调用。这样可以节省调试成本也更容易定位问题是出在流程编排还是模型输出。9.5 生产部署要重点关注持久化和可观测性LangGraph 入门到实战最终要落地的不仅是功能还包括可靠性。生产环境需要关注状态持久化使用 checkpointer让任务能断点续跑。日志每个节点记录输入、输出、耗时。监控图执行时间、失败率、重试次数。人工审批在关键节点设置中断等待人工确认后再继续执行。版本管理图的定义和依赖版本需要一起发布避免状态结构变更后旧任务无法恢复。9.6 学习路径建议如果你刚接触 LangGraph建议按这个顺序进阶完成本文的最简图和条件路由示例确保理解状态、节点、边的概念。自己设计一个三节点的流程加入循环和子图。把messages字段改成 reducer 追加理解状态合并规则。尝试接入一个真实 LLM让条件边根据模型输出路由。阅读 LangGraph 官方文档中关于 checkpointer 和持久化的部分实现一个带记忆的会话应用。在项目里抽象出可复用的子图并编写单元测试。LangGraph 的学习曲线不算陡难点在于从“链式思维”切换到“图思维”。只要把状态、边、条件路由这三件事理解透后面无论是构建 Agent、知识库问答还是复杂任务编排都能很快上手。 LangGraph实战：状态管理与条件路由工作流详解
LangGraph 是 LangChain 生态中用于构建有状态、可编排智能体工作流的核心框架。很多初学者从简单的链式调用开始学习 LangChain等真正进入 LangGraph 后才发现状态如何传递、条件分支如何路由、循环如何终止、子图如何与父图交互才是实际项目里最关键的难点。这篇文章不从概念堆砌开始而是沿着一条可复现的路线先理解图的执行模型再动手写一个带条件路由、循环控制、子图和状态更新的完整工作流最后给出调试路径和常见问题排查方法。读完以后你可以把 LangGraph 用于意图路由、多步骤任务编排、Agent 状态管理等实际场景而不再是只会在示例里复制粘贴。如果原始材料没有给出明确版本落地前要先确认依赖版本。文中所有代码都基于 Python 环境代码内容用于说明 LangGraph 的核心用法实际项目需要结合自己的包名、目录结构和依赖版本调整。1. 先把 LangGraph 的执行模型讲清楚节点、状态和边的运行顺序LangGraph 不是一个“调用一次就返回结果”的普通工具它把智能体流程建模成一张图。图里有节点node、边edge、状态state和条件边conditional edge执行时由运行时按照边的关系调度节点。理解这张图的运行机制是后续写条件路由、子图和并行分支的基础。1.1 LangGraph 解决什么问题传统 LangChain 的Chain是固定的顺序调用A 完成后执行 BB 完成后执行 C。但真实业务里流程往往不是线性的。用户问一个问题系统要先判断意图再决定是查数据库、调用外部 API 还是让模型直接回答回答之前可能还要检查历史会话、重试两次、人工介入。这种带分支、有状态、可能循环的流程用普通 Chain 写会非常别扭。LangGraph 把流程建模为状态机。节点函数负责处理状态并返回更新边决定下一个节点是谁状态对象在整张图执行期间持续存在。这样做的好处是流程分支不再通过if/else散落在代码里而是显式表达在图结构中。循环可以建模成边不需要递归地调用自身。状态集中管理调试时可以查看每一步之后状态发生了什么变化。1.2 StateGraph 中的 State 是唯一数据源在 LangGraph 里状态是整个图的“唯一数据源”。每一个节点函数的输入都是一个状态对象返回值会被合并回状态然后传给下一个节点。状态通常使用 TypedDict 定义。例如from typing import TypedDict class MyState(TypedDict): user_input: str intent: str messages: list这里的user_input只存字符串messages存历史消息。每次节点执行时LangGraph 会读取当前状态传给节点函数节点函数返回一个字典LangGraph 把它合并到当前状态里。这里有一个容易误解的地方节点返回的字典不是替换整个状态而是类似更新操作。默认情况下同名字段会被覆盖列表字段也默认被覆盖如果你希望messages字段是追加而不是覆盖需要给字段声明 reducer这会在后面详细介绍。1.3 节点函数和边的关系节点函数是普通 Python 函数通常长这样def analyze_input(state: MyState) - dict: text state[user_input] return {intent: classify(text)}它接收当前状态返回要更新的字段。LangGraph 不会要求节点函数必须修改状态如果你返回空字典也不会报错只是本轮没有状态更新。边则是节点之间的连接关系。普通边表示“执行完 A 后无条件执行 B”。条件边表示“执行完 A 后根据状态或函数返回值动态决定下一个节点是哪一个”。1.4 图编译与运行compile 和 invokeLangGraph 的图不是定义完就能直接调用需要先编译from langgraph.graph import StateGraph, START, END graph StateGraph(MyState) graph.add_node(analyze, analyze_input) graph.add_node(respond, respond) graph.add_edge(START, analyze) graph.add_edge(analyze, respond) graph.add_edge(respond, END) app graph.compile() result app.invoke({user_input: 你好})compile()会返回一个可执行的CompiledStateGraph。invoke()是同步执行接口适合常规业务调用异步场景可以使用ainvoke()。执行过程可以理解为传入初始状态。从START节点出发。进入analyze节点执行函数合并返回值。根据边关系进入respond。到达END返回最终状态。你不需要手动维护“当前走到哪一步”的变量LangGraph 运行时已经替你做了这件事。2. 环境准备安装 LangGraph 并确认 Python 版本在写第一个图之前先把环境准备好。小项目在 Jupyter Notebook 或本地 Python 虚拟环境里跑都可以但生产项目建议使用虚拟环境和明确的依赖锁定文件。2.1 安装前需要确认什么LangGraph 基于 Python 3.9 以上版本。如果你的机器上已经安装了 Anaconda建议创建一个独立环境不要直接装在系统 Python 里否则后续升级依赖时容易出现冲突。检查当前 Python 版本python --version建议使用 3.10 或 3.11。这两个版本对 LangGraph 的支持比较成熟而且许多 LangChain 生态包都默认兼容。2.2 安装 langgraph使用 pip 安装即可pip install langgraph如果需要使用消息列表、工具调用等能力通常会再安装 LangChain 核心包pip install langchain-core如果你的环境需要调用 OpenAI、Anthropic 等模型的接口还需要安装对应的 SDK。这里不绑定任何厂商只说明常见的依赖关系包名作用是否需要langgraph核心图执行引擎必须langchain-core提供消息、提示词、工具等基础抽象强烈建议langchain-openaiOpenAI 模型封装按需langchain-anthropicAnthropic 模型封装按需langgraph-checkpoint持久化状态和断点恢复生产环境建议2.3 验证安装是否成功安装完成后在 Python 里检查导入和版本信息import langgraph print(langgraph.__version__)如果输出版本号说明安装成功。如果提示模块不存在说明当前解释器不是 pip 安装时使用的那个环境需要检查虚拟环境是否激活。2.4 学习环境和生产环境的差异学习环境的重点是快速跑通一台普通笔记本、一个虚拟环境、一个 Jupyter Notebook 就够。生产环境还需要考虑状态是否要持久化。如果任务执行到一半进程重启如何恢复是否需要重试和超时控制。节点函数里的外部调用如何管理连接池。日志和监控是否能覆盖每个节点的执行时间。模型 API 的限流和配额。这些不是一开始就要全部实现但至少要在设计图结构时预留接口。后面专门有一节讲生产化建议。3. 编写第一个可运行工作流状态定义、节点和边的最简组合这一节的目标是写一个能真正跑起来的最简例子。不要急着加入复杂路由先把最基础的节点、状态、边的组合跑通。3.1 定义状态类型下面示例中State 包含两个字段question和answer。这个例子足够小能清楚看到状态如何传递。from typing import TypedDict class QnAState(TypedDict): question: str answer: str3.2 定义节点函数第一个节点负责记录进入时间或处理问题第二个节点负责回答。这里的回答用简单字符串拼接便于观察。from datetime import datetime def log_question(state: QnAState) - dict: return {answer: f收到问题{state[question]}处理时间{datetime.now()}} def generate_answer(state: QnAState) - dict: # 实际项目中这里可能调用 LLM 或查询知识库 return {answer: f答案你问的是『{state[question]}』}注意log_question和generate_answer都返回了answer字段。因为默认情况下后一个返回值会覆盖前一个所以最终结果来自generate_answer。如果你不确定“覆盖”行为运行后打印结果就能看到。3.3 创建图并连接边LangGraph 的执行顺序由边决定。下面的代码把流程定义为START - log_question - generate_answer - END。from langgraph.graph import StateGraph, START, END graph StateGraph(QnAState) graph.add_node(log, log_question) graph.add_node(answer, generate_answer) graph.add_edge(START, log) graph.add_edge(log, answer) graph.add_edge(answer, END) app graph.compile()3.4 运行并观察结果调用invoke并传入初始状态result app.invoke({question: LangGraph 是什么}) print(result)预期输出大致是{question: LangGraph 是什么, answer: 答案你问的是『LangGraph 是什么』}如果你在generate_answer前打印状态会发现第二个节点拿到的是已经被第一个节点更新过的状态。这就证明状态确实在节点之间传递。3.5 这段代码的执行流程拆解执行流程可以拆成五步调用方传入{question: ...}LangGraph 把这个字典作为初始状态。找到START顺着边进入log节点。log_question返回{answer: 收到问题...}运行时更新状态。顺着边进入answer节点generate_answer拿到的是更新后的状态。返回最终状态。这里最关键的理解是节点函数的输入是“当前状态”输出是“要更新的字段”而不是“下一个节点”。下一个节点由边决定不在节点函数内部决定。如果你想根据内容动态决定下一步就需要使用条件边这是下一节的内容。4. 条件路由与分支控制conditional_edges 深度解析条件边是 LangGraph 里最常用的能力之一也是从入门走向实战必须掌握的分水岭。很多复杂功能比如意图识别后分流、循环重试、并行分支都可以通过条件边实现。4.1 什么时候需要条件边如果你遇到以下场景普通边就不够用了用户输入先做意图分类分类结果决定走“查询订单”还是“转人工”。大模型回答前先检查是否有记忆有记忆走带上下文的节点没有记忆走普通回答。某个任务失败后根据重试次数决定是继续重试还是返回错误。多个节点之间需要形成循环直到某个条件满足才退出。条件边的作用是在一条有向边上挂一个路由函数执行完当前节点后调用路由函数根据返回值选择下一个节点。4.2 conditional_edges 的签名与返回值规则在 LangGraph 中条件边使用add_conditional_edges添加。典型形式如下graph.add_conditional_edges( analyze, route_by_intent, { order: handle_order, chat: chat_respond, human: transfer_human, } )这段代码的意思是analyze节点执行完成后调用route_by_intent(state)然后根据函数返回值在映射表里找到下一个节点。route_by_intent的返回值必须和映射表的键一致。如果返回order就走handle_order如果返回表里没有的键比如unknowLangGraph 默认会报错。所以路由函数里通常要给一个兜底值。4.3 条件路由示例意图分类后分流假设工作流分为三步识别意图、根据意图执行、生成最终回复。状态定义from typing import TypedDict, Literal class RouteState(TypedDict): user_input: str intent: str response: str路由函数def route_by_intent(state: RouteState) - Literal[order, chat, human]: if state[intent] 查订单: return order if state[intent] 闲聊: return chat return human节点函数def detect_intent(state: RouteState) - dict: # 生产环境会用 LLM 或分类模型这里直接示例 text state[user_input] if 订单 in text: return {intent: 查订单} return {intent: 闲聊} def handle_order(state: RouteState) - dict: return {response: 你有一个待发货订单。} def chat_respond(state: RouteState) - dict: return {response: 今天天气不错有什么可以帮你} def transfer_human(state: RouteState) - dict: return {response: 正在为你转接人工客服。}图结构from langgraph.graph import StateGraph, START, END graph StateGraph(RouteState) graph.add_node(detect, detect_intent) graph.add_node(order, handle_order) graph.add_node(chat, chat_respond) graph.add_node(human, transfer_human) graph.add_edge(START, detect) graph.add_conditional_edges( detect, route_by_intent, { order: order, chat: chat, human: human, } ) graph.add_edge(order, END) graph.add_edge(chat, END) graph.add_edge(human, END) app graph.compile()运行result app.invoke({user_input: 我想查订单}) print(result)这里要注意add_conditional_edges之后还要为每个目标节点添加到END的边否则图会缺少终止路径。如果后面还要接同一个汇总节点也可以统一连到汇总节点。4.4 循环检测用条件边实现安全终止条件边的另一个重要用途是构建循环。LangGraph 本身不做“最多执行几次”的限制如果条件边一直返回同一个节点图就会一直循环。一个典型场景是“检查回答质量不合格就重写最多重试三次”。实现方式是在图里加一个validator节点它判断结果是否合格同时记录重试次数。状态from typing import TypedDict class LoopState(TypedDict): input_text: str draft: str retry_count: int finished: bool节点def generate_draft(state: LoopState) - dict: return {draft: f草稿第{state[retry_count]}次} def validate_draft(state: LoopState) - dict: # 如果 retry_count 小于 2继续重写如果大于等于 2强制结束 return {finished: state[retry_count] 2, retry_count: state[retry_count] 1}路由def route_after_validate(state: LoopState) - str: if state[finished]: return end return generate图graph StateGraph(LoopState) graph.add_node(generate, generate_draft) graph.add_node(validate, validate_draft) graph.add_edge(START, generate) graph.add_edge(generate, validate) graph.add_conditional_edges( validate, route_after_validate, { generate: generate, end: END, } ) app graph.compile()执行时validate每次会把retry_count加 1第三次执行validate后finished变为 True路由函数返回end图终止。这里的核心是循环出口必须由状态数据决定而且状态里每次都要有变化否则就是死循环。4.5 并行分支Fan-out 的两种实现方式并行分支不是指多线程执行而是指 LangGraph 可以同时从一个节点扩展到多个目标节点。常见方式有两种一种是通过add_edge一对多连接另一种是通过SendAPI 在运行时批量生成目标节点。简单的一对多并列graph.add_edge(source_node, parallel_a) graph.add_edge(source_node, parallel_b) graph.add_edge(source_node, parallel_c)三个节点都能从source_node拿到同一份状态。它们的返回值会在后续合并时决定如何更新状态。如果三个节点都修改同一个字段LangGraph 需要注意reducer的合并策略。使用SendAPI 可以从一个节点动态创建多个任务适合对列表中的每一项分别处理。示例from langgraph.types import Send def dispatch_nodes(state: LoopState) - list[Send]: return [Send(process_item, {item: item}) for item in state[input_text]]dispatch_nodes是一个节点它返回一组Send对象。每个Send包含目标节点名称和要传给该节点的状态。这种写法适合数据并行场景比如把一个文档列表拆成多个子任务每个子任务独立处理。并行分支的常见坑是多个子节点同时修改同一个状态字段时最后一个写入方的结果可能覆盖前面的结果。如果需要保留所有结果状态字段应该设计成列表并使用 reducer 合并。5. 子图Subgraph把复杂工作流拆分到可管理单元当一张图包含十几个节点时全部平铺在一个StateGraph里会很难维护。子图允许你把一部分节点封装成一个独立的图再作为另一个图的节点嵌入。5.1 子图解决什么问题子图的主要作用是拆解复杂度。你可以在一个文件里定义“订单处理子图”在另一个文件里定义“售后子图”然后在主图里将它们作为节点挂载。这样每个子图可以独立测试主图只负责编排。另外当你需要复用一套流程时子图也比复制粘贴更合适。例如不同的入口节点都可能走“意图识别 - 槽位填充 - 执行动作”的流程就可以把这一段封装成子图。5.2 定义一个最简单的子图先定义一个内部子图from langgraph.graph import StateGraph, START, END class SubState(TypedDict): task: str sub_result: str def sub_node(state: SubState) - dict: return {sub_result: f子图处理{state[task]}} sub_graph StateGraph(SubState) sub_graph.add_node(sub_node, sub_node) sub_graph.add_edge(START, sub_node) sub_graph.add_edge(sub_node, END) sub_app sub_graph.compile()这里sub_app是一个CompiledStateGraph。5.3 在主图中嵌入子图主图的状态需要包含子图需要的数据且节点可以是一个CompiledStateGraph。class MainState(TypedDict): task: str sub_result: str final_result: str def after_sub(state: MainState) - dict: return {final_result: f最终结果{state[sub_result]}} main_graph StateGraph(MainState) main_graph.add_node(sub, sub_app) main_graph.add_node(after, after_sub) main_graph.add_edge(START, sub) main_graph.add_edge(sub, after) main_graph.add_edge(after, END) main_app main_graph.compile() result main_app.invoke({task: 分析用户情绪}) print(result)运行时主图进入sub节点相当于把当前状态交给子图执行子图执行完返回的状态会与主图状态合并。这里要特别注意子图的输入状态只能包含子图状态里定义的字段。如果子图里没有定义final_result即使主图有该字段子图也无法直接访问或修改。也就是说主图和子图之间是“按状态字段”解耦的而不是自动共享全部字段。5.4 子图与普通节点的区别从主图视角看子图就是一个节点。节点函数可以自定义输入输出而子图有自己完整的状态定义和执行流程。普通节点适合做轻量计算子图适合承载一套完整业务流程。使用子图时要注意状态匹配问题。如果主图状态字段名和子图状态字段名不一致要在传参时做转换。例如主图字段叫user_text子图字段叫task可以写一个转换节点def prepare_sub_state(state: MainState) - dict: return {task: state[user_text]}这样能更清晰地管理数据流而不会让字段名不一致导致运行时缺字段。6. 在节点函数中修改 State 的几种姿势很多初学者在写 LangGraph 节点函数时最大的困惑是“到底怎么修改状态”。 LangGraph 不允许你在节点函数内部直接对整个状态对象赋值它要求节点返回一个字典由运行时来合并。但这里的合并规则并不是每种字段都一样。6.1 最直接的方式返回字段字典当节点需要更新一个普通字段时直接返回包含该字段的字典即可def set_answer(state: MyState) - dict: return {answer: 新的答案}如果字段名不存在LangGraph 默认会新增一个字段。如果你不希望节点意外创建新字段可以在状态类型里明确声明所有字段。6.2 用 Annotated 声明 Reducer实现列表追加如果状态里有消息列表直接返回列表会覆盖而不是追加。比如class MessageState(TypedDict): messages: list某个节点返回{messages: [你好]}下一轮另一个节点也返回{messages: [有什么需要帮助吗]}最终状态里只会保留最后一个列表。要解决这个问题需要使用消息追加的 reducer。LangGraph 的常见写法是使用Annotated和operator.addfrom typing import TypedDict, Annotated import operator class MessageState(TypedDict): messages: Annotated[list, operator.add]此时每个节点返回的messages列表都会被合并到原列表后面。例如[hello] [world]最终是[hello, world]。如果你想自己控制合并逻辑也可以写一个自定义 reducer 函数。reducer 接收两个值旧值和新值返回合并后的结果。例如def merge_messages(old: list, new: list) - list: return old new然后用Annotated[list, merge_messages]声明即可。6.3 节点函数内部更新局部变量但不直接改 State节点函数里可以读取整个 state但不建议直接执行state[xxx] yyy这类语句。一个原因是 LangGraph 的执行环境可能对状态做并发控制直接修改外部字典对象容易产生不可预期的副作用另一个原因是调试时难以追踪状态变化。正确写法是def node_func(state: MyState) - dict: local_var state[user_input] 已处理 return {processed_input: local_var}如果你确实需要修改很多字段可以构建一个新字典统一返回def node_func(state: MyState) - dict: return { field_a: 1, field_b: 2, field_c: state[old] x, }6.4 常见错误与正确写法对比错误现象错误写法正确写法节点返回后状态没有变化函数内直接修改state对象不返回内容返回字典由运行时合并messages一直覆盖而不是追加返回{messages: [新消息]}且字段没有 reducer字段用Annotated[list, operator.add]子图返回后主图缺少字段子图状态没有声明该字段在子图状态类型中补充该字段循环里状态不更新每次返回相同的值在返回字典中包含自增次数或变化结果在调试状态更新问题时最有效的方法是打印每个节点的输入和输出状态。LangGraph 也支持在invoke时传入debug参数或者自定义回调来观察每个节点的状态变更。7. 一个完整实战构建带条件路由和子图的多轮问答工作流把前面的知识点串起来写一个更完整的示例。这个工作流模拟“智能客服”用户输入问题后先判断意图如果是订单相关问题进入订单子图查询如果要转人工直接转人工如果只是闲聊直接回复。7.1 需求拆解这个实战包含四个步骤输入用户问题。判断意图。根据意图分流订单子图、闲聊回复、人工转接。收集结果生成最终回复。用例可以通过但为了学习 LangGraph重点放在状态设计、条件路由和子图使用上。7.2 状态设计from typing import TypedDict, Annotated, Literal import operator class OrderState(TypedDict): order_id: str order_info: str class CustomerServiceState(TypedDict): user_input: str intent: Literal[order, chat, human] messages: Annotated[list, operator.add] order_result: str final_response: str这里messages使用追加 reducer便于记录整个对话过程。7.3 意图识别节点实际项目中可以用 LLM 分类这里用关键词模拟def detect_intent(state: CustomerServiceState) - dict: text state[user_input] if 订单 in text or 物流 in text: return {intent: order, messages: [识别为订单意图]} if 人工 in text or 客服 in text: return {intent: human, messages: [识别为人工意图]} return {intent: chat, messages: [识别为闲聊意图]}7.4 条件路由def route_after_detect(state: CustomerServiceState) - Literal[order, chat, human]: return state[intent]然后把三个分支节点都定义好def chat_reply(state: CustomerServiceState) - dict: return {final_response: 我是智能助手可以陪你聊聊天气、编程或日常问题。, messages: [执行闲聊回复]} def human_transfer(state: CustomerServiceState) - dict: return {final_response: 正在为你转接人工客服请稍候。, messages: [执行人工转接]}订单子图单独定义def query_order(state: OrderState) - dict: return {order_info: f订单 {state[order_id]} 已发货} order_graph StateGraph(OrderState) order_graph.add_node(query, query_order) order_graph.add_edge(START, query) order_graph.add_edge(query, END) order_app order_graph.compile()主图里接入子图前需要写一个适配节点把主图状态转换成子图需要的字段def order_subgraph_entry(state: CustomerServiceState) - dict: # 简化从输入中获取订单号 order_id SO123456 sub_result order_app.invoke({order_id: order_id}) return {order_result: sub_result[order_info], messages: [订单子图执行完成]}完整主图构造from langgraph.graph import StateGraph, START, END main_graph StateGraph(CustomerServiceState) main_graph.add_node(detect, detect_intent) main_graph.add_node(chat, chat_reply) main_graph.add_node(human, human_transfer) main_graph.add_node(order, order_subgraph_entry) main_graph.add_edge(START, detect) main_graph.add_conditional_edges( detect, route_after_detect, { order: order, chat: chat, human: human, } ) # 分支统一结束 main_graph.add_edge(order, END) main_graph.add_edge(chat, END) main_graph.add_edge(human, END) main_app main_graph.compile()这里子图不是完全黑盒主图通过order_subgraph_entry调用了order_app.invoke()相当于在一个普通节点里手动调用子图。这种方式的优点是控制精确缺点是该节点会阻塞等待子图完成。如果希望子图作为独立节点直接嵌入LangGraph 也支持把CompiledStateGraph作为节点加入主图但需要保证主图状态字段和子图状态字段兼容。7.5 运行验证运行三个测试输入test_cases [ 我想查一下订单号 SO123456 的物流, 你叫什么名字, 转人工客服 ] for case in test_cases: result main_app.invoke({user_input: case}) print(输入:, case) print(意图:, result[intent]) print(最终回复:, result[final_response]) print(消息记录:, result[messages]) print(---)预期结果大致是输入包含“订单”时走order分支返回订单子图结果。输入“你叫什么名字”时走chat分支。输入“转人工”时走human分支。如果某个分支没有按预期执行优先检查条件路由函数返回值是否与add_conditional_edges的映射表键一致。7.6 学习环境与生产环境的差异上面的代码已经把 LangGraph 的核心功能串起来了。但生产使用还需要补充以下能力能力学习环境生产环境状态存储内存进程重启即丢失使用 checkpointer 持久化到数据库模型调用本地或测试 key统一 API 网关、限流、重试日志print 临时查看结构化日志、链路追踪分支异常直接抛错兜底分支、人工介入、告警子图复用单文件示例独立模块化、单元测试检查点checkpointer是 LangGraph 的一个重要能力。它可以把图的状态保存下来支持断点恢复、时间旅行调试和人工审批流程。生产环境建议一开始就接入数据库存储比如使用SqliteSaver或生产级数据库实现。8. 常见问题排查与调试清单实际使用 LangGraph 时问题往往不在语法而在执行语义。这里整理几个高频问题和排查路径。8.1 节点执行了但状态里没有看到字段更新现象节点函数明明返回了字典但invoke结果里没有对应字段。排查步骤检查节点函数是否真的被调用了。可以在函数入口打印日志。检查返回的字典是否拼写错误。如果状态类型是 TypedDict字段名不一致不会报错。检查是否有其他节点在后续覆盖了同一个字段。检查是否使用了Annotated字段且 reducer 合并逻辑不符合预期。检查边是否跳过了该节点。如果节点连接到END后续节点自然拿不到它的返回值。8.2 条件边不执行或总是走默认分支现象条件路由后目标节点不是预期节点。排查步骤在路由函数里打印返回值确认返回的是字符串。检查返回值是不是包含空格、引号或类型不一致。比如返回Literal[order]但实际值是order 。检查add_conditional_edges的映射表里是否有该键。如果路由函数返回None也会导致映射失败需要给所有路径提供兜底返回值。确认使用的是add_conditional_edges而不是add_edge。8.3 图陷入死循环现象程序一直执行不返回。排查步骤查看循环节点的日志确认状态是否在每次循环中发生变化。检查循环出口条件。比如重试次数是否真的在递增。检查路由函数是否存在永远不会返回出口键的分支。为循环节点添加最大重试次数保护这是最稳妥的兜底方案。例如可以在循环节点内增加if state[retry_count] 5: return {finished: True}这样即使路由逻辑写错也能在有限次循环后强制退出。8.4 子图返回结果与父图状态不兼容现象子图执行完主图后续节点读不到子图返回的内容。排查步骤检查子图状态类型是否包含该字段。检查子图是否通过END正常返回。检查主图是否在子图后继续使用主图状态对象而不是子图返回的局部状态字典。如果通过节点函数调用子图需要把子图返回结果手动映射回主图字段。8.5 并行分支共享状态污染现象多个并行节点都修改messages结果顺序不确定或字段被覆盖。排查步骤为需要追加的字段添加 reducer比如Annotated[list, operator.add]。如果并行节点各自返回相同字段确认期望是覆盖还是合并。必要时为每个并行分支定义独立字段避免竞态。8.6 排查顺序清单当一张图运行结果和预期不一致时按下面顺序检查输入状态是否正确。节点函数是否都被调用。每个节点返回的字典是否正确。普通边和条件边的目标节点是否符合预期。状态字段是否被 reducer 正确合并。子图是否独立运行正常。最终返回状态里是否有预期字段。9. 最佳实践与进阶方向LangGraph 从示例到可用中间差的是工程化思维。以下实践建议同样适用于 LangChain 生态的其他工作流项目。9.1 状态设计先于节点设计很多初学者先写节点函数再思考状态类型导致状态字段混乱。建议先画一张表列出整个流程中哪些数据需要在节点之间共享哪些只是局部变量哪些需要追加哪些需要覆盖。可以参考这个状态字段设计清单字段名类型更新策略用途user_inputstr覆盖保存用户原始输入messageslist追加保存对话历史intentstr覆盖保存分类结果retry_countint覆盖控制循环次数tool_resultstr覆盖工具调用结果状态字段越少图越容易理解。能用局部变量解决的问题不要都放到全局状态里。9.2 节点函数尽量保持“纯函数”一个节点只做一件事输入和输出尽量可预测。不要在节点里写复杂的业务逻辑更不要让节点之间通过全局变量交换数据。推荐做法每个节点函数只读取自己需要的字段。每个节点函数返回明确的数据结构。对可能失败的调用增加 try/except再返回错误状态。这样既方便单元测试也方便后面接入日志和监控。9.3 设置最大执行步数和循环保护LangGraph 虽然提供了灵活的执行机制但生产环境必须防止失控循环。可以在编译后的图上设置递归限制或者在状态里增加步数计数器。具体限制值和业务相关但至少要有兜底。9.4 先本地调试再接入模型不要一开始就把所有节点都接上大模型。先用字符串拼接、规则匹配把流程跑通确认状态传递、条件路由、子图都正确再替换为真实模型调用。这样可以节省调试成本也更容易定位问题是出在流程编排还是模型输出。9.5 生产部署要重点关注持久化和可观测性LangGraph 入门到实战最终要落地的不仅是功能还包括可靠性。生产环境需要关注状态持久化使用 checkpointer让任务能断点续跑。日志每个节点记录输入、输出、耗时。监控图执行时间、失败率、重试次数。人工审批在关键节点设置中断等待人工确认后再继续执行。版本管理图的定义和依赖版本需要一起发布避免状态结构变更后旧任务无法恢复。9.6 学习路径建议如果你刚接触 LangGraph建议按这个顺序进阶完成本文的最简图和条件路由示例确保理解状态、节点、边的概念。自己设计一个三节点的流程加入循环和子图。把messages字段改成 reducer 追加理解状态合并规则。尝试接入一个真实 LLM让条件边根据模型输出路由。阅读 LangGraph 官方文档中关于 checkpointer 和持久化的部分实现一个带记忆的会话应用。在项目里抽象出可复用的子图并编写单元测试。LangGraph 的学习曲线不算陡难点在于从“链式思维”切换到“图思维”。只要把状态、边、条件路由这三件事理解透后面无论是构建 Agent、知识库问答还是复杂任务编排都能很快上手。