LangGraph实战：从Chain到图模型，构建可控AI工作流

最近几个月我把手头一个检索问答项目从原来堆满 if-else 的控制逻辑逐步重构到了 LangGraph前后差别非常明显。如果你正在用 LangChain 的 Chain 写流程却总觉得“线性的链不够用”那 LangGraph 大概率就是你需要的那块拼图。这篇内容围绕 LangGraph 展开把这个框架核心的东西一次性讲透它到底解决什么问题、关键 API 怎么用、实际踩过哪些坑以及什么情况下值得引入。示例基于 LangGraph 0.2/0.3 系列代码可直接照搬改一改就能跑。1. LangGraph 是什么从 Chain 到 Graph到底改变了什么1.1 LangChain 的 Chain 模式哪里不够用早期 LangChain 最常用的核心抽象是 ChainChain 的本质是“线性管道”数据从输入流向输出中间每步都执行相似的逻辑。比如最简单的 RAG 流程retriever ... # 向量检索器 llm ... # 大模型实例 chain ( {context: retriever, question: lambda x: x[question]} | prompt | llm | StrOutputParser() )这种写法在“检索一次、生成一次”的简单场景下非常顺手。可一旦流程复杂起来你会迅速撞到几个明显的天花板没有循环能力。模型生成答案后如果质量不合格想让它重新思考一次Chain 表达不了“回到上一步再跑一遍”的逻辑。你只能在外部写 while 循环把 Chain 包一层又一层的胶水逻辑。没有条件分支。同一个入口进来有的问题需要使用代码解释器有的问题只需直接回答。这类动态路由需求必须预先拆成多个 Chain再在外面用 if-else 手动调度状态变量还散落在外层。中间状态难以管理。每个步骤产生的中间结果都只是局部变量流程结束就丢了。想跟踪每一步用了哪些工具、查了哪些资料、模型评价如何需要自己造一套日志和内存管理。当时我遇到的实际场景是要给用户一个问答机器人除了直接回答还要在答案质量不高时自动触发“检索优化—再生成”并且在多次循环后强制收敛。用 Chain 加外层 while 也能写但代码可读性越来越差全局变量谁来负责清理重试次数放在哪里多轮对话的记录怎么融入每次改动都像在拔河。1.2 图模型为什么更能表达真实业务流LangGraph 换了一个角度把整个流程建模成一张有向图。图中每个节点是一个处理函数边代表从哪个状态过渡到哪个状态。关键差异在于两点第一图天然支持“环”。边不是只能往前走你可以让 A 节点处理完去 BB 发现结果不对再跳回 A。这种“回路”正好对应模型生成的反思、自我修正、迭代优化等常见需求。第二公共状态被显式建模。图里有一个全局 State 对象所有节点都能读它、更新它。节点 A 写入的信息在节点 B 拿到的是同一份数据不用靠返回值一级一级传递也不需要在函数外面挂一堆全局变量。这个设计解决了实际工程里一大类痛点流程控制和状态管理不再是各管各的而是绑定在一个统一的图模型上。状态在哪里被修改、边在什么条件下触发、循环何时终止这些逻辑都能直接在图定义中看到而不是散落在几十行的循环和判断里。2. 核心概念一次讲清State、Node、Edge、Checkpointer2.1 State整个图的“共享内存”LangGraph 中 State 的定义方式是用 TypedDict或者更常用的是在 TypedDict 上叠加字段注解。它描述的是图在任意时刻的全局数据快照from typing import TypedDict, Annotated class AgentState(TypedDict): question: str # 用户原始问题 context: list[str] # 检索到的上下文片段 answer: str # 模型最终答案 retry_count: int # 已重试次数 final: bool # 是否已到终止条件每个节点都会接收当前 State然后通过返回值告诉图“哪些字段需要更新”。LangGraph 并不是把所有字段都粗暴覆盖而是按字段名做合并。比如某节点返回{answer: xxx}图会把新值写到 State 的answer字段上其他字段保持不变。这里要注意一个细节如果你不声明任何 reducer合并函数默认行为就是“覆盖”。但如果某个字段需要“累加”而不是覆盖比如把多轮工具调用记录追加到一个列表里就可以用Annotated加 reducerfrom langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] # 新消息会追加而不是覆盖add_messages是官方提供的 reducer它会将返回的消息列表原有消息列表合并自动处理消息 ID 和顺序。这个设计很聪明把“如何合并”的语义从节点内部搬到了 State 定义里让结构更加清晰。2.2 Node真正干活的函数节点就是一个普通函数入参是当前 State返回值是待更新字段的字典。比如def retrieve_node(state: AgentState) - dict: docs vector_store.similarity_search(state[question], k3) return {context: [d.page_content for d in docs]} def generate_node(state: AgentState) - dict: answer llm.invoke(f基于以下资料回答问题\n{state[context]}\n问题{state[question]}) return {answer: answer.content}节点不一定要返回东西如果只是执行一个副作用比如写日志、发通知返回None或空字典即可。函数签名越简单越好不要试图在节点内部做大而全的控制初始化连接、读取配置这些事应该放到图外面做。2.3 Edge 与条件分支边控制节点之间的流转。LangGraph 内置了两种边普通边和条件边。普通边就是无条件跳转from langgraph.graph import StateGraph, START, END graph StateGraph(AgentState) graph.add_node(retrieve, retrieve_node) graph.add_node(generate, generate_node) graph.add_edge(START, retrieve) graph.add_edge(retrieve, generate) graph.add_edge(generate, END)条件边则是一个函数根据 State 内容决定跳转到哪个节点def should_continue(state: AgentState) - str: if state[retry_count] 3 and state[answer] : return retry return end graph.add_conditional_edge( generate, should_continue, { retry: retrieve, end: END, } )条件边函数返回的字符串会作为 key 去查映射表命中哪个 value 就跳转到对应节点。这个机制是 LangGraph 最灵活的部分循环、路由、终止判断都能用它表达。from langgraph.checkpoint.memory import InMemorySaver config {configurable: {thread_id: test-1}} app graph.compile(checkpointerInMemorySaver()) app.invoke({question: 今天天气怎么样}, configconfig)Checkpointer 是 LangGraph 的一个重要能力。它保存图的完整状态包括每一步的中间结果只要传入相同的thread_id下次invoke时可以从上次中断的地方继续。InMemorySaver适合本地实验生产环境可以用 SQLite 或 Redis 等持久化方案。实际项目里checkpointer 最常见的用途是实现“人工介入”图在某个节点前暂停等待用户确认确认完成后继续往后跑。这在人工审批、危险操作确认、内容审核等场景下非常实用。LangGraph 的interrupt_before和interrupt_after参数就是为此设计的后面来源操作时我会展开。3. 实操案例从零实现一个带反思能力的智能体3.1 场景设计与流程拆解我用一个具体例子来说明整个构建过程打造一个“会自查的 RAG Agent”。流程大致是用户提问先执行一次检索和生成。生成完成后让一个“评审节点”检查答案是否回答了问题是否引用了检索上下文如果答案不合格则对原问题重写然后重新检索、重新生成最多循环两轮。如果两轮后仍不合格直接用最后一次答案返回避免死循环。这个流程非常适合用 LangGraph 表现因为它同时包含分支合格/不合格和循环最多两轮。我用文字流程表示如下START - retrieve - generate - grade grade --pass-- END grade --fail-- rewrite -- retrieve -- generate -- grade grade --max_attempts-- END3.2 定义 State 与工具函数先定义状态。这里需要 markdown 的回答结果因此 state 里放了问题、上下文、答案、重写后的问题、重试次数等字段from typing import TypedDict class ReflectiveState(TypedDict): question: str rewritten_question: str context: list[str] answer: str retry_count: int grade: str # pass / fail / max再定义两个辅助函数模拟检索和评审def fake_retrieve(question: str) - list[str]: # 实际项目中这里会接向量数据库、搜索引擎或自定义工具 return [f关于 {question} 的参考文档片段 A, f相关片段 B] def grade_answer(question: str, answer: str) - str: # 用大模型判断更合理这里为演示用字符串长度模拟 if len(answer) 20: return fail return pass3.3 设计四个节点和一个条件边下面定义四个节点检索节点、生成节点、评审节点、重写节点。节点函数的返回值就是状态字段的更新。def retrieve_node(state: ReflectiveState) - dict: q state.get(rewritten_question) or state[question] return {context: fake_retrieve(q)} def generate_node(state: ReflectiveState) - dict: prompt f 基于检索到的资料回答问题 资料{state[context]} 问题{state[question]} 请给出完整回答。 # 实际项目里这里调用大模型 answer f回答{state[question]} 相关信息汇总。 return {answer: answer, retry_count: state[retry_count] 1} def grade_node(state: ReflectiveState) - dict: grade grade_answer(state[question], state[answer]) if state[retry_count] 2: grade max return {grade: grade} def rewrite_node(state: ReflectiveState) - dict: # 对问题做改写让后续检索更精准 rewritten state[question] 已优化版 return {rewritten_question: rewritten}需要说明的是这里用字符串长度模拟评审只是演示。真实场景中评审节点应该调用大模型让它输出pass、fail或修正建议本质是一样的一个节点输入状态返回状态更新。然后把这些节点加入图用条件边连接from langgraph.graph import StateGraph, START, END graph StateGraph(ReflectiveState) graph.add_node(retrieve, retrieve_node) graph.add_node(generate, generate_node) graph.add_node(grade, grade_node) graph.add_node(rewrite, rewrite_node) graph.add_edge(START, retrieve) graph.add_edge(retrieve, generate) graph.add_edge(generate, grade) graph.add_conditional_edge( grade, lambda state: state[grade], { pass: END, max: END, fail: rewrite, } ) graph.add_edge(rewrite, retrieve)条件边的写法是add_conditional_edge(source_node, routing_function, mapping)。路由函数返回值pass/fail/max分别映射到结束、结束、重写节点。“fail” 会回到 retrieve 节点完成一次循环。3.4 编译、调用与结果观察图定义完成之后编译成可执行对象app graph.compile() result app.invoke({ question: LangGraph 的核心概念是什么, retry_count: 0, context: [], answer: , grade: fail, rewritten_question: , }) print(result[answer]) print(重试次数, result[retry_count]) print(最终评分, result[grade])invoke会从 START 节点开始按边执行直到遇到 END。这里的result是最后的状态快照包含所有已更新的字段。如果你想观察中间过程不只是一头一尾用stream逐个步骤看for step in app.stream({ question: LangGraph 的核心概念是什么, retry_count: 0, context: [], answer: , grade: fail, rewritten_question: , }): print(step)stream会输出每个节点执行后的增量状态比如{retrieve: {context: [...]}}、{generate: {answer: ..., retry_count: 1}}。这在调试图逻辑时非常有用哪一步没走、哪一步返回值不对一眼就能看到。3.5 加入 Checkpointer让图状态能被保存和恢复如果要从“中途暂停”之后继续执行最简单的做法是加一个 Checkpointer。比如想在图执行到grade之前停下来等待人工确认答案质量可以这样from langgraph.checkpoint.memory import MemorySaver config {configurable: {thread_id: session-20241201}} app graph.compile(checkpointerMemorySaver(), interrupt_before[grade]) # 第一次调用会执行到 grade 之前暂停 first_result app.invoke({...}, configconfig) print(已暂停等待人工确认...) # 人工确认通过后继续执行invoke 需要用相同的 thread_id final_result app.invoke(None, configconfig) print(final_result[answer])这里的核心原理checkpointer 把每一步执行后的状态序列化保存在外部存储里。thread_id是会话标识相同的thread_id对应同一个会话的状态序列。后续invoke传None不代表清空状态而是从保存的检查点继续执行。这个机制用在“人工审批”“用户确认”“异步任务恢复”上非常顺手——状态不会因为进程重启而丢失。如果状态中有大量对象不好直接交给内存保存还可以换成 SQLite 连接器把状态持久化到本地文件系统重启服务后依然能恢复。4. 踩坑记录我实际使用中遇到的常见问题框架虽好坑也不少。这些问题我基本都亲历过写出来帮你少走弯路。4.1 节点返回值不是“整体替换”是“字段级合并”新手最容易犯的错以为节点返回的字典会整体替换 State。实际上 LangGraph 的默认行为是字段级合并——返回字典里出现的 key 才会更新没出现的 key 保持原样。举例return {answer: 新的答案}这会把 State 中的answer覆盖但context、retry_count等原字段不受影响。如果某字段需要累加必须显式用 reducer 声明。字段级合并这个设计其实很友好它让每个节点只需要关心自己负责修改的那几个字段不需要手动搬运整个 State。4.2 条件边返回字符串必须能“精确命中”条件边路由函数返回的字符串必须和映射表的 key 完全一致大小写、空格都不能差。一旦不匹配LangGraph 会直接抛异常而且报错信息在复杂图中不一定立刻指向问题源头。我踩过的一个实际场景路由函数返回retry映射表里写的是retry 或Try结果图跑一半报 “Invalid condition”。解决方式也很简单把路由返回值集中放到常量里别散落多个地方手写字符串。4.3 状态字段被误覆盖成 None节点返回{answer: None}时LangGraph 默认会真的把answer覆盖成None。这在设置了update_as_node之类的自定义 reducer 时尤其容易引起困惑。所以节点里如果某字段没有更新最稳妥的做法是根本不要返回这个 key。返回{retry_count: state[retry_count] 1}就只更新这一个字段不涉及其他 key。4.4 add_messages 的消息 ID 冲突问题用add_messages管理聊天历史时有一个隐藏规则如果两条消息有相同的id新消息会替换旧消息而不是追加。这本来是为了支持消息编辑但如果你手动构造消息时忘了生成唯一 ID列表就会发生“看似少了消息”的诡异现象。from langchain_core.messages import HumanMessage, AIMessage # 错误两条消息都默认为空 id可能互相覆盖 messages [HumanMessage(content你好)] new_messages [HumanMessage(content你好)]正确的做法是尽量让 LangChain 的消息对象自动管理 ID或者显式指定唯一 ID。4.5 状态体积越滚越大State 是一个会贯穿整个图生命周期的对象。如果你的检索上下文是大规模文本片段且每次循环都会重新把结果塞回 State它的体积会持续增加。开启 checkpointer 后每个检查点都会保存整份状态快照存储成本跟着涨。我的建议是State 里只放“流程控制必需的最小数据”。大体积数据完整文档、长图片编码可以存在外部对象存储里State 只放引用 ID。需要原始数据时节点通过 ID 去外部取。4.6 LangGraph 0.1 与 0.2/0.3 的 API 差异LangGraph 版本迭代速度很快API 有一定变化。早期版本里StateGraph的初始化是传一个 dict而 0.2 之后更推荐用TypedDict类。如果在老教程上照抄代码跑不通先检查版本pip show langgraph遇到TypeError: dict object is not callable之类的报错多半是版本不匹配。建议直接安装 0.2 以上版本并参照官方最新文档中的StateGraph(AgentState)风格。下面是常见问题速查方便你在报错时先瞄一眼现象大概率原因处理方式字段没更新节点返回的 key 没写对用stream查看每步增量条件边报错路由返回值不在映射表里统一常量管理返回值消息列表奇怪地缺失add_messages 的 ID 冲突显式分配唯一 ID图执行但无限循环终止条件不满足在路由函数中加最大次数判断状态体积过大State 里塞了大量长文本外部存储 ID 引用5. 实际应用场景与选型建议什么时候该上 LangGraph5.1 典型场景多智能体协作、循环反思、人工审批LangGraph 最擅长的场景有三个方向。方向一是多智能体协作。多个 Agent 分别承担不同职责彼此通过共享状态或消息通信。最典型的是“研究员 写作者”双 Agent一个负责检索资料一个负责汇总成文。两者之间不是简单串联而是可以来回传递信息、相互补充。LangGraph 用共享状态天然支持这种模式每个 Agent 就是一个节点路由逻辑写在边上。方向二是循环反思。模型输出后用一个评审节点判断结果质量不达标就回到前面的节点重新处理。这类流程用 Chain 很难优雅实现而图结构可以自然表达“循环直到满足条件”。比如自动修复代码让模型生成代码随后让另一个节点运行测试测试失败则回到生成节点重新修改。方向三是人工审批。借助 checkpointer 和interrupt_before图可以在关键动作前暂停等人点击确认后再继续。这在发邮件、执行高风险操作、消费金额等场景中非常有用。传统实现要依靠外部状态机或数据库表记录“当前走到哪一步”而 LangGraph 直接把“暂停、恢复”内置化开发成本低了一截。5.2 什么时候不应该用 LangGraph也不是所有项目都应该无脑上 LangGraph。如果流程只有两个顺序步骤比如“检索一次、调用一次模型”那用普通 Chain 就是最直接的选择引入图反而增加抽象层级和学习成本。我的判断标准大致是这样流程中含循环、分支、状态回溯 → 值得用图。流程是纯线性管道 → 用 Chain 或函数式调用即可。需要多轮人工介入状态需要跨请求保存 → 用图 checkpointer。单次无状态请求做简单分类或抽取 → 直接调用模型别加框架。另一个参考维度是团队熟悉度。LangGraph 的思维模型和传统“线性脚本”不太一样如果团队没有时间消化图编排概念仓促引入会让代码审查和调试变得困难。这种情况下先用朴素的 while 循环解决眼前问题也比强行上一个还没掌握的工具更稳。5.3 和同类编排框架的粗略对比做 Agent 编排的框架不少比如高层角色协作类框架、对话式多 Agent 框架等。LangGraph 的特点在于“底层、显式、可控”它不预设角色协作模式所有流程都由你自己用节点和边定义。它对状态的控制最细checkpointer、reducer、流式输出都有明确接口。学习成本相对更高需要你理解图模型中的各种细节。如果你的项目重点是“精确控制每一步做什么”LangGraph 的显式风格会非常顺手。如果你更希望开箱即用的角色分工那种高层框架可能更快但灵活性也会弱一些。我对 LangGraph 的定位是它不是要替代所有 Chain而是把“当 Chain 不够用时该怎么升级”这件事标准化了。现在我的习惯做法是先用最简单的方式跑通逻辑如果发现流程里出现了第二个 if-else或者需要把状态传到循环外部才用 LangGraph 重构成图。这样既避免了过度设计也能让每张图都确实确有其存在的必要。