Human-in-the-Loop AI — Oversight, Control & Accountability

from enum import Enum
from dataclasses import dataclass
from typing import Optional
import datetime

class ReviewOutcome(Enum):
    APPROVED     = "approved"
    REJECTED     = "rejected"
    NEEDS_REVIEW = "needs_human_review"

@dataclass
class ContentReview:
    content_id:   str
    content:      str
    ai_score:     float       # 0.0 = safe, 1.0 = harmful
    ai_label:     str
    confidence:   float       # How confident is the AI?
    outcome:      ReviewOutcome
    human_review: Optional[str] = None
    reviewer_id:  Optional[str] = None
    reviewed_at:  Optional[datetime.datetime] = None

class HumanInLoopModerationSystem:
    """
    HITL content moderation:
    - High confidence safe/harmful: AI decides automatically
    - Low confidence or borderline: routes to human review queue
    - Appeals: always go to human
    """

    def __init__(self,
                 auto_approve_threshold: float = 0.1,   # AI score below this = auto-approve
                 auto_reject_threshold: float  = 0.9,   # AI score above this = auto-reject
                 min_confidence: float         = 0.85): # Route to human if confidence < this
        self.auto_approve_threshold = auto_approve_threshold
        self.auto_reject_threshold  = auto_reject_threshold
        self.min_confidence         = min_confidence
        self.human_review_queue     = []
        self.audit_log              = []

    def moderate(self, content_id: str, content: str,
                 ai_score: float, confidence: float) -> ContentReview:
        ai_label = "harmful" if ai_score > 0.5 else "safe"

        # High confidence clear cases → automatic decision
        if confidence >= self.min_confidence:
            if ai_score <= self.auto_approve_threshold:
                outcome = ReviewOutcome.APPROVED
            elif ai_score >= self.auto_reject_threshold:
                outcome = ReviewOutcome.REJECTED
            else:
                outcome = ReviewOutcome.NEEDS_REVIEW  # Borderline even with confidence
        else:
            outcome = ReviewOutcome.NEEDS_REVIEW  # Low confidence → always human

        review = ContentReview(
            content_id=content_id, content=content,
            ai_score=ai_score, ai_label=ai_label,
            confidence=confidence, outcome=outcome
        )

        if outcome == ReviewOutcome.NEEDS_REVIEW:
            self.human_review_queue.append(review)
            print(f"  → Queued for human review (score={ai_score:.2f}, conf={confidence:.2f})")
        else:
            print(f"  → Auto-{outcome.value} (score={ai_score:.2f}, conf={confidence:.2f})")

        self.audit_log.append(review)    # Every decision is logged for accountability
        return review

    def human_review(self, content_id: str, decision: str, reviewer_id: str):
        """Human reviewer processes a queued item."""
        for review in self.human_review_queue:
            if review.content_id == content_id:
                review.outcome     = ReviewOutcome(decision)
                review.human_review = decision
                review.reviewer_id = reviewer_id
                review.reviewed_at = datetime.datetime.now()
                self.human_review_queue.remove(review)
                print(f"Human decision for {content_id}: {decision} by {reviewer_id}")
                return review
        print(f"Content {content_id} not found in queue")

# Test the system
system = HumanInLoopModerationSystem()
print("Content Moderation Results:")
system.moderate("post_001", "Have a great day!", ai_score=0.02, confidence=0.99)
system.moderate("post_002", "I will hurt you",   ai_score=0.95, confidence=0.98)
system.moderate("post_003", "Politics is ...",   ai_score=0.55, confidence=0.72)
system.moderate("post_004", "Borderline content", ai_score=0.6, confidence=0.90)

print(f"
Human review queue: {len(system.human_review_queue)} items")
print(f"Total audit log:    {len(system.audit_log)} decisions")