☰
LangGraph工作流编排实操:状态设计、人工介入与生产稳定性
2026/9/26 2:37:20 网站建设 项目流程

写这个系列到第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: str

messages字段用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 mode,method参数需要按实际模型调整。而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_path="workflow.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工作流,欢迎把各自的踩坑清单交换一下,这个领域每天都在长出新坑。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询