Mastering Agentic AI: Human-in-the-Loop

In my previous post, we covered Exception Handling and Recovery — how an agent detects that something went wrong and tries to dig itself out. That pattern keeps the agent autonomous: it fails, it retries, it falls back, all on its own.

But some decisions shouldn’t be the agent’s to make alone. Issuing a refund. Sending an email to a key account. Deleting production data. Escalating an angry enterprise customer to a senior specialist. For those, “the model decided” is not an acceptable audit trail.

That’s where the Human-in-the-Loop pattern comes in. It’s the difference between an agent you demo and an agent you deploy.

Pattern #13: Human-in-the-Loop
#

The Problem
#

Full autonomy is a great pitch and a terrible default. The moment an agent can take consequential actions — moving money, touching customer-facing systems, making irreversible changes — pure autonomy stops being a feature and becomes a liability.

The naive fix is to gate everything behind a human. But that just turns your agent into an expensive form with extra steps. Nobody wants to approve a password reset.

What you actually need is selective, stateful interruption: the agent runs autonomously for the 90% of cases that are routine, and pauses for human judgment only on the 10% that are sensitive, irreversible, or low-confidence. And critically, when it pauses, it has to stay paused — preserving its full state — until a human responds minutes, hours, or days later, then resume exactly where it stopped.

That last requirement is what makes this hard. A simple input() call blocks a thread; it doesn’t survive a process restart, it doesn’t scale to thousands of concurrent conversations, and it has no memory. You need durable state and a real resumption mechanism.

The Solution
#

The cleanest implementation models the workflow as a graph where one node is a deliberate pause point, backed by a checkpointer that persists state across the interruption. Here’s the support-agent example from the chapter.

State carries everything the agent and the human need to make a decision:

class SupportState(TypedDict):
    customer_name: str
    customer_tier: str
    issue: str
    diagnosis: str
    needs_escalation: bool
    human_approved: bool
    resolution: str

The first node diagnoses the issue and decides whether a human is needed — the model itself flags the cases that exceed its mandate:

def personalize_and_diagnose(state: SupportState) -> dict:
    """Troubleshoots the issue using customer context from state."""
    llm = get_llm(temperature=0)
    prompt = ChatPromptTemplate.from_messages([
        ("system",
         "You are a support agent. The customer context is:\n"
         "- Name: {customer_name}\n"
         "- Tier: {customer_tier}\n\n"
         "Diagnose the issue and determine if it needs human escalation.\n"
         "If the issue is complex, sensitive, or the customer is frustrated, "
         "recommend escalation.\n\n"
         "Output format:\n"
         "DIAGNOSIS: <your diagnosis>\n"
         "ESCALATE: <yes or no>"),
        ("user", "Customer issue: {issue}")
    ])
    chain = prompt | llm | StrOutputParser()
    result = chain.invoke({
        "customer_name": state["customer_name"],
        "customer_tier": state["customer_tier"],
        "issue": state["issue"],
    })

    needs_escalation = "escalate: yes" in result.lower()
    diagnosis = result.split("DIAGNOSIS:")[-1].split("ESCALATE:")[0].strip()
    return {"diagnosis": diagnosis, "needs_escalation": needs_escalation}

A conditional edge routes on that flag — autonomous resolution for the routine path, human approval for the sensitive one:

def route_after_diagnosis(state: SupportState) -> Literal["request_human_approval", "resolve_directly"]:
    if state["needs_escalation"]:
        return "request_human_approval"
    return "resolve_directly"

The pause itself lives in a dedicated node. In the demo it auto-approves, but the comment in the script is the whole point — this is exactly where you wire in the real human gate:

def request_human_approval(state: SupportState) -> dict:
    """Pauses execution for human review before escalation."""
    print(f"  AWAITING HUMAN APPROVAL for escalation of: {state['customer_name']}")
    print(f"  Diagnosis: {state['diagnosis'][:200]}")
    # In production, this would use LangGraph's interrupt() for real human input.
    # For demo purposes, we auto-approve.
    return {"human_approved": True}

The graph wires the diagnosis node to the conditional router, sends the routine path straight to END, and routes the sensitive path through approval and on to escalation. The detail that makes it production-grade is the checkpointer:

def build_support_graph():
    builder = StateGraph(SupportState)
    builder.add_node("personalize_and_diagnose", personalize_and_diagnose)
    builder.add_node("request_human_approval", request_human_approval)
    builder.add_node("resolve_directly", resolve_directly)
    builder.add_node("escalate_to_human", escalate_to_human)

    builder.add_edge(START, "personalize_and_diagnose")
    builder.add_conditional_edges(
        "personalize_and_diagnose", route_after_diagnosis,
        {"request_human_approval": "request_human_approval",
         "resolve_directly": "resolve_directly"}
    )
    builder.add_edge("request_human_approval", "escalate_to_human")
    builder.add_edge("resolve_directly", END)
    builder.add_edge("escalate_to_human", END)

    memory = MemorySaver()
    return builder.compile(checkpointer=memory)

MemorySaver() is the in-memory checkpointer — fine for a demo, swap it for SqliteSaver or PostgresSaver in production. And the thread_id is what ties it together:

result = graph.invoke({
    "customer_name": "Bob",
    "customer_tier": "Enterprise",
    "issue": "I've been charged incorrectly for 3 months and I'm extremely frustrated...",
    "needs_escalation": False,
    "human_approved": False,
}, {"configurable": {"thread_id": "case-2"}})

Here’s the mental model that matters. In real production code, request_human_approval would call LangGraph’s interrupt(). That call doesn’t block a thread — it stops the graph and persists the entire state to the checkpointer under that thread_id. Your process can die. Hours can pass. When the human finally clicks “approve” in some dashboard, you re-invoke the graph with the same thread_id and a Command(resume=...), and LangGraph rehydrates the state and continues from the exact node where it paused. The interruption is durable, not a blocked socket.

That single property — pause, persist, resume by ID — is what separates a real human-in-the-loop system from a script that happens to call input().

Why This Matters
#

This pattern is the gatekeeper for putting agents anywhere near real consequences:

Approval gates: refunds, contract changes, outbound comms to key accounts — the agent drafts, a human signs off.
Tiered escalation: route the frustrated Enterprise customer to a senior specialist; resolve the Standard password reset autonomously. Same graph, different paths.
Editing, not just approving: the human can correct the agent’s state mid-run (fix a diagnosis, adjust a plan) and let it continue from the corrected state.
Low-confidence fallback: when the model’s own confidence drops below a threshold, interrupt and ask rather than guess.
Compliance and audit: every pause is a logged decision point with a named human attached. That’s the audit trail “the model decided” never gives you.

The costs are real and worth naming. Latency: a paused workflow is only as fast as the human, which can mean hours. Your architecture has to be asynchronous and event-driven — no synchronous request can sit waiting on an approval. State management: durable checkpointing is non-negotiable, and persisted conversation state has privacy and retention implications you now own. UX overhead: someone has to build the dashboard where humans see pending interrupts and respond. Over-gating: gate too much and you’ve rebuilt the bottleneck you were trying to automate away.

Rule of thumb: interrupt on consequence and on uncertainty, not on principle. Reserve the human for decisions that are irreversible, sensitive, or genuinely beyond the model’s competence. Everything else should flow through autonomously — that’s the whole reason you built an agent.

The Bigger Picture
#

This is post #13 in my series documenting Antonio Gulli’s Agentic Design Patterns. All credit for the conceptual work goes to him — I’m focused on producing clean, runnable Python implementations that you can clone, modify, and ship.

Human-in-the-Loop is where several earlier patterns converge. The checkpointer is the same durability primitive that underpins Memory (Chapter 8). The conditional escalation edge is Routing (Chapter 2) applied to a person instead of a model. And the interrupt is, in a sense, the most honest form of Exception Handling (Chapter 12) — when recovery genuinely requires judgment the agent doesn’t have, the right move is to ask.

All the code from this post is in my repository: carlosprados/Agentic_Design_Patterns, specifically under 13_Human_in_the_Loop/. The example runs with uv run, supports Gemini and Ollama via the shared get_llm() abstraction, and is ready to fork.

What’s Next
#

In the next post we’ll tackle Knowledge Retrieval (RAG) — giving agents access to information they were never trained on by retrieving relevant documents at query time and grounding their answers in real, citable sources. It’s the pattern that turns a confident-sounding model into one you can actually trust on facts.

Stay tuned.

Pattern #13: Human-in-the-Loop#

The Problem#

The Solution#

Why This Matters#

The Bigger Picture#

What’s Next#