人机协作（Human-in-the-loop）¶

LangGraph 支持强大的 人机协作（HIL） 工作流程，允许在自动化流程的任何环节进行人工干预。这在大型语言模型（LLM）驱动的应用中尤其有用，因为模型的输出可能需要验证、更正或补充额外的上下文。

如需了解更多关于 LangGraph 人机协作功能的信息，请参阅 LangGraph 人机协作概述。

`interrupt`¶

LangGraph 中的 [interrupt 函数][langgraph.types.interrupt] 通过在特定节点暂停图、向人类展示信息并使用人类输入恢复图，从而支持人机协作的工作流程。它适用于审批、编辑或收集额外上下文等任务。

图通过一个 [Command][langgraph.types.Command] 对象恢复，该对象提供了人类的响应。

带有 interrupt 的图节点：

^{API Reference: interrupt | Command}

from langgraph.types import interrupt, Command

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

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

LangGraph API 调用与恢复：

PythonJavaScriptcURL

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

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

^{API Reference: InMemorySaver | START | StateGraph | interrupt | Command}

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 API 服务器，你可以使用 LangGraph SDK

PythonJavaScriptcURL

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

学习更多¶

LangGraph 人工参与概述: 了解 LangGraph 人工参与功能的更多信息。
设计模式: 学习如何实现批准/拒绝操作、请求用户输入等模式。
如何审查工具调用: 详细了解如何审查和批准/编辑工具调用或向调用工具的 LLM 提供反馈。

人机协作（Human-in-the-loop）¶

interrupt¶

学习更多¶

`interrupt`¶