스토어 (Stores)

스토어 (Stores)

LangGraph의 스토어는 스레드를 넘나드는(cross-thread) 장기 메모리를 제공해요. 스레드 단위로 고정되는 체크포인터의 영속성과는 짝을 이루는 개념이죠.

체크포인터가 한 스레드에 묶인 전체 그래프 상태를 저장하는 반면, 스토어는 어느 스레드에서든 접근할 수 있는 임의의 키-값(key-value) 데이터를 담아요. 다시 말해 단일 대화에 갇히지 않고, 사용자·세션에 걸쳐서 이어져야 하는 정보(기억)를 다룰 때 쓰는 저장소예요.

Info: 에이전트 서버(Agent Server)는 스토어를 자동으로 처리해요. 에이전트 서버를 사용할 때는 스토어를 직접 구현하거나 설정할 필요가 없어요. API가 모든 저장 인프라를 뒤에서 알아서 처리해 줍니다.

기본 사용법

아래 코드는 LangGraph 없이 단독으로 InMemoryStore를 사용하는 모습이에요.

from langgraph.store.memory import InMemoryStore
store = InMemoryStore()

메모리는 tuple로 네임스페이스(namespace)를 매기는데, 아래 예시에서는 (<user_id>, "memories") 형태를 써요. 네임스페이스의 길이는 얼마든지 될 수 있고 무엇이든 나타낼 수 있어요. 반드시 사용자별일 필요는 없습니다.

각 메모리는 특정 속성을 가진 파이썬 클래스(Item)예요. .dict로 변환하면 사전(dictionary)처럼 접근할 수 있죠. Item이 갖는 속성은 다음과 같아요.

LangGraph에서 사용하기

스토어는 체크포인터와 함께 사용할 때 빛을 발해요. 체크포인터는 앞서 이야기한 것처럼 스레드에 상태를 저장하고, 스토어는 스레드를 넘어 접근할 수 있는 임의 정보를 담아두는 역할을 하죠. 그래프를 컴파일할 때 체크포인터와 스토어를 함께 넘겨주면 됩니다.

from dataclasses import dataclass
from langgraph.checkpoint.memory import InMemorySaver

@dataclass
class Context:
    user_id: str

# 스레드(대화)를 활성화하려면 필요해요
checkpointer = InMemorySaver()

# ... 그래프 정의 ...

# 체크포인터와 스토어를 함께 컴파일
builder = StateGraph(MessagesState, context_schema=Context)
# ... 노드와 엣지 추가 ...
graph = builder.compile(checkpointer=checkpointer, store=store)

그리고 그래프를 호출할 때는 이전과 마찬가지로 thread_id를, 그리고 해당 사용자의 메모리 네임스페이스 역할을 하는 user_id를 함께 넘겨줘요.

# 그래프 호출
config = {"configurable": {"thread_id": "1"}}

# 먼저 AI에게 그냥 인사부터 해볼게요
for update in graph.stream(
    {"messages": [{"role": "user", "content": "hi"}]},
    config,
    stream_mode="updates",

어느 노드에서든 스토어와 user_id에 접근하려면 Runtime 객체를 사용하면 돼요. Runtime은 노드 함수의 파라미터로 추가하면 LangGraph가 자동으로 주입해 주죠. 이를 통해 메모리를 저장할 수 있어요:

from langgraph.runtime import Runtime
from dataclasses import dataclass

@dataclass
class Context:
    user_id: str

async def update_memory(state: MessagesState, runtime: Runtime[Context]):

    # 런타임 컨텍스트에서 사용자 ID를 가져와요
    user_id = runtime.context.user_id

    # 메모리에 네임스페이스를 매겨요
    namespace = (user_id, "memories")

    # ... 대화를 분석해 새 메모리를 만들기 ...

    # 새 메모리 ID 생성
    memory_id = str(uuid.uuid4())

    # 새 메모리 저장
    await runtime.store.aput(namespace, memory_id, {"memory": memory})

저장한 메모리는 검색해서 꺼내 쓸 수도 있어요. 가장 최근 메시지를 쿼리로 삼아 관련 메모리를 찾아보죠.

from dataclasses import dataclass
from langgraph.runtime import Runtime

@dataclass
class Context:
    user_id: str

async def call_model(state: MessagesState, runtime: Runtime[Context]):
    # 런타임 컨텍스트에서 사용자 ID를 가져와요
    user_id = runtime.context.user_id

    # 메모리에 네임스페이스를 매겨요
    namespace = (user_id, "memories")

    # 가장 최근 메시지를 기준으로 검색
    memories = await runtime.store.asearch(
        namespace,
        query=state["messages"][-1].content,
        limit=3
    )
    info = "\n".join([d.value["memory"] for d in memories])

    # ... 모델 호출에 메모리 사용 ...

새 스레드에서 그래프를 호출해도 저장된 메모리에 접근할 수 있다는 게 요점이에요.

# 새 스레드에서 그래프 호출
config = {"configurable": {"thread_id": "2"}}

# 다시 인사해볼게요
for update in graph.stream(
    {"messages": [{"role": "user", "content": "hi, tell me about my memories"}]},
    config,

LangSmith를 로컬(예: Studio)이나 호스팅으로 사용할 때는 **기본 스토어(base store)**가 기본으로 제공되므로 그래프 컴파일 시점에 별도로 지정할 필요가 없어요.

다만 **시맨틱 검색(semantic search)**을 활성화하려면 langgraph.json 파일에 인덱싱 설정을 구성해야 해요. 예를 들면 다음과 같습니다.

{
    ...
    "store": {
        "index": {
            "embed": "openai:text-embeddings-3-small",
            "dims": 1536,
            "fields": ["$"]
        }
    }
}

자세한 내용과 설정 옵션은 배포 가이드에서 확인할 수 있어요.

커스텀 스토어 만들기

내장 구현 말고 다른 저장 백엔드를 쓰고 싶다면 BaseStore를 상속해 필요한 메서드를 구현하면 돼요. 내장 InMemoryStore가 가장 단순한 참조 구현이 됩니다.

기본 계약 (Base contract)

다섯 개의 비동기(async) 메서드는 모두 필수예요. 동기(sync) 대응 메서드(put, get, delete, search, list_namespaces)는 선택 사항이지만, 동기 그래프 실행과의 호환성을 위해 권장합니다.

다음 단계