모델 기반 의사결정을 수행하는 비결정적 시스템(예: LLM으로 구동되는 agent)을 다룰 때, 의사결정 과정을 자세히 살펴보는 것이 유용할 수 있습니다:
  1. 추론 이해하기: 성공적인 결과로 이어진 단계를 분석합니다.
  2. 실수 디버깅하기: 오류가 발생한 위치와 이유를 파악합니다.
  3. 대안 탐색하기: 더 나은 솔루션을 찾기 위해 다양한 경로를 테스트합니다.
LangGraph는 이러한 사용 사례를 지원하기 위해 time travel 기능을 제공합니다. 구체적으로, 이전 checkpoint에서 실행을 재개할 수 있습니다 — 동일한 state를 재생하거나 수정하여 대안을 탐색할 수 있습니다. 모든 경우에 과거 실행을 재개하면 히스토리에 새로운 분기가 생성됩니다. LangGraph에서 time-travel을 사용하려면:
  1. graph 실행하기: invoke 또는 stream method를 사용하여 초기 입력으로 graph를 실행합니다.
  2. 기존 thread에서 checkpoint 식별하기: get_state_history method를 사용하여 특정 thread_id에 대한 실행 히스토리를 검색하고 원하는 checkpoint_id를 찾습니다. 또는 실행을 일시 중지하려는 node 앞에 interrupt를 설정할 수 있습니다. 그런 다음 해당 interrupt까지 기록된 가장 최근 checkpoint를 찾을 수 있습니다.
  3. graph state 업데이트하기 (선택사항): update_state method를 사용하여 checkpoint에서 graph의 state를 수정하고 대체 state에서 실행을 재개합니다.
  4. checkpoint에서 실행 재개하기: 입력을 None으로 하고 적절한 thread_idcheckpoint_id를 포함하는 configuration으로 invoke 또는 stream method를 사용합니다.
Time-travel의 개념적 개요는 Time travel을 참조하세요.

workflow에서

이 예제는 농담 주제를 생성하고 LLM을 사용하여 농담을 작성하는 간단한 LangGraph workflow를 구축합니다. graph를 실행하고, 과거 실행 checkpoint를 검색하고, 선택적으로 state를 수정하고, 선택한 checkpoint에서 실행을 재개하여 대체 결과를 탐색하는 방법을 보여줍니다.

Setup

먼저 필요한 패키지를 설치해야 합니다
%%capture --no-stderr
pip install --quiet -U langgraph langchain_anthropic
다음으로, Anthropic(사용할 LLM)에 대한 API key를 설정해야 합니다
import getpass
import os


def _set_env(var: str):
    if not os.environ.get(var):
        os.environ[var] = getpass.getpass(f"{var}: ")


_set_env("ANTHROPIC_API_KEY")
LangSmith에 가입하여 LangGraph 프로젝트의 문제를 빠르게 발견하고 성능을 개선하세요. LangSmith를 사용하면 trace 데이터를 활용하여 LangGraph로 구축된 LLM 앱을 디버그, 테스트 및 모니터링할 수 있습니다.
import uuid

from typing_extensions import TypedDict, NotRequired
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langgraph.checkpoint.memory import InMemorySaver


class State(TypedDict):
    topic: NotRequired[str]
    joke: NotRequired[str]


model = init_chat_model(
    "anthropic:claude-sonnet-4-5",
    temperature=0,
)


def generate_topic(state: State):
    """LLM call to generate a topic for the joke"""
    msg = model.invoke("Give me a funny topic for a joke")
    return {"topic": msg.content}


def write_joke(state: State):
    """LLM call to write a joke based on the topic"""
    msg = model.invoke(f"Write a short joke about {state['topic']}")
    return {"joke": msg.content}


# Build workflow
workflow = StateGraph(State)

# Add nodes
workflow.add_node("generate_topic", generate_topic)
workflow.add_node("write_joke", write_joke)

# Add edges to connect nodes
workflow.add_edge(START, "generate_topic")
workflow.add_edge("generate_topic", "write_joke")
workflow.add_edge("write_joke", END)

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

1. Run the graph

config = {
    "configurable": {
        "thread_id": uuid.uuid4(),
    }
}
state = graph.invoke({}, config)

print(state["topic"])
print()
print(state["joke"])
Output:
How about "The Secret Life of Socks in the Dryer"? You know, exploring the mysterious phenomenon of how socks go into the laundry as pairs but come out as singles. Where do they go? Are they starting new lives elsewhere? Is there a sock paradise we don't know about? There's a lot of comedic potential in the everyday mystery that unites us all!

# The Secret Life of Socks in the Dryer

I finally discovered where all my missing socks go after the dryer. Turns out they're not missing at all—they've just eloped with someone else's socks from the laundromat to start new lives together.

My blue argyle is now living in Bermuda with a red polka dot, posting vacation photos on Sockstagram and sending me lint as alimony.

2. Identify a checkpoint

# The states are returned in reverse chronological order.
states = list(graph.get_state_history(config))

for state in states:
    print(state.next)
    print(state.config["configurable"]["checkpoint_id"])
    print()
Output:
()
1f02ac4a-ec9f-6524-8002-8f7b0bbeed0e

('write_joke',)
1f02ac4a-ce2a-6494-8001-cb2e2d651227

('generate_topic',)
1f02ac4a-a4e0-630d-8000-b73c254ba748

('__start__',)
1f02ac4a-a4dd-665e-bfff-e6c8c44315d9
# This is the state before last (states are listed in chronological order)
selected_state = states[1]
print(selected_state.next)
print(selected_state.values)
Output:
('write_joke',)
{'topic': 'How about "The Secret Life of Socks in the Dryer"? You know, exploring the mysterious phenomenon of how socks go into the laundry as pairs but come out as singles. Where do they go? Are they starting new lives elsewhere? Is there a sock paradise we don\\'t know about? There\\'s a lot of comedic potential in the everyday mystery that unites us all!'}

3. Update the state

update_state는 새로운 checkpoint를 생성합니다. 새 checkpoint는 동일한 thread와 연결되지만 새로운 checkpoint ID를 갖습니다.
new_config = graph.update_state(selected_state.config, values={"topic": "chickens"})
print(new_config)
Output:
{'configurable': {'thread_id': 'c62e2e03-c27b-4cb6-8cea-ea9bfedae006', 'checkpoint_ns': '', 'checkpoint_id': '1f02ac4a-ecee-600b-8002-a1d21df32e4c'}}

4. Resume execution from the checkpoint

graph.invoke(None, new_config)
Output:
{'topic': 'chickens',
 'joke': 'Why did the chicken join a band?\n\nBecause it had excellent drumsticks!'}

Connect these docs programmatically to Claude, VSCode, and more via MCP for real-time answers.
I