使用服务器 API 的人工干预¶

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

动态中断¶

PythonJavaScriptcURL

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

这是一个可以在 LangGraph API 服务器中运行的示例图表。有关更多详细信息，请参阅LangGraph 平台快速入门。

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 暂停执行，将给定的有效负载呈现给人工。
任何可 JSON 序列化的值都可以传递给 interrupt 函数。这里是一个包含要修改文本的字典。
一旦恢复，interrupt(...) 的返回值是人工提供的输入，用于更新状态。

一旦您拥有一个正在运行的 LangGraph API 服务器，就可以使用LangGraph SDK与它进行交互

PythonJavaScriptcURL

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 指定了在节点执行后应该暂停执行的节点。

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

PythonJavaScriptcURL

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

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

PythonJavaScriptcURL

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 until the breakpoint
result = await client.runs.wait(
    thread_id,
    assistant_id,
    input=inputs   # (1)!
)

# Resume the graph
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> });

// 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 breakpoint
const result = await client.runs.wait(
  threadID,
  assistantID,
  { input: input }   // (1)!
);

// Resume the graph
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 人工干预功能。
常见模式：了解如何实现诸如批准/拒绝操作、请求用户输入、工具调用审查以及验证人工输入等模式。