如果你准备用 LangGraph 搭建带分支、循环、并行和子图的 LLM 应用却被网上零散的资料折磨得怀疑人生这篇文章应该能帮你把路铺平。我尽量用“能直接运行”的代码和“能直接改”的思路把 LangGraph 从核心概念讲到综合实战。无论你是第一次接触 LangGraph还是已经在 LangChain 里写过一些链式调用都可以按章节顺序读下去。1. LangGraph 到底是什么从 Chain 到 State Graph1.1 一个容易被忽视的问题流程控制在基于大模型的应用里最常见的开发方式是“按顺序调用几次大模型”。比如先分析用户意图再调用工具最后生成回答。这种线性流程在 Demo 阶段非常顺手但一旦需求变复杂问题就来了流程要分叉不同意图走不同分支流程要循环信息不足时需要追问流程要并行多个工具可以同时查询。用传统 if-else 硬写代码会迅速膨胀而且很难看清整体结构。LangGraph 解决的就是这类问题。它把一次应用执行过程建模成一张“状态图”节点是处理步骤边是步骤之间的跳转状态是全局共享的数据。这样分叉、循环、并行都能用图结构表达代码的可读性、可维护性也会明显提升。简单说LangGraph 是把“流程控制”从业务代码中抽离出来由图引擎统一管理。1.2 LangGraph 的核心模型LangGraph 的核心思想并不复杂你只需要记住四个词State状态通常是一个 TypedDict 或 Pydantic 模型保存整个流程共享的数据。Node节点一个普通函数接收当前 State返回一个字典表示需要更新的字段。Edge边连接两个节点决定执行顺序。StateGraph状态图把节点和边组装起来最后编译成可执行对象。流程看起来很直观先定义一个 State再实现若干节点函数然后用 StateGraph 把这些节点连起来。每次invoke时LangGraph 会从 START 节点出发沿着边执行所有节点直到 END 节点。这种方式比硬编码函数调用更好维护因为节点之间是解耦的你可以随时插入新的步骤或调整流程顺序。1.3 LangChain 与 LangGraph 的区别很多初学者会混淆 LangChain 和 LangGraph。简单来说两者解决的问题层次不同LangChain 更偏向“与大模型交互”提供模型封装、Prompt 模板、文档加载、向量库集成等能力而 LangGraph 更偏向“复杂流程编排”解决的是多个步骤之间如何组织、如何跳转、如何共享状态的问题。两者并不是替代关系。LangGraph 完全可以搭配 LangChain 的模型封装组件使用例如用langchain-openai调用大模型再用 LangGraph 控制调用次数和判断逻辑。可以这样理解LangChain 解决“怎么让大模型更好地工作”LangGraph 解决“怎么让多个大模型调用和工具调用按照业务规则跑起来”。在 Agent 类项目中LangGraph 的优势非常明显因为它内置了循环和条件跳转能力正好对应 Agent 的“思考-行动-观察”反复过程。2. LangGraph 环境准备与项目初始化2.1 环境要求LangGraph 是一个纯 Python 库跨平台支持比较好。开发环境建议满足以下条件Python 3.9 及以上推荐 3.10 或 3.11。操作系统不限Windows、macOS、Linux 都可以。包管理工具使用 pip、poetry 或 uv 均可。如果只需要跑图编排逻辑不需要额外安装大模型依赖。我这里不会写死某个具体版本号因为 LangGraph 的迭代速度比较快API 在不同小版本之间可能略有差异。你只需要保证能够正常import langgraph并按照你实际安装版本的官方文档来调整细节即可。2.2 安装依赖创建虚拟环境并安装 LangGraphpython -m venv .venv source .venv/bin/activate # Windows 下使用 .venv\Scripts\activate pip install langgraph如果之后要接入 OpenAI 或其他大模型可以再安装对应的集成包例如pip install langchain-openai但本文的核心示例不依赖具体模型先通过纯代码把图结构跑通避免混淆“框架用法”和“模型调用”两个层面的问题。2.3 建议的项目结构学习阶段的项目结构不用太复杂建议按“一个能力一个文件”来组织langgraph-demo/ ├── requirements.txt ├── demo_basic.py # 基础入门示例 ├── demo_route.py # 条件路由示例 ├── demo_parallel.py # 并行分支示例 ├── demo_subgraph.py # 子图示例 └── ticket_agent.py # 综合实战工单处理助手这种方式的好处是定位问题快。如果你在综合实战里跑出了诡异报错可以回退到单个示例文件逐层排查究竟是条件路由写错了还是状态字段类型不匹配。3. 5 分钟跑通第一个 LangGraph 程序3.1 定义 StateState 是 LangGraph 中所有节点共享的数据载体。最基础的做法是使用 TypedDict。例如from typing import TypedDict class DemoState(TypedDict): user_input: str messages: list[str]这里定义了两个字段user_input表示用户输入messages表示流程执行过程中累计的消息。需要特别注意的是LangGraph 并不会自动合并字段节点返回的字段默认会覆盖 State 中已有的值。后面如果要累积数据需要使用带 reducer 的字段比如Annotated[list, operator.add]。3.2 编写节点函数节点函数是 LangGraph 的基本执行单元。它接收当前 State返回一个字典字典中的字段会被更新到 State 中。下面的代码实现了两个节点def node_a(state: DemoState) - dict: print(f[node_a] 收到输入{state[user_input]}) return {messages: [node_a 处理完成]} def node_b(state: DemoState) - dict: print(f[node_b] 已处理 messages{state[messages]}) return {messages: [node_b 处理完成]}从代码可以看出节点并不需要把整个 State 都返回只需要返回“想要修改的字段”。node_a返回messagesnode_b也是返回messages。由于默认是覆盖更新所以最终messages只会保留最后一次赋的值。这里先不引入 reducer目的是让大家理解基本覆盖逻辑。3.3 连接边并编译有了节点就可以用 StateGraph 把它们连起来from langgraph.graph import StateGraph, START, END graph StateGraph(DemoState) graph.add_node(node_a, node_a) graph.add_node(node_b, node_b) graph.add_edge(START, node_a) graph.add_edge(node_a, node_b) graph.add_edge(node_b, END) app graph.compile()START和END是 LangGraph 内置的虚拟节点表示图的入口和出口。graph.compile()之后得到一个可执行对象app后续调用app.invoke(...)就是执行整条流程。3.4 运行结果分析使用invoke传入初始 Stateresult app.invoke({user_input: hello, messages: []}) print(result)预期输出类似下面这样[node_a] 收到输入hello [node_b] 已处理 messages[node_a 处理完成] {user_input: hello, messages: [node_b 处理完成]}从输出可以看到先执行node_a再执行node_b返回结果是最终 State。如果这里把messages默认覆盖逻辑改成累加逻辑node_b读取到的messages就会包含node_a和node_b的结果。这一步是理解 LangGraph 状态机制的关键。4. 条件路由与分支控制conditional_edges 深度解析4.1 为什么需要条件路由固定链路只能解决“顺序执行”的问题但现实业务更多是“根据不同条件走不同分支”。比如客服系统中用户输入“查订单”和“投诉建议”应该进入不同处理逻辑。LangGraph 通过add_conditional_edges支持条件路由让一个节点根据某个条件决定下一步跳转到哪个节点。条件路由的本质是编写一个路由函数接收当前 State返回一个字符串标志再提供一个映射字典把返回值对应到目标节点。4.2 基础条件路由写法下面是一个完整示例根据finish字段决定流程是继续执行work节点还是直接结束from typing import TypedDict, Annotated import operator from langgraph.graph import StateGraph, START, END class RouteState(TypedDict): steps: Annotated[list[str], operator.add] finish: bool def start_node(state: RouteState) - dict: return {steps: [start]} def route_by_state(state: RouteState) - str: if state.get(finish): return end return continue def work_node(state: RouteState) - dict: return {steps: [work], finish: True} graph StateGraph(RouteState) graph.add_node(start, start_node) graph.add_node(work, work_node) graph.add_edge(START, start) graph.add_conditional_edges( start, route_by_state, { continue: work, end: END, }, ) graph.add_edge(work, END) app graph.compile() result app.invoke({steps: [], finish: False}) print(result)初始状态下finish为 False所以路由函数返回continue流程进入work节点。work节点把finish设为 True下一次如果再次路由就会走到end。这里的Annotated[list[str], operator.add]表示steps字段使用operator.add作为 reducer节点返回的列表会追加到旧列表后面而不是直接覆盖。条件路由中的路由函数必须保持简单明确尽量只读 State 并返回一个字符串。不要在路由函数里做耗时 IO 操作因为路由函数本身不是业务节点它只负责“选择下一站”。4.3 循环检测与递归限制条件路由还有一个常见用途构造循环。比如某个节点需要对一个结果反复处理直到满足条件。下面这个例子展示了一个典型的循环结构from typing import TypedDict from langgraph.graph import StateGraph, START, END class LoopState(TypedDict): count: int def increment(state: LoopState) - dict: return {count: state[count] 1} def should_continue(state: LoopState) - str: if state[count] 3: return continue return stop graph StateGraph(LoopState) graph.add_node(increment, increment) graph.add_edge(START, increment) graph.add_conditional_edges( increment, should_continue, { continue: increment, stop: END, }, ) app graph.compile() result app.invoke({count: 0}, config{recursion_limit: 10}) print(result)这里increment节点执行后回到自身形成一个循环。每次进入increment后count加 1当count达到 3 时should_continue返回stop流程结束。LangGraph 默认会对递归深度做限制。如果你的图真的陷入了死循环LangGraph 会抛出递归超限异常。示例中通过config{recursion_limit: 10}显式允许最多执行 10 步避免误伤正常循环。实际开发中这个值要根据业务合理设计并确保条件路由最终可以收敛否则会因为死循环消耗大量资源。4.4 并行分支让多个节点同时工作LangGraph 支持从同一个节点出发连接多个下游节点从而让这些节点并行执行。在状态更新上如果多个并行节点都修改同一个字段就需要使用 reducer 合并结果。下面是一个例子from typing import TypedDict, Annotated import operator from langgraph.graph import StateGraph, START, END class ParallelState(TypedDict): results: Annotated[list[str], operator.add] def task_a(state: ParallelState) - dict: return {results: [A 的结果]} def task_b(state: ParallelState) - dict: return {results: [B 的结果]} def merge(state: ParallelState) - dict: return {results: [f合并完成{state[results]}]} graph StateGraph(ParallelState) graph.add_node(task_a, task_a) graph.add_node(task_b, task_b) graph.add_node(merge, merge) graph.add_edge(START, task_a) graph.add_edge(START, task_b) graph.add_edge(task_a, merge) graph.add_edge(task_b, merge) graph.add_edge(merge, END) app graph.compile() result app.invoke({results: []}) print(result)这里task_a和task_b都从 START 出发LangGraph 会并行执行它们然后都进入merge节点。因为results字段使用了Annotated[list[str], operator.add]并行节点返回的列表会被追加而不是互相覆盖。最终merge节点拿到的results包含两个任务的结果。这里需要注意的是并行节点的执行顺序并不固定。如果业务上必须区分顺序就不要依赖并行节点而是明确串成一条路径。并行执行主要是为了提升效率比如同时查询订单系统和物流系统。5. 子图Subgraph与 State 修改5.1 什么时候需要拆子图当流程越来越复杂时把所有节点都塞进同一张图会变得难以维护。LangGraph 允许把一张已经编译好的图作为子图嵌套到另一张父图中。这样既能把复杂流程拆成多个模块又能复用公共处理逻辑。比较适合拆子图的场景包括一段反复出现的固定处理流程、一个独立的业务模块、一组可以单独测试的节点集合。子图可以拥有独立的 State也可以和父图共享部分字段数据。5.2 嵌套子图示例先定义一个子图用于统计文本中的单词数量from typing import TypedDict from langgraph.graph import StateGraph, START, END class SubState(TypedDict): text: str count: int def count_words(state: SubState) - dict: return {count: len(state[text].split())} sub_graph StateGraph(SubState) sub_graph.add_node(count_words, count_words) sub_graph.add_edge(START, count_words) sub_graph.add_edge(count_words, END) sub_app sub_graph.compile()接着在父图节点中调用子图class ParentState(TypedDict): text: str words: int upper_text: str def upper_node(state: ParentState) - dict: return {upper_text: state[text].upper()} def run_subgraph(state: ParentState) - dict: sub_result sub_app.invoke({text: state[text]}) return {words: sub_result[count]} parent_graph StateGraph(ParentState) parent_graph.add_node(upper, upper_node) parent_graph.add_node(subgraph, run_subgraph) parent_graph.add_edge(START, upper) parent_graph.add_edge(START, subgraph) parent_graph.add_edge(upper, END) parent_graph.add_edge(subgraph, END) parent_app parent_graph.compile() result parent_app.invoke({text: hello world langgraph, words: 0}) print(result)在这个例子中父图并行执行upper_node和run_subgraphrun_subgraph内部再调用sub_app去执行子图。子图返回的结果通过普通字典传给父图 State。这种“父图节点里调子图”的方式最简单也最容易理解。5.3 节点函数修改 State 的几种方式掌握节点如何修改 State是 LangGraph 使用中最重要的一环。主要方式有三种。第一种直接返回需要更新的字段。节点函数返回字典字典中的 key 会被写入 State。比如return {count: 1}就会把 State 中count字段覆盖成 1。第二种使用 reducer 合并数据。在定义 State 时用Annotated[字段类型, reducer函数]声明累计逻辑。最常见的写法是Annotated[list, operator.add]。这样节点返回一个列表时LangGraph 会自动把新列表追加到旧列表后面。第三种返回空字典或 None。如果节点只是消费 State 的内容不需要修改任何状态可以直接返回空字典。不返回某个字段该字段就不会被修改。需要特别注意的是节点返回的 key 必须存在于 State 定义中否则 LangGraph 会报错。如果你确实需要一些临时变量尽可能在节点函数内部处理不要随意往 State 里塞无关字段。6. 综合实战带路由、循环、并行的工单助手6.1 需求与流程设计现在把前面几节的能力组合起来做一个“客服工单助手”综合案例。需求如下用户输入一段文本系统判断这次输入属于订单咨询还是投诉建议。如果属于订单咨询则并行查询订单信息和物流信息最后汇总结果如果属于投诉建议则直接生成投诉处理话术。整体流程是START - 意图分类 - 条件路由 | |--- order - 并行查询订单 物流 - 合并结果 | |--- complaint - 投诉处理话术 | - 最终输出 - END这个案例不依赖真实大模型所有节点都使用规则或模拟数据方便你把注意力放在图结构上。6.2 完整代码创建ticket_agent.py内容如下from typing import Annotated, TypedDict import operator from langgraph.graph import StateGraph, START, END class TicketState(TypedDict): user_input: str intent: str order_id: str results: Annotated[list[str], operator.add] final_answer: str def classify(state: TicketState) - dict: text state[user_input] intent order if (订单 in text or 物流 in text) else complaint return {intent: intent} def route_intent(state: TicketState) - str: return state[intent] def order_entry(state: TicketState) - dict: return {} def query_order(state: TicketState) - dict: order_id state.get(order_id, A10086) return {results: [f订单 {order_id} 状态已支付待发货]} def query_logistics(state: TicketState) - dict: return {results: [物流信息仓库已打包等待揽收]} def merge_order(state: TicketState) - dict: merged .join(state[results]) return {final_answer: f订单查询结果{merged}} def complaint_reply(state: TicketState) - dict: return {final_answer: 已收到您的投诉我们会在 1 个工作日内人工处理。} def output_answer(state: TicketState) - dict: return {final_answer: f客服助手{state[final_answer]}} graph StateGraph(TicketState) graph.add_node(classify, classify) graph.add_node(order_entry, order_entry) graph.add_node(query_order, query_order) graph.add_node(query_logistics, query_logistics) graph.add_node(merge_order, merge_order) graph.add_node(complaint_reply, complaint_reply) graph.add_node(output_answer, output_answer) graph.add_edge(START, classify) graph.add_conditional_edges( classify, route_intent, { order: order_entry, complaint: complaint_reply, }, ) graph.add_edge(order_entry, query_order) graph.add_edge(order_entry, query_logistics) graph.add_edge(query_order, merge_order) graph.add_edge(query_logistics, merge_order) graph.add_edge(merge_order, output_answer) graph.add_edge(complaint_reply, output_answer) graph.add_edge(output_answer, END) app graph.compile()这段代码里order_entry是一个空节点作用是给并行分支提供一个统一入口。query_order和query_logistics并行执行因为它们都从order_entry出发然后都汇聚到merge_order。results字段使用了operator.add作为 reducer保证并行节点返回的列表不会互相覆盖。6.3 运行与验证编写运行代码order_result app.invoke({ user_input: 我想查一下订单 A10086 的物流, order_id: A10086, results: [], }) print(order_result[final_answer])预期输出客服助手订单查询结果订单 A10086 状态已支付待发货物流信息仓库已打包等待揽收再测试投诉分支complaint_result app.invoke({ user_input: 我对这次服务非常不满意, order_id: , results: [], }) print(complaint_result[final_answer])预期输出客服助手已收到您的投诉我们会在 1 个工作日内人工处理。到这里综合案例就跑通了。你可以在这个基础上做两件事把classify节点替换成真实的大模型分类调用把query_order和query_logistics替换成真实系统的 HTTP 请求或数据库查询。图结构本身不需要大改。7. LangGraph 高频问题与排查清单LangGraph 的报错信息整体比较友好但很多新手会在几个固定位置踩坑。这里整理一份高频问题表你可以直接对照排查。问题现象常见原因解决思路ModuleNotFoundError: No module named langgraph未安装或没有激活正确的虚拟环境执行pip install langgraph检查当前 Python 环境KeyError或字段不存在的报错节点返回了 State 中未定义的 key检查节点返回字典中的 key 是否在 State 类中定义条件路由报错找不到目标节点路由函数返回值不在映射字典中核对路由函数的返回值与映射字典的 key 是否完全一致并行节点更新同一个字段时互相覆盖该字段没有配置 reducer改成Annotated[list, operator.add]或用其他 reducer 合并循环执行次数太多抛出递归限制异常条件边形成了无法收敛的循环检查路由条件或者在invoke时通过recursion_limit调整上限调用invoke返回的结果与预期不一致对 State 的覆盖逻辑理解有误记住默认是覆盖更新需要累加时使用 reducer子图状态和父图状态混淆父图节点与子图使用不同状态定义明确子图输入输出字段在节点函数中做显式映射排查问题时建议按照“State 定义 - 节点返回值 - 边的连接”这个顺序检查。大部分问题都出在这三个环节。尤其是条件路由路由函数的返回值如果多打了一个空格LangGraph 就找不到对应节点。8. 最佳实践与工程建议8.1 State 设计要“小而明确”State 是 LangGraph 应用的核心设计不好后面会很痛苦。不要把每个临时变量都塞进 StateState 中应该只保留“跨节点共享、需要被执行流程记住”的数据。字段命名尽量清晰避免使用data、info、tmp这类模糊名称。如果字段需要累加、追加或合并提前用Annotated加上 reducer而不是在节点函数里手工拼接后覆盖。8.2 命名与模块划分节点名称建议使用“动词 名词”的结构例如query_order、merge_result、classify_intent。这样图结构打印出来之后一眼就能看出流程在做什么。项目文件可以按业务模块划分每个模块负责一组相关节点。子图是拆解复杂流程的好工具但不要为了用子
LangGraph从入门到实战：构建分支、循环与并行的LLM应用
如果你准备用 LangGraph 搭建带分支、循环、并行和子图的 LLM 应用却被网上零散的资料折磨得怀疑人生这篇文章应该能帮你把路铺平。我尽量用“能直接运行”的代码和“能直接改”的思路把 LangGraph 从核心概念讲到综合实战。无论你是第一次接触 LangGraph还是已经在 LangChain 里写过一些链式调用都可以按章节顺序读下去。1. LangGraph 到底是什么从 Chain 到 State Graph1.1 一个容易被忽视的问题流程控制在基于大模型的应用里最常见的开发方式是“按顺序调用几次大模型”。比如先分析用户意图再调用工具最后生成回答。这种线性流程在 Demo 阶段非常顺手但一旦需求变复杂问题就来了流程要分叉不同意图走不同分支流程要循环信息不足时需要追问流程要并行多个工具可以同时查询。用传统 if-else 硬写代码会迅速膨胀而且很难看清整体结构。LangGraph 解决的就是这类问题。它把一次应用执行过程建模成一张“状态图”节点是处理步骤边是步骤之间的跳转状态是全局共享的数据。这样分叉、循环、并行都能用图结构表达代码的可读性、可维护性也会明显提升。简单说LangGraph 是把“流程控制”从业务代码中抽离出来由图引擎统一管理。1.2 LangGraph 的核心模型LangGraph 的核心思想并不复杂你只需要记住四个词State状态通常是一个 TypedDict 或 Pydantic 模型保存整个流程共享的数据。Node节点一个普通函数接收当前 State返回一个字典表示需要更新的字段。Edge边连接两个节点决定执行顺序。StateGraph状态图把节点和边组装起来最后编译成可执行对象。流程看起来很直观先定义一个 State再实现若干节点函数然后用 StateGraph 把这些节点连起来。每次invoke时LangGraph 会从 START 节点出发沿着边执行所有节点直到 END 节点。这种方式比硬编码函数调用更好维护因为节点之间是解耦的你可以随时插入新的步骤或调整流程顺序。1.3 LangChain 与 LangGraph 的区别很多初学者会混淆 LangChain 和 LangGraph。简单来说两者解决的问题层次不同LangChain 更偏向“与大模型交互”提供模型封装、Prompt 模板、文档加载、向量库集成等能力而 LangGraph 更偏向“复杂流程编排”解决的是多个步骤之间如何组织、如何跳转、如何共享状态的问题。两者并不是替代关系。LangGraph 完全可以搭配 LangChain 的模型封装组件使用例如用langchain-openai调用大模型再用 LangGraph 控制调用次数和判断逻辑。可以这样理解LangChain 解决“怎么让大模型更好地工作”LangGraph 解决“怎么让多个大模型调用和工具调用按照业务规则跑起来”。在 Agent 类项目中LangGraph 的优势非常明显因为它内置了循环和条件跳转能力正好对应 Agent 的“思考-行动-观察”反复过程。2. LangGraph 环境准备与项目初始化2.1 环境要求LangGraph 是一个纯 Python 库跨平台支持比较好。开发环境建议满足以下条件Python 3.9 及以上推荐 3.10 或 3.11。操作系统不限Windows、macOS、Linux 都可以。包管理工具使用 pip、poetry 或 uv 均可。如果只需要跑图编排逻辑不需要额外安装大模型依赖。我这里不会写死某个具体版本号因为 LangGraph 的迭代速度比较快API 在不同小版本之间可能略有差异。你只需要保证能够正常import langgraph并按照你实际安装版本的官方文档来调整细节即可。2.2 安装依赖创建虚拟环境并安装 LangGraphpython -m venv .venv source .venv/bin/activate # Windows 下使用 .venv\Scripts\activate pip install langgraph如果之后要接入 OpenAI 或其他大模型可以再安装对应的集成包例如pip install langchain-openai但本文的核心示例不依赖具体模型先通过纯代码把图结构跑通避免混淆“框架用法”和“模型调用”两个层面的问题。2.3 建议的项目结构学习阶段的项目结构不用太复杂建议按“一个能力一个文件”来组织langgraph-demo/ ├── requirements.txt ├── demo_basic.py # 基础入门示例 ├── demo_route.py # 条件路由示例 ├── demo_parallel.py # 并行分支示例 ├── demo_subgraph.py # 子图示例 └── ticket_agent.py # 综合实战工单处理助手这种方式的好处是定位问题快。如果你在综合实战里跑出了诡异报错可以回退到单个示例文件逐层排查究竟是条件路由写错了还是状态字段类型不匹配。3. 5 分钟跑通第一个 LangGraph 程序3.1 定义 StateState 是 LangGraph 中所有节点共享的数据载体。最基础的做法是使用 TypedDict。例如from typing import TypedDict class DemoState(TypedDict): user_input: str messages: list[str]这里定义了两个字段user_input表示用户输入messages表示流程执行过程中累计的消息。需要特别注意的是LangGraph 并不会自动合并字段节点返回的字段默认会覆盖 State 中已有的值。后面如果要累积数据需要使用带 reducer 的字段比如Annotated[list, operator.add]。3.2 编写节点函数节点函数是 LangGraph 的基本执行单元。它接收当前 State返回一个字典字典中的字段会被更新到 State 中。下面的代码实现了两个节点def node_a(state: DemoState) - dict: print(f[node_a] 收到输入{state[user_input]}) return {messages: [node_a 处理完成]} def node_b(state: DemoState) - dict: print(f[node_b] 已处理 messages{state[messages]}) return {messages: [node_b 处理完成]}从代码可以看出节点并不需要把整个 State 都返回只需要返回“想要修改的字段”。node_a返回messagesnode_b也是返回messages。由于默认是覆盖更新所以最终messages只会保留最后一次赋的值。这里先不引入 reducer目的是让大家理解基本覆盖逻辑。3.3 连接边并编译有了节点就可以用 StateGraph 把它们连起来from langgraph.graph import StateGraph, START, END graph StateGraph(DemoState) graph.add_node(node_a, node_a) graph.add_node(node_b, node_b) graph.add_edge(START, node_a) graph.add_edge(node_a, node_b) graph.add_edge(node_b, END) app graph.compile()START和END是 LangGraph 内置的虚拟节点表示图的入口和出口。graph.compile()之后得到一个可执行对象app后续调用app.invoke(...)就是执行整条流程。3.4 运行结果分析使用invoke传入初始 Stateresult app.invoke({user_input: hello, messages: []}) print(result)预期输出类似下面这样[node_a] 收到输入hello [node_b] 已处理 messages[node_a 处理完成] {user_input: hello, messages: [node_b 处理完成]}从输出可以看到先执行node_a再执行node_b返回结果是最终 State。如果这里把messages默认覆盖逻辑改成累加逻辑node_b读取到的messages就会包含node_a和node_b的结果。这一步是理解 LangGraph 状态机制的关键。4. 条件路由与分支控制conditional_edges 深度解析4.1 为什么需要条件路由固定链路只能解决“顺序执行”的问题但现实业务更多是“根据不同条件走不同分支”。比如客服系统中用户输入“查订单”和“投诉建议”应该进入不同处理逻辑。LangGraph 通过add_conditional_edges支持条件路由让一个节点根据某个条件决定下一步跳转到哪个节点。条件路由的本质是编写一个路由函数接收当前 State返回一个字符串标志再提供一个映射字典把返回值对应到目标节点。4.2 基础条件路由写法下面是一个完整示例根据finish字段决定流程是继续执行work节点还是直接结束from typing import TypedDict, Annotated import operator from langgraph.graph import StateGraph, START, END class RouteState(TypedDict): steps: Annotated[list[str], operator.add] finish: bool def start_node(state: RouteState) - dict: return {steps: [start]} def route_by_state(state: RouteState) - str: if state.get(finish): return end return continue def work_node(state: RouteState) - dict: return {steps: [work], finish: True} graph StateGraph(RouteState) graph.add_node(start, start_node) graph.add_node(work, work_node) graph.add_edge(START, start) graph.add_conditional_edges( start, route_by_state, { continue: work, end: END, }, ) graph.add_edge(work, END) app graph.compile() result app.invoke({steps: [], finish: False}) print(result)初始状态下finish为 False所以路由函数返回continue流程进入work节点。work节点把finish设为 True下一次如果再次路由就会走到end。这里的Annotated[list[str], operator.add]表示steps字段使用operator.add作为 reducer节点返回的列表会追加到旧列表后面而不是直接覆盖。条件路由中的路由函数必须保持简单明确尽量只读 State 并返回一个字符串。不要在路由函数里做耗时 IO 操作因为路由函数本身不是业务节点它只负责“选择下一站”。4.3 循环检测与递归限制条件路由还有一个常见用途构造循环。比如某个节点需要对一个结果反复处理直到满足条件。下面这个例子展示了一个典型的循环结构from typing import TypedDict from langgraph.graph import StateGraph, START, END class LoopState(TypedDict): count: int def increment(state: LoopState) - dict: return {count: state[count] 1} def should_continue(state: LoopState) - str: if state[count] 3: return continue return stop graph StateGraph(LoopState) graph.add_node(increment, increment) graph.add_edge(START, increment) graph.add_conditional_edges( increment, should_continue, { continue: increment, stop: END, }, ) app graph.compile() result app.invoke({count: 0}, config{recursion_limit: 10}) print(result)这里increment节点执行后回到自身形成一个循环。每次进入increment后count加 1当count达到 3 时should_continue返回stop流程结束。LangGraph 默认会对递归深度做限制。如果你的图真的陷入了死循环LangGraph 会抛出递归超限异常。示例中通过config{recursion_limit: 10}显式允许最多执行 10 步避免误伤正常循环。实际开发中这个值要根据业务合理设计并确保条件路由最终可以收敛否则会因为死循环消耗大量资源。4.4 并行分支让多个节点同时工作LangGraph 支持从同一个节点出发连接多个下游节点从而让这些节点并行执行。在状态更新上如果多个并行节点都修改同一个字段就需要使用 reducer 合并结果。下面是一个例子from typing import TypedDict, Annotated import operator from langgraph.graph import StateGraph, START, END class ParallelState(TypedDict): results: Annotated[list[str], operator.add] def task_a(state: ParallelState) - dict: return {results: [A 的结果]} def task_b(state: ParallelState) - dict: return {results: [B 的结果]} def merge(state: ParallelState) - dict: return {results: [f合并完成{state[results]}]} graph StateGraph(ParallelState) graph.add_node(task_a, task_a) graph.add_node(task_b, task_b) graph.add_node(merge, merge) graph.add_edge(START, task_a) graph.add_edge(START, task_b) graph.add_edge(task_a, merge) graph.add_edge(task_b, merge) graph.add_edge(merge, END) app graph.compile() result app.invoke({results: []}) print(result)这里task_a和task_b都从 START 出发LangGraph 会并行执行它们然后都进入merge节点。因为results字段使用了Annotated[list[str], operator.add]并行节点返回的列表会被追加而不是互相覆盖。最终merge节点拿到的results包含两个任务的结果。这里需要注意的是并行节点的执行顺序并不固定。如果业务上必须区分顺序就不要依赖并行节点而是明确串成一条路径。并行执行主要是为了提升效率比如同时查询订单系统和物流系统。5. 子图Subgraph与 State 修改5.1 什么时候需要拆子图当流程越来越复杂时把所有节点都塞进同一张图会变得难以维护。LangGraph 允许把一张已经编译好的图作为子图嵌套到另一张父图中。这样既能把复杂流程拆成多个模块又能复用公共处理逻辑。比较适合拆子图的场景包括一段反复出现的固定处理流程、一个独立的业务模块、一组可以单独测试的节点集合。子图可以拥有独立的 State也可以和父图共享部分字段数据。5.2 嵌套子图示例先定义一个子图用于统计文本中的单词数量from typing import TypedDict from langgraph.graph import StateGraph, START, END class SubState(TypedDict): text: str count: int def count_words(state: SubState) - dict: return {count: len(state[text].split())} sub_graph StateGraph(SubState) sub_graph.add_node(count_words, count_words) sub_graph.add_edge(START, count_words) sub_graph.add_edge(count_words, END) sub_app sub_graph.compile()接着在父图节点中调用子图class ParentState(TypedDict): text: str words: int upper_text: str def upper_node(state: ParentState) - dict: return {upper_text: state[text].upper()} def run_subgraph(state: ParentState) - dict: sub_result sub_app.invoke({text: state[text]}) return {words: sub_result[count]} parent_graph StateGraph(ParentState) parent_graph.add_node(upper, upper_node) parent_graph.add_node(subgraph, run_subgraph) parent_graph.add_edge(START, upper) parent_graph.add_edge(START, subgraph) parent_graph.add_edge(upper, END) parent_graph.add_edge(subgraph, END) parent_app parent_graph.compile() result parent_app.invoke({text: hello world langgraph, words: 0}) print(result)在这个例子中父图并行执行upper_node和run_subgraphrun_subgraph内部再调用sub_app去执行子图。子图返回的结果通过普通字典传给父图 State。这种“父图节点里调子图”的方式最简单也最容易理解。5.3 节点函数修改 State 的几种方式掌握节点如何修改 State是 LangGraph 使用中最重要的一环。主要方式有三种。第一种直接返回需要更新的字段。节点函数返回字典字典中的 key 会被写入 State。比如return {count: 1}就会把 State 中count字段覆盖成 1。第二种使用 reducer 合并数据。在定义 State 时用Annotated[字段类型, reducer函数]声明累计逻辑。最常见的写法是Annotated[list, operator.add]。这样节点返回一个列表时LangGraph 会自动把新列表追加到旧列表后面。第三种返回空字典或 None。如果节点只是消费 State 的内容不需要修改任何状态可以直接返回空字典。不返回某个字段该字段就不会被修改。需要特别注意的是节点返回的 key 必须存在于 State 定义中否则 LangGraph 会报错。如果你确实需要一些临时变量尽可能在节点函数内部处理不要随意往 State 里塞无关字段。6. 综合实战带路由、循环、并行的工单助手6.1 需求与流程设计现在把前面几节的能力组合起来做一个“客服工单助手”综合案例。需求如下用户输入一段文本系统判断这次输入属于订单咨询还是投诉建议。如果属于订单咨询则并行查询订单信息和物流信息最后汇总结果如果属于投诉建议则直接生成投诉处理话术。整体流程是START - 意图分类 - 条件路由 | |--- order - 并行查询订单 物流 - 合并结果 | |--- complaint - 投诉处理话术 | - 最终输出 - END这个案例不依赖真实大模型所有节点都使用规则或模拟数据方便你把注意力放在图结构上。6.2 完整代码创建ticket_agent.py内容如下from typing import Annotated, TypedDict import operator from langgraph.graph import StateGraph, START, END class TicketState(TypedDict): user_input: str intent: str order_id: str results: Annotated[list[str], operator.add] final_answer: str def classify(state: TicketState) - dict: text state[user_input] intent order if (订单 in text or 物流 in text) else complaint return {intent: intent} def route_intent(state: TicketState) - str: return state[intent] def order_entry(state: TicketState) - dict: return {} def query_order(state: TicketState) - dict: order_id state.get(order_id, A10086) return {results: [f订单 {order_id} 状态已支付待发货]} def query_logistics(state: TicketState) - dict: return {results: [物流信息仓库已打包等待揽收]} def merge_order(state: TicketState) - dict: merged .join(state[results]) return {final_answer: f订单查询结果{merged}} def complaint_reply(state: TicketState) - dict: return {final_answer: 已收到您的投诉我们会在 1 个工作日内人工处理。} def output_answer(state: TicketState) - dict: return {final_answer: f客服助手{state[final_answer]}} graph StateGraph(TicketState) graph.add_node(classify, classify) graph.add_node(order_entry, order_entry) graph.add_node(query_order, query_order) graph.add_node(query_logistics, query_logistics) graph.add_node(merge_order, merge_order) graph.add_node(complaint_reply, complaint_reply) graph.add_node(output_answer, output_answer) graph.add_edge(START, classify) graph.add_conditional_edges( classify, route_intent, { order: order_entry, complaint: complaint_reply, }, ) graph.add_edge(order_entry, query_order) graph.add_edge(order_entry, query_logistics) graph.add_edge(query_order, merge_order) graph.add_edge(query_logistics, merge_order) graph.add_edge(merge_order, output_answer) graph.add_edge(complaint_reply, output_answer) graph.add_edge(output_answer, END) app graph.compile()这段代码里order_entry是一个空节点作用是给并行分支提供一个统一入口。query_order和query_logistics并行执行因为它们都从order_entry出发然后都汇聚到merge_order。results字段使用了operator.add作为 reducer保证并行节点返回的列表不会互相覆盖。6.3 运行与验证编写运行代码order_result app.invoke({ user_input: 我想查一下订单 A10086 的物流, order_id: A10086, results: [], }) print(order_result[final_answer])预期输出客服助手订单查询结果订单 A10086 状态已支付待发货物流信息仓库已打包等待揽收再测试投诉分支complaint_result app.invoke({ user_input: 我对这次服务非常不满意, order_id: , results: [], }) print(complaint_result[final_answer])预期输出客服助手已收到您的投诉我们会在 1 个工作日内人工处理。到这里综合案例就跑通了。你可以在这个基础上做两件事把classify节点替换成真实的大模型分类调用把query_order和query_logistics替换成真实系统的 HTTP 请求或数据库查询。图结构本身不需要大改。7. LangGraph 高频问题与排查清单LangGraph 的报错信息整体比较友好但很多新手会在几个固定位置踩坑。这里整理一份高频问题表你可以直接对照排查。问题现象常见原因解决思路ModuleNotFoundError: No module named langgraph未安装或没有激活正确的虚拟环境执行pip install langgraph检查当前 Python 环境KeyError或字段不存在的报错节点返回了 State 中未定义的 key检查节点返回字典中的 key 是否在 State 类中定义条件路由报错找不到目标节点路由函数返回值不在映射字典中核对路由函数的返回值与映射字典的 key 是否完全一致并行节点更新同一个字段时互相覆盖该字段没有配置 reducer改成Annotated[list, operator.add]或用其他 reducer 合并循环执行次数太多抛出递归限制异常条件边形成了无法收敛的循环检查路由条件或者在invoke时通过recursion_limit调整上限调用invoke返回的结果与预期不一致对 State 的覆盖逻辑理解有误记住默认是覆盖更新需要累加时使用 reducer子图状态和父图状态混淆父图节点与子图使用不同状态定义明确子图输入输出字段在节点函数中做显式映射排查问题时建议按照“State 定义 - 节点返回值 - 边的连接”这个顺序检查。大部分问题都出在这三个环节。尤其是条件路由路由函数的返回值如果多打了一个空格LangGraph 就找不到对应节点。8. 最佳实践与工程建议8.1 State 设计要“小而明确”State 是 LangGraph 应用的核心设计不好后面会很痛苦。不要把每个临时变量都塞进 StateState 中应该只保留“跨节点共享、需要被执行流程记住”的数据。字段命名尽量清晰避免使用data、info、tmp这类模糊名称。如果字段需要累加、追加或合并提前用Annotated加上 reducer而不是在节点函数里手工拼接后覆盖。8.2 命名与模块划分节点名称建议使用“动词 名词”的结构例如query_order、merge_result、classify_intent。这样图结构打印出来之后一眼就能看出流程在做什么。项目文件可以按业务模块划分每个模块负责一组相关节点。子图是拆解复杂流程的好工具但不要为了用子
