human-in-the-loop-aiDesign AI systems with human oversight checkpoints, approval workflows, and escalation mechanisms. Covers when to require human intervention, how to build approval queues, confidence thresholds, and compliance-aware AI decision making.
Independent GitHub snapshot · Automated classification · Original documentation governs
Human-in-the-Loop (HITL) AI systems incorporate human judgment at critical decision points within automated workflows. As AI agents gain the ability to take real-world actions — sending emails, processing payments, modifying records — HITL patterns become essential for safety, compliance, and trust.
┌─────────────────────────────────────────────────┐
│ AI Action │
├─────────────────────────────────────────────────┤
│ │
│ Low Risk + High Confidence ──► Auto-execute │
│ Low Risk + Low Confidence ──► Auto + Log │
│ High Risk + High Confidence──► Request Approval │
│ High Risk + Low Confidence ──► Escalate + Block │
│ │
└─────────────────────────────────────────────────┘
| Action Type | Risk Level | HITL Required? | Example |
|---|---|---|---|
| Read data | None | No | Query database, fetch API |
| Format/transform | None | No | Generate report, summarize |
| Send notification | Low | Optional | Slack message, email alert |
| Modify record | Medium | Recommended | Update CRM, edit document |
| Financial transaction | High | Required | Process payment, refund |
| External comms (customer) | High | Required | Send client email |
| Delete/destroy | Critical | Required | Delete records, cancel subscription |
| Access escalation | Critical | Required | Grant permissions |
User Request
│
▼
┌──────────┐ ┌──────────────┐ ┌──────────┐
│ AI Agent │────►│ Risk Assessor│────►│ Executor │
│ (Plans) │ │ (Classifies) │ │ (Acts) │
└──────────┘ └──────┬───────┘ └──────────┘
│ ▲
High Risk? │
│Yes │
▼ │
┌─────────────┐ ┌──────┴──────┐
│ Approval │────►│ Decision │
│ Queue │ │ (Approve/ │
│ │ │ Reject) │
└─────────────┘ └─────────────┘
class ConfidenceRouter:
"""Route AI decisions based on confidence and risk."""
THRESHOLDS = {
"auto_execute": 0.95, # High confidence, auto-proceed
"soft_review": 0.80, # Execute but flag for review
"approval_required": 0.60, # Queue for human approval
"block": 0.0, # Below 0.60, block entirely
}
def route(self, action, confidence, risk_level):
if risk_level == "critical":
return "approval_required" # Always require approval
if confidence >= self.THRESHOLDS["auto_execute"]:
return "auto_execute"
elif confidence >= self.THRESHOLDS["soft_review"]:
return "soft_review" if risk_level == "low" else "approval_required"
elif confidence >= self.THRESHOLDS["approval_required"]:
return "approval_required"
else:
return "block"
interface ApprovalRequest {
id: string;
agentId: string;
action: string;
details: Record<string, unknown>;
riskLevel: "low" | "medium" | "high" | "critical";
confidence: number;
createdAt: string;
expiresAt: string;
status: "pending" | "approved" | "rejected" | "expired";
}
class ApprovalQueue {
async submit(request: Omit<ApprovalRequest, "id" | "status">): Promise<ApprovalRequest> {
const approval: ApprovalRequest = {
...request,
id: crypto.randomUUID(),
status: "pending",
};
await this.store.save(approval);
await this.notify(approval); // Slack, email, dashboard
return approval;
}
async waitForDecision(id: string, timeoutMs = 300000): Promise<Decision> {
const deadline = Date.now() + timeoutMs;
while (Date.now() < deadline) {
const approval = await this.store.get(id);
if (approval.status !== "pending") {
return {
approved: approval.status === "approved",
approver: approval.reviewedBy,
reason: approval.reviewNote,
};
}
await sleep(1000);
}
// Auto-expire if no response
await this.store.update(id, { status: "expired" });
return { approved: false, reason: "Approval request expired" };
}
}
class HITLFeedbackLoop:
"""Capture human decisions to improve future AI confidence."""
def record_decision(self, request_id, ai_recommendation, human_decision):
self.feedback_store.append({
"request_id": request_id,
"ai_said": ai_recommendation,
"human_said": human_decision,
"agreed": ai_recommendation == human_decision,
"timestamp": datetime.utcnow().isoformat(),
})
def get_accuracy_stats(self, action_type, window_days=30):
"""Track how often AI recommendations match human decisions."""
recent = self.feedback_store.query(
action_type=action_type,
since=datetime.utcnow() - timedelta(days=window_days),
)
agreed = sum(1 for r in recent if r["agreed"])
return {
"total_reviews": len(recent),
"agreement_rate": agreed / len(recent) if recent else 0,
"can_auto_approve": (agreed / len(recent)) > 0.95 if len(recent) > 50 else False,
}
| Version | Date | Changes |
|---|---|---|
| 1.0.0 | 2026-03-31 | Initial documentation |