일부 tool 작업은 민감할 수 있으며 실행 전에 사람의 승인이 필요할 수 있습니다. Deep agent는 LangGraph의 interrupt 기능을 통해 human-in-the-loop workflow를 지원합니다. interrupt_on parameter를 사용하여 승인이 필요한 tool을 구성할 수 있습니다.

​
기본 구성

interrupt_on parameter는 tool 이름을 interrupt 구성에 매핑하는 dictionary를 받습니다. 각 tool은 다음과 같이 구성할 수 있습니다:
  • True: 기본 동작으로 interrupt 활성화 (approve, edit, reject 허용)
  • False: 이 tool에 대한 interrupt 비활성화
  • {"allowed_decisions": [...]}: 특정 허용 결정이 있는 사용자 정의 구성
from langchain_core.tools import tool
from deepagents import create_deep_agent
from langgraph.checkpoint.memory import MemorySaver

@tool
def delete_file(path: str) -> str:
    """Delete a file from the filesystem."""
    return f"Deleted {path}"

@tool
def read_file(path: str) -> str:
    """Read a file from the filesystem."""
    return f"Contents of {path}"

@tool
def send_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-4-20250514",
    tools=[delete_file, read_file, send_email],
    interrupt_on={
        "delete_file": True,  # Default: approve, edit, reject
        "read_file": False,   # No interrupts needed
        "send_email": {"allowed_decisions": ["approve", "reject"]},  # No editing
    },
    checkpointer=checkpointer  # Required!
)

​
Decision 유형

allowed_decisions list는 tool 호출을 검토할 때 사람이 취할 수 있는 작업을 제어합니다:
  • "approve": agent가 제안한 원래 argument로 tool 실행
  • "edit": 실행 전에 tool argument 수정
  • "reject": 이 tool 호출 실행을 완전히 건너뜀
각 tool에 대해 사용 가능한 decision을 사용자 정의할 수 있습니다: 
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"]},
}

​
Interrupt 처리

Interrupt가 트리거되면 agent는 실행을 일시 중지하고 제어권을 반환합니다. 결과에서 interrupt를 확인하고 적절하게 처리하세요. 
import uuid
from langgraph.types import Command

# Create config with thread_id for state persistence
config = {"configurable": {"thread_id": str(uuid.uuid4())}}

# 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
    interrupts = result["__interrupt__"][0].value
    action_requests = interrupts["action_requests"]
    review_configs = interrupts["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": "approve"}  # User approved the 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"])

​
여러 tool 호출

Agent가 승인이 필요한 여러 tool을 호출하면 모든 interrupt가 단일 interrupt로 일괄 처리됩니다. 각각에 대해 순서대로 decision을 제공해야 합니다. 
config = {"configurable": {"thread_id": str(uuid.uuid4())}}

result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "Delete temp.txt and send an email to [email protected]"
    }]
}, config=config)

if result.get("__interrupt__"):
    interrupts = result["__interrupt__"][0].value
    action_requests = interrupts["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"}    # Second tool: send_email
    ]

    result = agent.invoke(
        Command(resume={"decisions": decisions}),
        config=config
    )

​
Tool argument 편집

허용된 decision에 "edit"이 있으면 실행 전에 tool argument를 수정할 수 있습니다: 
if result.get("__interrupt__"):
    interrupts = result["__interrupt__"][0].value
    action_request = interrupts["action_requests"][0]

    # Original args from the agent
    print(action_request["args"])  # {"to": "[email protected]", ...}

    # User decides to edit the recipient
    decisions = [{
        "type": "edit",
        "edited_action": {
            "name": action_request["name"],  # Must include the tool name
            "args": {"to": "[email protected]", "subject": "...", "body": "..."}
        }
    }]

    result = agent.invoke(
        Command(resume={"decisions": decisions}),
        config=config
    )

​
Subagent interrupt

각 subagent는 main agent의 설정을 재정의하는 자체 interrupt_on 구성을 가질 수 있습니다: 
agent = create_deep_agent(
    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
)
Subagent가 interrupt를 트리거하면 처리 방법은 동일합니다 – __interrupt__를 확인하고 Command로 재개하세요.

​
모범 사례

​
항상 checkpointer 사용

Human-in-the-loop는 interrupt와 재개 사이에 agent 상태를 유지하기 위해 checkpointer가 필요합니다: 
from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
agent = create_deep_agent(
    tools=[...],
    interrupt_on={...},
    checkpointer=checkpointer  # Required for HITL
)

​
동일한 thread ID 사용

재개할 때 동일한 thread_id를 가진 동일한 config를 사용해야 합니다: 
# 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)

​
Decision 순서를 action과 일치

Decision list는 action_requests의 순서와 일치해야 합니다: 
if result.get("__interrupt__"):
    interrupts = result["__interrupt__"][0].value
    action_requests = interrupts["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
    )

​
위험도에 따라 구성 조정

위험 수준에 따라 다른 tool을 구성하세요: 
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,
    "list_files": False,
}
Edit the source of this page on GitHub.
Connect these docs programmatically to Claude, VSCode, and more via MCP for real-time answers.
⌘I