☰
LangGraph复杂流程编排实战:分支、循环与人工审批设计
2026/9/26 7:52:52 网站建设 项目流程

最开始用 LangGraph 承接公司的内容工单系统,我有一个很直观的体会:真正难写的不是“调用大模型”,而是“让调用大模型的过程可以被控制”。比如同一份用户提问,有的可以直接让 AI 回复,有的必须转人工客服;同一篇稿子,AI 生成的初稿质量不够,需要回头重写;哪怕到了发布的最后一步,也要有负责人拍板确认。这些都是流程问题,不是模型问题。LangGraph 恰好把这个流程建模成一张图,在处理分支、回头、人工介入这三件事上,比以往那种“链式调用”顺手得多。想让聊天机器人、AI 助手、审核系统真正在业务里跑起来,这个框架值得花时间认真搞一遍。下面我把实际项目里的设计思路和踩坑记录一起写出来。

1. 普通 LLM 应用的“链式写法”,到底卡在哪里

1.1 链的本质是“预先铺好的固定轨道”

初学 LangChain 的时候,大家更多是在学习怎么用 Prompt 模板把外部数据填进去,一步步调用模型、解析结果、交给下游。这种写法的隐含假设是:流程从头到尾是线性的、确定的,每一步该做什么都提前写死了。在一个固定检索、固定回答的简单 Demo 里,这完全没有问题,跑起来干净利落。

但真实业务流程不会这么礼貌。用户从同一个入口进来,可能带的是图片链接,也可能带着投诉语气;后台任务既可能执行成功,也可能因为参数问题在第 3 步就失败;业务负责人既要看到 AI 的结果,又不希望它未经确认就发布。于是你会发现,控制流开始变得比单次模型调用更重要,而“链”这种结构对控制流的表达能力是偏弱的。

1.2 三个真实需求,直接把线性链逼到墙角

分支:同一个入口进来,根据意图、权限、用户身份走完全不同的处理路径。比如客服机器人,VIP 客户转人工优先处理,普通客户先自动应答;又比如内容系统的敏感词命中就送审核,不命中才递发布。你用 if 来接,逻辑上不会死,可一旦分支变多,整个主流程里到处都是散落的判断,代码判断会越来越难读。

回头:流程需要回到之前的某个节点重新执行。稿子质量分低于阈值,要回到生成节点重写;工具调用失败,要回到工具选择节点重新选工具;检索结果太分散,还要回到检索节点改查询词。这是“链”结构天然不擅长的,因为链只提供前进方向,回退要么靠递归、要么靠外部循环包裹。

人工介入:流程执行到某个环节,必须停下来等人确认。付款要审批、文章要主编终审、敏感操作要双重确认。人没回应之前,整个任务要挂起,恢复之后还要能接着跑,不是从头再跑一遍。这三个需求加在一起,线性链很快就不够用了。

1.3 也许你会想:在外层写 if-else 不就行了

从短期看,确实行。你可以在调用链的外面包一个大 while 循环,再塞一堆 if 判断。但等到流程复杂起来,需要自己维护的东西至少包括:每一步的执行状态、分支判断的日志、重试次数的计数、人工挂起时的状态快照、恢复之后的上下文拼装。一旦服务重启或者并发多个任务,这些状态还得被序列化进数据库,否则恢复就是一句空话。

这些恰恰是 LangGraph 这类图执行框架要解决的问题:把状态交还给框架,把流程画成图,让运行时的暂停、恢复、回环都发生在可观测的轨道上。下面我直接用一个最小图来拆开讲。

2. 用图来兜住复杂流程:State、Node、Edge 的分工

2.1 三件套对应的具体职责

LangGraph 最核心的抽象只有三个:State、Node、Edge。我用办公场景类比一下:State 是一张正在流转的审批单,上面记录了申请人、审批意见、当前进度等字段;Node 是流程中的岗位,每个岗位拿到单子后做一件事,再写回若干字段;Edge 是单子的流转路线,告诉系统下一个岗位是谁。当条件不同时,Edge 可以指向不同岗位,这就是分支;当路线绕回去,就是循环。

State 通常是一个 TypedDict 或自定义数据类,节点函数的签名是“接收整个 State,返回一个更新字典”。LangGraph 会把返回的字典和当前 State 做合并,所以节点最好只更新自己负责的字段,不要随意覆盖其他人的字段。这意味着节点之间通过 State 通信,而不是通过全局变量或函数返回值传参。

2.2 一个分支场景的最小落地

来看一个朴素的客服分流需求:输入提问后先做意图分类,如果识别到“人工”或用户是 VIP,就转人工;否则用 AI 自动回复。用 LangGraph 写出来大概是:

from typing import TypedDict from langgraph.graph import StateGraph, START, END class SupportState(TypedDict): user_input: str is_vip: bool need_human: bool answer: str def classify(state: SupportState) -> dict: # 模拟意图识别,实际项目里这里往往调 LLM if "人工" in state["user_input"] or state.get("is_vip"): return {"need_human": True} return {"need_human": False} def auto_reply(state: SupportState) -> dict: return {"answer": f"自动回复:{state['user_input']}"} def human_reply(state: SupportState) -> dict: return {"answer": "已转接人工客服,请稍候。"} def route(state: SupportState) -> str: if state.get("need_human"): return "human_reply" return "auto_reply" g = StateGraph(SupportState) g.add_node("classify", classify) g.add_node("auto_reply", auto_reply) g.add_node("human_reply", human_reply) g.add_edge(START, "classify") g.add_conditional_edges( "classify", route, {"auto_reply": "auto_reply", "human_reply": "human_reply"} ) g.add_edge("auto_reply", END) g.add_edge("human_reply", END) app = g.compile()

这里面值得注意的点是:add_node把业务函数注册成图里的节点,add_edge定义无条件跳转,add_conditional_edges则按照路由函数的返回值,从映射表里选择下一个节点。路由函数返回的是一个字符串 key,必须和第三个参数给出的映射对齐,否则框架会直接报错。这也是很多新手第一次跑不起来的原因。

2.3 设计分支时容易忽略的两件事

一是路由函数里不要做重逻辑。分类和打标属于节点责任,路由函数只负责看 State 里已有的字段,并选择下一跳。如果你在路由函数里又调了一次模型,整个图的执行时序会变得混乱,而且不好调试。

二是条件边的映射必须覆盖所有情况。LangGraph 不允许路由函数返回一个没有映射的值,这看起来是个限制,其实是一件好事:分支条件被强制显式声明,而不是散落在一堆 if 里。我在实际项目里,会把所有路由 key 定义成常量,避免手写字符串时拼错。

3. 回头:循环和重试的状态设计

3.1 看似简单的“回头重试”,往往没那么简单

不少人一开始以为循环就是让节点调用自己,或者在外层写一个 for 循环包住整个链。如果你只是机械地重试,第二次调用和第一次调用拿到的是同样的输入、同样的 prompt,结果大概率也一模一样。想让重试有价值,需要把“上一轮为什么失败,为什么不满意”作为下一次执行的上下文。这就是 State 里要维护 feedback、attempt、draft_history 这类字段的原因。

另一个容易忽视的点是:循环不只是看见“坏结果再重跑一遍”,它还可以改变下一步的策略。比如第一次生成稿子太短,第二次的 prompt 里就应该明确写“上一版只有 80 字,太短,请补充案例和数据”;第一次工具调用超时,第二次就应该直接换一个更快的工具。这些策略变化都应该由 State 来携带。

3.2 用一条边把流程绕回去

接下来用一个写作场景演示:写手节点负责生成文章草稿,评审节点负责对质量打分;不合格就回到写手节点改写,合格就往后走。核心代码如下:

from typing import TypedDict from langgraph.graph import StateGraph, START, END class DraftState(TypedDict): topic: str draft: str feedback: str attempt: int max_attempts: int def writer(state: DraftState) -> dict: attempt = state.get("attempt", 0) + 1 prev_feedback = state.get("feedback", "") prompt = f"写一篇关于{state['topic']}的稿子,第{attempt}版" if prev_feedback: prompt += f",参考上一版反馈:{prev_feedback}" return {"draft": f"(生成的第{attempt}版草稿)", "attempt": attempt} def reviewer(state: DraftState) -> dict: # 模拟质量判断,实际项目里这里调 LLM 或规则引擎 if state["attempt"] >= state["max_attempts"]: return {"feedback": "PASS"} if len(state["draft"]) < 10: return {"feedback": "内容太短,请补充细节。"} return {"feedback": "PASS"} def route_after_review(state: DraftState) -> str: if state["feedback"] == "PASS" or state["attempt"] >= state["max_attempts"]: return "accept" return "rewrite" g = StateGraph(DraftState) g.add_node("writer", writer) g.add_node("reviewer", reviewer) g.add_edge(START, "writer") g.add_edge("writer", "reviewer") g.add_conditional_edges( "reviewer", route_after_review, {"rewrite": "writer", "accept": END} ) app = g.compile()

这段代码最值得注意的地方是:循环不是靠递归调用节点函数,而是通过“评审节点 → 条件边 → 写手节点”的边回路实现。图中可以清楚看到哪一步会回头,审计的时候也方便。把add_conditional_edges的目标设成"writer",本质上就是把流程的路径绕回去了。

3.3 防死循环要放在第一版设计里

自纠正型 Agent 最容易犯的错就是忘记设置最大重试次数。模型对“再试一次”的服从度很高,如果没有上限,它会在同一件事上反复打转,最后账单和延迟双双爆炸。我通常会维护state["attempt"],在路由函数里与max_attempts比较,并把“用尽次数后走平庸但可靠的方案”作为一个显式分支。很多正规项目还会把每次 attempt 的产物追加到一个数组里,方便事后复盘,比只保留最后一次结果更能说清楚问题。

4. 让人在关键节点叫停,而不是把“审批逻辑”塞进节点

4.1 人工介入的本质,其实是“暂停和恢复”

有朋友一开始是这么做的:在一个节点内部用input()等人工输入,或者写一个 while 循环轮询数据库看有没有审批结果。这样在本地脚本里可以跑通,但放进异步服务、消息队列、多实例部署之后就很难受:谁暂停的?暂停到哪一步了?恢复时上下文还在吗?

LangGraph 提供了一个更干净的机制:执行器遇到interrupt就会暂停整个图,把控制权交还外部系统;外部系统拿到要展示的信息,等人处理完以后,再带着结果恢复这个图。整个过程依赖 checkpointer 做状态持久化。换句话说,人工介入不是“在代码里等人”,而是“把流程挂在某个节点上,等外部事件来唤醒”。

4.2 代码上看是这样

以下 API 以 0.2.x 版本为参考,不同版本会有细微差异,以官方文档为准:

from typing import TypedDict from langgraph.checkpoint.memory import MemorySaver from langgraph.types import interrupt, Command from langgraph.graph import StateGraph, START, END class ReviewState(TypedDict): draft: str editor_decision: str editor_comment: str def human_review(state: ReviewState) -> dict: # 执行到这里,图会暂停,把下面的 dict 交给外部系统展示 decision = interrupt({"draft": state["draft"]}) return { "editor_decision": decision.get("decision", ""), "editor_comment": decision.get("comment", "") } g = StateGraph(ReviewState) g.add_node("human_review", human_review) g.add_edge(START, "human_review") g.add_edge("human_review", END) app = g.compile(checkpointer=MemorySaver()) config = {"configurable": {"thread_id": "task-001"}} # 第一次调用会触发暂停,不会真正走到 END result = app.invoke({"draft": "这是待审批的初稿"}, config) # 外部系统拿到 result 里的中断信息,展示给审核人 # 审核人点了“通过”之后,用同一个 thread_id 恢复 result = app.invoke( Command(resume={"decision": "approve", "comment": "OK"}), config )

这里面的核心钥匙是thread_id。图执行到interrupt时会把检查点按 thread_id 存起来;恢复时只要带上同一个 thread_id,它就能回到当初暂停的节点,而不是从头再跑一遍。只要能把这个 thread_id 从前端传到后端,再传回图执行器,流程的“等人”和“唤醒”就闭环了。

4.3 审批流落地时,我总结的几个经验

第一,人工节点要尽可能地少。每个审批节点都意味着等待时间、超时处理和用户体验成本,能用规则兜底的先用规则兜底,只在真正需要人判断的位置插入interrupt。

第二,暂停后不要急着清理状态。把 draft、得分、候选意见一起给用户展示,恢复时把用户打回的备注写回 State,否则人工给出的反馈会丢失。

第三,要注意幂等。用户可能连续点了两次“批准”,恢复接口要能处理重复提交,避免同一条流程被发布两次。

第四,检查点要落到外部存储。MemorySaver是内存实现,进程一重启就没了,生产环境至少要换 Postgres 或 Redis 这类持久化检查点。

5. LangChain、CrewAI、扣子与 LangGraph 的关系,别被绕晕

5.1 LangChain 是工具箱,LangGraph 是流水线

既然标题里是 LangGraph,很多人会问它和 LangChain 到底什么关系。最粗的解释:LangChain 提供模型调用、提示词管理、检索、工具封装等一系列零件;LangGraph 提供把这些零件按图组织起来的运行时。你可以只用 LangGraph 而不碰 LangChain 的类,也可以两者一起用。面试时如果被问到,直接说“它们不是竞争关系,LangChain 偏开发组件,LangGraph 偏流程编排,两者共享生态”就算答到点子上了。

5.2 CrewAI 偏向多智能体协作模板,LangGraph 更通用

CrewAI 的设计思路更接近“给你一组 role、task、process 的批量配置”,适合快速搭建多 Agent 协作的固定剧本。LangGraph 则没有限定剧情,你可以画任何有状态的状态机:条件路由、并行、循环、人工审批都由你自己定义。选型时不必非黑即白:需要快速搭建标准多 Agent 合作,CrewAI 的上手速度可能更快;需要细分粒度控制与复杂生命周期,LangGraph 的灵活性更高。

5.3 扣子是不是用 LangGraph 实现的

这是被搜得比较多的问题之一。大家真正关心的其实是:可视化编排和代码编排是不是一回事。就目前公开信息来看,扣子并非由 LangGraph 实现,它的工作流画布用的是自己的底层平台。两者在“用节点和边描述流程”这件事上属于设计理念同源,但实现完全独立。你可以把扣子当成低代码工作台,把 LangGraph 当成可编程的状态图运行时。理念相似不代表同一个骨架,建议两个都学:低代码工具适合快速验证流程,LangGraph 适合把复杂流程做深做透。

5.4 选型速判

业务场景建议选择理由
只是“调模型 → 拿答案”不需要 LangGraph单次调用用普通代码更直接
需要分支、重试、人工审批、恢复LangGraph控制流和状态持久化都是强项
追求低代码可视化扣子这类平台拖拽节点比写代码更快验证
标准多 Agent 协作CrewAI 或 LangGraphCrewAI 上手快,LangGraph 控制更细

在一次完整的工作流里,LangGraph 最值钱的部分不是“把 Agent 拆成节点”,而是给流程提供了显式的“分支、回头和人工介入”能力。如果你没有这三个诉求,暂时不上 LangGraph 也没问题,别为了框架而框架。

6. 完整实例:内容发布的“自动生成—质量回流—人工确认”工作流

6.1 场景规则

为了让前面三部分都串起来,我举一个实际做过的例子:内容平台要发布一篇新闻通稿,流程规则如下:

  1. 运营提交标题和要点,AI 生成初稿。
  2. 质量校验:命中敏感词或质量分低于阈值,进入“自动改写”节点;改写完再次校验,最多改 3 次。
  3. 质量通过后,进入主编审批节点,流程在这里停下来等人操作。
  4. 主编批准则发布;打回则带着意见回到改写节点;拒绝则流程结束。

6.2 图的大致骨架和关键代码

定义一个 State 来装整个流程的全部信息:

from typing import TypedDict class PublishState(TypedDict): title: str bullet_points: list draft: str quality_score: float rewrite_count: int max_rewrites: int editor_decision: str editor_comment: str

节点部分我只演示两个关键节点逻辑。generate负责生成或改写,quality_check负责质量打分,human_review里放interrupt等主编处理:

def generate(state: PublishState) -> dict: rewrite_count = state.get("rewrite_count", 0) + 1 # 实际项目里根据 editor_comment 拼 prompt 再调 LLM return { "draft": f"基于要点生成的第{rewrite_count}版稿子", "rewrite_count": rewrite_count } def quality_check(state: PublishState) -> dict: score = 0.8 if len(state["draft"]) > 20 else 0.4 return {"quality_score": score} def human_review(state: PublishState) -> dict: # 主编在这里看到待审稿和当前版本,图会暂停等待外部恢复 decision = interrupt({"draft": state["draft"]}) return { "editor_decision": decision.get("decision", ""), "editor_comment": decision.get("comment", "") } def route_after_quality(state: PublishState) -> str: if state["quality_score"] < 0.7 and state["rewrite_count"] < state["max_rewrites"]: return "rewrite" return "human_review" def route_after_human(state: PublishState) -> str: if state["editor_decision"] == "approve": return "publish" if state["editor_decision"] == "rework": return "rewrite" return "reject"

图的组装就是把上面这些节点和边连起来:generate → quality_check → human_review → 终态,同时quality_check有一条回generate的边,human_review也有一条回generate的边。整个图有两条循环,分别对应自动返工和人工打回,这在传统链式代码里会写得很绕,但在图模型里就是两条显式的边。

6.3 踩坑记录:我在这个例子里实际遇到过的四个问题

第一,在节点里直接改 State 字段会“改不进去”。节点返回值要和当前 State 合并,如果你只写state["draft"] = "xxx"然后返回{},下一次节点读到的仍然是旧值。必须把所有改动放进返回的 dict。这是新手最容易忽略的一条,也是各种状态不同步问题的源头。

第二,循环改稿时只留下最后一版,没有保留历史。主编打回时往往想看“上一版为什么被打回”,所以 State 里最好维护一个draft_history列表,每次追加。文件、稿子这类版本化状态,尤其要避免用同一个字段不断覆盖。

第三,恢复流程时忘记带 thread_id。人工中断后,如果不带同一个 thread_id 再调用,框架会认为你要开启一条全新流程,而不是恢复原流程。所有前端交互接口都必须把 thread_id 传透,这个 id 在交互链路里就是流程的唯一身份。

第四,测试阶段只用了内存检查点。本地跑通不代表生产能跑通,MemorySaver重启即丢失。部署到多实例或多进程环境,要换成 Postgres 或 Redis 这类持久化检查点,否则“暂停后恢复”在生产环境根本不可用。

这些经验在官方文档的 checkpointing 和 human-in-the-loop 页面里都有,但只有自己跑过一遍,才会真正记住。

7. 社区高频问题速答(含面试用简答案)

7.1 LangGraph 官方文档与学习路径

以官方文档为起点,重点看四个页面:概念、持久化、人工介入、条件边。顺序上我建议:先跑通一个带条件路由的最小例子,再给它加上循环和计数器,最后把内存检查点换成持久化检查点。不要一上手就追求炫酷的多 Agent 编排,把这三步基础吃透,绝大多数业务需求都能套进去。中文社区里“菜鸟教程”类的文章可以参考,但版本迭代快,一定要和官方文档对照着看。

7.2 几个面试官常问的问题参考

LangGraph 和 LangChain 的区别是什么?LangChain 是 LLM 应用开发工具箱,LangGraph 是图状态机运行时,解决的是复杂流程编排、循环、持久化、人工中断等问题。

为什么需要图而不是链?链只能顺序执行,遇到条件路由、循环、人工暂停时需要自己在外部维护状态和跳转逻辑;图把这些控制流声明式地画出来,状态恢复和审计都归运行时管理。

如何防止 Agent 死循环?在 State 里维护重试次数和总预算,在路由函数里用阈值收口,把“超过预算走降级方案”作为显式分支。

人工介入怎么做?用 checkpointer 保存图状态,在需要人类确认的节点调用interrupt,流程暂停并返回待审信息;用户处理完以后通过同一个 thread_id 触发恢复。

节点返回值的约定是什么?节点接收整个 State,返回当前节点要更新的字段字典;框架负责合并,节点只修改自己负责的字段。

这些内容都是我在把自己第一个复杂的客服工单系统改造成 LangGraph 时踩出来的。当时如果我一开始就知道“State 增量合并”和“interrupt + thread_id 恢复”这两件事,至少能少熬两天夜。如果你只是写简单的问答脚本,现在真不急着上 LangGraph;但一旦业务开始要求分支、回环、人工审批,它就是值得认真选型的工具。我的建议是:先在纸上画出状态流转图,再动键盘;先把最简路径跑通,再加一个中断节点;任何时候都别省掉max_attempts这个字段。

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

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

立即咨询