做 LangGraph 系列写到第三篇我在后台收到的私信终于从节点怎么写变成了LangChain 和 LangGraph 到底什么关系我是不是学了要过时的东西。这两个问题其实指向同一个症结大家已经能跑通最简单的 hello world 图但一想到要把它用在真实业务里就不知道怎么设计了。这篇我不打算按官方文档的顺序平铺我想换一个更实战的视角按照我自己在项目里踩出来的路径把几个关键问题一次讲清楚——从链到图的思维转换、状态设计为什么是工作流的天花板、Planning 模式和 Human-in-the-loop 这些高级编排怎么落地以及最后多智能体协作和部署阶段的翻车点。这篇文章适合已经开始接触 LangGraph、能跑通基础图、但想在真实业务里用得更顺手的读者。如果你是纯零基础建议先补完系列前两篇再回来看不然部分概念会显得跳跃。1. 从链式调用到状态机思维LangChain 和 LangGraph 的本质差异1.1 同一个生态两种完全不同的执行模型先下结论LangChain 和 LangGraph 不是升级关系而是两种并存的编排范式。LangChain 的核心抽象是 Chain本质是预先定义好的一系列调用LLM 调用、工具调用、解析逻辑全部写在一个顺序列表里。一旦流程跑起来顺序基本就固定了想动态跳转、循环、回退代码会变得很难维护。LangGraph 的核心抽象是图Graph。节点Node是执行单元边Edge是跳转规则状态State是全局共享的数据。它的执行模型更像一个有限状态机每次把一个输入喂进图图根据当前节点的返回结果决定下一个节点是谁可以分支、可以循环、可以暂停、可以恢复。注意这里的可以循环是 LangChain 那条链路本质上做不到的链条是单向的图是带环的。我自己刚切换时感受特别明显以前写一个带多步工具调用的 Agent要在 Chain 里堆一堆 AgentExecutor 和中间解析逻辑出了问题根本分不清是哪一步换成图之后每个节点就是一张小卡片甚至可以直接把图画出来LangGraph 自带 draw 相关方法哪条边断了、哪个节点超时一眼就能看出来。维度LangChain ChainLangGraph Graph执行模型顺序链式调用有向图状态机可循环、分支、回退状态管理各步骤内部局部变量全局 State 对象节点间显式传递人机交互几乎不支持原生支持 interrupt / Human-in-the-loop持久化无内建恢复机制Checkpointer 可随时快照、恢复、回滚适用场景确定性简单流程复杂决策、多步推理、Agent 协作1.2 为什么真实业务最终都会走向图结构真实业务里的 AI 工作流几乎都是非线性的。一个文本处理任务可能要先去识别输入类型再决定走摘要分支还是翻译分支一个自动写稿任务可能需要写一版草稿 → 人工确认 → 不合格重写 → 合格后发布的循环。这些流程如果用 Chain 写本质上是在用 if-else 强行模拟状态机随着分支增多代码很快就变成一坨意大利面。LangGraph 把下一步取决于当前结果的能力变成了框架原语。你不需要自己维护流程控制变量只需要声明节点和边。这同时也解释了为什么 LangGraph 在 Agent 场景这么流行Agent 最核心的行为——LLM 决定调工具、工具返回结果、LLM 再决定下一步——天然就是图上的一个自环self-loopLangGraph 把这个模式直接做成了框架的一等公民。提示判断一个项目该用 Chain 还是 Graph我的经验非常简单只要流程有根据运行结果改变路径的可能就直接用 Graph。确定性顺序流程用 Chain 确实更简洁但业务需求几乎一定会变从 Chain 迁到 Graph 的成本远高于一开始就用 Graph。1.3 迁移过来时最容易犯的第一个错误很多从 LangChain 迁过来的同学包括我本人当年最常犯的错是把图节点写成了一个节点里塞进一整条子链。比如有人把写文章这一个节点里放了 5 次 LLM 调用、2 次工具调用、3 次解析逻辑。表面上是图实际上还是链State 里看不到中间过程出了问题依然没法定位。正确做法是给每个有独立语义的步骤单独建节点。一次 LLM 调用、一次工具执行、一次结构校验都值得拆成一个节点。宁可节点多一点也不要让一个节点干太多活。拆完之后图的表达能力才开始真正释放——你可以任意调整顺序、插入人工确认节点、给不同节点配不同的重试策略。这个原则在后面几节会反复用到。2. 状态设计决定工作流的天花板State、Reducer 与通道2.1 State 不是普通字典是带归约器的数据契约LangGraph 的 State 本质上是一个 TypedDict但它有一个很关键的机制每个字段都可以声明一个 Reducer归约器。Reducer 解决的是多个节点修改同一个字段时怎么合并的问题。很多人第一次接触会忽略它直到出现数据互相覆盖才回头看文档其实这一步应该一开始就想清楚。最常见的场景是聊天多个节点都往对话历史里追加消息。如果 messages 字段是普通覆盖那模型上下文会丢失如果每个节点自己写合并逻辑又全是重复代码。LangGraph 提供了现成的 add_messages 归约器它的语义是新消息追加到已有列表后面而不是覆盖。from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class ChatState(TypedDict): messages: Annotated[list, add_messages] current_role: str draft: str def generator_node(state: ChatState) - dict: return {messages: [{role: assistant, content: 草稿完成}]}这个设计为什么重要因为它把节点更新共享状态这种容易出错的操作收敛成字段级别的声明式规则。我打个比方普通字典像一张公共白板谁都可以乱写带 Reducer 的 State 像一份带版本控制的共享文档每个人只能做规定的编辑操作系统自动合并。你不需要在每个节点里写防御式判断数据一致性由框架保证。2.2 什么时候需要自定义 Reduceradd_messages 是最常用的归约器但远不是全部。真实项目里我常用到两类自定义归约器第一类是取最新值。状态里有个 current_query 字段多个前置节点都可能修改它但我们只关心最终结果。其实字典字段不声明 reducer 时默认行为就是覆盖所以取最新值常常不需要额外声明。第二类是累积求和。统计多个并行子任务的成功数、失败数就可以用 operator.add。import operator from typing import Annotated, TypedDict class WorkflowState(TypedDict): success_count: Annotated[int, operator.add] tags: Annotated[list, operator.add]每个并行节点返回 success_count: 1 或 0最终在汇聚节点只需要读一次 state就能拿到所有分支的统计信息不用再维护一个额外计数器。这正好弥补了各节点并行操作同一字段的核心痛点。2.3 状态设计的三条实操经验状态设计几乎决定了一个图后面好不好维护。分享三条经验第一字段宁可先少后多。不要一开始就把所有中间变量都塞进 State。每个新字段都意味着所有读写它的节点要一起维护字段一多图就很乱。先只放必须跨节点传递的数据局部计算用节点内部变量。第二消息类字段统一用 add_messages 管理。即使是单 Agent 场景也建议统一消息格式因为后期几乎一定会加人工审核节点、会话历史回放消息列表是你最重要的上下文载体。第三State 尽量用可序列化类型。int、str、list、dict 是安全的尽量避免把自定义类对象直接放进 State。原因在于 LangGraph 的 Checkpointer 会把状态快照序列化持久化存了自定义类会面临反序列化问题。实在要传对象就传 id让节点内部再去查。3. Planning 模式与条件分支让图自己决定下一步3.1 条件边是怎么工作的LangGraph 里下一个节点是谁由 add_edge 和 add_conditional_edges 决定。条件边允许你注册一个路由函数这个函数读取当前 State返回下一个节点的名称。官方文档习惯把这类节点叫 router。from langgraph.graph import StateGraph, START, END def route_after_classify(state: WorkflowState) - str: if state[category] translation: return translate_node if state[category] summary: return summarize_node return fallback_node graph.add_conditional_edges( classify_node, route_after_classify, { translate_node: translate_node, summarize_node: summarize_node, fallback_node: fallback_node, }, )这里第二个参数是路由函数第三个参数是逻辑标签 → 实际节点名的映射表。很多人不理解为什么要这个映射表其实目的是解耦路由函数里返回的是逻辑标签比如translation至于哪个节点叫 translate_node 还是 translator_agent由映射表统一管理。这样在图上重命名节点时不需要改路由函数内部逻辑。我经常用这种模式实现护栏分类节点先判断输入是否合法命中风险内容就走风险提示分支否则才进入正常处理节点。这比在业务节点里写一堆 if 判断要干净得多。3.2 Planning先规划后执行的两阶段结构Planning 模式是大模型工作流里非常实用的一种结构搜索热词里也频繁出现。思路很简单第一个节点让 LLM 把复杂任务拆解成步骤列表后面的执行节点按照步骤列表逐项完成。class PlanState(TypedDict): original_task: str plan: list[dict] # [{step: 1, action: search}, ...] executed_steps: Annotated[list, operator.add] def planner_node(state: PlanState) - dict: response chat_model.invoke(plan_prompt.format(taskstate[original_task])) return {plan: response.steps}然后在条件边里判断计划里还有没执行的步骤吗有就回到执行节点没有就进入汇总节点。这个执行 → 判断 → 再执行 → 再判断的循环就是 Planning 模式的骨架也是最基础的自环结构。这里有个关键点执行节点每轮最好只执行一步并把执行结果追加到状态里。这样可以随时插入人工确认、随时暂停恢复、随时跳过某一步。我见过有人把执行节点写成一个 for 循环把计划一次性跑完那就完全没有发挥出图的价值——一旦中间某一步出错整个任务就得从头再来而且没有任何中间状态可以审计。3.3 实际项目里的 Planning 模板我在项目里常用的 Planning 模板分三段第一段是任务理解让模型输出任务目标、约束条件、输入里有哪些关键信息。第二步是步骤拆解要求模型输出步骤列表每步包含动作类型搜索、生成、校验、总结和输出格式。第三段是动态调整每执行完一步都把结果反馈给规划节点允许模型重新规划剩余步骤。动态调整这步非常关键因为 LLM 最初生成的计划在拿到真实工具结果之后往往会变。比如计划第一步是搜索三个关键词实际搜索后只找到两个有价值来源那么规划节点看到搜索结果后应该把后续步骤改成补充搜索或直接进入写作。这正是图比链的优势体现规划节点和执行节点之间形成闭环计划不是一次性生成后就闷头执行完。3.4 条件边的边界条件处理条件边有一个很容易被忽略的问题路由函数返回的标签如果不在映射表里程序会直接报错。我在生产环境遇到过一次原因是某个节点数据的枚举值出现了训练数据里没见过的类型路由函数没法映射到已有分支。处理办法是路由函数永远要有一个 fallback 分支。哪怕你觉得这种情况不可能出现也要把未知值统一导到一个兜底处理节点去这个节点可以记录日志、通知人工、或者给一个固定的安全回复。把这个习惯养成之后线上事故能少一半。4. Human-in-the-loop 与 interrupt把人类放进 AI 工作流4.1 为什么需要中断机制搜索热词里 HITLHuman-in-the-loop出现频率极高说明大家都在关注同一个问题AI 全自动执行不可靠关键节点必须有人确认。LangGraph 对这个场景做了原生支持核心 API 是 interrupt()。interrupt 的原理可以理解为图在某个节点执行时遇到 interrupt会把当前状态存储到 Checkpointer然后向外部调用方暴露一个暂停信号外部程序把这个暂停信号展示给人类收集人工反馈再把反馈“喂”回断点图从断点继续执行。每一步都说得通但实际写的时候有几个细节容易懵下面展开。from langgraph.types import interrupt, Command def approval_node(state: WorkflowState): user_input interrupt( { draft: state[draft], question: 是否批准发布 } ) decision user_input.get(decision, reject) return {approved: decision approve}代码里有个很关键的行为interrupt 被打断时节点后续代码不会执行状态停留在那一刻当外部把结果通过 Command(resume...) 送回来时interrupt 的返回值才会被填上节点继续往下走。注意它不是重新运行整个节点而是真正地从断点恢复。这意味着你可以在断点前已经做过的副作用操作比如调用过的 API不会重复执行。4.2 三种常见的人机协作模式我总结过三种最常见的 HITL 使用方式新手建议从第一种开始。第一种是草稿审核。LLM 生成内容后停住人工在界面上修改、确认确认后继续发布流程。这是最简单也最稳的模式适合内容生成类业务。第二种是可执行动作确认。Agent 在调用外部工具发送邮件、调用支付接口、修改数据库之前中断向人类展示即将执行的操作和参数等确认后才真正调用。对于有副作用的工具我强烈建议一律接入这个模式这是安全底线。第三种是数据修正。当 Agent 从文档提取的信息置信度低或校验不通过时暂停并让人工直接修改提取结果再继续后续流程。这种模式对 RAG、文档整理类场景特别好用与其让模型猜不如给人一个明确的修正入口准确率提升非常明显。4.3 一个要踩过才知道的坑断点恢复时内容重复我早期项目里遇到过一个问题人工确认之后LLM 节点把之前已生成的内容又生成了一遍导致上下文里出现重复。排查了很久才发现原因是审核节点把人工修改的内容作为新消息追加到了 state[messages]但重新进入生成节点时模型看到的上下文里既有旧草稿又有新修改自然就重复了。解决方式有两个一是审核节点不要往 messages 里追加内容只更新一个 approved 字段人工的修改通过 Command(resume...) 单独传二是如果必须把修正内容写进 messages要确保 reducer 是追加而不是覆盖并且人工节点只追加一次。我建议优先用第一种让审核结果和对话历史分开存储职责更清晰。注意HITL 节点一定要单独拆开不要和生成节点合在一起。中断恢复的语义一旦和业务逻辑纠缠调试会变得非常痛苦。图的价值恰恰在于让每个停顿点都成为一等公民可以单独配置超时、日志、重试。4.4 异步场景怎么等人确认还有一点实战中常遇到HITL 节点中断后人可能几分钟甚至几小时才确认同步等待显然不现实。LangGraph 官方推荐的模式是服务端把图执行挂起把断点信息存到你的应用数据库里前端轮询或通过 WebSocket 通知用户有待确认项用户确认后后端拿着之前保存的 thread_id 和 config调用 graph.invoke 或 graph.update_state 把结果送回。这里的关键是 thread_id 和 checkpoint 必须可靠保存。生产环境我一般用 Postgres 作为 Checkpointer这样即使服务重启挂起的任务也不会丢。这属于工程层面的设计留给后面讲服务化部署时再展开但你心里需要有这个数HITL 不只是一个 API它需要配套的存储和通知机制。5. 多智能体协作与并行从单 Agent 到 Agent 团队5.1 为什么单 Agent 不够用当任务复杂度上升单个 Agent 会出现几个问题上下文越来越长导致注意力分散工具数量太多导致工具选择错误率上升权限难以隔离一个 Agent 既读文件又发外部请求风险很大。多 Agent 架构核心思路是把不同职责的 Agent 拆开由一个协调者Supervisor决定把任务交给谁、什么时候回收结果。在 LangGraph 里每个 Agent 本身可以是一个子图Subgraph大图把子图当成普通节点调用。图中有图这是 LangGraph 很实用的能力。writer_subgraph StateGraph(WorkflowState) # ... 定义 writer 子图的节点和边 compiled_writer writer_subgraph.compile() main_graph.add_node(writer, compiled_writer) main_graph.add_node(reviewer, reviewer_agent) main_graph.add_edge(writer, reviewer)这样做的好处非常直接子图可以独立测试可以在不同项目里复用还可以单独给子图配 Checkpointer。比如写作者子图可以在超时时单独重跑而不会影响整个大图的全局状态。5.2 监督者模式的实现思路监督者模式是最经典的多 Agent 协作方式。监督者节点本身也是一个 LLM 调用它的任务不是干活而是调度读取当前状态和各个子 Agent 的最近输出决定下一次行动由谁执行。def supervisor_node(state: WorkflowState) - dict: response chat_model.invoke( supervisor_prompt.format( taskstate[original_task], recent_outputsstate[agent_outputs][-3:] ) ) return {next_agent: response.agent_name} main_graph.add_conditional_edges( supervisor, lambda state: state[next_agent], { researcher: researcher, writer: writer, reviewer: reviewer, finish: END, }, )这里的路由逻辑和前面 3.1 节完全一致。监督者节点不直接修改业务数据只输出下一步调谁这样整个工作流的走向是可控的、可审计的。你会清楚地看到每一轮调度决策是谁做的、依据是什么。我在实际项目里会让监督者在连续 N 次调度后强制休息给审核节点让路避免两个 Agent 来回踢皮球的情况。比如writer 写完了 → reviewer 说重写 → writer 又写 → reviewer 又说重写这种循环除了浪费 token 没有意义加上轮次上限之后超限就直接转人工。5.3 并行分支fan-out 与 fan-in多 Agent 之后的下一步就是并行。LangGraph 支持从同一个节点同时向多个节点发出执行信号对应图论里的 fan-out多个分支结束后汇集到同一个节点对应 fan-in。from langgraph.types import Send def fan_out(state: WorkflowState): return [Send(summarize_one, {doc: doc}) for doc in state[documents][:5]] graph.add_conditional_edges( fan_out, fan_out, [summarize_one], )这个 Send API 是 LangGraph 并行执行的核心它允许动态决定要复制出多少个并行任务每个任务带上各自的输入。注意这里的条件边返回的不是节点名而是 Send 对象列表框架会为每个 Send 创建一次独立的执行。并行分支结束后用 operator.add 之类的 reducer 把结果汇总到同一个字段。我实测下来5 个并行子任务相比串行能快 4 倍以上理论极限 5 倍实际受限于 LLM API 并发配额。但要注意并行分支越多LLM API 的 rate limit 越容易被打满建议在调用层做信号量限流给每个 provider 客户端设置最大并发数。6. 部署阶段的几个坑Checkpointer、重试与流式输出6.1 Checkpointer 选型和记忆分层LangGraph 的 Checkpointer 负责持久化图的运行状态是所有高级特性的地基。默认的 InMemoryCheckpointer 只适合本地调试生产环境至少要用 SQLite 或 Postgres。选型的核心指标是你的服务是多实例部署吗会做蓝绿发布吗服务重启需要恢复挂起任务吗如果答案都是是那直接上 Postgres。我推荐把记忆分成两层会话记忆用 Checkpointer保存图执行的全部状态长期记忆用外部数据库或向量库保存用户偏好、事实摘要。不要把长期产物全塞进 Checkpointer否则每次恢复图都要加载超大状态性能会很难看。from langgraph.checkpoint.sqlite import SqliteSaver with SqliteSaver.from_conn_string(checkpoints.db) as saver: graph builder.compile(checkpointersaver) config {configurable: {thread_id: user-123}} result graph.invoke(initial_input, configconfig)thread_id 是运行时隔离的关键参数。同一个 thread_id 的多次 invoke 会自动接着上一轮状态继续不同 thread_id 之间互不影响。这个设计对多用户 Web 应用极其友好你不需要自己管理每个用户的会话变量LangGraph 已经替你做好了。6.2 节点重试与超时以及幂等性大模型调用是极不稳定的远程 IO不设重试策略的生产系统就是在裸奔。LangGraph 的节点支持直接配置重试和超时。graph.add_node( llm_call, llm_call_node, retryRetryPolicy(max_attempts3, initial_interval1.0), timeout60, )但我更想强调重试的幂等性问题。如果节点本身有副作用比如已经调用过发送 API 又重试就会重复发送自动重试反而是灾难。我的处理办法是分两类纯计算和只读类节点放心配置自动重试有副作用的节点不配自动重试而是在节点内部自己检查外部状态查数据库看这笔操作是否已完成或者把副作用做成幂等接口。记住重试机制只能解决调用失败不能解决调用成功但结果未知。6.3 流式输出别只看默认参数LangGraph 的 stream() 方法支持多种模式最常见的坑是直接调默认模式只能拿到最终结果中间节点的过程输出全丢了。如果你要给前端展示正在思考中正在调用工具这类渐进提示必须指定 stream_mode。for chunk in graph.stream(initial_input, configconfig, stream_modeupdates): # updates 模式下chunk 是 {node_name: node_output} 格式 for node_name, output in chunk.items(): print(f{node_name}: {output})values 模式可以看每一步完成后的完整状态messages 模式专门用于流式输出 LLM token。我通常的组合是messages 模式给前端做 Token 流式渲染updates 模式给前端展示节点状态变化。两者可以同时启用stream_mode[updates, messages]。如果你只传一个 stream_mode默认行为是 values也就是每一步结束时的完整状态而不是增量的中间输出别搞混了。6.4 调试利器把图画出来最后分享一个非常实用的小技巧LangGraph 编译后的图对象可以一键导出图结构我每次改完图都会先导出看一眼再跑测试。compiled_graph.get_graph().draw_mermaid_png(output_file_pathworkflow.png)虽然生成的图不一定美观但能一目了然地确认边有没有连对、有没有出现意外的并行分支。真实开发里我见过太多逻辑看着没问题但图结构全乱的事故一张图能省半小时排查时间。还有个细节如果你在 Jupyter 或本地调试时环境缺 Graphviz上面这个 draw_mermaid_png 不一定能直接跑通。备选方案是调 compiled_graph.get_graph().to_json()把图结构转成 JSON 再对着看也能快速定位边和节点的关系。做完整整一圈项目之后我自己最大的体会是LangGraph 值得花时间学不是因为它比 LangChain 多了几个花哨 API而是它逼着你用状态机而不是脚本的视角去看 AI 工作流。以前写 Agent 像是在堆 if-else现在更像是在画一张流程图——哪些步骤可以并行、哪些节点需要人工介入、哪些状态要持久化在设计阶段就想清楚而不是等事故发生了再补。最后再分享一个小技巧新项目上手时别急着写复杂图先把你最核心的业务路径画成一张最简图三五节点跑通之后再逐步加分支、加 HITL、加并行。图的架构和代码一样一开始设计得好后面加功能是越来越顺的一开始贪多后面每次改动都像在解环。这篇先聊到这儿下一篇我准备写 LangGraph 与 Web 服务集成时的改造实录包括长任务异步处理、状态通知这些部署中躲不开的硬骨头。
