1. 从 LangChain 到 LangGraph为什么需要一张“图”如果你已经用 LangChain 写过几个 Demo大概率经历过这样的时刻一个简单的问答链跑得挺顺可一旦想加入“先判断意图、再决定调哪个工具、工具失败要重试、重试超过三次就转人工”这类逻辑代码就开始变成一堆if/else嵌套调试时根本说不清上一步到底走到哪了。LangGraph 出现的直接原因就是给这种“有状态、有分支、有循环”的流程提供一个显式的结构——用图Graph来描述节点和边而不是靠代码里的条件判断硬堆。我最初接触 LangGraph 时的第一反应是这不就是把工作流画成流程图吗但用下来发现它真正解决的不是“画图好看”而是状态管理和控制流这两件在复杂 Agent 里最要命的事。LangChain 的 Chain 本质是一条单向流水线数据从 A 流到 B 再到 C中间很难回头也很难根据中间结果动态改变路线。而 LangGraph 允许你定义节点Node、边Edge和条件边Conditional Edge并且所有节点共享一份可持久化的状态State。这意味着循环、重试、人工介入这些在真实业务里绕不开的需求终于有了一个相对干净的实现方式。这篇文章面向的是已经了解 LangChain 基础、想进一步搞懂 LangGraph 到底怎么用、什么时候该用的人。我会从核心概念讲起把状态、节点、边、检查点这些机制拆开揉碎再给出一套可以直接跑起来的实操流程最后聊聊我在实际项目里踩过的坑。全文不堆概念尽量用“为什么这么设计”的角度来讲让你看完能自己判断某个场景到底该不该上 LangGraph。2. LangGraph 的核心抽象状态、节点与边到底怎么协作2.1 State 不是普通变量而是贯穿全图的共享内存LangGraph 里最核心的概念是 State。你可以把它理解成一张全局的“黑板”每个节点都能读到上面的内容也能往上写东西。和普通函数传参最大的区别在于State 是跨节点持久存在的而且它的更新方式决定了整个图的运行逻辑。State 通常用一个 TypedDict 或者 Pydantic 模型来定义。比如一个最简单的对话 AgentState 里可能包含messages对话历史、user_intent识别出的意图、retry_count重试次数这几个字段。每个节点执行完返回一个字典LangGraph 会把这个字典合并回全局 State。这里有个关键点很多人一开始会忽略State 的字段可以定义“归约函数”reducer。默认情况下节点返回的新值会直接覆盖旧值。但像messages这种列表你往往希望是“追加”而不是“覆盖”。这时候就要用Annotated配合operator.add来声明from typing import Annotated, TypedDict import operator class AgentState(TypedDict): messages: Annotated[list, operator.add] retry_count: int这样每次节点返回新的 message都会被追加到列表末尾而不是把整个历史冲掉。我见过不少人第一次写 LangGraph 时消息莫名其妙丢失八成就是忘了配 reducer。这个设计其实很符合直觉状态更新语义应该由字段自己决定而不是让每个节点都去手动处理合并逻辑。2.2 节点是计算单元但它的返回值有讲究节点本质上就是一个 Python 函数接收当前 State返回一个字典表示要更新的字段。听起来简单但有几个细节决定了你的图能不能正常工作。第一节点返回的字典只需要包含要更新的字段不需要把整个 State 都返回。LangGraph 会负责合并。第二节点可以是同步函数也可以是异步函数这在调用外部 API 时很实用。第三节点内部不要直接修改传入的 State 对象而是通过返回值来更新——这是函数式编程的思路保证了状态变更的可追踪性。举个实际例子一个“意图识别”节点大概长这样def classify_intent(state: AgentState): last_message state[messages][-1] intent llm.invoke(f判断这句话的意图{last_message}) return {user_intent: intent}它只更新了user_intent一个字段其他字段原封不动。这种“只声明增量”的方式让每个节点的职责非常清晰也方便后续做单元测试——你给一个固定的输入 State断言输出字典是否符合预期就行。2.3 边决定了“下一步去哪”条件边才是真正的灵魂普通边Edge是确定性的从节点 A 直接连到节点 B。条件边Conditional Edge则根据当前 State 动态决定下一个节点是谁。这是 LangGraph 区别于普通工作流引擎的关键。条件边的写法是给一个路由函数它读取 State返回一个字符串这个字符串对应你预先注册的节点名。比如def route_after_classify(state: AgentState): if state[user_intent] 查询订单: return query_order elif state[user_intent] 退款: return refund else: return fallback然后在构图时graph.add_conditional_edges( classify, route_after_classify, { query_order: query_order, refund: refund, fallback: fallback } )这个路由函数就是整个 Agent 的“大脑”。我个人的经验是把路由逻辑集中在一个或少数几个函数里比散落在各个节点内部要好得多。因为当流程出问题时你只需要看路由函数就能知道分支条件不用满图找if。2.4 检查点机制让“断点续跑”和“人工介入”成为可能LangGraph 另一个让我觉得设计得很到位的地方是 Checkpointer。每次图执行到一个节点结束它可以把当前 State 存下来。这个存储可以是内存、SQLite、Postgres 等。存下来之后能干什么两件事一是失败后从上一个检查点恢复不用从头跑二是实现 Human-in-the-loop也就是在某个节点前暂停等人工确认后再继续。比如一个自动发邮件的流程你可以在“发送”节点前设置中断把草稿展示给用户用户点确认后再恢复执行。这在 LangChain 的 Chain 里几乎没法优雅实现而在 LangGraph 里只需要配置interrupt_before[send_email]然后用graph.update_state()修改状态再graph.invoke(None, config)继续跑。提示Checkpointer 在生产环境建议用持久化后端内存版只适合本地调试。因为一旦进程重启内存里的检查点就没了断点续跑也就失效了。3. 手把手搭一个带分支和重试的 LangGraph 流程3.1 环境准备与依赖安装先把环境搭起来。LangGraph 是独立包不需要装整个 LangChain 全家桶但通常会配合 langchain-core 和某个模型 SDK 一起用。pip install langgraph langchain-core langchain-openai如果你要用持久化检查点再加一个pip install langgraph-checkpoint-sqlite版本方面LangGraph 迭代挺快建议锁定一个较新的稳定版。我写这篇时用的是 0.2.x 系列API 相对稳定。装完后先跑一个最小示例验证环境没问题别一上来就写复杂图不然报错都分不清是环境问题还是逻辑问题。3.2 定义 State 与三个基础节点我们来做一个“客服工单处理”的小流程用户发一句话先分类意图如果是“查订单”就去查订单如果是“退款”就走退款流程其他情况走兜底回复。查订单失败要重试最多两次。先定义 Statefrom typing import Annotated, TypedDict import operator class TicketState(TypedDict): messages: Annotated[list, operator.add] intent: str order_result: str retry_count: int然后写三个节点分类、查订单、兜底。def classify_node(state: TicketState): user_msg state[messages][-1] intent llm.invoke(f分类以下客服消息的意图只返回查订单、退款或其他{user_msg}) return {intent: intent.strip()} def query_order_node(state: TicketState): # 模拟查询可能失败 if state[retry_count] 2 and random.random() 0.5: return {order_result: 查询失败, retry_count: state[retry_count] 1} return {order_result: 订单号 12345已发货} def fallback_node(state: TicketState): return {order_result: 抱歉我暂时无法处理已转人工}注意query_order_node里我用了retry_count来控制重试次数并且每次失败都自增。这个字段就是靠 State 在节点间传递的。3.3 用条件边实现“失败重试”的循环重试的本质是一个循环查订单失败就回到查订单节点直到成功或超过次数。在 LangGraph 里循环就是一条指回自己的条件边。def route_after_query(state: TicketState): if state[order_result] 查询失败 and state[retry_count] 2: return retry return done构图from langgraph.graph import StateGraph, END builder StateGraph(TicketState) builder.add_node(classify, classify_node) builder.add_node(query_order, query_order_node) builder.add_node(fallback, fallback_node) builder.set_entry_point(classify) builder.add_conditional_edges(classify, lambda s: s[intent], { 查订单: query_order, 退款: fallback, 其他: fallback }) builder.add_conditional_edges(query_order, route_after_query, { retry: query_order, done: END }) builder.add_edge(fallback, END) graph builder.compile()这里route_after_query返回retry时边指向query_order自己形成循环。返回done时指向END流程结束。整个图跑起来后如果查询一直失败最多循环两次就会退出不会死循环。3.4 加上检查点让流程可以暂停和恢复现在给图加上 SQLite 检查点from langgraph.checkpoint.sqlite import SqliteSaver memory SqliteSaver.from_conn_string(checkpoints.db) graph builder.compile(checkpointermemory)调用时要传一个config里面带thread_id用来区分不同的会话config {configurable: {thread_id: ticket-001}} result graph.invoke({messages: [我的订单到哪了], retry_count: 0}, config)有了thread_id同一个会话的状态会被持久化。你可以在任意时刻用graph.get_state(config)查看当前状态也可以用graph.update_state(config, {...})手动改状态。这个能力在做“人工审核后继续”时特别有用先跑到审核节点前中断人工看完改一下 State再继续跑。注意retry_count这种计数字段一定要在 State 里初始化否则第一次读取会 KeyError。我习惯在入口调用时把所有非列表字段都显式传一遍。4. 那些文档里不会明说的坑与调优经验4.1 循环没有出口是最常见的翻车方式LangGraph 给了你写循环的能力但也意味着你得自己保证循环能结束。我见过最典型的事故是条件边在某个边界情况下一直返回“重试”而重试计数因为 reducer 配置问题没有正确累加结果图跑了几百轮把额度耗光。避免这个问题的办法有两个。第一所有循环都必须有一个基于 State 的硬性退出条件比如retry_count max_retry而且这个判断要放在路由函数的最前面。第二给图设置递归限制。LangGraph 编译时可以传recursion_limit默认是 25超过会抛异常。这个默认值其实是个保护别轻易调太高调高之前先确认你的退出条件真的可靠。graph builder.compile(checkpointermemory) result graph.invoke(inputs, config{recursion_limit: 50, **config})4.2 State 字段设计要“够用但不臃肿”State 是全局共享的字段越多每个节点能看到的上下文就越多但同时也越容易产生意料之外的耦合。我的经验是只把需要跨节点传递的数据放进 State节点内部的临时变量不要塞进去。比如某个节点里算了一个中间结果只有它自己用那就用局部变量别加到 State。另外列表类字段一定要想清楚是覆盖还是追加。messages用追加但像tool_results这种每次只关心最新一次的可能覆盖更合适。reducer 配错了轻则数据冗余重则逻辑错乱。4.3 调试时善用 stream 和 get_stateLangGraph 提供了stream方法可以逐节点输出执行过程。调试复杂图时这比invoke好用得多for event in graph.stream(inputs, config, stream_modeupdates): print(event)stream_modeupdates会打印每个节点返回的增量字典你能清楚看到每一步 State 变了什么。如果某个分支没走到或者某个字段没更新一眼就能定位。配合graph.get_state(config)查看完整状态快照基本能覆盖大部分调试场景。4.4 和 LangChain 的关系不是替代是互补经常有人问 LangGraph 和 LangChain 到底什么关系。我的理解是LangChain 提供的是“组件”比如模型封装、工具、检索器、提示模板LangGraph 提供的是“编排”把这些组件按图的方式组织起来。你完全可以在 LangGraph 的节点里调用 LangChain 的 Chain 或 Agent两者不冲突。那什么时候该从 LangChain 迁到 LangGraph我的判断标准是当你的流程出现“需要根据中间结果决定下一步”“需要循环”“需要人工介入”“需要持久化状态”这四个特征中的任意一个时就该考虑 LangGraph 了。如果只是简单的“输入→模型→输出”用 LangChain 的 LCEL 反而更简洁。5. 从 Planning 模式看 LangGraph 的进阶玩法5.1 Planning 模式为什么天然适合用图来表达Planning 模式的核心是先让模型生成一个计划一串步骤然后逐步执行每执行完一步根据结果决定是否调整后续计划。这个“生成计划→执行→评估→可能重新规划”的过程本质上就是一个带循环和条件分支的图。用 LangGraph 实现 Planning通常会有一个planner节点负责生成或更新计划一个executor节点负责执行当前步骤一个evaluator节点判断是否完成或需要重新规划。条件边根据 evaluator 的结果决定回到 planner 还是继续 executor 还是结束。这种结构用 Chain 写会非常别扭因为 Chain 不支持“回到前面某一步”这种控制流。5.2 多 Agent 协作其实是同一套机制的延伸LangGraph 常被用来做多 Agent 系统比如一个“主管 Agent”分配任务给“研究员 Agent”和“写作 Agent”。听起来很高级但拆开看还是 State 节点 条件边。主管节点读取 State 决定派谁条件边路由到对应 Agent 节点Agent 执行完把结果写回 State再回到主管节点判断是否继续。关键设计点在于每个 Agent 的职责边界要清晰State 里要有明确的“当前任务”和“完成标志”字段。否则多个 Agent 互相覆盖状态很容易乱套。我一般会给每个 Agent 的输出单独开一个字段而不是都往messages里塞这样回溯问题时能清楚知道哪条信息是谁产生的。5.3 性能与成本上的取舍图越复杂节点越多调用模型的次数就越多成本和延迟都会上升。我做过一个对比同样一个客服场景用单次 Prompt 直接回答平均 1 次模型调用用 LangGraph 做意图分类工具调用结果润色平均 3 到 4 次调用。延迟大概翻倍成本也翻倍。所以我的建议是不要为了用 LangGraph 而用。先用最简单的方案跑通当简单方案确实撑不住业务复杂度时再引入图编排。引入之后也要定期回看哪些节点是必要的能不能合并。比如意图分类如果准确率够高可以和小模型结合不一定每次都调大模型。6. 我实际落地时总结的几条硬经验第一条先把图画在纸上再写代码。LangGraph 的代码结构和图结构是一一对应的如果图本身没想清楚代码写出来一定是乱的。我现在习惯先用纸笔画出节点和边标清楚每个节点的输入输出字段确认没有死循环和孤立节点再动手写。第二条State 字段命名要能自解释。data、result、info这种名字在单节点里没问题但在跨节点的 State 里就是灾难。用order_query_result、intent_label、retry_count这种一看就知道是什么的名字能省下大量调试时间。第三条检查点不是万能的状态迁移要幂等。断点续跑的前提是节点可以重复执行而不产生副作用。如果你的节点里有“发邮件”“扣款”这类操作一定要做幂等处理比如用唯一 ID 去重否则恢复执行时可能重复触发。第四条别忽略异常处理。LangGraph 的节点抛异常会中断整个图。对于可能失败的外部调用我习惯在节点内部 try/except把错误信息写进 State然后通过条件边决定是重试还是走兜底。这样图不会直接崩用户也能得到反馈。第五条版本升级要谨慎。LangGraph 还在快速演进小版本之间偶尔会有 API 调整。生产项目里建议锁定版本升级前先在测试环境跑一遍完整流程重点看条件边和检查点相关的行为有没有变化。这套东西我用了大半年从最初的小心翼翼到现在基本能比较顺手地设计中等复杂度的 Agent 流程。它确实不是银弹但对于那些“流程本身就很复杂”的场景LangGraph 提供的结构化表达能力比在代码里堆条件判断要清晰太多。如果你正在被复杂的 Agent 逻辑折磨不妨从一个小图开始试试先跑通一个带条件分支的流程感受一下状态在节点间流动的方式后面再逐步加循环、加检查点、加人工介入。