서버 API를 사용한 타임 트래블
서버 API를 사용한 타임 트래블
LangGraph는 이전 체크포인트에서 실행을 재개하는 타임 트래블 기능을 제공해요. 동일한 상태를 재생하거나, 대안을 탐색하기 위해 상태를 수정할 수 있어요. 어느 경우든 과거 실행을 재개하면 이력에 새로운 포크가 생성돼요.
LangSmith Deployment API(LangGraph SDK를 통해)를 사용해 타임 트래블하려면:
- **LangGraph SDK**의 client.runs.wait 또는 client.runs.stream API를 사용해 초기 입력으로 그래프를 실행해요.
- 기존 스레드에서 체크포인트 식별: client.threads.get_history 메서드를 사용해 특정
thread_id의 실행 이력을 검색하고 원하는checkpoint_id를 찾아요. 또는 실행을 일시 중지하려는 노드 앞에 브레이크포인트를 설정할 수도 있어요. 그 브레이크포인트까지 기록된 가장 최근 체크포인트를 찾을 수 있어요. - (선택) 그래프 상태 수정: client.threads.update_state 메서드를 사용해 체크포인트에서 그래프 상태를 수정하고 대체 상태에서 실행을 재개해요.
- 체크포인트에서 실행 재개: client.runs.wait 또는 client.runs.stream API를
None입력과 적절한thread_id및checkpoint_id로 사용해요.
출처: 문서
본문
워크플로에서 타임 트래블 사용하기
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)