最开始用 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(checkpointerMemorySaver()) 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 场景规则为了让前面三部分都串起来我举一个实际做过的例子内容平台要发布一篇新闻通稿流程规则如下运营提交标题和要点AI 生成初稿。质量校验命中敏感词或质量分低于阈值进入“自动改写”节点改写完再次校验最多改 3 次。质量通过后进入主编审批节点流程在这里停下来等人操作。主编批准则发布打回则带着意见回到改写节点拒绝则流程结束。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这个字段。