Enterprise AI Architecture Blueprint Part 3: AI Governance and Automated Regulatory Compliance

from datetime import datetime
from enum import Enum

class ModelStatus(Enum):
    EXPERIMENTAL = "experimental"
    STAGING = "staging"
    PRODUCTION = "production"
    DEPRECATED = "deprecated"
    RETIRED = "retired"

class ModelRegistry:
 """ AI """

    def register(self, model_id: str, version: str, metadata: dict) -> str:
        record = {
            "model_id": model_id,
            "version": version,
            "status": ModelStatus.EXPERIMENTAL.value,
            "registered_at": datetime.utcnow().isoformat(),
            "owner_team": metadata["owner"],
            "use_case": metadata["use_case"],
            "data_classification": metadata["data_class"],
            "approved_by": None,
            "retired_at": None,
        }
        return self.db.insert("model_registry", record)

    def promote(self, model_id: str, to_status: ModelStatus, approver: str):
        if to_status == ModelStatus.PRODUCTION:
            self._require_approval(model_id, approver)
        self.db.update("model_registry",
            {"model_id": model_id},
            {"status": to_status.value, "approved_by": approver}
        )

    def deprecate(self, model_id: str, successor_id: str, sunset_date: str):
        self.db.update("model_registry",
            {"model_id": model_id},
            {
                "status": ModelStatus.DEPRECATED.value,
                "successor_model": successor_id,
                "sunset_date": sunset_date,
            }
        )

class ComplianceAutomator:
 """GDPR/AI Act """

    REQUIRED_CHECKS = [
        "data_minimization",
        "consent_verified",
        "audit_log_enabled",
        "pii_masked",
        "cross_border_approved",
    ]

    async def pre_flight(self, request: AIRequest) -> ComplianceResult:
        results = {}
        for check in self.REQUIRED_CHECKS:
            handler = getattr(self, f"check_{check}")
            results[check] = await handler(request)

        failed = [k for k, v in results.items() if not v]
        if failed:
 raise ComplianceError(f" failed: {', '.join(failed)}")
        return ComplianceResult(passed=True, checks=results)

    async def check_data_minimization(self, req: AIRequest) -> bool:
 # Request
        allowed_fields = self.policy.get_allowed_fields(req.use_case)
        return all(f in allowed_fields for f in req.data_fields)

    async def check_audit_log_enabled(self, req: AIRequest) -> bool:
        return req.audit_session_id is not None

CREATE TABLE ai_audit_log (
    id              BIGSERIAL PRIMARY KEY,
    session_id      UUID NOT NULL,
    user_id         VARCHAR(128),
    model_id        VARCHAR(128) NOT NULL,
    use_case        VARCHAR(256),
    data_class      VARCHAR(64),
    input_tokens    INTEGER,
    output_tokens   INTEGER,
    latency_ms      INTEGER,
    compliance_tags JSONB,
    pii_detected    BOOLEAN DEFAULT FALSE,
    blocked         BOOLEAN DEFAULT FALSE,
    created_at      TIMESTAMPTZ DEFAULT NOW()
);

-- 2 (GDPR )
CREATE POLICY audit_retention ON ai_audit_log
    USING (created_at > NOW() - INTERVAL '2 years');

class AIFinOps:
    PRICE_PER_1K = {
        "claude-haiku-4-5":  {"input": 0.00025, "output": 0.00125},
        "claude-sonnet-4-6": {"input": 0.003,   "output": 0.015},
        "claude-opus-4-7":   {"input": 0.015,   "output": 0.075},
    }

    def calculate_cost(self, model: str, in_tokens: int, out_tokens: int) -> float:
        p = self.PRICE_PER_1K[model]
        return (in_tokens * p["input"] + out_tokens * p["output"]) / 1000

    async def enforce_budget(self, team: str, model: str, est_tokens: int):
        monthly_spend = await self.get_monthly_spend(team)
        budget = await self.get_budget(team)
        est_cost = self.calculate_cost(model, est_tokens, est_tokens)

        if monthly_spend + est_cost > budget * 0.9:
            raise BudgetExceededError(
 f"{team} 90% . : ${monthly_spend:.2f}, : ${budget:.2f}"
            )

Level 1 — Ad hoc
  ∘ Independent adoption by each team, no standards

Level 2 — Repeatable
  ∘ Security guidelines exist; reviews are manual

Level 3 — Defined  ← the goal for most enterprises
  ∘ Central registry, automated compliance checks

Level 4 — Managed
  ∘ Real-time monitoring, KPI dashboards

Level 5 — Optimized
  ∘ Automatic policy updates, AI governing AI