AI Governance in Practice: FastAPI on EKS with Model Cards, Audit Logging, and Helm

Client
  └── POST /predict → FastAPI (EKS pod)
                           │
                           ├── LogisticRegression inference
                           ├── Audit log entry (request_id, features, result)
                           └── Response: churn_probability, prediction, request_id

Kubernetes (EKS)
  ├── Helm chart  →  Deployment + Service + HPA (2–10 replicas)
  └── Terraform   →  VPC, EKS cluster, node groups, IAM OIDC

Governance Layer
  ├── GET /governance/model-card   → version, metrics, fairness, EU AI Act tier
  └── GET /governance/audit-log   → full prediction audit trail

class PredictRequest(BaseModel):
    customer_id: str
    tenure_months: float
    monthly_charges: float
    total_charges: float
    num_complaints: int

@app.post("/predict", response_model=PredictResponse)
def predict(request: PredictRequest) -> PredictResponse:
    features = np.array([[
        request.tenure_months,
        request.monthly_charges,
        request.total_charges,
        request.num_complaints,
    ]])
    scaled = _scaler.transform(features)
    churn_prob = float(_model.predict_proba(scaled)[0][1])
    prediction = "churn" if churn_prob >= 0.5 else "retain"
    request_id = str(uuid.uuid4())

    audit_log.append({
        "request_id": request_id,
        "customer_id": request.customer_id,
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "features": request.model_dump(exclude={"customer_id"}),
        "prediction": prediction,
        "churn_probability": churn_prob,
        "model_version": MODEL_VERSION,
    })

    return PredictResponse(
        customer_id=request.customer_id,
        churn_probability=round(churn_prob, 4),
        prediction=prediction,
        model_version=MODEL_VERSION,
        request_id=request_id,
        timestamp=datetime.now(timezone.utc).isoformat(),
    )

MODEL_CARD = {
    "model_id": "telecom-churn-v1",
    "version": "1.0.0",
    "type": "LogisticRegression",
    "training_date": "2026-04-29",
    "features": ["tenure_months", "monthly_charges", "total_charges", "num_complaints"],
    "metrics": {
        "accuracy": 0.89,
        "roc_auc": 0.92,
        "precision": 0.85,
        "recall": 0.81,
    },
    "fairness": {
        "demographic_parity_difference": 0.03,
        "equalized_odds_difference": 0.02,
        "evaluation_date": "2026-04-29",
    },
    "governance_tier": "Medium Risk",
    "eu_ai_act_classification": "Limited Risk",
    "approved_by": "AI Governance Board",
    "next_review": "2026-10-29",
}

@app.get("/governance/audit-log")
def get_audit_log() -> dict:
    return {"total": len(audit_log), "entries": list(audit_log)}

@app.get("/healthz")
def healthz():
    return {"status": "ok", "model_version": MODEL_VERSION}

@app.get("/ready")
def ready():
    return {"status": "ready", "model_id": MODEL_ID}

_rng = np.random.default_rng(42)
_X_train = np.column_stack([
    _rng.uniform(1, 72, 2000),     # tenure_months
    _rng.uniform(20, 150, 2000),   # monthly_charges
    _rng.uniform(20, 10000, 2000), # total_charges
    _rng.integers(0, 8, 2000),     # num_complaints
])
# Rule: ≥3 complaints, or new customer + high charge → churn
_y_train = (
    (_X_train[:, 3] >= 3) |
    ((_X_train[:, 0] < 6) & (_X_train[:, 1] > 80))
).astype(int)

_scaler = StandardScaler()
_model = LogisticRegression(max_iter=200, random_state=42)
_model.fit(_scaler.fit_transform(_X_train), _y_train)

tests/test_app.py::TestHealthEndpoints        4 passed
  - /healthz returns 200 with status "ok"
  - /ready returns 200 with status "ready"

tests/test_app.py::TestPredictEndpoint        9 passed
  - 200 on valid payload
  - customer_id, churn_probability (0–1), prediction, model_version, request_id present
  - 422 on missing field
  - high-risk profile (1 month tenure, 5 complaints) → churn
  - low-risk profile (60 months, 0 complaints) → retain

tests/test_app.py::TestAuditLog               5 passed
  - empty initially
  - records entry after prediction
  - entry contains customer_id and prediction
  - total count increments correctly

tests/test_app.py::TestModelCard              5 passed
  - 200 response
  - version, metrics, fairness, governance_tier present

FROM python:3.12-slim AS builder
WORKDIR /app
RUN apt-get update && apt-get install -y --no-install-recommends gcc libpq-dev
COPY requirements.txt .
RUN pip install --no-cache-dir --prefix=/install -r requirements.txt

FROM python:3.12-slim AS production
RUN groupadd -r appuser && useradd -r -g appuser -d /app -s /sbin/nologin appuser
WORKDIR /app
COPY --from=builder /install /usr/local
COPY . .
RUN chown -R appuser:appuser /app
USER appuser
EXPOSE 8080
HEALTHCHECK CMD curl -f http://localhost:8080/healthz || exit 1
CMD ["python", "-m", "uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8080"]

docker build --platform linux/amd64 -t ai-governance-platform:local .
# Successfully built d1a906f056b5 — 146MB

# helm/templates/hpa.yaml
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
spec:
  minReplicas: {{ .Values.autoscaling.minReplicas }}   # 2
  maxReplicas: {{ .Values.autoscaling.maxReplicas }}   # 10
  metrics:
    - type: Resource
      resource:
        name: cpu
        target:
          averageUtilization: {{ .Values.autoscaling.targetCPUUtilizationPercentage }}   # 70
    - type: Resource
      resource:
        name: memory
        target:
          averageUtilization: {{ .Values.autoscaling.targetMemoryUtilizationPercentage }} # 80

{{- define "ai-platform.fullname" -}}
{{- if .Values.fullnameOverride }}
{{- .Values.fullnameOverride | trunc 63 | trimSuffix "-" }}
{{- else }}
{{- printf "%s-%s" .Release.Name .Chart.Name | trunc 63 | trimSuffix "-" }}
{{- end }}
{{- end }}

helm lint ./helm/
# ==> Linting ./helm/
# 1 chart(s) linted, 0 chart(s) failed

helm install ai-platform ./helm -f helm/values.yaml
kubectl get pods    # 2–10 replicas depending on load
kubectl get hpa     # watch scaling events

module "eks" {
  source          = "terraform-aws-modules/eks/aws"
  version         = "~> 20.0"
  cluster_name    = "ai-governance-cluster"
  cluster_version = "1.30"
  vpc_id          = module.vpc.vpc_id
  subnet_ids      = module.vpc.private_subnets

  eks_managed_node_groups = {
    general = {
      instance_types = ["m5.large"]
      min_size       = 2
      max_size       = 6
      desired_size   = 3
    }
  }

  enable_irsa = true
}

module "vpc" {
  source  = "terraform-aws-modules/vpc/aws"
  version = "~> 5.0"
  name    = "ai-governance-vpc"
  cidr    = "10.0.0.0/16"

  azs             = ["eu-west-1a", "eu-west-1b", "eu-west-1c"]
  private_subnets = ["10.0.1.0/24", "10.0.2.0/24", "10.0.3.0/24"]
  public_subnets  = ["10.0.101.0/24", "10.0.102.0/24", "10.0.103.0/24"]

  enable_nat_gateway = true
}