使用服务器 API 的人在回路

要在智能体或工作流中审查、编辑和批准工具调用，请使用 LangGraph 的人在回路功能。

动态中断

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

这是一个可以在 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. 任何可 JSON 序列化的值都可以传给 interrupt 函数。这里传入的是一个包含待修订文本的字典。
3. 恢复后，interrupt(...) 的返回值就是人类提供的输入，用于更新状态。

当你的 Agent Server 运行起来后，你可以使用 LangGraph SDK 与它交互

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=...) 恢复图，注入人类的输入并继续执行。

静态中断

静态中断（也称为静态断点）会在节点执行之前或之后触发。

WARNING

不建议将静态中断用于人在回路工作流。它们最适合用于调试和测试。

你可以通过在编译时指定 interrupt_before 和 interrupt_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 指定执行应在节点执行之后暂停的节点。

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

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_before 和 interrupt_after 参数。这是运行时配置，可以在每次调用时更改。
2. interrupt_before 指定执行应在节点执行之前暂停的节点。
3. interrupt_after 指定执行应在节点执行之后暂停的节点。

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

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

1. 运行图，直到遇到第一个断点。
2. 通过为输入传入 None 来恢复图。这会运行图，直到遇到下一个断点。

了解更多

• 人在回路概念指南：进一步了解 LangGraph 的人在回路功能。
• 常见模式：了解如何实现批准/拒绝操作、请求用户输入、工具调用审查以及校验人类输入等模式。

---