타임트래블 사용하기

타임트래블 사용하기 (Use time-travel)

과거 실행을 재생하고, 포크를 통해 대안 경로를 탐색하는 LangGraph의 타임트래블 기능을 다뤄요.

개요 (Overview)

LangGraph는 체크포인트를 통해 타임트래블을 지원해요.

  • Replay(재생): 이전 체크포인트부터 다시 시도해요.
  • Fork(포크): 수정된 상태로 이전 체크포인트에서 분기해 대안 경로를 탐색해요.

둘 다 이전 체크포인트에서 재개하는 방식으로 동작해요. 체크포인트 이전의 노드는 다시 실행되지 않아요(결과가 이미 저장돼 있어요). 체크포인트 이후의 노드는 LLM 호출, API 요청, 인터럽트를 포함해 다시 실행돼요. (결과가 달라질 수 있어요.)

출처: 문서

본문

Replay (재생)

이전 체크포인트의 config로 그래프를 호출하면 그 지점부터 다시 재생해요.

Replay는 노드를 다시 실행해요. 캐시에서 읽는 것만이 아니에요. LLM 호출, API 요청, 인터럽트가 다시 발화되어 다른 결과를 반환할 수 있어요. 마지막 체크포인트(다음(next) 노드가 없음)에서 재생하는 것은 no-op이에요.

get_state_history를 사용해 재생할 체크포인트를 찾고, 그 체크포인트의 config로 invoke를 호출해요.

from langgraph.graph import StateGraph, START
from langgraph.checkpoint.memory import InMemorySaver
from typing_extensions import TypedDict, NotRequired
from langchain_core.utils.uuid import uuid7

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


def generate_topic(state: State):
    return {"topic": "socks in the dryer"}


def write_joke(state: State):
    return {"joke": f"Why do {state['topic']} disappear? They elope!"}


checkpointer = InMemorySaver()
graph = (
    StateGraph(State)
    .add_node("generate_topic", generate_topic)
    .add_node("write_joke", write_joke)
    .add_edge(START, "generate_topic")
    .add_edge("generate_topic", "write_joke")
    .compile(checkpointer=checkpointer)
)

# Step 1: Run the graph
config = {"configurable": {"thread_id": str(uuid7())}}
result = graph.invoke({}, config)

# Step 2: Find a checkpoint to replay from
history = list(graph.get_state_history(config))
# History is in reverse chronological order
for state in history:
    print(f"next={state.next}, checkpoint_id={state.config['configurable']['checkpoint_id']}")

# Step 3: Replay from a specific checkpoint
# Find the checkpoint before write_joke
before_joke = next(s for s in history if s.next == ("write_joke",))
replay_result = graph.invoke(None, before_joke.config)
# write_joke re-executes (runs again), generate_topic does not

Fork (포크)

Fork는 수정된 상태로 과거 체크포인트에서 새 분기를 만들어요. 이전 체크포인트에 update_state를 호출해 포크를 만들고, None으로 invoke를 호출해 실행을 계속해요.

update_state는 스레드를 롤백하지 않아요. 지정한 지점에서 분기하는 새 체크포인트를 만들어요. 원래 실행 기록은 그대로 남아 있어요.

# Find checkpoint before write_joke
history = list(graph.get_state_history(config))
before_joke = next(s for s in history if s.next == ("write_joke",))

# Fork: update state to change the topic
fork_config = graph.update_state(
    before_joke.config,
    values={"topic": "chickens"},
)

# Resume from the fork — write_joke re-executes with the new topic
fork_result = graph.invoke(None, fork_config)
print(fork_result["joke"])  # A joke about chickens, not socks

특정 노드에서 (From a specific node)

update_state를 호출하면 값이 지정한 노드의 라이터(writers, 리듀서 포함)로 적용돼요. 체크포인트는 그 노드가 업데이트를 만들었다고 기록하고, 실행은 그 노드의 후속 노드들에서 재개돼요.

기본적으로 LangGraph는 체크포인트의 버전 이력에서 as_node를 추론해요. 특정 체크포인트에서 포크하면 이 추론이 거의 항상 맞아요.

다음 경우에 as_node를 명시하세요.

  • 병렬 분기(Parallel branches): 같은 단계에서 여러 노드가 상태를 갱신했고 LangGraph가 어떤 것이 마지막인지 판단할 수 없을 때 (InvalidUpdateError).
  • 실행 이력 없음(No execution history): 새 스레드에서 상태를 설정할 때 (테스팅에서 흔해요).
  • 노드 건너뛰기(Skipping nodes): as_node를 더 뒤의 노드로 설정하면 그래프가 그 노드가 이미 실행됐다고 생각하게 해요.
# graph: generate_topic -> write_joke

# Treat this update as if generate_topic produced it.
# Execution resumes at write_joke (the successor of generate_topic).
fork_config = graph.update_state(
    before_joke.config,
    values={"topic": "chickens"},
    as_node="generate_topic",
)

인터럽트 (Interrupts)

그래프가 human-in-the-loop 워크플로에 interrupt를 사용한다면, 인터럽트는 타임트래블 중에 항상 다시 트리거돼요. 인터럽트를 포함한 노드가 다시 실행되고, interrupt()는 새 Command(resume=...)를 기다리며 일시정지돼요.

from langgraph.types import interrupt, Command

class State(TypedDict):
    value: list[str]

def ask_human(state: State):
    answer = interrupt("What is your name?")
    return {"value": [f"Hello, {answer}!"]}

def final_step(state: State):
    return {"value": ["Done"]}

graph = (
    StateGraph(State)
    .add_node("ask_human", ask_human)
    .add_node("final_step", final_step)
    .add_edge(START, "ask_human")
    .add_edge("ask_human", "final_step")
    .compile(checkpointer=InMemorySaver())
)

config = {"configurable": {"thread_id": "1"}}

# First run: hits interrupt
graph.invoke({"value": []}, config)
# Resume with answer
graph.invoke(Command(resume="Alice"), config)

# Replay from before ask_human
history = list(graph.get_state_history(config))
before_ask = [s for s in history if s.next == ("ask_human",)][-1]

replay_result = graph.invoke(None, before_ask.config)
# Pauses at interrupt — waiting for new Command(resume=...)

# Fork from before ask_human
fork_config = graph.update_state(before_ask.config, {"value": ["forked"]})
fork_result = graph.invoke(None, fork_config)
# Pauses at interrupt — waiting for new Command(resume=...)

# Resume the forked interrupt with a different answer
graph.invoke(Command(resume="Bob"), fork_config)
# Result: {"value": ["forked", "Hello, Bob!", "Done"]}

여러 인터럽트 (Multiple interrupts)

그래프가 여러 지점에서 입력을 수집한다면(예: 다단계 폼), 인터럽트 사이에서 포크해 이전 질문을 다시 묻지 않고 이후 답변을 바꿀 수 있어요.

def ask_name(state):
    name = interrupt("What is your name?")
    return {"value": [f"name:{name}"]}

def ask_age(state):
    age = interrupt("How old are you?")
    return {"value": [f"age:{age}"]}

# Graph: ask_name -> ask_age -> final
# After completing both interrupts:

# Fork from BETWEEN the two interrupts (after ask_name, before ask_age)
history = list(graph.get_state_history(config))
between = [s for s in history if s.next == ("ask_age",)][-1]

fork_config = graph.update_state(between.config, {"value": ["modified"]})
result = graph.invoke(None, fork_config)
# ask_name result preserved ("name:Alice")
# ask_age pauses at interrupt — waiting for new answer

서브그래프 (Subgraphs)

서브그래프와 함께 타임트래블하려면 서브그래프가 자신의 checkpointer를 갖는지가 관건이에요. 이는 타임트래블할 수 있는 체크포인트의 세밀함(granularity)을 결정해요.

상속된 checkpointer (기본값, Inherited checkpointer)

기본적으로 서브그래프는 부모의 checkpointer를 상속해요. 부모는 서브그래프 전체를 하나의 **슈퍼스텝(super-step)**으로 취급해요. 서브그래프 실행 전체에 대해 부모 수준의 체크포인트가 하나만 있어요. 서브그래프 이전에서 타임트래블하면 서브그래프 전체를 처음부터 다시 실행해요.

기본 서브그래프에서는 노드 사이 지점으로 타임트래블할 수 없어요. 오직 부모 수준에서만 타임트래블할 수 있어요.

# Subgraph without its own checkpointer (default)
subgraph = (
    StateGraph(State)
    .add_node("step_a", step_a)       # Has interrupt()
    .add_node("step_b", step_b)       # Has interrupt()
    .add_edge(START, "step_a")
    .add_edge("step_a", "step_b")
    .compile()  # No checkpointer — inherits from parent
)

graph = (
    StateGraph(State)
    .add_node("subgraph_node", subgraph)
    .add_edge(START, "subgraph_node")
    .compile(checkpointer=InMemorySaver())
)

config = {"configurable": {"thread_id": "1"}}

# Complete both interrupts
graph.invoke({"value": []}, config)            # Hits step_a interrupt
graph.invoke(Command(resume="Alice"), config)  # Hits step_b interrupt
graph.invoke(Command(resume="30"), config)     # Completes

# Time travel from before the subgraph
history = list(graph.get_state_history(config))
before_sub = [s for s in history if s.next == ("subgraph_node",)][-1]

fork_config = graph.update_state(before_sub.config, {"value": ["forked"]})
result = graph.invoke(None, fork_config)
# The entire subgraph re-executes from scratch
# You cannot time travel to a point between step_a and step_b

서브그래프 checkpointer (Subgraph checkpointer)

서브그래프에 checkpointer=True를 설정하면 자신만의 체크포인트 이력을 갖게 돼요. 이는 서브그래프 내부의 각 단계에서 체크포인트를 만들어, 그 안의 특정 지점(예: 두 인터럽트 사이)에서 타임트래블할 수 있게 해줘요.

subgraphs=True로 get_state를 사용해 서브그래프 자신의 체크포인트 config에 접근한 뒤 거기서 포크해요.

# Subgraph with its own checkpointer
subgraph = (
    StateGraph(State)
    .add_node("step_a", step_a)       # Has interrupt()
    .add_node("step_b", step_b)       # Has interrupt()
    .add_edge(START, "step_a")
    .add_edge("step_a", "step_b")
    .compile(checkpointer=True)  # Own checkpoint history
)

graph = (
    StateGraph(State)
    .add_node("subgraph_node", subgraph)
    .add_edge(START, "subgraph_node")
    .compile(checkpointer=InMemorySaver())
)

config = {"configurable": {"thread_id": "1"}}

# Run until step_a interrupt
graph.invoke({"value": []}, config)

# Resume step_a -> hits step_b interrupt
graph.invoke(Command(resume="Alice"), config)

# Get the subgraph's own checkpoint (between step_a and step_b)
parent_state = graph.get_state(config, subgraphs=True)
sub_config = parent_state.tasks[0].state.config

# Fork from the subgraph checkpoint
fork_config = graph.update_state(sub_config, {"value": ["forked"]})
result = graph.invoke(None, fork_config)
# step_b re-executes, step_a's result is preserved

서브그래프 checkpointer 구성에 대한 자세한 내용은 서브그래프 영속성을 참고해요.

더 알아보기 (Learn more)