第一次听说 LangGraph 的时候,我其实没有太当回事。当时手里已经用 LangChain 写了不少自动化流程,链式调用用得也算顺手,觉得无非又是生态里多了一个新概念。真正开始被它吸引,是某次在做多步骤任务编排的时候,被一个非常基础的问题反复折磨:流程走到一半,怎么让系统根据上一步的结果决定接下来走哪条路?是循环调用工具,还是直接结束?用传统链式写法,代码会越堆越别扭,判断逻辑散落在各个环节,状态也理不清楚。后来认真把 LangGraph 源码和官方文档啃了一遍,才发现这东西解决的不只是“分支判断”这么简单——它等于把一套完整的状态机机制引入了 AI 应用开发,让 Agent 的行为变得可控、可回溯、可持久化。这篇就聊聊我对 LangGraph 的理解和实际使用经验,从核心概念到工程落地,尽量把你需要知道的东西讲透。
1. 从链式调用到状态图:LangGraph 到底解决了什么问题
1.1 链式调用的线性局限
先回忆一下我们最熟悉的方式。用 LangChain 写一个简单的问答流程,通常是这样:加载文档、切块、向量化、检索、扔给大模型生成答案。每一步的输出是下一步的输入,整条链路是固定的、单向的、线性的,代码看起来也足够清晰。
但一旦任务复杂起来,问题就出现了。假设你要做一个能够自主决定“要不要调用联网搜索、要不要查数据库、要不要重写问题”的 Agent,这个时候流程不再是直线了,而是带条件分支、带循环、甚至需要多轮工具调用才能收敛的图结构。你当然可以用 if-else 硬写,但每加一个分支,代码的可读性就下降一截。更麻烦的是状态管理:每一轮执行到哪了、已经调用过哪些工具、当前会话的上下文是什么,这些信息如果不集中管理,很快就会失控。
LangGraph 的核心理念,就是把这些流程显式地建模成一张图。节点是你要执行的动作,边是状态之间的转移条件,状态则贯穿整个图的运行周期。听起来像传统的状态机,但它和 LLM 应用结合在一起之后,能做的东西就远比传统状态机灵活得多。
1.2 用图结构管理 Agent 运行状态
我们可以用一个生活化的类比来理解。做饭的时候,流程通常不是一条直线走到黑的。你要先看看冰箱里有什么菜,再决定做什么菜、需不需要下楼买调料,甚至可能炒到一半发现锅不够用,得调整菜谱。每一步都依赖当前的“厨房状态”:食材列表、炊具情况、时间预算。如果你把所有决策都用 if-else 硬编码写在一条长长的菜谱里,换一个人来就很难维护。
LangGraph 的做法是:把“厨房状态”定义成一个结构化的对象,把每个动作(洗菜、切菜、炒菜、决定下一步)定义成节点,再把节点之间的转换关系定义成边。这样整个烹饪过程就是状态在这些节点之间流动的过程。图结构的好处是天然支持分支、回环、暂停和恢复,而且状态是唯一的、可序列化的,随时可以保存下来,下次接着用。
1.3 LangGraph 的定位与适用人群
LangGraph 并不是 LangChain 的替代品,而是它面向复杂 Agent 场景的补充。如果你只是做简单的文档问答、基础提示词模板,LangChain 足够用;一旦你想实现一个自主规划、多工具调度、可人工介入、需要记忆上下文的智能体,不上图结构管理基本会在某个阶段被复杂度反噬。
适合学习 LangGraph 的读者大概是这几类:已经用过 LangChain、对 Agent 概念有基础认识但苦于流程编排的朋友;准备做生产级智能体应用、需要持久化和断点容错的工程师;以及对状态机模式感兴趣、想看看如何把传统软件工程思路搬进 LLM 应用里的开发者。这篇文章会尽量把基础概念讲到小白也能理解,同时保留足够的深度给有经验的读者。
2. LangGraph 五大核心概念拆解
2.1 State:所有节点共享的一块黑板
理解 LangGraph,第一个要抓住的概念就是 State。它是贯穿整个图运行过程的全局数据结构,所有节点都可以读取它、修改它、并且把修改后的内容传回给图的状态管理器。
在 LangGraph 中,State 通常用 TypedDict 定义。比如:
from typing import TypedDict, Annotated, List import operator class AgentState(TypedDict): question: str messages: Annotated[List[str], operator.add] intermediate_steps: List[dict] final_answer: str这段代码里有几个细节值得注意。question和final_answer是普通的字段,后一个节点写进去会把前一个节点的值覆盖。而messages字段用了Annotated[List[str], operator.add],这是 LangGraph 的约定了:这个字段的合并方式是“把新值追加到旧值后面”,而不是替换。也就是说,多个节点往messages里写内容时,它们会自动累积成列表。
这种设计思路我特别喜欢,因为它把状态更新的语义显式地放在了类型定义层面。你不需要在每个节点里手动把旧消息取出来再拼接,框架帮你把常见的 reducer 逻辑自动化了。
我一开始踩过一个很小的坑:忘了给messages加 reducer,结果多次节点运行后,前面的消息全部被覆盖了。后来才意识到,普通的字段语义是“写入”,带 reducer 的字段语义才是“累积”。如果你希望某个状态字段在多个节点之间共享并持续增长,记得加上合适的 reducer。
2.2 Node:单个业务动作的封装单位
节点就是图里的一个函数。它接收当前完整的 State,执行一段逻辑,然后返回一个字典,字典里是要更新的字段值。一个最基本的节点长这样:
def retrieve_node(state: AgentState) -> dict: question = state["question"] docs = retriever.invoke(question) return {"intermediate_steps": [{"type": "retrieve", "content": docs}]}函数签名很干净:入参是 State,出参是状态更新。好处是每个节点都是独立的、可测试的单元。你可以单独调用一个节点函数,传入一个模拟的 State,检查它返回的更新是否符合预期,不需要把整张图跑起来。
我在实际项目里的习惯是:一个节点只做一件事。比如专门负责改写问题,专门负责调用检索,专门负责生成最终回复。这不只是代码风格问题,而是状态流可观测性的基础。节点越细,出错时就越容易定位是哪一步坏了。很多 LangGraph 示例项目喜欢把几个动作合在一个节点里以图省事,但工程上还是建议拆开。
2.3 Edge:连接节点与条件路由的“水管”
边决定了节点之间的流转方式。LangGraph 里有两种边:普通边和条件边。
普通边表示固定转移,A 结束后一定走 B。条件边则引入了一段判断函数,它接收 State,返回下一个要进入的节点名称。这个函数可以用任意 Python 逻辑实现,也可以调用大模型来做决策,这也是 LangGraph 能表达“Agent 自主决定行为”的关键。
from langgraph.graph import END def route_after_tools(state: AgentState) -> str: if state.get("need_more_info", False): return "retrieve_node" return "generate_node"条件边返回的是字符串,这个字符串必须和图中注册的节点名严格对应。如果你返回了不存在的节点名,运行时会直接报错,这点要注意。我习惯把所有节点名定义成常量,避免手打字符串出错。
2.4 StateGraph:构图、编译与运行的完整流程
把节点和边组装到一起的就是 StateGraph。构建流程分成四步:创建图对象、添加节点、添加边、编译。
from langgraph.graph import StateGraph graph = StateGraph(AgentState) graph.add_node("query_rewrite", rewrite_node) graph.add_node("retrieve_node", retrieve_node) graph.add_node("generate_node", generate_node) graph.set_entry_point("query_rewrite") graph.add_edge("query_rewrite", "retrieve_node") graph.add_conditional_edges( "retrieve_node", route_after_tools, {"retrieve_node": "retrieve_node", "generate_node": "generate_node"} ) graph.add_edge("generate_node", END) compiled_graph = graph.compile()compile 之后得到的对象可以直接调用。编译过程会做校验:所有边的目标节点是否存在、是否设置了入口点、是否可能有无法到达的死角。在开发和调试阶段,会把明显错误前置暴露出来,而不是运行到一半才爆炸。
运行方式也很简单:
result = compiled_graph.invoke({"question": "什么是状态机?"})invoke 传入的就是初始 State 里需要填充的字段,返回值是最终 State。如果想拿中间过程,LangGraph 是支持开启 stream 模式的,后面我也会讲到调试时怎么用这个能力。
2.5 和 LangChain 生态的关系:不是替代,是升级
LangGraph 在底层复用了很多 LangChain 的概念和组件:工具接口、消息抽象、模型封装等。它的节点函数里完全可以自由使用 LangChain 的链、模型调用、工具装饰器。两者更像是同一个生态在两个不同复杂度层次上的呈现。
简单说,LangChain 提供了丰富的积木,LangGraph 则提供了把这些积木灵活组合成“图”的框架。我的建议是不要把它们割裂开看,在实际项目中经常是混着用的。
3. 手把手构建一个可复用的检索问答 Agent
3.1 场景设定与整体架构
这部分我们用一个小而完整的案例来走一遍流程。场景是这样的:构建一个能够围绕内部知识库做问答的 Agent,它具备两个行为:第一,直接基于已有知识回答;第二,如果判断知识不足或需要联网补充,就调用检索工具,再基于检索结果回答。这个 Agent 需要支持多轮对话,并能记录当前用户问题的历史上下文。
整体图结构设计如下:
- 入口节点:检查是否需要检索
- 条件分支:需要则进入检索节点,不需要则进入生成节点
- 检索节点:调用向量检索工具,并把结果写入状态
- 生成节点:整合历史消息和检索结果,调用大模型生成答案
- 结束节点:输出最终答案
这个结构很简单,却是 LangGraph 条件路由和状态管理的最小完整案例,后续无论多复杂的 Agent 都是在此基础上的扩展。
3.2 定义状态与检索工具
先定义状态和工具函数。我用的是伪代码风格,重点在于逻辑结构清晰,你可以直接替换成自己环境里的实现。
from typing import TypedDict, Annotated, List import operator class ChatState(TypedDict): question: str chat_history: Annotated[List[dict], operator.add] retrieved_docs: List[str] needs_search: bool answer: str def check_search_node(state: ChatState) -> dict: last_message = state["chat_history"][-1] if state["chat_history"] else "" needs_search = should_search(state["question"], last_message) return {"needs_search": needs_search} def retrieve_node(state: ChatState) -> dict: docs = knowledge_retriever(state["question"]) return {"retrieved_docs": docs} def generate_node(state: ChatState) -> dict: context = "\n\n".join(state.get("retrieved_docs", [])) if not context: reply = normal_llm_chat(state["question"]) else: reply = rag_llm_chat(state["question"], context) return {"answer": reply}这里should_search可以是一个简单规则,比如检测到“最新”“实时”“内部文档没提到”等信号,也可以用一个小模型做意图判断。在实际生产项目中,我倾向于用后者,准确率高很多。
3.3 编排条件路由循环
接下来把节点组装成图,并加上条件路由。这里有一个易踩的坑:LangGraph 的条件边是我们自己写判断逻辑,判断逻辑不要在多个节点里各处都有,尽量集中在条件函数里,否则状态流会变成一团乱麻。
from langgraph.graph import StateGraph, END def decide_next(state: ChatState) -> str: if not state["needs_search"]: return "generate_node" return "retrieve_node" graph = StateGraph(ChatState) graph.add_node("check_search_node", check_search_node) graph.add_node("retrieve_node", retrieve_node) graph.add_node("generate_node", generate_node) graph.set_entry_point("check_search_node") graph.add_conditional_edges( "check_search_node", decide_next, {"generate_node": "generate_node", "retrieve_node": "retrieve_node"} ) graph.add_edge("retrieve_node", "generate_node") graph.add_edge("generate_node", END) app = graph.compile()这里有一个细节:条件边的映射字典里键是decide_next返回的字符串,值才是真正的节点名。如果你希望返回值和节点名保持一致,可以直接这样写。我自己的习惯是让返回值和节点名解耦,这样以后改节点名时不用动判断函数,安全性高一些。
3.4 编译运行与参数调整
运行这段代码:
result = app.invoke({ "question": "我们的退款政策最近有什么变化?", "chat_history": [], }) print(result["answer"])如果你开启了检索模式,最终答案会综合知识库内容和问题来生成。这里有一个参数上的取舍需要说明:retrieved_docs的条数、chat_history保留的长度,都会直接影响上下文窗口占用和模型成本。我在实际项目中会把检索出的文档数限制在 4 到 6 条,历史消息限制在最近 10 轮,并且按 token 截断。控制上下文长度不是可选项,是生产级应用必须考虑的。
另外一个小技巧:invoke 支持传入config参数,里面可以带recursion_limit,控制图最多执行多少步。这个参数在调试无限循环时特别好用,比如一场意外的循环卡死,如果不设上限,请求会挂很久。默认上限足够,但我在复杂图结构里都会主动设一个更小的值,比如 20,让异常快速暴露。
4. 持久化、Checkpoint 与断点续跑
4.1 记忆从哪来:Checkpoint 机制解读
单纯的图执行是一次性的:invoke 结束,所有状态都丢了。但实际业务里,用户不可能每次都把所有历史重新说一遍。LangGraph 提供的答案是 Checkpoint。
Checkpoint 机制简单说就是:图上每执行到一个节点,框架就把当前完整 State 保存成一份快照。这份快照存在内存或数据库里,关联一个线程 ID。下次要恢复这个会话,只要传入同一个线程 ID,图就能从上次中断处继续跑。
这会带来两个很明显的好处。第一是多轮对话的天然记忆:上一轮的状态在下一轮自动可用。第二是容错:如果某一步执行失败或用户主动终止了流程,后续可以从断点恢复,不用重头开始。这在长任务场景里价值巨大。
4.2 多轮会话的 thread_id 用法
使用 Checkpoint 并不需要大幅改代码,主要是在编译和调用时多传一个配置:
config = {"configurable": {"thread_id": "user_123_session_456"}} result1 = app.invoke({"question": "你们的退货周期是几天?"}, config=config) result2 = app.invoke({"question": "如果超过这个时间怎么办?"}, config=config)第二次调用时,输入的 State 会自动带上第一轮产生的历史消息。你会发现chat_history字段里自动积累了前面的内容,而你的生成节点只需要正常读取即可。这个机制解决了我之前手动维护历史记录的大量样板代码。
使用 thread_id 有个原则需要想清楚:thread_id 是线程的标识,不是用户 ID。一个用户可以有多个线程,比如围绕不同问题开的会话。把用户 ID 和会话 ID 拼起来作为 thread_id 是比较常见的做法。如果只拿用户 ID 当 thread_id,会导致所有历史混在一个会话里,后续状态会越来越乱。
除了内存级 Checkpoint 外,LangGraph 也支持把状态存到外部持久化存储里。这样即使服务重启,历史也能恢复。我建议生产环境一定要配置外部持久化,否则服务一挂,用户的历史对话全没了,体验会很糟糕。
4.3 Human-in-the-loop:让人插一脚的设计
Checkpoint 机制还解锁了一个冷门但重要的能力:人机协同。LangGraph 允许你在图中设置断点,比如在执行某个敏感动作之前暂停,等人工确认后再继续。这非常适合审核、审批、危险操作确认等场景。
实现思路大概是:在编译时把某个节点设置成断点节点,运行时图执行到该节点前会自动暂停,并返回当前待执行步骤的信息。你可以自行做人工审核流程,然后再调接口让图继续跑。
我在实际项目里用这个模式实现过一个生成内容的发布前审核流程:大模型生成初稿,进入审核节点时暂停,运营人员在后台查看、修改或退回,确认后图继续执行发布动作。整个过程用户无感知,但对业务方来说,风险控制能力增强了一个量级。
5. 常见问题与排查技巧实录
5.1 三个高频踩坑场景
场景一:状态字段被覆盖而不是累积
这是最容易踩的坑。如果某个字段需要由多个节点往里追加内容,比如历史消息列表,一定要在定义 State 时给它加上Annotated和 reducer。否则每个节点写入时都会整体覆盖,最终只保留最后一个节点的内容。排查这种问题,可以把最终 State 打印出来,看看字段值是预期的累积列表还是只有单条内容。
场景二:条件路由返回了不存在的节点名
条件边函数返回的内容如果拼错了一个字母,运行时会直接报错 “Node not found”。这种问题在字符串没有常量管理时非常容易发生,而且错误信息只有在实际执行到该条件边时才出现。我强烈建议把所有节点名字定义成常量,并且条件函数返回时直接用常量,不要手写字符串字面量。
场景三:循环节点停不下来
某些业务流程里,为了处理复杂任务,会让图在“检索—生成—工具调用—再检索”之间循环。如果循环终止条件写得太宽松,或依赖的模型判断不稳定,就可能出现死循环。LangGraph 有 recursion_limit,但默认值很高,不会及时触发。排查时先把 recursion_limit 调小一些,比如 20,让循环快速中断,再检查终止条件。另一个思路是在状态里加一个计数器字段,超过阈值强制跳转到最终节点,这是工程上比较稳的兜底方案。
5.2 调试建议与日志技巧
调试 LangGraph 应用有一个明显优势:图结构本身就是很好的调试地图。出错时可以先确认是哪两个节点之间的转换出了问题,再针对节点单独测试。
我在项目里通常会在每个节点的入口打一条结构化日志,记录当前节点名和关键状态字段的摘要;在节点出口再打一条日志,记录返回的更新内容。这样整个执行路径一目了然。不要只记录节点名,还要记录状态里的关键值,否则日志只能告诉你“走到哪了”,不能告诉你“为什么走了这条路”。
LangGraph 还提供了app.stream()方法,可以逐节点流式输出执行过程。调试阶段我会经常用它替代 invoke,能直观看到每一步的状态变化:
for event in app.stream({"question": "..."}, config=config): for node_name, state in event.items(): print(f"节点: {node_name}") print(f"状态摘要: {state}")流式执行在线上也能派上用场,用户侧可以实时看到 Agent 当前在做什么,体验上会“智能感”强很多。但要注意不要直接把完整状态流给用户,里面可能包含检索中间结果、内部判断逻辑,容易暴露实现细节。
5.3 问题速查表
| 现象 | 可能原因 | 排查思路 |
|---|---|---|
| 历史消息不累积 | State 字段没配 reducer | 检查 TypedDict 中的 Annotated 定义 |
| 条件路由报节点不存在 | 返回字符串与节点名不一致 | 用常量管理节点名 |
| 图运行超时 | 循环终止条件过弱 | 调小 recursion_limit,加计数器兜底 |
| 多轮对话串号 | thread_id 设计粒度不对 | 检查是否把用户 ID 直接当 thread_id |
| 某节点状态为 None 引发异常 | 前置节点未返回该字段 | 在节点入口加字段默认值 |
| 修改 State 后下游读不到新值 | 返回的字段名与状态字段不同 | 检查节点返回的 dict 键名 |
这张速查表基本覆盖了早期使用 LangGraph 的高频问题。很多问题形式上五花八门,本质都是对 State 更新语义或节点命名约定理解不到位。
6. 工具选型解析与进阶方向
6.1 什么场景必须上 LangGraph
不是所有项目都需要 LangGraph。判断标准主要看三点:有没有分支决策、有没有循环依赖、需不需要持久化和断点恢复。
如果你的应用只是“接一个模型,给一个答案”,LangChain 或其他轻量封装完全够用。一旦你的 Agent 需要根据中间结果决定下一步、需要调用多个工具且调用的顺序不固定、需要在长时间任务中反复比较和筛选信息,那么图结构的价值就体现出来了。
从维护角度说,图结构让流程可见。一份代码接手过来,只要画出节点和边的连接关系,就能理解业务逻辑。而 if-else 嵌套的链式代码,这种可见性是很差的。团队协作时,这个优势会越来越明显。
6.2 与其他编排方案的对比
市面上做 Agent 流程编排的选项不少,大致有三类思路。一类是在原生 Python 代码里自己维护状态机和循环;一类是用 LangGraph 这类专门的图编排库;另一类是拖拽式流程编排平台。
单说 LangGraph 和自维护代码的区别。自维护代码最大的问题是:状态不集中、恢复逻辑要自己造轮子、异常处理分散。你可能自己写出了一个能用的流程,但等到要加“断点恢复”“人工审批”“历史会话”,就会发现每一块都要手搓。LangGraph 把这些从框架层面解决了,代码量不一定会少,但复杂度可控得多。
与其他图编排库相比,LangGraph 的优势是离 LangChain 生态近、对 AI 场景的抽象更贴切,比如消息、工具、模型调用这些概念都是内置的。局限是它要求你按照它的思维模式去设计状态,对没有状态机经验的开发者来说,有一定的学习成本。
6.3 可以继续扩展的方向
LangGraph 本身的能力边界还在快速扩展。几个方向值得大家保持关注。
一个是多智能体协作模式。单个 Agent 处理不了的任务,可以拆成多个子 Agent 并行或接力完成。LangGraph 的状态机制很自然地支持这种组织结构,每个子图可以作为一个节点挂在更大的图上。
另一个是与外部事件流的集成。结合消息队列和事件触发器,图节点可以由外部事件激活,而不再局限于用户请求驱动。这个方向比较适合构建自动化运维、异常处理这类响应式系统。
还有一个是图的动态生成。目前多数应用的图结构是静态的,开发时写死了。如果能让图结构随着任务变化而动态生成,通用性和灵活度会大幅提升。这也意味着 Agent 的自主规划能力会上一个新台阶,当然工程达成难度也要高不少。
最后再分享一个小技巧。如果你刚开始学 LangGraph,不要一上来就追求“大而全”的复杂图,先拿一个三四个节点的最小流程图跑通,再逐步加入条件路由、持久化、人工介入这些机制。状态机的思维和传统面向过程编程很不一样,需要一段时间适应。等你想清楚“状态是什么、变化如何发生、谁能改变状态”这三件事,LangGraph 就不再是一个需要学的新框架,而是一个顺手的表达工具。