LangGraph + PostgreSQL Checkpoint：打造可恢复的Agent运行时架构

开篇：从一个让你抓狂的“断点”说起

我现在的日常已经离不开 LangGraph，但真正让我下定决心重构整个 Agent 项目架构的，是一次印象极深的线上事故。当时我用一个最朴素的 while 循环，在代码里让大模型反复调用工具、把结果塞回上下文再继续跑。业务方管这个叫“Agent”，我也没多想，直到某个凌晨，任务跑到一半进程崩了。重启之后，前面十几轮的工具调用记录、中间结果、已经写入数据库的半成品订单，全部付之东流。

那一刻我才意识到：这个所谓的 Agent 运行时，本质上就是一个会丢状态的内存循环。只要进程一死，所有干活到一半的步骤全没了，而且没有任何“接着干”的可能。那次事故之后，我开始认真研究怎么把 Loop 从“手写的 while True”升级成一个具备恢复能力的可控 Runtime。整个改造的核心，就是三样东西：LangGraph 的状态图、PostgreSQL 的 Checkpoint 持久化，以及 AG-UI 的协议化事件输出。

如果你现在还在用普通 Python 代码硬写 Agent 循环，或者刚接触 LangGraph 但不知道它比 LangChain 好在哪，又或者已经被“human in the loop”这个词绕晕了——这篇文章就是给你准备的。我会完整拆解这套方案的选型逻辑、落地步骤和踩坑记录，文末还有我私藏的排查清单，希望能帮你跳过我在凌晨三点踩过的那些坑。

1. 整体设计与思路拆解：为什么不再手写 Loop

1.1 手写 Loop 的真相：它不是不能用，是扛不住事

先说结论：手写循环在 demo 和单机脚本里完全没问题，但一旦进入生产环境，它会暴露三个致命弱点。

第一个弱点是状态全在内存。所有中间变量、模型回复、工具返回值，都堆在一个 Python dict 或 SQLite 临时表里，进程一退出，什么都没了。第二个弱点是没有“暂停”的概念。你想在某个节点停下来等用户确认，只能靠 sleep 轮询，或者硬生生拆成两段代码，中间状态靠数据库手工同步。第三个弱点最隐蔽，循环跑飞了没人知道。大模型偶尔会陷入死循环，比如反复调用同一个工具十几次，返回结果都是一样的错，你的手写循环根本来不及拦截，只会把账单和错误一起滚雪球。

那 LangGraph 是怎么解决这三个问题的？它把整个 Agent 流程建模成一个显式的状态图，每个节点接收一个共享状态（State），处理后返回一个状态增量。这种“单状态流转 + 节点纯函数化”的设计，天然就给持久化、恢复、人为干预留好了位置。你不需要自己设计消息队列或状态快照，框架层面已经提供了。

1.2 LangGraph 与 LangChain：别再纠结它们谁代替谁

网上经常有人问 langchain 和 langgraph 的区别。我说一句可能得罪人的话：LangChain 的定位是工具集，LangGraph 的定位是运行时。LangChain 提供了一大堆封装好的组件——模型调用、检索器、Tool 接口、Prompt 模板——但它的链式调用（Chain）模式更像是顺序脚本，并不强调状态管理和循环控制。你可以在 LangGraph 的节点里放心使用 LangChain 的模型和工具，这完全不冲突。

从工程实践的角度看，LangGraph 真正吸引我的是它对“状态”做了显式建模。你可以定义一个 TypedDict 作为全局状态，也可以定义节点之间的消息传递协议。LangChain 的 AgentExecutor 本质上也是循环，但它的循环是封装死的，你对它的控制只有几个参数。而 LangGraph 允许你把循环画出来，甚至允许你在某个条件边（conditional edge）上做任意分支。这个自由度在生产环境里非常值钱。

1.3 为什么选 PostgreSQL Checkpoint 而不是 Redis 或本地文件

LangGraph 官方提供了多种 Checkpoint 存储后端，包括内存、SQLite、PostgreSQL、Redis 等。我在选型时的判断标准很简单：生产环境要保证两点——持久化可靠，以及并发恢复时不丢状态。Redis 当然快，但如果不做 AOF 持久化，宕机一样丢数据。SQLite 适合单机，但多实例部署时锁竞争很头疼。PostgreSQL 在行业里几乎是“默认的可靠数据库”，而且 LangGraph 的langgraph-checkpoint-postgres库写得很成熟，基于它做状态恢复基本是开箱即用。

我们团队的生产环境已经有一套自建的 PostgreSQL 集群，所以选它还有一个现实理由：不用额外引入基础设施。Checkpoint 本质上就是把图的执行快照存下来，包括当前的节点位置、状态数据和待处理的任务队列。后端的区别只在于序列化格式和存储介质，而这个存储介质的可靠性，直接决定了整个 Runtime 能不能扛住宕机。

接下来的内容，我会围绕三个重点展开：LangGraph 图的搭建、PostgreSQL Checkpoint 的接入与恢复、AG-UI 协议化输出的设计与事件流规范。这三块拼在一起，才构成一个真正“可恢复 Runtime”的完整闭环，缺一个都不行。

2. 核心细节解析与实操要点：LangGraph 图、Checkpoint 与 AG-UI 三件套

2.1 LangGraph 状态图的核心概念，不扯术语直接讲

LangGraph 的核心就四个东西：State、Node、Edge 和 Conditional Edge。State 是全局共享的数据结构，我一般用 TypedDict 定义，这样 IDE 提示和运行时校验都比较舒服。Node 是处理逻辑的单元，它接收整个 State，返回一个部分 State 的 dict，框架会自动合并回去。

Edge 把 Node 串起来，表示“这一步执行完，下一步固定走哪个节点”。Conditional Edge 则是一个函数，输入是当前的 State，输出是下一跳节点的名称。整个 Agent 主循环在我这里就是一条带条件边的图：入口是agent节点，生成一次模型回复；如果回复里有工具调用，就跳tools节点，把工具结果写回 State，再跳回agent节点继续生成；如果没有工具调用，条件边直接指向end节点。

看一段伪代码，应该比读十篇文档都有用：

from typing import TypedDict, Literal from langgraph.graph import StateGraph, END class AgentState(TypedDict): messages: list tool_results: dict def agent_node(state: AgentState) -> dict: # 调用 LLM，返回新的消息列表 ... def tools_node(state: AgentState) -> dict: # 执行工具调用，把结果写入 state ... def route_after_agent(state: AgentState) -> Literal["tools", "end"]: if state["messages"][-1].tool_calls: return "tools" return "end" g = StateGraph(AgentState) g.add_node("agent", agent_node) g.add_node("tools", tools_node) g.add_edge("tools", "agent") g.add_conditional_edges("agent", route_after_agent) g.set_entry_point("agent") g.add_edge("end", END) app = g.compile()

这段代码跑起来就是一个标准的 Agent Loop。但它和手写循环最大的区别是：这个 Loop 的每一步都是可观测、可暂停、可恢复的。判断依据很简单——只要你的图结构足够清晰，Checkpoint 就能在任意两步之间保存快照，下次从快照位置继续跑。手写循环做不到这点，因为在while里执行到哪一行并不构成一个可恢复的边界。

提示：节点函数里尽量不要直接修改传入的 state 对象，而是返回一个部分更新的 dict。LangGraph 的底层合并逻辑更可靠，也不容易串数据。

2.2 PostgreSQL Checkpoint 接入：不只是存个快照

Checkpoint 这个词容易让人误解成“数据库备份”。实际上，LangGraph 的 Checkpoint 是每个超步（super-step）的细粒度状态记录，它保存的信息包括：当前正在执行的节点、节点间传递的消息、状态数据的序列化结果、以及图执行的时间线。

要接入 PostgreSQL，先安装依赖：

pip install langgraph-checkpoint-postgres

然后在代码里创建连接池和 Checkpoint 实例：

from langgraph.checkpoint.postgres import PostgresSaver from psycopg import Connection conn = Connection.connect( "postgresql://user:password@localhost:5432/agent_runtime", autocommit=True, ) checkpointer = PostgresSaver(conn) # 首次使用需要初始化数据表 checkpointer.setup() app = g.compile(checkpointer=checkpointer)

关键点在于compile(checkpointer=...)这一步。一旦编译时传入了 Checkpointer，LangGraph 就会自动在每个节点执行完之后，把当前状态写入 PostgreSQL。你不需要在业务代码里手动埋点，save 的时机是框架控制的。

还有一个我必须提醒你的细节：连接参数必须开启 autocommit。我第一次接入时没注意，结果状态一直写不进数据库，卡了整整一个下午排查，最后发现是事务没有自动提交，所有 Checkpoint 都回滚了。psycopg 的连接默认不是 autocommit 模式，这在普通 SQL 操作里没事，但 LangGraph 的 saver 会频繁写入快照，如果不自动提交，事务会越积越多，最终要么锁死，要么丢失。

2.3 AG-UI 在 Runtime 中的定位：把“内部状态”变成“可消费事件”

AG-UI 你可能不熟，它不是 LangGraph 的一部分，而是一套定义Agent 与用户/前端之间事件流格式的协议规范。运行时内部的状态是图节点自己用的，可前端界面却需要实时显示“哪个工具正在调用”“当前轮到谁说话”“是否在等待用户确认”。如果没有统一协议，你就会陷入自己设计消息格式的泥潭，前端一个字段一个字段跟你对，烦不胜烦。

AG-UI 的核心是提供了一套标准的事件类型，比如工具调用开始、工具调用结束、Agent 消息增量、会话暂停等。我在 LangGraph 节点里会写一个回调函数，把这些事件按 AG-UI 协议输出到前端：

from ag_ui.core import AGUISession def agent_node(state: AgentState) -> dict: session = AGUISession.from_state(state) result = llm.invoke(session.messages) session.emit("agent_message", content=result.content) return {"messages": [result]}

有了 AG-UI，可恢复 Runtime 的价值才能真正显现：不仅后端能恢复，前端的展示状态也能同步恢复。你刷新页面之后，可以从数据库中读取历史 Checkpoint 对应的事件流，把界面恢复到崩溃前的样子。这一点在手写 Loop 里想都不敢想。

2.4 超时、中断与边界条件：生产环境的三个隐藏炸弹

除了核心概念，还有三个边界条件，我希望你从一开始就想清楚，不然上线后早晚出事。

第一个是单步超时。OpenAI 这样的大模型接口偶尔会超时，工具调用也可能卡死。我在每个节点上会包一层超时控制，不让任何一步无限阻塞。LangGraph 本身没有原生的 timeout 参数，但你可以用asyncio.timeout或者functools.partial结合concurrent.futures实现，反正不能裸调 API。

第二个是中断拦截。LangGraph 的interrupt机制是专门的“暂停”节点设计，调用它会抛出中断，让图执行暂停并等待外部输入。这个特性天然适合人工确认场景。同时你需要自己判断“循环何时该停”，如果某个工具连续被调用超过 N 次，就应该让条件边返回中断，而不是傻乎乎地继续喂给模型。

第三个是死循环防御。有时候模型会自己陷入循环，即使没有工具调用，它也会说一堆空话，导致状态无限膨胀。我的做法是在状态里加一个step_count字段，每经过一次 agent 节点就 +1，条件边检查如果超过最大步数，直接跳到 END。

这三个炸弹不解决，你的 Runtime 即使能恢复，也会在一个小时内被生产流量打垮。

3. 实操过程与核心环节实现：一步步搭出可恢复 Runtime

3.1 定义状态结构与图结构，我不建议一上来就写代码

好的开始不是敲代码，而是先梳理状态结构。我的做法是先画一张状态字段表，明确哪些数据需要被持久化，哪些只是在单次执行中临时用。

表里“是否持久化”的判断标准很简单：只要是恢复后续执行所需的数据，就必须持久化；只影响 UI 或日志的临时数据，可以放在 State 之外。这里我踩过一个坑：把大模型的原始 response 对象直接塞进了 messages，但这个对象里有不可序列化的字段，导致 PostgreSQL Checkpoint 写入时报错。所以我在消息入 State 之前，都会转成统一的 dict 结构。

3.2 LangGraph 图的完整代码实现，可以直接抄

下面这段代码基本是我生产环境下的最小版本，去掉业务细节，保留骨架。我会把关键步骤讲透。

from typing import TypedDict, Literal from langgraph.graph import StateGraph, END from langgraph.checkpoint.postgres import PostgresSaver from psycopg import Connection class AgentState(TypedDict): messages: list tool_results: dict step_count: int pending_action: str MAX_STEPS = 10 def agent_node(state: AgentState) -> dict: # 1. 组织 prompt，把历史消息交给 LLM # 2. 如果有工具调用，就把调用信息附加到 messages # 3. step_count += 1 return { "messages": new_messages, "step_count": state.get("step_count", 0) + 1, } def tools_node(state: AgentState) -> dict: # 批量执行 messages[-1].tool_calls results = [] for call in state["messages"][-1].tool_calls: results.append(execute_tool(call)) return {"tool_results": {"calls": results}} def route_after_agent(state: AgentState) -> Literal["tools", "end"]: last_msg = state["messages"][-1] if state.get("step_count", 0) >= MAX_STEPS: return "end" if last_msg.get("tool_calls"): return "tools" return "end" graph = StateGraph(AgentState) graph.add_node("agent", agent_node) graph.add_node("tools", tools_node) graph.add_edge("tools", "agent") graph.add_conditional_edges("agent", route_after_agent) graph.set_entry_point("agent") graph.add_edge("end", END) # PostgreSQL 连接与 Checkpoint conn = Connection.connect( "postgresql://agent:secret@localhost:5432/agent_runtime", autocommit=True, ) saver = PostgresSaver(conn) saver.setup() app = graph.compile(checkpointer=saver)

执行时，你需要给每次运行分配一个thread_id，这个 ID 是关键中的关键。它标识的不只是会话，更是一条独立的执行流。后续恢复，就是拿着这个 thread_id 去查历史 Checkpoint。

config = {"configurable": {"thread_id": "order-20250101-001"}} result = app.invoke( {"messages": [{"role": "user", "content": "帮我查一下订单 12345 的状态"}]}, config=config, )

写入 PostgreSQL 的具体结构长什么样？你可以去数据表里看一眼，主要表有两张：checkpoints存储每个超步的快照，checkpoint_blobs存储序列化后的二进制内容。通过这两个表你能清楚看到整个图在什么时间执行到哪个节点，这为问题定位提供了非常有力的证据。

3.3 中断恢复的完整流程：从崩溃到原地复活

假设你的服务在工具调用节点执行过程中崩溃了，重启后你想要恢复这个订单流程。代码逻辑非常简单，依然拿着原来的 thread_id 调用invoke：

config = {"configurable": {"thread_id": "order-20250101-001"}} app.invoke(None, config=config)

有没有发现，我传给invoke的是None。这是 LangGraph 官方设计的行为模式——当传入 None 且存在 Checkpoint 时，它会自动从上次停止的位置继续。如果 Checkpoint 中没有记录，它会从 entry_point 重新开始。

实际项目里，重启后你会发现进程恢复的位置不一定从崩溃的节点中间，而是从崩溃前最后一个完成写入的快照点继续。这是一个非常重要的工程认知：Checkpoint 恢复不是“精确到字节”的还原，而是“恢复到最近一个可靠边界”。那些“执行了一半但还没形成快照”的副作用（比如已经调用了外部 API 但结果没写库），可能在恢复后不会重新执行，也可能会因外部 API 幂等性不足而重复执行。所以，工具本身一定要设计成幂等的，否则即使有 Checkpoint，也无法保证端到端的一致性。

3.4 人工确认节点怎么做：interrupt 与 update_state 的正确姿势

结合前面的两个核心机制，我把“人工确认”这个动作做成了图中一个反复使用的模式。

from langgraph.types import interrupt def human_review_node(state: AgentState) -> dict: action = state["pending_action"] # 挂起图，等待外部输入 user_decision = interrupt({"action": action}) return {"pending_action": None, "review_result": user_decision}

当节点执行到interrupt时，LangGraph 会抛出一个中断异常，整个图的执行暂停，状态已经持久化。你可以在另一个接口里拿到暂停信息，然后给用户展示确认按钮。用户点“同意”后，调用：

app.update_state( config, {"review_result": "approved"}, as_node="human_review_node", )

这一步会把新的状态注入到图中，并解除中断。随后再次invoke即可继续。这个流程对手写 Loop 来说，实现起来相当繁琐；而用 LangGraph 的标准机制，前后代码量不足 20 行。

注意一个细节：update_state的as_node参数很重要。如果你不指定它，LangGraph 可能在你恢复后找不到要执行的节点位置。我遇到过一次把 as_node 写成空，结果它直接从入口节点从头跑了一遍，所有的历史消息全乱了。

3.5 AG-UI 事件流如何和前端对接，完整示例

为了让前端能实时感知后端运行状态，我会把 AG-UI 事件通过 WebSocket 推给前端。事件流的时序如下：

{"type": "AGENT_MESSAGE_START", "message_id": "m1"} {"type": "TOOL_CALL_START", "tool": "order_query"} {"type": "TOOL_CALL_END", "tool": "order_query", "output": "订单状态: 已发货"} {"type": "AGENT_MESSAGE_DELTA", "content": "根据查询结果"} {"type": "AGENT_MESSAGE_DELTA", "content": "您的订单已发货"}

前端收到这些事件后，可以做流式展示，也可以在终端用户告知“刚刚卡住了”时，直接根据后端返回的历史事件流恢复界面。

这套协议的好处是标准、自解释。你不需要和下一位前端开发者吵架，告诉对方“我给你发一个 modelOutput，还有一个 toolOutput，还有一个 pendingAction”，而是让对方直接对照 AG-UI 的事件表取数据。在实际协作中，这个协议帮我节省的沟通成本远超预期。

4. 常见问题与排查技巧实录：这些坑我替你踩过了

4.1 问题速查表，直接对照症状找方案

4.2 排查实录：一次“恢复后状态错乱”的完整分析

有一次我遇到了一个很诡异的现象：系统恢复后，工具调用的返回结果里居然出现了上一次会话的数据。查了很久，最后定位到根因，是我的工具函数用了全局缓存，缓存 key 只包含参数，没包含会话 ID。第一次会话查询订单 111，第二次恢复查询订单 222，但工具读到了缓存里的 111 结果，直接返回了。

这个问题的启示是：接入 Checkpoint 只是让“状态”可恢复，但你的工具调用必须是纯粹的函数——相同的输入产生相同的输出，或者至少完全上下文无关。凡是依赖隐式上下文的工具，在恢复场景下一定会出问题。后来我把所有工具函数都改成显式传参会话 ID，并把缓存策略改成按会话隔离，问题就再没出现过。

4.3 我的独家避坑经验，提前说给你听

第一，不要滥用 checkpointer 的 setup 方法。它只应该在首次部署时执行一次，如果在每次启动时都调用，就会在高并发场景下产生 DDL 锁竞争。生产环境我通常用 ORM 迁移或手动 SQL 来建表。

第二，给 PostgreSQL 表建索引。LangGraph 官方建表语句里，checkpoints 表和 checkpoint_blobs 表的查询条件都基于 thread_id，所以一定要在 thread_id 字段上建立索引。我项目刚上线时没有加索引，恢复一次要扫全表，数据量一旦上去就明显感觉到查询变慢。

第三，不要把 Checkpoint 表和业务表混在一个连接池里。如果你让业务模块和 Checkpointer 共用同一个连接池，一旦业务表有大事务，Checkpoint 的写入可能被阻塞。我在生产环境给 LangGraph 单独开了一个连接池，大小控制在 10 个连接以内，效果很稳定。

第四，监控 Checkpoint 写入延迟。Checkpoint 是我们这套可恢复 Runtime 的“命根子”，如果写入延迟突然飙高，意味着系统在崩溃发生时可能丢掉更多状态。我给 Checkpoint 写入加了指标上报和告警，阈值设置在 200ms，超过就报警，这会让你在状态真正丢失之前就介入处理。

4.4 从 LangChain Agent 迁移到 LangGraph 时，COM 层错误与 Runtime 报错速览

社区里有不少朋友是从 langchain agent executor 迁移过来的，常会碰到一个看起来很吓人的报错，比如在 Windows 环境下偶发 runtime error 相关字样，或者 webview2 runtime 相关的报错。这些报错其实和 LangGraph 本身没关系，多半是系统环境问题，比如缺少 Microsoft Edge WebView2 Runtime，或者 Visual C++ 运行库。LangGraph 只是纯 Python 包，它不依赖这些系统组件，所以遇到这类报错时，先检查你的部署环境，别把锅甩给框架。

有一个容易混淆的点：langgraph 依赖 langchain-core，但不强制依赖 langchain 全套。如果你在项目里只用 LangGraph 和模型接口，哪怕完全没有 LangChain 的 Chain 也完全没问题。反过来，如果你已经在用 LangChain 的各种工具集成，LangGraph 也可以作为一个外层运行时把它们挂载进图里，两者是互补关系。

5. 可恢复 Runtime 的架构边界与扩展空间：不吹不黑，讲点实际价值

5.1 它能解决什么问题，不能解决什么问题

做完这套改造，最直接的价值有三个。

第一，进程崩溃后可以原地恢复，不丢中间状态，这是当初做这件事的初衷，实际效果也是在一次真的宕机后验证的。当时某个订单处理任务执行到第三步，服务重启后我拿着 thread_id 调了一次 invoke，它从断点继续把订单状态更新完成了，整个过程没有让用户重新发起请求。

第二，人工介入成为一等公民。审核、确认、多轮对话暂停，都可以通过 interrupt 和 update_state 实现，不再需要自己用临时表去设计中转状态。

第三，执行过程完全可观测。由于 Checkpoint 落库了每个超步的状态，你可以像回放录像一样查看任何一个任务的执行轨迹。这在排查问题、审计合规场景下价值巨大。

但也要说清楚局限性。首先，它不会帮你自动解决外部系统的副作用一致性问题；如果你调用了一个只能成功不能失败的下游系统，Checkpoint 并不能让它支持事务回滚。其次，如果你的节点函数不是纯函数，有随机性或全局状态，那么从 Checkpoint 恢复后，行为可能和崩溃前不一致。最后，LangGraph 本身不支持跨进程分布式图执行——多实例部署时，同一线程 ID 最好只由一个实例处理，否则两个实例同时从同一个 Checkpoint 恢复，就不只是状态冲突，而是直接逻辑错乱了。

5.2 后续可以怎么扩展：消息队列、子图拆分、流式输出

目前这套 Runtime 已经能支撑单条执行流的完整生命周期。如果业务量起来，我会考虑做三个扩展方向。

方向一是接入消息队列做异步任务分发。LangGraph 原生是同步阻塞调用，但如果你的任务本身适合放进任务队列，就可以在图的入口节点把任务投递到队列，由 worker 进程异步执行图，结果通过回调回来。这样能把 Runtime 和 Web 服务生命周期解耦，适合长时间运行的任务，也能避开“同步 HTTP 接口等待 5 分钟”这种尴尬。

方向二是子图拆分。如果单个图节点数超过十来个，维护成本会急剧上升。LangGraph 支持把一部分节点封装成子图，在主图的节点里调用它。子图也有自己的 State 和 Checkpoint，这样每个功能域可以独立开发和测试，调试体验会舒服很多。

方向三是流式输出增强。目前 AG-UI 事件已经支持流式，你可以进一步结合 SSE 或 WebSocket 把事件推送做得更平滑。官方提供的 stream 方法和 astream_events 接口就是为此设计的，节点内部的 token 级输出都可以作为事件发出去，配合前端打字机效果再合适不过。

结尾：最后一次迁移之后，我再也不写裸 while 循环了

从手写 Loop 到 LangGraph 可恢复 Runtime，这次重构对我的项目来说，不只是一个技术栈替换，更像是一次思维方式的转变。现在每次写 Agent 流程，我下意识的第一步永远是“这个流程的状态边界在哪”，第二步是“如果这一步挂了，从哪一步能爬起来”。带着这两个问题去设计，任何一个 Agent 项目都不会太走样。

如果你也打算做类似的改造，我给的建议是千万别开头就追求完美。先用最简的图结构，接入 PostgreSQL Checkpoint，把thread_id统一管理起来，再去加人工确认和 AG-UI 事件流。一步一步来，每一步都能独立验证，这样即使出问题，你也能很快定位到是图结构的问题、存储的问题，还是事件协议的问题。

最后分享一个小技巧：每次部署新版本前，我会用一份历史 Checkpoint 做一次完整的恢复演练，脚本会自动检查恢复后的状态数据和原状态数据是否一致。如果检查不过，我宁愿延迟发布，也不让一个不可恢复的新版本悄悄上线。这个习惯帮我避免了至少三次线上事故，希望能对你有用。