서버 API를 사용한 타임 트래블

서버 API를 사용한 타임 트래블

LangGraph는 이전 체크포인트에서 실행을 재개하는 타임 트래블 기능을 제공해요. 동일한 상태를 재생하거나, 대안을 탐색하기 위해 상태를 수정할 수 있어요. 어느 경우든 과거 실행을 재개하면 이력에 새로운 포크가 생성돼요.

LangSmith Deployment 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_idcheckpoint_id로 사용해요.

출처: 문서

본문

워크플로에서 타임 트래블 사용하기

```python 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( "claude-sonnet-4-6", 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()

</Accordion>

### 1. 그래프 실행

<Tabs>
<Tab title="Python">
  ```python
  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={}
  )
  ```
</Tab>

<Tab title="JavaScript">
  ```js
  import { Client } from "@langchain/langgraph-sdk";
  const client = new Client({ apiUrl: <DEPLOYMENT_URL> });

  // Using the graph deployed with the name "agent"
  const assistantID = "agent";

  // create a thread
  const thread = await client.threads.create();
  const threadID = thread["thread_id"];

  // Run the graph
  const result = await client.runs.wait(
    threadID,
    assistantID,
    { input: {}}
  );
  ```
</Tab>

<Tab title="cURL">
  스레드 생성:

  ```bash
  curl --request POST \
  --url <DEPLOYMENT_URL>/threads \
  --header 'Content-Type: application/json' \
  --data '{}'
  ```

  그래프 실행:

  ```bash
  curl --request POST \
  --url <DEPLOYMENT_URL>/threads/<THREAD_ID>/runs/wait \
  --header 'Content-Type: application/json' \
  --data "{
    \"assistant_id\": \"agent\",
    \"input\": {}
  }"
  ```
</Tab>
</Tabs>

### 2. 체크포인트 식별

<Tabs>
<Tab title="Python">
  ```python
  # The states are returned in reverse chronological order.
  states = await client.threads.get_history(thread_id)
  selected_state = states[1]
  print(selected_state)
  ```
</Tab>

<Tab title="JavaScript">
  ```js
  // The states are returned in reverse chronological order.
  const states = await client.threads.getHistory(threadID);
  const selectedState = states[1];
  console.log(selectedState);
  ```
</Tab>

<Tab title="cURL">
  ```bash
  curl --request GET \
  --url <DEPLOYMENT_URL>/threads/<THREAD_ID>/history \
  --header 'Content-Type: application/json'
  ```
</Tab>
</Tabs>

<a id="optional" />

### 3. 상태 업데이트

[`update_state`](https://reference.langchain.com/python/langgraph/graphs/#langgraph.graph.state.CompiledStateGraph.update_state)는 새 체크포인트를 만들어요. 새 체크포인트는 동일한 스레드에 연결되지만 새 체크포인트 ID를 가져요.

<Tabs>
<Tab title="Python">
  ```python
  new_config = await client.threads.update_state(
      thread_id,
      {"topic": "chickens"},
      checkpoint_id=selected_state["checkpoint_id"]
  )
  print(new_config)
  ```
</Tab>

<Tab title="JavaScript">
  ```js
  const newConfig = await client.threads.updateState(
    threadID,
    {
      values: { "topic": "chickens" },
      checkpointId: selectedState["checkpoint_id"]
    }
  );
  console.log(newConfig);
  ```
</Tab>

<Tab title="cURL">
  ```bash
  curl --request POST \
  --url <DEPLOYMENT_URL>/threads/<THREAD_ID>/state \
  --header 'Content-Type: application/json' \
  --data "{
    \"assistant_id\": \"agent\",
    \"checkpoint_id\": <CHECKPOINT_ID>,
    \"values\": {\"topic\": \"chickens\"}
  }"
  ```
</Tab>
</Tabs>

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

<Tabs>
<Tab title="Python">
  ```python
  await client.runs.wait(
      thread_id,
      assistant_id,
      input=None,
      checkpoint_id=new_config["checkpoint_id"]
  )
  ```
</Tab>

<Tab title="JavaScript">
  ```javascript
  await client.runs.wait(
    threadID,
    assistantID,
    {
      input: null,
      checkpointId: newConfig["checkpoint_id"]
    }
  );
  ```
</Tab>

<Tab title="cURL">
  ```bash
  curl --request POST \
  --url <DEPLOYMENT_URL>/threads/<THREAD_ID>/runs/wait \
  --header 'Content-Type: application/json' \
  --data "{
    \"assistant_id\": \"agent\",
    \"checkpoint_id\": <CHECKPOINT_ID>
  }"
  ```
</Tab>
</Tabs>

## 더 알아보기

* [**LangGraph 타임 트래블 가이드**](/oss/python/langgraph/use-time-travel): LangGraph에서 타임 트래블을 사용하는 방법에 대해 자세히 알아보기.

## 더 알아보기 (Learn more)

- [LangGraph 타임 트래블 가이드](/oss/python/langgraph/use-time-travel)
- [LangGraph Python SDK](/langsmith/langgraph-python-sdk)