使用服务器 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'}

图以初始状态调用。
遇到中断时会返回包含载荷与元数据的中断对象。
通过 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`

下方示例图可在 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)!
    }


# Build the graph
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 暂停执行，并将载荷暴露给人工。
interrupt 可接收任意可 JSON 序列化的值；此处传入需修改的文本。
恢复后，interrupt(...) 的返回值即人工输入，用于更新状态。

Once you have a running Agent Server, you can interact with it using LangGraph SDK

Python
JavaScript
cURL

from langgraph_sdk import get_client
from langgraph_sdk.schema import Command
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 until the interrupt is hit.
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'
# >     }
# > ]


# Resume the graph
print(await client.runs.wait(
    thread_id,
    assistant_id,
    command=Command(resume="Edited text")   # (3)!
))
# > {'some_text': 'Edited text'}

图以初始状态调用。
遇到中断时会返回包含载荷与元数据的中断对象。
通过 Command(resume=...) 恢复图，注入人类输入后继续执行。

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 until the interrupt is hit.
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'
# >     }
# > ]

// Resume the graph
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)!
)

client.runs.wait 在运行时传入 interrupt_before 与 interrupt_after，可为每次调用单独配置。
interrupt_before 指定执行前需暂停的节点。
interrupt_after 指定执行后需暂停的节点。

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

client.runs.wait 在运行时传入 interruptBefore/interruptAfter，可按需配置。
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 的人在回路能力。
常见模式：学习如何实现动作审批/拒绝、请求用户输入、工具调用审核、人类输入校验等模式。

Edit the source of this page on GitHub.

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

​动态中断

​静态中断

​进一步了解

动态中断

静态中断

进一步了解