先说一个我自己的真实经历。早先用 LangChain 搭 Agent 做企业内部文档问答时Demo 跑得非常顺畅接一个检索器挂一个 LLM配上 ReAct 循环几乎没费劲就能让模型自己选工具、查资料、给答案。可一旦放进生产环境问题就接二连三地冒出来——它会在一个简单问题上来回打转同一段检索结果重复读好几遍明明工具已经返回了足够信息它却开始凭“记忆”编答案最头疼的是线上跑挂了之后根本说不清它到底执行了哪几步。后来我把架构从 LangChain 的 AgentExecutor 迁到 LangGraph用显式的图结构重写了整套流程才算真正把“可控”这件事抓在了手里。这篇文章不是框架介绍而是我迁移过程中的完整工程笔记。我会先用几个真实症状说明“不可控”到底长什么样再拆解 LangGraph 的核心设计为什么能解决这些问题然后是完整迁移实录、人工审批和成本控制的落地方案最后给出我的取舍建议。内容面向已经用 LangChain 做过一些 Agent 开发、但对生产环境可控性开始有要求的工程师也适合刚接触 LangGraph、想搞清楚它和 LangChain 到底差在哪儿的读者。1. 先别谈框架先把“不可控”的三个症状说透很多团队引入 LangChain 的时候看中的是它“快”和“灵活”。确实几行代码就能让模型调用工具、完成多步推理这对原型验证来说太友好了。但生产环境和笔记本上的 Demo 完全是两码事。我自己踩过、也帮其他团队复盘过不少 Agent 事故总结下来不可控主要体现在三个非常具体的症状上。1.1 症状一Agent 在一个问题上打转token 像开了水龙头这是最常见的失控表现。Agent 拿到一个问题后调了一次搜索工具模型觉得结果不够又调一次回来还是不够于是再调一次……你设置了max_iterations5它就堂堂正正地把 5 次全用完然后给你一句“抱歉我无法回答”。不仅问题没解决一次对话的 token 消耗可能是正常回答的十几倍。我见过一个比较极端的案例一个简单的“报销流程是什么”的问题Agent 连续调用了三次不同的文档检索工具每次结果明明已经覆盖了答案模型却还在“确认”最后触发循环上限强行终止。更麻烦的是这个行为是随机的——同一套代码换个问题就正常换回来又疯跑。你很难用“加一句提示词”来根治因为在 LangChain 的默认循环里模型每一次迭代都会把之前完整的推理过程和工具结果重新塞进上下文迭代次数越多上下文越乱模型越容易丢失最初的意图。1.2 症状二“以为它在调工具其实它在编答案”这是更隐蔽也更危险的一类问题。模型在某一轮循环里拿到了工具返回结果但这个结果被塞进了长长的消息列表里模型“读”的时候注意力可能已经被前排冗长的思考过程分散掉了。于是它跳过了真实结果直接基于自己的训练记忆给出了一段看起来很专业、其实完全脱离工具返回内容的“幻觉答案”。我自己的项目里就遇到过一个检索 Agent 明明查到了 2024 年最新版的公司政策回答里却引用了 2022 年的旧条款而且语气十分笃定。事后追溯消息列表才发现最新版内容确实出现在工具返回结果里只是模型在那一次生成时压根没参考它。这个问题比“打转”更致命因为它不报错、不超时、不触发任何兜底逻辑你会以为一切正常直到业务方拿着错误的答案来找你。1.3 症状三出了故障没法复盘线上无法干预假设 Agent 在线上真的出了问题你想知道它这次执行经历了哪些步骤、哪一步开始偏离、中间调用了哪些工具、每个节点的耗时和 token 是多少。在 LangChain 的 AgentExecutor 模型下你只能从调用的 message 列表里一层层扒历史记录工具调用链混在一起状态覆盖了状态没有一个干净的“执行轨迹”概念。同样难受的是“无法干预”。Agent 已经跑起来了你发现它正在走向错误的方向比如打算把财务数据写入文档、或者准备调用一个高成本的付费接口你没有任何机制在运行中途拦住它。要么眼睁睁看着它跑完要么直接杀掉整个进程然后从头再来。这些在生产环境里都是不可接受的。如果你也遇到过其中任何一个症状就不用怀疑了问题的根子不在提示词、不在模型而在框架的抽象方式把 Agent 设计成“一个不断循环的黑盒”。接下来我们看 LangGraph 是怎么从底层改掉这件事的。2. LangGraph 到底改了什么从“循环黑盒”到“显式状态机”LangGraph 最核心的变化一句话就能讲清楚它不再把 Agent 当成一个函数或循环而是当成一张有状态的计算图。你从“写提示词驱动模型自由发挥”变成了“定义节点和边让模型在其中按照规则执行”。2.1 一个核心心智模型把 Agent 当成一张图而不是一个循环我在给团队做内部分享时喜欢用交通系统来类比。LangChain 的 AgentExecutor 像是让一辆无人车在一个环形路口里反复绕圈每次绕到决策点都由模型决定“是继续绕还是开走”。这个决策本身没有任何约束它想绕几圈完全看当天的状态。而 LangGraph 更像是你直接修了一条有红绿灯和闸机的路你想让它走哪几条街、在哪个路口允许掉头、最多掉头几次、哪里需要人工刷一下卡全部在修路的时候就定好了。这个心智模型非常重要。一旦你开始“修路”你对待 Agent 的态度就会发生根本性变化从“信任它会自己找到路”变成“我来修路让它跑但跑不偏”。可控的东西是你提前规划好的图结构不可控的部分被压缩到节点内部模型的单步生成上。2.2 状态、节点、边LangGraph 的三个基础抽象LangGraph 的编程模型围绕三个东西展开。State状态是贯穿整个图运行的数据对象相当于整个执行过程的“唯一事实来源”。你可以把它理解成一个结构化的“运行记录本”里面可以放问题原文、检索结果、生成答案、当前重试次数、累计 token 等。每个节点都能读取和更新这个状态但更新的方式必须遵循你在 State 里定义好的规则比如用add_messages追加消息而不是粗暴地覆盖整个消息列表。这一点直接解决了 LangChain 里上下文越来越乱的问题——每个节点只认状态里自己关心的字段模型不再需要把全部历史推理过程都背负在上下文中。Node节点就是图里的一个执行步骤本质上是一个函数输入某个状态视图执行你定义好的逻辑调用 LLM、调用检索器、调用外部 API然后把更新后的状态返回出来。和 LangChain 里“模型自己决定下一步调哪个函数”的方式不同LangGraph 里的节点由你显式定义每个节点做什么、不该做什么边界非常清楚。Edge边定义了节点之间的“路”。普通边表示“这条路必走”条件边表示“走到这里要根据状态值做判断然后走某一条分支”。这个设计非常关键——你在边里写的是一个 Python 函数函数读一下当前状态返回下一步要去哪个节点。模型的“自主选择权”被降级成“只能在给定的几个分支里选”而不是“从所有工具里随便挑一个”。2.3 可控性的来源条件边、循环上限、中断点有了上面三个抽象“可控性”就不是一句口号了而是三个非常具体可操作的机制。第一条件边允许你给循环加上限。比如“检索-生成-质量检查”这个循环你可以规定最多循环 3 次每次走完检查节点都判断一下“答案质量达标了就结束没达标并且重试次数还没用尽就回到检索节点”。这个“重试次数”就放在状态里每轮循环加一。模型再也没机会无限打转了因为循环终止条件已经完全被你的代码接管。第二中断点允许你把人的判断嵌入执行流。LangGraph 提供了interrupt机制可以在任意节点运行之前或之后暂停整个图的执行保存当前状态把控制权交给你或者你的业务系统。你判断没问题就调resume继续往下走有问题就直接终止或者改一下状态再继续。这个机制在自动化流程里价值巨大后面我会专门讲。第三全过程的检查点机制让你永远可以加载和执行回放。图执行的每一步都会通过 checkpointer 保存状态快照。这意味着你可以完全复现某次线上执行的过程查到到底是哪一步产生了错误结果也可以随时从某个中间状态“续跑”。这在事故复盘时的价值谁用谁知道。从工程角度看LangGraph 给我的感觉是它没有牺牲 LangChain 的灵活而是把灵活放在了“图结构设计”这个层面让开发者重新拿回了 Agent 流程的掌控权。接下来我拿一个实际迁移案例把从 LangChain 到 LangGraph 的每一步改动完整呈现出来。3. 迁移实录把一个检索问答 Agent 改写成 LangGraph 工作流为了讲得具体我带大家过一遍我实际迁移的一个项目。场景是一个内部文档问答 Agent用户提出自然语言问题Agent 决定是否需要检索内部知识库、检索后基于结果生成答案、然后做一遍质量标准自查不合格就重新检索再生成。整个过程期望是可追踪、可限制、可干预的。我把这个项目叫做“某企业文档问答系统”下面所有代码都来自这个项目只是做了脱敏。3.1 改造前的 LangChain 版本几分钟就能跑通但问题也明显迁移前使用的是 LangChain 的create_react_agent代码大致长这样from langchain.agents import create_react_agent, AgentExecutor from langchain.tools import Tool tools [ Tool( namedoc_search, funcsearch_docs, descriptionSearch internal docs. Use this when you need info. ) ] agent create_react_agent(llm, tools, prompt_template) executor AgentExecutor(agentagent, toolstools, max_iterations5) result executor.invoke({input: 报销流程是什么})这段代码短得让人心动但其内部是一个标准的“模型自我驱动循环”模型决定要不要用doc_search工具返回结果模型再看结果生成下一步直到它自己觉得可以结束了。max_iterations只是一个最大圈数保险丝模型如果执意绕圈你也没办法在中间插手。在实际运行中这个“黑盒”让我很难判断当前走到了哪一步模型是准备调工具还是正在写答案要不要限制它只能搜一次能不能在写答案前插一道质量检查这些问题在 ReAct 模式里都不好解决。所以迁到 LangGraph 后我把流程直接画成了下面这张图用户输入 - 意图判断 - 决定分支 [检索 / 直接回答 / 结束] | 检索节点 - 生成节点 - 质量检查节点 | 条件边 [达标 - 返回 / 不达标 - 重试]这张图一旦跑起来每一步都可以监控而且循环次数和分支逻辑全部掌握在自己手里。下面我就按照图的定义、节点编写、边与路由、运行效果四个部分完整讲一下改写过程。3.2 改造后的图定义状态设计是第一步LangGraph 的第一步是定义State。我建议状态字段不要贪多够用就行但每个字段的更新方式一定要设计清楚。这个项目里我定义了一个QAStateimport operator from typing import TypedDict, Annotated, List from langchain_core.messages import AnyMessage class QAState(TypedDict): question: str # 用户原始问题不变 intent: str # 意图判断结果retrieve / answer_direct retrieval_count: int # 已检索次数用于限制循环 max_retrievals: int # 最多检索次数 documents: Annotated[List[AnyMessage], operator.add] # 追加式累积检索结果 answer: str # 最终答案 quality_score: float # 质量自检得分 passed_quality: bool # 是否通过质量检查几个设计要点question和intent属于“全局只写一次”的字段后续节点只读不写避免被意外改掉。retrieval_count是循环控制计数器每轮检索加 1条件边靠它决定是否继续重试。documents我特意使用了operator.add作为 reducer这样每次检索的新结果会被追加到历史列表而不是覆盖旧结果。这一步非常关键因为它能保留“整个检索过程都看到了什么”复盘时特别有用。max_retrievals的最大值我定为 3超过后即使质量检查不通过也直接强制结束避免死循环。3.3 节点函数的写法把“提示词模板”变成“可审计步骤”LangGraph 的节点就是普通 Python 函数签名固定为(state: QAState) - dict返回值会以 reducer 规则合并进新状态。我把四个核心节点拆开写。意图判断节点先判断问题需不需要检索文档。这不是可有可无的一步——很多问题模型自己就能答没必要白白调一次检索器增加延迟和成本。def intent_node(state: QAState) - dict: prompt f 判断以下问题是否需要检索内部知识库才能回答。 如果问题涉及公司政策、流程、制度等内部文档内容回答 retrieve 如果是通用问题或根据常识即可回答回答 answer_direct。 只输出一个词retrieve 或 answer_direct。 问题{state[question]} response llm.invoke(prompt) intent response.content.strip().lower() return {intent: intent}检索节点根据intent判断结果决定是否执行检索。这里我额外做了一件事每次检索都会先增加retrieval_count再调外部检索接口确保所有分支共用同一个计数器。def retrieval_node(state: QAState) - dict: count state.get(retrieval_count, 0) 1 docs search_docs(state[question], top_k3) # 自定义检索函数 return { retrieval_count: count, documents: docs, # 使用 operator.add 追加不会覆盖历史 }生成节点只有在状态里明确存在检索结果时才允许生成否则拒绝回答。这个“无结果不回答”的硬约束是 LangChain 时代很难实现的——模型可以不管有没有检索结果就凭记忆生成现在我用节点逻辑直接堵死了这条路。def generate_node(state: QAState) - dict: docs state.get(documents, []) if not docs: return {answer: 未获取到有效资料无法生成答案。, passed_quality: False} qa_prompt f 基于以下检索资料回答问题不得使用资料之外的信息。 如果资料不足以回答问题请明确说明“资料不足”。 问题{state[question]} 资料 {docs[-1]} # 只取最新一次检索结果避免上下文过长 回答 answer llm.invoke(qa_prompt).content return {answer: answer}质量检查节点这是 Angent 可控性里最有价值的一个节点。生成答案后不直接返回先让模型自评一轮判断答案是否与检索资料一致、是否有遗漏。def quality_check_node(state: QAState) - dict: check_prompt f 你是质量检查员。根据检索资料判断以下答案是否合格。 检查维度1) 是否使用了资料中的信息2) 是否有遗漏关键点3) 是否包含资料外的编造内容。 合格输出 passed不合格输出 failed并给一个 0 到 1 的分数。 问题{state[question]} 资料{state[documents]} 答案{state[answer]} result llm.invoke(check_prompt).content passed result.strip().lower().startswith(passed) return {passed_quality: passed}四个节点都只做一件事职责单一到可以直接对着状态字段审计。这正是图结构的好处每个函数改起来都无风险不会像以前那样改一处 Loop 逻辑就影响全局。3.4 条件边与回退策略把“模型自由发挥”变成“显式路由”节点只是图里的执行单元真正决定流程走向的是边。LangGraph 里我可以定义普通边和条件边条件边是一个返回目标节点名的 Python 函数。入口路由根据intent_node的结果分流。def route_after_intent(state: QAState) - str: return retrieval if state[intent] retrieve else generate回退路由质量检查后判断是直接结束还是回到检索节点。def route_after_quality(state: QAState) - str: if state[passed_quality]: return end if state[retrieval_count] state[max_retrievals]: return end # 达到循环上限强制结束 return retrieval # 允许重试回到检索节点最后把节点和边拼成图。from langgraph.graph import StateGraph graph StateGraph(QAState) graph.add_node(intent, intent_node) graph.add_node(retrieval, retrieval_node) graph.add_node(generate, generate_node) graph.add_node(quality, quality_check_node) graph.set_entry_point(intent) graph.add_edge(intent, retrieval) # 这里是条件边实际用 add_conditional_edges graph.add_conditional_edges(intent, route_after_intent, {retrieval: retrieval, generate: generate}) graph.add_edge(retrieval, generate) graph.add_edge(generate, quality) graph.add_conditional_edges(quality, route_after_quality, {end: END, retrieval: retrieval}) app graph.compile()这里要特别说明一下add_conditional_edges的第一个参数是起点节点第二个参数是路由函数第三个参数是一个映射表把路由函数的返回值映射到目标节点。END是 LangGraph 内置的特殊标记表示整个流程结束。图编译完成之后每一次调用app.invoke(initial_state)的执行轨迹就完全透明了。我习惯把完整状态打出来看每一步走到哪个节点、带上了什么数据、最后谁终止的一目了然。这个可追踪性是之前 LangChain 的 AgentExecutor 给不了我的。4. 真正让 Agent 可控的工程细节检查点、人工审批与成本护栏图结构重写之后“可控”迈出了第一步。但部署到真实业务里你会发现还不够——你还得解决“中断续跑”“人工审核”“成本预防”这些工程问题。这一章我把 LangGraph 相对 LangChain 最硬核的三个能力拆开讲透。4.1 检查点机制让每一次执行都有状态快照检查点checkpoint是 LangGraph 里一个容易被低估的能力。即使你不主动配置任何东西图在编译时也会维护一个状态的历史。但一旦加上checkpointer它才会真正持久化到存储介质中。from langgraph.checkpoint.memory import MemorySaver checkpointer MemorySaver() app graph.compile(checkpointercheckpointer) # 执行时带上 thread_id让同一线程的多轮消息可以共享状态 config {configurable: {thread_id: conversation-001}} result app.invoke({question: 报销流程是什么}, configconfig)thread_id是一个重要参数。它相当于给一次完整的执行过程起了一个名字。只要 thread_id 相同那么你就可以在不同的调用中共享同一个状态也可以随时随地调用get_state查看当前某个线程的内部分数。我在生产实践中最常用到检查点的场景是事故复盘。线上一次执行突然在生成节点给出了奇怪的答案过去只能对着日志猜现在我可以直接加载那个线程的状态快照看看检索阶段拿回来的文档是什么、质量检查节点的评分是多少马上就能定位是检索源的问题还是模型提示词的走偏。状态快照是结构化的数据不是自然语言的日志检索和分析都方便得多。4.2 中断与恢复把“执行权”还给人和业务系统LangGraph 最有价值的生产功能我认为是interrupt。它允许你主动在一个节点内暂停图的执行流程并把当前状态存下来然后把控制权交回给调用方。这个过程做到了一半会有“人工审批”的效果。举个例子我的文档问答系统在“生成答案后、发送给用户前”增加了一个审批流程。实现方式是定义一个approval_nodefrom langgraph.types import interrupt def approval_node(state: QAState) - dict: # 在节点内部主动挂起等待外部确认 user_decision interrupt({ question: state[question], answer: state[answer], docs_count: len(state[documents]), }) # interrupt 返回的值是外部审批方通过 resume 传回来的 if user_decision.get(approved) is True: return {passed_quality: True} return {passed_quality: False}你可以在图里加入这个节点让流程变成“生成节点 → 人工审批节点 → 质量结束”。当interrupt被触发时invoke会返回一个特殊的结果你把这个结果包括问题、答案、文档数量展示给业务人员看。业务人员看完后用新的配置调用invoke并传入审批结论# 业务人员进行人工审批后继续执行 result app.invoke( {approved: True}, # 或者 {approved: False} configconfig )为什么要活着用resume而不是简单地“杀掉进程”呢因为resume会从暂停的确切位置继续执行所有中间状态都不会丢失。对于生产环境来说这意味着你可以把人工决策嵌入到自动化流程里既不牺牲“无人值守”的效率又保证了关键节点有人的把控。我在内部系统里将这个机制用来审批高危工具调用比如写入操作、对外发送消息等。凡是权利大的工具必经人工审批。4.3 成本护栏限制循环、用量采集和预警LangGraph 把循环次数控制权完全交给了你自己这为成本管理提供了巨大便利。在 LangChain 里你只能设定max_iterations一个粗糙的上限而且该上限是针对整个循环的次数没法细分到具体哪个环节。在 LangGraph 里你可以精细控制每一个回退分支的次数。拿我的系统来举例三个独立上限意图判断最多运行 1 次不设计重试检索最多执行 3 次每次失败才能重试质量检查最多执行 3 次超过上限且仍不达标则直接结束带着未通过状态返回给调用方。这个“分层上限”能精确控制最坏情况下的 token 支出。假设一次问题检索最多 3 次生成最多 3 次质量检查 3 次那么一次对话最坏也就是调用约 10 次 LLM成本完全可估再也不会出现动辄十几次循环的事情。我还习惯在节点之间嵌入一个简单的 token 采集逻辑每次调用 LLM 之后把usage_metadata累加到状态里from langchain_core.callbacks import BaseCallbackHandler class TokenCounter(BaseCallbackHandler): def __init__(self): self.total_tokens 0 def on_llm_end(self, response, **kwargs): self.total_tokens response.llm_output.get(token_usage, {}).get(total_tokens, 0)配合 LangGraph 的检查点每次执行结束后我都能拿到“本次对话各节点分别消耗了多少 token”的数据。这正是所有成本优化场景的起点没有数据你没法告诉老板钱花在哪了也没法针对性优化。5. 迁移之后的实测感受什么时候该迁什么时候劝你别迁我讲完了迁移过程可能已经有不少人摩拳擦掌准备把所有 LangChain 代码都改写了。但我得提醒一句LangGraph 不是银弹它解决的是“可控性”问题而不是“Agent 好不好用”的问题。有些场景用 LangChain 就行有些场景甚至不该用 Agent下面我实话实说。5.1 几个真实对比数据我的文档问答系统迁移前后对比可以作为一个参考。横向对比三个字段首轮调用成功率、单次执行平均 token 消耗、问题可定位率。指标迁移前LangChain迁移后LangGraph异常消耗次数一周约 8 次出现超过 5 轮循环的情况0 次循环上限内不会失控平均 token 浪费约 4000 tokens / 次包含重复检索约 1200 tokens / 次“答案来自检索结果”的比例约 87%100%生成节点强制只读文档线上问题原因定位速度平均 1.5 小时人工翻日志不到 10 分钟直接看状态快照当然这个对比不完全公平因为我在迁移同时重写了提示词和节点逻辑。但它反映了一个趋势当你显式定义流程以后很多性能问题都是结构性地解决的而不是靠碰运气修提示词修出来的。另外迁移开发成本是客观存在的。LangGraph 的图结构开发量明显比 LangChain 的 AgentExecutor 大节点函数、条件边、状态定义、测试覆盖每一项都要花时间。我自己的节奏是简单原型仍然用 LangChain 的 Agent 快速验证一旦流程进入生产、需要稳定性和可观测性就近乎无条件选 LangGraph。5.2 我的取舍建议把合适的问题交给合适的框架如果你是以下情况我劝你别急着迁移只是做一个内部工具、跑一个给团队展示的 Demo流程很简单、失败也没什么损失你的 Agent 只有单次工具调用、没有复杂的多步推理和回退需求你愿意接受“模型自己瞎跑但偶尔出稀奇结果”的现状。以上场景LangChain 的轻量循环完全够用强行上 LangGraph 只会增加开发负担。如果你是以下情况LangGraph 几乎是你绕不开的必选项你需要让 Agent 按照明确的业务流程执行且流程里有多个互斥分支你需要在执行过程中加入人工审批或业务系统介入点你希望每一次运行都留下完整的、可回放的状态审计轨迹你的 Agent 会调用权限较高的工具写数据库、发邮件、触发订单等需要强制审计你需要对 token 成本做精细监控和上限管控。归根结底我是这么理解 LangChain 和 LangGraph 的关系的LangChain 负责的是“让模型表达更丰富”它把模型、提示词、工具黏合在一起快速产出结果LangGraph 负责的是“让流程更可靠”它用图计算的方式把 Agent 的执行过程变成可设计、可审计、可干预的工程产物。好的实践不是二选一而是把 LangChain 生态里的好组件检索器、提示词、工具封装保留下来再套上 LangGraph 的图骨架。我从这次迁移里收获最大的其实不是那几百行代码而是一种思维方式的转变与其费尽心思写提示词让模型“别乱来”不如直接修改工程结构让它“根本没法乱来”。最后给大家一个实操小建议——如果你也准备迁移别一上来就重新设计整个流程先找个你现有系统里最容易失控的 Agent把它一个节点一个节点画出来哪怕一开始就画三个节点起步跑通了再逐渐加分支这样心理负担小踩坑也少。