使用服务器 API 实现人机交互

要审查、编辑和批准代理或工作流中的工具调用，请使用 LangGraph 的人机交互功能。

动态中断

Python
JavaScript
cURL

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'}

使用一些初始状态调用图。
当图遇到中断时，它会返回一个包含有效负载和元数据的中断对象。 3. 使用 Command(resume=...) 恢复图的执行，注入人类输入并继续执行。

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'}

使用一些初始状态调用图。
当图遇到中断时，它会返回一个包含有效负载和元数据的中断对象。
使用 { resume: ... } 命令对象恢复图的执行，注入人类输入并继续执行。

创建一个线程：

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`

这是一个可以在代理服务器中运行的示例图。有关更多详细信息，请参阅 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()

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

一旦你有了一个运行的代理服务器，你可以使用 LangGraph SDK 与之交互。

Python
JavaScript
cURL

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'}

使用一些初始状态调用图。
当图遇到中断时，它会返回一个包含有效负载和元数据的中断对象。 3. 使用 Command(resume=...) 恢复图的执行，注入人类输入并继续执行。

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'}

使用一些初始状态调用图。
当图遇到中断时，它会返回一个包含有效负载和元数据的中断对象。
使用 { resume: ... } 命令对象恢复图的执行，注入人类输入并继续执行。

创建一个线程：

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)!
)

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

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

Python
JavaScript
cURL

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

使用 interrupt_before 和 interrupt_after 参数调用 client.runs.wait。这是一个运行时配置，可以为每次调用更改。
interrupt_before 指定应在节点执行之前暂停执行的节点。
interrupt_after 指定应在节点执行之后暂停执行的节点。

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

使用 interruptBefore 和 interruptAfter 参数调用 client.runs.wait。这是一个运行时配置，可以为每次调用更改。
interruptBefore 指定应在节点执行之前暂停执行的节点。
interruptAfter 指定应在节点执行之后暂停执行的节点。

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
JavaScript
cURL

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)!
)

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

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)!
);

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

创建一个线程：

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 人机交互功能的信息。
常见模式：学习如何实现批准/拒绝操作、请求用户输入、工具调用审查和验证人类输入等模式。

将这些文档连接到 Claude、VSCode 等，通过 MCP 获取实时答案。

在 GitHub 上编辑此页面或提交问题。

Agent server

Core capabilities

Develop agents

Deployment guides

Studio

Auth & access control

Server customization

使用服务器 API 实现人机交互

动态中断

静态中断

了解更多

​动态中断

​静态中断

​了解更多

动态中断

静态中断

了解更多