YouTipLangchain Human In The Loop
In production environments, some operations require human confirmation—such as sending emails, executing deletions, or processing payments.
Human-in-the-Loop (HITL) allows agents to pause at critical moments and wait for human approval before continuing.
* * *
## interrupt()——Pause Execution in Tools
The interrupt() function allows tool execution to pause halfway and wait for external input before continuing:
## Example
from langgraph.types import interrupt
# Use interrupt() to pause in tools
def send_email(to: str, subject: str, body: str) ->str:
"""Send email (requires human approval)"""
# Pause execution and send approval request to external system
approval = interrupt({
"action": "send_email",
"to": to,
"subject": subject,
"body": body,
"message": "Please confirm whether to send this email?"
})
# Wait for external approval input before continuing
if approval.get("approved"):
return f"Email has been sent to {to}"
else:
return f"Email sending was rejected: {approval.get('reason', 'User cancelled')}"
interrupt() workflow:
1. Tool calls interrupt() → Agent pauses execution
2. External system retrieves interrupt information and displays it to user
3. After user makes a decision, resume execution via Command(resume=...)
4. interrupt() returns the value passed by the user, and the tool continues execution
* * *
## Complete Example——Approval Workflow
## Example
from dotenv import load_dotenv
load_dotenv()
from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import InMemorySaver
from langchain.tools import tool
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
@tool
def delete_course(course_name: str) ->str:
"""Delete course (requires approval).
Args:
course_name: The name of the course to delete
"""
# Pause and wait for approval
approval = interrupt({
"action": "delete_course",
"course": course_name,
"message": f"Confirm deletion of course \\"{course_name}\\"? This action cannot be undone."
})
if approval.get("confirmed"):
return f"Course \\"{course_name}\\" has been deleted"
else:
return f"Deletion operation cancelled"
checkpointer = InMemorySaver()
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=,
checkpointer=checkpointer,
system_prompt="You are the administrator assistant for Tutorial TUTORIAL.",
)
config ={"configurable": {"thread_id": "admin-001"}}
# Step 1: Initiate deletion request (will trigger interrupt)
print("=== Start Execution ===")
result = agent.invoke(
{"messages": [HumanMessage(content="Please delete the course \\"Outdated Java Tutorial\\"")]},
config=config,
)
# Check if agent has paused
state = agent.get_state(config)
print(f"Status: {state.next}")# ('tools',) means paused at tools node
print(f"Interrupt Info: {state.tasks.interrupts}")
# Step 2: Human approval (simulate user clicking "Confirm")
print("
=== Human Approval ===")
resume_value ={"confirmed": True,"operator": "Admin Zhang San"}
result = agent.invoke(
Command(resume=resume_value),
config=config,
)
print(f"Final Response: {result['messages'].content}")
Run result:
=== Start Execution ===Status: ('tools',)Interrupt Info: (Interrupt(value={'action': 'delete_course', ...}),)=== Human Approval ===Final Response: Course "Outdated Java Tutorial" has been deleted.
* * *
## interrupt_before / interrupt_after Parameters
In addition to using interrupt() in tools, you can also set global interrupt points in create_agent():
## Example
from langgraph.checkpoint.memory import InMemorySaver
checkpointer = InMemorySaver()
agent = create_agent(
model="deepseek:deepseek-v4-flash",
tools=,
checkpointer=checkpointer,
# Pause before tools node (requires approval before each tool call)
interrupt_before=,
# Pause after model node (can inspect after each model response)
# interrupt_after=,
)
| Parameter | Pause Timing | Use Case |
| --- | --- | --- |
| interrupt_before= | Before each tool execution | All tool calls require approval |
| interrupt_before= | Before each model call | Human reviews message before model processes it |
| interrupt_after= | After each model response | Review model output before deciding whether to continue |
| interrupt_after= | After each tool execution | Check tool results before deciding next step |
* * *
## Typical HITL Architecture
In actual web applications, HITL is typically implemented as follows:
## Example
# Backend: Receive user message, process to interrupt point, return interrupt info
def handle_user_message(thread_id: str, message: str):
config ={"configurable": {"thread_id": thread_id}}
result = agent.invoke(
{"messages": [HumanMessage(content=message)]},
config=config,
)
state = agent.get_state(config)
# Check if waiting for approval
if state.tasks and state.tasks.interrupts:
return{
"status": "pending_approval",
"interrupt": state.tasks.interrupts.value,
"thread_id": thread_id,
}
return{
"status": "completed",
"reply": result.content,
}
# Backend: Handle user approval
def handle_approval(thread_id: str, approved: bool, reason: str=""):
config ={"configurable": {"thread_id": thread_id}}
result = agent.invoke(
Command(resume={"confirmed": approved,"reason": reason}),
config=config,
)
return{"status": "completed","reply": result.content}
> HITL requires Checkpointer to work together. Because when an agent pauses at interrupt(), its state must be persisted in order to correctly continue execution