先回答一个很多人问过的问题LangChain 学了很久示例能跑但一遇到“答案不满意就重试”“检索完判断要不要调用工具”“多个任务并行执行再汇总”这种真实 Agent 需求代码就变成了一堆散落的 if-else 和全局变量维护成本直线上升。LangGraph 解决的正是这个问题它把 Agent 的流程变成一张有向图把中间数据变成显式的 State让开发者重新拿回控制权。本文会从零讲清楚 LangGraph 的核心模型并给出一套可运行的实战样例覆盖条件路由、循环检测、State 修改、并行分支、子图和状态持久化。读完你能自己搭建一个带重试机制和并行任务的 Agent 图也知道该去哪里排查问题。文章配套的代码尽量保持最小化方便你复制后直接运行验证。1. 为什么 LangGraph 值得专门花时间学先看一个真实场景。你有一个 RAG 问答功能用户输入问题系统检索知识库大模型生成答案。用普通 Python 写大概是“检索函数 生成函数 主流程”。看起来不难但加上以下需求后就开始失控答案质量不合格时要改写问题重新检索最多重试 3 次。某些问题需要调用外部工具查天气、查订单、查库存调完之后还要再生成一次。用户连续追问时要保留历史消息避免模型“失忆”。多个检索源要并行执行全部完成后再汇总。如果用普通代码管理你需要自己维护循环条件、状态传递、超时控制、重试上限稍不注意就出现无限循环或状态覆盖。很多项目最后变成“能用但不敢改”。LangGraph 的价值在于它把“流程控制”从业务代码里剥离出来交给图执行引擎。节点函数只关心“输入 State 是什么输出什么更新”至于下一步走到哪、是否循环、是否并行、是否中断由图的边和配置决定。适合学 LangGraph 的人通常已经具备以下特征之一用 LangChain 做过 Demo但对 Chain 的线性执行模型不满意。正在开发客服、RAG 重写、工具调用型 Agent需要复杂的控制流。希望把 Agent 流程做得可测试、可恢复、可观测。不适合的情况也有如果只是调一次大模型接口返回结果不需要图如果业务流程非常简单且不会增长上 LangGraph 是过度设计。2. LangGraph 核心概念一张图、四样东西LangGraph 的学习曲线不在于 API 数量而在于思维方式的转变。它的核心抽象可以浓缩成四个概念。2.1 State图运行过程中的数据中枢State 是一个TypedDict定义了图在运行期间需要维护的全部数据。所有节点都能读取当前 State节点返回的更新会被合并回 State。例如一个问答 Agent 的 State 可能是这样的from typing import TypedDict class QAState(TypedDict): question: str answer: str retry_count: int设计 State 时有一个原则只放节点之间需要共享的数据能推导出来的字段不要存。比如answer_length这种可以由answer算出来的字段就不应该进入 State。2.2 Node图中的一个处理单元Node 就是一个普通的 Python 函数或可调用对象。它接收当前 State返回一个 dict表示要更新的字段。def generate(state: QAState) - dict: return {answer: 这是生成的答案}这里有一个新手最容易踩的坑不要在节点函数里直接修改传入的 state 对象而是返回需要更新的字段字典。LangGraph 的推荐做法是返回一个 partial state由框架负责合并。2.3 Edge节点之间的流转关系Edge 定义节点之间的连接。最基本的边是无条件边A 执行完之后一定去 B。graph.add_edge(node_a, node_b)边的类型有两种add_edge无条件跳转。add_conditional_edges条件路由根据 State 动态决定下一步。2.4 Conditional Edge真正的控制流入口条件路由是 LangGraph 区别于普通链式框架的关键。它允许一个节点执行完后由路由函数决定跳转到哪个节点。graph.add_conditional_edges( evaluate, router, # 路由函数接收 state返回节点名 { continue: generate, end: END, } )路由函数是一个纯函数输入 State输出一个字符串 keyLangGraph 根据 key 在映射表里找到目标节点。这个设计让“判断”和“执行”分离逻辑非常清晰。2.5 LangGraph 和 LangChain 的关系很多人在搜“LangGraph 和 LangChain 的区别”。准确地说它们解决的不是同一层的问题。维度LangChainLangGraph核心抽象Chain、Tool、RetrieverState、Node、Edge、Graph执行模型偏线性链式编排有向图支持循环、分支、并行状态管理通常靠显式传参或内存对象显式 State Schema可持久化循环/重试实现起来别扭图本身就是循环结构适用场景简单流水线、组件调用需要控制流的 Agent 应用LangGraph 并不替代 LangChain它们可以配合使用。你完全可以在 LangGraph 的节点里调用 LangChain 的 Retriever、Tool、ChatModel。LangGraph 管“流程怎么走”LangChain 管“每一步用什么组件”。3. 环境准备与最小示例实操之前先准备好环境。本文使用的是 Python 生态代码以 LangGraph 的核心 API 为主。为了避免版本绑定问题我建议你按照实际操作时的官方安装方式为准。pip install langgraph如果要在节点里使用 LangChain 的模型消息类型额外安装pip install langchain-core环境要求建议 Python 3.9 及以上具体版本以你安装的 LangGraph 依赖要求为准。下面是最小示例两个节点从 START 进入 node_a再到 node_b最后到 END。# graph_minimal.py from typing import TypedDict from langgraph.graph import StateGraph, START, END class GraphState(TypedDict): text: str def node_a(state: GraphState) - dict: return {text: state[text] - A} def node_b(state: GraphState) - dict: return {text: state[text] - B} builder StateGraph(GraphState) builder.add_node(node_a, node_a) builder.add_node(node_b, node_b) builder.add_edge(START, node_a) builder.add_edge(node_a, node_b) builder.add_edge(node_b, END) app builder.compile() result app.invoke({text: start}) print(result)运行python graph_minimal.py预期输出{text: start - A - B}这个示例虽然简单但已经把最核心的执行逻辑讲清楚了图接收初始 State按边执行节点每个节点返回的 dict 更新 State最终输出完整的 State。如果运行失败优先检查两点安装的 langgraph 是否成功Python 版本是否满足依赖要求。4. 条件路由与循环检测做一个带重试的 QA Agent条件路由最典型的应用场景就是“判断结果是否合格决定下一步动作”。下面用一个带重试机制的 QA Agent 来演示。业务逻辑是生成答案。评估答案质量。如果答案质量不合格且还没超过最大重试次数回到生成节点重新生成。超过次数后无条件结束。代码如下# graph_conditional.py from typing import TypedDict from langgraph.graph import StateGraph, START, END class QAState(TypedDict): question: str answer: str retry_count: int route: str def generate(state: QAState) - dict: current_try state[retry_count] 1 return { answer: f第 {current_try} 次生成的答案, retry_count: current_try, } def evaluate(state: QAState) - dict: # 模拟质量评估超过最大次数或答案包含关键词则结束 if state[retry_count] 3: return {route: end} if 可用 in state[answer]: return {route: end} return {route: regenerate} def router(state: QAState) - str: return state[route] builder StateGraph(QAState) builder.add_node(generate, generate) builder.add_node(evaluate, evaluate) builder.add_edge(START, generate) builder.add_edge(generate, evaluate) builder.add_conditional_edges( evaluate, router, { regenerate: generate, end: END, } ) app builder.compile() result app.invoke({ question: LangGraph 是什么, answer: , retry_count: 0, route: , }) print(result)这里的关键在于add_conditional_edges的用法evaluate 节点执行完后不会直接走固定的下一条边而是调用 router 函数router 返回regenerate就回到 generate返回end就结束。运行这个示例你会看到 retry_count 从 1 增长到 3最终停在 END。它证明了 LangGraph 天然支持循环不需要你用 while 自己控制。循环检测有两个层面。第一个层面是业务层面在 State 中维护计数由节点函数判断是否继续。第二个层面是框架层面LangGraph 有recursion_limit配置防止因逻辑 bug 导致无限循环。result app.invoke( {question: LangGraph 是什么, answer: , retry_count: 0, route: }, config{recursion_limit: 10}, )当图的执行步数超过recursion_limit时框架会抛出异常。生产环境建议两个层面都设置业务逻辑里做循环上限判断invoke 时也设置一个大于预期最大步数的 limit 作为兜底。5. 在节点函数中正确修改 State 状态值这是 LangGraph 新手问得最多的问题之一节点函数里到底怎么改 State很多人的第一反应是def my_node(state): state[answer] xxx # 不推荐LangGraph 的推荐做法是节点函数不直接原地修改 State 对象而是返回一个 dict包含想要更新的字段。LangGraph 会把返回值合并进全局 State。def my_node(state): return {answer: xxx}为什么这样设计因为 LangGraph 需要知道 State 的变更历史才能支持断点恢复、时间旅行、并行分支合并等能力。如果所有节点都原地改同一个字典框架很难追踪“谁改了什么”。但如果字段是列表、消息序列这类需要“追加”而不是“覆盖”的数据直接返回新值会把旧值整个替换掉。解决办法是使用 reducer。下面是一个消息追加的示例# graph_state_update.py from typing import TypedDict, Annotated from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langchain_core.messages import HumanMessage, AIMessage class ChatState(TypedDict): messages: Annotated[list, add_messages] status: str def chat_node(state: ChatState) - dict: user_text state[messages][-1].content return { messages: [AIMessage(contentf收到{user_text})], status: processed, } builder StateGraph(ChatState) builder.add_node(chat, chat_node) builder.add_edge(START, chat) builder.add_edge(chat, END) app builder.compile() result app.invoke({ messages: [HumanMessage(content你好 LangGraph)], status: pending, }) print(result)输出结果中messages 会同时包含 HumanMessage 和 AIMessage而不是把 HumanMessage 覆盖掉。这是因为messages字段的类型是Annotated[list, add_messages]LangGraph 发现该字段有 reducer就会用 reducer 合并新旧值。不同字段的更新行为对比字段定义更新行为典型场景answer: str新值覆盖旧值状态字段messages: Annotated[list, add_messages]追加而非覆盖对话消息列表results: Annotated[list, merge_list]自定义合并逻辑并行任务收集结果如果你需要自定义合并逻辑只要在类型标注里写Annotated[list, my_reducer]即可。reducer 是一个接收两个列表并返回一个新列表的函数。6. 并行分支与子图复杂 Agent 的组装方式真实 Agent 经常需要同时做多件事。比如用户输入一个问题后系统要并行检索知识库、查询历史会话、调用天气工具等所有结果都返回后再汇总。LangGraph 支持两种并行方式静态并行和动态并行。6.1 静态并行fan-out 与 fan-in在一个 StateGraph 中多个节点可以同时从一个起始节点流出然后汇聚到一个节点。下面是一个并行检索的示例。# graph_parallel.py from typing import TypedDict, Annotated from langgraph.graph import StateGraph, START, END def merge_list(left: list, right: list) - list: return left right class ParallelState(TypedDict): question: str results: Annotated[list, merge_list] def search_kb(state: ParallelState) - dict: return {results: [f知识库结果{state[question]}]} def search_history(state: ParallelState) - dict: return {results: [f历史会话结果{state[question]}]} def aggregate(state: ParallelState) - dict: return {results: [f汇总{state[results]}]} builder StateGraph(ParallelState) builder.add_node(search_kb, search_kb) builder.add_node(search_history, search_history) builder.add_node(aggregate, aggregate) builder.add_edge(START, search_kb) builder.add_edge(START, search_history) builder.add_edge(search_kb, aggregate) builder.add_edge(search_history, aggregate) builder.add_edge(aggregate, END) app builder.compile() result app.invoke({question: LangGraph 并行, results: []}) print(result)这段代码的关键在于results字段使用了merge_listreducer。两个并行节点都会向results追加内容如果没有 reducer后返回的节点会覆盖先返回的节点aggregate 只能看到一个来源的数据。并行节点更新同一个 State 字段时必须通过 reducer 定义合并规则这是并行写法里最重要的注意事项。6.2 动态并行使用 Send API如果并行任务的数量是动态的比如用户上传了 5 个文件每个文件需要一个处理节点这时静态写边就不够了。LangGraph 提供了SendAPI用于在条件路由中动态创建多个并行任务。from langgraph.types import Send def continue_to_process(state): return [ Send(process_one, {item: item}) for item in state[items] ]然后在图中使用条件边builder.add_conditional_edges( distribute, continue_to_process, [process_one], )Send的第一个参数是目标节点名第二个参数是传给该节点的 state。这样就能把一个分发节点拆成 N 个并行任务。具体写法在不同版本中可能略有差异建议以官方文档为准。6.3 子图把一张图塞进另一个节点当一个图变得复杂时你应该把它拆成多个子图。LangGraph 支持把一张已编译的图作为另一个图的节点来使用这很适合模块化设计。inner_builder StateGraph(InnerState) # ... 定义 inner 图的节点和边 ... inner_app inner_builder.compile() outer_builder StateGraph(OuterState) outer_builder.add_node(inner, inner_app)这种做法相当于把一个完整的子流程封装成一个节点。外层图只看到“有一个节点叫 inner”内部逻辑完全隔离。注意子图的输入输出字段需要和外层 State 对齐如果字段名不一致可以在子图入口和出口加转换节点。7. 状态持久化与断点恢复Agent 应用一旦进入生产环境“执行到一半进程挂了怎么办”“用户的会话状态怎么保存”就是必须面对的问题。LangGraph 的 Checkpointer 机制解决的就是这个问题。引入 checkpointer 后LangGraph 会在每一步执行后保存 State 快照。配合thread_id可以区分不同会话。from langgraph.checkpoint.memory import MemorySaver memory MemorySaver() app builder.compile(checkpointermemory) config {configurable: {thread_id: thread-1}} result1 app.invoke({question: 你好}, configconfig) result2 app.invoke({question: 继续刚才的话题}, configconfig)有了 checkpointer同一个thread_id的多次 invoke 之间State 是共享的。这意味着多轮对话不再需要自己手动拼接历史消息checkpoint 会帮你维护。生产环境不建议使用MemorySaver因为数据只存在内存里进程重启就丢了。更稳妥的做法是使用基于 SQLite 或 Postgres 的持久化 saver并将存储会话数据的数据库单独备份管理。具体 saver 的导入路径和配置方式会随版本变化接入前务必查阅官方文档。需要注意的安全边界checkpoint 里可能包含用户敏感消息。如果用于生产存储端要做访问控制遵循最小权限原则不要把所有用户的 thread 放在一个无鉴权的存储里。8. 常见问题与排查思路问题现象可能原因排查方式解决方案运行时报 Recursion limit 错误条件路由形成死循环或循环次数超过框架默认配置查看报错信息中的执行步数检查路由函数和 State 中的计数逻辑在业务 State 中加循环上限判断并在 invoke 时合理设置 recursion_limit节点返回后 State 没有变化返回 dict 的 key 与 State schema 中的字段不一致打印返回 dict 和 graph 的 State schema 对比确保返回的 key 在 State 中定义并行节点更新同一个字段结果互相覆盖该字段没有定义 reducer检查字段的类型标注是否有 Annotated给字段添加 merge 类型的 reducer消息列表只保留了最后一条messages 字段没有使用 add_messages reducer检查 TypedDict 中的字段定义使用Annotated[list, add_messages]条件路由返回的节点不存在router 返回值与 add_conditional_edges 的映射 key 不一致打印 router 返回值核对映射表中的 key 和节点名调用图执行很慢但没有报错节点内部可能有网络重试或分支实际是串行执行给节点函数加日志确认是否并行检查并行边是否真的存在任务内部增加超时控制提示模型 API Key 未配置节点内调用模型时环境变量不存在检查环境变量是否加载使用环境变量或密钥服务管理 Key不要硬编码排错时有一个通用顺序先看异常发生在“图执行层”还是“节点内部”。如果是图层面错误信息通常会提到节点名和边如果是节点内部通常能看到业务代码的堆栈。给每个节点函数加一行入口日志是定位问题最便宜的手段。9. 工程落地与最佳实践9.1 设计 State 时保持克制State 是图的数据协议字段越少越容易维护。每新增一个字段前问自己这个字段真的需要跨节点共享吗能否由其他字段推导出来9.2 节点函数尽量写成纯函数节点函数只依赖传入的 State不读写全局变量不直接操作外部状态。这样每个节点都可以单独测试也方便在失败时重放 State。9.3 条件路由函数保持轻量路由函数只负责根据 State 返回节点名不要在里面做重 IO、调模型、查数据库。路由函数执行频率高一旦变慢会影响整个图的调度。9.4 永远做循环防护即使业务上认为不会无限循环也要设置recursion_limit并在 State 里维护计数。线上 Agent 的异常往往不是模型报错而是流程失控。9.5 密钥与安全边界模型 API Key、数据库密码等敏感信息放在环境变量或密钥管理服务中不要提交到代码库。节点函数如果访问外部系统要考虑超时、限流和失败降级。Agent 工具调用还可能涉及用户数据权限工具层要做鉴权不能因为“流程跑通了”就跳过权限校验。9.6 测试策略对条件路由函数做单元测试是最划算的给定 State 输入断言返回的节点名正确。对完整图做集成测试时用 mock 模型固定返回内容避免测试结果不稳定。涉及数据库或外部服务的变更先在测试环境验证再走发布流程。9.7 不要过度设计如果一个流程只需要三步线性调用直接写函数调用更清晰。Graph 的价值在复杂度出现以后才显现。一张图如果超过十几个节点建议拆成多个子图否则排查问题时会很难受。9.8 关注版本兼容性LangGraph 迭代速度不慢API 偶尔会有调整。学习时以官方文档为准不要照搬过时的博客代码。搜索“LangGraph 是否有 Rust 版本”这类问题时也要注意实际开发中 Python 生态最为完整除非你有明确的跨语言需求否则优先走 Python 路线把核心的状态机设计思想学扎实。10. 总结与后续学习方向本文的核心判断是LangGraph 不值得死记 API真正需要理解的是“State 定义数据、Node 处理状态、Edge 控制流转、Conditional Edge 实现决策”这套状态机模型。把最小示例跑通之后再逐步加入条件路由、循环检测、并行分支、子图和 checkpointer就是一条很平滑的学习路径。如果你今天只做一件事建议把第 4 节的条件路由示例亲自跑一遍并试着把“重试判断”改造为自己的业务逻辑。跑通了之后再去看官方文档里的高级主题你会发现很多概念已经自然理解了。下一步可以从这几个方向继续深入用 LangGraph 重写一个已有的 LangChain 线性 Demo从链式调用改造成图结构。给图加上 checkpointer体验多轮对话和历史恢复。实现一个人工审核节点Agent 生成结果后暂停人工确认后再继续执行。把多个子图组合成一个完整的上层应用。关于网上各种“2026 全套教程”的说法有一点可以放心LangGraph 的核心模型短期内不会发生颠覆性变化。把 Graph、State、Node、Edge 这四个概念吃透再多的版本更新对你来说都只是增量知识。