체크포인터 (Checkpointers)

LangGraph의 체크포인터는 그래프 실행의 각 단계에서 그래프 상태를 체크포인트로 저장해요. 이를 통해 지속성(persistence), human-in-the-loop(사람 개입), 장애에 강한(fault-tolerant) 실행이 가능해집니다.

Tip: 체크포인트된 상태를 추적하고 에이전트가 세션을 넘어 어떻게 이어지는지 디버깅하려면 LangSmith를 활용하세요. tracing quickstart를 따라 설정할 수 있어요.

왜 체크포인터를 쓰나요

Human-in-the-loop: 체크포인터는 사람이 그래프 단계를 검사·중단·승인할 수 있게 해주는 human-in-the-loop 워크플로를 가능하게 해요. 이런 워크플로에는 체크포인터가 꼭 필요합니다. 사람은 언제든지 그래프의 현재 상태를 볼 수 있어야 하고, 그래프는 사람이 상태를 수정한 뒤 실행을 이어갈 수 있어야 하기 때문이에요. 예시는 Interrupts에서 확인할 수 있습니다.
대화 메모리를 체크포인터로 추가하고 관리하는 방법은 Add memory를 참고하세요.
Time travel(시간 여행): 체크포인터는 "time travel"을 지원해요. 사용자가 과거 그래프 실행을 재생(replay)해서 특정 그래프 단계를 검토하거나 디버깅할 수 있습니다.

핵심 개념

스레드 (Threads)

스레드는 체크포인터가 저장한 각 체크포인트에 부여되는 고유 ID(스레드 식별자)예요. 스레드는 일련의 runs가 쌓아온 누적 상태를 담고 있습니다. 실행(run)이 발생하면 어시스턴트의 그래프 state가 스레드에 영속됩니다.

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

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

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

체크포인트 (Checkpoints)

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

Super-steps

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

graph = workflow.compile(checkpointer=checkpointer)

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

상태 가져오기와 갱신하기

상태 이력 가져오기

이 예시에서 get_state_history의 출력은 다음과 같은 형태가 돼요.

재생 (Replay)

과거 실행을 재생하는 전체 예시와 코드는 Time travel에서 확인할 수 있어요.

상태 갱신 (Update state)

update_state를 사용해 그래프 상태를 수정할 수 있어요. 이 메서드는 갱신된 값으로 새 체크포인트를 생성하지, 원래 체크포인트를 수정하지는 않습니다.

갱신은 노드 갱신과 동일하게 처리돼요: reducer 함수가 정의되어 있으면 값을 reducer에 통과시키므로, reducer가 있는 채널은 값을 덮어쓰지 않고 누적(accumulate) 합니다.

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

지속성 모드 (Durability modes)

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

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

"exit": LangGraph는 그래프 실행이 끝날 때(성공, 오류, 또는 human-in-the-loop 인터럽트로 인해)만 변경 사항을 영속합니다.

체크포인트 저장 최적화

Warning: DeltaChannel는 langgraph>=1.2가 필요하며 현재 베타 상태예요. API는 향후 릴리스에서 변경될 수 있습니다.

체크포인터 라이브러리

내부적으로 체크포인팅은 BaseCheckpointSaver 인터페이스를 따르는 체크포인터 객체로 구현돼요. LangGraph는 여러 체크포인터 구현을 제공하는데, 모두 독립적으로 설치 가능한 라이브러리로 구현되어 있습니다.

langgraph-checkpoint: 체크포인터 saver의 기본 인터페이스(BaseCheckpointSaver)와 직렬화/역직렬화 인터페이스입니다.

체크포인터 인터페이스

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

AsyncSqliteSaver / AsyncPostgresSaver 체크포인터에 대해 자세히 알아볼 수 있어요.

직렬화 도구 (Serializer)

langgraph_checkpoint는 직렬화 도구를 구현하기 위한 protocol을 정의하고, 기본 구현으로 langchain과 LangGraph 프리미티브, datetime, enum 등 다양한 타입을 처리하는 JsonPlusSerializer를 제공합니다.

`pickle`을 이용한 직렬화

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

# ... 그래프 정의 ...
graph.compile(
    checkpointer=InMemorySaver(serde=JsonPlusSerializer(pickle_fallback=True))
)

암호화 (Encryption)

체크포인터는 저장되는 모든 상태를 선택적으로 암호화할 수 있어요. 활성화하려면 EncryptedSerializer 인스턴스를 아무 BaseCheckpointSaver 구현의 serde 인자로 전달하면 됩니다.

커스텀 체크포인터 만들기

기본 계약 (Base contract)

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,
)

put_writes / aput_writes

langgraph.checkpoint.base에서 WRITES_IDX_MAP을 import하세요. 이 매핑은 __error__, __interrupt__ 같은 특수 채널을 예약된 음수 인덱스로 연결해서, 일반 write 인덱스와 충돌하지 않게 해줍니다.

직렬화

JsonPlusSerializer는 모든 LangGraph 네이티브 타입을 자동으로 처리해요:

_DeltaSnapshot — delta 채널이 사용하는 sentinel blob (msgpack ext code 7)
Pydantic v2 모델, dataclass, numpy 배열, datetime, enum 등

Delta 채널 지원

Info: DeltaChannel은 베타 상태입니다. 디자인이 안정되기 전까지 API와 디스크 표현이 변경될 수 있어요.

런타임에 필요한 것

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

기본 구현

                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
    }

체크포인터 (Checkpointers)