Skip to main content
LangGraph는 이전 체크포인트에서 실행을 재개하여 동일한 상태를 재현하거나 대안을 탐색하기 위해 수정할 수 있는 시간 여행 기능을 제공합니다. 모든 경우에서 과거 실행을 재개하면 히스토리에 새로운 분기가 생성됩니다. LangGraph Server API를 사용하여 시간 여행을 수행하려면 (LangGraph SDK를 통해):
  1. 그래프 실행: LangGraph SDK의 client.runs.wait 또는 client.runs.stream API를 사용하여 초기 입력으로 그래프를 실행합니다.
  2. 기존 스레드에서 체크포인트 식별: client.threads.get_history 메서드를 사용하여 특정 thread_id의 실행 히스토리를 가져오고 원하는 checkpoint_id를 찾습니다. 또는 실행을 일시 중지할 노드 앞에 중단점을 설정할 수 있습니다. 그런 다음 해당 중단점까지 기록된 가장 최근 체크포인트를 찾을 수 있습니다.
  3. (선택 사항) 그래프 상태 수정: client.threads.update_state 메서드를 사용하여 체크포인트에서 그래프 상태를 수정하고 대안 상태에서 실행을 재개합니다.
  4. 체크포인트에서 실행 재개: client.runs.wait 또는 client.runs.stream API를 None 입력과 적절한 thread_id 및 checkpoint_id와 함께 사용합니다.

​
워크플로우에서 시간 여행 사용하기

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
builder = StateGraph(State)

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

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

# Compile
graph = builder.compile()

​
1. 그래프 실행

  • Python
  • JavaScript
  • cURL
from langgraph_sdk import get_client
client = get_client(url=<DEPLOYMENT_URL>)

# Using the graph deployed with the name "agent"
assistant_id = "agent"

# create a thread
thread = await client.threads.create()
thread_id = thread["thread_id"]

# Run the graph
result = await client.runs.wait(
    thread_id,
    assistant_id,
    input={}
)

​
2. 체크포인트 식별

  • Python
  • JavaScript
  • cURL
# The states are returned in reverse chronological order.
states = await client.threads.get_history(thread_id)
selected_state = states[1]
print(selected_state)

​
3. 상태 업데이트

update_state는 새로운 체크포인트를 생성합니다. 새 체크포인트는 동일한 스레드와 연결되지만, 새로운 체크포인트 ID를 갖습니다.
  • Python
  • JavaScript
  • cURL
new_config = await client.threads.update_state(
    thread_id,
    {"topic": "chickens"},
    checkpoint_id=selected_state["checkpoint_id"]
)
print(new_config)

​
4. 체크포인트에서 실행 재개

  • Python
  • JavaScript
  • cURL
await client.runs.wait(
    thread_id,
    assistant_id,
    input=None,
    checkpoint_id=new_config["checkpoint_id"]
)

​
더 알아보기

Edit the source of this page on GitHub.
Connect these docs programmatically to Claude, VSCode, and more via MCP for real-time answers.
⌘I