做 Agent 最痛苦的不是模型答错而是跑了一半程序挂了所有上下文都得从头再来。我前几年做 AI Agent 项目时第一版是标准的手写 Loopwhile True里拿 LLM 先生成一轮再解析工具调用跑完工具把结果拼回去继续循环。这套东西在小 Demo 里怎么玩都顺一上生产就原形毕露——多轮对话的状态全堆在内存变量里进程一重启就是失忆遇到需要人工审批的中间环节整个循环要么死等要么靠什么 Redis 临时锁加任务表硬磨。折腾到半夜一两点盯着终端里那一串丢失的对话记录我是真真切切地意识到Agent 不能靠手写循环它需要一个能暂停、能落地存储、能恢复的 Runtime。这篇文章就是记录我怎么从手写 Loop 走出来用 LangGraph 做有状态编排用 PostgreSQL 做 Checkpoint 持久化再用 AG-UI 那一套思路把人工介入与中断恢复接起来的完整过程。适合正在做生产级 Agent 应用、被状态管理和大模型不确定性搞得焦头烂额的人。里面都是我自己踩过的坑和可以直接抄走的代码思路。1. 手写 Loop 的痛点和可恢复 Runtime 的设计目标1.1 手写 Loop 到底输在哪里所谓手写 Loop大概是下面这种结构messages [] while True: response llm.invoke(messages) messages.append(response) if response.tool_calls: for tool_call in response.tool_calls: result execute_tool(tool_call) messages.append(result) else: break这段代码本身没问题问题都在它活不到生产环境的地方。第一是状态维度太单一messages只是模型对话历史但真正有用的状态还包括当前执行到哪一个逻辑分支、哪些中间结果已经算好、哪些外部申请已经拿到、用户上一次给过什么约束条件。这些状态在手写 Loop 里往往是散落在一堆局部变量和全局变量里的你甚至没法完整地、原子地快照它们。第二个问题是循环结构本身对中断极不友好。我这边最典型的一个场景是“生成内容后需要管理员审批”。LLM 已经把内容生成好了但它不能直接发出去系统要停下来等管理员点一下确认。手写 Loop 里这种等待就意味着我要把循环拆开把中间结果存到数据库再写一套轮询任务表来唤醒后续逻辑。每一次拆解都伴随一个新的维护点拆着拆着业务逻辑就散了。但凡中间步骤再叠加一下失败重试、并发触发手写代码瞬间变成一坨谁都不想碰的状态机。还有最扎心的就是进程崩溃。凌晨三点内存被挤爆服务自动重启所有对话上下文清零。用户第二天早上接着问“昨天那个审批通过了吗”系统只能一脸无辜地回答“我不记得了。”这种体验对于 To B 产品来说基本是灾难。1.2 从 Loop 到 Runtime什么叫真正的可恢复我把这套目标拆成了四层图状编排不再用单一循环线性地跑而是把任务分解为节点和边每个节点有明确职责边表达流转条件。流程可以包含分支、并行、循环而不是一杆子到底。全局状态持久化整个执行上下文包括消息、变量、中间产物都要能在任意一个节点边界上落盘。中断点支持在人工审批、外部依赖等待这些位置主动暂停等待外部信号恢复。崩溃恢复进程意外退出后能从最近一次持久化的 Checkpoint 恢复执行而不是从头再来。LangGraph 正好就是这个思路下的框架。它的核心抽象是 StateGraph每个节点接收当前状态、返回状态更新图内部通过边来流转。更重要的是 LangGraph 原生支持 Checkpoint 机制默认用内存存储也可以扩展到 Redis、SQLite、PostgreSQL 这些外部存储上。1.3 为什么我最后还是选了 PostgreSQL 存 CheckpointLangGraph 官方支持的 Checkpoint 后端有好几个我简单说一下我比对过的方案。内存存储MemorySaver最快但进程一重启就没了只适合写单测。SqliteSaver轻量适合单机原型不过并发能力、数据管理能力都弱生产环境下要额外上锁控制。Redis 性能很好但 Agent 的 Checkpoint 数据往往需要查询和审计Redis 数据结构不够直观而且我们公司已经有成套的 PostgreSQL 运维体系再引一个 Redis 反而增加维护成本。PostgreSQL 修正了我对“持久化”的认知Checkpoint 不只是为了恢复还意味着我可以写 SQL 去查这轮对话之前在什么状态执行过哪些工具什么时候人工中断过。这些信息拿来排查线上问题、做数据分析、回放历史会话都极其方便。LangGraph 官方提供了langgraph-checkpoint-postgres库用起来也简单所以最终就敲定了。2. 架构选型LangGraph、PostgreSQL Checkpoint、AG-UI 各自扮演什么角色2.1 LangGraph 和 LangChain 的关系别再傻傻分不清用 LangGraph 之前很多人的第一反应是“这不又是 LangChain 的另一个封装吗”我之前面试也经常被问到 LangChain 和 LangGraph 到底什么区别。这里我统一用一个比喻解释LangChain 是工具箱它给你提供了大量封装好的模型封装、工具、模板和链LangGraph 是流水线调度器它关心的是你这些步骤之间的依赖、状态流转、暂停和恢复。LangChain 里的Runnable和AgentExecutor。AgentExecutor其实也是一个循环实现但它内部是隐式的你想在中间插一个人工审核步骤要 break 内部循环控制粒度非常难受。LangGraph 把执行显式化成图你想在哪里暂停就在哪里interrupt想在哪个节点恢复就从哪个节点继续。这才是它和 LangChain 最大的价值差异。2.2 PostgreSQL 不只是数据库它是 Agent 的“黑匣子”Checkpoint 落到 PostgreSQL 后我看到的直接收益有三个。恢复闭环。LangGraph 会把每个超步结束后节点的状态、写入的 key、相关的 blob 都备份到表里。只要拿到thread_id随时能把整张图恢复到那个位置。审计价值。PG 里的 Checkpoint 表是结构化的我可以查询“这个 Agent 上一轮调度到什么节点”“它调用了哪几个工具输入输出各是什么”“人工什么时候给了什么反馈”。在做合规审计或者用户投诉回溯时这些数据比让用户重新录一遍操作过程靠谱得多。运维的可观测性。Agent 代码 bug 了需要调试直接把库里的 Checkpoint 导出对比不同时间点的状态就能定位是哪一步把关键变量改坏了。这个优势是内存存储绝对给不了的。2.3 AG-UI 是什么它在整个链路里接在哪一层标题里的 AG-UI我理解的是 Agent 与用户之间的交互界面层核心用途是承接“人工介入”。它不是一个单一组件而是一套交互流程把 Agent 当前的推理结果、待确认信息、工具执行记录呈现给用户收集用户的确认、修改或补充指令再把用户的决策送回 Agent 运行时让 Agent 从暂停点继续执行。在我们的架构里LangGraph 负责的是 Agent 内部的调度和状态PG 负责状态持久化AG-UI 则负责外界信号如何安全地注入到运行时。它听起来“面”很大落地时我会把它收敛成一个轻量的后端 API 加一个前端面板后端暴露“获取待审批事项”“提交审批结果”两个接口前端就是简单的表格加按钮。真正核心的衔接点是 LangGraph 的interrupt()函数和Command(resume...)。所有 AG-UI 层收集到的人工决策最终都会变成一次 resume 操作。3. 从手写 Loop 到 LangGraph核心状态图设计与中断点实现3.1 先定义状态再定义节点LangGraph 的图不是从边开始的而是从状态开始的。我那个“内容审核后发送”的场景状态大概是这样的from typing import TypedDict, Annotated, List from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[List[dict], add_messages] draft_content: str approval_result: str send_status: str这里的Annotated[List[dict], add_messages]是对消息字段做一个合并函数的标注节点返回新的消息时不是覆盖而是追加LangGraph 内部会处理重复消息的合并。其他几个字段就是普通的业务状态每个节点都可以读和改。我强烈建议在设计早期就把状态字段定义清楚尤其是哪些是“累计型”字段哪些是“覆盖型”字段。累计型字段要用合并器否则节点并发执行时可能会出现覆盖丢失的问题覆盖型字段则保持简单赋值即可。这个设计直接决定了后续 Checkpoint 里存的数据是不是完整。3.2 节点拆分生成、审批、发送图里我拆了四个节点from langgraph.graph import StateGraph, START, END def generate_content(state): prompt state[messages] response llm.invoke(prompt) return {draft_content: response.content, messages: [response]} def human_approval(state): # 先用 interrupt 暂停等人工结果 decision interrupt({draft_content: state[draft_content]}) return {approval_result: decision} def send_content(state): if state[approval_result].get(approved): notification.send(state[draft_content]) return {send_status: sent} return {send_status: rejected} graph_builder StateGraph(AgentState) graph_builder.add_node(generate_content, generate_content) graph_builder.add_node(human_approval, human_approval) graph_builder.add_node(send_content, send_content) graph_builder.add_edge(START, generate_content) graph_builder.add_edge(generate_content, human_approval) graph_builder.add_edge(human_approval, send_content) graph_builder.add_edge(send_content, END)human_approval是中断关键点。interrupt()会在该节点执行时抛出中断信号LangGraph 会将当前状态保存为 Checkpoint然后把图的执行状态标记为__interrupt__整个并发任务挂起。你可以把它理解为“水位线”图跑到了这里当前状态已经完整落盘但它不会再往下执行了直到外部恢复信号到来。3.3 为什么中断点不能放在节点之间这里有个设计经验值得讲讲中断点应该放在节点内部而不是两个节点之间。如果你试图在边上加条件去拦截那这个条件本身就是图逻辑的一部分一旦外部恢复信号到达你很难区分这次执行是“新触发”还是“恢复触发”。用interrupt()的好处是LangGraph 会把“中断与恢复”当作一个显式的图状态来处理。恢复时它会重新执行包含interrupt()的那个节点并让interrupt()函数返回外部传入的恢复值。这就像你在函数里放了个 return下次有人传参进来函数才继续往下跑。这种语义在手写 Loop 里面是非常难模拟的。4. 用 PostgreSQL 做 Checkpoint安装、配置与核心表结构4.1 环境准备PostgreSQL 安装与连接假设你还没有装 PostgreSQL我这里快速过一遍。Windows 用户建议直接下载官方安装器安装时把pgAdmin和Stack Builder勾上Linux 用户用包管理器# Ubuntu / Debian sudo apt update sudo apt install postgresql postgresql-contrib # 启动服务 sudo systemctl start postgresql sudo systemctl enable postgresql装好之后先用默认的 postgres 用户登录并创建数据库sudo -u postgres psql CREATE DATABASE agent_runtime; CREATE USER agent_user WITH PASSWORD your_strong_password; GRANT ALL PRIVILEGES ON DATABASE agent_runtime TO agent_user;这里有一个易错点GRANT ALL PRIVILEGES ON DATABASE只授了数据库层级的权限但后续 LangGraph 会在库里建表、读写表你还得给这个用户授public schema的操作权限GRANT CREATE, USAGE ON SCHEMA public TO agent_user;不然后面连接时十有八九会碰到permission denied for schema public。4.2 LangGraph 的 PostgreSQL Checkpoint 配置官方库是langgraph-checkpoint-postgres底层用psycopg连接数据库。示例代码from langgraph.checkpoint.postgres import PostgresSaver from langgraph.graph import StateGraph, START, END def build_graph(): builder StateGraph(AgentState) # ... 添加节点和边 ... return builder.compile(checkpointerpostgres_saver)注意compile()里传的是checkpointerpostgres_saver这个参数不能漏。我之前见过有人把PostgresSaver实例化了但忘了传给compile()跑起来一看怎么每次调用状态都不对。原因就是没挂 Checkpoint图退化成每次独立执行。PostgresSaver的连接串格式是标准的postgresql://agent_user:your_strong_passwordlocalhost:5432/agent_runtime我还建议把setup()显式调用一下。LangGraph 官方库内部有自动建表机制但我从来只相信显式初始化postgres_saver PostgresSaver.from_conn_string(CONN_STRING) postgres_saver.setup() # 显式初始化表结构4.3 笔记LangGraph 在 PostgreSQL 里建了哪些表如果你们 DBA 问起你“这个 UI 库在数据库里做了什么”你要能答得上来。LangGraph 的 PG Checkpoint 实现会创建一些内部表不同版本表名略有差异核心是 checkpoint、checkpoint_writes、checkpoint_blobs。checkpoint表保存每一次 Checkpoint 的主记录包括thread_id、checkpoint_id、parent_checkpoint_id、checkpointJSONB 内容、metadata。这里相当于 Agent 执行状态的“墓碑”记录着完整图状态。checkpoint_writes则记录每个节点对状态的具体写入操作也就是每步任务的执行痕迹。checkpoint_blobs存储大块的数据比如长文本和二进制内容。理解这个结构对排查问题很有用。比如你发现恢复后某个变量丢了就可以去checkpoint_writes里看看那个节点当时有没有产生写入记录如果压根没有说明节点在执行过程中提前 return 了LangGraph 就不会写入对应字段。4.4 并发和连接池的注意点生产环境不要每次都新建连接。PostgresSaver.from_conn_string()可以接受连接串但更推荐用连接池方式。psycopg 的AsyncConnectionPool或 SQLAlchemy 的连接池都可以关键在于一个 Agent Runtime 进程里保持一组长连接而不是每次线程进来都去psycopg.connect()。并发写入时 PG 自带事务锁不用担心数据错乱但要注意单次 Checkpoint 写入的数据量。有些 Agent 会把大段文本塞进状态里JSONB 字段再大也能扛不过批量的checkpoint_writes可能会显著拉长单步耗时。我的经验是在状态里只保留必要字段像工具返回的原始大文件内容该存对象存储就存对象存储状态里只放引用。5. 中断恢复实战用 AG-UI 接入人工审批流5.1 AG-UI 的最小闭环设计我们的 AG-UI 不需要做成很复杂的前端框架核心是三条链路待办拉取前端拿到thread_id向后端查这个线程有没有处于__interrupt__状态的 Checkpoint。决策提交前端把用户填写的审批意见提交给后端后端调用 LangGraph 的CommandAPI 向图注入恢复信号。进度跟踪前端轮询或通过 WebSocket 订阅拿到恢复后的最新状态展示给用户。后端我起的是 FastAPI 服务核心接口长这样from fastapi import FastAPI from pydantic import BaseModel from langgraph.types import Command app FastAPI() class ApprovalPayload(BaseModel): thread_id: str approved: bool comment: str app.post(/agent/resume) async def resume_agent(payload: ApprovalPayload): command Command(resume{ approved: payload.approved, comment: payload.comment, }) config {configurable: {thread_id: payload.thread_id}} result await graph.ainvoke(None, config, commandcommand) return {status: resumed, result: result}这里有个关键点await graph.ainvoke(None, config, commandcommand)的第一个参数传None。你可能会奇怪为什么不是传输入因为现在不是从头开始跑而是从 Checkpoint 恢复。LangGraph 根据thread_id找到之前的 Checkpoint然后执行resume指令。如果这里我们重新传了一堆输入LangGraph 会把它当成一个新的图输入来更新状态容易把原来保存的业务状态给污染掉。前端部分我直接用了一个简单 React 页面把待审批内容展示出来。审批按钮会回填approved和comment。这个交互天然适合人工介入场景我重点想强调的是人工审批状态下前端展示的数据应该从 Checkpoint 里读而不是从某个内存缓存里读。否则你展示的可能是陈旧的或已变更的数据。5.2 用 thread_id 管理多会话并发thread_id是整个中断恢复机制的主键。每个独立任务都要有唯一的thread_id比如用户会话 ID、工单号、订单号。启动任务时统一约定config {configurable: {thread_id: ticket-1024}} result await graph.ainvoke({messages: initial_messages}, config)后续所有对该任务的中断恢复、数据查询都是拿着同一个thread_id去操作。两个不同thread_id的图执行是相互隔离的共用一个数据库也没关系LangGraph 层的状态隔离和 PG 行锁能保证互不干扰。如果thread_id设计不当比如把时间戳直接当thread_id那恢复时你就找不到上次跑的那个分支了。我建议业务层分配一个稳定且可读的 ID并在前端 URL 参数、日志、数据库主键中都沿用同一个 ID这样排查问题时链路很顺滑。5.3 时间旅行Checkpoint 带来的隐藏功能PostgreSQL 里存了 CheckpointLangGraph 就支持了一个特别酷的操作时间旅行。你可以指定恢复到一个旧的checkpoint_id而不是最近那个。也就是说如果这次审批内容后来发现不对用户可以回退到上次审批之前重新跑而不是只能继续往下推。config { configurable: { thread_id: ticket-1024, checkpoint_id: specific_checkpoint_uuid } } result await graph.ainvoke(None, config, commandcommand)这个功能很强大但也容易把人绕晕。我自己的经验是时间旅行一定要配合界面层的明确操作入口最好加一个二次确认弹窗因为一旦回退到旧的 Checkpoint之后的执行记录虽然还在库里但逻辑上就不是当前主线了。没有 UI 约束就开放这个能力测试同学很容易创造出各种诡异的状态分支。6. 常见问题与排查技巧实录6.1 问题速查表我这一年多实践下来遇到最多的问题都集中在下面这几类。现象大概率原因排查与解决graph.invoke执行后没有任何输出传入的thread_id对应的 Checkpoint 已经处于中断态但又没有提供Command(resume...)检查graph.get_state(config)是否返回__interrupt__提示permission denied for schema publicPostgreSQL 用户没有 schema 权限执行GRANT CREATE, USAGE ON SCHEMA public TO agent_user表找不到relation does not exist新库没有初始化 Checkpoint 表显式调用postgres_saver.setup()恢复时变量丢失节点返回了None或状态字段名写错查看checkpoint_writes核对节点返回的 key中断后重复执行同一个节点interrupt()返回的值没有被正确处理确认恢复传入的 resume 数据结构与节点内消费结构一致并发大量任务时性能下降没有使用连接池频繁创建 PG 连接换用连接池并开启事务批量写入6.2 断点调试实录一次诡异的“审批结果丢了”说一个真实事故。曾经有个同事在human_approval节点里写代码他写的是def human_approval(state): decision interrupt({draft_content: state[draft_content]}) return {approval_result: decision}从 API 层提交Command(resume{approved: True})后看起来图也恢复继续跑了但最终send_content判断时永远拿不到approved字段。我猜他把decision的类型搞错了。LangGraph 的interrupt()在恢复时返回的是resume参数整体也就是{approved: True, comment: ...}。结果他用decision.get(approved)取的时候没取到因为decision是 dict而state[approval_result]应该存整个 dict。他写的却是一个字符串等于把整个 dict 转成字符串存进去了。后来我用 SQL 查了一下checkpoint_writes看到那一步确实写入了一个字符串马上就定位了。这就是为什么我始终强调要结合 Checkpoint 数据来调试而不是只盯着业务日志。6.3 状态字段“膨胀”的监控还有一种问题不是报错是慢慢变慢。Checkpoint 会把状态整体序列化存进 PG。如果你的 Agent 特别啰嗦每次工具调用都要把完整文件内容塞进 messages即使 PG 是 JSONB处理几万字段的速度也会下滑。我建议给状态大小设个上限超过警戒线就报警。简单一点直接在代码里检查import sys def check_state_size(state): size sys.getsizeof(state) if size 1024 * 1024: logger.warning(state too large: %s bytes, thread_id%s, size, state.get(thread_id))内容特别大时把大的data_url、文件内容抽出去放到对象存储或者单独的表里状态里只保留一个file_id。恢复的时候节点再根据file_id拉取真实内容这样 Checkpoint 永远轻量。7. 最后的经验从“能跑”到“稳如狗”还差的几步写完这个 Runtime 之后我自己的感受是框架选对之后代码量反而变少了。过去手写 Loop 加七七八八的持久化逻辑零零散散写了一千多行现在 LangGraph 加 Checkpoint核心逻辑可能三百行就搞定。但代码量下降不代表生产就稳了还有几件事必须做第一要建立“Checkpoint 恢复演练”的机制。每两周人工制造一次崩溃杀掉 Agent 进程重启服务用之前的thread_id继续任务。不演练你根本不知道哪一步会因为缺依赖而恢复失败。首次恢复数据要用与生产一致的表结构和数据不要拿本地内存库糊弄。第二PostgreSQL 的备份和清理策略要跟上。Checkpoint 数据会累积不做生命周期管理三个月后 PG 表体积会大得吓人。可以根据业务保留最近 N 天数据或者用分区表按日期归档。第三AG-UI 层要留操作权限和审计。人工审批功能上线后外部使用者可能会误操作、甚至恶意操作。每一条审批决策都要记录操作人、操作时间、审批前后状态。这些记录同步写回到 PG 的业务表里跟 Agent 的 Checkpoint 形成两条平行证据链。第四工具调用要有幂等性。恢复执行时LangGraph 会把节点重新执行一遍如果你的工具在第一次执行时已经产生了副作用比如已经发了邮件、扣了库存、调了外部 API恢复时重复执行就会出事。我给所有外部副作用工具都加了幂等键用业务请求 ID 做去重这样即使节点被重新执行外部系统也能识别出这是同一笔操作的重复提交直接返回原结果。做可恢复 Runtime 这件事本质上就是在承认一个事实大模型驱动的 Agent 不可能永远一次跑通。模型可能抽风工具可能超时用户可能反悔进程可能被杀。既然意外是常态那设计上就应该默认意外会发生而不是侥幸地认为“这次应该没事”。LangGraph 加 PostgreSQL Checkpoint 加 AG-UI 这套组合至少让我在面对这些意外时还能从容地打开数据库看一眼中断点给用户一个明确的交代。这份从容就是手写 Loop 永远给不了的。
