메모리

메모리 (Memory)

AI 애플리케이션은 여러 상호작용 사이에서 맥락을 공유하기 위해 memory가 필요해요. LangGraph에서는 두 종류의 메모리를 추가할 수 있습니다:

• 에이전트의 state 일부로 단기 메모리를 추가해 다중 턴 대화를 가능하게.
• 세션을 넘어 사용자별·애플리케이션별 데이터를 저장하는 장기 메모리 추가.

출처: 문서

본문

단기 메모리 추가 (Add short-term memory)

단기 메모리(스레드 수준 persistence)는 에이전트가 다중 턴 대화를 추적하게 해줍니다. 단기 메모리를 추가하려면:

from langgraph.checkpoint.memory import InMemorySaver  # [!code highlight]
from langgraph.graph import StateGraph

checkpointer = InMemorySaver()  # [!code highlight]

builder = StateGraph(...)
graph = builder.compile(checkpointer=checkpointer)  # [!code highlight]

graph.invoke(
    {"messages": [{"role": "user", "content": "hi! i am Bob"}]},
    {"configurable": {"thread_id": "1"}},  # [!code highlight]
)

프로덕션에서 사용 (Use in production)

프로덕션에서는 데이터베이스로 백업된 체크포인터를 사용하세요:

from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://postgres:***@localhost:5432/postgres?sslmode=disable"
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:  # [!code highlight]
    builder = StateGraph(...)
    graph = builder.compile(checkpointer=checkpointer)  # [!code highlight]

pip install -U "psycopg[binary,pool]" langgraph langgraph-checkpoint-postgres

Postgres 체크포인터를 처음 사용할 때는 `checkpointer.setup()`을 호출해야 해요.

동기(Sync) 버전:

from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.checkpoint.postgres import PostgresSaver  # [!code highlight]

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:  # [!code highlight]
    # 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)  # [!code highlight]

    config = {
        "configurable": {
            "thread_id": "1"  # [!code highlight]
        }
    }

    stream = graph.stream_events(
        {"messages": [{"role": "user", "content": "hi! I'm bob"}]},
        config,  # [!code highlight]
        version="v3",
    )
    for snapshot in stream.values:
        print(snapshot)

    stream = graph.stream_events(
        {"messages": [{"role": "user", "content": "what's my name?"}]},
        config,  # [!code highlight]
        version="v3",
    )
    for snapshot in stream.values:
        print(snapshot)

비동기(Async) 버전:

from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver  # [!code highlight]

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:  # [!code highlight]
    # 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)  # [!code highlight]

    config = {
        "configurable": {
            "thread_id": "1"  # [!code highlight]
        }
    }

    stream = await graph.astream_events(
        {"messages": [{"role": "user", "content": "hi! I'm bob"}]},
        config,  # [!code highlight]
        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,  # [!code highlight]
        version="v3",
    )
    async for message in stream.messages:
        async for token in message.text:
            print(token, end="", flush=True)

MongoDB, Redis, Oracle 체크포인터

MongoDB, Redis, Oracle 체크포인터도 같은 패턴을 사용합니다. 설치와 새이버 클래스만 다릅니다:

pip install -U pymongo langgraph langgraph-checkpoint-mongodb
pip install -U langgraph langgraph-checkpoint-redis
pip install -U langgraph langgraph-oracledb

• MongoDB: MongoDBSaver / AsyncMongoDBSaver (동기: from langgraph.checkpoint.mongodb import MongoDBSaver, 비동기: from langgraph.checkpoint.mongodb.aio import AsyncMongoDBSaver). MongoDB 클러스터가 필요하며, 생성 가이드는 여기를 참고하세요. 에이전트 중심 워크스루는 MongoDB Atlas로 단기 메모리를 참고.
• Redis: RedisSaver / AsyncRedisSaver. 처음 사용 시 checkpointer.setup() 호출 필요.
• Oracle: from langgraph_oracledb.checkpoint.oracle import OracleSaver / AsyncOracleSaver. Oracle AI Database 인스턴스 필요(예: gvenzl/oracle-free:23-slim 로컬 컨테이너 또는 OCI의 Oracle Autonomous Database). 처음 사용 시 checkpointer.setup() 호출 필요.

서브그래프에서 사용 (Use in 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()  # [!code highlight]

# Parent graph

builder = StateGraph(State)
builder.add_node("node_1", subgraph)  # [!code highlight]
builder.add_edge(START, "node_1")

checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)  # [!code highlight]

서브그래프별 체크포인팅 동작을 구성할 수 있어요. 인터럽트 지원과 상태 유지 연속을 포함한 영속성 수준에 대한 자세한 내용은 subgraph persistence를 참고하세요.

subgraph_builder = StateGraph(...)
subgraph = subgraph_builder.compile(checkpointer=True)  # [!code highlight]

장기 메모리 추가 (Add long-term memory)

장기 메모리는 대화를 넘어 사용자별·애플리케이션별 데이터를 저장하는 데 사용해요.

from langgraph.store.memory import InMemoryStore  # [!code highlight]
from langgraph.graph import StateGraph

store = InMemoryStore()  # [!code highlight]

builder = StateGraph(...)
graph = builder.compile(store=store)  # [!code highlight]

노드 안에서 스토어 접근 (Access the store inside nodes)

스토어로 그래프를 컴파일하면 LangGraph가 노드 함수에 스토어를 자동 주입합니다. 스토어에 접근하는 권장 방법은 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]):  # [!code highlight]
    user_id = runtime.context.user_id  # [!code highlight]
    namespace = (user_id, "memories")

    # Search for relevant memories
    memories = await runtime.store.asearch(  # [!code highlight]
        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(  # [!code highlight]
        namespace, str(uuid.uuid4()), {"data": "User prefers dark mode"}
    )

builder = StateGraph(MessagesState, context_schema=Context)  # [!code highlight]
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"),  # [!code highlight]
)

프로덕션에서 사용 (Use in production)

프로덕션에서는 데이터베이스로 백업된 스토어를 사용하세요:

from langgraph.store.postgres import PostgresStore

DB_URI = "postgresql://postgres:***@localhost:5432/postgres?sslmode=disable"
with PostgresStore.from_conn_string(DB_URI) as store:  # [!code highlight]
    builder = StateGraph(...)
    graph = builder.compile(store=store)  # [!code highlight]

pip install -U "psycopg[binary,pool]" langgraph langgraph-checkpoint-postgres

Postgres 스토어를 처음 사용할 때는 `store.setup()`을 호출해야 해요.

비동기(Async) 예시 — Runtime[Context]로 스토어 접근:

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  # [!code highlight]
from langgraph.runtime import Runtime  # [!code highlight]
import uuid

model = init_chat_model(model="claude-haiku-4-5-20251001")

@dataclass
class Context:
    user_id: str

async def call_model(  # [!code highlight]
    state: MessagesState,
    runtime: Runtime[Context],  # [!code highlight]
):
    user_id = runtime.context.user_id  # [!code highlight]
    namespace = ("memories", user_id)
    memories = await runtime.store.asearch(namespace, query=str(state["messages"][-1].content))  # [!code highlight]
    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})  # [!code highlight]

    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,  # [!code highlight]
    AsyncPostgresSaver.from_conn_string(DB_URI) as checkpointer,
):
    # await store.setup()
    # await checkpointer.setup()

    builder = StateGraph(MessagesState, context_schema=Context)  # [!code highlight]
    builder.add_node(call_model)
    builder.add_edge(START, "call_model")

    graph = builder.compile(
        checkpointer=checkpointer,
        store=store,  # [!code highlight]
    )

    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"),  # [!code highlight]
    )
    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"),  # [!code highlight]
    )
    async for message in stream.messages:
        async for token in message.text:
            print(token, end="", flush=True)

동기(Sync) 예시:

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  # [!code highlight]
from langgraph.runtime import Runtime  # [!code highlight]
import uuid

model = init_chat_model(model="claude-haiku-4-5-20251001")

@dataclass
class Context:
    user_id: str

def call_model(  # [!code highlight]
    state: MessagesState,
    runtime: Runtime[Context],  # [!code highlight]
):
    user_id = runtime.context.user_id  # [!code highlight]
    namespace = ("memories", user_id)
    memories = runtime.store.search(namespace, query=str(state["messages"][-1].content))  # [!code highlight]
    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})  # [!code highlight]

    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,  # [!code highlight]
    PostgresSaver.from_conn_string(DB_URI) as checkpointer,
):
    # store.setup()
    # checkpointer.setup()

    builder = StateGraph(MessagesState, context_schema=Context)  # [!code highlight]
    builder.add_node(call_model)
    builder.add_edge(START, "call_model")

    graph = builder.compile(
        checkpointer=checkpointer,
        store=store,  # [!code highlight]
    )

    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"),  # [!code highlight]
    )
    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"),  # [!code highlight]
    )
    for snapshot in stream.values:
        print(snapshot)

Redis의 경우 AsyncRedisStore/RedisStore(langgraph.store.redis), Oracle의 경우 OracleStore/AsyncOracleStore(langgraph_oracledb.store.oracle)를 쓰면 같은 패턴입니다. Oracle 스토어는 의미 search용 벡터 인덱스가 Oracle AI Vector Search를 요구합니다.

pip install -U langgraph langgraph-checkpoint-redis
pip install -U langgraph langgraph-oracledb langchain-openai

의미 검색 사용 (Use semantic search)

그래프의 메모리 스토어에서 의미 검색(semantic search)을 활성화하면, 그래프 에이전트가 의미 유사도로 스토어의 항목을 검색할 수 있어요.

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
)

Runtime으로 대화 중에 의미 검색:

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  # [!code highlight]

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):  # [!code highlight]
    # Search based on user's last message
    items = await runtime.store.asearch(  # [!code highlight]
        ("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의 컨텍스트 창을 초과할 수 있어요. 일반적인 해결책:

• 메시지 트리밍: 처음 또는 마지막 N개 메시지 제거 (LLM 호출 전)
• 메시지 삭제: LangGraph 상태에서 영구 삭제
• 메시지 요약: 이력의 이전 메시지를 요약하고 그 요약으로 대체
• 체크포인트 관리: 메시지 이력 저장·조회
• 커스텀 전략 (예: 메시지 필터링 등)

이렇게 하면 에이전트가 LLM 컨텍스트 창을 초과하지 않고 대화를 추적할 수 있어요.

메시지 트리밍 (Trim messages)

대부분의 LLM은 최대 지원 컨텍스트 창(토큰 단위)이 있어요. 메시지를 언제 잘라낼지 정하는 한 가지 방법은 메시지 이력의 토큰을 세고 그 한도에 가까워지면 잘라내는 것입니다. LangChain을 쓴다면 trim_messages 유틸리티를 사용해 목록에서 유지할 토큰 수와, 경계 처리에 사용할 strategy(예: 마지막 max_tokens 유지)를 지정할 수 있어요.

메시지 이력을 트리밍하려면 trim_messages 함수를 사용하세요:

from langchain_core.messages.utils import (  # [!code highlight]
    trim_messages,  # [!code highlight]
    count_tokens_approximately  # [!code highlight]
)  # [!code highlight]

def call_model(state: MessagesState):
    messages = trim_messages(  # [!code highlight]
        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,  # [!code highlight]
    count_tokens_approximately  # [!code highlight]
)
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(  # [!code highlight]
        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가 동작하려면 add_messages reducer가 있는 상태 키(예: MessagesState)를 사용해야 합니다.

특정 메시지를 제거하려면:

from langchain.messages import RemoveMessage  # [!code highlight]

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]]}  # [!code highlight]

모든 메시지를 제거하려면:

from langgraph.graph.message import REMOVE_ALL_MESSAGES  # [!code highlight]

def delete_messages(state):
    return {"messages": [RemoveMessage(id=REMOVE_ALL_MESSAGES)]}  # [!code highlight]

메시지를 삭제할 때 결과 메시지 이력이 유효한지 **반드시** 확인하세요. 사용하는 LLM 프로바이더의 제한을 확인하세요. 예를 들어: * 일부 프로바이더는 메시지 이력이 `user` 메시지로 시작할 것을 기대합니다. * 대부분의 프로바이더는 도구 호출이 있는 `assistant` 메시지 뒤에 해당 `tool` 결과 메시지가 오도록 요구합니다.

체크포인터와 함께 삭제하는 완전한 예시:

from langchain.messages import RemoveMessage  # [!code highlight]

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]]}  # [!code highlight]

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

그런 다음 기존 요약을 다음 요약의 맥락으로 사용해 채팅 이력 요약을 생성할 수 있어요. messages 상태 키에 일정 수의 메시지가 쌓인 뒤 이 summarize_conversation 노드를 호출합니다.

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}

langmem의 SummarizationNode로 요약하는 예시:

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  # [!code highlight]

model = init_chat_model("claude-sonnet-4-6")
summarization_model = model.bind(max_tokens=128)

class State(MessagesState):
    context: dict[str, RunningSummary]  # [!code highlight]

class LLMInputState(TypedDict):  # [!code highlight]
    summarized_messages: list[AnyMessage]
    context: dict[str, RunningSummary]

summarization_node = SummarizationNode(  # [!code highlight]
    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):  # [!code highlight]
    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)  # [!code highlight]
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)

설명:

1. SummarizationNode가 기대하는 context 필드에 실행 중인 요약(running summary)을 추적합니다.
2. call_model 노드의 입력을 필터링하는 데만 쓰일 비공개(private) 상태를 정의합니다.
3. 여기서 비공개 입력 상태를 전달해 요약 노드가 반환한 메시지를 격리합니다.

================================== 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로:

config = {
    "configurable": {
        "thread_id": "1",  # [!code highlight]
        # optionally provide an ID for a specific checkpoint,
        # otherwise the latest checkpoint is shown
        # "checkpoint_id": "1f029ca3-1f5b-6704-8004-820c16b69a5a"  # [!code highlight]

    }
}
graph.get_state(config)  # [!code highlight]

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=()
)

Checkpointer API로:

config = {
    "configurable": {
        "thread_id": "1",  # [!code highlight]
        # optionally provide an ID for a specific checkpoint,
        # otherwise the latest checkpoint is shown
        # "checkpoint_id": "1f029ca3-1f5b-6704-8004-820c16b69a5a"  # [!code highlight]

    }
}
checkpointer.get_tuple(config)  # [!code highlight]

스레드 이력 조회 (View the history of the thread)

Graph/Functional API로:

config = {
    "configurable": {
        "thread_id": "1"  # [!code highlight]
    }
}
list(graph.get_state_history(config))  # [!code highlight]

Checkpointer API로:

config = {
    "configurable": {
        "thread_id": "1"  # [!code highlight]
    }
}
list(checkpointer.list(config))  # [!code highlight]

스레드의 모든 체크포인트 삭제 (Delete all checkpoints for a thread)

thread_id = "1"
checkpointer.delete_thread(thread_id)

데이터베이스 관리 (Database management)

단기 및/또는 장기 메모리를 저장하기 위해 데이터베이스 기반 영속성 구현(Postgres, Redis, Oracle 등)을 사용한다면, 데이터베이스에서 사용하기 전에 필수 스키마를 설정하기 위한 마이그레이션을 실행해야 합니다.

관례상 대부분의 데이터베이스 특화 라이브러리는 필수 마이그레이션을 실행하는 setup() 메서드를 체크포인터나 스토어 인스턴스에 정의해요. 하지만 사용 중인 BaseCheckpointSaver 또는 BaseStore 구현에서 정확한 메서드 이름과 사용법을 확인하세요.

마이그레이션은 전용 배포 단계로 실행하거나, 서버 시작 시 실행되도록 하는 것을 권장합니다.

더 알아보기 (Learn more)

• Persistence — 스레드 수준 영속화.
• Checkpointers — 체크포인터 심층.
• Subgraph persistence — 서브그래프 영속성.
• MongoDB Atlas 단기 메모리 — MongoDB 에이전트 워크스루.