如何实现LangGraph历史状态数据在外部数据库（PostgreSQL/MongoDB）的存储与加载？

嗨，我刚好折腾过这个需求！你现在用的内存存储确实只能临时存当前执行的状态，想把历史状态持久化到PostgreSQL或者MongoDB这种外部库，核心就是替换掉默认的内存存储，改用适配外部数据库的Checkpointer实现，下面给你分两种数据库讲具体的操作步骤，都是亲测可行的：

针对PostgreSQL的实现方案 #

LangGraph官方提供了基于SQLAlchemy的Checkpoint实现，能直接对接PostgreSQL：

1. 先装必要的依赖包

pip install langgraph sqlalchemy psycopg2-binary

这里psycopg2-binary是PostgreSQL的Python驱动，sqlalchemy是ORM工具，用来和数据库交互。

2. 初始化PostgreSQL版的Checkpointer

from sqlalchemy import create_engine
from langgraph.checkpoint.sqlalchemy import SQLAlchemyCheckpoint

# 替换成你的PostgreSQL连接信息，格式是：postgresql+psycopg2://用户名:密码@主机:端口/数据库名
engine = create_engine("postgresql+psycopg2://my_user:my_password@localhost:5432/langgraph_state_db")
# 初始化Checkpointer，第一次运行会自动在数据库里创建必要的表（不用手动建表！）
pg_checkpointer = SQLAlchemyCheckpoint(engine)

3. 创建Agent时传入这个Checkpointer
   把你原来代码里的内存Checkpointer换成这个PostgreSQL的实例，以后所有执行状态都会自动存在PostgreSQL里：

order_information_subagent = create_react_agent(
    llm,
    tools=OrderToolset.get_all_tools(),
    name="order_information_subagent",
    prompt=OrderAgentPrompts.order_subagent_prompt,
    state_schema=State,
    checkpointer=pg_checkpointer,  # 用PostgreSQL的Checkpointer
)

4. 加载历史状态
   只要你记得历史会话的thread_id（每个会话的唯一标识，你可以自己生成，比如用户ID加时间戳），就能直接加载恢复：

# 替换成你要恢复的历史会话ID
target_thread_id = "user_123_session_202405201430"
# 用这个Checkpointer创建Agent，然后传入thread_id恢复会话
result = order_information_subagent.invoke(
    {"input": "我要查之前的订单状态"},
    config={"configurable": {"thread_id": target_thread_id}}
)

针对MongoDB的实现方案 #

同样，LangGraph有官方的MongoDB Checkpointer实现，操作流程类似：

1. 安装依赖包

pip install langgraph pymongo

2. 初始化MongoDB版的Checkpointer

from pymongo import MongoClient
from langgraph.checkpoint.mongodb import MongoDBCheckpoint

# 替换成你的MongoDB连接字符串
client = MongoClient("mongodb://my_user:my_password@localhost:27017/")
# 指定存储状态的数据库和集合，不存在会自动创建
db = client["langgraph_state_db"]
checkpoint_collection = db["agent_checkpoints"]
# 初始化Checkpointer
mongo_checkpointer = MongoDBCheckpoint(checkpoint_collection)

3. 创建Agent时传入这个Checkpointer
   和PostgreSQL的步骤一样，替换原来的Checkpointer实例即可：

order_information_subagent = create_react_agent(
    llm,
    tools=OrderToolset.get_all_tools(),
    name="order_information_subagent",
    prompt=OrderAgentPrompts.order_subagent_prompt,
    state_schema=State,
    checkpointer=mongo_checkpointer,  # 用MongoDB的Checkpointer
)

4. 加载历史状态
   和PostgreSQL逻辑完全一致，用thread_id恢复：

target_thread_id = "user_123_session_202405201430"
result = order_information_subagent.invoke(
    {"input": "我要查之前的订单状态"},
    config={"configurable": {"thread_id": target_thread_id}}
)

一些实用注意事项 #

• 第一次使用时，不管是PostgreSQL还是MongoDB，Checkpointer都会自动创建所需的表/集合，不用手动提前准备
• thread_id是区分不同会话的关键，建议用业务相关的唯一标识生成（比如用户ID+会话启动时间），方便后续查找
• 如果你的状态里有自定义的复杂对象，可以在初始化Checkpointer时传入自定义的序列化器，比如用Pydantic来处理
• 记得给数据库用户配置足够的读写权限，避免出现连接或操作失败的情况

内容来源于stack exchange