Human-in-Loop in Machine Learning

from dataclasses import dataclass
from enum import Enum
from typing import Any, Optional
import logging
import time
import uuid


class ReviewTier(Enum):
    AUTO_APPROVE = "auto_approve"
    SOFT_REVIEW = "soft_review"     # async, non-blocking
    HARD_GATE = "hard_gate"         # sync, blocks execution


@dataclass
class EscalationPolicy:
    """Configurable thresholds per action type."""
    action_type: str
    auto_approve_threshold: float   # above this -> auto approve
    soft_review_threshold: float    # above this but below auto -> soft review
    # below soft_review_threshold -> hard gate
    max_auto_approve_amount: Optional[float] = None  # e.g., INR 10000
    require_audit: bool = True


@dataclass
class ReviewDecision:
    decision_id: str
    action_id: str
    tier: ReviewTier
    confidence: float
    reviewer_id: Optional[str] = None
    approved: Optional[bool] = None
    modified_action: Optional[Any] = None
    review_timestamp: Optional[float] = None
    feedback_notes: Optional[str] = None


class HITLEscalationRouter:
    """Routes model predictions to the appropriate review tier."""

    def __init__(self, policies: list[EscalationPolicy], audit_logger=None):
        self.policies = {p.action_type: p for p in policies}
        self.audit_logger = audit_logger or logging.getLogger("hitl_audit")
        self.review_queue = []  # In production, use Redis or Kafka

    def route(self, action_type: str, confidence: float,
              amount: Optional[float] = None, metadata: dict = None) -> ReviewDecision:
        """Determine the review tier for a given action."""
        policy = self.policies.get(action_type)
        if policy is None:
            # Unknown action type -> hard gate by default (fail safe)
            return self._create_decision(action_type, confidence, ReviewTier.HARD_GATE)

        # Amount-based override: large transactions always need review
        if (policy.max_auto_approve_amount is not None
                and amount is not None
                and amount > policy.max_auto_approve_amount):
            tier = ReviewTier.HARD_GATE
        elif confidence >= policy.auto_approve_threshold:
            tier = ReviewTier.AUTO_APPROVE
        elif confidence >= policy.soft_review_threshold:
            tier = ReviewTier.SOFT_REVIEW
        else:
            tier = ReviewTier.HARD_GATE

        decision = self._create_decision(action_type, confidence, tier)

        # Log to audit trail
        if policy.require_audit:
            self._log_audit(decision, amount, metadata)

        # Enqueue if review needed
        if tier != ReviewTier.AUTO_APPROVE:
            self.review_queue.append(decision)

        return decision

    def _create_decision(self, action_type: str, confidence: float,
                         tier: ReviewTier) -> ReviewDecision:
        return ReviewDecision(
            decision_id=str(uuid.uuid4()),
            action_id=action_type,
            tier=tier,
            confidence=confidence,
        )

    def _log_audit(self, decision: ReviewDecision,
                   amount: Optional[float], metadata: dict):
        self.audit_logger.info(
            f"HITL_ROUTING | id={decision.decision_id} | "
            f"action={decision.action_id} | tier={decision.tier.value} | "
            f"confidence={decision.confidence:.4f} | amount={amount} | "
            f"metadata={metadata}"
        )


# --- Usage Example ---
policies = [
    EscalationPolicy(
        action_type="payment_transfer",
        auto_approve_threshold=0.95,
        soft_review_threshold=0.80,
        max_auto_approve_amount=10000.0,  # INR 10,000
    ),
    EscalationPolicy(
        action_type="content_publish",
        auto_approve_threshold=0.90,
        soft_review_threshold=0.70,
    ),
]

router = HITLEscalationRouter(policies)

# High confidence, small amount -> auto approve
d1 = router.route("payment_transfer", confidence=0.97, amount=5000)
print(d1.tier)  # ReviewTier.AUTO_APPROVE

# High confidence but large amount -> hard gate
d2 = router.route("payment_transfer", confidence=0.97, amount=500000)
print(d2.tier)  # ReviewTier.HARD_GATE

# Low confidence -> hard gate
d3 = router.route("payment_transfer", confidence=0.55, amount=2000)
print(d3.tier)  # ReviewTier.HARD_GATE

from langgraph.graph import StateGraph, START, END
from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import MemorySaver
from typing import TypedDict


class AgentState(TypedDict):
    task: str
    plan: str
    tool_calls: list[dict]
    human_approved: bool
    result: str


def plan_step(state: AgentState) -> AgentState:
    """Agent generates a plan and proposed tool calls."""
    # In production, this calls an LLM to generate the plan
    plan = f"Plan for: {state['task']}"
    tool_calls = [
        {"tool": "database_write", "args": {"table": "orders", "action": "update"}},
        {"tool": "send_email", "args": {"to": "[email protected]"}},
    ]
    return {**state, "plan": plan, "tool_calls": tool_calls}


def human_review_step(state: AgentState) -> AgentState:
    """Pause execution and wait for human approval."""
    # This is where the magic happens -- interrupt() pauses the graph
    # and surfaces the current state to the human reviewer
    review_context = (
        f"Agent proposes the following actions:\n"
        f"Plan: {state['plan']}\n"
        f"Tool calls: {state['tool_calls']}\n"
        f"Please approve (yes/no/modify):"
    )
    human_response = interrupt(review_context)

    if human_response.get("approved"):
        return {**state, "human_approved": True}
    elif human_response.get("modified_calls"):
        return {
            **state,
            "tool_calls": human_response["modified_calls"],
            "human_approved": True,
        }
    else:
        return {**state, "human_approved": False, "result": "Rejected by human reviewer"}


def execute_step(state: AgentState) -> AgentState:
    """Execute the approved actions."""
    if not state.get("human_approved"):
        return state
    # Execute tool calls here
    results = [f"Executed {tc['tool']}" for tc in state["tool_calls"]]
    return {**state, "result": "; ".join(results)}


def should_execute(state: AgentState) -> str:
    """Conditional edge: only execute if human approved."""
    return "execute" if state.get("human_approved") else "end"


# Build the graph with HITL breakpoint
builder = StateGraph(AgentState)
builder.add_node("plan", plan_step)
builder.add_node("human_review", human_review_step)
builder.add_node("execute", execute_step)

builder.add_edge(START, "plan")
builder.add_edge("plan", "human_review")
builder.add_conditional_edges("human_review", should_execute, {
    "execute": "execute",
    "end": END,
})
builder.add_edge("execute", END)

# Compile with checkpointing (required for interrupt)
checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

# Run the agent -- it will pause at human_review_step
config = {"configurable": {"thread_id": "task-001"}}
initial_state = {"task": "Update order status and notify customer",
                 "plan": "", "tool_calls": [],
                 "human_approved": False, "result": ""}

# First invocation: runs until interrupt
result = graph.invoke(initial_state, config)
# Graph is now paused, waiting for human input

# Human reviews and approves
human_input = Command(resume={"approved": True})
final_result = graph.invoke(human_input, config)
print(final_result["result"])
# Output: "Executed database_write; Executed send_email"

import numpy as np
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score
from label_studio_sdk import Client


class ActiveLearningLoop:
    """Implements uncertainty-based active learning with human annotation."""

    def __init__(self, model, ls_url: str, ls_api_key: str, project_id: int):
        self.model = model
        self.ls_client = Client(url=ls_url, api_key=ls_api_key)
        self.project = self.ls_client.get_project(project_id)
        self.labeled_X, self.labeled_y = [], []
        self.iteration = 0

    def compute_uncertainty(self, X_pool: np.ndarray) -> np.ndarray:
        """Compute prediction uncertainty using entropy."""
        proba = self.model.predict_proba(X_pool)
        entropy = -np.sum(proba * np.log(proba + 1e-10), axis=1)
        return entropy

    def select_for_annotation(self, X_pool: np.ndarray,
                               n_samples: int = 50) -> np.ndarray:
        """Select the most uncertain samples for human review."""
        uncertainty = self.compute_uncertainty(X_pool)
        top_indices = np.argsort(uncertainty)[-n_samples:]
        return top_indices

    def send_to_label_studio(self, samples: list[dict]):
        """Push selected samples to Label Studio for annotation."""
        tasks = [{"data": sample} for sample in samples]
        self.project.import_tasks(tasks)
        print(f"Sent {len(tasks)} tasks to Label Studio for annotation")

    def fetch_annotations(self) -> list[dict]:
        """Retrieve completed annotations from Label Studio."""
        tasks = self.project.get_labeled_tasks()
        annotations = []
        for task in tasks:
            if task.get("annotations"):
                latest = task["annotations"][-1]
                annotations.append({
                    "data": task["data"],
                    "label": latest["result"][0]["value"]["choices"][0],
                    "annotator": latest.get("completed_by"),
                })
        return annotations

    def retrain(self, X_new: np.ndarray, y_new: np.ndarray):
        """Retrain model with newly annotated data."""
        self.labeled_X.append(X_new)
        self.labeled_y.append(y_new)
        X_all = np.vstack(self.labeled_X)
        y_all = np.concatenate(self.labeled_y)
        self.model.fit(X_all, y_all)
        self.iteration += 1
        print(f"Retrained model (iteration {self.iteration}) "
              f"on {len(y_all)} total samples")

    def run_iteration(self, X_pool: np.ndarray, n_samples: int = 50):
        """Execute one active learning iteration."""
        # 1. Select most uncertain samples
        indices = self.select_for_annotation(X_pool, n_samples)
        selected = X_pool[indices]

        # 2. Send to humans for annotation
        samples = [{"features": x.tolist()} for x in selected]
        self.send_to_label_studio(samples)

        # 3. Wait for annotations (in production, this is async)
        print(f"Waiting for {n_samples} annotations...")
        # annotations = self.fetch_annotations()  # called after humans complete

        return indices


# --- Usage ---
model = RandomForestClassifier(n_estimators=100)
# model.fit(initial_X, initial_y)  # fit on seed data first

al_loop = ActiveLearningLoop(
    model=model,
    ls_url="http://localhost:8080",
    ls_api_key="your-api-key",
    project_id=1,
)

# Run active learning iteration
# indices = al_loop.run_iteration(X_unlabeled_pool, n_samples=50)

# HITL Escalation Policy Configuration (YAML)
escalation_policies:
  - action_type: payment_transfer
    auto_approve_threshold: 0.95
    soft_review_threshold: 0.80
    max_auto_approve_amount_inr: 10000
    require_two_reviewers_above_inr: 500000
    sla_seconds: 30
    audit: true

  - action_type: content_publish
    auto_approve_threshold: 0.92
    soft_review_threshold: 0.75
    categories_always_review:
      - hate_speech
      - self_harm
      - child_safety
    sla_seconds: 300
    audit: true

  - action_type: agent_tool_call
    auto_approve_threshold: 0.98
    soft_review_threshold: 0.85
    irreversible_actions_always_gate:
      - database_delete
      - send_email
      - api_post
    sla_seconds: 60
    audit: true

review_settings:
  max_queue_depth: 500
  reviewer_session_limit_minutes: 120
  min_reviewers_per_task: 1
  high_stakes_min_reviewers: 2
  inter_rater_agreement_threshold: 0.80

feedback_pipeline:
  batch_size: 256
  retrain_trigger: every_1000_annotations
  min_agreement_for_training: 0.85
  store: redis
  audit_log: immutable_append_only