메모리 추가 (Add Memory)
메모리 추가 (Add Memory)
AI 애플리케이션은 여러 상호작용에 걸쳐 컨텍스트를 공유하기 위해 메모리가 필요합니다. LangGraph에서는 두 가지 유형의 메모리를 추가할 수 있습니다:
- 에이전트의 상태(state)의 일부로 단기 메모리(short-term memory)를 추가하여 다중 턴 대화(multi-turn conversation)를 가능하게 합니다.
- 세션에 걸쳐 사용자별 또는 애플리케이션 수준의 데이터를 저장하기 위해 장기 메모리(long-term memory)를 추가합니다.
단기 메모리 추가 (Add short-term memory)
단기 메모리(스레드 수준의 영속성(persistence))는 에이전트가 다중 턴 대화를 추적할 수 있게 해줍니다. 단기 메모리를 추가하려면:
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph
checkpointer = InMemorySaver()
builder = StateGraph(...)
graph = builder.compile(checkpointer=checkpointer)
graph.invoke(
{"messages": [{"role": "user", "content": "hi! i am Bob"}]},
{"configurable": {"thread_id": "1"}},
)
프로덕션에서 사용하기 (Use in production)
프로덕션에서는 데이터베이스로 백업된 체크포인터(checkpointer)를 사용하세요:
from langgraph.checkpoint.postgres import PostgresSaver
DB_URI = "postgresql://postgres:***@localhost:5432/postgres?sslmode=disable"
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
builder = StateGraph(...)
graph = builder.compile(checkpointer=checkpointer)
예시: Postgres 체크포인터 사용
pip install -U "psycopg[binary,pool]" langgraph langgraph-checkpoint-postgres
Postgres 체크포인터를 처음 사용할 때는 checkpointer.setup()을 호출해야 합니다.
- 동기(Sync)
- 비동기(Async)
from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.checkpoint.postgres import PostgresSaver
model = init_chat_model(model="claude-haiku-4-5-20251001")
DB_URI = "postgresql://postgres:***@localhost:5432/postgres?sslmode=disable"
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
# checkpointer.setup()
def call_model(state: MessagesState):
response = model.invoke(state["messages"])
return {"messages": response}
builder = StateGraph(MessagesState)
builder.add_node(call_model)
builder.add_edge(START, "call_model")
graph = builder.compile(checkpointer=checkpointer)
config = {
"configurable": {
"thread_id": "1"
}
}
stream = graph.stream_events(
{"messages": [{"role": "user", "content": "hi! I'm bob"}]},
config,
version="v3",
)
for snapshot in stream.values:
print(snapshot)
stream = graph.stream_events(
{"messages": [{"role": "user", "content": "what's my name?"}]},
config,
version="v3",
)
for snapshot in stream.values:
print(snapshot)
from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
model = init_chat_model(model="claude-haiku-4-5-20251001")
DB_URI = "postgresql://postgres:***@localhost:5432/postgres?sslmode=disable"
async with AsyncPostgresSaver.from_conn_string(DB_URI) as checkpointer:
# await checkpointer.setup()
async def call_model(state: MessagesState):
response = await model.ainvoke(state["messages"])
return {"messages": response}
builder = StateGraph(MessagesState)
builder.add_node(call_model)
builder.add_edge(START, "call_model")
graph = builder.compile(checkpointer=checkpointer)
config = {
"configurable": {
"thread_id": "1"
}
}
stream = await graph.astream_events(
{"messages": [{"role": "user", "content": "hi! I'm bob"}]},
config,
version="v3",
)
async for message in stream.messages:
async for token in message.text:
print(token, end="", flush=True)
stream = await graph.astream_events(
{"messages": [{"role": "user", "content": "what's my name?"}]},
config,
version="v3",
)
async for message in stream.messages:
async for token in message.text:
print(token, end="", flush=True)
예시: MongoDB 체크포인터 사용
pip install -U pymongo langgraph langgraph-checkpoint-mongodb
설정 (Setup) MongoDB 체크포인터를 사용하려면 MongoDB 클러스터가 필요합니다. 이미 클러스터가 없다면 이 가이드를 따라 클러스터를 만드세요. 에이전트 중심의 워크스루는 MongoDB Atlas를 사용한 단기 메모리를 참조하세요.
- 동기(Sync)
- 비동기(Async)
from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.checkpoint.mongodb import MongoDBSaver
model = init_chat_model(model="claude-haiku-4-5-20251001")
MONGODB_URI = "localhost:27017"
with MongoDBSaver.from_conn_string(MONGODB_URI) as checkpointer:
def call_model(state: MessagesState):
response = model.invoke(state["messages"])
return {"messages": response}
builder = StateGraph(MessagesState)
builder.add_node(call_model)
builder.add_edge(START, "call_model")
graph = builder.compile(checkpointer=checkpointer)
config = {
"configurable": {
"thread_id": "1"
}
}
stream = graph.stream_events(
{"messages": [{"role": "user", "content": "hi! I'm bob"}]},
config,
version="v3",
)
for snapshot in stream.values:
print(snapshot)
stream = graph.stream_events(
{"messages": [{"role": "user", "content": "what's my name?"}]},
config,
version="v3",
)
for snapshot in stream.values:
print(snapshot)
from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.checkpoint.mongodb.aio import AsyncMongoDBSaver
model = init_chat_model(model="claude-haiku-4-5-20251001")
MONGODB_URI = "localhost:27017"
async with AsyncMongoDBSaver.from_conn_string(MONGODB_URI) as checkpointer:
async def call_model(state: MessagesState):
response = await model.ainvoke(state["messages"])
return {"messages": response}
builder = StateGraph(MessagesState)
builder.add_node(call_model)
builder.add_edge(START, "call_model")
graph = builder.compile(checkpointer=checkpointer)
config = {
"configurable": {
"thread_id": "1"
}
}
stream = await graph.astream_events(
{"messages": [{"role": "user", "content": "hi! I'm bob"}]},
config,
version="v3",
)
async for message in stream.messages:
async for token in message.text:
print(token, end="", flush=True)
stream = await graph.astream_events(
{"messages": [{"role": "user", "content": "what's my name?"}]},
config,
version="v3",
)
async for message in stream.messages:
async for token in message.text:
print(token, end="", flush=True)
예시: Redis 체크포인터 사용
pip install -U langgraph langgraph-checkpoint-redis
Redis 체크포인터를 처음 사용할 때는 checkpointer.setup()을 호출해야 합니다.
- 동기(Sync)
- 비동기(Async)
from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.checkpoint.redis import RedisSaver
model = init_chat_model(model="claude-haiku-4-5-20251001")
DB_URI = "redis://localhost:6379"
with RedisSaver.from_conn_string(DB_URI) as checkpointer:
# checkpointer.setup()
def call_model(state: MessagesState):
response = model.invoke(state["messages"])
return {"messages": response}
builder = StateGraph(MessagesState)
builder.add_node(call_model)
builder.add_edge(START, "call_model")
graph = builder.compile(checkpointer=checkpointer)
config = {
"configurable": {
"thread_id": "1"
}
}
stream = graph.stream_events(
{"messages": [{"role": "user", "content": "hi! I'm bob"}]},
config,
version="v3",
)
for snapshot in stream.values:
print(snapshot)
stream = graph.stream_events(
{"messages": [{"role": "user", "content": "what's my name?"}]},
config,
version="v3",
)
for snapshot in stream.values:
print(snapshot)
from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.checkpoint.redis.aio import AsyncRedisSaver
model = init_chat_model(model="claude-haiku-4-5-20251001")
DB_URI = "redis://localhost:6379"
async with AsyncRedisSaver.from_conn_string(DB_URI) as checkpointer:
# await checkpointer.asetup()
async def call_model(state: MessagesState):
response = await model.ainvoke(state["messages"])
return {"messages": response}
builder = StateGraph(MessagesState)
builder.add_node(call_model)
builder.add_edge(START, "call_model")
graph = builder.compile(checkpointer=checkpointer)
config = {
"configurable": {
"thread_id": "1"
}
}
stream = await graph.astream_events(
{"messages": [{"role": "user", "content": "hi! I'm bob"}]},
config,
version="v3",
)
async for message in stream.messages:
async for token in message.text:
print(token, end="", flush=True)
stream = await graph.astream_events(
{"messages": [{"role": "user", "content": "what's my name?"}]},
config,
version="v3",
)
async for message in stream.messages:
async for token in message.text:
print(token, end="", flush=True)
예시: Oracle 체크포인터 사용
pip install -U langgraph langgraph-oracledb
설정 (Setup)
Oracle 체크포인터를 사용하려면 Oracle AI Database 인스턴스가 필요합니다. 로컬 컨테이너(예: gvenzl/oracle-free:23-slim) 또는 OCI의 Oracle Autonomous Database가 모두 작동합니다.
Oracle 체크포인터를 처음 사용할 때는 checkpointer.setup()을 호출해야 합니다.
- 동기(Sync)
- 비동기(Async)
from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, MessagesState, START
from langgraph_oracledb.checkpoint.oracle import OracleSaver
model = init_chat_model(model="claude-haiku-4-5-20251001")
DB_URI = "user/password@localhost:1521/FREEPDB1"
with OracleSaver.from_conn_string(DB_URI) as checkpointer:
# checkpointer.setup()
def call_model(state: MessagesState):
response = model.invoke(state["messages"])
return {"messages": response}
builder = StateGraph(MessagesState)
builder.add_node(call_model)
builder.add_edge(START, "call_model")
graph = builder.compile(checkpointer=checkpointer)
config = {
"configurable": {
"thread_id": "1"
}
}
stream = graph.stream_events(
{"messages": [{"role": "user", "content": "hi! I'm bob"}]},
config,
version="v3",
)
for snapshot in stream.values:
print(snapshot)
stream = graph.stream_events(
{"messages": [{"role": "user", "content": "what's my name?"}]},
config,
version="v3",
)
for snapshot in stream.values:
print(snapshot)
from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, MessagesState, START
from langgraph_oracledb.checkpoint.oracle import AsyncOracleSaver
model = init_chat_model(model="claude-haiku-4-5-20251001")
DB_URI = "user/password@localhost:1521/FREEPDB1"
async with AsyncOracleSaver.from_conn_string(DB_URI) as checkpointer:
# await checkpointer.setup()
async def call_model(state: MessagesState):
response = await model.ainvoke(state["messages"])
return {"messages": response}
builder = StateGraph(MessagesState)
builder.add_node(call_model)
builder.add_edge(START, "call_model")
graph = builder.compile(checkpointer=checkpointer)
config = {
"configurable": {
"thread_id": "1"
}
}
stream = await graph.astream_events(
{"messages": [{"role": "user", "content": "hi! I'm bob"}]},
config,
version="v3",
)
async for message in stream.messages:
async for token in message.text:
print(token, end="", flush=True)
stream = await graph.astream_events(
{"messages": [{"role": "user", "content": "what's my name?"}]},
config,
version="v3",
)
async for message in stream.messages:
async for token in message.text:
print(token, end="", flush=True)
서브그래프에서 사용하기 (Use in subgraphs)
그래프에 서브그래프(subgraphs)가 포함된 경우, 부모 그래프를 컴파일할 때만 체크포인터를 제공하면 됩니다. LangGraph는 자동으로 체크포인터를 하위 서브그래프로 전파합니다.
from langgraph.graph import START, StateGraph
from langgraph.checkpoint.memory import InMemorySaver
from typing import TypedDict
class State(TypedDict):
foo: str
# Subgraph
def subgraph_node_1(state: State):
return {"foo": state["foo"] + "bar"}
subgraph_builder = StateGraph(State)
subgraph_builder.add_node(subgraph_node_1)
subgraph_builder.add_edge(START, "subgraph_node_1")
subgraph = subgraph_builder.compile()
# Parent graph
builder = StateGraph(State)
builder.add_node("node_1", subgraph)
builder.add_edge(START, "node_1")
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)
서브그래프별 체크포인팅 동작을 구성할 수 있습니다. interrupts 지원과 상태 저장 연속(stateful continuation)을 포함한 영속성 수준에 대한 자세한 내용은 서브그래프 영속성을 참조하세요.
subgraph_builder = StateGraph(...)
subgraph = subgraph_builder.compile(checkpointer=True)
장기 메모리 추가 (Add long-term memory)
대화에 걸쳐 사용자별 또는 애플리케이션별 데이터를 저장하려면 장기 메모리를 사용하세요.
from langgraph.store.memory import InMemoryStore
from langgraph.graph import StateGraph
store = InMemoryStore()
builder = StateGraph(...)
graph = builder.compile(store=store)
노드 내부에서 스토어에 접근하기 (Access the store inside nodes)
store로 그래프를 컴파일하면 LangGraph는 자동으로 store를 노드 함수에 주입합니다. store에 접근하는 권장 방법은 Runtime 객체를 통하는 것입니다.
from dataclasses import dataclass
from langgraph.runtime import Runtime
from langgraph.graph import StateGraph, MessagesState, START
import uuid
@dataclass
class Context:
user_id: str
async def call_model(state: MessagesState, runtime: Runtime[Context]):
user_id = runtime.context.user_id
namespace = (user_id, "memories")
# Search for relevant memories
memories = await runtime.store.asearch(
namespace, query=state["messages"][-1].content, limit=3
)
info = "\n".join([d.value["data"] for d in memories])
# ... Use memories in model call
# Store a new memory
await runtime.store.aput(
namespace, str(uuid.uuid4()), {"data": "User prefers dark mode"}
)
builder = StateGraph(MessagesState, context_schema=Context)
builder.add_node(call_model)
builder.add_edge(START, "call_model")
graph = builder.compile(store=store)
# Pass context at invocation time
graph.invoke(
{"messages": [{"role": "user", "content": "hi"}]},
{"configurable": {"thread_id": "1"}},
context=Context(user_id="1"),
)
프로덕션에서 사용하기 (Use in production)
프로덕션에서는 데이터베이스로 백업된 store을 사용하세요:
from langgraph.store.postgres import PostgresStore
DB_URI = "postgresql://postgres:***@localhost:5432/postgres?sslmode=disable"
with PostgresStore.from_conn_string(DB_URI) as store:
builder = StateGraph(...)
graph = builder.compile(store=store)
예시: Postgres store 사용
pip install -U "psycopg[binary,pool]" langgraph langgraph-checkpoint-postgres
Postgres store를 처음 사용할 때는 store.setup()을 호출해야 합니다.
- 비동기(Async)
- 동기(Sync)
from dataclasses import dataclass
from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
from langgraph.store.postgres.aio import AsyncPostgresStore
from langgraph.runtime import Runtime
import uuid
model = init_chat_model(model="claude-haiku-4-5-20251001")
@dataclass
class Context:
user_id: str
async def call_model(
state: MessagesState,
runtime: Runtime[Context],
):
user_id = runtime.context.user_id
namespace = ("memories", user_id)
memories = await runtime.store.asearch(namespace, query=str(state["messages"][-1].content))
info = "\n".join([d.value["data"] for d in memories])
system_msg = f"You are a helpful assistant talking to the user. User info: {info}"
# Store new memories if the user asks the model to remember
last_message = state["messages"][-1]
if "remember" in last_message.content.lower():
memory = "User name is Bob"
await runtime.store.aput(namespace, str(uuid.uuid4()), {"data": memory})
response = await model.ainvoke(
[{"role": "system", "content": system_msg}] + state["messages"]
)
return {"messages": response}
DB_URI = "postgresql://postgres:***@localhost:5432/postgres?sslmode=disable"
async with (
AsyncPostgresStore.from_conn_string(DB_URI) as store,
AsyncPostgresSaver.from_conn_string(DB_URI) as checkpointer,
):
# await store.setup()
# await checkpointer.setup()
builder = StateGraph(MessagesState, context_schema=Context)
builder.add_node(call_model)
builder.add_edge(START, "call_model")
graph = builder.compile(
checkpointer=checkpointer,
store=store,
)
config = {"configurable": {"thread_id": "1"}}
stream = await graph.astream_events(
{"messages": [{"role": "user", "content": "Hi! Remember: my name is Bob"}]},
config,
version="v3",
context=Context(user_id="1"),
)
async for message in stream.messages:
async for token in message.text:
print(token, end="", flush=True)
config = {"configurable": {"thread_id": "2"}}
stream = await graph.astream_events(
{"messages": [{"role": "user", "content": "what is my name?"}]},
config,
version="v3",
context=Context(user_id="1"),
)
async for message in stream.messages:
async for token in message.text:
print(token, end="", flush=True)
from dataclasses import dataclass
from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.store.postgres import PostgresStore
from langgraph.runtime import Runtime
import uuid
model = init_chat_model(model="claude-haiku-4-5-20251001")
@dataclass
class Context:
user_id: str
def call_model(
state: MessagesState,
runtime: Runtime[Context],
):
user_id = runtime.context.user_id
namespace = ("memories", user_id)
memories = runtime.store.search(namespace, query=str(state["messages"][-1].content))
info = "\n".join([d.value["data"] for d in memories])
system_msg = f"You are a helpful assistant talking to the user. User info: {info}"
# Store new memories if the user asks the model to remember
last_message = state["messages"][-1]
if "remember" in last_message.content.lower():
memory = "User name is Bob"
runtime.store.put(namespace, str(uuid.uuid4()), {"data": memory})
response = model.invoke(
[{"role": "system", "content": system_msg}] + state["messages"]
)
return {"messages": response}
DB_URI = "postgresql://postgres:***@localhost:5432/postgres?sslmode=disable"
with (
PostgresStore.from_conn_string(DB_URI) as store,
PostgresSaver.from_conn_string(DB_URI) as checkpointer,
):
# store.setup()
# checkpointer.setup()
builder = StateGraph(MessagesState, context_schema=Context)
builder.add_node(call_model)
builder.add_edge(START, "call_model")
graph = builder.compile(
checkpointer=checkpointer,
store=store,
)
config = {"configurable": {"thread_id": "1"}}
stream = graph.stream_events(
{"messages": [{"role": "user", "content": "Hi! Remember: my name is Bob"}]},
config,
version="v3",
context=Context(user_id="1"),
)
for snapshot in stream.values:
print(snapshot)
config = {"configurable": {"thread_id": "2"}}
stream = graph.stream_events(
{"messages": [{"role": "user", "content": "what is my name?"}]},
config,
version="v3",
context=Context(user_id="1"),
)
for snapshot in stream.values:
print(snapshot)
예시: MongoDB store 사용
예시: Redis store 사용
pip install -U langgraph langgraph-checkpoint-redis
Redis store를 처음 사용할 때는 store.setup()을 호출해야 합니다.
- 비동기(Async)
- 동기(Sync)
from dataclasses import dataclass
from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.checkpoint.redis.aio import AsyncRedisSaver
from langgraph.store.redis.aio import AsyncRedisStore
from langgraph.runtime import Runtime
import uuid
model = init_chat_model(model="claude-haiku-4-5-20251001")
@dataclass
class Context:
user_id: str
async def call_model(
state: MessagesState,
runtime: Runtime[Context],
):
user_id = runtime.context.user_id
namespace = ("memories", user_id)
memories = await runtime.store.asearch(namespace, query=str(state["messages"][-1].content))
info = "\n".join([d.value["data"] for d in memories])
system_msg = f"You are a helpful assistant talking to the user. User info: {info}"
# Store new memories if the user asks the model to remember
last_message = state["messages"][-1]
if "remember" in last_message.content.lower():
memory = "User name is Bob"
await runtime.store.aput(namespace, str(uuid.uuid4()), {"data": memory})
response = await model.ainvoke(
[{"role": "system", "content": system_msg}] + state["messages"]
)
return {"messages": response}
DB_URI = "redis://localhost:6379"
async with (
AsyncRedisStore.from_conn_string(DB_URI) as store,
AsyncRedisSaver.from_conn_string(DB_URI) as checkpointer,
):
# await store.setup()
# await checkpointer.asetup()
builder = StateGraph(MessagesState, context_schema=Context)
builder.add_node(call_model)
builder.add_edge(START, "call_model")
graph = builder.compile(
checkpointer=checkpointer,
store=store,
)
config = {"configurable": {"thread_id": "1"}}
stream = await graph.astream_events(
{"messages": [{"role": "user", "content": "Hi! Remember: my name is Bob"}]},
config,
version="v3",
context=Context(user_id="1"),
)
async for snapshot in stream.values:
snapshot["messages"][-1].pretty_print()
config = {"configurable": {"thread_id": "2"}}
stream = await graph.astream_events(
{"messages": [{"role": "user", "content": "what is my name?"}]},
config,
version="v3",
context=Context(user_id="1"),
)
async for snapshot in stream.values:
snapshot["messages"][-1].pretty_print()
from dataclasses import dataclass
from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.checkpoint.redis import RedisSaver
from langgraph.store.redis import RedisStore
from langgraph.runtime import Runtime
import uuid
model = init_chat_model(model="claude-haiku-4-5-20251001")
@dataclass
class Context:
user_id: str
def call_model(
state: MessagesState,
runtime: Runtime[Context],
):
user_id = runtime.context.user_id
namespace = ("memories", user_id)
memories = runtime.store.search(namespace, query=str(state["messages"][-1].content))
info = "\n".join([d.value["data"] for d in memories])
system_msg = f"You are a helpful assistant talking to the user. User info: {info}"
# Store new memories if the user asks the model to remember
last_message = state["messages"][-1]
if "remember" in last_message.content.lower():
memory = "User name is Bob"
runtime.store.put(namespace, str(uuid.uuid4()), {"data": memory})
response = model.invoke(
[{"role": "system", "content": system_msg}] + state["messages"]
)
return {"messages": response}
DB_URI = "redis://localhost:6379"
with (
RedisStore.from_conn_string(DB_URI) as store,
RedisSaver.from_conn_string(DB_URI) as checkpointer,
):
store.setup()
checkpointer.setup()
builder = StateGraph(MessagesState, context_schema=Context)
builder.add_node(call_model)
builder.add_edge(START, "call_model")
graph = builder.compile(
checkpointer=checkpointer,
store=store,
)
config = {"configurable": {"thread_id": "1"}}
stream = graph.stream_events(
{"messages": [{"role": "user", "content": "Hi! Remember: my name is Bob"}]},
config,
version="v3",
context=Context(user_id="1"),
)
for snapshot in stream.values:
snapshot["messages"][-1].pretty_print()
config = {"configurable": {"thread_id": "2"}}
stream = graph.stream_events(
{"messages": [{"role": "user", "content": "what is my name?"}]},
config,
version="v3",
context=Context(user_id="1"),
)
for snapshot in stream.values:
snapshot["messages"][-1].pretty_print()
예시: Oracle store 사용
pip install -U langgraph langgraph-oracledb langchain-openai
설정 (Setup)
Oracle store를 사용하려면 Oracle AI Database 인스턴스가 필요합니다. 의미 기반 search에 사용되는 벡터 인덱스는 Oracle AI Vector Search가 필요합니다.
Oracle store와 체크포인터를 처음 사용할 때는 store.setup()과 checkpointer.setup()을 호출해야 합니다.
- 동기(Sync)
- 비동기(Async)
import uuid
from langchain.chat_models import init_chat_model
from langchain.embeddings import init_embeddings
from langchain_core.runnables import RunnableConfig
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.store.base import BaseStore
from langgraph_oracledb.checkpoint.oracle import OracleSaver
from langgraph_oracledb.store.oracle import OracleStore
model = init_chat_model(model="claude-haiku-4-5-20251001")
embeddings = init_embeddings("openai:text-embedding-3-small")
DB_URI = "user/password@localhost:1521/FREEPDB1"
with (
OracleStore.from_conn_string(
DB_URI,
index={"embed": embeddings, "dims": 1536},
) as store,
OracleSaver.from_conn_string(DB_URI) as checkpointer,
):
store.setup()
checkpointer.setup()
def call_model(
state: MessagesState,
config: RunnableConfig,
*,
store: BaseStore,
):
user_id = config["configurable"]["user_id"]
namespace = ("memories", user_id)
memories = store.search(namespace, query=str(state["messages"][-1].content))
info = "\n".join([d.value["data"] for d in memories])
system_msg = f"You are a helpful assistant talking to the user. User info: {info}"
# Store new memories if the user asks the model to remember
last_message = state["messages"][-1]
if "remember" in last_message.content.lower():
memory = "User name is Bob"
store.put(namespace, str(uuid.uuid4()), {"data": memory})
response = model.invoke(
[{"role": "system", "content": system_msg}] + state["messages"]
)
return {"messages": response}
builder = StateGraph(MessagesState)
builder.add_node(call_model)
builder.add_edge(START, "call_model")
graph = builder.compile(
checkpointer=checkpointer,
store=store,
)
config = {
"configurable": {
"thread_id": "1",
"user_id": "1",
}
}
stream = graph.stream_events(
{"messages": [{"role": "user", "content": "Hi! Remember: my name is Bob"}]},
config,
version="v3",
)
for snapshot in stream.values:
snapshot["messages"][-1].pretty_print()
config = {
"configurable": {
"thread_id": "2",
"user_id": "1",
}
}
stream = graph.stream_events(
{"messages": [{"role": "user", "content": "what is my name?"}]},
config,
version="v3",
)
for snapshot in stream.values:
snapshot["messages"][-1].pretty_print()
import uuid
from langchain.chat_models import init_chat_model
from langchain.embeddings import init_embeddings
from langchain_core.runnables import RunnableConfig
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.store.base import BaseStore
from langgraph_oracledb.checkpoint.oracle import AsyncOracleSaver
from langgraph_oracledb.store.oracle import AsyncOracleStore
model = init_chat_model(model="claude-haiku-4-5-20251001")
embeddings = init_embeddings("openai:text-embedding-3-small")
DB_URI = "user/password@localhost:1521/FREEPDB1"
async with (
AsyncOracleStore.from_conn_string(
DB_URI,
index={"embed": embeddings, "dims": 1536},
) as store,
AsyncOracleSaver.from_conn_string(DB_URI) as checkpointer,
):
await store.setup()
await checkpointer.setup()
async def call_model(
state: MessagesState,
config: RunnableConfig,
*,
store: BaseStore,
):
user_id = config["configurable"]["user_id"]
namespace = ("memories", user_id)
memories = await store.asearch(namespace, query=str(state["messages"][-1].content))
info = "\n".join([d.value["data"] for d in memories])
system_msg = f"You are a helpful assistant talking to the user. User info: {info}"
# Store new memories if the user asks the model to remember
last_message = state["messages"][-1]
if "remember" in last_message.content.lower():
memory = "User name is Bob"
await store.aput(namespace, str(uuid.uuid4()), {"data": memory})
response = await model.ainvoke(
[{"role": "system", "content": system_msg}] + state["messages"]
)
return {"messages": response}
builder = StateGraph(MessagesState)
builder.add_node(call_model)
builder.add_edge(START, "call_model")
graph = builder.compile(
checkpointer=checkpointer,
store=store,
)
config = {
"configurable": {
"thread_id": "1",
"user_id": "1",
}
}
stream = await graph.astream_events(
{"messages": [{"role": "user", "content": "Hi! Remember: my name is Bob"}]},
config,
version="v3",
)
async for snapshot in stream.values:
snapshot["messages"][-1].pretty_print()
config = {
"configurable": {
"thread_id": "2",
"user_id": "1",
}
}
stream = await graph.astream_events(
{"messages": [{"role": "user", "content": "what is my name?"}]},
config,
version="v3",
)
async for snapshot in stream.values:
snapshot["messages"][-1].pretty_print()
의미 검색 사용하기 (Use semantic search)
그래프의 메모리 store에서 의미 검색(semantic search)을 활성화하면 그래프 에이전트가 의미적 유사성(semantic similarity)을 기준으로 store에서 항목을 검색할 수 있습니다.
from langchain.embeddings import init_embeddings
from langgraph.store.memory import InMemoryStore
# Create store with semantic search enabled
embeddings = init_embeddings("openai:text-embedding-3-small")
store = InMemoryStore(
index={
"embed": embeddings,
"dims": 1536,
}
)
store.put(("user_123", "memories"), "1", {"text": "I love pizza"})
store.put(("user_123", "memories"), "2", {"text": "I am a plumber"})
items = store.search(
("user_123", "memories"), query="I'm hungry", limit=1
)
의미 검색을 사용한 장기 메모리
from langchain.embeddings import init_embeddings
from langchain.chat_models import init_chat_model
from langgraph.store.memory import InMemoryStore
from langgraph.graph import START, MessagesState, StateGraph
from langgraph.runtime import Runtime
model = init_chat_model("gpt-5.4-mini")
# Create store with semantic search enabled
embeddings = init_embeddings("openai:text-embedding-3-small")
store = InMemoryStore(
index={
"embed": embeddings,
"dims": 1536,
}
)
store.put(("user_123", "memories"), "1", {"text": "I love pizza"})
store.put(("user_123", "memories"), "2", {"text": "I am a plumber"})
async def chat(state: MessagesState, runtime: Runtime):
# Search based on user's last message
items = await runtime.store.asearch(
("user_123", "memories"), query=state["messages"][-1].content, limit=2
)
memories = "\n".join(item.value["text"] for item in items)
memories = f"## Memories of user\n{memories}" if memories else ""
response = await model.ainvoke(
[\
{"role": "system", "content": f"You are a helpful assistant.\n{memories}"},\
*state["messages"],\
]
)
return {"messages": [response]}
builder = StateGraph(MessagesState)
builder.add_node(chat)
builder.add_edge(START, "chat")
graph = builder.compile(store=store)
stream = await graph.astream_events(
{"messages": [{"role": "user", "content": "I'm hungry"}]},
version="v3",
)
async for message in stream.messages:
async for token in message.text:
print(token, end="", flush=True)
단기 메모리 관리하기 (Manage short-term memory)
단기 메모리가 활성화되면, 긴 대화는 LLM의 컨텍스트 윈도우를 초과할 수 있습니다. 일반적인 해결 방법은 다음과 같습니다:
- 메시지 자르기(Trim messages): (LLM 호출 전에) 처음 또는 마지막 N개 메시지를 제거합니다.
- LangGraph 상태에서 메시지 삭제(Delete messages)를 영구적으로 수행합니다.
- 메시지 요약(Summarize messages): 히스토리의 이전 메시지를 요약하고 이를 요약문으로 교체합니다.
- 메시지 히스토리를 저장하고 검색하기 위한 체크포인트 관리(Manage checkpoints).
- 커스텀 전략(예: 메시지 필터링 등).
이를 통해 에이전트는 LLM의 컨텍스트 윈도우를 초과하지 않으면서 대화를 계속 추적할 수 있습니다.
메시지 자르기 (Trim messages)
대부분의 LLM에는 최대 지원 컨텍스트 윈도우(토큰 단위로 표시)가 있습니다. 언제 메시지를 잘라낼지 결정하는 한 가지 방법은 메시지 히스토리의 토큰 수를 세어 그 한계에 근접할 때 잘라내는 것입니다. LangChain을 사용한다면 trim_messages 유틸리티를 사용하고 유지할 토큰 수와 경계 처리에 사용할 strategy(예: 마지막 max_tokens 유지)를 지정할 수 있습니다. 메시지 히스토리를 자르려면 trim_messages 함수를 사용하세요:
from langchain_core.messages.utils import (
trim_messages,
count_tokens_approximately
)
def call_model(state: MessagesState):
messages = trim_messages(
state["messages"],
strategy="last",
token_counter=count_tokens_approximately,
max_tokens=128,
start_on="human",
end_on=("human", "tool"),
)
response = model.invoke(messages)
return {"messages": [response]}
builder = StateGraph(MessagesState)
builder.add_node(call_model)
...
전체 예시: 메시지 자르기
from langchain_core.messages.utils import (
trim_messages,
count_tokens_approximately
)
from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, START, MessagesState
model = init_chat_model("claude-sonnet-4-6")
summarization_model = model.bind(max_tokens=128)
def call_model(state: MessagesState):
messages = trim_messages(
state["messages"],
strategy="last",
token_counter=count_tokens_approximately,
max_tokens=128,
start_on="human",
end_on=("human", "tool"),
)
response = model.invoke(messages)
return {"messages": [response]}
checkpointer = InMemorySaver()
builder = StateGraph(MessagesState)
builder.add_node(call_model)
builder.add_edge(START, "call_model")
graph = builder.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "1"}}
graph.invoke({"messages": "hi, my name is bob"}, config)
graph.invoke({"messages": "write a short poem about cats"}, config)
graph.invoke({"messages": "now do the same but for dogs"}, config)
final_response = graph.invoke({"messages": "what's my name?"}, config)
final_response["messages"][-1].pretty_print()
================================== Ai Message ==================================
Your name is Bob, as you mentioned when you first introduced yourself.
메시지 삭제 (Delete messages)
그래프 상태에서 메시지를 삭제하여 메시지 히스토리를 관리할 수 있습니다. 특정 메시지를 제거하거나 전체 메시지 히스토리를 비우고 싶을 때 유용합니다. 그래프 상태에서 메시지를 삭제하려면 RemoveMessage를 사용할 수 있습니다. RemoveMessage가 작동하려면 MessagesState처럼 add_messages 리듀서(reducer)가 있는 상태 키를 사용해야 합니다.
특정 메시지를 제거하려면:
from langchain.messages import RemoveMessage
def delete_messages(state):
messages = state["messages"]
if len(messages) > 2:
# remove the earliest two messages
return {"messages": [RemoveMessage(id=m.id) for m in messages[:2]]}
모든 메시지를 제거하려면:
from langgraph.graph.message import REMOVE_ALL_MESSAGES
def delete_messages(state):
return {"messages": [RemoveMessage(id=REMOVE_ALL_MESSAGES)]}
메시지를 삭제할 때는 결과로 만들어지는 메시지 히스토리가 유효한지 확인하세요. 사용 중인 LLM 제공자의 제한 사항을 확인하세요. 예를 들어:
- 일부 제공자는 메시지 히스토리가
user메시지로 시작하기를 기대합니다. - 대부분의 제공자는 도구 호출이 있는
assistant메시지 뒤에 해당하는tool결과 메시지가 이어지도록 요구합니다.
전체 예시: 메시지 삭제
from langchain.messages import RemoveMessage
def delete_messages(state):
messages = state["messages"]
if len(messages) > 2:
# remove the earliest two messages
return {"messages": [RemoveMessage(id=m.id) for m in messages[:2]]}
def call_model(state: MessagesState):
response = model.invoke(state["messages"])
return {"messages": response}
builder = StateGraph(MessagesState)
builder.add_sequence([call_model, delete_messages])
builder.add_edge(START, "call_model")
checkpointer = InMemorySaver()
app = builder.compile(checkpointer=checkpointer)
stream = app.stream_events(
{"messages": [{"role": "user", "content": "hi! I'm bob"}]},
config,
version="v3"
)
for snapshot in stream.values:
print([(message.type, message.content) for message in snapshot["messages"]])
stream = app.stream_events(
{"messages": [{"role": "user", "content": "what's my name?"}]},
config,
version="v3"
)
for snapshot in stream.values:
print([(message.type, message.content) for message in snapshot["messages"]])
[('human', "hi! I'm bob")]
[('human', "hi! I'm bob"), ('ai', 'Hi Bob! How are you doing today? Is there anything I can help you with?')]
[('human', "hi! I'm bob"), ('ai', 'Hi Bob! How are you doing today? Is there anything I can help you with?'), ('human', "what's my name?")]
[('human', "hi! I'm bob"), ('ai', 'Hi Bob! How are you doing today? Is there anything I can help you with?'), ('human', "what's my name?"), ('ai', 'Your name is Bob.')]
[('human', "what's my name?"), ('ai', 'Your name is Bob.')]
메시지 요약 (Summarize messages)
위에서 본 메시지 자르기나 삭제의 문제점은 메시지 큐를 정리하면서 정보를 잃을 수 있다는 것입니다. 따라서 일부 애플리케이션은 채팅 모델을 사용해 메시지 히스토리를 요약하는 더 정교한 접근 방식이 더 유리합니다.
프롬프팅과 오케스트레이션 로직을 사용해 메시지 히스토리를 요약할 수 있습니다. 예를 들어 LangGraph에서는 MessagesState를 확장하여 summary 키를 포함할 수 있습니다:
from langgraph.graph import MessagesState
class State(MessagesState):
summary: str
그런 다음 기존 요약을 다음 요약의 컨텍스트로 사용하여 채팅 히스토리의 요약을 생성할 수 있습니다. 이 summarize_conversation 노드는 messages 상태 키에 특정 수의 메시지가 쌓인 후 호출할 수 있습니다.
def summarize_conversation(state: State):
# First, we get any existing summary
summary = state.get("summary", "")
# Create our summarization prompt
if summary:
# A summary already exists
summary_message = (
f"This is a summary of the conversation to date: {summary}\n\n"
"Extend the summary by taking into account the new messages above:"
)
else:
summary_message = "Create a summary of the conversation above:"
# Add prompt to our history
messages = state["messages"] + [HumanMessage(content=summary_message)]
response = model.invoke(messages)
# Delete all but the 2 most recent messages
delete_messages = [RemoveMessage(id=m.id) for m in state["messages"][:-2]]
return {"summary": response.content, "messages": delete_messages}
전체 예시: 메시지 요약
from typing import Any, TypedDict
from langchain.chat_models import init_chat_model
from langchain.messages import AnyMessage
from langchain_core.messages.utils import count_tokens_approximately
from langgraph.graph import StateGraph, START, MessagesState
from langgraph.checkpoint.memory import InMemorySaver
from langmem.short_term import SummarizationNode, RunningSummary
model = init_chat_model("claude-sonnet-4-6")
summarization_model = model.bind(max_tokens=128)
class State(MessagesState):
context: dict[str, RunningSummary]
class LLMInputState(TypedDict):
summarized_messages: list[AnyMessage]
context: dict[str, RunningSummary]
summarization_node = SummarizationNode(
token_counter=count_tokens_approximately,
model=summarization_model,
max_tokens=256,
max_tokens_before_summary=256,
max_summary_tokens=128,
)
def call_model(state: LLMInputState):
response = model.invoke(state["summarized_messages"])
return {"messages": [response]}
checkpointer = InMemorySaver()
builder = StateGraph(State)
builder.add_node(call_model)
builder.add_node("summarize", summarization_node)
builder.add_edge(START, "summarize")
builder.add_edge("summarize", "call_model")
graph = builder.compile(checkpointer=checkpointer)
# Invoke the graph
config = {"configurable": {"thread_id": "1"}}
graph.invoke({"messages": "hi, my name is bob"}, config)
graph.invoke({"messages": "write a short poem about cats"}, config)
graph.invoke({"messages": "now do the same but for dogs"}, config)
final_response = graph.invoke({"messages": "what's my name?"}, config)
final_response["messages"][-1].pretty_print()
print("\nSummary:", final_response["context"]["running_summary"].summary)
context필드에 실행 중인 요약(running summary)을 추적합니다. (SummarizationNode가 기대하는 방식입니다.)call_model노드의 입력을 필터링하는 데만 사용될 비공개 상태를 정의합니다.- 요약 노드가 반환한 메시지를 격리하기 위해 여기서 비공개 입력 상태를 전달합니다.
================================== Ai Message ==================================
From our conversation, I can see that you introduced yourself as Bob. That's the name you shared with me when we began talking.
Summary: In this conversation, I was introduced to Bob, who then asked me to write a poem about cats. I composed a poem titled "The Mystery of Cats" that captured cats' graceful movements, independent nature, and their special relationship with humans. Bob then requested a similar poem about dogs, so I wrote "The Joy of Dogs," which highlighted dogs' loyalty, enthusiasm, and loving companionship. Both poems were written in a similar style but emphasized the distinct characteristics that make each pet special.
체크포인트 관리 (Manage checkpoints)
체크포인터가 저장한 정보를 조회하고 삭제할 수 있습니다.
스레드 상태 조회 (View thread state)
- Graph/Functional API
- Checkpointer API
config = {
"configurable": {
"thread_id": "1",
# optionally provide an ID for a specific checkpoint,
# otherwise the latest checkpoint is shown
# "checkpoint_id": "1f029ca3-1f5b-6704-8004-820c16b69a5a"
}
}
graph.get_state(config)
StateSnapshot(
values={'messages': [HumanMessage(content="hi! I'm bob"), AIMessage(content='Hi Bob! How are you doing today?), HumanMessage(content="what's my name?"), AIMessage(content='Your name is Bob.')]}, next=(),
config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f029ca3-1f5b-6704-8004-820c16b69a5a'}},
metadata={
'source': 'loop',
'writes': {'call_model': {'messages': AIMessage(content='Your name is Bob.')}},
'step': 4,
'parents': {},
'thread_id': '1'
},
created_at='2025-05-05T16:01:24.680462+00:00',
parent_config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f029ca3-1790-6b0a-8003-baf965b6a38f'}},
tasks=(),
interrupts=()
)
config = {
"configurable": {
"thread_id": "1",
# optionally provide an ID for a specific checkpoint,
# otherwise the latest checkpoint is shown
# "checkpoint_id": "1f029ca3-1f5b-6704-8004-820c16b69a5a"
}
}
checkpointer.get_tuple(config)
CheckpointTuple(
config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f029ca3-1f5b-6704-8004-820c16b69a5a'}},
checkpoint={
'v': 3,
'ts': '2025-05-05T16:01:24.680462+00:00',
'id': '1f029ca3-1f5b-6704-8004-820c16b69a5a',
'channel_versions': {'__start__': '00000000000000000000000000000005.0.5290678567601859', 'messages': '00000000000000000000000000000006.0.3205149138784782', 'branch:to:call_model': '00000000000000000000000000000006.0.14611156755133758'}, 'versions_seen': {'__input__': {}, '__start__': {'__start__': '00000000000000000000000000000004.0.5736472536395331'}, 'call_model': {'branch:to:call_model': '00000000000000000000000000000005.0.1410174088651449'}},
'channel_values': {'messages': [HumanMessage(content="hi! I'm bob"), AIMessage(content='Hi Bob! How are you doing today?), HumanMessage(content="what's my name?"), AIMessage(content='Your name is Bob.')]},
},
metadata={
'source': 'loop',
'writes': {'call_model': {'messages': AIMessage(content='Your name is Bob.')}},
'step': 4,
'parents': {},
'thread_id': '1'
},
parent_config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f029ca3-1790-6b0a-8003-baf965b6a38f'}},
pending_writes=[]
)
스레드 히스토리 조회 (View the history of the thread)
- Graph/Functional API
- Checkpointer API
config = {
"configurable": {
"thread_id": "1"
}
}
list(graph.get_state_history(config))
[\
StateSnapshot(\
values={'messages': [HumanMessage(content="hi! I'm bob"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?'), HumanMessage(content="what's my name?"), AIMessage(content='Your name is Bob.')]},\
next=(),\
config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f029ca3-1f5b-6704-8004-820c16b69a5a'}},\
metadata={'source': 'loop', 'writes': {'call_model': {'messages': AIMessage(content='Your name is Bob.')}}, 'step': 4, 'parents': {}, 'thread_id': '1'},\
created_at='2025-05-05T16:01:24.680462+00:00',\
parent_config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f029ca3-1790-6b0a-8003-baf965b6a38f'}},\
tasks=(),\
interrupts=()\
),\
StateSnapshot(\
values={'messages': [HumanMessage(content="hi! I'm bob"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?'), HumanMessage(content="what's my name?")]},\
next=('call_model',),\
config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f029ca3-1790-6b0a-8003-baf965b6a38f'}},\
metadata={'source': 'loop', 'writes': None, 'step': 3, 'parents': {}, 'thread_id': '1'},\
created_at='2025-05-05T16:01:23.863421+00:00',\
parent_config={...}\
tasks=(PregelTask(id='8ab4155e-6b15-b885-9ce5-bed69a2c305c', name='call_model', path=('__pregel_pull', 'call_model'), error=None, interrupts=(), state=None, result={'messages': AIMessage(content='Your name is Bob.')}),),\
interrupts=()\
),\
StateSnapshot(\
values={'messages': [HumanMessage(content="hi! I'm bob"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?')]},\
next=('__start__',),\
config={...},\
metadata={'source': 'input', 'writes': {'__start__': {'messages': [{'role': 'user', 'content': "what's my name?"}]}}, 'step': 2, 'parents': {}, 'thread_id': '1'},\
created_at='2025-05-05T16:01:23.863173+00:00',\
parent_config={...}\
tasks=(PregelTask(id='24ba39d6-6db1-4c9b-f4c5-682aeaf38dcd', name='__start__', path=('__pregel_pull', '__start__'), error=None, interrupts=(), state=None, result={'messages': [{'role': 'user', 'content': "what's my name?"}]}),),\
interrupts=()\
),\
StateSnapshot(\
values={'messages': [HumanMessage(content="hi! I'm bob"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?')]},\
next=(),\
config={...},\
metadata={'source': 'loop', 'writes': {'call_model': {'messages': AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?')}}, 'step': 1, 'parents': {}, 'thread_id': '1'},\
created_at='2025-05-05T16:01:23.862295+00:00',\
parent_config={...}\
tasks=(),\
interrupts=()\
),\
StateSnapshot(\
values={'messages': [HumanMessage(content="hi! I'm bob")]},\
next=('call_model',),\
config={...},\
metadata={'source': 'loop', 'writes': None, 'step': 0, 'parents': {}, 'thread_id': '1'},\
created_at='2025-05-05T16:01:22.278960+00:00',\
parent_config={...}\
tasks=(PregelTask(id='8cbd75e0-3720-b056-04f7-71ac805140a0', name='call_model', path=('__pregel_pull', 'call_model'), error=None, interrupts=(), state=None, result={'messages': AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?')}),),\
interrupts=()\
),\
StateSnapshot(\
values={'messages': []},\
next=('__start__',),\
config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f029ca3-0870-6ce2-bfff-1f3f14c3e565'}},\
metadata={'source': 'input', 'writes': {'__start__': {'messages': [{'role': 'user', 'content': "hi! I'm bob"}]}}, 'step': -1, 'parents': {}, 'thread_id': '1'},\
created_at='2025-05-05T16:01:22.277497+00:00',\
parent_config=None,\
tasks=(PregelTask(id='d458367b-8265-812c-18e2-33001d199ce6', name='__start__', path=('__pregel_pull', '__start__'), error=None, interrupts=(), state=None, result={'messages': [{'role': 'user', 'content': "hi! I'm bob"}]}),),\
interrupts=()\
)\
]
config = {
"configurable": {
"thread_id": "1"
}
}
list(checkpointer.list(config))
[\
CheckpointTuple(\
config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f029ca3-1f5b-6704-8004-820c16b69a5a'}},\
checkpoint={\
'v': 3,\
'ts': '2025-05-05T16:01:24.680462+00:00',\
'id': '1f029ca3-1f5b-6704-8004-820c16b69a5a',\
'channel_versions': {'__start__': '00000000000000000000000000000005.0.5290678567601859', 'messages': '00000000000000000000000000000006.0.3205149138784782', 'branch:to:call_model': '00000000000000000000000000000006.0.14611156755133758'},\
'versions_seen': {'__input__': {}, '__start__': {'__start__': '00000000000000000000000000000004.0.5736472536395331'}, 'call_model': {'branch:to:call_model': '00000000000000000000000000000005.0.1410174088651449'}},\
'channel_values': {'messages': [HumanMessage(content="hi! I'm bob"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?'), HumanMessage(content="what's my name?"), AIMessage(content='Your name is Bob.')]},\
},\
metadata={'source': 'loop', 'writes': {'call_model': {'messages': AIMessage(content='Your name is Bob.')}}, 'step': 4, 'parents': {}, 'thread_id': '1'},\
parent_config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f029ca3-1790-6b0a-8003-baf965b6a38f'}},\
pending_writes=[]\
),\
CheckpointTuple(\
config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f029ca3-1790-6b0a-8003-baf965b6a38f'}},\
checkpoint={\
'v': 3,\
'ts': '2025-05-05T16:01:23.863421+00:00',\
'id': '1f029ca3-1790-6b0a-8003-baf965b6a38f',\
'channel_versions': {'__start__': '00000000000000000000000000000005.0.5290678567601859', 'messages': '00000000000000000000000000000006.0.3205149138784782', 'branch:to:call_model': '00000000000000000000000000000006.0.14611156755133758'},\
'versions_seen': {'__input__': {}, '__start__': {'__start__': '00000000000000000000000000000004.0.5736472536395331'}, 'call_model': {'branch:to:call_model': '00000000000000000000000000000005.0.1410174088651449'}},\
'channel_values': {'messages': [HumanMessage(content="hi! I'm bob"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?'), HumanMessage(content="what's my name?")], 'branch:to:call_model': None}\
},\
metadata={'source': 'loop', 'writes': None, 'step': 3, 'parents': {}, 'thread_id': '1'},\
parent_config={...},\
pending_writes=[('8ab4155e-6b15-b885-9ce5-bed69a2c305c', 'messages', AIMessage(content='Your name is Bob.'))]\
),\
CheckpointTuple(\
config={...},\
checkpoint={\
'v': 3,\
'ts': '2025-05-05T16:01:23.863173+00:00',\
'id': '1f029ca3-1790-616e-8002-9e021694a0cd',\
'channel_versions': {'__start__': '00000000000000000000000000000004.0.5736472536395331', 'messages': '00000000000000000000000000000003.0.7056767754077798', 'branch:to:call_model': '00000000000000000000000000000003.0.22059023329132854'},\
'versions_seen': {'__input__': {}, '__start__': {'__start__': '00000000000000000000000000000001.0.7040775356287469'}, 'call_model': {'branch:to:call_model': '00000000000000000000000000000002.0.9300422176788571'}},\
'channel_values': {'__start__': {'messages': [{'role': 'user', 'content': "what's my name?"}]}, 'messages': [HumanMessage(content="hi! I'm bob"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?')]}\
},\
metadata={'source': 'input', 'writes': {'__start__': {'messages': [{'role': 'user', 'content': "what's my name?"}]}}, 'step': 2, 'parents': {}, 'thread_id': '1'},\
parent_config={...},\
pending_writes=[('24ba39d6-6db1-4c9b-f4c5-682aeaf38dcd', 'messages', [{'role': 'user', 'content': "what's my name?"}]), ('24ba39d6-6db1-4c9b-f4c5-682aeaf38dcd', 'branch:to:call_model', None)]\
),\
CheckpointTuple(\
config={...},\
checkpoint={\
'v': 3,\
'ts': '2025-05-05T16:01:23.862295+00:00',\
'id': '1f029ca3-178d-6f54-8001-d7b180db0c89',\
'channel_versions': {'__start__': '00000000000000000000000000000002.0.18673090920108737', 'messages': '00000000000000000000000000000003.0.7056767754077798', 'branch:to:call_model': '00000000000000000000000000000003.0.22059023329132854'},\
'versions_seen': {'__input__': {}, '__start__': {'__start__': '00000000000000000000000000000001.0.7040775356287469'}, 'call_model': {'branch:to:call_model': '00000000000000000000000000000002.0.9300422176788571'}},\
'channel_values': {'messages': [HumanMessage(content="hi! I'm bob"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?')]}\
},\
metadata={'source': 'loop', 'writes': {'call_model': {'messages': AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?')}}, 'step': 1, 'parents': {}, 'thread_id': '1'},\
parent_config={...},\
pending_writes=[]\
),\
CheckpointTuple(\
config={...},\
checkpoint={\
'v': 3,\
'ts': '2025-05-05T16:01:22.278960+00:00',\
'id': '1f029ca3-0874-6612-8000-339f2abc83b1',\
'channel_versions': {'__start__': '00000000000000000000000000000002.0.18673090920108737', 'messages': '00000000000000000000000000000002.0.30296526818059655', 'branch:to:call_model': '00000000000000000000000000000002.0.9300422176788571'},\
'versions_seen': {'__input__': {}, '__start__': {'__start__': '00000000000000000000000000000001.0.7040775356287469'}},\
'channel_values': {'messages': [HumanMessage(content="hi! I'm bob")], 'branch:to:call_model': None}\
},\
metadata={'source': 'loop', 'writes': None, 'step': 0, 'parents': {}, 'thread_id': '1'},\
parent_config={...},\
pending_writes=[('8cbd75e0-3720-b056-04f7-71ac805140a0', 'messages', AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?'))]\
),\
CheckpointTuple(\
config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f029ca3-0870-6ce2-bfff-1f3f14c3e565'}},\
checkpoint={\
'v': 3,\
'ts': '2025-05-05T16:01:22.277497+00:00',\
'id': '1f029ca3-0870-6ce2-bfff-1f3f14c3e565',\
'channel_versions': {'__start__': '00000000000000000000000000000001.0.7040775356287469'},\
'versions_seen': {'__input__': {}},\
'channel_values': {'__start__': {'messages': [{'role': 'user', 'content': "hi! I'm bob"}]}}\
},\
metadata={'source': 'input', 'writes': {'__start__': {'messages': [{'role': 'user', 'content': "hi! I'm bob"}]}}, 'step': -1, 'parents': {}, 'thread_id': '1'},\
parent_config=None,\
pending_writes=[('d458367b-8265-812c-18e2-33001d199ce6', 'messages', [{'role': 'user', 'content': "hi! I'm bob"}]), ('d458367b-8265-812c-18e2-33001d199ce6', 'branch:to:call_model', None)]\
)\
]
스레드의 모든 체크포인트 삭제 (Delete all checkpoints for a thread)
thread_id = "1"
checkpointer.delete_thread(thread_id)
데이터베이스 관리 (Database management)
Postgres, Redis, Oracle 같은 데이터베이스 기반 영속성 구현을 사용해 단기 및/또는 장기 메모리를 저장한다면, 데이터베이스와 함께 사용하기 전에 필요한 스키마를 설정하기 위해 마이그레이션을 실행해야 합니다. 관례적으로 대부분의 데이터베이스 전용 라이브러리는 필수 마이그레이션을 실행하는 setup() 메서드를 체크포인터 또는 store 인스턴스에 정의합니다. 그러나 정확한 메서드 이름과 사용법을 확인하려면 자신이 사용하는 BaseCheckpointSaver 또는 BaseStore 구현을 확인해야 합니다. 마이그레이션을 전용 배포 단계로 실행하거나 서버 시작 시 실행되도록 하는 것을 권장합니다.