Store
Store (장기 메모리)
LangGraph store는 스레드 간에 걸쳐 있는 장기 메모리(cross-thread long-term memory)를 제공해요. 스레드별 영속성을 담당하는 checkpointer와 서로 보완하는 관계예요.
Store를 쓰면 에이전트가 여러 스레드에 걸쳐 정보를 유지할 수 있어요. 사용자 선호도, 축적된 지식, 그리고 한 번의 대화를 넘어 살아남아야 하는 사실들 같은 것들이죠. 하나의 스레드에 한정된 전체 그래프 상태를 저장하는 checkpointer와 달리, store는 아무 스레드에서나 접근할 수 있는 임의의 키-값 데이터를 담아요.
Agent Server는 store를 자동으로 처리해요 Agent Server를 쓸 때는 store를 직접 구현하거나 설정할 필요가 없어요. API가 모든 스토리지 인프라를 백그라운드에서 처리해 줘요.
InMemoryStore는 개발·테스트에 적합해요. 프로덕션에서는
PostgresStore,MongoDBStore,RedisStore,UpstashStore같은 영구 store를 쓰세요. 모든 구현체는 BaseStore를 확장하는데, 이게 노드 함수 시그니처에서 사용할 타입 어노테이션이에요.
사용 가능한 프로바이더의 전체 목록은 store 통합을 참고해요.
출처: 문서
본문
기본 사용법 (Basic usage)
다음 코드 스니펫은 LangGraph를 쓰지 않고 InMemoryStore를 단독으로 보여줘요.
from langgraph.store.memory import InMemoryStore
store = InMemoryStore()
메모리는 tuple로 네임스페이스가 정해져요. 아래 예시에서는 (<user_id>, "memories")예요. 네임스페이스는 어떤 길이든 될 수 있고 무엇이든 나타낼 수 있어요. 꼭 사용자와 관련될 필요는 없어요.
user_id = "1"
namespace_for_memory = (user_id, "memories")
store.put 메서드로 store의 네임스페이스에 메모리를 저장해요. 위에서 정의한 네임스페이스와 메모리의 키-값 쌍을 지정해요. 키는 단순히 메모리의 고유 식별자(memory_id)고, 값(딕셔너리)이 메모리 자체예요.
memory_id = str(uuid.uuid4())
memory = {"food_preference" : "I like pizza"}
store.put(namespace_for_memory, memory_id, memory)
store.search 메서드로 네임스페이스에서 메모리를 읽어내요. 주어진 사용자의 메모리를 limit 인자(기본 10)까지 목록으로 반환해요. InMemoryStore에서는 항목이 삽입 순서로 반환되므로, 가장 최근 메모리가 목록의 마지막에 있어요. 다른 백엔드는 순서가 다를 수 있어요 (네임스페이스의 항목 나열하기 참고).
memories = store.search(namespace_for_memory)
memories[-1].dict()
{'value': {'food_preference': 'I like pizza'},
'key': '07e0caf4-1631-47b7-b15f-65515d4c1843',
'namespace': ['1', 'memories'],
'created_at': '2024-10-02T17:22:31.590602+00:00',
'updated_at': '2024-10-02T17:22:31.590605+00:00'}
각 메모리는 특정 속성을 가진 Python 클래스(Item)예요. .dict로 변환해 딕셔너리로 접근할 수 있어요.
속성은 다음과 같아요.
value: 이 메모리의 값 (그 자체로 딕셔너리)key: 이 네임스페이스에서 이 메모리의 고유 키namespace: 문자열의 튜플, 이 메모리 타입의 네임스페이스타입은
tuple[str, ...]이지만, JSON으로 변환하면 리스트로 직렬화될 수 있어요. (예:['1', 'memories'])created_at: 이 메모리가 만들어진 타임스탬프updated_at: 이 메모리가 갱신된 타임스탬프
네임스페이스의 항목 나열하기 (Listing items in a namespace)
store.search(또는 비동기 store.asearch)를 query와 filter 없이 호출하면 namespace_prefix 아래 저장된 항목을 limit까지 반환해요. 의미적 순위(semantic ranking)가 필요 없을 때 네임스페이스의 모든 것을 나열하는 데 써요.
# Return up to 100 items stored under ("alice", "memories").
items = store.search(("alice", "memories"), limit=100)
세 가지 동작을 기억하세요.
namespace_prefix는 정확히가 아니라 프리픽스로 매칭돼요.("alice",)는("alice", "memories"),("alice", "preferences")등 아래의 항목도 반환해요. 단일 레벨로 제한하려면 전체 네임스페이스를 넘기거나, 반환된 항목을 클라이언트 쪽에서item.namespace로 필터링하세요.limit를 넘는 결과는 조용히 잘려요. 오버플로 신호가 없어요.limit를 예상 최대값보다 높게 설정하거나offset으로 페이지를 나누세요.- 기본 순서는 store 백엔드에 따라 달라요.
PostgresStore와AsyncPostgresStore는updated_at내림차순(가장 최근 갱신이 먼저)으로 결과를 반환해요.InMemoryStore는 삽입 순서(가장 최근 삽입이 마지막)로 반환해요. 구현체들 간에 특정 순서를 기대하지 마세요. 순서가 중요하면 클라이언트 쪽에서item.updated_at으로 정렬하세요.
큰 네임스페이스를 페이지로 넘기려면:
page_size = 50
offset = 0
while True:
page = store.search(("alice", "memories"), limit=page_size, offset=offset)
if not page:
break
for item in page:
pass
offset += page_size
어떤 네임스페이스가 존재하는지 알아내려면(예: 사용자의 메모리를 나열하기 전에 모든 사용자를 순회), store.list_namespaces 또는 store.alist_namespaces를 사용해요.
# All namespaces that start with ("alice",), truncated to two levels deep.
namespaces = store.list_namespaces(prefix=("alice",), max_depth=2)
의미 검색 (Semantic search)
단순 검색을 넘어, store는 의미 검색(semantic search)도 지원해요. 정확한 일치가 아니라 의미를 기준으로 메모리를 찾을 수 있게 해줘요. 이 기능을 켜려면 임베딩 모델로 store를 구성해요.
from langchain.embeddings import init_embeddings
store = InMemoryStore(
index={
"embed": init_embeddings("openai:text-embedding-3-small"), # Embedding provider
"dims": 1536, # Embedding dimensions
"fields": ["food_preference", "$"] # Fields to embed
}
)
이제 검색할 때 자연어 쿼리로 관련 메모리를 찾을 수 있어요.
# Find memories about food preferences
# (This can be done after putting memories into the store)
memories = store.search(
namespace_for_memory,
query="What does the user like to eat?",
limit=3 # Return top 3 matches
)
메모리의 어느 부분을 임베딩할지는 fields 파라미터를 구성하거나, 메모리를 저장할 때 index 파라미터를 지정해 제어할 수 있어요.
# Store with specific fields to embed
store.put(
namespace_for_memory,
str(uuid.uuid4()),
{
"food_preference": "I love Italian cuisine",
"context": "Discussing dinner plans"
},
index=["food_preference"] # Only embed "food_preferences" field
)
# Store without embedding (still retrievable, but not searchable)
store.put(
namespace_for_memory,
str(uuid.uuid4()),
{"system_info": "Last updated: 2024-01-01"},
index=False
)
LangGraph에서 사용하기 (Using in LangGraph)
store는 checkpointer와 함께 동작해요. 위에서 말한 것처럼 checkpointer는 상태를 스레드에 저장하고, store는 스레드 간 접근을 위해 임의 정보를 저장하게 해줘요. checkpointer와 store를 모두 그래프 컴파일에 넣으면 돼요.
from dataclasses import dataclass
from langgraph.checkpoint.memory import InMemorySaver
@dataclass
class Context:
user_id: str
# We need this because we want to enable threads (conversations)
checkpointer = InMemorySaver()
# ... Define the graph ...
# Compile the graph with the checkpointer and store
builder = StateGraph(MessagesState, context_schema=Context)
# ... add nodes and edges ...
graph = builder.compile(checkpointer=checkpointer, store=store)
그런 다음 전처럼 thread_id로, 그리고 특정 사용자의 메모리 네임스페이스가 되는 user_id와 함께 그래프를 호출해요.
# Invoke the graph
config = {"configurable": {"thread_id": "1"}}
# First let's just say hi to the AI
for update in graph.stream(
{"messages": [{"role": "user", "content": "hi"}]},
config,
stream_mode="updates",
context=Context(user_id="1"),
):
print(update)
Runtime 객체를 사용하면 어느 노드에서든 store와 user_id에 접근할 수 있어요. 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]):
# Get the user id from the runtime context
user_id = runtime.context.user_id
# Namespace the memory
namespace = (user_id, "memories")
# ... Analyze conversation and create a new memory
# Create a new memory ID
memory_id = str(uuid.uuid4())
# We create a new memory
await runtime.store.aput(namespace, memory_id, {"memory": memory})
어느 노드에서든 store에 접근해 store.search 메서드로 메모리를 가져올 수도 있어요. 메모리는 딕셔너리로 변환할 수 있는 객체 목록으로 반환돼요.
memories[-1].dict()
{'value': {'food_preference': 'I like pizza'},
'key': '07e0caf4-1631-47b7-b15f-65515d4c1843',
'namespace': ['1', 'memories'],
'created_at': '2024-10-02T17:22:31.590602+00:00',
'updated_at': '2024-10-02T17:22:31.590605+00:00'}
메모리에 접근해 모델 호출에서 사용할 수 있어요.
from dataclasses import dataclass
from langgraph.runtime import Runtime
@dataclass
class Context:
user_id: str
async def call_model(state: MessagesState, runtime: Runtime[Context]):
# Get the user id from the runtime context
user_id = runtime.context.user_id
# Namespace the memory
namespace = (user_id, "memories")
# Search based on the most recent message
memories = await runtime.store.asearch(
namespace,
query=state["messages"][-1].content,
limit=3
)
info = "\n".join([d.value["memory"] for d in memories])
# ... Use memories in the model call
새 스레드를 만들어도 user_id가 같다면 같은 메모리에 접근할 수 있어요.
# Invoke the graph on a new thread
config = {"configurable": {"thread_id": "2"}}
# Let's say hi again
for update in graph.stream(
{"messages": [{"role": "user", "content": "hi, tell me about my memories"}]},
config,
stream_mode="updates",
context=Context(user_id="1"),
):
print(update)
LangSmith를 로컬(예: Studio)이나 호스팅으로 쓸 때는 기본 store를 추가 설정 없이 쓸 수 있어서 그래프 컴파일 시 지정할 필요가 없어요. 다만 의미 검색을 켜려면 langgraph.json 파일에서 인덱싱 설정을 구성해야는 해요. 예:
{
...
"store": {
"index": {
"embed": "openai:text-embeddings-3-small",
"dims": 1536,
"fields": ["$"]
}
}
}
자세한 내용과 구성 옵션은 배포 가이드를 참고해요.
커스텀 store 만들기 (Build a custom store)
내장 구현체가 아닌 다른 스토리지 백엔드를 쓰려면 BaseStore를 상속하고 필요한 메서드를 구현해요. 내장 InMemoryStore가 가장 간단한 참조 구현체예요.
기본 계약 (Base contract)
다섯 개의 async 메서드가 모두 필요해요. 동기 대응 메서드(put, get, delete, search, list_namespaces)는 선택이지만 동기 그래프 실행과의 호환성을 위해 권장돼요.
| Method | Description |
|---|---|
aput(namespace, key, value, index=None) |
Store or overwrite a single item |
aget(namespace, key) |
Retrieve a single item by key; return None if missing |
adelete(namespace, key) |
Delete a single item |
asearch(namespace_prefix, *, query=None, filter=None, limit=10, offset=0) |
Search items under a namespace prefix; optionally by semantic query |
alist_namespaces(*, prefix=None, suffix=None, max_depth=None, limit=100, offset=0) |
List namespaces matching a prefix/suffix pattern |
구현 전에 정확한 시그니처를 확인해요.
import inspect
from langgraph.store.base import BaseStore
print(inspect.getsource(BaseStore))
네임스페이스 설계 (Namespace design)
네임스페이스는 문자열 튜플이에요. 예: ("user_id", "memories"). store 구현은 다음을 지원해야 해요.
- 프리픽스 매칭:
asearch(("alice",))는("alice",),("alice", "memories")및 다른 하위 네임스페이스 아래의 항목을 반환해요. - 정확한 키 조회:
aget(("alice", "memories"), "some-key")는 O(1) 또는 그에 가까워야 해요.
SQL 백엔드의 공통 스키마:
CREATE TABLE store_items (
namespace TEXT[] NOT NULL,
key TEXT NOT NULL,
value JSONB NOT NULL,
created_at TIMESTAMPTZ DEFAULT now(),
updated_at TIMESTAMPTZ DEFAULT now(),
PRIMARY KEY (namespace, key)
);
CREATE INDEX ON store_items USING gin(namespace);
직렬화 (Serialization)
store 값은 평범한 Python 딕셔너리예요. 특별한 직렬화기가 필요 없어요. json.dumps / json.loads나 JSONB 컬럼으로 직접 직렬화하세요. JSON 직렬화가 불가능한 raw Python 객체는 저장하지 마세요.
의미 검색 지원 (Semantic search support)
백엔드가 벡터 검색을 지원한다면 asearch의 query 파라미터를 구현해요.
query: str | None인자를 받아요.query가None이 아니면 임베딩하고 코사인 유사도로 순위를 매겨요.query가 주어지면 각Item에score필드를 포함해야 해요.
백엔드가 벡터 검색을 지원하지 않는다면 query가 전달될 때 NotImplementedError를 발생시켜요.
테스트 (Testing)
현재 커스텀 store용 적합성 스위트(conformance suite)는 없어요. InMemoryStore를 참조로 테스트해요.
import pytest
from langgraph.store.memory import InMemoryStore
from your_module import YourStore
@pytest.fixture
async def store():
async with YourStore.create() as s:
yield s
@pytest.fixture
def reference():
return InMemoryStore()
async def test_put_and_get(store, reference):
ns = ("test", "ns")
for s in [store, reference]:
await s.aput(ns, "k1", {"val": 1})
item = await s.aget(ns, "k1")
assert item is not None
assert item.value == {"val": 1}
async def test_delete(store, reference):
ns = ("test", "ns")
for s in [store, reference]:
await s.aput(ns, "k1", {"val": 1})
await s.adelete(ns, "k1")
assert await s.aget(ns, "k1") is None
async def test_search_prefix(store, reference):
for s in [store, reference]:
await s.aput(("user", "memories"), "m1", {"text": "likes pizza"})
results = await s.asearch(("user",))
assert any(r.key == "m1" for r in results)
다음 단계 (Next steps)
- Agent Server에 커스텀 store 추가하기 — 구현체 배포
- Checkpointers — 스레드 범위 상태 영속성
더 알아보기 (Learn more)
- store 통합: 사용 가능한 store 프로바이더의 전체 목록을 다뤄요.
- 배포 가이드: 의미 검색 구성 옵션을 다뤄요.
- Checkpointers: 스레드 범위 상태 영속성과 checkpointer를 다뤄요.