可以采用MongoDB作为Agent长久记忆的存储空间。与LangGraph结合时,有两种写法.
uv add langgraph-checkpoint-mongodb
uv add langgraph-store-mongodb
from langgraph.checkpoint.mongodb import MongoDBSaver
from langgraph.store.mongodb import MongoDBStore
from pymongo import MongoClient
# 写法一:
mongouri = f'mongodb://{username}:{password}@127.0.0.1:27017/news?authSource=admin'
with (
MongoDBSaver.from_conn_string(mongouri, db_name="news", collection_name="checkpointer") as checkpointer,
MongoDBStore.from_conn_string(mongouri, db_name="news", collection_name="store", ) as store, # 长期记忆数据存放的集合名
):
# 编译图:挂接检查点 + 长期存储
app = builder.compile(checkpointer=checkpointer, store=store)
# config 中的 thread_id 是检查点的唯一标识(必填)
# 同一个 thread_id 会恢复之前的状态;不同 thread_id 相互隔离
# config = {"configurable": {"thread_id": "thread-001"}}
config = {"configurable": {"thread_id": "thread-001"}, "user_id": "68"}
# invoke: 一次性执行完整图
result = app.invoke({"messages": ["hello"], "count": 0}, config=config)
# 写法二
mongouri = f'mongodb://{username}:{password}@127.0.0.1:27017/graph?authSource=admin'
mongo_client = MongoClient(mongouri)
collection = mongo_client["graph"]["checkpoints"] # 库名["graph"] + 集合名["checkpoints"]
collection2 = mongo_client["graph"]["store"]
checkpointer = MongoDBSaver(collection)
store = MongoDBStore(collection2)
## 查询记忆
for memory in store.search(("memories", "68")): # namespace
print(333, memory.value)
# 333 {'data': '您好,我是刘明'}
# 333 {'data': '您好,刘明!很高兴认识您。请问有什么可以帮您?'}
# 查看历史(检查点记录)
for checkpoint in app.get_state_history(config):
print(563, checkpoint)
# StateSnapshot(values={'messages': ['hello', '来自 node_a,第 0 次调用', '来自 node_b'], 'count': 1}, next=(), config={'configurable': {'thread_id': 'thread-001', 'checkpoint_ns': '', 'checkpoint_id': '1f1c7820-73a7-652d-8002-5b2891609753'}}, metadata={'source': 'loop', 'step': 2, 'parents': {}}, created_at='2026-10-14T03:47:50.614300+00:00', parent_config={'configurable': {'thread_id': 'thread-001', 'checkpoint_ns': '', 'checkpoint_id': '1f1c7820-739b-6022-8001-3bf17713d514'}}, tasks=(), interrupts=())