使用 MongoDB Atlas 的长期记忆

使用 MongoDBStore 和 MongoDB Atlas 持久化智能体的长期记忆。

长期记忆让你的智能体能够跨不同对话和会话存储并回忆信息。本指南展示如何使用 MongoDB Atlas 作为持久化 store 后端。

MongoDB 将记忆作为文档存储在集合中。借助 Atlas Vector Search 索引，同一个 store 可以对已存储的记忆运行语义搜索。

TIP

要深入了解记忆类型和写入记忆的策略，请参阅记忆概念指南。

设置

安装

安装 store 包（langgraph-store-mongodb）：

pip

pip install -U langgraph-store-mongodb pymongo

uv

uv add langgraph-store-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"

用量

创建一个 MongoDBStore 并将其传给 create_agent。from_conn_string 上下文管理器在打开时会创建所需的索引。你无需调用单独的 setup() 方法。

import os

from langchain.agents import create_agent
from langgraph.store.mongodb import MongoDBStore

MONGODB_ATLAS_URI = os.environ["MONGODB_ATLAS_URI"]

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

记忆存储

LangGraph 将记忆存储为按 namespace 和 key 组织的 JSON 文档。命名空间通常包含用户或组织 ID，以便限定记忆的作用范围。

下面的示例配置向量索引，以便 store.search 能够运行语义查询。使用 create_vector_index_config 并将其作为 index_config 传入：

import os
from collections.abc import Sequence

from langgraph.store.mongodb import MongoDBStore, create_vector_index_config

def embed(texts: Sequence[str]) -> list[list[float]]:
    # Replace with an embeddings model or LangChain Embeddings instance
    return [[1.0, 2.0] for _ in texts]

MONGODB_ATLAS_URI = os.environ["MONGODB_ATLAS_URI"]

with MongoDBStore.from_conn_string(
    MONGODB_ATLAS_URI,
    index_config=create_vector_index_config(embed=embed, dims=2),
) as store:
    user_id = "my-user"
    application_context = "chitchat"
    namespace = (user_id, application_context)
    store.put(
        namespace,
        "a-memory",
        {
            "rules": [
                "User likes short, direct language",
                "User only speaks English and Python",
            ],
            "my-key": "my-value",
        },
    )
    item = store.get(namespace, "a-memory")
    items = store.search(
        namespace, filter={"my-key": "my-value"}, query="language preferences"
    )

有关 store 操作的更多信息，请参阅 Stores 指南。

在工具中读取长期记忆

工具可以使用 runtime 参数从 store 读取数据，LangGraph 会自动注入该参数：

import os
from dataclasses import dataclass

from langchain.agents import create_agent
from langchain.tools import ToolRuntime, tool
from langgraph.store.mongodb import MongoDBStore

@dataclass
class Context:
    user_id: str

MONGODB_ATLAS_URI = os.environ["MONGODB_ATLAS_URI"]

with MongoDBStore.from_conn_string(MONGODB_ATLAS_URI) as store:
    store.put(("users",), "user_123", {"name": "John Smith", "language": "English"})

    @tool
    def get_user_info(runtime: ToolRuntime[Context]) -> str:
        """Look up user info."""
        assert runtime.store is not None
        user_info = runtime.store.get(("users",), runtime.context.user_id)
        return str(user_info.value) if user_info else "Unknown user"

    agent = create_agent(
        "anthropic:claude-sonnet-4-6",
        tools=[get_user_info],
        store=store,
        context_schema=Context,
    )

    agent.invoke(
        {"messages": [{"role": "user", "content": "look up user information"}]},
        context=Context(user_id="user_123"),
    )

从工具写入长期记忆

工具也可以使用 runtime.store.put 写入 store，从而持久化数据，使其能够跨线程存续：

import os
from dataclasses import dataclass

from langchain.agents import create_agent
from langchain.tools import ToolRuntime, tool
from langgraph.store.mongodb import MongoDBStore
from typing_extensions import TypedDict

@dataclass
class Context:
    user_id: str

class UserInfo(TypedDict):
    name: str

@tool
def save_user_info(user_info: UserInfo, runtime: ToolRuntime[Context]) -> str:
    """Save user info."""
    assert runtime.store is not None
    runtime.store.put(("users",), runtime.context.user_id, dict(user_info))
    return "Successfully saved user info."

MONGODB_ATLAS_URI = os.environ["MONGODB_ATLAS_URI"]

with MongoDBStore.from_conn_string(MONGODB_ATLAS_URI) as store:
    agent = create_agent(
        "anthropic:claude-sonnet-4-6",
        tools=[save_user_info],
        store=store,
        context_schema=Context,
    )

    agent.invoke(
        {"messages": [{"role": "user", "content": "My name is John Smith"}]},
        context=Context(user_id="user_123"),
    )

另请参阅

• 长期记忆
• 存储集成
• 使用 MongoDB Atlas 的短期记忆

---