State Vs Checkpointer 정리

• State: 작업대 위의 현재 작업 도구들
• Checkpointer: 작업 내역을 기록하는 노트/데이터베이스

State (현재 런타임)

• 범위: 단일 실행(single run) 내에서의 데이터 흐름
• 생명주기: 그래프 실행이 시작되고 끝날 때까지
• 목적: 노드 간 데이터 전달 및 공유
• 저장 위치: 메모리 (런타임 동안만 존재)

# State는 현재 실행 중인 워크플로우 내에서만 유효
class State(TypedDict):
    messages: List[str]
    current_step: int
    
def node1(state):
    # 이 state는 현재 실행 중에만 존재
    return {"current_step": state["current_step"] + 1}

Checkpointer (세션 간 영속성)

• 범위: 여러 실행(multiple runs)에 걸친 상태 보존
• 생명주기: 애플리케이션/프로세스를 넘어서 지속
• 목적: 대화 히스토리, 상태 복원, 시간 여행
• 저장 위치: 데이터베이스, 파일, Redis 등 영속 저장소

# Checkpointer는 여러 실행에 걸쳐 상태를 유지
checkpointer = SqliteSaver("chat_history.db")
app = workflow.compile(checkpointer=checkpointer)

# 첫 번째 실행 (월요일)
config = {"configurable": {"thread_id": "user_123"}}
app.invoke({"input": "안녕하세요"}, config)

# 두 번째 실행 (화요일) - 이전 대화 기억
app.invoke({"input": "어제 우리가 뭐라고 얘기했지?"}, config)

실제 사용 예시

챗봇 시나리오

# 1. State 정의 - 런타임 데이터 구조
class ChatState(TypedDict):
    messages: List[dict]      # 현재 대화
    current_topic: str        # 현재 주제
    sentiment: str           # 현재 감정 상태

# 2. Checkpointer 설정 - 영속성
checkpointer = SqliteSaver("chatbot.db")

# 3. 워크플로우 구성
workflow = StateGraph(ChatState)

def process_message(state: ChatState):
    # 현재 런타임에서 state 처리
    latest_message = state["messages"][-1]
    sentiment = analyze_sentiment(latest_message)
    
    return {
        "sentiment": sentiment,
        "current_topic": extract_topic(latest_message)
    }

workflow.add_node("process", process_message)
app = workflow.compile(checkpointer=checkpointer)

# 4. 실행
config = {"configurable": {"thread_id": "conversation_456"}}

# Day 1
result1 = app.invoke({
    "messages": [{"role": "user", "content": "오늘 기분이 좋아요"}]
}, config)
# State는 메모리에서 사라지지만, Checkpointer가 저장

# Day 2 - 새로운 프로세스/세션
result2 = app.invoke({
    "messages": [{"role": "user", "content": "어제 제가 뭐라고 했죠?"}]
}, config)
# Checkpointer가 이전 상태를 복원하여 State에 로드