체크포인터

체크포인터 (Checkpointers)

LangGraph 체크포인터는 각 단계에서 그래프 상태를 체크포인트로 저장해, 영속성(persistence), human-in-the-loop, 장애 허용 실행을 가능하게 해요.

체크포인터(checkpointer)는 각 슈퍼스텝(super-step)에서 그래프 상태의 스냅샷을 스레드(threads) 단위로 저장합니다. 그래프를 체크포인터와 함께 컴파일하면 human-in-the-loop 워크플로우, time travel 디버깅, 장애 허용 실행, 대화 메모리를 쓸 수 있어요.

출처: 문서

본문

**Agent Server는 체크포인팅을 자동으로 처리합니다** — [Agent Server](/langsmith/agent-server)를 사용할 때는 체크포인터를 직접 구현하거나 구성할 필요가 없어요. 서버가 모든 영속성 인프라를 뒤에서 처리해 줍니다. 체크포인트된 상태를 추적하고 에이전트가 세션을 넘어 어떻게 재개되는지 디버그하려면 [LangSmith](https://smith.langchain.com)를 쓰세요. [tracing quickstart](/langsmith/trace-with-langgraph)를 따라 설정해 보세요.

체크포인터를 왜 쓸까 (Why use checkpointers)

체크포인터는 다음 기능에 필수입니다:

• Human-in-the-loop: 체크포인터는 사람이 그래프 단계를 검사·중단·승인할 수 있도록 human-in-the-loop 워크플로우를 지원해요. 이 워크플로우에는 체크포인터가 필요한데, 사람이 어떤 시점에서든 그래프 상태를 볼 수 있어야 하고, 사람이 상태를 갱신한 뒤 그래프가 실행을 재개할 수 있어야 하기 때문입니다. 예시는 Interrupts를 참고하세요.
• 메모리: 체크포인터는 상호작용 사이의 "메모리"를 가능하게 해요. 반복되는 인간 상호작용(예: 대화)에서는 후속 메시지를 그 스레드로 보낼 수 있고, 스레드는 이전 메시지의 기억을 유지합니다. 체크포인터로 대화 메모리를 추가·관리하는 방법은 Add memory를 참고하세요.
• Time travel: 체크포인터는 "time travel"을 가능하게 해서, 이전 그래프 실행을 재생(replay)해 특정 그래프 단계를 검토·디버그할 수 있어요. 또한 임의의 체크포인트에서 그래프 상태를 포크(fork)해 대안 경로를 탐색할 수 있게 해줍니다.
• 장애 허용(Fault-tolerance): 체크포인팅은 장애 허용과 오류 복구를 제공합니다. 주어진 슈퍼스텝에서 하나 이상의 노드가 실패하면 마지막 성공 단계에서 그래프를 재시작할 수 있어요.

• 보류 중인 쓰기(Pending writes): 그래프 노드가 어떤 슈퍼스텝에서 실행 중 실패하면, LangGraph는 같은 슈퍼스텝에서 성공적으로 완료된 다른 노드들의 보류 중인 체크포인트 쓰기를 저장해요. 그 슈퍼스텝에서 그래프 실행을 재개할 때 성공한 노드를 다시 실행하지 않습니다.

핵심 개념 (Core concepts)

스레드 (Threads)

스레드는 체크포인터가 저장한 각 체크포인트에 할당된 고유 ID, 즉 스레드 식별자예요. 일련의 런들의 누적 상태를 담고 있습니다. 런이 실행되면 어시스턴트의 기반 그래프 state가 그 스레드에 영속화됩니다.

체크포인터가 있는 그래프를 호출할 때 config의 configurable 부분에 thread_id를 반드시 지정해야 해요:

{"configurable": {"thread_id": "1"}}

스레드의 현재·과거 상태를 조회할 수 있어요. 상태를 영속화하려면 런을 실행하기 전에 스레드를 먼저 생성해야 합니다. LangSmith API는 스레드와 스레드 상태를 만들고 관리하는 여러 엔드포인트를 제공합니다. 자세한 내용은 API reference를 참고하세요.

체크포인터는 체크포인트를 저장하고 조회하는 기본 키(primary key)로 thread_id를 사용합니다. thread_id가 없으면 체크포인터는 상태를 저장하거나 interrupt 후에 실행을 재개할 수 없어요. 체크포인터가 thread_id로 저장된 상태를 로드하기 때문입니다.

체크포인트 (Checkpoints)

특정 시점의 스레드 상태를 체크포인트라고 해요. 체크포인트는 각 슈퍼스텝에 저장된 그래프 상태의 스냅샷이며, StateSnapshot 객체로 표현됩니다(전체 필드 참조는 StateSnapshot 필드 참고).

슈퍼스텝 (Super-steps)

LangGraph는 각 슈퍼스텝 경계에서 체크포인트를 만듭니다. 슈퍼스텝은 그 단계에 예약된 모든 노드가 (잠재적으로 병렬로) 실행되는 그래프의 단일 "틱(tick)"이에요. START -> A -> B -> END 같은 순차 그래프라면 입력, 노드 A, 노드 B에 각각 별도의 슈퍼스텝이 있어 각각 후에 체크포인트를 만듭니다. 슈퍼스텝 경계를 이해하는 것은 time travel에 중요합니다. 체크포인트(즉, 슈퍼스텝 경계)에서만 실행을 재개할 수 있기 때문입니다.

슈퍼스텝 체크포인트 외에도 LangGraph는 노드(태스크) 수준에서 쓰기를 영속화해요. 슈퍼스텝 내의 각 노드가 끝나면 그 출력은 진행 중인 체크포인트에 연결된 태스크 항목으로 체크포인터의 checkpoint_writes 테이블에 기록됩니다. 이 태스크별 쓰기가 보류 중인 쓰기 복구를 가능하게 해요. 같은 슈퍼스텝의 다른 노드가 실패해도 성공한 노드의 쓰기는 이미 내구성이 있어 재개 시 다시 실행할 필요가 없습니다. 전체 상태 스냅샷은 슈퍼스텝이 완료되면 커밋됩니다.

LangGraph는 슈퍼스텝 내 개별 노드 실행의 쓰기도 영속화합니다. 이 쓰기는 태스크로 저장되어 장애 허용에 사용됩니다. 같은 슈퍼스텝의 다른 노드가 실패하면 성공한 노드의 쓰기는 재개 시 다시 계산할 필요가 없어요. 이 태스크 쓰기는 완전한 StateSnapshot 체크포인트가 아니므로, time travel은 슈퍼스텝 경계의 완전한 체크포인트에서 재개합니다.

체크포인트는 영속화되며 나중에 스레드 상태를 복원하는 데 사용할 수 있어요.

단순한 그래프를 다음과 같이 호출했을 때 어떤 체크포인트가 저장되는지 살펴볼게요:

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langchain_core.runnables import RunnableConfig
from typing import Annotated
from typing_extensions import TypedDict
from operator import add

class State(TypedDict):
    foo: str
    bar: Annotated[list[str], add]

def node_a(state: State):
    return {"foo": "a", "bar": ["a"]}

def node_b(state: State):
    return {"foo": "b", "bar": ["b"]}


workflow = StateGraph(State)
workflow.add_node(node_a)
workflow.add_node(node_b)
workflow.add_edge(START, "node_a")
workflow.add_edge("node_a", "node_b")
workflow.add_edge("node_b", END)

checkpointer = InMemorySaver()
graph = workflow.compile(checkpointer=checkpointer)

config: RunnableConfig = {"configurable": {"thread_id": "1"}}
graph.invoke({"foo": "", "bar":[]}, config)

그래프를 실행하면 정확히 4개의 체크포인트가 생깁니다:

• 다음 실행 노드가 START인 빈 체크포인트
• 사용자 입력 {'foo': '', 'bar': []}을 담고 다음 실행 노드가 node_a인 체크포인트
• node_a의 출력 {'foo': 'a', 'bar': ['a']}을 담고 다음 실행 노드가 node_b인 체크포인트
• node_b의 출력 {'foo': 'b', 'bar': ['a', 'b']}을 담고 다음 실행 노드가 없는 체크포인트

bar 채널에는 두 노드의 출력이 모두 들어 있습니다. 이 예시가 bar 채널에 리듀서(reducer)를 쓰기 때문입니다.

체크포인트 네임스페이스 (Checkpoint namespace)

각 체크포인트는 어떤 그래프 또는 서브그래프에 속하는지 식별하는 checkpoint_ns(체크포인트 네임스페이스) 필드를 가져요:

• "" (빈 문자열): 체크포인트는 부모(루트) 그래프에 속합니다.
• "node_name:uuid": 체크포인트는 주어진 노드로 호출된 서브그래프에 속합니다. 중첩 서브그래프에서는 네임스페이스가 | 구분자로 연결됩니다(예: "outer_node:uuid|inner_node:uuid").

config를 통해 노드 안에서 체크포인트 네임스페이스에 접근할 수 있어요:

from langchain_core.runnables import RunnableConfig

def my_node(state: State, config: RunnableConfig):
    checkpoint_ns = config["configurable"]["checkpoint_ns"]
    # "" for the parent graph, "node_name:uuid" for a subgraph

서브그래프 상태와 체크포인트 작업에 대한 자세한 내용은 Subgraphs를 참고하세요.

상태 조회 및 갱신 (Get and update state)

상태 조회 (Get state)

저장된 그래프 상태와 상호작용할 때는 스레드 식별자를 반드시 지정해야 해요. graph.get_state(config)를 호출하면 그래프의 최신 상태를 볼 수 있습니다. 이는 config에 제공된 스레드 ID와 연관된 최신 체크포인트, 또는 제공된 경우 그 스레드의 특정 checkpoint ID와 연관된 체크포인트에 해당하는 StateSnapshot 객체를 반환합니다.

# get the latest state snapshot
config = {"configurable": {"thread_id": "1"}}
graph.get_state(config)

# get a state snapshot for a specific checkpoint_id
config = {"configurable": {"thread_id": "1", "checkpoint_id": "1ef663ba-28fe-6528-8002-5a559208592c"}}
graph.get_state(config)

이 예시에서 get_state의 출력은 다음과 같습니다:

StateSnapshot(
    values={'foo': 'b', 'bar': ['a', 'b']},
    next=(),
    config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1ef663ba-28fe-6528-8002-5a559208592c'}},
    metadata={'source': 'loop', 'writes': {'node_b': {'foo': 'b', 'bar': ['b']}}, 'step': 2},
    created_at='2024-08-29T19:19:38.821749+00:00',
    parent_config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1ef663ba-28f9-6ec4-8001-31981c2c39f8'}}, tasks=()
)

StateSnapshot 필드 (StateSnapshot fields)

필드	타입	설명
values	dict	이 체크포인트에서의 상태 채널 값.
next	tuple[str, ...]	다음에 실행할 노드 이름. 빈 ()는 그래프가 완료됐다는 뜻입니다.
config	dict	thread_id, checkpoint_ns, checkpoint_id를 담습니다.
metadata	dict	실행 메타데이터. source ("input", "loop", 또는 "update"), writes(노드 출력), step(슈퍼스텝 카운터)를 담습니다.
created_at	str	이 체크포인트가 생성된 ISO 8601 타임스탬프.
parent_config	dict | None	이전 체크포인트의 config. 첫 체크포인트에서는 None.
tasks	tuple[PregelTask, ...]	이 단계에서 실행할 태스크. 각 태스크는 id, name, error, interrupts, 그리고 선택적으로 state(서브그래프 스냅샷, subgraphs=True 사용 시)를 가집니다.

상태 기록 조회 (Get state history)

특정 스레드의 그래프 실행 전체 기록은 graph.get_state_history(config)로 가져올 수 있어요. config에 제공된 스레드 ID와 연관된 StateSnapshot 객체 목록을 반환합니다. 중요하게도 체크포인트는 시간순으로 정렬되어, 가장 최근 체크포인트/StateSnapshot이 목록의 첫 번째입니다.

config = {"configurable": {"thread_id": "1"}}
list(graph.get_state_history(config))

이 예시에서 get_state_history의 출력은 다음과 같습니다:

[
    StateSnapshot(
        values={'foo': 'b', 'bar': ['a', 'b']},
        next=(),
        config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1ef663ba-28fe-6528-8002-5a559208592c'}},
        metadata={'source': 'loop', 'writes': {'node_b': {'foo': 'b', 'bar': ['b']}}, 'step': 2},
        created_at='2024-08-29T19:19:38.821749+00:00',
        parent_config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1ef663ba-28f9-6ec4-8001-31981c2c39f8'}},
        tasks=(),
    ),
    StateSnapshot(
        values={'foo': 'a', 'bar': ['a']},
        next=('node_b',),
        config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1ef663ba-28f9-6ec4-8001-31981c2c39f8'}},
        metadata={'source': 'loop', 'writes': {'node_a': {'foo': 'a', 'bar': ['a']}}, 'step': 1},
        created_at='2024-08-29T19:19:38.819946+00:00',
        parent_config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1ef663ba-28f4-6b4a-8000-ca575a13d36a'}},
        tasks=(PregelTask(id='6fb7314f-f114-5413-a1f3-d37dfe98ff44', name='node_b', error=None, interrupts=()),),
    ),
    StateSnapshot(
        values={'foo': '', 'bar': []},
        next=('node_a',),
        config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1ef663ba-28f4-6b4a-8000-ca575a13d36a'}},
        metadata={'source': 'loop', 'writes': None, 'step': 0},
        created_at='2024-08-29T19:19:38.817813+00:00',
        parent_config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1ef663ba-28f0-6c66-bfff-6723431e8481'}},
        tasks=(PregelTask(id='f1b14528-5ee5-579c-949b-23ef9bfbed58', name='node_a', error=None, interrupts=()),),
    ),
    StateSnapshot(
        values={'bar': []},
        next=('__start__',),
        config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1ef663ba-28f0-6c66-bfff-6723431e8481'}},
        metadata={'source': 'input', 'writes': {'foo': ''}, 'step': -1},
        created_at='2024-08-29T19:19:38.816205+00:00',
        parent_config=None,
        tasks=(PregelTask(id='6d27aa2e-d72b-5504-a36f-8620e54a76dd', name='__start__', error=None, interrupts=()),),
    )
]

특정 체크포인트 찾기 (Find a specific checkpoint)

상태 기록을 필터링해 특정 기준과 일치하는 체크포인트를 찾을 수 있어요:

history = list(graph.get_state_history(config))

# Find the checkpoint before a specific node executed
before_node_b = next(s for s in history if s.next == ("node_b",))

# Find a checkpoint by step number
step_2 = next(s for s in history if s.metadata["step"] == 2)

# Find checkpoints created by update_state
forks = [s for s in history if s.metadata["source"] == "update"]

# Find the checkpoint where an interrupt occurred
interrupted = next(
    s for s in history
    if s.tasks and any(t.interrupts for t in s.tasks)
)

재생 (Replay)

재생은 이전 체크포인트에서 단계를 다시 실행합니다. 이전 checkpoint_id로 그래프를 호출하면 그 체크포인트 이후의 노드를 다시 실행해요. 체크포인트 이전의 노드는 건너뜁니다(결과가 이미 저장되어 있으므로). 체크포인트 이후의 노드는 재실행되는데, LLM 호출, API 요청, interrupts를 포함하며 이들은 재생 중 항상 다시 트리거됩니다.

이전 실행 재생에 대한 전체 내용과 코드 예시는 Time travel을 참고하세요.

상태 갱신 (Update state)

update_state로 그래프 상태를 편집할 수 있어요. 이는 갱신된 값으로 새 체크포인트를 만들며, 원래 체크포인트를 수정하지는 않습니다. 갱신은 노드 갱신과 동일하게 취급됩니다. 값은 reducer 함수가 정의돼 있을 때 통과되므로, 리듀서가 있는 채널은 값을 덮어쓰기보다 누적합니다.

as_node를 선택적으로 지정해 갱신이 어느 노드에서 온 것인지 제어할 수 있고, 이는 다음에 실행할 노드에 영향을 줍니다. 자세한 내용은 Time travel: as_node를 참고하세요.

내구성 모드 (Durability modes)

LangGraph는 성능과 데이터 일관성을 조절할 수 있는 세 가지 내구성 모드를 지원해요. 그래프 실행 메서드를 호출할 때 내구성 모드를 지정할 수 있습니다:

graph.stream(
    {"input": "test"},
    durability="sync"
)

내구성 모드는 내구성이 낮은 것부터 높은 순으로 다음과 같습니다:

• "exit": LangGraph는 그래프 실행이 끝날 때(성공, 오류, 또는 human-in-the-loop 인터럽트로 인해)만 변경사항을 영속화해요. 장기 실행 그래프에서 최고의 성능을 제공하지만, 중간 상태가 저장되지 않으므로 실행 중 시스템 장애(예: 프로세스 크래시)에서 복구할 수 없습니다.
• "async": LangGraph는 다음 단계가 실행되는 동안 비동기로 변경사항을 영속화해요. 좋은 성능과 내구성을 제공하지만, 실행 중 프로세스가 크래시하면 체크포인트를 쓰지 못할 작은 위험이 있습니다.
• "sync": LangGraph는 다음 단계를 시작하기 전에 동기적으로 변경사항을 영속화해요. 실행을 계속하기 전에 모든 체크포인트를 쓰도록 보장해 높은 내구성을 제공하며, 일부 성능 오버헤드가 있습니다.

체크포인트 저장 최적화 (Optimize checkpoint storage)

기본적으로 LangGraph 체크포인트는 각 슈퍼스텝에서 모든 상태 채널의 전체 값을 씁니다. 다중 턴 대화처럼 큰 누적이 있는 장기 실행 스레드에서는 시간이 지나며 저장 공간이 크게 늘 수 있어요.

DeltaChannel은 전체 누적 값 대신 증분 델타만 저장하므로, append 중심 채널의 체크포인트 크기를 크게 줄여줍니다. 사용법과 저장공간-대-지연시간 트레이드오프는 DeltaChannel을 참고하세요.

`DeltaChannel`은 `langgraph>=1.2`가 필요하며 현재 베타 상태입니다. API는 향후 릴리스에서 바뀔 수 있어요.

체크포인터 라이브러리 (Checkpointer libraries)

내부적으로 체크포인팅은 BaseCheckpointSaver 인터페이스를 따르는 체크포인터 객체가 구동합니다. LangGraph는 모두 독립적으로 설치 가능한 라이브러리로 구현된 여러 체크포인터 구현을 제공합니다.

사용 가능한 프로바이더 목록은 [checkpointer integrations](/oss/python/integrations/checkpointers/index)를 참고하세요.

• langgraph-checkpoint: 체크포인터 세이버(BaseCheckpointSaver)와 직렬화/역직렬화 인터페이스(SerializerProtocol)의 기본 인터페이스. 실험용 인메모리 체크포인터 구현(InMemorySaver)을 포함합니다. LangGraph에는 langgraph-checkpoint가 포함되어 있어요.
• langgraph-checkpoint-sqlite: SQLite 데이터베이스를 사용하는 LangGraph 체크포인터 구현(SqliteSaver / AsyncSqliteSaver). 실험과 로컬 워크플로우에 이상적. 별도 설치 필요.
• langgraph-checkpoint-postgres: LangSmith에서 사용하는 Postgres 데이터베이스 기반 고급 체크포인터(PostgresSaver / AsyncPostgresSaver). 프로덕션 사용에 이상적. 별도 설치 필요.
• langgraph-checkpoint-mongodb: MongoDB를 사용하는 고급 체크포인터(MongoDBSaver / AsyncMongoDBSaver). 프로덕션 사용에 이상적. 별도 설치 필요.
• langchain-azure-cosmosdb: Azure Cosmos DB for NoSQL을 사용하는 LangGraph 체크포인터 구현(CosmosDBSaverSync / CosmosDBSaver). Azure와 함께 프로덕션 사용에 이상적. 동기·비동기 연산을 모두 지원하며 Microsoft Entra ID 인증을 사용. 별도 설치 필요.

체크포인터 인터페이스 (Checkpointer interface)

각 체크포인터는 BaseCheckpointSaver 인터페이스를 따르며 다음 메서드를 구현합니다:

• .put - 체크포인트를 그 구성과 메타데이터와 함께 저장합니다.
• .put_writes - 체크포인트에 연결된 중간 쓰기(즉 보류 중인 쓰기)를 저장합니다.
• .get_tuple - 주어진 구성(thread_id와 checkpoint_id)으로 체크포인트 튜플을 가져옵니다. graph.get_state()에서 StateSnapshot을 채우는 데 사용됩니다.
• .list - 주어진 구성과 필터 기준과 일치하는 체크포인트를 나열합니다. graph.get_state_history()의 상태 기록을 채우는 데 사용됩니다.

체크포인터를 비동기 그래프 실행(.ainvoke, .astream, .abatch로 실행)과 함께 사용하면 위 메서드의 비동기 버전(.aput, .aput_writes, .aget_tuple, .alist)이 사용됩니다.

그래프를 비동기로 실행할 때는 [`InMemorySaver`](https://reference.langchain.com/python/langgraph/checkpoints/#langgraph.checkpoint.memory.InMemorySaver) 또는 Sqlite/Postgres 체크포인터의 비동기 버전([`AsyncSqliteSaver`](https://reference.langchain.com/python/langgraph/checkpoints/#langgraph.checkpoint.sqlite.aio.AsyncSqliteSaver) / [`AsyncPostgresSaver`](https://reference.langchain.com/python/langgraph/checkpoints/#langgraph.checkpoint.postgres.aio.AsyncPostgresSaver))을 사용할 수 있어요.

직렬화 (Serializer)

체크포인터가 그래프 상태를 저장할 때 상태의 채널 값을 직렬화해야 합니다. 이는 직렬화 객체(serializer)로 처리됩니다.

langgraph_checkpoint는 직렬화기 구현을 위한 protocol을 정의하며, LangChain·LangGraph 원시형, datetime, enum 등 다양한 타입을 처리하는 기본 구현(JsonPlusSerializer)을 제공합니다.

pickle을 이용한 직렬화 (Serialization with pickle)

기본 직렬화기인 JsonPlusSerializer는 내부적으로 ormsgpack과 JSON을 사용하므로 모든 타입의 객체에는 적합하지 않습니다.

msgpack 인코더가 지원하지 않는 객체(Pandas dataframe 등)에 pickle로 폴백하려면 JsonPlusSerializer의 pickle_fallback 인자를 사용할 수 있어요:

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer

# ... Define the graph ...
graph.compile(
    checkpointer=InMemorySaver(serde=JsonPlusSerializer(pickle_fallback=True))
)

암호화 (Encryption)

체크포인터는 모든 영속화된 상태를 선택적으로 암호화할 수 있습니다. 활성화하려면 EncryptedSerializer 인스턴스를 어떤 BaseCheckpointSaver 구현의 serde 인자에 전달하세요. 암호화 직렬화기를 만드는 가장 쉬운 방법은 from_pycryptodome_aes를 사용하는 것으로, LANGGRAPH_AES_KEY 환경 변수에서 AES 키를 읽습니다(또는 key 인자를 받습니다):

import sqlite3

from langgraph.checkpoint.serde.encrypted import EncryptedSerializer
from langgraph.checkpoint.sqlite import SqliteSaver

serde = EncryptedSerializer.from_pycryptodome_aes()  # reads LANGGRAPH_AES_KEY
checkpointer = SqliteSaver(sqlite3.connect("checkpoint.db"), serde=serde)

from langgraph.checkpoint.serde.encrypted import EncryptedSerializer
from langgraph.checkpoint.postgres import PostgresSaver

serde = EncryptedSerializer.from_pycryptodome_aes()
checkpointer = PostgresSaver.from_conn_string("postgresql://...", serde=serde)
checkpointer.setup()

LangSmith에서 실행할 때는 LANGGRAPH_AES_KEY가 있을 때마다 암호화가 자동으로 활성화되므로, 환경 변수만 제공하면 됩니다. 다른 암호화 방식을 쓰려면 CipherProtocol을 구현해 EncryptedSerializer에 제공하세요.

커스텀 체크포인터 만들기 (Build a custom checkpointer)

만들면서 [적합성 테스트 스위트](#testing-with-the-conformance-suite)로 구현을 검증하세요. 다섯 가지 기본 메서드와 delta channel 같은 확장 기능을 모두 다룹니다. 배포 전에 CI에서 실행하세요.

이 섹션에서는 커스텀 스토리지 백엔드를 위해 BaseCheckpointSaver를 처음부터 구현하는 방법을 다룹니다. 이미 동작하는 체크포인터가 있고 delta channel 지원만 추가하려면 Delta channel 지원로 건너뛰세요.

개요 (Overview)

LangGraph의 영속성 레이어는 두 가지 저장소 추상화 위에 구축됩니다:

• 체크포인트 테이블(Checkpoints table) — 슈퍼스텝당 한 행. 직렬화된 그래프 상태(channel_values, channel_versions, versions_seen)를 저장하고 부모 체크포인트에 연결합니다.
• 쓰기 테이블(Writes table) — 슈퍼스텝 내 노드 출력당 한 행. 체크포인트에 연결된 (task_id, channel, value) 튜플을 저장합니다.

체크포인터는 두 테이블을 모두 관리합니다. put은 체크포인트 행을 쓰고, put_writes는 노드 출력 행을 쓰며, get_tuple은 둘 다 다시 CheckpointTuple으로 읽어 옵니다.

기본 계약 (Base contract)

BaseCheckpointSaver를 서브클래스화하고 다음 다섯 메서드를 구현하세요. 모두 필수입니다 — 기본 메서드가 빠지면 런타임에 NotImplementedError가 발생합니다.

from collections.abc import AsyncIterator, Iterator, Sequence
from typing import Any
from langchain_core.runnables import RunnableConfig
from langgraph.checkpoint.base import (
    BaseCheckpointSaver,
    ChannelVersions,
    Checkpoint,
    CheckpointMetadata,
    CheckpointTuple,
)

class MyCheckpointer(BaseCheckpointSaver):
    async def aput(
        self,
        config: RunnableConfig,
        checkpoint: Checkpoint,
        metadata: CheckpointMetadata,
        new_versions: ChannelVersions,
    ) -> RunnableConfig:
        ...

    async def aput_writes(
        self,
        config: RunnableConfig,
        writes: Sequence[tuple[str, Any]],
        task_id: str,
        task_path: str = "",
    ) -> None:
        ...

    async def aget_tuple(self, config: RunnableConfig) -> CheckpointTuple | None:
        ...

    async def alist(
        self,
        config: RunnableConfig | None,
        *,
        filter: dict[str, Any] | None = None,
        before: RunnableConfig | None = None,
        limit: int | None = None,
    ) -> AsyncIterator[CheckpointTuple]:
        ...
        yield  # make this an async generator

    async def adelete_thread(self, thread_id: str) -> None:
        ...

put / aput

체크포인트 행 하나를 저장하고, 저장된 checkpoint_id가 담긴 갱신된 config를 반환합니다.

핵심 요구사항:

• self.serde.dumps_typed(checkpoint)로 체크포인트를 직렬화하세요 — 이것이 delta channel이 쓰는 _DeltaSnapshot blob을 포함한 모든 LangGraph 네이티브 타입을 처리합니다.
• metadata는 전체를 저장하세요 — 알 수 없는 키를 버리지 마세요. LangGraph는 마이너 릴리스에서 새 메타데이터 필드(예: delta channel용 counters_since_delta_snapshot)를 추가하는데, 이를 조용히 버리면 기능이 깨집니다.
• 부모 체크포인트 ID로 config["configurable"].get("checkpoint_id")를 저장해서 get_tuple이 parent_config를 채울 수 있게 하세요.

async def aput(self, config, checkpoint, metadata, new_versions):
    thread_id = config["configurable"]["thread_id"]
    checkpoint_ns = config["configurable"]["checkpoint_ns"]
    checkpoint_id = checkpoint["id"]
    parent_id = config["configurable"].get("checkpoint_id")

    type_, blob = self.serde.dumps_typed(checkpoint)
    serialized_metadata = self.serde.dumps_typed(metadata)

    await self.db.execute(
        "INSERT INTO checkpoints (...) VALUES (...)",
        thread_id, checkpoint_ns, checkpoint_id, parent_id,
        type_, blob, *serialized_metadata,
    )
    return {
        "configurable": {
            "thread_id": thread_id,
            "checkpoint_ns": checkpoint_ns,
            "checkpoint_id": checkpoint_id,
        }
    }

put_writes / aput_writes

현재 슈퍼스텝 내 단일 태스크에 대한 노드 출력 행을 저장합니다. 이 행들은 (thread_id, checkpoint_ns, checkpoint_id)로 체크포인트에 연결됩니다.

async def aput_writes(self, config, writes, task_id, task_path=""):
    thread_id = config["configurable"]["thread_id"]
    checkpoint_ns = config["configurable"]["checkpoint_ns"]
    checkpoint_id = config["configurable"]["checkpoint_id"]

    rows = []
    for idx, (channel, value) in enumerate(writes):
        type_, blob = self.serde.dumps_typed(value)
        final_idx = WRITES_IDX_MAP.get(channel, idx)
        rows.append((thread_id, checkpoint_ns, checkpoint_id,
                      task_id, task_path, final_idx, channel, type_, blob))

    await self.db.executemany("INSERT INTO writes (...) VALUES (...)", rows)

WRITES_IDX_MAP을 langgraph.checkpoint.base에서 import하세요. 특수 채널(__error__, __interrupt__ 등)을 예약된 음수 인덱스로 매핑해 일반 쓰기 인덱스와 충돌하지 않게 합니다.

get_tuple / aget_tuple

체크포인트를 조회합니다. config에는 다음이 포함될 수 있습니다:

• checkpoint_id 없음 — 스레드 + 네임스페이스에 대한 최신 체크포인트를 반환합니다.
• 특정 checkpoint_id — 해당 정확한 체크포인트를 반환합니다.

두 경로 모두 올바르게 동작해야 합니다. 특정 ID 경로는 time travel에, 그리고 결정적으로 매 그래프 호출의 delta channel 상태 재구성에 사용됩니다(Delta channel 지원 참고). 특정 ID 조회가 깨지면 delta channel 상태가 조용히 손상됩니다.

async def aget_tuple(self, config):
    thread_id = config["configurable"]["thread_id"]
    checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
    checkpoint_id = config["configurable"].get("checkpoint_id")

    if checkpoint_id:
        row = await self.db.fetchone(
            "SELECT * FROM checkpoints "
            "WHERE thread_id=? AND checkpoint_ns=? AND checkpoint_id=?",
            thread_id, checkpoint_ns, checkpoint_id,
        )
    else:
        row = await self.db.fetchone(
            "SELECT * FROM checkpoints "
            "WHERE thread_id=? AND checkpoint_ns=? "
            "ORDER BY checkpoint_id DESC LIMIT 1",
            thread_id, checkpoint_ns,
        )

    if row is None:
        return None

    writes = await self.db.fetchall(
        "SELECT task_id, channel, type, value FROM writes "
        "WHERE thread_id=? AND checkpoint_ns=? AND checkpoint_id=? "
        "ORDER BY task_id, idx",
        thread_id, checkpoint_ns, row["checkpoint_id"],
    )
    pending_writes = [
        (w["task_id"], w["channel"], self.serde.loads_typed((w["type"], w["value"])))
        for w in writes
    ]

    checkpoint = self.serde.loads_typed((row["type"], row["blob"]))
    metadata = self.serde.loads_typed((row["metadata_type"], row["metadata"]))

    parent_config = None
    if row["parent_checkpoint_id"]:
        parent_config = {
            "configurable": {
                "thread_id": thread_id,
                "checkpoint_ns": checkpoint_ns,
                "checkpoint_id": row["parent_checkpoint_id"],
            }
        }

    return CheckpointTuple(
        config={
            "configurable": {
                "thread_id": thread_id,
                "checkpoint_ns": checkpoint_ns,
                "checkpoint_id": row["checkpoint_id"],
            }
        },
        checkpoint=checkpoint,
        metadata=metadata,
        parent_config=parent_config,
        pending_writes=pending_writes,
    )

**행 키/인덱스 설계는 특정 ID 조회에 중요합니다.** 저장소가 `checkpoint_id`를 포함하지 않는 시간순 키(예: 역순 타임스탬프)를 사용한다면 id로 직접 행을 읽을 수 없습니다. 행 키에 `checkpoint_id`를 인코딩하거나 보조 인덱스를 만드세요. 매 조회마다 값 필터로 스캔하는 방식은 동작하지만 확장되지 않습니다.

list / alist

스레드의 체크포인트를 최신순으로 반환합니다. before(해당 config의 checkpoint_id보다 오래된 체크포인트만 반환)와 limit을 존중합니다.

delete_thread / adelete_thread

스레드의 모든 체크포인트와 쓰기를 삭제합니다. 체크포인트 행과 쓰기 행을 모두 삭제해야 해요.

행 키 / 인덱스 설계 (Row key / index design)

체크포인트를 저장하고 인덱싱하는 방식은 정확성과 성능에 직접 영향을 줍니다.

권장 스키마(SQL):

CREATE TABLE checkpoints (
    thread_id          TEXT NOT NULL,
    checkpoint_ns      TEXT NOT NULL DEFAULT '',
    checkpoint_id      TEXT NOT NULL,   -- ULID, lexicographically sortable newest-last
    parent_checkpoint_id TEXT,
    type               TEXT,
    checkpoint         BYTEA,
    metadata           JSONB,
    PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id)
);

CREATE TABLE writes (
    thread_id     TEXT NOT NULL,
    checkpoint_ns TEXT NOT NULL DEFAULT '',
    checkpoint_id TEXT NOT NULL,
    task_id       TEXT NOT NULL,
    task_path     TEXT NOT NULL DEFAULT '',
    idx           INTEGER NOT NULL,
    channel       TEXT NOT NULL,
    type          TEXT,
    value         BYTEA,
    PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id, task_id, task_path, idx)
);

checkpoint_id가 ULID이므로 사전순으로 정렬됩니다 — 값이 클수록 더 새 것입니다. "최신 가져오기"는 ORDER BY checkpoint_id DESC LIMIT 1, "id로 가져오기"는 기본 키에 대한 동등 조회입니다.

비-SQL 저장소의 경우: 같은 원칙이 적용됩니다. 어떤 키 방식을 쓰든 (thread_id, checkpoint_ns, checkpoint_id)로 직접 조회하는 것이 O(1) 또는 그에 근접해야 합니다. 스레드의 모든 행을 스캔해야만 id로 체크포인트를 찾는 설계는 피하세요.

직렬화 (Serialization)

체크포인트, 쓰기, 메타데이터에는 항상 self.serde(BaseCheckpointSaver에서 상속, 기본은 JsonPlusSerializer)를 사용하세요. 메타데이터에 pickle을 직접 쓰지 마세요 — 동작은 하지만 JsonPlusSerializer가 사람이 읽을 수 있는 출력을 만들고 버전 관리를 더 잘 처리합니다.

JsonPlusSerializer는 모든 LangGraph 네이티브 타입을 자동으로 처리합니다:

• _DeltaSnapshot — delta channel이 쓰는 sentinel blob (msgpack ext code 7)
• Pydantic v2 모델, dataclass, numpy 배열, datetime, enum 등

커스텀 직렬화기를 작성한다면 langgraph.checkpoint.serde.types의 _DeltaSnapshot을 왕복(round-trip)할 수 있는지 확인하세요.

확장 기능 (Extended capabilities)

이 메서드들은 선택 사항이지만 추가 Agent Server 기능을 활성화합니다. 스토리지 백엔드가 효율적으로 지원할 수 있다면 구현하세요.

메서드	활성화하는 것
adelete_for_runs	Rollback multitask strategy
acopy_thread	효율적인 스레드 포킹
aprune	스레드 기록 정리(pruning)
aget_delta_channel_history	효율적인 delta channel 상태 재구성 (아래 참고)

Agent Server는 시작 시 체크포인터가 구현한 기능을 자동 감지하고 해당 기능을 활성화합니다.

Delta channel 지원 (Delta channel support)

**DeltaChannel은 베타입니다.** 설계가 안정화되면서 API와 온디스크 표현이 바뀔 수 있어요.

DeltaChannel은 체크포인트 blob에 전체 채널 값 대신 sentinel(MISSING)만 저장하는 리듀서 채널이에요. 상태는 리듀서를 통해 조상의 쓰기를 재생함으로써 재구성됩니다. 이렇게 하면 시간이 지나며 누적되는 messages 같은 채널에서 체크포인트 blob을 단계당 O(1)로 만들어줍니다(N 대신).

런타임이 필요한 것 (What the runtime needs)

channel_values에 없는 delta channel이 있는 체크포인트를 로드할 때 LangGraph는 saver.get_delta_channel_history(config=config, channels=[...])를 호출합니다. 이는 채널마다 다음을 반환합니다:

• writes — 조상 체인의 해당 채널에 대한 모든 쓰기, 가장 오래된 것부터, 가장 가까운 스냅샷까지.
• seed (선택) — 하나가 있는 가장 가까운 조상에 저장된 _DeltaSnapshot blob; 스냅샷 없이 루트까지 도달하면 없음.

런타임은 그런 다음 channel.from_checkpoint(seed)와 channel.replay_writes(writes)를 호출해 실제 값을 재구성합니다.

기본 구현 (Default implementation)

BaseCheckpointSaver는 올바른 get_tuple 구현과 함께 동작하는 기본 get_delta_channel_history를 제공합니다:

# Simplified from BaseCheckpointSaver
def get_delta_channel_history(self, *, config, channels):
    target = self.get_tuple(config)          # load the head checkpoint
    cursor = target.parent_config            # walk from its parent
    collected = {ch: [] for ch in channels}
    seed = {}
    remaining = set(channels)

    while cursor and remaining:
        tup = self.get_tuple(cursor)         # ← requires correct by-id lookup
        if tup is None:
            break
        for write in reversed(tup.pending_writes or []):
            if write[1] in remaining:
                collected[write[1]].append(write)
        for ch in list(remaining):
            if ch in tup.checkpoint["channel_values"]:
                seed[ch] = tup.checkpoint["channel_values"][ch]
                remaining.discard(ch)
        cursor = tup.parent_config

    return {
        ch: {"writes": list(reversed(collected[ch])), **({"seed": seed[ch]} if ch in seed else {})}
        for ch in channels
    }

결정적 의존성: get_tuple(cursor)는 항상 특정 checkpoint_id(부모의 id)로 호출됩니다. 그 조회가 None을 반환하면 워크(stroll)는 즉시 멈추고 모든 delta channel이 오류 없이 조용히 빈 상태로 재구성됩니다. 그래서 get_tuple의 특정 ID 경로가 올바르게 동작해야 합니다.

성능 오버라이드 (Performance override)

기본 워크는 조상 체크포인트마다 get_tuple 호출을 하나씩 사용합니다. 쿼리 지원이 좋은 백엔드는 get_delta_channel_history(및 그 비동기 쌍)를 오버라이드해 조상 체인과 쓰기를 두 개의 쿼리로 조회하세요:

async def aget_delta_channel_history(self, *, config, channels):
    if not channels:
        return {}

    thread_id = config["configurable"]["thread_id"]
    checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
    checkpoint_id = config["configurable"]["checkpoint_id"]

    # Stage 1: stream ancestors newest-first until every channel has a seed
    ancestors = await self.db.fetchall(
        "SELECT checkpoint_id, parent_checkpoint_id, type, checkpoint "
        "FROM checkpoints "
        "WHERE thread_id=? AND checkpoint_ns=? AND checkpoint_id < ? "
        "ORDER BY checkpoint_id DESC",
        thread_id, checkpoint_ns, checkpoint_id,
    )

    chain_by_ch: dict[str, list[str]] = {ch: [] for ch in channels}
    seed_by_ch: dict[str, Any] = {}
    remaining = set(channels)
    cur_id = config["configurable"]["checkpoint_id"]

    for row in ancestors:
        if not remaining:
            break
        parent_id = row["parent_checkpoint_id"]
        ckpt = self.serde.loads_typed((row["type"], row["checkpoint"]))
        cv = ckpt.get("channel_values") or {}
        for ch in list(remaining):
            chain_by_ch[ch].append(row["checkpoint_id"])
            if ch in cv:
                seed_by_ch[ch] = cv[ch]
                remaining.discard(ch)
        cur_id = parent_id

    # Stage 2: fetch writes for each channel's ancestor chain in one query
    result: dict[str, DeltaChannelHistory] = {}
    for ch in channels:
        chain = chain_by_ch[ch]
        if not chain:
            entry: DeltaChannelHistory = {"writes": []}
            if ch in seed_by_ch:
                entry["seed"] = seed_by_ch[ch]
            result[ch] = entry
            continue

        write_rows = await self.db.fetchall(
            f"SELECT checkpoint_id, task_id, idx, type, value FROM writes "
            f"WHERE thread_id=? AND checkpoint_ns=? AND channel=? "
            f"AND checkpoint_id IN ({','.join('?' * len(chain))})"
            f"ORDER BY checkpoint_id, task_id, idx",
            thread_id, checkpoint_ns, ch, *chain,
        )
        writes_by_cid: dict[str, list[PendingWrite]] = {}
        for row in write_rows:
            cid = row["checkpoint_id"]
            value = self.serde.loads_typed((row["type"], row["value"]))
            writes_by_cid.setdefault(cid, []).append((row["task_id"], ch, value))

        # chain is newest-first; iterate oldest-first to get correct replay order
        collected: list[PendingWrite] = []
        for cid in reversed(chain):
            collected.extend(writes_by_cid.get(cid, []))

        entry = {"writes": collected}
        if ch in seed_by_ch:
            entry["seed"] = seed_by_ch[ch]
        result[ch] = entry

    return result

delta channel과 함께 정리하기 (Pruning with delta channels)

DeltaChannel 상태는 단일 체크포인트에 자족적이지 않아요 — 가장 가까운 _DeltaSnapshot까지의 조상 쓰기 체인에 의존합니다. prune이나 delete_for_runs를 구현한다면, 살아남는 체크포인트의 delta channel이 의존하는 쓰기 행을 삭제하면 안 됩니다.

안전한 옵션:

1. 정리 전에 워크 — 유지하려는 각 체크포인트에 대해 조상 체인을 워크하고 가장 가까운 _DeltaSnapshot까지의 모든 쓰기 행을 삭제 불가로 표시합니다.
2. 정리 전에 스냅샷 강제 — 유지하는 체크포인트에 channel_values[ch] = _DeltaSnapshot(reconstructed_value)를 다시 쓰고 조상을 자유롭게 삭제합니다.
3. delta channel 스레드에 대한 정리 건너뛰기 — 아직 정리가 필요 없다면 가장 안전한 단기 옵션입니다.

delta channel로 스레드 복사 (Copy thread with delta channels)

copy_thread를 구현할 때는 헤드 체크포인트만이 아니라 완전한 조상 체인을 복사하세요. 대상 스레드는 모든 delta channel에 대해 적어도 하나의 _DeltaSnapshot까지 거슬러 올라가는 쓰기 행이 있어야 합니다. 그렇지 않으면 복사 후 그 채널들이 빈 상태로 재구성됩니다.

적합성 스위트로 테스트 (Testing with the conformance suite)

langgraph-checkpoint-conformance는 delta channel 기록을 포함한 전체 계약에 대해 구현을 검증합니다:

pip install langgraph-checkpoint-conformance

import asyncio
from langgraph.checkpoint.conformance import checkpointer_test, validate

@checkpointer_test(name="MyCheckpointer")
async def my_checkpointer():
    async with MyCheckpointer.create() as saver:
        yield saver

async def main():
    report = await validate(my_checkpointer)
    report.print_report()
    # Fails the process if any base capability is missing or broken
    if not report.passed_all_base():
        raise RuntimeError("Checkpointer failed conformance suite")

asyncio.run(main())

이 스위트는 체크포인터가 구현한 확장 기능(aget_delta_channel_history 포함)을 자동 감지하고 각각에 대한 관련 테스트를 실행합니다. 배포 전에 CI의 일부로 실행하세요.

더 알아보기 (Learn more)

• Interrupts — human-in-the-loop 워크플로우.
• Add memory — 체크포인터로 대화 메모리 관리.
• Time travel — 과거 실행 재생·포킹.
• DeltaChannel — 델타 채널 사용과 트레이드오프.