interrupt_on 参数配置哪些工具需要批准。当 interrupt_on 被设置时，HumanInTheLoopMiddleware 会被添加到 深度智能体栈 中。如果一个运行在工具返回结果之前被取消或中断，同一个栈中的 @[PatchToolCallsMiddleware] 会自动修复消息历史记录。
:::python
:::
:::js
:::
基本配置
interrupt_on 参数接受一个将工具名称映射到中断配置的字典。每个工具可以配置为：
True：启用具有默认行为的中断 :::python (允许批准、编辑、拒绝和响应) ::: :::js (允许批准、编辑和拒绝) :::False：禁用此工具的中断InterruptOnConfig：自定义配置。设置allowed_decisions来控制审核选项。 :::python 在 Python 中，添加一个可选的when谓词来仅中断特定的调用（参见 条件中断）。 :::
from langchain.tools import tool
from deepagents import create_deep_agent
from langgraph.checkpoint.memory import MemorySaver
@tool
def remove_file(path: str) -> str:
"""Delete a file from the filesystem."""
return f"Deleted {path}"
@tool
def fetch_file(path: str) -> str:
"""Read a file from the filesystem."""
return f"Contents of {path}"
@tool
def notify_email(to: str, subject: str, body: str) -> str:
"""Send an email."""
return f"Sent email to {to}"
# Checkpointer is REQUIRED for human-in-the-loop
checkpointer = MemorySaver()
agent = create_deep_agent(
model="google_genai:gemini-3.6-flash",
tools=[remove_file, fetch_file, notify_email],
interrupt_on={
"remove_file": True, # Default: approve, edit, reject, respond
"fetch_file": False, # No interrupts needed
"notify_email": {"allowed_decisions": ["approve", "reject"]}, # No editing
},
checkpointer=checkpointer, # Required!
)
from langchain.tools import tool
from deepagents import create_deep_agent
from langgraph.checkpoint.memory import MemorySaver
@tool
def remove_file(path: str) -> str:
"""Delete a file from the filesystem."""
return f"Deleted {path}"
@tool
def fetch_file(path: str) -> str:
"""Read a file from the filesystem."""
return f"Contents of {path}"
@tool
def notify_email(to: str, subject: str, body: str) -> str:
"""Send an email."""
return f"Sent email to {to}"
# Checkpointer is REQUIRED for human-in-the-loop
checkpointer = MemorySaver()
agent = create_deep_agent(
model="openai:gpt-5.5",
tools=[remove_file, fetch_file, notify_email],
interrupt_on={
"remove_file": True, # Default: approve, edit, reject, respond
"fetch_file": False, # No interrupts needed
"notify_email": {"allowed_decisions": ["approve", "reject"]}, # No editing
},
checkpointer=checkpointer, # Required!
)
from langchain.tools import tool
from deepagents import create_deep_agent
from langgraph.checkpoint.memory import MemorySaver
@tool
def remove_file(path: str) -> str:
"""Delete a file from the filesystem."""
return f"Deleted {path}"
@tool
def fetch_file(path: str) -> str:
"""Read a file from the filesystem."""
return f"Contents of {path}"
@tool
def notify_email(to: str, subject: str, body: str) -> str:
"""Send an email."""
return f"Sent email to {to}"
# Checkpointer is REQUIRED for human-in-the-loop
checkpointer = MemorySaver()
agent = create_deep_agent(
model="anthropic:claude-sonnet-5",
tools=[remove_file, fetch_file, notify_email],
interrupt_on={
"remove_file": True, # Default: approve, edit, reject, respond
"fetch_file": False, # No interrupts needed
"notify_email": {"allowed_decisions": ["approve", "reject"]}, # No editing
},
checkpointer=checkpointer, # Required!
)
from langchain.tools import tool
from deepagents import create_deep_agent
from langgraph.checkpoint.memory import MemorySaver
@tool
def remove_file(path: str) -> str:
"""Delete a file from the filesystem."""
return f"Deleted {path}"
@tool
def fetch_file(path: str) -> str:
"""Read a file from the filesystem."""
return f"Contents of {path}"
@tool
def notify_email(to: str, subject: str, body: str) -> str:
"""Send an email."""
return f"Sent email to {to}"
# Checkpointer is REQUIRED for human-in-the-loop
checkpointer = MemorySaver()
agent = create_deep_agent(
model="openrouter:z-ai/glm-5.2",
tools=[remove_file, fetch_file, notify_email],
interrupt_on={
"remove_file": True, # Default: approve, edit, reject, respond
"fetch_file": False, # No interrupts needed
"notify_email": {"allowed_decisions": ["approve", "reject"]}, # No editing
},
checkpointer=checkpointer, # Required!
)
from langchain.tools import tool
from deepagents import create_deep_agent
from langgraph.checkpoint.memory import MemorySaver
@tool
def remove_file(path: str) -> str:
"""Delete a file from the filesystem."""
return f"Deleted {path}"
@tool
def fetch_file(path: str) -> str:
"""Read a file from the filesystem."""
return f"Contents of {path}"
@tool
def notify_email(to: str, subject: str, body: str) -> str:
"""Send an email."""
return f"Sent email to {to}"
# Checkpointer is REQUIRED for human-in-the-loop
checkpointer = MemorySaver()
agent = create_deep_agent(
model="fireworks:accounts/fireworks/models/glm-5p2",
tools=[remove_file, fetch_file, notify_email],
interrupt_on={
"remove_file": True, # Default: approve, edit, reject, respond
"fetch_file": False, # No interrupts needed
"notify_email": {"allowed_decisions": ["approve", "reject"]}, # No editing
},
checkpointer=checkpointer, # Required!
)
from langchain.tools import tool
from deepagents import create_deep_agent
from langgraph.checkpoint.memory import MemorySaver
@tool
def remove_file(path: str) -> str:
"""Delete a file from the filesystem."""
return f"Deleted {path}"
@tool
def fetch_file(path: str) -> str:
"""Read a file from the filesystem."""
return f"Contents of {path}"
@tool
def notify_email(to: str, subject: str, body: str) -> str:
"""Send an email."""
return f"Sent email to {to}"
# Checkpointer is REQUIRED for human-in-the-loop
checkpointer = MemorySaver()
agent = create_deep_agent(
model="baseten:zai-org/GLM-5.2",
tools=[remove_file, fetch_file, notify_email],
interrupt_on={
"remove_file": True, # Default: approve, edit, reject, respond
"fetch_file": False, # No interrupts needed
"notify_email": {"allowed_decisions": ["approve", "reject"]}, # No editing
},
checkpointer=checkpointer, # Required!
)
from langchain.tools import tool
from deepagents import create_deep_agent
from langgraph.checkpoint.memory import MemorySaver
@tool
def remove_file(path: str) -> str:
"""Delete a file from the filesystem."""
return f"Deleted {path}"
@tool
def fetch_file(path: str) -> str:
"""Read a file from the filesystem."""
return f"Contents of {path}"
@tool
def notify_email(to: str, subject: str, body: str) -> str:
"""Send an email."""
return f"Sent email to {to}"
# Checkpointer is REQUIRED for human-in-the-loop
checkpointer = MemorySaver()
agent = create_deep_agent(
model="ollama:north-mini-code-1.0",
tools=[remove_file, fetch_file, notify_email],
interrupt_on={
"remove_file": True, # Default: approve, edit, reject, respond
"fetch_file": False, # No interrupts needed
"notify_email": {"allowed_decisions": ["approve", "reject"]}, # No editing
},
checkpointer=checkpointer, # Required!
)
import { tool } from "langchain";
import { createDeepAgent } from "deepagents";
import { MemorySaver } from "@langchain/langgraph";
import { z } from "zod";
const removeFile = tool(
async ({ path }: { path: string }) => {
return `Deleted ${path}`;
},
{
name: "remove_file",
description: "Delete a file from the filesystem.",
schema: z.object({
path: z.string(),
}),
},
);
const fetchFile = tool(
async ({ path }: { path: string }) => {
return `Contents of ${path}`;
},
{
name: "fetch_file",
description: "Read a file from the filesystem.",
schema: z.object({
path: z.string(),
}),
},
);
const notifyEmail = tool(
async ({
to,
subject,
body,
}: {
to: string;
subject: string;
body: string;
}) => {
return `Sent email to ${to}`;
},
{
name: "notify_email",
description: "Send an email.",
schema: z.object({
to: z.string(),
subject: z.string(),
body: z.string(),
}),
},
);
// Checkpointer is REQUIRED for human-in-the-loop
const checkpointer = new MemorySaver();
const agent = createDeepAgent({
model: "google:gemini-3.6-flash",
tools: [removeFile, fetchFile, notifyEmail],
interruptOn: {
remove_file: true, // Default: approve, edit, reject, respond
fetch_file: false, // No interrupts needed
notify_email: { allowedDecisions: ["approve", "reject"] }, // No editing
},
checkpointer, // Required!
});
决策类型
allowed_decisions 列表控制当人类在审核工具调用时可以采取的操作：
| 决策类型 | 描述 | 用例示例 |
|---|---|---|
✅ approve | 使用智能体提出的原始参数执行工具。 | 按原样发送电子邮件草稿 |
✏️ edit | 在执行前修改工具参数。 | 发送电子邮件前更改收件人 |
❌ reject | 完全跳过执行此工具调用，并将拒绝反馈返回给智能体。 | 拒绝文件删除并说明原因 |
💬 respond | 直接返回用户消息作为合成工具结果，跳过执行（适用于“ask user”类型工具）。 | 用直接回复回答一个 "ask_user" 提示 |
reject。仅当人类作为工具行事时（例如，回答一个 ask_user 提示词）才使用 respond。不要使用 respond 来拒绝具有副作用的工具，因为其消息可能会被模型视为成功的工具结果。
:::
:::js
当人类拒绝提议的操作时，使用 reject。
:::
当 编辑 工具参数时，保守地做出更改。对原始参数进行重大修改可能会导致模型重新评估其方法，并可能多次执行工具或采取意外操作。
interrupt_on = {
# Sensitive operations: allow all options
"delete_file": {"allowed_decisions": ["approve", "edit", "reject"]},
# Moderate risk: approval or rejection only
"write_file": {"allowed_decisions": ["approve", "reject"]},
# Must approve (no rejection allowed)
"critical_operation": {"allowed_decisions": ["approve"]},
}
const interruptOn = {
// Sensitive operations: allow all options
delete_file: { allowedDecisions: ["approve", "edit", "reject"] },
// Moderate risk: approval or rejection only
write_file: { allowedDecisions: ["approve", "reject"] },
// Must approve (no rejection allowed)
critical_operation: { allowedDecisions: ["approve"] },
};
条件中断
默认情况下，interrupt_on 中列出的每个工具调用都会暂停以供审核。要仅暂停某些调用，请向工具的 InterruptOnConfig 添加一个 when 谓词。谓词接收一个 @[ToolCallRequest] 并返回 True 以中断或 False 以自动批准，因此您可以根据工具的参数进行控制。
条件中断需要
langchain>=1.3.3。from deepagents import create_deep_agent
from langchain.agents.middleware import ToolCallRequest
from langgraph.checkpoint.memory import MemorySaver
def writes_outside_workspace(request: ToolCallRequest) -> bool:
"""Pause writes to paths outside the workspace directory."""
path = request.tool_call["args"].get("file_path", "")
return not path.startswith("/workspace/")
agent = create_deep_agent(
model="google_genai:gemini-3.6-flash",
interrupt_on={
"write_file": {
"allowed_decisions": ["approve", "edit", "reject"],
"when": writes_outside_workspace,
},
},
checkpointer=MemorySaver(),
)
from deepagents import create_deep_agent
from langchain.agents.middleware import ToolCallRequest
from langgraph.checkpoint.memory import MemorySaver
def writes_outside_workspace(request: ToolCallRequest) -> bool:
"""Pause writes to paths outside the workspace directory."""
path = request.tool_call["args"].get("file_path", "")
return not path.startswith("/workspace/")
agent = create_deep_agent(
model="openai:gpt-5.5",
interrupt_on={
"write_file": {
"allowed_decisions": ["approve", "edit", "reject"],
"when": writes_outside_workspace,
},
},
checkpointer=MemorySaver(),
)
from deepagents import create_deep_agent
from langchain.agents.middleware import ToolCallRequest
from langgraph.checkpoint.memory import MemorySaver
def writes_outside_workspace(request: ToolCallRequest) -> bool:
"""Pause writes to paths outside the workspace directory."""
path = request.tool_call["args"].get("file_path", "")
return not path.startswith("/workspace/")
agent = create_deep_agent(
model="anthropic:claude-sonnet-5",
interrupt_on={
"write_file": {
"allowed_decisions": ["approve", "edit", "reject"],
"when": writes_outside_workspace,
},
},
checkpointer=MemorySaver(),
)
from deepagents import create_deep_agent
from langchain.agents.middleware import ToolCallRequest
from langgraph.checkpoint.memory import MemorySaver
def writes_outside_workspace(request: ToolCallRequest) -> bool:
"""Pause writes to paths outside the workspace directory."""
path = request.tool_call["args"].get("file_path", "")
return not path.startswith("/workspace/")
agent = create_deep_agent(
model="openrouter:z-ai/glm-5.2",
interrupt_on={
"write_file": {
"allowed_decisions": ["approve", "edit", "reject"],
"when": writes_outside_workspace,
},
},
checkpointer=MemorySaver(),
)
from deepagents import create_deep_agent
from langchain.agents.middleware import ToolCallRequest
from langgraph.checkpoint.memory import MemorySaver
def writes_outside_workspace(request: ToolCallRequest) -> bool:
"""Pause writes to paths outside the workspace directory."""
path = request.tool_call["args"].get("file_path", "")
return not path.startswith("/workspace/")
agent = create_deep_agent(
model="fireworks:accounts/fireworks/models/glm-5p2",
interrupt_on={
"write_file": {
"allowed_decisions": ["approve", "edit", "reject"],
"when": writes_outside_workspace,
},
},
checkpointer=MemorySaver(),
)
from deepagents import create_deep_agent
from langchain.agents.middleware import ToolCallRequest
from langgraph.checkpoint.memory import MemorySaver
def writes_outside_workspace(request: ToolCallRequest) -> bool:
"""Pause writes to paths outside the workspace directory."""
path = request.tool_call["args"].get("file_path", "")
return not path.startswith("/workspace/")
agent = create_deep_agent(
model="baseten:zai-org/GLM-5.2",
interrupt_on={
"write_file": {
"allowed_decisions": ["approve", "edit", "reject"],
"when": writes_outside_workspace,
},
},
checkpointer=MemorySaver(),
)
from deepagents import create_deep_agent
from langchain.agents.middleware import ToolCallRequest
from langgraph.checkpoint.memory import MemorySaver
def writes_outside_workspace(request: ToolCallRequest) -> bool:
"""Pause writes to paths outside the workspace directory."""
path = request.tool_call["args"].get("file_path", "")
return not path.startswith("/workspace/")
agent = create_deep_agent(
model="ollama:north-mini-code-1.0",
interrupt_on={
"write_file": {
"allowed_decisions": ["approve", "edit", "reject"],
"when": writes_outside_workspace,
},
},
checkpointer=MemorySaver(),
)
when 谓词返回 False 时，调用不会中断。当它返回 True，或当您省略 when 时，调用会像往常一样暂停。计算结果为 False 的调用永远不会被添加到中断批次中，因此审核者只看到需要决策的操作。
有关其他配置选项和示例，请参阅 LangChain 人工介入文档。
:::
处理中断
当中断被触发时，智能体会暂停执行并返回控制权。检查结果中的中断，并相应地处理它们。如果用户拒绝操作，请包含一个清晰的message，告诉智能体工具未被执行以及下一步该做什么。
:::python
from langchain_core.utils.uuid import uuid7
from langgraph.types import Command
# Create config with thread_id for state persistence
config = {"configurable": {"thread_id": str(uuid7())}}
# Invoke the agent
result = agent.invoke(
{"messages": [{"role": "user", "content": "Delete the file temp.txt"}]},
config=config
)
# Check if execution was interrupted
if result.get("__interrupt__"):
# Extract interrupt information
interrupt_value = result["__interrupt__"][0].value
action_requests = interrupt_value["action_requests"]
review_configs = interrupt_value["review_configs"]
# Create a lookup map from tool name to review config
config_map = {cfg["action_name"]: cfg for cfg in review_configs}
# Display the pending actions to the user
for action in action_requests:
review_config = config_map[action["name"]]
print(f"Tool: {action['name']}")
print(f"Arguments: {action['args']}")
print(f"Allowed decisions: {review_config['allowed_decisions']}")
# Get user decisions (one per action_request, in order)
decisions = [
{
"type": "reject",
"message": "User rejected deleting temp.txt. Do not retry deletion.",
}
]
# Resume execution with decisions
result = agent.invoke(
Command(resume={"decisions": decisions}),
config=config # Must use the same config!
)
# Process final result
print(result["messages"][-1].content)
import { v7 as uuid7 } from "uuid";
import { Command } from "@langchain/langgraph";
// Create config with thread_id for state persistence
const config = { configurable: { thread_id: uuid7() } };
// Invoke the agent
let result = await agent.invoke({
messages: [{ role: "user", content: "Delete the file temp.txt" }],
}, config);
// Check if execution was interrupted
if (result.__interrupt__) {
// Extract interrupt information
const interrupts = result.__interrupt__[0].value;
const actionRequests = interrupts.actionRequests;
const reviewConfigs = interrupts.reviewConfigs;
// Create a lookup map from tool name to review config
const configMap = Object.fromEntries(
reviewConfigs.map((cfg) => [cfg.actionName, cfg])
);
// Display the pending actions to the user
for (const action of actionRequests) {
const reviewConfig = configMap[action.name];
console.log(`Tool: ${action.name}`);
console.log(`Arguments: ${JSON.stringify(action.args)}`);
console.log(`Allowed decisions: ${reviewConfig.allowedDecisions}`);
}
// Get user decisions (one per actionRequest, in order)
const decisions = [
{
type: "reject",
message: "User rejected deleting temp.txt. Do not retry deletion.",
}
];
// Resume execution with decisions
result = await agent.invoke(
new Command({ resume: { decisions } }),
config // Must use the same config!
);
}
// Process final result
console.log(result.messages[result.messages.length - 1].content);
使用流式输出处理中断
对于实时 UI，使用.stream() 而不是 .invoke() 来在处理 HITL 中断时显示 Token 和工具调用进度。中断会作为 __interrupt__ 条目出现在 updates 流中。流完成后，收集人类决策并通过 Command(resume=...) 再次流式传输以恢复。
对于典型的
interrupt_on HITL 暂停，使用与 .invoke() 相同形状的扁平 {"decisions": [...]} 负载来恢复。当同时有多个中断待处理时（例如，并行分支），将每个中断 ID 映射到其恢复值。参见 处理多个中断。from langchain_core.utils.uuid import uuid7
from langgraph.types import Command
config = {"configurable": {"thread_id": str(uuid7())}}
stream_input = {"messages": [{"role": "user", "content": "Delete the file temp.txt"}]}
while True:
pending_interrupt = None
for mode, data in agent.stream(
stream_input,
stream_mode=["messages", "updates"],
config=config,
):
if mode == "messages":
token, metadata = data
# Display tokens in real time
if token.content:
print(token.content, end="", flush=True)
elif mode == "updates" and isinstance(data, dict):
if "__interrupt__" in data:
pending_interrupt = data["__interrupt__"][0]
# If no interrupt occurred, the agent finished
if pending_interrupt is None:
break
interrupt_value = pending_interrupt.value
action_requests = interrupt_value["action_requests"]
review_configs = interrupt_value["review_configs"]
config_map = {cfg["action_name"]: cfg for cfg in review_configs}
for action in action_requests:
review = config_map[action["name"]]
print(f"\nTool: {action['name']}")
print(f"Arguments: {action['args']}")
print(f"Allowed: {review['allowed_decisions']}")
# Collect one decision per action_request, in order
decisions = [{"type": "approve"} for _ in action_requests]
stream_input = Command(resume={"decisions": decisions})
print()
import { v7 as uuid7 } from "uuid";
import { Command } from "@langchain/langgraph";
const config = { configurable: { thread_id: uuid7() } };
let streamInput: Record<string, any> | Command = {
messages: [{ role: "user", content: "Delete the file temp.txt" }],
};
while (true) {
let pendingInterrupt: { value: any } | undefined;
for await (const chunk of await agent.stream(
streamInput,
{
streamMode: ["messages", "updates"],
...config,
}
)) {
const [mode, data] = chunk;
if (mode === "messages") {
const [token] = data;
// Display tokens in real time
if (token.text) {
process.stdout.write(token.text);
}
}
if (mode === "updates" && data?.__interrupt__) {
pendingInterrupt = data.__interrupt__[0];
}
}
// If no interrupt occurred, the agent finished
if (!pendingInterrupt) break;
const interruptValue = pendingInterrupt.value;
const actionRequests = interruptValue.actionRequests;
const reviewConfigs = interruptValue.reviewConfigs;
const configMap = Object.fromEntries(
reviewConfigs.map((cfg: any) => [cfg.actionName, cfg])
);
for (const action of actionRequests) {
const review = configMap[action.name];
console.log(`\nTool: ${action.name}`);
console.log(`Arguments: ${JSON.stringify(action.args)}`);
console.log(`Allowed: ${review.allowedDecisions}`);
}
// Collect one decision per actionRequest, in order
const decisions = actionRequests.map(() => ({ type: "approve" }));
streamInput = new Command({ resume: { decisions } });
}
process.stdout.write("\n");
结合
"messages" 和 "updates" 流式模式，在捕获 HITL 中断的同时向用户显示实时 LLM Token。"messages" 模式流式传输 Token；"updates" 模式发送中断负载。多个工具调用
当智能体调用多个需要批准的工具时，所有中断都会被批量组合在一个中断中。您必须按顺序为每个提供决策。 :::pythonconfig = {"configurable": {"thread_id": str(uuid7())}}
result = agent.invoke(
{"messages": [{
"role": "user",
"content": "Delete temp.txt and send an email to admin@example.com"
}]},
config=config
)
if result.get("__interrupt__"):
interrupt_value = result["__interrupt__"][0].value
action_requests = interrupt_value["action_requests"]
# Two tools need approval
assert len(action_requests) == 2
# Provide decisions in the same order as action_requests
decisions = [
{"type": "approve"}, # First tool: delete_file
{
"type": "reject",
"message": "User rejected this action. Do not retry this tool call.",
} # Second tool: send_email
]
result = agent.invoke(
Command(resume={"decisions": decisions}),
config=config
)
const config = { configurable: { thread_id: uuid7() } };
let result = await agent.invoke({
messages: [{
role: "user",
content: "Delete temp.txt and send an email to admin@example.com"
}]
}, config);
if (result.__interrupt__) {
const interrupts = result.__interrupt__[0].value;
const actionRequests = interrupts.actionRequests;
// Two tools need approval
console.assert(actionRequests.length === 2);
// Provide decisions in the same order as actionRequests
const decisions = [
{ type: "approve" }, // First tool: delete_file
{
type: "reject",
message: "User rejected this action. Do not retry this tool call.",
} // Second tool: send_email
];
result = await agent.invoke(
new Command({ resume: { decisions } }),
config
);
}
拒绝消息
当审核者返回一个reject 决策时，深度智能体会跳过工具调用并将拒绝反馈发送回智能体。如果您省略 message，默认反馈会告诉模型工具未被执行，除非用户请求，否则不要重试同一个工具调用。
对于敏感或具有副作用的工具，在决策时传递一个与领域相关的 message。明确说明智能体是否应该放弃操作、询问后续问题或尝试更安全的替代方案。
:::python
decisions = [
{
"type": "reject",
"message": "User rejected deleting this file. Do not retry deletion. Ask which file to archive instead.",
}
]
const decisions = [
{
type: "reject",
message: "User rejected deleting this file. Do not retry deletion. Ask which file to archive instead.",
},
];
编辑工具参数
当"edit" 在允许的决策中时，您可以在执行前修改工具参数：
:::python
if result.get("__interrupt__"):
interrupt_value = result["__interrupt__"][0].value
action_request = interrupt_value["action_requests"][0]
# Original args from the agent
print(action_request["args"]) # {"to": "everyone@company.com", ...}
# User decides to edit the recipient
decisions = [{
"type": "edit",
"edited_action": {
"name": action_request["name"], # Must include the tool name
"args": {"to": "team@company.com", "subject": "...", "body": "..."}
}
}]
result = agent.invoke(
Command(resume={"decisions": decisions}),
config=config
)
if (result.__interrupt__) {
const interrupts = result.__interrupt__[0].value;
const actionRequest = interrupts.actionRequests[0];
// Original args from the agent
console.log(actionRequest.args); // { to: "everyone@company.com", ... }
// User decides to edit the recipient
const decisions = [{
type: "edit",
editedAction: {
name: actionRequest.name, // Must include the tool name
args: { to: "team@company.com", subject: "...", body: "..." }
}
}];
result = await agent.invoke(
new Command({ resume: { decisions } }),
config
);
}
子智能体中断
在使用子智能体时，您可以在工具调用 上 和 内 中使用中断。工具调用上的中断
每个子智能体都可以有自己的interrupt_on 配置，该配置会覆盖主智能体的设置：
:::python
agent = create_deep_agent(
model="google_genai:gemini-3.6-flash",
tools=[delete_file, read_file],
interrupt_on={
"delete_file": True,
"read_file": False,
},
subagents=[{
"name": "file-manager",
"description": "Manages file operations",
"system_prompt": "You are a file management assistant.",
"tools": [delete_file, read_file],
"interrupt_on": {
# Override: require approval for reads in this subagent
"delete_file": True,
"read_file": True, # Different from main agent!
}
}],
checkpointer=checkpointer
)
const agent = createDeepAgent({
tools: [deleteFile, readFile],
interruptOn: {
delete_file: true,
read_file: false,
},
subagents: [{
name: "file-manager",
description: "Manages file operations",
systemPrompt: "You are a file management assistant.",
tools: [deleteFile, readFile],
interruptOn: {
// Override: require approval for reads in this subagent
delete_file: true,
read_file: true, // Different from main agent!
}
}],
checkpointer
});
__interrupt__ 并使用 Command 恢复。
当流式传输时，子智能体中断也会作为 __interrupt__ 条目出现在父 updates 流中。您不需要 subgraphs=True 来传递中断。如果您还想流式传输子智能体 Token 和进度事件，请启用 subgraphs=True。参见 使用流式输出处理中断 以获取完整模式。
工具调用内的中断
子智能体工具可以直接调用interrupt() 来暂停执行并等待批准：
:::python
from langchain.agents import create_agent
from langchain_anthropic import ChatAnthropic
from langchain.messages import HumanMessage
from langchain.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command, interrupt
from deepagents.graph import create_deep_agent
from deepagents.middleware.subagents import CompiledSubAgent
@tool(description="Request human approval before proceeding with an action.")
def request_approval(action_description: str) -> str:
"""Request human approval using the interrupt() primitive."""
# interrupt() pauses execution and returns the value passed to Command(resume=...)
approval = interrupt({
"type": "approval_request",
"action": action_description,
"message": f"Please approve or reject: {action_description}",
})
if approval.get("approved"):
return f"Action '{action_description}' was APPROVED. Proceeding..."
else:
return f"Action '{action_description}' was REJECTED. Reason: {approval.get('reason', 'No reason provided')}"
def main():
checkpointer = InMemorySaver()
model = ChatAnthropic(
model_name="claude-sonnet-4-6",
max_tokens=4096,
)
compiled_subagent = create_agent(
model=model,
tools=[request_approval],
name="approval-agent",
)
parent_agent = create_deep_agent(
model="google_genai:gemini-3.6-flash",
checkpointer=checkpointer,
subagents=[
CompiledSubAgent(
name="approval-agent",
description="An agent that can request approvals",
runnable=compiled_subagent,
)
],
)
thread_id = "test_interrupt_directly"
config = {"configurable": {"thread_id": thread_id}}
print("Invoking agent - sub-agent will use request_approval tool...")
result = parent_agent.invoke(
{
"messages": [
HumanMessage(
content="Use the task tool to launch the approval-agent sub-agent. "
"Tell it to use the request_approval tool to request approval for 'deploying to production'."
)
]
},
config=config
)
# Check for interrupt
if result.get("__interrupt__"):
interrupt_value = result["__interrupt__"][0].value
print(f"\nInterrupt received!")
print(f" Type: {interrupt_value.get('type')}")
print(f" Action: {interrupt_value.get('action')}")
print(f" Message: {interrupt_value.get('message')}")
print("\nResuming with Command(resume={'approved': True})...")
result2 = parent_agent.invoke(
Command(resume={"approved": True}),
config=config
)
if "__interrupt__" not in result2:
print("\nExecution completed!")
# Find the tool response
tool_msgs = [m for m in result2.get("messages", []) if m.type == "tool"]
if tool_msgs:
print(f" Tool result: {tool_msgs[-1].content}")
else:
print("\nAnother interrupt occurred")
else:
print("\n No interrupt - the model may not have called request_approval")
if __name__ == "__main__":
main()
Invoking agent - sub-agent will use request_approval tool...
Interrupt received!
Type: approval_request
Action: deploying to production
Message: Please approve or reject: deploying to production
Resuming with Command(resume={'approved': True})...
Execution completed!
Tool result: Great! The approval request has been processed. The action **"deploying to production"** was **APPROVED**. You can now proceed with the production deployment.
import { createAgent, tool } from "langchain";
import { ChatOpenAI } from "@langchain/openai";
import { HumanMessage } from "@langchain/core/messages";
import { MemorySaver, Command, interrupt } from "@langchain/langgraph";
import { createDeepAgent } from "deepagents";
import { z } from "zod";
const requestApproval = tool(
async ({ actionDescription }: { actionDescription: string }) => {
const approval = interrupt({
type: "approval_request",
action: actionDescription,
message: `Please approve or reject: ${actionDescription}`,
}) as { approved?: boolean; reason?: string };
if (approval.approved) {
return `Action '${actionDescription}' was APPROVED. Proceeding...`;
} else {
return `Action '${actionDescription}' was REJECTED. Reason: ${
approval.reason || "No reason provided"
}`;
}
},
{
name: "request_approval",
description: "Request human approval before proceeding with an action.",
schema: z.object({
actionDescription: z
.string()
.describe("The action that requires approval"),
}),
}
);
async function main() {
const checkpointer = new MemorySaver();
const model = new ChatOpenAI({
model: "gpt-5.4-mini",
maxTokens: 4096,
});
const compiledSubagent = createAgent({
model: model,
tools: [requestApproval],
name: "approval-agent",
});
const parentAgent = await createDeepAgent({
checkpointer: checkpointer,
subagents: [
{
name: "approval-agent",
description: "An agent that can request approvals",
runnable: compiledSubagent as any,
},
],
});
const threadId = "test_interrupt_directly";
const config = { configurable: { thread_id: threadId } };
console.log("Invoking agent - sub-agent will use request_approval tool...");
let result = await parentAgent.invoke(
{
messages: [
new HumanMessage({
content:
"Use the task tool to launch the approval-agent sub-agent. " +
"Tell it to use the request_approval tool to request approval for 'deploying to production'.",
}),
],
},
config
);
if (result.__interrupt__) {
const interruptValue = result.__interrupt__[0].value as {
type?: string;
action?: string;
message?: string;
};
console.log("\nInterrupt received!");
console.log(` Type: ${interruptValue.type}`);
console.log(` Action: ${interruptValue.action}`);
console.log(` Message: ${interruptValue.message}`);
console.log("\nResuming with Command(resume={'approved': true})...");
const result2 = await parentAgent.invoke(
new Command({ resume: { approved: true } }),
config
);
if (!result2.__interrupt__) {
console.log("\nExecution completed!");
// Find the tool response
const toolMsgs = result2.messages?.filter((m) => m.type === "tool") || [];
if (toolMsgs.length > 0) {
const lastToolMsg = toolMsgs[toolMsgs.length - 1];
console.log(` Tool result: ${lastToolMsg.content}`);
}
} else {
console.log("\nAnother interrupt occurred");
}
} else {
console.log(
"\n No interrupt - the model may not have called request_approval"
);
}
}
main().catch(console.error);
Invoking agent - sub-agent will use request_approval tool...
Interrupt received!
Type: approval_request
Action: deploying to production
Message: Please approve or reject: deploying to production
Resuming with Command(resume={'approved': true})...
Execution completed!
Tool result: Approval for "deploying to production" has been granted. You can proceed with the deployment.
文件系统权限中断
文件系统权限中断需要
deepagents>=0.6.8。interrupt_on 之外，你可以通过将 权限规则 标记为 mode="interrupt" 来暂停内置文件系统工具。当智能体在匹配中断模式规则的路径上调用 write_file 或 edit_file 时，create_deep_agent 会像配置的工具一样引发相同的人工介入中断，并使用文件系统工具的名称作为操作名称。
from deepagents import FilesystemPermission, create_deep_agent
from langgraph.checkpoint.memory import MemorySaver
agent = create_deep_agent(
model=model,
permissions=[
FilesystemPermission(
operations=["write"],
paths=["/secrets/**"],
mode="interrupt",
),
],
checkpointer=MemorySaver(), # Required to pause and resume
)
from langgraph.types import Command
config = {"configurable": {"thread_id": "fs-thread-1"}}
result = agent.invoke(
{"messages": [{"role": "user", "content": "Save the API key to /secrets/key.txt"}]},
config=config,
)
if result.get("__interrupt__"):
action = result["__interrupt__"][0].value["action_requests"][0]
print(f"Approve {action['name']} on {action['args']}?")
# Resume with the human decision (approve, edit, or reject).
result = agent.invoke(
Command(resume={"decisions": [{"type": "approve"}]}),
config=config, # Same thread ID
)
interrupt_on 合并，因此单个审查步骤可以覆盖自定义工具和受保护的文件系统路径。
:::
:::python
最佳实践
始终使用检查点器
人工介入需要检查点器来在中断和恢复之间持久化智能体状态：from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()
agent = create_deep_agent(
model="google_genai:gemini-3.6-flash",
tools=[...],
interrupt_on={...},
checkpointer=checkpointer # Required for HITL
)
使用相同的线程 ID
恢复时，你必须使用相同配置且具有相同thread_id：
# First call
config = {"configurable": {"thread_id": "my-thread"}}
result = agent.invoke(input, config=config)
# Resume (use same config)
result = agent.invoke(Command(resume={...}), config=config)
决策顺序与操作匹配
决策列表必须与action_requests 的顺序匹配：
if result.get("__interrupt__"):
interrupt_value = result["__interrupt__"][0].value
action_requests = interrupt_value["action_requests"]
# Create one decision per action, in order
decisions = []
for action in action_requests:
decision = get_user_decision(action) # Your logic
decisions.append(decision)
result = agent.invoke(
Command(resume={"decisions": decisions}),
config=config
)
根据风险调整配置
根据工具的风险级别配置不同的工具：interrupt_on = {
# High risk: full control (approve, edit, reject)
"delete_file": {"allowed_decisions": ["approve", "edit", "reject"]},
"send_email": {"allowed_decisions": ["approve", "edit", "reject"]},
# Medium risk: no editing allowed
"write_file": {"allowed_decisions": ["approve", "reject"]},
# Low risk: no interrupts
"read_file": False,
"ls": False,
}
官方原文：src/oss/deepagents/human-in-the-loop.mdx · 非官方翻译，译文由 AI 生成，以英文原文为准。