LangGraph Checkpointer와 실행 재개

Checkpointer의 필요성

checkpointer(상태 저장기) 없이 compile한 그래프는 실행이 종료되면 상태가 보존되지 않으므로, 실행 도중 오류가 발생하면 처음부터 다시 실행해야 한다. checkpointer를 지정하면 각 단계가 끝날 때마다 상태가 체크포인트로 저장된다.

설정과 상태 확인

pythonrun.py

from typing import TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph


class State(TypedDict):
    step: str


builder = StateGraph(State)
builder.add_node("load", lambda state: {"step": "load"})
builder.add_node("save", lambda state: {"step": "save"})
builder.add_edge(START, "load")
builder.add_edge("load", "save")
builder.add_edge("save", END)
graph = builder.compile(checkpointer=InMemorySaver())

config = {"configurable": {"thread_id": "line-3"}}
graph.invoke({"step": ""}, config)

snapshot = graph.get_state(config)
print(snapshot.values)  # {'step': 'save'}
print(snapshot.next)    # ()  남은 노드가 없다

thread_id는 체크포인트를 식별하는 키이다. thread_id가 다르면 체크포인트가 별도로 저장되고, 같으면 저장된 상태에 이어서 실행된다.

단계마다 thread_id 단위로 저장되는 체크포인트

실행 재개

노드가 오류로 중단된 경우, 원인을 수정한 뒤 같은 config로 입력 자리에 None을 전달한다. 이미 완료된 노드는 다시 실행되지 않는다.

python

graph.invoke(None, config)

오류로 중단된 실행을 같은 config로 재개하는 흐름

저장소 선택

클래스	저장 위치	용도
`InMemorySaver`	메모리	시험용. 프로세스가 종료되면 소멸
`SqliteSaver`	SQLite 파일	로컬 작업
`PostgresSaver`	PostgreSQL	운영

뒤의 두 클래스를 사용하려면 langgraph-checkpoint-sqlite, langgraph-checkpoint-postgres를 별도로 설치한다.