使用 MongoDB Atlas 的短期内存

使用 MongoDBSaver 和 MongoDB Atlas 持久化智能体短期记忆。

短期记忆允许您的智能体记住单个线程或对话中的先前交互。本指南展示了如何使用MongoDB Atlas作为持久检查点后端。

MongoDB将对话状态存储为集合中的文档，因此线程可以在进程重启和部署之间继续进行。

需要在对话之间记住信息？使用MongoDB Atlas的长期记忆来存储和回忆不同线程和会话中的数据。

设置

安装

pip

pip install -U langgraph-checkpoint-mongodb pymongo

凭证

设置您的Atlas连接字符串：

import getpass
import os

if not os.environ.get("MONGODB_ATLAS_URI"):
    os.environ["MONGODB_ATLAS_URI"] = getpass.getpass("Enter your MongoDB Atlas URI: ")

如果您希望自动跟踪您的模型调用，设置您的LangSmith API密钥：

if not os.environ.get("LANGSMITH_API_KEY"):
    os.environ["LANGSMITH_API_KEY"] = getpass.getpass("Enter your LangSmith API key: ")
os.environ["LANGSMITH_TRACING"] = "true"

使用

创建一个MongoDBSaver，使用from_conn_string，并将其传递给create_agent。上下文管理器在打开时创建所需的集合和索引。您不会调用单独的setup()方法（与Postgres检查点不同）。

import os

from langchain.agents import create_agent
from langgraph.checkpoint.mongodb import MongoDBSaver

def get_user_info() -> str:
    """Look up information about the current user."""
    return "No user profile on file."

MONGODB_ATLAS_URI = os.environ["MONGODB_ATLAS_URI"]

with MongoDBSaver.from_conn_string(MONGODB_ATLAS_URI) as checkpointer:
    agent = create_agent(
        "anthropic:claude-sonnet-4-6",
        tools=[get_user_info],
        checkpointer=checkpointer,
    )

    thread_config = {"configurable": {"thread_id": "1"}}
    response = agent.invoke(
        {"messages": [{"role": "user", "content": "Hi! My name is Bob."}]},
        thread_config,
    )["messages"][-1].content

    print(response)

    response = agent.invoke(
        {"messages": [{"role": "user", "content": "What's my name?"}]},
        thread_config,
    )["messages"][-1].content

    print(response)

MongoDBSaver还暴露了异步方法，如aget和aput，用于在异步图中使用。没有单独的AsyncMongoDBSaver类。

下一步

• 在短期记忆指南中自定义智能体状态、修剪或总结历史记录，并从工具中访问记忆。将相同的MongoDBSaver传递给checkpointer。
• 查看检查点集成列表checkpointer integrations以获取其他后端。
• 对于跨线程持久化，请参阅MongoDB Atlas的长期记忆。

---

官方原文：src/oss/python/integrations/memory/mongodb-short-term-memory.mdx · 非官方翻译，译文由 AI 生成，以英文原文为准。