主题
要审查、编辑和批准智能体或工作流中的工具调用，请使用 LangGraph 的 人机交互 功能。
动态中断
Python
JavaScript
cURL
python
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. 当图遇到中断时，它会返回一个包含有效负载和元数据的中断对象。
- 使用
Command(resume=...)恢复图，注入人工输入并继续执行。
扩展示例：使用 interrupt
这是一个你可以在 Agent Server 中运行的示例图。 更多详情请参阅 LangSmith 快速入门。
python
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(...)的返回值是人工提供的输入，用于更新状态。
一旦你有一个运行中的 Agent Server，就可以使用 LangGraph SDK 与之交互。
Python
JavaScript
cURL
python
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=...)恢复图，注入人工输入并继续执行。
静态中断
静态中断（也称为静态断点）在节点执行之前或之后触发。
不推荐将静态中断用于人机交互工作流。它们最适合用于调试和测试。
你可以在编译时通过指定 interrupt_before 和 interrupt_after 来设置静态中断：
python
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
python
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` 指定在执行节点之后应暂停执行的节点。
以下示例展示了如何添加静态中断：
Python
JavaScript
cURL
python
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` 作为输入来恢复图的执行。这将运行图直到遇到下一个断点。