Why this matters

Production AI systems fail in ways that are hard to detect from standard application metrics. Response latency looks normal, error rates are low, but the system has quietly drifted: safety checks are blocking more than usual, the model is producing lower-quality outputs, or costs are spiking on a specific user segment.

Governance is the operational discipline that catches these drifts before they become incidents. Audit logging gives you the record. Monitoring gives you the signal. Approval gates give you the ability to stop high-risk actions before they execute.

This is not compliance theater — it is the operational infrastructure that lets you run AI features in production with confidence.

Core concepts

What to log for AI systems

Standard application logging covers HTTP status codes and response times. AI systems need more:

from dataclasses import dataclass, field
from datetime import datetime
from typing import Any
import uuid


@dataclass
class AIAuditRecord:
    # Identity
    record_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    session_id: str = ""
    user_id: str = ""

    # Timing
    timestamp: datetime = field(default_factory=datetime.utcnow)
    latency_ms: int = 0

    # Model call
    model: str = ""
    prompt_tokens: int = 0
    completion_tokens: int = 0
    estimated_cost_usd: float = 0.0

    # Safety checks
    input_threat_category: str = "safe"
    input_blocked: bool = False
    output_validation_passed: bool = True
    output_failed_checks: list[str] = field(default_factory=list)
    pii_detected_in_input: bool = False
    canary_leaked: bool = False

    # Feature context
    feature_name: str = ""
    path_version: str = ""  # prompt template version
    retrieval_doc_count: int = 0

    # Outcome
    response_delivered: bool = True
    fallback_used: bool = False
    error: str | None = None

Log every field for every request. This record is your audit trail, your debugging tool, and your safety metric source.

Human-in-the-loop approval gates

For high-risk actions — sending emails on behalf of users, making purchases, modifying database records — require explicit human approval before the AI proceeds:

from enum import Enum


class ActionRiskLevel(str, Enum):
    LOW = "low"        # AI proceeds automatically
    MEDIUM = "medium"  # Log and proceed, flag for async review
    HIGH = "high"      # Require real-time human approval
    CRITICAL = "critical"  # Never execute automatically


ACTION_RISK_REGISTRY = {
    "answer_faq": ActionRiskLevel.LOW,
    "summarize_document": ActionRiskLevel.LOW,
    "draft_email": ActionRiskLevel.MEDIUM,
    "send_email": ActionRiskLevel.HIGH,
    "process_refund": ActionRiskLevel.HIGH,
    "delete_account": ActionRiskLevel.CRITICAL,
    "make_payment": ActionRiskLevel.CRITICAL,
}


class ApprovalGate:
    def __init__(self, approval_callback=None) -> None:
        # approval_callback: async fn(action, context) -> bool
        # In production, this sends to a UI, Slack bot, or review queue
        self.approval_callback = approval_callback

    async def check(self, action_name: str, context: dict[str, Any]) -> bool:
        risk = ACTION_RISK_REGISTRY.get(action_name, ActionRiskLevel.HIGH)

        if risk == ActionRiskLevel.LOW:
            return True
        if risk == ActionRiskLevel.CRITICAL:
            return False  # Never auto-approve critical actions
        if risk == ActionRiskLevel.MEDIUM:
            # Proceed but log for async review
            self._log_for_review(action_name, context)
            return True
        if risk == ActionRiskLevel.HIGH:
            if self.approval_callback:
                return await self.approval_callback(action_name, context)
            return False  # Default: deny if no approval mechanism configured

        return False

    def _log_for_review(self, action_name: str, context: dict) -> None:
        import logging
        logging.getLogger("ai_governance").warning(
            "medium_risk_action_executed",
            extra={"action": action_name, "context_keys": list(context.keys())},
        )

Safety metric monitoring

Five metrics to track continuously:

@dataclass
class SafetyMetrics:
    window_minutes: int
    total_requests: int
    blocked_input_rate: float    # % of inputs blocked by threat classifier
    output_fail_rate: float      # % of outputs failing validation
    pii_detection_rate: float    # % of inputs containing PII
    fallback_rate: float         # % of requests that used fallback response
    approval_gate_denial_rate: float  # % of high-risk actions denied


class SafetyMonitor:
    ALERT_THRESHOLDS = {
        "blocked_input_rate": 0.05,     # Alert if >5% inputs blocked
        "output_fail_rate": 0.02,       # Alert if >2% outputs fail validation
        "pii_detection_rate": 0.10,     # Alert if >10% inputs contain PII
        "fallback_rate": 0.08,          # Alert if >8% responses use fallback
    }

    def check_alerts(self, metrics: SafetyMetrics) -> list[str]:
        alerts = []
        for metric_name, threshold in self.ALERT_THRESHOLDS.items():
            value = getattr(metrics, metric_name, 0.0)
            if value > threshold:
                alerts.append(
                    f"SAFETY ALERT: {metric_name} = {value:.1%} "
                    f"(threshold: {threshold:.1%})"
                )
        return alerts

The GovernanceLayer

import time
import logging

logger = logging.getLogger("ai_governance")


class GovernanceLayer:
    def __init__(
        self,
        feature_name: str,
        approval_gate: ApprovalGate | None = None,
        threat_classifier=None,
        output_validator=None,
        pii_protector=None,
    ) -> None:
        self.feature_name = feature_name
        self.approval_gate = approval_gate or ApprovalGate()
        self.threat_classifier = threat_classifier
        self.output_validator = output_validator
        self.pii_protector = pii_protector

    async def run(
        self,
        user_input: str,
        action_name: str,
        llm_callable,  # async fn(clean_input: str) -> (response_text, model, tokens)
        session_id: str = "",
        user_id: str = "",
    ) -> tuple[str, AIAuditRecord]:
        record = AIAuditRecord(
            session_id=session_id,
            user_id=user_id,
            feature_name=self.feature_name,
        )
        start = time.monotonic()

        # 1. PII protection on input
        clean_input = user_input
        if self.pii_protector:
            pii_result = self.pii_protector.protect(user_input)
            clean_input = pii_result.redacted_text
            record.pii_detected_in_input = len(pii_result.pii_found) > 0

        # 2. Threat classification
        if self.threat_classifier:
            threat = self.threat_classifier(clean_input)
            record.input_threat_category = threat.category
            if threat.category != "safe":
                record.input_blocked = True
                record.latency_ms = int((time.monotonic() - start) * 1000)
                logger.warning("input_blocked", extra={"record_id": record.record_id, "category": threat.category})
                return "I'm not able to help with that request.", record

        # 3. Approval gate
        approved = await self.approval_gate.check(action_name, {"input": clean_input[:100]})
        if not approved:
            record.response_delivered = False
            record.latency_ms = int((time.monotonic() - start) * 1000)
            return "This action requires human approval.", record

        # 4. LLM call
        try:
            response_text, model, tokens = await llm_callable(clean_input)
            record.model = model
            record.prompt_tokens = tokens.get("prompt", 0)
            record.completion_tokens = tokens.get("completion", 0)
        except Exception as exc:
            record.error = str(exc)
            record.fallback_used = True
            record.latency_ms = int((time.monotonic() - start) * 1000)
            logger.error("llm_call_failed", extra={"record_id": record.record_id, "error": str(exc)})
            return "I encountered an error processing your request.", record

        # 5. Output validation
        if self.output_validator:
            val_result = self.output_validator.validate(response_text)
            record.output_validation_passed = val_result.passed
            record.output_failed_checks = val_result.failed_checks
            if not val_result.passed:
                record.fallback_used = True
                logger.warning("output_validation_failed", extra={"record_id": record.record_id, "checks": val_result.failed_checks})
                return "I wasn't able to generate a safe response. Please try rephrasing.", record

        record.latency_ms = int((time.monotonic() - start) * 1000)
        logger.info("request_completed", extra={"record_id": record.record_id, "model": record.model, "latency_ms": record.latency_ms})
        return response_text, record

Common mistakes

Logging inputs and outputs in full without redaction. Operational logs are often stored with weaker access controls than your primary database. PII in logs is PII at risk. Redact before logging.

Alert thresholds that never change. A blocked_input_rate threshold of 5% might be reasonable at launch and entirely wrong six months later when your user base changes. Review thresholds quarterly.

Approval gates without a timeout. If your approval callback waits indefinitely for a human to respond, a high-risk action can keep a request hanging forever. Set a timeout after which the gate denies the action.

No incident playbook. When a safety alert fires at 3 AM, your on-call engineer needs a runbook: what does this metric mean, what caused it last time, what is the rollback procedure. Write it before launch.

Try it yourself

1. Implement the GovernanceLayer class and run it with a mock LLM callable. Verify the audit record is populated correctly for a safe request.
2. Run a second test with an input that triggers the threat classifier. Verify the input is blocked, the record captures the threat category, and the LLM callable is never invoked.
3. Add a SafetyMonitor and generate a SafetyMetrics object from a simulated set of 100 requests where 8 were blocked. Verify that the monitor triggers an alert on the blocked_input_rate.