要审查、编辑和批准智能体或工作流程中的工具调用,请使用 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'}
  1. 图以初始状态调用。
  2. 遇到中断时会返回包含载荷与元数据的中断对象。
  3. 通过 Command(resume=...) 恢复图,注入人类输入后继续执行。
下方示例图可在 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()
  1. interrupt(...) 会在 human_node 暂停执行,并将载荷暴露给人工。
  2. interrupt 可接收任意可 JSON 序列化的值;此处传入需修改的文本。
  3. 恢复后,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'}
  1. 图以初始状态调用。
  2. 遇到中断时会返回包含载荷与元数据的中断对象。
  3. 通过 Command(resume=...) 恢复图,注入人类输入后继续执行。

静态中断

静态中断(亦称静态断点)会在节点执行前或执行后触发。
静态中断适合人在回路场景,主要用于调试与测试。
在编译阶段可通过 interrupt_beforeinterrupt_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
  • 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)!
)
  1. client.runs.wait 在运行时传入 interrupt_beforeinterrupt_after,可为每次调用单独配置。
  2. interrupt_before 指定执行前需暂停的节点。
  3. interrupt_after 指定执行后需暂停的节点。
下面示例展示如何添加静态中断:
  • 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)!
)
  1. 图运行至首个断点。
  2. 将输入设为 None 以恢复,并继续运行至下一个断点。

进一步了解

  • 人在回路概念指南:深入了解 LangGraph 的人在回路能力。
  • 常见模式:学习如何实现动作审批/拒绝、请求用户输入、工具调用审核、人类输入校验等模式。

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