MongoDB Atlas로 장기 기억 구현하기

MongoDB Atlas로 장기 기억 구현하기 (Long-term memory with MongoDB Atlas)

장기 기억(long-term memory)은 에이전트가 서로 다른 대화·세션에 걸쳐 정보를 저장하고 불러올 수 있게 해줘요. 이 가이드에서는 MongoDB Atlas를 그 영구 저장소 백엔드로 사용하는 방법을 보여줘요. MongoDB는 기억을 컬렉션의 문서로 저장하는데, Atlas Vector Search 인덱스를 쓰면 같은 저장소에서 저장된 기억에 대해 시맨틱 검색도 할 수 있어요.

출처: 공식문서

설정

설치

스토어 패키지 langgraph-store-mongodb를 설치해요.

자격 증명

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"
    )

툴에서 장기 기억 읽기

툴은 LangGraph가 자동으로 주입하는 runtime 파라미터로 스토어를 읽을 수 있어요:

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으로 스토어에 쓰면서 스레드를 넘어 살아남는 데이터를 저장할 수도 있어요:

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"),
    )

더 알아보기 (Learn more)

장기 기억(Long-term memory) 개념 문서
스토어 통합(Store integrations) 문서
MongoDB Atlas 단기 기억(Short-term memory with MongoDB Atlas)