1. 为什么我要认真聊聊 LangGraph 这个“状态图式”框架

第一次看到 LangGraph 这个名字，很多人会下意识把它和 LangChain 混为一谈，觉得不过是又一个封装库。我最初也是这么想的，直到在一个多轮对话项目里被链式调用的状态管理折磨得够呛——上下文丢失、分支逻辑写成一坨 if-else、出错之后根本不知道卡在哪一步——才回头认真研究了它。LangGraph 解决的核心问题其实很朴素：当你的 Agent 不再是“一问一答”的直线流程，而是需要循环、分支、回退、人工介入的时候，传统的链式编排就不够用了。它把整个 Agent 的执行过程抽象成一张“状态图”，节点是处理步骤，边是流转方向，状态在节点之间传递和累积。这个思路听起来简单，但真正用起来，你会发现它把很多原本藏在代码深处的控制流显式化了。

这篇文章适合谁？如果你已经能写简单的 LLM 调用、对 LangChain 有过一点接触，但一遇到“让 Agent 自己决定下一步做什么”“多轮对话里保持上下文”“出错后重试某个环节”这类需求就犯怵，那这篇就是写给你的。我会从概念讲到代码，把状态图到底怎么建、状态怎么设计、条件边怎么写、循环怎么控制这些关键点掰开揉碎，配上能直接跑的最小示例。全程用从业者之间聊天的口吻，不堆术语，遇到绕的地方就用生活化的类比讲清楚。你不需要是算法专家，只要会写 Python、调过 API，就能跟着走下来。

我踩过的坑会重点标出来，比如状态字段设计不当导致的内存泄漏、条件边返回值写错导致的死循环、以及 checkpoint 配置不当引发的重复执行。这些在官方文档里往往一笔带过，但实际项目里能让你调半天。文章结构上，先讲清楚“状态图”这个心智模型，再拆核心概念，然后上手写代码，最后集中处理常见问题和排查技巧。你可以按顺序读，也可以直接跳到实操部分抄作业，但我建议至少把第二节的概念看完，不然后面代码里的StateGraph、add_conditional_edges这些会看得云里雾里。

2. 先把心智模型建起来：状态图到底在解决什么问题

2.1 从“链”到“图”，差的不只是画法

传统链式编排（比如 LangChain 的Chain）本质是一条直线：输入进 A，A 的输出进 B，B 的输出进 C，结束。这种结构适合“固定流程”的任务，比如“翻译→润色→格式化”。但 Agent 的真实需求往往不是直线。举个我实际遇到的场景：一个客服 Agent，用户提问后要先判断意图，是查订单还是问售后；查订单需要调用数据库，如果查不到还要追问用户确认信息；问售后需要走另一套流程，中间可能还要人工审核。这里面有分支（按意图走不同路径）、有循环（追问直到信息齐全）、有中断（人工审核后继续）。你用链式编排硬写，代码会变成嵌套的 if-else 地狱，而且每一步的状态（用户说了什么、查到了什么、还缺什么）都得手动传来传去，极易出错。

状态图把这个过程显式化了。你可以把它想象成一张地铁线路图：节点是站点（处理步骤），边是轨道（流转方向），状态是乘客随身携带的行李（在站点之间传递的数据）。乘客从起点站上车，每到一站，站台工作人员（节点函数）会打开行李看看里面有什么，往里面加点东西（比如查到的订单信息），然后根据行李内容决定下一站去哪（条件边）。如果行李里缺关键信息，就绕回上一站再问一遍（循环）。这个模型的好处是：控制流和数据流都看得见、改得动。你想加一个分支，就加一条边；你想让某一步可以重试，就加一条指回自己的边；你想在中间插一个人工确认，就加一个中断点。不用再在代码里挖地雷。

2.2 状态（State）是整个图的心脏

在 LangGraph 里，状态是一个贯穿始终的数据结构，通常用TypedDict或 Pydantic 模型定义。所有节点函数都接收当前状态、返回状态更新。这里有个关键设计：状态不是被“修改”，而是被“合并”。每个节点返回的是一个“增量”，框架负责把它合并进全局状态。这个设计的好处是节点之间解耦——A 节点不需要知道 B 节点会往状态里加什么字段，它只管返回自己那部分。但这也带来一个坑：如果你在状态里放了一个列表，多个节点都往里面 append，默认行为是覆盖而不是追加，除非你显式声明合并策略（比如用Annotated加operator.add）。我一开始没注意这点，结果多轮对话的历史消息总是只剩最后一条，排查了半天才发现是合并策略的问题。

状态的字段设计直接决定了图的复杂度。字段太少，节点之间传递信息不够，你得反复调用外部服务；字段太多，状态臃肿，序列化和 checkpoint 都变慢，还容易出并发问题。我的经验是：只放“跨节点需要共享”的数据。比如用户原始输入、当前意图、已收集的关键参数、中间结果、错误信息。那些只在单个节点内部用的临时变量，别往状态里塞。另外，状态字段尽量用不可变类型或者明确合并策略，避免多个分支同时写同一个字段导致覆盖。

2.3 节点、边、条件边：三个必须分清的角色

节点（Node）就是一个普通函数，签名通常是def node_name(state: State) -> dict。它接收完整状态，返回一个字典表示要更新的字段。节点里可以做任何事：调 LLM、查数据库、跑工具、甚至什么都不做只做判断。节点应该是幂等的——同样的输入状态，多次执行结果应该一致（除非你故意设计成有副作用，比如发邮件）。这个原则在重试和 checkpoint 恢复时特别重要。

边（Edge）分两种。普通边是“无条件跳转”：A 执行完直接去 B。条件边（Conditional Edge）是“根据状态决定去哪”：A 执行完，调用一个路由函数，路由函数看状态里的某个字段，返回下一个节点的名字。路由函数本身不修改状态，只做判断。这里最常见的错误是路由函数返回值写错——比如返回了节点名但拼写不对，或者返回了None，框架会直接报错或者静默走默认路径。我建议路由函数的返回值用常量或者枚举，别手写字符串。

还有一个容易忽略的点：图的入口和出口。入口用set_entry_point指定，出口是END常量。如果你的图有多个可能的终点（比如“成功结束”和“失败结束”），可以都指向END，也可以在END之前加一个汇总节点做收尾。循环的终止条件一定要在条件边里明确写出来，否则就是无限循环。我见过有人把“判断是否完成”的逻辑写在节点里，节点返回状态更新但边还是无条件跳回，结果图跑飞了。记住：循环靠边控制，不靠节点内部 return。

3. 核心概念拆解：StateGraph、Checkpoint、中断与恢复

3.1 StateGraph 的构建流程：从定义到编译

用 LangGraph 建图的标准流程分四步：定义状态 → 创建StateGraph实例 → 添加节点和边 → 编译。编译这一步很关键，它会做静态检查（比如有没有孤立节点、入口是否可达），并返回一个可执行对象。编译后的图可以invoke（同步）或ainvoke（异步），传入初始状态，返回最终状态。

定义状态时，我习惯用TypedDict加Annotated来声明合并策略。比如消息历史用Annotated[list, operator.add]，这样每个节点返回新消息列表时是追加而不是覆盖。其他标量字段（比如intent、retry_count）直接声明类型即可，默认是覆盖语义。这里有个细节：状态的默认值。TypedDict本身不支持默认值，所以初始状态得在invoke时传全，或者在第一个节点里做兜底。我一般会在入口节点里检查关键字段是否存在，不存在就初始化，避免后续节点 KeyError。

添加节点用add_node(name, func)，name 是字符串标识，func 是处理函数。添加边用add_edge(source, target)，条件边用add_conditional_edges(source, router_func, mapping)。mapping是个可选参数，把路由函数的返回值映射到节点名，这样路由函数可以返回更语义化的值（比如"need_retry"映射到"retry_node"），而不是直接返回节点名。这个技巧在节点名比较长或者路由逻辑复杂时很有用，能让代码可读性提升不少。

3.2 Checkpoint：让图可以“存档读档”

Checkpoint 是 LangGraph 里我觉得最实用的功能之一。它把每一步执行后的状态存下来，存到内存、SQLite 或 Postgres 里。好处有三个：断点续跑（程序崩了能从最后一步恢复）、人工介入（在某个节点前暂停，等人确认后再继续）、时间旅行（回退到之前某个状态重新走分支）。配置 checkpoint 需要传入一个checkpointer对象和thread_id。thread_id是会话标识，同一个 thread 的状态会累积，不同 thread 互相隔离。

这里有个坑我踩过：checkpoint 存的是状态快照，不是执行位置。如果你在节点里做了有副作用的操作（比如发了一条消息），恢复后重新执行这个节点，副作用会再来一次。所以有副作用的节点要么设计成幂等，要么在节点开头检查状态里有没有“已执行”标记。另外，checkpoint 的存储后端选择要看场景：开发调试用内存版（MemorySaver）最方便，生产环境用 SQLite 或 Postgres，注意定期清理旧 checkpoint，不然数据库会膨胀。

3.3 中断与人工介入：把“人”放进流程里

LangGraph 支持在节点执行前或执行后中断（interrupt_before/interrupt_after），中断后图暂停，状态存进 checkpoint，等外部输入后再恢复。这个机制在需要人工审核的场景特别有用。比如一个内容生成 Agent，生成完草稿后中断，等人审核通过再继续发布流程。实现上，编译时传入interrupt_before=["publish_node"]，然后invoke会停在publish_node之前，返回当前状态。外部系统拿到状态、人工确认后，用update_state修改状态（比如加一个approved: True），再invoke(None, config)继续执行。

注意：中断恢复时传入的输入是None，因为状态已经从 checkpoint 恢复了，不需要重新传初始状态。这个细节文档里写得不明显，我第一次用的时候传了初始状态，结果状态被覆盖了。另外，中断点的选择要合理——别在纯计算节点前中断，没意义；在需要外部决策或者有副作用的节点前中断才有价值。

4. 上手实操：从零写一个带循环和分支的最小 Agent

4.1 环境准备与依赖安装

先装依赖。LangGraph 本身依赖不多，核心是langgraph包，LLM 调用我用的langchain-openai（你可以换成任何兼容 OpenAI 接口的客户端）。Python 版本建议 3.10 以上，因为用到了TypedDict的一些新特性。

pip install langgraph langchain-openai

如果你要用 SQLite 做 checkpoint，还需要langgraph-checkpoint-sqlite。开发阶段先用内存版，省事。

pip install langgraph-checkpoint-sqlite

环境变量里配好 API key 和 base url（如果你用的是兼容接口）。我习惯用.env文件加python-dotenv加载，避免 key 硬编码。

4.2 定义状态：字段设计与合并策略

我们做一个“信息收集 Agent”：用户输入一个查询，Agent 判断意图，如果是“查天气”就收集城市名，缺城市就追问，收集齐了调用天气工具返回结果；如果是“闲聊”就直接回复。这个场景有分支（意图判断）、有循环（追问城市）、有工具调用，足够覆盖核心概念。

状态定义如下：

from typing import TypedDict, Annotated, Literal import operator class AgentState(TypedDict): messages: Annotated[list, operator.add] # 对话历史，追加合并 intent: Literal["weather", "chat", "unknown"] # 当前意图 city: str | None # 收集到的城市 retry_count: int # 追问次数，防死循环 final_answer: str | None # 最终回复

messages用operator.add声明追加合并，这样每个节点返回新消息时不会覆盖历史。intent和city是覆盖语义，后写的覆盖先写的。retry_count用来限制追问次数，防止用户一直不给城市导致无限循环——这是实际项目里必须加的保险。

4.3 写节点函数：每个节点只做一件事

节点一：意图识别。调 LLM 判断用户想干嘛。

def classify_intent(state: AgentState) -> dict: last_msg = state["messages"][-1]["content"] # 实际项目里这里调 LLM，示例用简单规则模拟 if "天气" in last_msg: return {"intent": "weather"} elif "城市" in last_msg or state.get("intent") == "weather": return {"intent": "weather"} else: return {"intent": "chat"}

节点二：提取城市。如果意图是天气，从消息里抽城市名。

def extract_city(state: AgentState) -> dict: last_msg = state["messages"][-1]["content"] # 简化：假设用户直接说了城市名 known_cities = ["北京", "上海", "广州", "深圳"] for c in known_cities: if c in last_msg: return {"city": c} return {"city": None}

节点三：追问。城市缺失时让用户补充。

def ask_for_city(state: AgentState) -> dict: count = state.get("retry_count", 0) + 1 return { "messages": [{"role": "assistant", "content": "请告诉我你想查哪个城市的天气？"}], "retry_count": count }

节点四：调天气工具。这里用假数据模拟。

def call_weather(state: AgentState) -> dict: city = state["city"] # 实际项目调真实 API fake_weather = {"北京": "晴 25度", "上海": "多云 22度"} result = fake_weather.get(city, "暂无数据") return { "final_answer": f"{city}的天气：{result}", "messages": [{"role": "assistant", "content": f"{city}的天气：{result}"}] }

节点五：闲聊回复。

def chat_reply(state: AgentState) -> dict: return { "final_answer": "你好，有什么可以帮你的？", "messages": [{"role": "assistant", "content": "你好，有什么可以帮你的？"}] }

每个节点都很短，只做一件事。这是刻意的——节点越单一，越容易测试和复用。别把意图识别和城市提取塞一个节点里，不然条件边没法精细控制。

4.4 连边与条件路由：让图跑起来

现在建图。入口是意图识别，然后根据意图分支。

from langgraph.graph import StateGraph, END def route_by_intent(state: AgentState) -> str: if state["intent"] == "weather": return "extract_city" return "chat_reply" def route_after_extract(state: AgentState) -> str: if state.get("city"): return "call_weather" if state.get("retry_count", 0) >= 3: return "chat_reply" # 追问太多次，兜底 return "ask_for_city" builder = StateGraph(AgentState) builder.add_node("classify_intent", classify_intent) builder.add_node("extract_city", extract_city) builder.add_node("ask_for_city", ask_for_city) builder.add_node("call_weather", call_weather) builder.add_node("chat_reply", chat_reply) builder.set_entry_point("classify_intent") builder.add_conditional_edges("classify_intent", route_by_intent) builder.add_conditional_edges("extract_city", route_after_extract) builder.add_edge("ask_for_city", "extract_city") # 追问后回到提取 builder.add_edge("call_weather", END) builder.add_edge("chat_reply", END) graph = builder.compile()

注意ask_for_city的边指回extract_city，形成循环。循环的终止靠route_after_extract里的retry_count判断，以及城市提取成功后走向call_weather。这个设计里，retry_count是防死循环的关键——没有它，用户一直不说城市，图就无限转。

跑一下：

result = graph.invoke({ "messages": [{"role": "user", "content": "我想查天气"}], "intent": "unknown", "city": None, "retry_count": 0, "final_answer": None }) print(result["final_answer"])

第一次跑，意图识别为 weather，提取城市失败，追问，retry_count变 1，回到提取，还是失败……直到第三次追问后兜底走chat_reply。你可以改初始消息带上城市名，验证成功路径。

4.5 加上 Checkpoint 和中断：让流程可暂停可恢复

给图加内存 checkpoint：

from langgraph.checkpoint.memory import MemorySaver memory = MemorySaver() graph_with_memory = builder.compile(checkpointer=memory, interrupt_before=["call_weather"])

interrupt_before=["call_weather"]表示在调天气工具前暂停。执行时传config：

config = {"configurable": {"thread_id": "user-001"}} result = graph_with_memory.invoke(initial_state, config) # 此时停在 call_weather 之前，result 是当前状态 print(result["city"]) # 应该已经提取到城市

人工确认后继续：

graph_with_memory.invoke(None, config) # 传 None 表示从 checkpoint 恢复

这个机制在需要人工审核、或者调外部付费 API 前确认的场景非常实用。注意thread_id要唯一，不同用户用不同 id，否则状态会串。

5. 常见问题与排查技巧实录

5.1 状态合并踩坑：为什么我的列表总是被覆盖

最常见的报错是“消息历史只剩最后一条”。原因就是状态字段没声明合并策略。TypedDict默认是覆盖语义，节点返回{"messages": [new_msg]}会把整个列表替换掉。解决方法是Annotated[list, operator.add]。但注意，operator.add对列表是拼接，如果你返回的是单个消息而不是列表，会报类型错误。所以节点返回时统一用列表包裹。

另一个坑是嵌套字典的合并。operator.add只对顶层列表生效，如果状态里有个dict字段，多个节点往里加 key，默认还是覆盖。这种情况要么把 dict 拆成多个标量字段，要么自定义合并函数。我一般倾向拆字段，简单直接。

5.2 死循环排查：图跑飞了怎么定位

死循环的表现是invoke一直不返回，或者日志疯狂刷同一个节点。排查步骤：第一，看条件边的路由函数，是不是某个分支永远返回同一个节点名；第二，看循环的终止条件有没有被正确更新（比如retry_count是不是真的在递增）；第三，加日志打印每次路由的返回值。LangGraph 支持stream模式，可以逐步看执行过程：

for step in graph.stream(initial_state, config): print(step)

这样能看到每一步进了哪个节点、状态怎么变。我调试循环问题时基本都用这招，比打断点直观。

5.3 Checkpoint 相关：恢复后重复执行怎么办

前面提过，checkpoint 恢复的是状态，不是执行位置。如果节点有副作用（发消息、写数据库），恢复后重新执行会重复。解决办法是在节点开头检查状态里的标记：

def send_email(state: AgentState) -> dict: if state.get("email_sent"): return {} # 已发过，跳过 # 发邮件逻辑 return {"email_sent": True}

另外，interrupt_before和interrupt_after的选择也影响重复执行。interrupt_before是节点执行前暂停，恢复后节点会执行一次；interrupt_after是执行后暂停，恢复后不会重复执行该节点。根据副作用的位置选对中断类型。

5.4 常见问题速查表

5.5 几个我踩过的独家坑

第一个：条件边的 mapping 参数别省。路由函数直接返回节点名虽然能跑，但节点名一改就得全局搜替换。用 mapping 把语义值和节点名解耦，改节点名只改 mapping 一处。

第二个：状态里别放不可序列化的对象。checkpoint 要序列化状态，如果你塞了个数据库连接或者文件句柄，序列化会失败。状态里只放基本类型和可序列化的结构。

第三个：异步节点和同步节点别混用。LangGraph 支持ainvoke，但如果你图里有的节点是async def有的不是，行为可能不符合预期。统一用同步或统一用异步，省心。

第四个：调试时把debug=True打开。编译时传debug=True，框架会打印详细的执行日志，包括每个节点的输入输出状态。开发阶段开着，上线关掉。

6. 从能跑到好用：几个让图更稳的工程习惯

6.1 节点粒度：宁可多拆，不要少拆

我见过有人把整个 Agent 逻辑塞一个节点里，条件边只用来判断“结束还是继续”。这样图是建了，但完全没发挥状态图的价值。节点拆得细，好处是每个节点可单独测试、可单独重试、可单独加中断。比如“调 LLM 生成”和“解析 LLM 输出”应该拆成两个节点——生成可能失败要重试，解析可能格式错要兜底，拆开后各自处理，逻辑清晰。拆的代价是节点多了图变复杂，但这个复杂度是显式的、可控的，比藏在函数内部的 if-else 好维护得多。

6.2 状态设计：只放跨节点共享的数据

前面提过，但值得再强调。状态是全局共享的，放进去的每个字段都要考虑：谁写、谁读、合并策略是什么、序列化成本多少。临时变量、大对象（比如完整的 API 响应原文）尽量别放状态，放外部存储，状态里只存引用或摘要。我一般会在状态定义旁边写注释，标明每个字段的写入节点和读取节点，方便后续维护。

6.3 错误处理：每个节点都要有兜底

节点里调外部服务，失败是常态。我的习惯是每个节点内部 try-except，失败时返回一个带错误标记的状态更新，而不是抛异常让整个图崩掉。然后在条件边里判断错误标记，走重试分支或降级分支。这样图不会因为单点失败而中断，用户体验也好。比如调天气 API 失败，返回{"error": "api_failed"}，路由到“告知用户稍后重试”的节点，而不是直接 500。

6.4 测试策略：单节点测试 + 整图集成测试

单节点测试很简单，构造一个状态字典，调节点函数，断言返回的更新字段。整图测试用invoke跑几条典型路径：正常路径、缺参数路径、重试超限路径。LangGraph 的stream模式可以断言执行顺序，确保条件边走了预期分支。我一般会写一个测试文件，覆盖每个条件边的每个分支，改图之后跑一遍，防止回归。

6.5 性能考量：checkpoint 频率和状态大小

checkpoint 每步都存，状态大的话 IO 开销不小。如果状态里有大列表（比如长对话历史），考虑定期截断或者只存摘要。另外，生产环境用持久化 checkpoint 时，注意清理策略——按thread_id和时间清理旧记录，不然存储会涨得很快。内存版 checkpoint 只适合开发，重启就丢，别用在生产。

7. 最后聊几句实际体会

LangGraph 这个框架，我用了大半年，最大的感受是它把“控制流”这件事从代码细节里拎出来了。以前写 Agent，状态管理和分支逻辑混在一起，改一个分支要动好几处；现在图一画，哪里进哪里出一目了然，加分支就是加条边，加循环就是加条回边，心智负担小很多。但它也不是银弹——如果你的流程就是一条直线，用 Chain 更简单；如果你的 Agent 逻辑简单到不需要状态共享，硬上状态图反而增加复杂度。工具选型要看场景，别为了用而用。

另外，状态图这个思路本身不限于 LangGraph。你完全可以用类似的模型自己实现一套轻量编排，核心就是“状态 + 节点 + 路由”三件套。理解了这套心智模型，换任何框架都能快速上手。我后续打算把 checkpoint 和外部存储结合，做一个支持多轮长会话的 Agent，到时候再分享踩坑记录。如果你也在用状态图做 Agent，欢迎交流，尤其是循环控制和状态合并这两块，坑最多，经验最值钱。