AI agents are powerful — but sometimes too powerful. What if your agent is about to send an email on your behalf and you want a chance to say "wait, hold on"? That's exactly what Human-in-the-Loop (HITL) is for.
In this article, we'll build an AI agent that can send emails, but pauses and asks for your approval before actually doing it. If you approve, it goes through. If you reject, it doesn't. Simple, safe, and production-ready.
What We're Building
A LangChain agent powered by an LLM that:
- Receives a user request like "Send an email to talha@gmail.com"
- Decides to call the
send_emailtool - Pauses and waits for human approval
- Sends the email (or drops it) based on your decision
Prerequisites
pip install langchain langgraph
You'll also need an LLM. In this article we'll use OpenAI's GPT-4o-mini via langchain-openai, but you can swap in any LangChain-compatible model.
pip install langchain-openai
Set your API key:
export OPENAI_API_KEY="your-key-here"
Step 1: Set Up the Model
from langchain_openai import ChatOpenAI
# You can swap this for any LangChain-compatible chat model
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
Using a different provider? LangChain supports Anthropic, Mistral, Groq, and many others. Just replace
ChatOpenAIwith the appropriate class and install its package (e.g.langchain-anthropicfor Claude).
Step 2: Define the Tool
This is the action the agent will want to take. We keep it simple — just a function that prints the email details.
def send_email(email: str, message: str) -> str:
"""Send email"""
print(f"\nEMAIL SENT TO: {email}")
print(f"MESSAGE: {message}")
return "Email sent successfully"
In a real app, you'd integrate this with SendGrid, SES, or your SMTP server.
Step 3: Create the Agent with Human-in-the-Loop Middleware
Here's where the magic happens. We use:
-
create_agent— builds the agent with tools attached -
HumanInTheLoopMiddleware— tells the agent to pause before calling specific tools -
InMemorySaver— a checkpointer that saves agent state so it can be resumed after the pause
import warnings
warnings.filterwarnings("ignore", category=UserWarning)
warnings.filterwarnings("ignore", category=DeprecationWarning)
from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command
agent = create_agent(
model=model,
tools=[send_email],
middleware=[HumanInTheLoopMiddleware(interrupt_on={"send_email": True})],
checkpointer=InMemorySaver(),
)
The key line is interrupt_on={"send_email": True}. This tells the middleware: "Before you actually run send_email, freeze and hand control back to the human."
The InMemorySaver checkpointer is what makes resuming possible — it snapshots the agent's state at the interruption point so you can pick up exactly where you left off.
Step 4: First Invocation — Let the Agent Plan
We kick things off with the user's request. The agent will reason about what to do and reach the point of calling send_email — then stop.
config = {"configurable": {"thread_id": "test-thread"}}
response = agent.invoke(
{
"messages": [
{
"role": "user",
"content": "Send an email to talha@gmail.com with message 'hello world'",
}
]
},
config=config,
)
print("\nAGENT PAUSED FOR APPROVAL\n")
The thread_id in config is how LangGraph tracks this specific conversation. Use the same thread ID when you resume so it picks up the right checkpoint.
Step 5: Ask the Human
Now we prompt the user in the terminal. This is the "human" part of human-in-the-loop.
decision = input("Approve email? (yes/no): ")
Step 6: Resume Based on the Decision
This is where Command(resume=...) comes in. We send the agent's paused state a signal telling it whether to proceed or abort.
If Approved
if decision.lower() == "yes":
response = agent.invoke(
Command(resume={"decisions": [{"type": "approve"}]}),
config=config,
)
print("\nFINAL RESPONSE:")
print(response["messages"][-1].content)
The agent resumes from the checkpoint, actually calls send_email, and continues to completion.
If Rejected
else:
response = agent.invoke(
Command(
resume={
"decisions": [{"type": "reject", "message": "Human rejected the email"}]
}
),
config=config,
)
print("\nEMAIL REJECTED")
The agent is told the action was rejected. It can use the rejection message to inform its final response (e.g., "The email was not sent as per your instruction.").
Full Code
Here's everything together:
import warnings
warnings.filterwarnings("ignore", category=UserWarning)
warnings.filterwarnings("ignore", category=DeprecationWarning)
from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command
# ── Model ──────────────────────────────────────────────────────────────────────
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# ── Tool ───────────────────────────────────────────────────────────────────────
def send_email(email: str, message: str) -> str:
"""Send email"""
print(f"\nEMAIL SENT TO: {email}")
print(f"MESSAGE: {message}")
return "Email sent successfully"
# ── Agent ──────────────────────────────────────────────────────────────────────
agent = create_agent(
model=model,
tools=[send_email],
middleware=[HumanInTheLoopMiddleware(interrupt_on={"send_email": True})],
checkpointer=InMemorySaver(),
)
config = {"configurable": {"thread_id": "test-thread"}}
# ── First call: agent plans and pauses ─────────────────────────────────────────
response = agent.invoke(
{
"messages": [
{
"role": "user",
"content": "Send an email to talha@gmail.com with message 'hello world'",
}
]
},
config=config,
)
print("\nAGENT PAUSED FOR APPROVAL\n")
# ── Human decision ─────────────────────────────────────────────────────────────
decision = input("Approve email? (yes/no): ")
if decision.lower() == "yes":
response = agent.invoke(
Command(resume={"decisions": [{"type": "approve"}]}),
config=config,
)
print("\nFINAL RESPONSE:")
print(response["messages"][-1].content)
else:
response = agent.invoke(
Command(
resume={
"decisions": [{"type": "reject", "message": "Human rejected the email"}]
}
),
config=config,
)
print("\nEMAIL REJECTED")
Sample Output
When approved:
AGENT PAUSED FOR APPROVAL
Approve email? (yes/no): yes
EMAIL SENT TO: talha@gmail.com
MESSAGE: hello world
FINAL RESPONSE:
The email has been sent to talha@gmail.com with the message "hello world".
When rejected:
AGENT PAUSED FOR APPROVAL
Approve email? (yes/no): no
EMAIL REJECTED
Why This Pattern Matters
Human-in-the-loop isn't just a safety net — it's an architectural pattern for building trust between AI and users. Here's when you'd want it:
| Use Case | Why HITL Helps |
|---|---|
| Sending emails / messages | Prevents accidental or hallucinated sends |
| Database writes | Confirms destructive operations |
| API calls with side effects | Gives humans a veto before real-world actions |
| Financial transactions | Compliance and audit requirements |
| Customer-facing responses | Quality control before publishing |
Taking It Further
-
Async approval: Instead of a terminal
input(), send an approval request to Slack, email, or a web dashboard. - Timeout handling: Auto-reject if no human responds within N minutes.
- Audit logs: Persist every approval/rejection to a database for compliance.
-
Multiple tools: Pass multiple tool names to
interrupt_onto gate any subset of your agent's capabilities.
Wrapping Up
LangGraph's checkpointing + LangChain's middleware make it surprisingly clean to add human oversight to any AI agent. The agent does the thinking; you keep the final say.
If you found this useful, drop a ❤️ and let me know what kinds of agents you're building in the comments!
Top comments (0)