AI Engineering Academy · Lesson

Human-in-the-Loop Escalation

Define escalation triggers that pause the agent and request human guidance when confidence is low, when a destructive action is about to occur, or when retry budget is exhausted.

Human-in-the-Loop Escalation is a free AI Engineering Academy lesson on CoddyKit — lesson 4 of 4. You can read the complete lesson below for free — then practise it hands-on in the browser with a built-in code editor and a 24/7 AI tutor. It is part of the AI Engineering Academy learning path, one of 4 lessons in the course, and your progress syncs across the web and the CoddyKit app.

When Agents Need Human Guidance

Fully autonomous agents are appropriate for well-defined, low-risk tasks. But some situations require human judgment: ambiguous instructions, low model confidence, destructive actions that cannot be undone, or tasks where failure has serious consequences. Human-in-the-loop (HITL) escalation pauses the agent at these decision points and requests human input before proceeding, combining the efficiency of automation with the judgment of humans.

Defining Escalation Triggers

Escalation should be triggered by specific, measurable conditions rather than vague intuition. Define explicit escalation triggers for your application. Common triggers include: model confidence below a threshold, a destructive action about to be taken, a policy boundary being approached, the retry budget exhausted, or the task exceeding its time limit. Document triggers in code as named constants so they can be tuned without changing control flow logic.

from enum import Enum

class EscalationReason(Enum):
    LOW_CONFIDENCE = 'low_confidence'           # model uncertainty
    DESTRUCTIVE_ACTION = 'destructive_action'   # irreversible change
    AMBIGUOUS_TASK = 'ambiguous_task'           # unclear instructions
    RETRY_BUDGET_EXHAUSTED = 'retry_exhausted'  # too many failures
    POLICY_BOUNDARY = 'policy_boundary'         # approaching limit
    HUMAN_REQUESTED = 'human_requested'         # explicit request
    TIMEOUT = 'timeout'                         # took too long

ESCALATION_THRESHOLDS = {
    'min_confidence': 0.6,
    'max_retries': 5,
    'max_runtime_minutes': 30,
}

Detecting Low Confidence

Ask the model to express its confidence in a proposed action before executing it. A confidence score below your threshold triggers escalation. Use a structured confidence check with both a numeric score and a brief rationale so the human reviewer understands exactly why the agent was uncertain. The rationale helps the human provide targeted guidance rather than having to review the entire task history.

from pydantic import BaseModel

class ConfidenceCheck(BaseModel):
    proposed_action: str
    confidence: float  # 0.0 to 1.0
    uncertainty_reason: str | None
    proceed: bool

async def check_confidence(context: str, proposed_action: str) -> ConfidenceCheck:
    return await judge_client.chat.completions.create(
        model='gpt-4o',
        response_model=ConfidenceCheck,
        messages=[{
            'role': 'user',
            'content': f'Context: {context}\n\nI am about to: {proposed_action}\n\nHow confident am I that this is correct? Be honest about uncertainty.'
        }]
    )

Detecting Destructive Actions

Tag tools that perform irreversible actions with a destructive=True flag and require human confirmation before executing them. Examples: deleting files, sending emails to real users, making database changes that cannot be rolled back, charging a customer, or publishing content publicly. The agent must pause at these actions and wait for explicit human approval even if it is otherwise operating autonomously.

from dataclasses import dataclass
from typing import Callable

@dataclass
class Tool:
    name: str
    func: Callable
    destructive: bool = False
    description: str = ''

tools = [
    Tool('search_web',     search_web,      destructive=False),
    Tool('read_file',      read_file,       destructive=False),
    Tool('write_file',     write_file,      destructive=True,  description='Overwrites existing file'),
    Tool('send_email',     send_email,      destructive=True,  description='Sends real email to user'),
    Tool('delete_records', delete_records,  destructive=True,  description='Permanent DB deletion'),
]

def requires_approval(tool: Tool) -> bool:
    return tool.destructive

Pausing the Agent and Awaiting Input

When an escalation trigger fires, save the checkpoint (so the task can resume), create an escalation request record, and notify the human reviewer. The agent stops processing and waits. The human reviews the escalation via a dashboard or notification, provides guidance or approval, and the agent resumes from the checkpoint with that guidance included as a new message in the history.

import asyncio

async def escalate_and_wait(task_id: str, reason: EscalationReason, context: str,
                             question: str, timeout_hours: int = 24) -> str:
    # Save checkpoint
    save_checkpoint(load_checkpoint(task_id))
    # Create escalation record
    escalation_id = create_escalation(task_id, reason, context, question)
    # Notify reviewer
    notify_reviewer(escalation_id, question)
    # Wait for response (polling with timeout)
    deadline = asyncio.get_event_loop().time() + timeout_hours * 3600
    while asyncio.get_event_loop().time() < deadline:
        response = get_escalation_response(escalation_id)
        if response:
            return response.guidance
        await asyncio.sleep(60)  # check every minute
    raise TimeoutError(f'Escalation {escalation_id} not answered within {timeout_hours}h')

Building the Reviewer Interface

Human reviewers need a simple interface to respond to escalations. At minimum, show: the task description, the agent's progress so far, the specific question or proposed action requiring approval, and buttons for Approve, Reject, and Provide Guidance. Log every reviewer decision with the reviewer's identity and timestamp for audit purposes. A Slack bot or a simple web form both work well for internal teams.

# FastAPI escalation endpoint
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class EscalationResponse(BaseModel):
    escalation_id: str
    decision: str  # 'approve', 'reject', 'guide'
    guidance: str | None = None
    reviewer_id: str

@app.post('/escalations/{escalation_id}/respond')
async def respond_to_escalation(esc_id: str, response: EscalationResponse):
    escalation = get_escalation(esc_id)
    if not escalation or escalation.status != 'pending':
        return {'error': 'Escalation not found or already resolved'}
    save_escalation_response(esc_id, response)
    return {'status': 'response_recorded', 'task_will_resume': True}

Injecting Human Guidance into Agent Context

After the human responds, inject their guidance as a new message in the agent's conversation history before resuming. Frame it as coming from a 'supervisor' to distinguish it from the agent's own observations. The agent can then reference this guidance in its next step. If the human rejected the proposed action, include instructions on what to do instead.

def inject_human_guidance(messages: list, decision: str, guidance: str | None) -> list:
    if decision == 'approve':
        messages.append({
            'role': 'user',
            'content': 'Supervisor: Your proposed action has been approved. Proceed.'
        })
    elif decision == 'reject':
        messages.append({
            'role': 'user',
            'content': f'Supervisor: Your proposed action was rejected. Instead: {guidance}'
        })
    elif decision == 'guide':
        messages.append({
            'role': 'user',
            'content': f'Supervisor: Additional guidance: {guidance}'
        })
    return messages

Tracking Escalation Metrics

Monitor escalation volume, reasons, and response times. High escalation volume indicates the agent is not confident enough — either the task is too ambiguous, the model needs better instructions, or confidence thresholds are set too low. Long response times indicate reviewer workload problems. These metrics help you tune the automation-human balance to minimize unnecessary interruptions while keeping humans in the loop for genuinely risky decisions.

def escalation_report(db_connection, days: int = 7) -> dict:
    # SQL query (pseudocode)
    rows = db_connection.execute('''
        SELECT
            reason,
            COUNT(*) as count,
            AVG(EXTRACT(EPOCH FROM (responded_at - created_at)) / 3600) as avg_response_hours,
            SUM(CASE WHEN decision = 'approve' THEN 1 ELSE 0 END) as approvals,
            SUM(CASE WHEN decision = 'reject' THEN 1 ELSE 0 END) as rejections
        FROM escalations
        WHERE created_at > NOW() - INTERVAL '%s days'
        GROUP BY reason
        ORDER BY count DESC
    ''' % days).fetchall()
    return [dict(r) for r in rows]

Gradual Autonomy Expansion

Start with high escalation sensitivity (low confidence threshold, escalate for all destructive actions) and gradually reduce escalation frequency as you gain confidence in the agent's behavior. Track which escalations result in Approve decisions versus actual corrections. A consistently high approval rate for a specific trigger type means you can safely automate that trigger, reducing human workload while maintaining oversight where it truly matters.

# Autonomy expansion strategy:
# Week 1: escalate for ALL destructive actions
# Week 2: auto-approve file writes to /tmp (low-risk), escalate others
# Week 4: auto-approve all file writes, escalate only email/DB changes
# Week 8: auto-approve emails under 10 recipients, escalate mass emails

# Track approval rates per trigger type:
# Tool: write_file    -> 98% approve -> safe to automate
# Tool: send_email    -> 89% approve -> near-automate with content check
# Tool: delete_records -> 43% approve -> always escalate

Emergency Override and Task Cancellation

Always provide an emergency override mechanism that allows a human to immediately cancel a running agent task. If an agent is misbehaving — calling tools it should not, or taking actions outside its intended scope — a human must be able to stop it within seconds. Implement a cancel signal (a database flag the agent checks at each step) and ensure tool call results are discarded if the agent is cancelled mid-step.

async def run_agent_with_cancel(task_id: str, messages: list) -> str:
    for step in range(MAX_ITERATIONS):
        # Check cancel flag at start of every step
        if redis_client.get(f'agent:cancel:{task_id}'):
            save_final_status(task_id, 'cancelled')
            return 'Task cancelled by operator.'
        response = await get_next_action(messages)
        if response.is_final:
            return response.answer
        result = await execute_tool(response.tool, response.args)
        messages.append({'role': 'user', 'content': result})
        save_checkpoint_after_step(task_id, step, messages)
    return 'Max iterations reached'

Calibrating Escalation Thresholds

Escalation thresholds need tuning. If confidence threshold is too high, the agent escalates on nearly every action, overwhelming reviewers. If it is too low, risky actions slip through. Start with conservative thresholds in week 1, track escalation volume and reviewer approval rate, and adjust. A stable system should escalate 5-15% of tasks for confidence issues and near-100% for destructive actions, with an overall approval rate above 80%.

# Threshold tuning guide:
# Escalation rate vs quality trade-off:
#
# confidence_threshold=0.8  -> 35% escalation rate (too many)
# confidence_threshold=0.6  -> 12% escalation rate (target)
# confidence_threshold=0.4  ->  4% escalation rate (too few)
#
# Weekly review of escalation decisions:
# - Approval rate > 90%: lower threshold (too conservative)
# - Approval rate < 70%: raise threshold (not catching real issues)
# - Target: 75-85% approval rate

Quick Check

Test your understanding of human-in-the-loop escalation design.

Lesson Recap

In this lesson you learned: escalation triggers define precise conditions under which the agent must pause and seek human guidance, destructive action flags on tools enforce approval requirements for irreversible operations, and gradual autonomy expansion lets you safely increase automation as the agent earns trust. Next up we design the production architecture for our capstone project.

Frequently asked questions

Is the “Human-in-the-Loop Escalation” lesson free?

Yes — the full text of “Human-in-the-Loop Escalation” is free to read here on the web, and the AI Engineering Academy course includes 4 lessons in total. To practise it interactively (a built-in code editor and a 24/7 AI tutor) and unlock the rest of the AI Engineering Academy course, upgrade to CoddyKit PRO.

What will I learn in “Human-in-the-Loop Escalation”?

Define escalation triggers that pause the agent and request human guidance when confidence is low, when a destructive action is about to occur, or when retry budget is exhausted. You practise AI Engineering Academy with hands-on code you run directly in the browser, and a 24/7 AI tutor answers your questions as you work through the lesson.

Do I need any experience to start AI Engineering Academy?

No prior experience is required. AI Engineering Academy on CoddyKit is structured for beginners through advanced learners; this is — lesson 4 of 4, so you can start here or from the beginning and move at your own pace.

How long does the “Human-in-the-Loop Escalation” lesson take?

Most CoddyKit lessons take about 5–10 minutes. Each one is bite-sized and interactive, so you make steady progress and pick up exactly where you left off across the web and the app.

Can I write and run code in this AI Engineering Academy lesson?

Yes. Every AI Engineering Academy lesson includes a built-in code editor, so you write and run real code right in your browser and get instant AI feedback — no local setup required.

All lessons in this course

Classifying Agent Failure Modes
Self-Correction and Reflective Prompting
Checkpointing and Task Resumption
Human-in-the-Loop Escalation

← Back to AI Engineering Academy