使用服务器 API 实现人机协同

要在智能体或工作流中审查、编辑和批准工具调用，请使用 LangGraph 的人机协同功能。

动态中断

Python

from langgraph_sdk import get_client
from langgraph_sdk.schema import Command
client = get_client(url=<DEPLOYMENT_URL>)

# 使用已部署且名为 "agent" 的图
assistant_id = "agent"

# 创建会话线程
thread = await client.threads.create()
thread_id = thread["thread_id"]

# 运行图直到遇到中断点
result = await client.runs.wait(
    thread_id,
    assistant_id,
    input={"some_text": "original text"}   # (1)!
)

print(result['__interrupt__']) # (2)!
# > [
# >     {
# >         'value': {'text_to_revise': 'original text'},
# >         'resumable': True,
# >         'ns': ['human_node:fc722478-2f21-0578-c572-d9fc4dd07c3b'],
# >         'when': 'during'
# >     }
# > ]


# 恢复图执行
print(await client.runs.wait(
    thread_id,
    assistant_id,
    command=Command(resume="Edited text")   # (3)!
))
# > {'some_text': 'Edited text'}

1. 使用初始状态调用图。
2. 当图遇到中断时，返回包含负载和元数据的中断对象。 3. 通过 Command(resume=...) 恢复图执行，注入人工输入并继续执行。

JavaScript

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

// 使用已部署且名为 "agent" 的图
const assistantID = "agent";

// 创建会话线程
const thread = await client.threads.create();
const threadID = thread["thread_id"];

// 运行图直到遇到中断点
const result = await client.runs.wait(
  threadID,
  assistantID,
  { input: { "some_text": "original text" } }   # (1)!
);

console.log(result['__interrupt__']); # (2)!
// > [
# >     {
# >         'value': {'text_to_revise': 'original text'},
# >         'resumable': True,
# >         'ns': ['human_node:fc722478-2f21-0578-c572-d9fc4dd07c3b'],
# >         'when': 'during'
# >     }
# > ]

// 恢复图执行
console.log(await client.runs.wait(
    threadID,
    assistantID,
    { command: { resume: "Edited text" }}   # (3)!
));
# > {'some_text': 'Edited text'}

1. 使用初始状态调用图。
2. 当图遇到中断时，返回包含负载和元数据的中断对象。
3. 通过 { resume: ... } 命令对象恢复图执行，注入人工输入并继续执行。

cURL

创建会话线程：

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

运行图直到遇到中断点：

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

恢复图执行：

curl --request POST \
 --url <DEPLOYMENT_URL>/threads/<THREAD_ID>/runs/wait \
 --header 'Content-Type: application/json' \
 --data "{
   \"assistant_id\": \"agent\",
   \"command\": {
     \"resume\": \"Edited text\"
   }
 }"

扩展示例：使用 `interrupt`

这是一个可以在 Agent Server 中运行的图示例。 更多详情请参阅 LangSmith 快速入门。

from typing import TypedDict
import uuid

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.constants import START
from langgraph.graph import StateGraph
from langgraph.types import interrupt, Command

class State(TypedDict):
    some_text: str

def human_node(state: State):
    value = interrupt( # (1)!
        {
            "text_to_revise": state["some_text"] # (2)!
        }
    )
    return {
        "some_text": value # (3)!
    }


# 构建图
graph_builder = StateGraph(State)
graph_builder.add_node("human_node", human_node)
graph_builder.add_edge(START, "human_node")

graph = graph_builder.compile()

1. interrupt(...) 在 human_node 处暂停执行，将给定负载呈现给人工处理。
2. 任何可 JSON 序列化的值都可以传递给 interrupt 函数。这里传递的是包含待修订文本的字典。
3. 恢复后，interrupt(...) 的返回值是人工提供的输入，用于更新状态。

运行 Agent Server 后，可以使用 LangGraph SDK 与其交互。

Python

from langgraph_sdk import get_client
from langgraph_sdk.schema import Command
client = get_client(url=<DEPLOYMENT_URL>)

# 使用已部署且名为 "agent" 的图
assistant_id = "agent"

# 创建会话线程
thread = await client.threads.create()
thread_id = thread["thread_id"]

# 运行图直到遇到中断点
result = await client.runs.wait(
    thread_id,
    assistant_id,
    input={"some_text": "original text"}   # (1)!
)

print(result['__interrupt__']) # (2)!
# > [
# >     {
# >         'value': {'text_to_revise': 'original text'},
# >         'resumable': True,
# >         'ns': ['human_node:fc722478-2f21-0578-c572-d9fc4dd07c3b'],
# >         'when': 'during'
# >     }
# > ]


# 恢复图执行
print(await client.runs.wait(
    thread_id,
    assistant_id,
    command=Command(resume="Edited text")   # (3)!
))
# > {'some_text': 'Edited text'}

1. 使用初始状态调用图。
2. 当图遇到中断时，返回包含负载和元数据的中断对象。 3. 通过 Command(resume=...) 恢复图执行，注入人工输入并继续执行。

JavaScript

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

// 使用已部署且名为 "agent" 的图
const assistantID = "agent";

// 创建会话线程
const thread = await client.threads.create();
const threadID = thread["thread_id"];

// 运行图直到遇到中断点
const result = await client.runs.wait(
  threadID,
  assistantID,
  { input: { "some_text": "original text" } }   # (1)!
);

console.log(result['__interrupt__']); # (2)!
# > [
# >     {
# >         'value': {'text_to_revise': 'original text'},
# >         'resumable': True,
# >         'ns': ['human_node:fc722478-2f21-0578-c572-d9fc4dd07c3b'],
# >         'when': 'during'
# >     }
# > ]

// 恢复图执行
console.log(await client.runs.wait(
    threadID,
    assistantID,
    { command: { resume: "Edited text" }}   # (3)!
));
# > {'some_text': 'Edited text'}

1. 使用初始状态调用图。
2. 当图遇到中断时，返回包含负载和元数据的中断对象。
3. 通过 { resume: ... } 命令对象恢复图执行，注入人工输入并继续执行。

cURL

创建会话线程：

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

运行图直到遇到中断点：

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

恢复图执行：

curl --request POST \
--url <DEPLOYMENT_URL>/threads/<THREAD_ID>/runs/wait \
--header 'Content-Type: application/json' \
--data "{
  \"assistant_id\": \"agent\",
  \"command\": {
    \"resume\": \"Edited text\"
  }
}"

静态中断

静态中断（也称为静态断点）在节点执行前或执行后触发。

静态中断不推荐用于人机协同工作流。它们最适合调试和测试场景。

您可以在编译时通过指定 interrupt_before 和 interrupt_after 来设置静态中断：

graph = graph_builder.compile( # (1)!
    interrupt_before=["node_a"], # (2)!
    interrupt_after=["node_b", "node_c"], # (3)!
)

1. 断点在 compile 时设置。
2. interrupt_before 指定在节点执行前应暂停执行的节点。
3. interrupt_after 指定在节点执行后应暂停执行的节点。

或者，您也可以在运行时设置静态中断：

Python

await client.runs.wait( # (1)!
    thread_id,
    assistant_id,
    inputs=inputs,
    interrupt_before=["node_a"], # (2)!
    interrupt_after=["node_b", "node_c"] # (3)!
)

1. 调用 client.runs.wait 时传入 interrupt_before 和 interrupt_after 参数。这是运行时配置，每次调用都可以更改。
2. interrupt_before 指定在节点执行前应暂停执行的节点。
3. interrupt_after 指定在节点执行后应暂停执行的节点。

JavaScript

await client.runs.wait( // (1)!
    threadID,
    assistantID,
    {
    input: input,
    interruptBefore: ["node_a"], // (2)!
    interruptAfter: ["node_b", "node_c"] // (3)!
    }
)

1. 调用 client.runs.wait 时传入 interruptBefore 和 interruptAfter 参数。这是运行时配置，每次调用都可以更改。
2. interruptBefore 指定在节点执行前应暂停执行的节点。
3. interruptAfter 指定在节点执行后应暂停执行的节点。

cURL

curl --request POST \
--url <DEPLOYMENT_URL>/threads/<THREAD_ID>/runs/wait \
--header 'Content-Type: application/json' \
--data "{
    \"assistant_id\": \"agent\",
    \"interrupt_before\": [\"node_a\"],
    \"interrupt_after\": [\"node_b\", \"node_c\"],
    \"input\": <INPUT>
}"

以下示例展示如何添加静态中断：

Python

from langgraph_sdk import get_client
client = get_client(url=<DEPLOYMENT_URL>)

# 使用已部署且名为 "agent" 的图
assistant_id = "agent"

# 创建会话线程
thread = await client.threads.create()
thread_id = thread["thread_id"]

# 运行图直到遇到断点
result = await client.runs.wait(
    thread_id,
    assistant_id,
    input=inputs   # (1)!
)

# 恢复图执行
await client.runs.wait(
    thread_id,
    assistant_id,
    input=None   # (2)!
)

1. 运行图直到遇到第一个断点。
2. 通过传入 None 作为输入来恢复图执行。这将运行图直到遇到下一个断点。

JavaScript

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

// 使用已部署且名为 "agent" 的图
const assistantID = "agent";

# 创建会话线程
const thread = await client.threads.create();
const threadID = thread["thread_id"];

# 运行图直到遇到断点
const result = await client.runs.wait(
  threadID,
  assistantID,
  { input: input }   # (1)!
);

# 恢复图执行
await client.runs.wait(
  threadID,
  assistantID,
  { input: null }   # (2)!
);

1. 运行图直到遇到第一个断点。
2. 通过传入 null 作为输入来恢复图执行。这将运行图直到遇到下一个断点。

cURL

创建会话线程：

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

运行图直到遇到断点：

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

恢复图执行：

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

了解更多

• 人机协同概念指南：深入了解 LangGraph 人机协同功能。
• 常见模式：学习如何实现批准/拒绝操作、请求用户输入、工具调用审查和验证人工输入等模式。

---

Edit this page on GitHub or file an issue.

Connect these docs to Claude, VSCode, and more via MCP for real-time answers.