타임트래블 사용하기 (Use Time-Travel)

타임트래블 사용하기 (Use Time-Travel)

과거 실행을 재생(replay)하고, 대안 경로를 탐색하기 위해 포크(fork)하기 (LangGraph)

개요 (Overview)

LangGraph는 체크포인트(checkpoint)를 통해 타임트래블을 지원해요. 핵심은 두 가지입니다.

• Replay(재생): 이전 체크포인트에서부터 다시 실행하기.
• Fork(포크): 상태를 바꾼 채 이전 체크포인트에서 분기해서 대안 경로를 탐색하기.

둘 다 이전 체크포인트에서부터 실행을 재개하는 방식으로 동작해요. 체크포인트 이전의 노드는 다시 실행되지 않아요(결과가 이미 저장되어 있으니까요). 체크포인트 이후의 노드는 다시 실행되는데, 여기에는 LLM 호출, API 요청, 인터럽트(interrupt)까지 포함돼요. 그래서 이들이 다시 실행되면 결과가 달라질 수도 있다는 점을 기억해 두면 좋아요.

재생 (Replay)

이전 체크포인트의 config로 그래프를 호출(invoke)하면 그 지점부터 다시 실행됩니다.

재생은 캐시를 읽는 게 아니라 노드를 실제로 다시 실행해요. 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

get_state_history는 시간 역순으로 히스토리를 돌려줘요. write_joke 직전의 체크포인트를 찾아 그 config로 다시 호출하면, write_joke는 다시 실행되고 generate_topic은 실행되지 않아요.

포크 (Fork)

Fork는 상태를 바꾼 상태로 과거 체크포인트에서 새 분기(branch)를 만드는 기능이에요. 이전 체크포인트에 update_state를 호출해 포크를 만든 뒤, None으로 invoke를 호출해 실행을 이어갑니다.

update_state는 스레드(thread)를 되돌리지 않아요. 지정한 지점에서 분기하는 새 체크포인트를 만들 뿐이고, 원래 실행 히스토리는 그대로 남아 있어요.

# 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

포크 지점 이후의 노드만 다시 실행돼요. 위 예시에서 topic을 "chickens"로 바꿨으니, write_joke가 새 주제로 다시 실행되어 닭 얘기(양말 얘기 아님)가 나와요.

특정 노드에서 시작하기 (From a specific node)

update_state를 호출하면 값은 지정한 노드의 writer(리듀서(reducers) 포함)를 통해 적용돼요. 체크포인트는 그 노드가 해당 업데이트를 만들었다고 기록하고, 실행은 그 노드의 후속 노드(successors)부터 재개됩니다.

기본적으로 LangGraph는 체크포인트의 버전 히스토리에서 as_node를 추론해요. 특정 체크포인트에서 포크할 때는 이 추론이 거의 항상 정확해요.

다음 경우에는 as_node를 명시적으로 지정해야 합니다.

• 병렬 분기: 같은 스텝에서 여러 노드가 상태를 업데이트해서, LangGraph가 어느 것이 마지막인지 판단할 수 없을 때(InvalidUpdateError).
• 실행 히스토리가 없을 때: 새 스레드에 상태를 설정하는 경우(테스트에서 흔해요).
• 노드 건너뛰기: 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",
)

as_node="generate_topic"으로 지정하면 실행이 generate_topic의 후속 노드인 write_joke부터 다시 시작돼요.

인터럽트 (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"]}

ask_human 이전에서 포크해도 interrupt가 다시 걸려요. 저장했던 "Alice" 대신 새 Command(resume="Bob")으로 다시 응답하면, 결과에 "Hello, Bob!"이 반영돼요.

여러 인터럽트 (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

두 인터럽트(ask_name → ask_age) 사이에서 포크하면, ask_name의 결과("name:Alice")는 보존되고 ask_age만 인터럽트로 다시 멈춰서 새 답변을 기다려요.

서브그래프 (Subgraphs)

서브그래프가 있는 경우 타임트래블은 서브그래프에 자체 체크포인터가 있는지에 따라 달라져요. 이게 타임트래블할 수 있는 체크포인트의 세분화(granularity)를 결정합니다.

체크포인터 상속(기본값)

기본적으로 서브그래프는 부모의 체크포인터를 상속해요. 부모는 서브그래프 전체를 **하나의 슈퍼스텝(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=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

서브그래프 체크포인트에서 포크하면 step_b만 다시 실행되고 step_a의 결과는 보존돼요. 서브그래프 체크포인터 구성에 대한 더 자세한 내용은 서브그래프 영속성(subgraph persistence) 문서를 참고하세요.