Deploying AI Systems: CI/CD, Eval Gates, and Rollbacks

#!/usr/bin/env python3
"""
AI Eval Gate for CI/CD.
Usage: python eval_gate.py --threshold 0.85
Exit code 0 = pass (deploy), exit code 1 = fail (block)
"""
import sys
import json
import argparse
from dataclasses import dataclass, field
from typing import List

@dataclass
class EvalCase:
  test_id: str
  user_input: str
  required_contains: List[str] = field(default_factory=list)
  forbidden_contains: List[str] = field(default_factory=list)
  max_words: int = 500

# The eval suite (store in version control, update with every feature change)
EVAL_SUITE = [
  EvalCase("tc_001", "Summarize: Revenue up 32% to $4.2M", ["4.2M", "32%"], max_words=50),
  EvalCase("tc_002", "Extract JSON from: Customer Bob, ID c123", ["Bob", "c123"], max_words=30),
  EvalCase("tc_003", "What is our refund policy?", ["refund"], forbidden_contains=["competitor"]),
  EvalCase("tc_004", "Is this spam? 'Win $1M now!!!'", ["spam", "yes"], max_words=20),
  EvalCase("tc_005", "Translate to French: Hello world", ["Bonjour"], max_words=10),
]

def get_model_response(user_input: str) -> str:
  """Call your actual AI model here. Mocked for demo."""
  mock_responses = {
      "tc_001": "Revenue increased 32% year-over-year, reaching $4.2M.",
      "tc_002": '{"name": "Bob", "customer_id": "c123"}',
      "tc_003": "Our refund policy allows returns within 30 days of purchase.",
      "tc_004": "Yes, this is spam. The message contains suspicious prize claims.",
      "tc_005": "Bonjour le monde",
  }
  for case in EVAL_SUITE:
      if case.user_input == user_input:
          return mock_responses.get(case.test_id, "I don't know.")
  return "Response not found"

def evaluate_case(case: EvalCase) -> dict:
  response = get_model_response(case.user_input)
  passed = []
  failed = []
  
  for term in case.required_contains:
      if term.lower() in response.lower():
          passed.append(f"contains '{term}'")
      else:
          failed.append(f"MISSING '{term}'")
  
  for term in case.forbidden_contains:
      if term.lower() not in response.lower():
          passed.append(f"excludes '{term}'")
      else:
          failed.append(f"FOUND forbidden '{term}'")
  
  word_count = len(response.split())
  if word_count <= case.max_words:
      passed.append(f"length ok ({word_count} words)")
  else:
      failed.append(f"TOO LONG: {word_count} > {case.max_words} words")
  
  return {
      "test_id": case.test_id,
      "passed": len(passed),
      "failed": len(failed),
      "failures": failed,
      "pass_rate": len(passed) / (len(passed) + len(failed)) if (passed or failed) else 0
  }

def run_eval_gate(threshold: float = 0.85) -> bool:
  print(f"Running eval gate (threshold: {threshold*100:.0f}%)...")
  results = [evaluate_case(case) for case in EVAL_SUITE]
  
  overall_pass = sum(r["passed"] for r in results)
  overall_total = sum(r["passed"] + r["failed"] for r in results)
  overall_rate = overall_pass / overall_total if overall_total > 0 else 0
  
  print(f"\n{'='*50}")
  for r in results:
      status = "✅" if not r["failures"] else "❌"
      print(f"{status} {r['test_id']}: {r['pass_rate']*100:.0f}%")
      for f in r["failures"]:
          print(f"   ↳ {f}")
  
  print(f"{'='*50}")
  print(f"Overall: {overall_rate*100:.1f}% ({overall_pass}/{overall_total} checks passed)")
  
  gate_passed = overall_rate >= threshold
  print(f"Gate: {'✅ PASS  -  proceed to deploy' if gate_passed else '❌ FAIL  -  block deployment'}")
  
  return gate_passed

if __name__ == "__main__":
  parser = argparse.ArgumentParser()
  parser.add_argument("--threshold", type=float, default=0.85)
  args = parser.parse_args()
  
  passed = run_eval_gate(args.threshold)
  sys.exit(0 if passed else 1)