写这个系列到第7篇我明显感觉关注点变了从怎么写一个节点变成了整个工作流怎么组织才不会崩。用LangGraph做AI工作流编排玩到后面拼的根本不是提示词而是状态管理、流程控制、人工介入和生产稳定性这些偏架构的功夫。这篇我把最近在真实项目里反复折腾过的几个点整理了一遍LangGraph和LangChain的关系到底怎么理解、State怎么设计才不膨胀、分类路由怎么把输出约束成固定几种类型、Human-in-the-Loop怎么落地以及让Agent自己拆任务的Planning模式。内容不太适合完全零基础的读者但如果你正在搭一条超过五六个节点的业务流或者纠结要不要把LangChain项目迁到LangGraph上这篇应该对你有用。1. LangGraph和LangChain的关系辨析工具箱与流水线设计图langchain和langgraph的区别这个热搜词几乎每次写LangGraph都会出现。老实说这个混淆太正常了因为LangGraph本身就是LangChain团队出的名字又像文档里还经常互相引用。但这两个东西根本不在一个层面理解清楚之后你会发现很多要不要迁移的纠结会自动消失。1.1 组件库负责单元能力编排框架负责状态流转LangChain的核心价值是提供了大量好用的组件DocumentLoader、TextSplitter、各类VectorStore封装、Retriever、Prompt模板、ChatModel封装、OutputParser以及把这些组件用LCEL铰链起来的链式调用。你可以在完全不引入LangGraph的情况下用prompt | llm | output_parser这种写法构建一条固定顺序的调用链这就是LangChain最擅长干的事。LangGraph解决的则是复杂流程的状态流转。它关注的是节点之间怎么跳转、分支怎么走、循环怎么退、数据在节点之间怎么传递、执行到一半断了怎么恢复。你可以把LangGraph理解成流水线的设计图把LangChain理解成工具箱。两者不在一个维度更谈不上谁替代谁。维度LangChainLangGraph核心定位组件库与链式调用有状态的可视化编排框架控制流线性LCEL为主分支、循环、并行、中断状态管理无内建状态持久化State Checkpointer适用场景固定链路的原子能力组装复杂业务流与Agent系统我在实际项目里LangGraph的节点内部经常直接使用LangChain的Retriever和ChatModel但也遇到过有同事在节点里用OpenAI SDK裸写的情况一样跑得很好。LangGraph并不绑定LangChain它是一种通用的图执行框架只是和LangChain组件配合得最顺滑。1.2 什么样的流程才值得上LangGraph很多人会问我的项目就是用户提问-召回-生成回答三步有必要上LangGraph吗我的答案是没必要。三步纯线性调用LCEL或者直接写几十行代码就搞定了引入LangGraph反而增加了状态序列化和checkpoint的开销。但如果你发现流程开始出现以下信号就该认真考虑同一段逻辑要在一次请求里执行多次也就是循环后续步骤要依赖前面多种结果做条件分支判断有人工审核、人工补录、审批确认的需求希望具备断点续跑和时间回溯能力多个并行子任务需要汇总和冲突处理。一个典型例子是客服工单系统拿到工单先做意图分类如果是查订单走自动查询如果是申请退款则进入人工审核节点审核不通过要退回重新生成话术。这种流程用LCEL写会非常别扭因为每一步都在围着状态绕圈。用LangGraph的条件边和中断机制整个过程是显式的图一画出来谁都能看懂。我现在做AI工作流编排几乎一半的时间是在画状态跳转图而不是写prompt这个比例希望大家有心理准备。2. State设计是工作流的命门字段合并、分层结构与上下文膨胀LangGraph里面最容易被低估的概念就是State。很多人一开始把它当成一个简单的字典随手塞点数据进去结果节点一多就乱套。实际上State就是整个图的唯一事实来源它的设计好坏直接决定了工作流能撑到多大规模。2.1 默认覆盖写与Reducer机制所有节点共享同一个State对象每个节点接收当前State返回一个部分更新框架负责把这些更新合并回全局State。这里有个初学者特别容易踩的坑默认情况下State的更新是覆盖操作。假设两个节点都返回{user_name: ...}后执行的会覆盖先执行的而不是把它们拼起来。如果确实需要合并呢LangGraph提供了字段级Reducer机制from typing import Annotated, TypedDict from operator import add from langgraph.graph.message import add_messages class WorkflowState(TypedDict): messages: Annotated[list, add_messages] total_tokens: Annotated[int, add] status: strmessages字段用add_messages会把新消息追加到列表同时按消息ID去重total_tokens用内置add做累加status不配置Reducer走默认覆盖。这个差异看似简单但在分支合并或并行节点执行时选错策略就是数据互相覆盖的事故。还可以自定义Reducer。比如多个并行分支节点都返回一个置信度分数最后要取最大值来决定路由就可以写def merge_max(a: float, b: float) - float: return max(a, b)这种自定义合并策略特别适合多个候选方案选最优的场景。团队里我一般会定一条规矩所有具有累积语义的字段日志、消息、计数、打分必须配Reducer不允许靠碰巧只有一个节点写它来维持正确性。2.2 分层State设计给上下文留条活路State设计最经典的问题就是一股脑全塞进去。我见过有人把每个节点的中间结果、大段markdown、甚至图片base64都塞进State图是跑通了但每次checkpoint序列化都慢得惊人LLM上下文也被撑爆。我现在的做法是给State分成三层输入层原始用户请求、业务参数中间计算层各节点的临时结果、分类置信度、检索结果摘要、工具调用记录输出层最终回复、需要展示给用户的数据、审计日志。中间计算层最关键的是控制体积。检索出的文本切片建议先做截断再放进State不要直接塞原始全文需要大模型生成的中间长文档只存摘要或引用ID就好。一个实用的经验是大模型上下文里只放与当前任务强相关的消息历史通常最近几轮就够全量历史放在State里作为审计追踪但不要无脑都塞给LLM。另外一定要给State写类型注解。TypedDict配合IDE的自动补全和静态检查在节点越来越多的时候能省掉大量低级错误。类型清晰了每个节点的入参出参才不会糊成一锅粥。这里我推荐在项目一开始就坚持后面补类型真的是地狱难度。3. 分类路由节点把输出约束成固定几种类型并读出置信度AI工作流里最常用到的第一个节点往往是分类。热搜词里有一句修改大模型架构做分类,输出限制只有这几种类型的概率这个需求我太熟悉了。但这里要澄清一个容易误解的点大多数业务场景根本不需要真的去改模型结构替换分类头那是微调阶段才可能考虑的事情。在LangGraph工作流里做分类路由核心是在推理期把输出约束在一个枚举集合内拿到可解析的类别结果和置信度。3.1 推理期结构化约束而不是改模型具体做法分四步prompt中明确写出枚举值以及每个枚举的说明用pydantic定义输出结构枚举字段用Literal限定调用with_structured_output绑定结构温度设为0减少随机性。from typing import Literal from pydantic import BaseModel class IntentResult(BaseModel): intent: Literal[query_order, apply_refund, transfer_human, other] confidence: float classifier llm.with_structured_output(IntentResult) def classify_node(state): result classifier.invoke( 请判断这条用户输入的意图只允许返回枚举值 query_order(查订单), apply_refund(申请退款), transfer_human(转人工), other(其他)。 f\n用户输入{state[input_text]} ) return {intent: result.intent, confidence: result.confidence}注意with_structured_output具体走哪种底层实现取决于模型支持function calling还是JSON modemethod参数需要按实际模型调整。而confidence这个值是模型在文本层面自评的并非softmax后的真实概率。如果模型提供了logprobs你可以进一步读取各类别token的logprob当作概率参考但多数情况下结构化输出给的confidence配合规则阈值已经够用。3.2 低置信度降级与条件边路由拿到的intent和confidence怎么用在LangGraph里通过条件边决定下一步走向def route_by_intent(state): if state[confidence] 0.6: return transfer_human if state[intent] query_order: return query_order_node if state[intent] apply_refund: return refund_approval_node return fallback_node workflow.add_conditional_edges(classify_node, route_by_intent)低置信度不硬走直接转人工或澄清追问这是工作流稳定运行的关键。我在实际项目里统计过加了置信度阈值之后错误路由率降了差不多一半代价是转人工比例高了一点点但整体体感反而更好。宁可让人多看一眼也不能让错误分支一路跑到黑。分类路由的常见坑还有一个枚举类别太多。实测下来类别一旦超过七到十个模型就开始互相混淆。如果业务类型真的很多就做两层分类先粗分四五个大类再在子节点里细分。给每个枚举附上一句简短说明也很有帮助这比把类别定义写得又长又绕效果好得多。另一个小技巧是other之类的兜底类永远放在枚举最后很多模型对第一个和最后一个枚举值的概率分配有微妙偏好把最常见的意图放前面通常准确率更高。4. Human-in-the-Loop人工介入节点设计不只是加一个确认框AI工作流做到全自动之后下一个需求必然是有些环节必须让人看一眼——AI生成的对外文案要审核、高风险操作要审批、缺少某个必要参数要人工补录。这就是Human-in-the-Loop的典型场景。没有编排框架的时候只能拆成多个服务靠数据库状态字段轮询或者写一堆回调接口非常痛苦。LangGraph原生支持interrupt机制图执行到某个节点时主动停住把控制权交还给外部程序人工处理完再恢复这个图继续往下跑。这个真暂停和外部系统打标记的体验差别很大。4.1 用interrupt实现流程内暂停与恢复在节点里这样写from langgraph.types import interrupt, Command def approval_node(state): result interrupt({ stage: human_review, draft: state[draft_reply], order_id: state[order_id], }) return {final_reply: result}外部程序拿到interrupt抛出的信息把人审结果传回来图继续执行graph.invoke( Command(resume同意措辞再温和一点), config{configurable: {thread_id: order_1024_001}} )这里有个很多人忽略的前提interrupt必须配合checkpointer才能工作。断点信息、当前节点位置、State快照全都存在checkpointer里没有它恢复根本无从谈起。本地调试可以用MemorySaver生产环境至少换SqliteSaver或PostgresSaver。我自己上线时直接用的PostgresSaver好处是状态持久化在数据库里服务重启、多实例部署都不影响断点恢复。4.2 生产落地的人工介入坑副作用、超时与安全人工介入看着简单落地全是细节。第一个坑是副作用节点重复执行。假设图中某个节点已经调用过发送邮件API之后才进入人工审核节点而审核不通过又退回重跑前面的步骤重新执行那一段时邮件可能又被发了一遍。解决方案是把有副作用的操作设计成预检查确认提交两步或者让外部接口天然支持幂等。写工作流的时候我习惯给这类节点加一个执行标记字段比如email_sent: bool重跑前先判断。第二个坑是人工响应超时。人工可能十分钟、一小时甚至两天后才处理图一直挂着不是个办法。我的做法是给thread配置外部超时机制超过时限自动走默认批准或默认驳回分支或者转给下一级负责人。具体走哪条路要按业务风险定但必须有兜底。第三个坑是安全。人工审核接口一定不能裸奔要有鉴权展示给审核人的State信息要脱敏敏感字段一律不传给人审页面。这些属于技术常识但在很多demo项目里确实没有。另外人工介入的每一步操作最好留审计日志谁审的、改了什么、什么时候改的全都要能回溯这在复杂业务流里不是可选项而是底线。5. Planning模式让工作流自己拆任务再分流执行单Agent用ReAct模式跑简单问题没问题但任务一复杂就暴露两个毛病走一步看一步缺少全局规划经常绕远路过程太长还会超出上下文限制。所以现在复杂任务普遍采用Plan-and-Execute的思路先规划再执行最后汇总。这个模式和LangGraph的图结构天然匹配。5.1 Plan-and-Execute的节点拆解实现层面对应三个角色Planner节点把用户请求拆成有序、可执行的步骤ExecuteStep循环节点按步骤逐个调用工具或子图Finalizer节点把各步骤结果整合成最终输出。Planner的prompt是成败关键。我在prompt里强制要求每一步必须绑定一个具体工具或子图并且只产出可执行的步骤否则模型很容易拆出思考市场趋势与用户共情这种无法落地的伪步骤。解析阶段也要做校验对不满足格式的步骤直接打回给Planner重新生成。举个例子一个竞品分析报告生成器Planner拆出三个步骤搜集竞品公开信息、分析定价策略差异、生成对比报告每一步分别绑定一个检索子图和一个分析子图ExecuteStep循环依次执行Finalizer把三个子任务的结果汇总生成最终报告。5.2 子图嵌套与递归限制子图是LangGraph里特别好用的封装手段。如果在多个工作流里都要用到通用检索通用去重通用质量校验这些能力就应该把它们各自构建成子图再作为节点嵌进不同的父图。复杂一点子图内部也可以有自己的Planner和执行循环实现多级规划这在AI Agent系统里已经很常见。用LangGraph搭Planning模式时还有一个必须提前认识的机制递归限制RecursionLimit。LangGraph默认的递归步数上限是25步就是为了防止Agent死循环。我一般保持默认只在业务确实需要长流程时才调大同时要实时监控每条工作流的结束状态。死循环是最可怕的线上事故宁可让流程偶尔没跑完也不能放任一个循环在半夜把token烧光。Planner模式的另一个坑是子任务结果冲突。并行返回的检索信息可能互相矛盾最后生成报告时模型会左右摇摆。我的做法是增加一个专门的一致性裁决节点把冲突点逐个列出用投票或置信度加权选优而不是盲目汇总。这一步对报告类、分析类工作流的最终质量影响非常大。6. 从Demo到生产稳定性优先的几个关键配置Demo能跑起来只是开始。我把这套工作流推到生产环境后发现最有价值的几件事全是一些看不见的配置。这一节整理几个我认为最该提前做的事。6.1 可视化、异常兜底与超时可视化排第一。LangGraph支持直接把图结构画出来app.get_graph().draw_mermaid_png(output_file_pathworkflow.png)或者用官方Studio。节点少的时候觉得多余节点一多、分支一乱这张图就是debug的救命稻草。每次改动架构我都会生成一次图贴在项目文档里对照着走一遍全流程。异常兜底排第二。生产环境里任何一个LLM调用都可能超时、限流、返回格式错误。我的习惯是每个节点内部用try/except包住失败时把错误信息记录到State并返回一个error标记图上设置条件边一旦发现error就转入降级分支。降级分支可以是直接转人工也可以是返回固定话术原则只有一个不能让异常变成未捕获的crashed状态。超时控制排第三。每一条LLM调用必须显式设置timeout工具调用也要设。工作流层面的防守是RecursionLimit和节点级超时。线上出现过一次低级事故某一个外部API无响应整个图卡在那里进度条就停在95%。后来给所有外部调用都加了timeout再也没犯过。6.2 黄金数据集回归与运行日志最后我想强烈建议的是黄金数据集回归。具体做法是准备几十条覆盖各分支的历史请求每改一次图结构、每调一次prompt就批量跑一遍自动检查几条硬指标分类路由是否符合预期、人工介入节点是否在需要的时候触发、关键节点耗时是否异常、最终回复格式是否为合法JSON等。这套回归不复杂但能让你改得放心不用提心吊胆怕顺手改坏了路由。运行日志是另一个容易被忽略的东西。LangGraph的State天然就是一个快照流我会在每个节点完成时把State的diff、token用量、节点耗时、异常信息结构化成JSON落盘。排查问题的时候直接从日志里回放整条工作流比对哪一步的State和预期不一致比盯着模型输出猜高效得多。还有一个和成本相关的小经验分类、路由这类高重复节点输入经常相同值得加一层Redis缓存。同一个意图判断反复调用LLM既慢又费钱缓存命中后直接拿结果走条件边整条链路会轻快不少。我做下来分类节点的缓存命中率能到三成以上积少成多还是很可观的。多说一句。我做完这套LangGraph工作流之后最深的体会是这个框架真正的门槛不在API怎么调而在你有没有把状态当回事。State设计、人工介入、异常兜底、回归测试这些都不在官方教程的首页上但它们决定了你的工作流是玩具还是生产系统。如果你们团队也正在用LangGraph编排AI工作流欢迎把各自的踩坑清单交换一下这个领域每天都在长出新坑。
