1. 从"串行链条"到"图上编排：LangGraph 到底改变了什么
在聊 LangGraph 之前，我想先说说一个很多人问过我的问题：LangChain 本身不是已经能构建智能体了吗，为什么还需要 LangGraph？换句话说，标题里这六个字——"有状态智能体"，到底指的是什么？
如果你写过几个基于 LangChain 的 Agent，大概率会碰到这类场景：你调用了两次模型，第二次调用时模型完全不记得第一次说了什么。你明明把对话历史传给了大模型，但只要中间穿插了一次工具调用、一次条件分支或者一次人工确认，状态就"断"了。传统 LangChain 的链式抽象擅长处理"线性流程"，却很难优雅地表达"循环""条件分支""临时中断"和"记忆持久化"这些真实智能体必需的控制流。
LangGraph 解决的就是这个问题。它把智能体的运行过程建模成一张有向图——节点是处理逻辑，边是流转规则，图本身承载一个可以被读写、被保存、被恢复的状态容器。你从"链"的思维切换到"图"的思维之后，一切突然变得顺理成章：模型循环调用、工具派发、人工介入检查点、异常回退路径，全都变成了图上普通的一条边。
这篇文章我会从一个实际可运行的项目出发，完整演示如何用 Python 借助 LangGraph 打造一个具备持久记忆、支持中断恢复、带工具调用的有状态智能体。适合的读者是已经跑通过 LangChain 基础链路、想更进一步掌握智能体编排技术的同学；如果你是完全零基础，建议先了解 Python 函数装饰器和 Dict 数据结构，理解成本会低很多。
我最后反复强调一点：LangGraph 不是要替代 LangChain，而是把"控制流"层面缺失的那块拼图补上了。接下来我们直接上手。
2. 动手搭状态机：StateGraph 的核心构件与状态流设计
2.1 项目结构和基础依赖
为了不让你在版本地狱里浪费时间，这里给出我实测可用的依赖组合。我用的是 LangGraph 0.2 系列版本（较新版本 API 有部分调整，但核心概念一致）：
pip install langgraph langchain-core langchain-openai注意一个版本问题：langgraph的StateGraph早期包名是langgraph.graph，在 0.2 版本之后建议统一使用 TypedDict 定义状态；alias或条件边的参数签名在 0.3 中也有变化，如果你在运行时收到TypeError报错，大概率是两个版本混用了。
项目的目录结构也很简单：
stateful_agent/ ├── graph.py # 核心图定义 ├── state.py # 状态类型定义 ├── tools.py # 工具函数 ├── main.py # 交互入口 └── memory.db # SQLite 持久化文件（运行时生成）2.2 状态定义：不只是字典，而是整个智能体的"内存骨架"
LangGraph 的状态核心是一个 TypedDict（或 Pydantic BaseModel）。这个状态容器贯穿全图的所有节点：每个节点函数接收当前状态的快照，处理后返回一个字典（或可选地返回状态切片），LangGraph 内部会负责"合并"新返回的字段到主状态中。
我个人习惯先定义一个状态类型模块state.py：
from typing import Annotated, TypedDict, Literal from langgraph.graph.message import add_messages class AgentState(TypedDict): # 消息列表，add_messages 注解是关键，它决定了新消息如何合并到已有列表 messages: Annotated[list, add_messages] # 自定义业务字段：当前用户需求描述 user_intent: str # 状态机流转标志：用于控制条件判断 stage: Literal["start", "agent", "tool", "confirm", "done"] # 对话轮次计数器 turn_count: int看到Annotated[list, add_messages]可能有些陌生。它的含义是：当某个节点返回新的messages时，不是直接覆盖旧消息列表，而是调用add_messages函数把新旧消息合并成一份完整的消息历史。这对大模型对话场景至关重要——如果不做合并，每轮节点返回都会把之前的对话记录冲掉。
自定义的业务字段如user_intent、stage、turn_count则是普通覆盖式合并：谁最后返回，谁的值就生效。这些字段构成了智能体的"短期工作记忆"，可以用于业务状态判断。
2.3 节点和边的底层逻辑：你会写函数，就会写节点
LangGraph 的节点就是一个普通 Python 函数，接收state字典，返回一个待合并的字典。
def agent_node(state: AgentState) -> dict: # 调用模型生成回复 response = call_llm_with_history(state["messages"]) return {"messages": [response], "stage": "agent"}边则分为两类：直接边和条件边。
- 直接边：
graph.add_edge("agent_node", "tool_node")，表示无条件从 A 到 B。 - 条件边：
graph.add_conditional_edges("agent_node", route_function, {...})，表示根据某个函数返回值动态选择下一个节点。
这就是 LangGraph 有别于 LangChain 的核心之一——它让你用最原生的 Python 表达控制流：
def route_after_agent(state: AgentState) -> str: if state["stage"] == "tool": return "call_tool" elif state["stage"] == "confirm": return "human_confirm" return "finish"节点、边、状态三者的关系，我打个比方：状态是整个团队的共享白板，节点是各个工位的员工，边则是公司规定好的流程路线。员工可以在白板上写东西、改东西；写完之后按路线把活儿传给下一个工位。但凡中间某个工位想临时插一段人工审批（interrupt机制），流程图就能清晰表达这个岔路。
2.4 把图编译成可执行应用
LangGraph 中StateGraph是绘制蓝图，调用compile()之后才获得一个可执行的CompiledStateGraph：
from langgraph.graph import StateGraph, END builder = StateGraph(AgentState) builder.add_node("agent", agent_node) builder.add_node("tools", tool_node) builder.set_entry_point("agent") builder.add_conditional_edges("agent", route_after_agent, { "call_tool": "tools", "human_confirm": "confirm_node", "finish": END, }) builder.add_edge("tools", "agent") # 工具结果回传给 agent app = builder.compile()这段代码几乎是 LangGraph 的"hello world"，但它已经包含了一个智能体最核心的骨架：模型节点、工具节点、回环路径、条件出口。后面所有复杂状态流都是在这个骨架上的扩展。
3. 有状态的关键：Checkpointer 持久化与跨会话记忆
3.1 内存态智能体为什么撑不起真实业务
如果只做到上面这一步，你的智能体仍然是无状态的——每次执行app.invoke()，LangGraph 都会从头开始跑图，之前的对话记录、中间计算结果全部丢失。这在真实业务中几乎不可用：用户说"查一下当前库存"，你回答完，下一条消息"再查下上周销售"就无法关联上下文了。
让状态跨轮次存活，核心依赖的是 LangGraph 的Checkpointer（检查点机制）。它就像给整个图运行过程拍了快照：每次节点执行之后，系统都会把当前状态持久化到一个存储后端。下次调用时，你可以传入一个thread_id来指定"续读"哪一份快照。
3.2 SQLite 持久化的完整实现
LangGraph 自带多种 Checkpointer 后端，我最常用的是 SQLite（单机可靠、零运维、易备份）。下面是一个直接把SqliteSaver集成到应用里的示例：
from langgraph.checkpoint.sqlite import SqliteSaver # 注意：使用 SqliteSaver 时通常用上下文管理器，确保连接生命周期正常 with SqliteSaver.from_conn_string("memory.db") as checkpointer: app = builder.compile(checkpointer=checkpointer) config = {"configurable": {"thread_id": "user-001"}} # 第一轮会话 result1 = app.invoke({ "messages": [{"role": "user", "content": "帮我看看目前有哪些待办事项"}], "user_intent": "query_todo", "stage": "start", "turn_count": 0, }, config) # 第二轮会话，同一个 thread_id，模型能记住上一轮内容 result2 = app.invoke({ "messages": [{"role": "user", "content": "好的，再提醒我明天开会"}], "user_intent": "add_reminder", "stage": "start", "turn_count": 0, }, config)关键就在第二个invoke使用了相同的thread_id。LangGraph 会从 SQLite 里自动加载该线程的历史状态，并把它作为新一轮图的初始状态。result2中的模型自然就能看到第一轮的对话内容——这就实现了跨轮次的记忆。
如果你的场景还需要跨天记忆、甚至跨用户记忆，一个常见做法是thread_id直接使用用户 ID 拼接日期：
config = {"configurable": {"thread_id": f"user-{user_id}-{date.today()}"}}3.3 手动快照与"分叉"恢复：被很多人忽略的高级用法
除了自动打点，LangGraph 还允许你手动获取和恢复检查点，这在"多方案对比"场景中非常有用。
# 获取当前所有快照 snapshots = list(app.get_state_history(config)) if snapshots: # 回滚到倒数第二个状态（比如模型回答完、但还没调用工具之前） target = snapshots[-2] app.update_state(config, {"messages": target.values["messages"]}) # 之后再次 invoke，就能从该节点重新执行这种"从某个历史状态重新分叉"的能力，很多传统链式框架完全做不到。它让我做智能体评测时特别方便：同一个状态可以喂给不同节点的变体，对比不同工具调度策略的输出效果。
4. 一个完整实战：可中断恢复的多步骤研究助手
4.1 需求设定与状态流转方案
光讲概念没感觉，我带你完整走一个能跑的项目。我们要做一个"多步骤研究助手"：用户给出一个研究主题，智能体先调用搜索工具收集素材，然后生成大纲，最后在进入终稿生成前请求人工确认；如果用户觉得大纲不满意，可以回退重写。整个流程里有两次关键机制：
- 中断机制：生成大纲后暂停图执行，等待人工输入确认或修改意见。
- 回退机制：收到修改意见后，不是重新启动流程，而是从"大纲生成"节点重新执行。
状态流转设计图如下（用文字描述）：入口节点收到用户研究主题 → 调用搜索工具 → 模型提炼大纲 → 触发interrupt暂停 → 人工确认 → 如果确认则进入终稿节点 → 结束；如果要求修改，则把修改意见并入状态，重新走向大纲节点。
4.2 动态工具注册与节点实现
首先定义两个搜索工具（为了示例简单，这里用模拟数据，实际可替换成真实的搜索引擎 API 或向量检索）：
def search_wiki(query: str) -> str: """搜索百科知识，返回简要结果""" # 实际项目中可替换成真实搜索 API return f"关于『{query}』的百科结果：这是一段模拟返回内容，涵盖背景、历史与现状。" def search_news(query: str) -> str: """搜索最新资讯""" return f"关于『{query}}』的最新资讯：模拟返回2025年相关进展..."然后把两个工具放到一个列表中，LangGraph 中模型节点内部可以用bind_tools把这些工具注册给模型：
tools = [search_wiki, search_news] # 模型初始化示例（使用兼容 OpenAI 协议的任意模型服务） from langchain_openai import ChatOpenAI model = ChatOpenAI( model="your-model-name", # 替换为你的模型标识 api_key="your-api-key", base_url="your-api-endpoint", # 兼容 OpenAI /api 的端点 temperature=0 ).bind_tools(tools)实际生产中你可以接入任意兼容 OpenAI 协议的模型服务，LangGraph 只负责编排，模型本身谁提供并不重要。
4.3 三节点图：搜索、大纲、终稿
下面定义研究助手的三个核心节点：
def search_node(state: AgentState) -> dict: """根据当前用户问题调用多个工具，并把结果追加到 messages""" user_message = state["messages"][-1].content wiki_res = search_wiki(user_message) news_res = search_news(user_message) return { "messages": [ {"role": "tool", "content": wiki_res}, {"role": "tool", "content": news_res}, ], "stage": "agent", } def outline_node(state: AgentState) -> dict: """根据工具结果生成大纲；这里通过模型调用来完成，并插入中断点""" system_prompt = "你是资深研究助理，基于检索到的资料生成一份结构化大纲，输出为 markdown 格式。" response = model.invoke([ {"role": "system", "content": system_prompt}, *state["messages"] ]) return { "messages": [{"role": "assistant", "content": response.content}], "stage": "confirm", } def final_node(state: AgentState) -> dict: """根据确认后的大纲生成完整终稿""" system_prompt = "你是资深写作专家，请根据大纲扩展为完整研究文章，保留结构并补充细节。" response = model.invoke([ {"role": "system", "content": system_prompt}, *state["messages"] ]) return { "messages": [{"role": "assistant", "content": response.content}], "stage": "done", }4.4 interrupt 中断与人工确认的关键代码
LangGraph 的interrupt是一个非常重要、也容易被误解的机制。它不是简单的"暂停线程"，而是把控制权交还给用户进程，应用状态会被持久化；在后续调用中，你可以通过Command(resume=...)把人工输入的结果重新注入到中断点。
基于较新的 API 写法：
from langgraph.types import interrupt, Command def confirm_node(state: AgentState) -> dict: """在大纲生成后暂停，向用户请求确认或修改意见""" outline = state["messages"][-1].content # 触发中断，把提示信息抛出；应用层会捕获并等待用户输入 user_feedback = interrupt({ "type": "confirm_outline", "outline": outline, "question": "大纲是否符合预期？直接确认请输入 confirm，需要修改请输入修改意见。" }) # 用户通过 resume() 返回输入 if user_feedback and user_feedback.strip().lower() != "confirm": return { "messages": [{"role": "user", "content": f"大纲修改意见：{user_feedback}"}], "stage": "agent", } return {"stage": "confirm_passed"}路由函数根据stage决定下一步走向：
def route_after_outline(state: AgentState) -> str: if state.get("stage") == "confirm_passed": return "final" return "outline" # 如果用户提了修改意见，我们回退到 outline_node 重新生成大纲这里有个重要设计：回退时outline_node再次执行，但此时state["messages"]里已经有了用户的修改意见，所以重新生成的大纲会基于新意见推演。这正是图结构相比链式结构的巨大优势——循环不是 bug，而是特性。
4.5 运行入口与恢复会话代码
以下是main.py的交互式入口，完整演示了中断如何恢复：
import sqlite3 from langgraph.checkpoint.sqlite import SqliteSaver from graph import build_research_graph def main(): graph = build_research_graph() with SqliteSaver.from_conn_string("research_memory.db") as checkpointer: app = graph.compile(checkpointer=checkpointer) config = {"configurable": {"thread_id": "research-session-001"}} # 第一轮触发 user_input = "请研究一下：大语言模型在医疗诊断中的应用与挑战" events = app.stream( {"messages": [{"role": "user", "content": user_input}], "user_intent": "research", "stage": "start", "turn_count": 0}, config, stream_mode="updates" ) for event in events: print("事件：", event) # 当检测到 __interrupt__ 事件时，说明图已暂停 if "__interrupt__" in event: interrupt_payload = event["__interrupt__"][0] print("需要人工确认：", interrupt_payload.value["question"]) # 实际应用中，这里可以接入 WebSocket、HTTP 回调或 CLI 输入 feedback = input("请输入确认意见：") # 第二个 stream 调用会从上次中断处恢复，并携带人工输入 resumed_events = app.stream( Command(resume=feedback), config, stream_mode="updates" ) for resumed in resumed_events: print("恢复事件：", resumed) if __name__ == "__main__": main()这里有几个实操要点：
- 第一次
stream执行到confirm_node时，模型已经生成大纲，但图暂停在中断处，后续节点（如 final_node）不会执行。 - 第二次
app.stream时传入的不是普通字典，而是Command(resume=feedback)，这会直接把反馈注入到中断处，confirm_node拿到返回值后继续往下跑。 - 如果你不希望每次都用
input()阻塞等待，完全可以把中断消息投递到消息队列、钉钉/企微回调或 Webhook 中，架构上完全解耦。
5. 几个最容易踩的坑：并发、隔离与调试技巧
5.1 thread_id 并不保证业务隔离
很多新手以为thread_id是"不同会话之间天然隔离"，这大错特错。Checkpointer 只负责按thread_id存取状态，它不会阻止你在不同的thread_id下调用同一个外部数据库、同一个文件句柄或同一个全局变量。如果你的工具函数里修改了全局缓存或共享连接，多个thread_id之间仍然会互相污染。
我的处理习惯是：所有工具函数都设计成纯函数——入参只有消息内容和业务字段，返回值只有新消息和状态更新。任何需要外部资源的对象要么在节点内部创建，要么以依赖注入的方式传入，绝不用模块级全局变量。
5.2 中断恢复时不要重传整个输入
使用Command(resume=...)恢复时，你只需要传入"中断点需要的那一小块数据"，而不必再次传入完整的 messages。但这里有另一个常见问题：如果恢复时业务上下文变了（比如用户改了需求），你可能想更新某些业务字段。这时候可以配合update_state在恢复前修改状态：
app.update_state(config, {"user_intent": "changed_intent"}) app.stream(Command(resume=feedback), config)否则，新传入的resume值将基于旧的业务字段继续执行，容易产生逻辑错位。
5.3 共享 Checkpointer 并发写冲突
我用 SQLite 做持久化早期踩过一个坑：多个进程（比如 Web 服务的多个 Worker）同时写同一个memory.db时，偶尔会遇到database is locked报错。原因是 SQLite 的默认事务隔离级别不适合高并发写。
解决办法有两个方向：
- 把写操作串行化：在应用层用一个连接池，或者给所有
invoke调用加一个全局锁（代价是并发吞吐下降）。 - 换 Postgres 后端：LangGraph 支持
AsyncPostgresSaver（需要安装langgraph-checkpoint-postgres），并发能力强得多，适合做线上服务。
如果你只是本地 Demo 或内网工具，SQLite 完全够用；如果要上生产，建议直接上 Postgres。
5.4 用 get_state/history 做调试，不要只靠打印日志
调试 LangGraph 时，最多人上来就打印每个节点返回的 dict，但这样很难看到时间线。我推荐的做法是：
# 获取某线程当前完整状态 current_state = app.get_state(config) print("当前状态值：", current_state.values) # 获取全部历史快照 for snap in app.get_state_history(config): print("快照元数据：", snap.metadata) print("节点ID：", snap.next)app.get_state_history()返回的是一个倒序列表，能看到每次节点执行之后的状态存档和下一个将执行的节点。排查"为什么走到了错误的分支"时，这个命令比任何日志都直观。
另外，app.get_graph().draw_mermaid()（如果你用 Jupyter）可以把整张图以流图形式可视化出来，看路由关系一目了然。如果你用的环境中没有 Mermaid 渲染能力，也可以用print(app.get_graph().to_json())查看节点和边的 JSON 结构。
5.5 工具异常不应该让整个图崩溃
真实环境里，搜索引擎可能限流、数据库可能超时。如果不做异常处理，工具节点抛异常会直接中断整张图。我惯用的做法是在工具节点内部捕获异常并返回一个"工具错误消息"，让模型自己判断下一步：
def safe_search_node(state: AgentState) -> dict: try: # 调用多个检索工具 ... return {"messages": [...]} except Exception as e: return { "messages": [{"role": "tool", "content": f"检索失败: {str(e)}"}], "stage": "agent", }这样模型会看到错误文本，大概率会尝试换一个工具或者向用户解释失败原因，而不是整个智能体僵死。
写在最后的一点个人体会
LangGraph 这几轮迭代我一路跟下来，最大的感受是：它把智能体从"一段顺序执行的脚本"变成了"一张可以被检查、被恢复、被调试的图"。刚开始迁移思维确实有点别扭，尤其习惯了 Chain 的线性心智模型后，总会下意识地想把所有东西排成一条直线。但一旦你尝到"中断恢复""状态回放""分支回退"的甜头，就再也回不去了。
我个人最建议你动手实验的第一个案例，就是这篇文章的三节点研究助手：搜索 → 大纲 → 终稿，中间加一个人工确认中断点。这个场景足够简单、覆盖了 LangGraph 约 80% 的高频特性，又不会因为逻辑复杂而让你在调试中迷失。把它在本地跑通之后，再逐步往里面加入更多工具、更多条件分支、甚至嵌套子图，你会发现自己对"有状态智能体"的理解完全不一样了。
