Skip to main content

​
상태 관리 (Persistence)

LangGraph의 상태 관리는 체크포인터(Checkpointer)를 통해 그래프 실행 상태를 저장하고 복원하는 메커니즘입니다. 이를 통해 대화 이력 유지, 실행 재개, 타임 트래블 등이 가능합니다.

​
체크포인터

매 노드 실행 후 자동으로 상태가 저장되며, 이를 통해:
  • 대화 이력 유지: thread_id로 이전 대화 계속
  • 실행 재개: 중단된 지점에서 다시 시작
  • 타임 트래블: 과거 상태로 돌아가기

​
체크포인터 종류

체크포인터저장소영속성적합한 용도
MemorySaver메모리X개발, 테스트
SqliteSaverSQLiteO단일 서버 프로덕션
PostgresSaverPostgreSQLO분산 프로덕션

​
MemorySaver (개발용)

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

​
PostgresSaver (프로덕션)

pip install langgraph-checkpoint-postgres

from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://user:password@localhost:5432/langgraph"

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()  # 테이블 생성 (최초 1회)
    app = graph.compile(checkpointer=checkpointer)

​
thread_id 기반 대화 관리

thread_id로 독립적인 대화 세션을 관리합니다. 
from langchain_core.messages import HumanMessage

# 대화 1: 사용자 A
config_a = {"configurable": {"thread_id": "user-a-session-1"}}

app.invoke(
    {"messages": [HumanMessage(content="안녕, 나는 민수야")]},
    config=config_a,
)
app.invoke(
    {"messages": [HumanMessage(content="내 이름이 뭐였지?")]},
    config=config_a,
)
# → "민수"라고 답변 (같은 thread_id이므로 이전 대화 기억)

# 대화 2: 사용자 B (독립된 세션)
config_b = {"configurable": {"thread_id": "user-b-session-1"}}

app.invoke(
    {"messages": [HumanMessage(content="내 이름이 뭐였지?")]},
    config=config_b,
)
# → 모름 (다른 thread_id이므로 이전 대화 없음)

​
타임 트래블

과거 체크포인트로 돌아가서 다른 경로로 실행할 수 있습니다. 
# 현재까지의 모든 체크포인트 조회
states = list(app.get_state_history(config_a))

for state in states:
    print(f"체크포인트: {state.config}")
    print(f"메시지 수: {len(state.values.get('messages', []))}")

# 특정 체크포인트로 돌아가기
past_config = states[2].config  # 3번째 과거 상태
result = app.invoke(
    {"messages": [HumanMessage(content="다른 질문")]},
    config=past_config,
)

​
Durable Execution

장시간 실행되는 워크플로우에서도 중단과 재개가 가능합니다. 체크포인터가 매 노드 실행 후 상태를 저장하므로, 서버 재시작이나 장애 발생 시에도 마지막 체크포인트에서 실행을 재개할 수 있습니다. 
# 서버 재시작 후 실행 재개
config = {"configurable": {"thread_id": "long-running-task"}}

# 마지막 저장된 상태 확인
state = app.get_state(config)
if state.next:  # 아직 완료되지 않은 작업이 있음
    # 마지막 체크포인트에서 재개
    result = app.invoke(None, config=config)
프로덕션 체크리스트: 1) PostgresSaver 사용, 2) 의미 있는 thread_id 체계 설계 (예: user-{id}-session-{n}), 3) 오래된 체크포인트 정기 삭제.