Human-in-the-loop
Human-in-the-loop (사람 개입 승인)
민감한 툴 연산에 대해 실행 전 사람의 승인을 구성하는 방법을 배워요. 파일 삭제나 이메일 전송 같은 일부 툴 연산은 실행 전에 사람의 승인이 필요할 수 있죠. 딥 에이전트(Deep Agents)는 LangGraph의 인터럽트(interrupt) 기능을 통해 human-in-the-loop 워크플로우를 지원해요.
출처: 공식문서
interrupt_on 파라미터로 승인이 필요한 툴을 구성해요. interrupt_on을 설정하면 HumanInTheLoopMiddleware가 Deep Agents 스택에 추가돼요. 툴이 결과를 반환하기 전에 실행이 취소되거나 인터럽트되면, 같은 스택의 PatchToolCallsMiddleware가 메시지 히스토리를 자동으로 복구해줘요.
graph LR
Agent[Agent] --> Check{Interrupt?}
Check --> |no| Execute[Execute]
Check --> |yes| Human{Human}
Human --> |approve| Execute
Human --> |edit| Execute
Human --> |reject| ToolMessage[ToolMessage]
Human --> |respond| ToolMessage
Execute --> Agent
ToolMessage --> Agent
기본 구성 (Basic configuration)
interrupt_on 파라미터는 툴 이름을 인터럽트 구성으로 매핑하는 딕셔너리를 받아요. 각 툴은 다음 중 하나로 구성할 수 있어요.
True: 기본 동작으로 인터럽트 활성화 (approve, edit, reject, respond 허용)False: 이 툴에 대해 인터럽트 비활성화InterruptOnConfig: 커스텀 구성.allowed_decisions를 설정해 검토 옵션을 제어해요. Python에서는 선택적으로when프레디케이트를 추가해 특정 호출만 인터럽트해요 ("Conditional interrupts" 참고).
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-4-6",
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!
)
모델명만 바꾸면 다른 프로바이더에서도 같은 코드를 그대로 쓸 수 있어요. 예를 들어 model="openai:gpt-5.5", model="google_genai:gemini-3.6-flash", model="openrouter:z-ai/glm-5.2", model="fireworks:accounts/fireworks/models/glm-5p2", model="baseten:zai-org/GLM-5.2", model="ollama:north-mini-code-1.0" 로 교체하면 되죠.
결정 유형 (Decision types)
allowed_decisions 목록은 툴 호출을 검토할 때 사람이 취할 수 있는 동작을 제어해요.
| 결정 유형 | 설명 | 예시 사용 사례 |
|---|---|---|
✅ approve |
에이전트가 제안한 원래 인자 그대로 툴을 실행 | 이메일 초안을 그대로 전송 |
✏️ edit |
실행 전 툴 인자 수정 | 이메일 보내기 전에 수신자 변경 |
❌ reject |
이 툴 호출을 완전히 건너뛰고 에이전트에 거부 피드백 반환 | 파일 삭제를 거부하고 이유 설명 |
💬 respond |
실행을 건너뛰고 사람의 메시지를 합성 툴 결과로 직접 반환 ("ask user" 스타일 툴용) | "ask_user" 프롬프트에 직접 답변 |
reject는 사람이 제안된 동작을 거부할 때 써요. respond는 사람이 툴 역할을 할 때(예: ask_user 프롬프트에 답변)만 사용하세요. 부작용이 있는 툴을 거부할 때 respond를 쓰면 안 돼요. 그 메시지가 모델에 의해 성공적인 툴 결과로 처리될 수 있기 때문이에요.
편집 시 주의: 툴 인자를 편집할 때는 보수적으로 변경하세요. 원래 인자를 크게 변경하면 모델이 접근 방식을 재평가해 툴을 여러 번 실행하거나 예상 밖의 동작을 할 수 있어요.
각 툴에 대해 어떤 결정을 허용할지 커스터마이즈할 수 있어요.
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"]},
}
조건부 인터럽트 (Conditional interrupts)
기본적으로 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="anthropic:claude-sonnet-4-6",
interrupt_on={
"write_file": {
"allowed_decisions": ["approve", "edit", "reject"],
"when": writes_outside_workspace,
},
},
checkpointer=MemorySaver(),
)
when 프레디케이트가 False를 반환하면 호출은 인터럽트 없이 실행돼요. True를 반환하거나 when을 생략하면 평소처럼 호출이 멈춰요. False로 평가된 호출은 인터럽트 배치에 절대 추가되지 않아서, 검토자는 결정이 필요한 동작만 보게 돼요.
추가 구성 옵션과 예제는 LangChain human-in-the-loop 문서를 참고해요.
인터럽트 처리 (Handle interrupts)
인터럽트가 트리거되면 에이전트는 실행을 멈추고 제어권을 반환해요. 결과에서 인터럽트를 확인하고 그에 따라 처리해요. 사용자가 동작을 거부하면, 에이전트에게 툴이 실행되지 않았고 다음에 무엇을 해야 하는지 알려주는 명확한 message를 포함하세요.
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,
version="v2",
)
# Check if execution was interrupted
if result.interrupts:
# Extract interrupt information
interrupt_value = result.interrupts[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!
version="v2",
)
# Process final result
print(result.value["messages"][-1].content)
여러 툴 호출 (Multiple tool calls)
에이전트가 승인이 필요한 여러 툴을 호출하면 모든 인터럽트가 단일 인터럽트로 배치돼요. 각각에 대해 순서대로 결정을 제공해야 해요.
config = {"configurable": {"thread_id": str(uuid7())}}
result = agent.invoke(
{"messages": [{
"role": "user",
"content": "Delete temp.txt and send an email to [email protected]"
}]},
config=config,
version="v2",
)
if result.interrupts:
interrupt_value = result.interrupts[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,
version="v2",
)
거부 메시지 (Rejection messages)
검토자가 reject 결정을 반환하면 딥 에이전트는 툴 호출을 건너뛰고 에이전트에게 거부 피드백을 보내요. message를 생략하면 기본 피드백이 모델에게 툴이 실행되지 않았고 사용자가 요청하지 않는 한 같은 툴 호출을 재시도하지 말라고 알려줘요.
민감하거나 부작용이 있는 툴에는 결정과 함께 도메인 특화 message를 전달하세요. 에이전트가 동작을 포기할지, 후속 질문을 할지, 더 안전한 대안을 시도할지를 명확히 하세요.
decisions = [
{
"type": "reject",
"message": "User rejected deleting this file. Do not retry deletion. Ask which file to archive instead.",
}
]
툴 인자 편집 (Edit tool arguments)
allowed_decisions에 "edit"가 있으면 실행 전에 툴 인자를 수정할 수 있어요.
if result.interrupts:
interrupt_value = result.interrupts[0].value
action_request = interrupt_value["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,
version="v2",
)
서브에이전트 인터럽트 (Subagent interrupts)
서브에이전트를 사용하면 툴 호출에 대한 인터럽트와 툴 호출 내부 인터럽트를 사용할 수 있어요.
툴 호출에 대한 인터럽트 (Interrupts on tool calls)
각 서브에이전트는 메인 에이전트 설정을 오버라이드하는 자체 interrupt_on 구성을 가질 수 있어요.
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
)
서브에이전트가 인터럽트를 트리거하면 처리는 동일해요 — 결과의 interrupts를 확인하고 Command로 재개하면 됩니다.
툴 호출 내부 인터럽트 (Interrupts within tool calls)
서브에이전트 툴은 interrupt()를 직접 호출해 실행을 멈추고 승인을 기다릴 수 있어요.
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,
version="v2",
)
# Check for interrupt
if result.interrupts:
interrupt_value = result.interrupts[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,
version="v2",
)
if not result2.interrupts:
print("\nExecution completed!")
# Find the tool response
tool_msgs = [m for m in result2.value.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.
파일시스템 권한 인터럽트 (Filesystem permission interrupts)
파일시스템 권한 인터럽트는
deepagents>=0.6.8이 필요해요.
interrupt_on 외에도 내장 파일시스템 툴을 퍼미션 규칙에 mode="interrupt"를 표시해 멈추게 할 수 있어요. 에이전트가 인터럽트 모드 규칙과 일치하는 경로에 write_file 또는 edit_file을 호출하면, create_deep_agent는 구성된 툴과 동일한 human-in-the-loop 인터럽트를 일으키고 파일시스템 툴의 이름을 액션 이름으로 사용해요.
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,
version="v2",
)
if result.interrupts:
action = result.interrupts[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
version="v2",
)
파일시스템 권한 인터럽트는 전달한 interrupt_on과 병합돼서, 단일 검토 단계로 커스텀 툴과 보호된 파일시스템 경로를 모두 다룰 수 있어요.
모범 사례 (Best practices)
항상 체크포인터 사용 (Always use a checkpointer)
Human-in-the-loop은 인터럽트와 재개 사이의 에이전트 상태를 유지하려면 체크포인터가 필요해요.
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
)
같은 thread ID 사용 (Use the same thread ID)
재개할 때는 반드시 같은 thread_id를 가진 같은 config를 사용해야 해요.
# First call
config = {"configurable": {"thread_id": "my-thread"}}
result = agent.invoke(input, config=config, version="v2")
# Resume (use same config)
result = agent.invoke(Command(resume={...}), config=config, version="v2")
결정 순서를 액션 순서와 일치 (Match decision order to actions)
결정 목록은 action_requests의 순서와 일치해야 해요.
if result.interrupts:
interrupt_value = result.interrupts[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,
version="v2",
)
위험도별로 구성 다듬기 (Tailor configurations by risk)
툴마다 위험 수준에 따라 다르게 구성해요.
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,
}
더 알아보기 (Learn more)
- LangChain human-in-the-loop — 추가 구성 옵션과 예제
- Permissions — 파일시스템 권한 규칙
- Customization — Deep Agents 스택 구성