从手写 Loop 到可恢复 Runtime这个转折点我摸索了小半年。早期做 Agent 应用时一个带循环的自动任务跑起来不难难的是它跑到一半崩了、断网了、数据库连接超时了你到底是从头再来还是能从断点续上。后来我用 LangGraph 重写了整个调度层把 PostgreSQL 当作外部持久化 Checkpoint 的存储再配合 AG-UI 把多轮交互事件统一收口这才算真正把机器挂了人还在这件事落地了。这篇文章就把我踩过的坑、拆过的细节、最终跑通的方案完整写出来适合那些正在从手撸循环往健壮 Runtime 架构迁移的团队参考也适合想理解 LangGraph 状态恢复机制到底怎么工作的朋友。1. 从手写 Loop 到 Runtime为什么必须换1.1 手写循环最痛的不是跑不了而是断了接不上先说说我之前是怎么写智能体循环的。最简单直接的方式就是用一个while True包住调用大模型 → 解析输出 → 执行工具 → 再把结果拼接回 messages这个过程直到模型认为任务完成。代码不多逻辑看起来也直接但是在真实生产环境里跑上一段时间后问题就一件接一件浮出来。最疼的是崩溃恢复。假设循环在第三轮工具调用后工具返回了一个异常数据结构或者网络闪断导致 API 调用超时进程直接退出。等到你重启进程内存里的消息队列、当前状态、已经执行过的工具结果全部归零。你不得不把任务从头开始重新跑一遍前两轮对话白费了对用户来说体验极其糟糕。更麻烦的是如果你的循环里有副作用操作——比如已经扣过款、已经发过通知——重试时你根本不知道哪些操作重复执行了。为什么会出现这种状态丢失核心原因是手写循环把状态全部放在进程内存里变量、列表、上下文对象换个进程就什么都没了。这就像你写文章写到一半断电而你的草稿箱还停留在第一句话这种开发方式只适合演示 demo做不了正经的异步长任务。另一个痛点是循环逻辑和业务逻辑耦合。随着需求变多你会往循环里塞各种判断要不要重试、什么时候该问用户、工具结果怎么过滤。加上日志、计数器、定时器这些横切逻辑之后代码很快就变成一个面条循环改一处要担心另外三处。这种结构下你根本没办法做精细控制比如暂停一段时间等用户输入再继续或者某一步失败后走另一个分支而不是整体退出。1.2 LangGraph 带来的核心转变状态机和持久化LangGraph 解决这个问题的思路我觉得最精髓的就是把 Agent 循环从一段代码改写成了一张有状态的状态机图。在这个图里节点是函数边是跳转逻辑而状态对象是整个图运行的核心。每一步运行时读取当前状态执行节点更新状态再决定下一步走向。状态从内存局部变量提升为一等公民。这就带来了立竿见影的好处。因为状态对象是显式的你想持久化它、序列化它、在进程重启后重新加载它都是顺理成章的事情。即使循环中间程序崩了只要把最新状态的 Checkpoint 保存到外部存储重启后从 Checkpoint 恢复整个 Agent 就能从崩溃前那一刻继续跑而不是从头再来。这和我以前手写循环完全不是一个维度的能力。再说图结构的优势。LangGraph 的节点是 Function状态是 TypedDict图可以动态编译。你可以显式定义条件边比如如果工具执行结果不符合预期跳去重试节点否则继续主流程。这比一堆if-else嵌套写在循环体里清晰得多而且每一步都能被单独测试和调试。更重要的是LangGraph 内置了 Checkpoint 机制配合外部存储比如 PostgreSQL它天然支持断点续跑而不需要自己设计状态序列化协议。LangGraph 和 LangChain 的关系很多刚接触的人容易混淆。LangGraph 并不是 LangChain 的升级版本它是 LangChain 团队推出的一个独立库专注于构建有状态、可编排的 Agent 应用。LangChain 的核心价值在模型调用链路的抽象和工具的集成而 LangGraph 更关心你如何编排这些调用、如何管理状态、如何处理复杂分支和循环。在实际项目中两者经常配合使用LangChain 负责跟模型打交道和封装各种工具LangGraph 负责把整个流程组织成一张可控的图。1.3 AG-UI把用户交互从回调地狱变成事件流Loop 写到最后还有一个绕不开的问题Agent 在关键节点需要向用户提问确认。手写实现通常是回调函数把用户响应传回循环里但这也是一种隐式状态一旦恢复或重试回调很容易丢失交互上下文也容易错乱。AG-UI Agent-User Interface是我在这套架构里用来统一交互协议的方案。它把 Agent 和前端 UI 之间的消息定义成标准事件流核心概念是事件Event和指令Instruction。Agent 输出的事件统一走一个协议格式事件类型包括任务开始、工具调用、需要用户输入、最终回复等。前端只需要解析这个事件流渲染对应组件然后通过标准指令把用户的操作回报给 Runtime。这样设计的价值在于前端不需要关心 Agent 内部循环的细节只需要消费标准协议事件。而 Runtime 也只需要维护一个事件流通道无论是重连、恢复还是多端同步都是往这个统一通道发消息。AG-UI 的事件结构天然适合和 LangGraph 的interrupt()机制配合——当图执行到需要用户确认的节点时Runtime 暂停并发出事件前端渲染并返回用户指令Runtime 通过Command(resume...)把指令喂回图。这也让交互环节成为可恢复的持久化状态而不是脆弱的临时回调。所以整条链路就清晰了LangGraph 负责把 Agent 循环变成有状态图PostgreSQL Checkpoint 负责持久化快照AG-UI 负责把人和机器的交互统一为可恢复事件流。这就是我从手写 Loop 迁移到可恢复 Runtime 的核心三步。2. 环境搭建与工具选型先配好三件套2.1 LangGraph 与 LangChain 的版本选择如果你是从零开始搭这套 Runtime我建议直接用 LangGraph 的最新稳定版。以我写这篇时的经验LangGraph 的 API 已经比早期版本稳定很多核心抽象基本固定但是不同小版本之间的参数名和导入路径偶尔会有调整所以最好锁定一个版本不要盲目追新。安装 LangGraph 有几个可选依赖要留意。基础安装只包含核心图执行引擎如果要支持 PostgreSQL Checkpoint就需要安装langgraph-checkpoint-postgres这个库。这个库内部依赖 psycopg 驱动它会负责把 checkpoint 数据写入 PostgreSQL 的指定表中。安装命令大致是这样pip install langgraph langgraph-checkpoint-postgres如果你还需要配合 LangChain 使用已有的工具和模型封装那可以把langchain-openai、langchain-community这类包一起装上。但是记住LangGraph 本身并不强制依赖 LangChain它的节点函数只要是普通 Python 函数就行你完全可以自己封装模型调用和工具执行逻辑不一定非要走 LangChain 的链式 API。我一直建议团队在正式开始写代码之前先想清楚一个问题你要的是 LangGraph 的状态机和恢复能力还是 LangChain 的工具链生态如果你已经有自己的一套工具封装LangGraph 单独用完全没问题。如果把两者混在一起用反而容易搞不清某些报错是来自哪一层。2.2 PostgreSQL Checkpoint 的存储设计接着是 PostgreSQL 的准备。如果你本地没装 PostgreSQL用 Docker 起一个实例是最快的docker run --name pg-checkpoint -e POSTGRES_PASSWORDpostgres -p 5432:5432 -d postgres:16连接字符串我习惯放在环境变量里避免写死在代码中export POSTGRES_URIpostgresql://postgres:postgreslocalhost:5432/postgres然后建一个专门的数据库不和业务数据混在一起CREATE DATABASE agent_runtime;在 LangGraph 中使用 PostgreSQL Checkpoint 的核心用法是这样的from langgraph.checkpoint.postgres import PostgresSaver from psycopg import Connection conn Connection.connect(postgresql://postgres:postgreslocalhost:5432/agent_runtime, autocommitTrue) checkpointer PostgresSaver(conn) checkpointer.setup() # 创建必要的表这里有几个细节需要说明。autocommitTrue是必须的因为 LangGraph 在执行图的过程中需要在节点完成后立即把 checkpoint 写入数据库如果使用事务包裹又没及时提交可能导致状态写不进去。setup()方法会创建 checkpoints 表以及相关索引只需要执行一次之后每次启动应用都不用重复建表。如果你用的是异步版本记得用AsyncPostgresSaver连接方式也要改成AsyncConnection.connect()。我实际运行中最常用的是每执行完一个节点就保存一次 checkpoint也就是checkpoint频率设置成默认的__end__之前的每个 step。这样即使整个图运行到中间某个环节突然崩溃重启后加载最近的完整 checkpoint也能找到当时的状态。但是 checkpoint 写得太频繁也有问题接下来会讲。2.3 AG-UI 的接入方式AG-UI 并不是一个具体的软件包而是一套协议规范所以接入方式取决于你怎么实现这个协议。如果你用 Python 后端最常见的做法是引入官方 SDK 或者自己按照规范生成事件对象。目前社区里比较常用的是在 FastAPI 应用里暴露一个 SSEServer-Sent Events端点把 Runtime 产生的事件流实时推送到前端。在 LangGraph 的项目中我会把 AG-UI 事件层放在 Runtime 和 FastAPI 之间后端 Runtime 层负责执行 LangGraph 图生成事件。AG-UI 层将内部事件转换成标准协议消息通过 SSE 推给前端。前端根据事件类型渲染 UI用户操作通过标准指令消息回传。之所以选择 SSE 而不是 WebSocket是因为 Agent 会话主要是单向推送为主用户交互频率不高SSE 实现简单、浏览器原生支持、断线重连也容易处理。如果你的场景需要高频双向交互WebSocket 也可以只是协议序列化要做更多处理。我这里用的是 SSE 标准 JSON 事件实测足够稳定。3. 核心实现把 Agent 循环改造成可恢复 Runtime3.1 定义状态对象和图结构整个改造的第一步是把原来那个隐式的循环变量变成显式的状态定义。我用 TypedDict 描述状态数据结构from typing_extensions import TypedDict from langchain_core.messages import AnyMessage class AgentState(TypedDict): messages: list[AnyMessage] current_tool_calls: list[dict] pending_user_input: bool retry_count: int这里的messages是核心对话历史每次节点执行后追加新消息current_tool_calls记录当前待执行的工具调用信息pending_user_input是一个标记表示当前是否正在等待用户输入retry_count统计工具调用失败后的重试次数防止无限循环。接着定义节点函数。以调用模型和执行工具两个核心节点为例from langchain_openai import ChatOpenAI model ChatOpenAI(modelgpt-4o) def call_model(state: AgentState) - dict: response model.invoke(state[messages]) return {messages: [response], current_tool_calls: response.tool_calls} def execute_tool(state: AgentState) - dict: tool_calls state[current_tool_calls] results [] for call in tool_calls: tool tools_by_name[call[name]] result tool.invoke(call[args]) results.append(ToolMessage(contentstr(result), tool_call_idcall[id])) return {messages: results, retry_count: 0}最后构建图并加上条件判断from langgraph.graph import StateGraph, END from langgraph.graph.message import add_message def should_continue(state: AgentState) - str: if state[current_tool_calls]: return continue return end graph StateGraph(AgentState) graph.add_node(agent, call_model) graph.add_node(tools, execute_tool) graph.set_entry_point(agent) graph.add_conditional_edges( agent, should_continue, {continue: tools, end: END} ) graph.add_edge(tools, agent)这段代码很容易理解模型节点和工具节点循环交替模型有工具调用就执行工具然后回到模型没有工具调用就结束。相比手写while循环这里每个节点都是独立函数状态流是显式维护的。3.2 把 Compile 变成可恢复的 Runtime图定义好之后关键一步是编译时挂载 checkpointercompiled_graph graph.compile(checkpointercheckpointer)这时的compiled_graph已经具备持久化能力。你要运行一个 Agent 任务操作方式如下config {configurable: {thread_id: user-session-001}} events compiled_graph.stream( {messages: [HumanMessage(content帮我查一下北京的天气)]}, config, stream_modevalues ) for event in events: # 处理每个事件并推送到 AG-UI 前端 ...这段代码的核心在thread_id。一个thread_id代表一个完整的会话流程所有 checkpoint 都以 thread_id 为维度保存。如果你在同一个 thread_id 上继续调用streamRuntime 会先把上次的状态加载回来再配合新的输入继续执行而不是从零开始。这就是可恢复 Runtime的空间含义——时间维度同样重要。如果你想让一个长时间运行的 Agent 任务支持暂停和恢复能力LangGraph 内部还会为每次 checkpoint 打上时间戳在加载的时候会自动选择特定thread_id的最新 checkpoint。同时它还会在捕获事件的时候暴露 checkpoint 的checkpoint_id你可以通过这个 ID 精确指认恢复点。配合config里的checkpoint_id参数你可以指定回滚到某个历史版本。这个机制怎么理解类比一下thread_id 是客户的会话 IDcheckpoint_id 是每一次快照的版本号。你可以在任意一个 checkpoint 上重新分支、重放、分叉整个 Runtime 不再是一条道走到黑更像一个可以随时存取进度的游戏存档。3.3 中断与恢复让 Runtime 停得住现在的架构已经解决了崩溃恢复问题但正常运行中的暂停是另一个场景。例如 Agent 执行到需要用户确认的步骤它不应该继续往下走而应该把控制权交还给用户。LangGraph 提供的答案是interrupt()函数。当图执行到某个节点时调用interrupt()它会暂停执行并抛出一个特殊信号Runtime 捕获后可以等待外部输入。比如在发送邮件之前需要用户确认from langgraph.types import interrupt, Command def require_confirmation(state: AgentState) - dict: user_confirmation interrupt({ question: 是否要发送这封邮件, draft: state[messages][-1].content }) if user_confirmation yes: # 执行发送操作 return {messages: [SystemMessage(content用户已确认发送)]} return {messages: [SystemMessage(content用户取消发送)]}这时候compiled_graph.stream()会先产出事件表示节点暂停等待输入一直等到你向同一个thread_id发送恢复指令events compiled_graph.stream( Command(resumeyes), configconfig, stream_modevalues )这个Command(resume...)就是 AG-UI 协议中用户指令回传的入口。interrupt()和Command(resume)配对使用构成了人和机器之间的握手协议。前端通过 AG-UI 事件流看到需要确认的事件渲染确认按钮用户点击后把 yes 通过指令回传Runtime 把值交给interrupt()的返回值继续执行后续图逻辑。这种状态恢复比起回调函数可靠得多因为所有等待中的状态都保存在 PostgreSQL checkpoint 里即使恢复过程中 Runtime 重启Command(resumeyes)仍然可以找到正确的 checkpoint 继续执行。用户不会察觉后台发生了什么只是看到发送中……和已发送的界面变化。3.4 把 AG-UI 协议接到 Runtime 上为了让前端能消费事件我写了如下简单的 AG-UI 事件适配层def to_agui_event(event: dict) - dict: # 将 LangGraph 的 state 事件转换为 AG-UI 协议标准事件 if messages in event: last_msg event[messages][-1] if hasattr(last_msg, tool_calls) and last_msg.tool_calls: return {type: tool_call, data: last_msg.tool_calls} if hasattr(last_msg, content) and last_msg.content: return {type: assistant_message, data: last_msg.content} return {type: state_update, data: event}然后在 FastAPI 里提供一个 SSE 流式端点推送事件给前端from fastapi import FastAPI, Request from fastapi.responses import StreamingResponse import json app FastAPI() app.post(/agent/run) async def run_agent(request: Request): body await request.json() thread_id body.get(thread_id) user_input body.get(input) config {configurable: {thread_id: thread_id}} async def event_stream(): async for event in compiled_graph.astream( {messages: [HumanMessage(contentuser_input)]}, config, stream_modevalues ): yield fdata: {json.dumps(to_agui_event(event), ensure_asciiFalse)}\n\n return StreamingResponse(event_stream(), media_typetext/event-stream) app.post(/agent/resume) async def resume_agent(request: Request): body await request.json() thread_id body.get(thread_id) user_decision body.get(resume) config {configurable: {thread_id: thread_id}} async def event_stream(): async for event in compiled_graph.astream( Command(resumeuser_decision), config, stream_modevalues ): yield fdata: {json.dumps(to_agui_event(event), ensure_asciiFalse)}\n\n return StreamingResponse(event_stream(), media_typetext/event-stream)到这里一个具备运行 → 暂停 → 用户输入 → 恢复 → 完成能力的整体 Runtime 就搭出来了。前端不管用什么框架只要实现 SSE 客户端解析事件流再按照协议回传用户指令即可。这套方案内部实现了运行和交互的解耦后续换 UI、加多端支持都不用动 Runtime 一层。4. 踩坑实录与排查心得实测中最该防的几个雷4.1 PostgreSQL 连接池与并发写入冲突起初我每创建一个 Runtime 实例就新建一个数据库连接结果压测时发现连接数飙高数据库报too many connections。后来改成用一个全局连接池每个线程从池里取连接。问题是 LangGraph 的PostgresSaver在写入 checkpoint 时需要一个数据库连接多线程并发执行多个 thread_id 的图时连接如果被其他线程占用会阻塞甚至报错。我的最终方案是每个thread_id对应一个独立的PostgresSaver实例共享底层连接池from psycopg_pool import ConnectionPool pool ConnectionPool(conninfopostgresql://postgres:postgreslocalhost:5432/agent_runtime, min_size1, max_size10) def get_saver_for_thread(thread_id: str) - PostgresSaver: conn pool.getconn() return PostgresSaver(conn)注意用完连接一定要归还连接池最好用try/finally包裹否则连接泄露是迟早的事。4.2 LangMem 作为替代方案的经历为什么我没深入用中途我还试过 LangChain 官方出的 LangMem 库它把长期记忆做得更抽象存储的粒度不再是整个状态快照而是按记忆类型分别落库还能用语义检索。它的思路很吸引人例如把用户偏好单独管理Agent 下次启动时只需要检索相关记忆片段而不是加载全部消息历史。我没有深入用下去因为当时团队没有向量数据库的服务端资源而且语义索引和结构化事务存储的数据一致性更难调试。如果你的项目还停留在恢复整个会话上下文这一层PostgreSQL Checkpoint 已经足够只有当你需要做跨会话的记忆抽取时才值得引入 LangMem。但它确实给我提了个醒Checkpoint 存储不只是为了崩溃恢复它也可以作为长期记忆的基础选择持久化方案时最好留一点扩展余地别把路径写死。4.3 模型返回中的流式与中断冲突使用流式输出时有一个容易忽略的问题LangGraph 的astream默认会产生多个类型的流事件包括values、updates以及内部执行的细节。如果你只是想把最终状态推给前端stream_modevalues是最省心的选择它会产出每个节点执行完成后的完整状态快照不会太碎。如果你需要细粒度地展示模型正在生成字这类流式效果那stream_modemessages更合适它会逐 token 产出增量消息。但注意在这种流式模式下interrupt()的暂停行为可能会被吞掉需要前端根据事件类型自行判断。我最终的做法是双通道主通道用values模式做状态同步和恢复附带一个可选的messages模式做打字机效果两个独立 SSE 连接避免互相干扰。另外PostgreSQL 写入的并发问题折腾最久的是版本冲突。LangGraph 内部保存 checkpoint 时带一个版本号如果两个并发请求试图更新同一个 thread_id 的 checkpoint后写入的会收到version_id mismatch错误。这不是程序 bug而是图中thread_id被重复使用造成的竞态。解决方法是一个 thread_id 同一时间只允许一个正在运行的实例操作。如果要支持用户多路操作每个操作路径要分配独立的 thread_id否则就要实现队列串行化。4.4 AG-UI 事件顺序问题AG-UI 协议里事件类型有先后约束比如常有任务开始必须在工具调用之前。我发现 LangGraph 的原生 stream 事件并不天然保证这个顺序有时候节点执行完成的事件还没发出去内部工具产生的增量日志已经先到了。于是我在适配层加了一个简易的事件排序器class EventSequencer: def __init__(self): self.sequence 0 self.pending {} def push(self, event_type: str, data: dict) - dict: self.sequence 1 return {sequence: self.sequence, type: event_type, data: data}前端消费时严格按照sequence递增渲染如果发现跳号可以主动请求reconnect从断点重新拉取事件流。这个机制和 AG-UI 的 resume 语义是兼容的实际线上使用效果比较稳。5. 常见问题速查表与复现步骤5.1 问题速查表症状原因解决方案checkpoint 写入失败未启用 autocommit 或数据库连接数超限使用连接池设置autocommitTrue恢复时状态丢失thread_id不一致或未挂载 checkpointer确认编译时传入checkpointer运行 config 中 thread_id 一致并发运行时版本冲突同一 thread_id 被多个实例同时写入同一 thread_id 串行执行或分配独立 thread_idinterrupt 后无法恢复恢复指令未通过Command(resume...)或 config 错误使用compiled_graph.stream(Command(resume...), config)AG-UI 事件顺序错乱LangGraph 事件流与协议顺序不完全一致适配层增加 sequence 排序模型调用超时导致整个图卡住没有配置超时或重试节点内对模型调用设置超时和有限重试计入状态字段5.2 一个最小可复现的恢复示例我简化出一个最小复现用例你复制下来可以直接跑通整个中断-恢复链路from typing_extensions import TypedDict from langgraph.graph import StateGraph, END from langgraph.checkpoint.postgres import PostgresSaver from langgraph.types import interrupt, Command from psycopg import Connection from langchain_core.messages import HumanMessage, SystemMessage class State(TypedDict): messages: list def step1(state: State): user_input interrupt({question: 请确认是否继续}) return {messages: [SystemMessage(contentf用户确认{user_input})]} def step2(state: State): return {messages: [SystemMessage(content流程完成)]} builder StateGraph(State) builder.add_node(step1, step1) builder.add_node(step2, step2) builder.set_entry_point(step1) builder.add_edge(step1, step2) builder.add_edge(step2, END) conn Connection.connect(postgresql://postgres:postgreslocalhost:5432/agent_runtime, autocommitTrue) saver PostgresSaver(conn) saver.setup() graph builder.compile(checkpointersaver) config {configurable: {thread_id: demo-1}} events list(graph.stream({messages: []}, config, stream_modevalues)) print(第一次执行后状态, events[-1]) # 此时图在 step1 中断等待用户输入 events list(graph.stream(Command(resume继续执行), config, stream_modevalues)) print(恢复后最终状态, events[-1])这个例子虽然简单但把 LangGraph 中断恢复三个要素全部展示出来了interrupt()暂停、Command(resume...)恢复、PostgresSaver持久化。你把第一次执行后的进程杀掉重启再执行恢复命令一样能继续。5.3 关于 Runtime 的部署形态最后说下部署形态。我把这套 LangGraph Runtime 封装成一个独立的服务进程通过 FastAPI 对外提供 HTTP SSE 接口前端和后端业务服务都通过标准协议访问它。Runtime 本身无状态——这不是说它内部没有状态而是它的状态全部在 PostgreSQL 里进程可以随时杀掉重建不影响任何正在进行的会话。容器化部署时Runtime 镜像里只需要 Python 环境、LangGraph 依赖和 FastAPI数据库独立在外。这样的好处是扩容非常方便用户量上来以后直接多起几个 Runtime 副本负载均衡按 thread_id 做一致性哈希路由同一个会话始终打到同一个副本即可。如果副本挂了另一个副本也可以从 checkpoint 恢复该会话不会产生不可恢复的故障。我在实际压测中用 4 个 Runtime 副本 1 个 PostgreSQL 实例撑住了 200 个并发会话的日常负载每个会话平均有 8 轮工具调用。PostgreSQL 的 checkpoint 表增长也还好主要是每个会话每节点一条记录及时清理超期会话就能控制表大小。6. 一点体会Runtime 化的关键是状态思维的转变整套方案从设计到上线前后迭代了三个版本我最大的感触是技术选型反而没那么难最难的是思维方式从写循环到维护状态机的转变。手写 Loop 时你的注意力在下一步做什么用了 LangGraph 后你的注意力必须放在当前状态是什么、怎么持久化、怎么恢复。AG-UI 协议在这一过程中起到了定海神针的作用。没有这套统一交互协议之前每次前端加一个交互形式后端就要改一次接口。有了 AG-UI所有交互都标准化成事件流里的消息和指令前端可以按协议实现任意交互方式后端 Runtime 也只认标准协议。如果你正在纠结要不要从手写 Loop 迁移到这种可恢复 Runtime 架构我的建议是先把最小闭环跑通也就是图执行 → PostgreSQL 保存 checkpoint → 中断 → 恢复 → 继续执行体验一次省心省力的恢复过程你就再也不想回到那个while True的流浪时代了。最后分享一个小技巧调试这类 Runtime 时给每个 checkpoint 打上自定义元数据比如用户 ID、业务类型、当前意图标签排查问题时能省很多力气。PostgreSQL 的 checkpoints 表支持存 JSON 字段我不会只用它存状态快照还会把一些侧面信息塞进去配合查询分析几乎无往不利。