如何从之前的态重新运行和分支¶

使用LangGraph Cloud，您可以返回到任何先前的状态，并重新运行图以重现测试期间注意到的问题，或者从之前的状态以不同的方式分支出去。在本指南中，我们将展示如何重新运行过去的态以及如何从先前的态分支出去的一个快速示例。

设置¶

我们不会展示托管图的完整代码，但如果你想查看，可以在这里查看：此处。一旦图被托管，我们就可以调用它并等待用户输入。

SDK初始化¶

首先，我们需要设置客户端，以便能够与托管的图进行通信：

PythonJavaScriptCURL

from langgraph_sdk import get_client
client = get_client(url=<DEPLOYMENT_URL>)
# 使用名称为 "agent" 部署的图
assistant_id = "agent"
thread = await client.threads.create()

import { Client } from "@langchain/langgraph-sdk";

const client = new Client({ apiUrl: <DEPLOYMENT_URL> });
// 使用名称为 "agent" 部署的图
const assistantId = "agent";
const thread = await client.threads.create();

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

重新播放状态¶

初始调用¶

在重新播放状态之前，我们需要创建一些状态来重新播放！为了做到这一点，让我们用一条简单的消息调用我们的图：

PythonJavascriptCURL

input = {"messages": [{"role": "user", "content": "请搜索旧金山的天气"}]}

async for chunk in client.runs.stream(
    thread["thread_id"],
    assistant_id,
    input=input,
    stream_mode="updates",
):
    if chunk.data and chunk.event != "metadata": 
        print(chunk.data)

const input = { "messages": [{ "role": "user", "content": "请搜索旧金山的天气" }] }

const streamResponse = client.runs.stream(
  thread["thread_id"],
  assistantId,
  {
    input: input,
    streamMode: "updates",
  }
);
for await (const chunk of streamResponse) {
  if (chunk.data && chunk.event !== "metadata") {
    console.log(chunk.data);
  }
}

curl --request POST \
 --url <DEPLOYMENT_URL>/threads/<THREAD_ID>/runs/stream \
 --header 'Content-Type: application/json' \
 --data "{
   \"assistant_id\": \"agent\",
   \"input\": {\"messages\": [{\"role\": \"human\", \"content\": \"请搜索旧金山的天气\"}]},
   \"stream_mode\": [
     \"updates\"
   ]
 }" | \
 sed 's/\r$//' | \
 awk '
 /^event:/ {
     if (data_content != "" && event_type != "metadata") {
         print data_content "\n"
     }
     sub(/^event: /, "", $0)
     event_type = $0
     data_content = ""
 }
 /^data:/ {
     sub(/^data: /, "", $0)
     data_content = $0
 }
 END {
     if (data_content != "" && event_type != "metadata") {
         print data_content "\n"
     }
 }
 '

输出：

{'agent': {'messages': [{'content': [{'text': "当然！我将使用搜索功能为您查询旧金山当前的天气。现在让我来查一下。", 'type': 'text'}, {'id': 'toolu_011vroKUtWU7SBdrngpgpFMn', 'input': {'query': 'current weather in San Francisco'}, 'name': 'search', 'type': 'tool_use'}], 'additional_kwargs': {}, 'response_metadata': {}, 'type': 'ai', 'name': None, 'id': 'run-ee639877-d97d-40f8-96dc-d0d1ae22d203', 'example': False, 'tool_calls': [{'name': 'search', 'args': {'query': 'current weather in San Francisco'}, 'id': 'toolu_011vroKUtWU7SBdrngpgpFMn'}], 'invalid_tool_calls': [], 'usage_metadata': None}]}}
{'action': {'messages': [{'content': '["我查到了：旧金山当前的天气。结果：旧金山天气晴朗，但是双子座的朋友们要注意哦 😈."]', 'additional_kwargs': {}, 'response_metadata': {}, 'type': 'tool', 'name': 'search', 'id': '7bad0e72-5ebe-4b08-9b8a-b99b0fe22fb7', 'tool_call_id': 'toolu_011vroKUtWU7SBdrngpgpFMn'}]}} 
{'agent': {'messages': [{'content': "根据搜索结果，我可以为您提供旧金山当前天气的信息：\n\n旧金山当前天气晴朗。这对户外活动和享受城市美景来说是个好消息。\n\n值得注意的是，搜索结果中包含了一个关于双子座的不寻常评论，这通常不会出现在天气报告中。这可能是由于搜索引擎在结果中包含了某些占星信息或笑话。然而，为了回答您关于天气的问题，我们可以专注于旧金山天气晴朗这一事实。\n\n如果您需要更多关于旧金山天气的具体信息，如温度、风速或未来几天的预报，请告诉我，我很乐意为您查询这些信息。", 'additional_kwargs': {}, 'response_metadata': {}, 'type': 'ai', 'name': None, 'id': 'run-dbac539a-33c8-4f0c-9e20-91f318371e7c', 'example': False, 'tool_calls': [], 'invalid_tool_calls': [], 'usage_metadata': None}]}}

现在让我们获取状态列表，并从第三个状态（即工具调用之前的状态）重新调用：

PythonJavascriptCURL

states = await client.threads.get_history(thread['thread_id'])

# 我们可以通过检查'next'属性来确认这个状态是正确的，可以看到它是工具调用节点
state_to_replay = states[2]
print(state_to_replay['next'])

const states = await client.threads.getHistory(thread['thread_id']);

// 我们可以通过检查'next'属性来确认这个状态是正确的，可以看到它是工具调用节点
const stateToReplay = states[2];
console.log(stateToReplay['next']);

curl --request GET --url <DEPLOYMENT_URL>/threads/<THREAD_ID>/history | jq -r '.[2].next'

输出：

['action']

为了从某个状态重新运行，我们首先需要向线程状态发出一个空更新。然后我们需要将生成的checkpoint_id作为以下参数传递：

PythonJavascriptCURL

state_to_replay = states[2]
updated_config = await client.threads.update_state(
    thread["thread_id"],
    {"messages": []},
    checkpoint_id=state_to_replay["checkpoint_id"]
)
async for chunk in client.runs.stream(
    thread["thread_id"],
    assistant_id, # graph_id
    input=None,
    stream_mode="updates",
    checkpoint_id=updated_config["checkpoint_id"]
):
    if chunk.data and chunk.event != "metadata": 
        print(chunk.data)

const stateToReplay = states[2];
const config = await client.threads.updateState(thread["thread_id"], { values: {"messages": [] }, checkpointId: stateToReplay["checkpoint_id"] });
const streamResponse = client.runs.stream(
  thread["thread_id"],
  assistantId,
  {
    input: null,
    streamMode: "updates",
    checkpointId: config["checkpoint_id"]
  }
);
for await (const chunk of streamResponse) {
  if (chunk.data && chunk.event !== "metadata") {
    console.log(chunk.data);
  }
}

curl --request GET --url <DEPLOYMENT_URL>/threads/<THREAD_ID>/history | jq -c '
    .[2] as $state_to_replay |
    {
        values: { messages: .[2].values.messages[-1] },
        checkpoint_id: $state_to_replay.checkpoint_id
    }' | \
curl --request POST \
    --url <DEPLOYMENT_URL>/threads/<THREAD_ID>/state \
    --header 'Content-Type: application/json' \
    --data @- | jq .checkpoint_id | \
curl --request POST \
 --url <DEPLOYMENT_URL>/threads/<THREAD_ID>/runs/stream \
 --header 'Content-Type: application/json' \
 --data "{
   \"assistant_id\": \"agent\",
   \"checkpoint_id\": \"$1\",
   \"stream_mode\": [
     \"updates\"
   ]
 }" | \
 sed 's/\r$//' | \
 awk '
 /^event:/ {
     if (data_content != "" && event_type != "metadata") {
         print data_content "\n"
     }
     sub(/^event: /, "", $0)
     event_type = $0
     data_content = ""
 }
 /^data:/ {
     sub(/^data: /, "", $0)
     data_content = $0
 }
 END {
     if (data_content != "" && event_type != "metadata") {
         print data_content "\n"
     }
 }
 '

输出：

{'action': {'messages': [{'content': '["我查到了：旧金山当前的天气。结果：旧金山天气晴朗，但是双子座的朋友们要注意哦 😈."]', 'additional_kwargs': {}, 'response_metadata': {}, 'type': 'tool', 'name': 'search', 'id': 'eba650e5-400e-4938-8508-f878dcbcc532', 'tool_call_id': 'toolu_011vroKUtWU7SBdrngpgpFMn'}]}} 
{'agent': {'messages': [{'content': "根据搜索结果，我可以为您提供旧金山当前天气的信息：\n\n旧金山当前天气晴朗。这对户外活动和享受城市美景来说是个好消息。\n\n值得注意的是，搜索结果中包含了一个关于双子座的不寻常评论，这通常不会出现在天气报告中。这可能是由于搜索引擎在结果中包含了某些占星信息或笑话。\n\n您是否还需要了解关于旧金山天气的其他信息或需要其他信息？", 'additional_kwargs': {}, 'response_metadata': {}, 'type': 'ai', 'name': None, 'id': 'run-bc6dca3f-a1e2-4f59-a69b-fe0515a348bb', 'example': False, 'tool_calls': [], 'invalid_tool_calls': [], 'usage_metadata': None}]}}

如我们所见，图从工具节点重新启动，并使用与我们原始图运行相同的输入。

从先前状态分支¶

使用LangGraph的检查点功能，您可以做更多的事情，不仅仅是重播过去的状态。您可以从先前的位置分支出来，让代理探索替代轨迹，或者让用户对工作流中的更改进行“版本控制”。

让我们展示如何编辑特定时间点的状态。我们将更新状态以更改工具的输入。

PythonJavaScriptCURL

# 现在获取状态中的最后一条消息
# 这是包含我们想要更新的工具调用的消息
last_message = state_to_replay['values']['messages'][-1]

# 更新该工具调用的参数
last_message['tool_calls'][0]['args'] = {'query': 'current weather in SF'}

config = await client.threads.update_state(thread['thread_id'],{"messages":[last_message]},checkpoint_id=state_to_replay['checkpoint_id'])

// 现在获取状态中的最后一条消息
// 这是包含我们想要更新的工具调用的消息
let lastMessage = stateToReplay['values']['messages'][-1];

// 更新该工具调用的参数
lastMessage['tool_calls'][0]['args'] = { 'query': 'current weather in SF' };

const config = await client.threads.updateState(thread['thread_id'], { values: { "messages": [lastMessage] }, checkpointId: stateToReplay['checkpoint_id'] });

curl -s --request GET --url <DEPLOYMENT_URL>/threads/<THREAD_ID>/history | \
jq -c '
    .[2] as $state_to_replay |
    .[2].values.messages[-1].tool_calls[0].args.query = "current weather in SF" |
    {
        values: { messages: .[2].values.messages[-1] },
        checkpoint_id: $state_to_replay.checkpoint_id
    }' | \
curl --request POST \
    --url <DEPLOYMENT_URL>/threads/<THREAD_ID>/state \
    --header 'Content-Type: application/json' \
    --data @-

现在，我们可以使用新的配置重新运行我们的图，从new_state开始，这是从state_to_replay分支出来的：

PythonJavaScriptCURL

async for chunk in client.runs.stream(
    thread["thread_id"],
    assistant_id,
    input=None,
    stream_mode="updates",
    checkpoint_id=config['checkpoint_id']
):
    if chunk.data and chunk.event != "metadata": 
        print(chunk.data)

const streamResponse = client.runs.stream(
  thread["thread_id"],
  assistantId,
  {
    input: null,
    streamMode: "updates",
    checkpointId: config['checkpoint_id'],
  }
);
for await (const chunk of streamResponse) {
  if (chunk.data && chunk.event !== "metadata") {
    console.log(chunk.data);
  }
}

curl -s --request GET --url <DEPLOYMENT_URL>/threads/<THREAD_ID>/state | \
jq -c '.checkpoint_id' | \
curl --request POST \
 --url <DEPLOYMENT_URL>/threads/<THREAD_ID>/runs/stream \
 --header 'Content-Type: application/json' \
 --data "{
   \"assistant_id\": \"agent\",
   \"checkpoint_id\": \"$1\",
   \"stream_mode\": [
     \"updates\"
   ]
 }" | \
 sed 's/\r$//' | \
 awk '
 /^event:/ {
     if (data_content != "" && event_type != "metadata") {
         print data_content "\n"
     }
     sub(/^event: /, "", $0)
     event_type = $0
     data_content = ""
 }
 /^data:/ {
     sub(/^data: /, "", $0)
     data_content = $0
 }
 END {
     if (data_content != "" && event_type != "metadata") {
         print data_content "\n"
     }
 }
 '

输出：

{'action': {'messages': [{'content': '["I looked up: current weather in SF. Result: It\'s sunny in San Francisco, but you better look out if you\'re a Gemini 😈."]', 'additional_kwargs': {}, 'response_metadata': {}, 'type': 'tool', 'name': 'search', 'id': '2baf9941-4fda-4081-9f87-d76795d289f1', 'tool_call_id': 'toolu_011vroKUtWU7SBdrngpgpFMn'}]}} 
{'agent': {'messages': [{'content': "Based on the search results, I can provide you with information about the current weather in San Francisco (SF):\n\nThe weather in San Francisco is currently sunny. This means it's a clear day with plenty of sunshine. \n\nIt's worth noting that the specific temperature wasn't provided in the search result, but sunny weather in San Francisco typically means comfortable temperatures. San Francisco is known for its mild climate, so even on sunny days, it's often not too hot.\n\nThe search result also included a playful reference to astrological signs, mentioning Gemini. However, this is likely just a joke or part of the search engine's presentation and not related to the actual weather conditions.\n\nIs there any specific information about the weather in San Francisco you'd like to know more about? I'd be happy to perform another search if you need details on temperature, wind conditions, or the forecast for the coming days.", 'additional_kwargs': {}, 'response_metadata': {}, 'type': 'ai', 'name': None, 'id': 'run-a83de52d-ed18-4402-9384-75c462485743', 'example': False, 'tool_calls': [], 'invalid_tool_calls': [], 'usage_metadata': None}]}}

如我们所见，搜索查询从“San Francisco”变更为“SF”，正如我们所期望的那样！

如何从之前的态重新运行和分支¶

设置¶

SDK初始化¶

重新播放状态¶

初始调用¶

从先前状态分支¶

Comments