LangGraph持久化第二部分：PostgresSavers、HITL与生产环境后端

从RAM检查点到持久化后端

第一部分介绍了LangGraph持久化的必要性——状态、检查点、检查点管理器、线程以及InMemorySaver。这一部分则将这些理念转化为可用于生产环境的技术。内存中的保存机制在进程重启时会消失；而真正的智能体需要能够超越Python进程生命周期的存储方案。

用于生产环境持久性的PostgresSaver

数据库位于应用进程之外。无论发生崩溃、重新部署还是机器更换，检查点历史记录都能完好保留。PostgresSaver会将这些检查点写入PostgreSQL而非RAM中。

PostgreSQL是常见的选择，因为它具有持久性、可作为托管云服务广泛提供、适合并发应用服务器使用，并且能够随着流量增加而扩展以应对更多的检查点数据。

请安装专用包：

pip install langgraph-checkpoint-postgres

基本配置步骤：

from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://user:password@localhost:5432/mydb"with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    # This creates the required tables the first time you use it.
    # Only needs to run once per database.
    checkpointer.setup()    graph = builder.compile(checkpointer=checkpointer)    config = {"configurable": {"thread_id": "session-99"}}
    result = graph.invoke(
        {"messages": [{"role": "user", "content": "Hello"}]},
        config=config,
    )

from_conn_string用于创建上下文管理的连接。setup()会一次性创建如checkpoints和checkpoint_blobs这样的表——应将其视为迁移操作，而非每次请求都需执行的步骤。compile(checkpointer=...)的接口与内存模式相同；更换后端并不会重新生成数据结构。

连接池

每次请求都打开新的连接会浪费时间，建议使用已准备好的连接池：

from psycopg_pool import ConnectionPool
from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://user:password@localhost:5432/mydb"with ConnectionPool(DB_URI) as pool:
    checkpointer = PostgresSaver(pool)
    checkpointer.setup()  # one-time setup
    graph = builder.compile(checkpointer=checkpointer)

异步版本

采用FastAPI风格的应用应使用AsyncPostgresSaver：

from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver

async with AsyncPostgresSaver.from_conn_string(DB_URI) as checkpointer:
    await checkpointer.setup()
    graph = builder.compile(checkpointer=checkpointer)

如果手动创建 psycopg 连接，请设置 autocommit=True，否则使用 row_factory=dict_row 可能会导致表无法持久化。thread_id 的长度应保持在255字符以内；UUID是安全的默认选择。

其他后端

SQLite

from langgraph.checkpoint.sqlite import SqliteSaver

with SqliteSaver.from_conn_string("./agent_memory.db") as checkpointer:
    graph = builder.compile(checkpointer=checkpointer)

单个本地文件可在重启后依然存在，适合用于原型设计或单进程脚本。但由于文件锁的存在，它不太适用于多工作线程的Web部署。

Redis

以内存为主，支持可选的磁盘持久化功能——非常适合需要处理大量并发线程的高吞吐量聊天应用。

托管型云 Postgres

RDS、Cloud SQL、Supabase、Neon等服务都使用标准的PostgreSQL协议，因此只要由服务提供商负责备份和系统运行时间保障，PostgresSaver就能正常工作。

自定义检查点工具

若必须使用内部存储，则继承 BaseCheckpointSaver 类并实现 put、get_tuple、list 方法（以及对应的异步版本）。

快速选择指南

本地持久化脚本 → SQLite；多工作进程生产环境 → Postgres/AsyncPostgres；极高并发的检查点处理需求 → Redis；已有托管式 Postgres 环境 → 云托管的 Postgres；强制要求使用内部存储 → 自定义保存器。

五个逐步示例

示例1 — 无持久化（遗忘机制）

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    count: intdef increment(state: State) -> dict:
    return {"count": state["count"] + 1}builder = StateGraph(State)
builder.add_node("increment", increment)
builder.add_edge(START, "increment")
builder.add_edge("increment", END)graph = builder.compile()  # No checkpointer passed in!print(graph.invoke({"count": 0}))  # {'count': 1}
print(graph.invoke({"count": 0}))  # {'count': 1} again — no memory of the last call!

{'count': 1}
{'count': 1}

第二次调用无法看到第一次的记录：既没有检查点信息，也没有线程标识。

示例2 — InMemorySaver 的连续性

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver

class State(TypedDict):
    count: intdef increment(state: State) -> dict:
    return {"count": state["count"] + 1}builder = StateGraph(State)
builder.add_node("increment", increment)
builder.add_edge(START, "increment")
builder.add_edge("increment", END)checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)config = {"configurable": {"thread_id": "counter-1"}}print(graph.invoke({"count": 0}, config))  # {'count': 1}
print(graph.invoke({"count": 0}, config))  # {'count': 2}

{'count': 1}
{'count': 2}

相同的输入数据会从上一次的检查点继续处理，而不会重置。

示例3 — 模拟崩溃后的恢复

from langgraph.checkpoint.memory import InMemorySaver

# Shared storage dictionary simulates a database that survives across "restarts"
shared_storage = {}checkpointer_1 = InMemorySaver()
checkpointer_1.storage = shared_storage  # pretend this is durable storagegraph = builder.compile(checkpointer=checkpointer_1)
config = {"configurable": {"thread_id": "resume-demo"}}graph.invoke({"count": 0}, config)  # count becomes 1
print("Process 'crashes' here...")# Simulate a restart: a brand new checkpointer object, same underlying storage
checkpointer_2 = InMemorySaver()
checkpointer_2.storage = shared_storage
graph_after_restart = builder.compile(checkpointer=checkpointer_2)# Resume using the SAME thread_id
result = graph_after_restart.invoke({"count": 0}, config)
print(result)  # {'count': 2} — continued, did NOT reset to 0

共享进程内的字典可模拟重启；在生产环境中，只需让新进程指向同一个Postgres数据库，再通过thread_id继续运行即可。

示例4 —— 使用interrupt()实现人工干预

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt, Command

class State(TypedDict):
    task: str
    approved: booldef request_approval(state: State) -> dict:
    decision = interrupt(f"Approve this task? '{state['task']}' (yes/no)")
    return {"approved": decision == "yes"}builder = StateGraph(State)
builder.add_node("request_approval", request_approval)
builder.add_edge(START, "request_approval")
builder.add_edge("request_approval", END)checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)config = {"configurable": {"thread_id": "approval-1"}}# Step 1: Start the graph — it will pause at interrupt()
result = graph.invoke({"task": "Send $500 refund", "approved": False}, config)
print(result)  # Contains an '__interrupt__' key with our question# ... hours later, a human reviews and approves ...# Step 2: Resume with the human's answer
final_result = graph.invoke(Command(resume="yes"), config)
print(final_result)  # {'task': 'Send $500 refund', 'approved': True}

interrupt(...)会暂停当前操作并创建检查点。调用方可以在__interrupt__下看到相关数据。之后通过Command(resume=...)继续执行。恢复运行时，LangGraph会从顶部重新执行整个节点——中断后的副作用应被保留，而非在中断之前。

示例5 —— 基于Postgres的聊天机器人

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.postgres import PostgresSaver

class State(TypedDict):
    messages: listdef chatbot_node(state: State) -> dict:
    last_message = state["messages"][-1]
    reply = f"You said: {last_message}"
    return {"messages": state["messages"] + [reply]}builder = StateGraph(State)
builder.add_node("chatbot", chatbot_node)
builder.add_edge(START, "chatbot")
builder.add_edge("chatbot", END)DB_URI = "postgresql://user:password@localhost:5432/mydb"with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()  # run once
    graph = builder.compile(checkpointer=checkpointer)    config = {"configurable": {"thread_id": "user-42"}}
    result = graph.invoke({"messages": ["Hi there"]}, config)
    print(result)

{'messages': ['Hi there', 'You said: Hi there']}

停止脚本后，次日使用相同的线程ID重新启动，由于数据库保存了检查点，对话可以继续进行。

为何HITL依赖于持久化存储

共有三种方式：interrupt()用于携带数据暂停执行，状态存储在内存中时以零CPU占用进行无限等待，以及Command(resume=...)用于继续执行。如果没有检查点机制，暂停等待时便没有地方存储被冻结的状态图。

实际的审批流程——超过阈值的贷款决策、记录病历前的临床建议、法律条款审核、费用超支处理——往往需要等待数分钟到数周。持久化机制使得这种等待成本更低且更安全。

跨代理能力的持久化

长时间的对话、能在崩溃后依然保留的多步骤计划、不能忘记先前结果的工具循环，以及需要分支后再重新连接的 Workflow，全都依赖于可靠的检查点机制。线程ID就是操作员和用户界面用来恢复正确执行流程的标识。

屡次出现的错误

• 在每次请求时都调用setup()函数。

在多个 Gunicorn/Uvicorn 工作进程下使用 SQLite。

thread_id 值过大。

在调用 interrupt() 之前执行不可逆的副作用操作。

恢复运行时忘记传递相同的 thread_id。

误以为内存中的演示环境具有与生产环境相同的稳定性。

值得采用的实践方法

将架构设置作为迁移任务来执行。优先使用线程池。在异步应用中采用异步保存机制。保持 thread_id 的长度较短且稳定。记录不同环境所使用的后端类型。添加能够触发崩溃并恢复运行的烟雾测试。监控检查点表的增长情况及其保留策略。

总结

LangGraph 的生产环境持久化方案基本上是“相同的图 API，不同的持久化存储”。Postgres 是默认的企业级选择；SQLite 和 Redis 则用于填补特定场景的需求；而自定义的持久化机制则用于满足特定的基础设施要求。人工干预与崩溃恢复并非可选功能——正因如此才需要检查点机制。应在早期就连接好后端，这样在代理与用户交互之前就能实现真正的中断恢复功能。

检查点后端的操作手册

应将检查点存储视为应用程序的核心数据。备份计划、恢复演练以及数据保留策略应与客户记录一同被纳入管理范畴。如果一个聊天产品在区域故障转移后无法恢复对话记录，那它实际上并未真正实现数据存储——它只是给了人们希望而已。

容量规划应跟踪每日每个活跃线程的检查点数量、消息量较大的状态下的平均数据块大小，以及高并发场景下的增长情况。在行数还未达到惊人程度之前，大量存储在状态中的工具数据就已经会让 Postgres 的体积变得庞大。建议将大型数据存放在对象存储中，仅在状态中保存指向这些数据的指针。

访问控制非常重要：用于生成检查点的数据库用户不应是拥有 DDL 权限、能够迁移无关架构的用户。应定期更换凭证，尽可能使用云服务提供商支持的 IAM 认证机制，并审计谁有权SELECT对话状态。线程 ID 通常与用户相对应，因此应将数据导出内容视为个人敏感信息加以处理。

在实时系统中将 InMemorySaver 迁移至 Postgres 时，只要接受在处理过程中 RAM 线程会重置一次这一事实，通常就无需使用双写机制。需规划好维护时间窗口，部署新的检查点器，并在恢复后监控错误率。对于蓝/绿部署切换，需确保两个环境指向同一个数据库，否则线程历史记录会出现混乱。

测试建议：在持续集成环境中启动 Postgres（或使用一次性容器），执行一次 setup() 函数，以五种示例场景作为集成测试，并验证第二个进程能够继续执行第一个进程创建的线程。这一跨进程测试能发现的实际生产环境缺陷，比任何内存中的单元测试都要多。

最后，要向值班人员讲解“图表失败”与“检查点无法访问”之间的区别。这两种情况的表现相似——都会引发错误——但解决方法却不同。将应用程序异常与数据库超时区分开来的控制面板能在故障处理过程中节省大量时间。

线程标识与产品用户体验

线程ID不仅仅是数据库键，它们也是产品对象。应刻意将它们与用户会话、支持工单或文档ID关联起来。通过复用这些ID在界面中实现“继续对话”的功能，而生成新ID则用于实现“重新开始”的功能。若将这两种功能混用在一个ID中，就会导致旧有的工具结果不断重现，造成沟通困扰。

对于多设备用户而言，要在不同客户端之间保持线程ID的稳定性，就需要进行基于身份验证的查询。在未确认所有权之前，绝不能接受客户端提供的线程ID。检查点会存储你输入的任何数据，它们不会强制实施租户隔离机制。

带检查点的状态模式演化

State TypedDicts会不断演变。添加可选字段十分简单；而重命名或删除字段则会破坏旧的检查点。建议采用渐进式修改方式，为缺失的键设置默认值，并对具有破坏性的变更进行显式的迁移处理。应在状态中存储schema_version，并在需要时根据该版本调整节点逻辑。

对于大量的消息列表，应当制定相应的截断策略：将较旧的消息汇总到一个字段中，仅保留最近的原始消息，并在截断后测试能否继续处理。无限制的历史记录是导致Postgres检查点表容量激增的最常见原因。

中断用户体验的模式

审批界面应显示中断载荷、相关状态信息以及截止时间。在恢复时将审批者身份存储在状态中。如果审批期限已过，应运行监视机制来取消任务或升级处理，而非让线程永久处于暂停状态。持久化存储会导致长时间等待，产品策略必须对此进行限制。

灾难恢复演练

每季度将备份数据恢复到测试环境，并重新启动已知的线程标识。需确认消息内容、中断状态以及工具残留信息是否一致。同时要记录检查点数据库与主应用数据库的RPO/RTO值。能够记住用户信息的代理系统应具备与邮件系统相同的连续性保障要求。

性能注意事项

连接池的规模应根据并发运行的图处理任务数量来设定，而不仅仅是网页处理器数量。需注意N+1模式，即每个节点都会不必要的写入检查点——应依据LangGraph版本的默认设置来调整检查点的使用频率。对于体积较大的二进制工具输出，应进行压缩或外部存储。

安全注意事项

应对磁盘进行加密，使用TLS连接Postgres数据库，限制工作节点与存储系统之间的网络访问路径；如果工具可能接触到原始密钥，则应在生成检查点之前清除状态中的敏感信息。可考虑使用字段级数据脱敏中间件。合规审查时会提出相关问题，需准备好包含从用户→图结构→检查点→磁盘的数据流示意图作为答复。

这些运营层面的措施能让代码示例真正转化为可稳定运行的系统。虽然API本身较为简单，但配套的规范管理才是实现实际生产环境应用的关键。

将检查点机制与水平自动扩展相结合

Kubernetes将图处理工作节点从两个Pod扩展到二十个时，每个Pod都必须与同一个持久化检查点器通信。如果每次请求都会携带线程ID，那么就无需使用粘性会话机制。温热节点池不应依赖本地的SQLite文件。健康检查应验证数据库连接状态，而不仅仅是进程是否存活。在部署过程中，正在运行的工作节点应在完成操作或中途创建检查点，这样才能确保中断后能够继续运行。可以通过在interrupt()等待期间终止Pod，然后确认能在另一个Pod上继续运行来进行混沌测试。这样的单一实验比十几项单元测试更能验证架构的可靠性。

文档背压问题：如果检查点存储变慢，工作节点应该减轻负载还是排队等待？相比在不同容器间产生不一致结果的静默内存回退机制，更应选择明确的故障处理方式。运维人员绝不能在生产环境中“暂时”切换到 InMemorySaver 来“解决延迟问题”——这种切换会破坏产品的可靠性承诺。

宣布持久化操作完成前的检查清单

确认setup()以迁移模式运行；已配置数据池；异步与同步机制与Web框架相匹配；严格限制线程ID的所有权；已在不同进程间测试中断恢复功能；备份已成功恢复一次；保留策略已有文档记录；个人信息处理流程已过审查；控制面板能区分数据库错误与图表相关错误；运行手册将线程ID与用户支持工具关联起来。当所有项都得到确认后，PostgresSaver（或您选择的后端）就不再只是代码片段——它已成为产品性能保障的一部分。在此之前，请将演示版本视为演示内容，对于长期运行的HITL流程，仍需保持功能标志处于关闭状态。

一次失败的恢复尝试的小故事

有一位值班工程师曾目睹审批流程卡住数小时。图表本身没有问题，检查点密码已更新，但工作人员仍使用旧凭证缓存数据。新的中断请求无法触发，而旧的线程仍在继续读取数据。解决办法是刷新凭证，并在操作手册中添加一条规则：在重启图表容器之前先验证检查点认证是否正常。请将这一经验与 PostgresSaver 的相关代码片段一并保存，以避免下次再次出现故障。持久化记忆还包括操作记忆——那些记录在操作手册中的经验教训。

那次事件也明确了监控策略：应将检查点认证失败的警报单独显示在一个页面上，而非归入通用的 500 错误类别。当内存是核心产品时，数据库路径便属于一级依赖项。有了这样的规范，用 PostgresSaver 替换 InMemorySaver 就不再是一项令人畏惧的迁移工作，而只是从笔记本到服务架构转型过程中的常规升级步骤而已。