:::tip 🎮 Interactive Playground Visualize this concept: Try the Benchmark Explorer demo on the EngineersOfAI Playground - no code required. :::
Continuous Eval in CI/CD
Reading time: 40 minutes | Interview relevance: Very High | Target roles: AI Engineer, MLOps, Platform Engineer, ML Engineer
The Post-Mortem Treadmill
The company had 200 engineers and shipped 10–15 prompt changes per week. Their AI system - a customer-facing assistant that handled product questions, support escalations, and account management - was central to their user experience. They were good at catching regressions in code review; experienced engineers knew what to look for, the review culture was strong, and they had an informal rule that any prompt change needed at least two reviewers.
But two or three regressions slipped through every month. Each one followed the same pattern: a change shipped on Tuesday, degraded behavior accumulated through Wednesday and Thursday, a customer escalation surfaced Friday, an emergency rollback consumed the weekend, and a post-mortem meeting filled the following Monday. The slip rate was not catastrophic, but it was expensive. Each incident cost roughly 20 engineer-hours (diagnosis, fix, rollback, post-mortem), generated user trust erosion that was hard to quantify, and required a post-mortem report that leadership read carefully.
The engineering team proposed a solution: a CI/CD pipeline where every prompt change went through automated evaluation before merge, and every production deployment ran continuous quality monitoring. The proposal was straightforward in concept but required solving several real engineering problems: how do you run meaningful evaluation in under 5 minutes at the PR stage without spending too much money? How do you set statistical thresholds for production anomaly detection that are sensitive enough to catch real problems without generating too many false alarms? How do you make rollback fast enough to limit user impact when an anomaly is detected?
They implemented the system over four weeks. After it went live, the slip rate dropped from 2–3 per month to 2 in the following six months. The two that did slip were both caught by production monitoring within 45 minutes and rolled back before generating customer escalations. The post-mortem treadmill stopped.
Why CI/CD for AI Systems
Traditional software CI/CD is well-understood: code changes go through automated tests, merge checks, and deployment pipelines before reaching production. The same principle applies to AI systems, with three important differences.
The test artifact is behavioral, not functional. Software tests check deterministic behavior: function returns the expected value, API returns the correct status code. AI system tests check probabilistic behavior: does the response exhibit expected properties 70%+ of the time? Does the faithfulness score stay above 0.8? This means every evaluation has inherent variance, and the CI gate must account for it through multi-run tests and statistical thresholds.
The "code" is heterogeneous. An AI system change might be: a new model version, a prompt change, a retrieval configuration change, a post-processing rule change, or a combination. Each type of change has different failure modes and different evaluation requirements. A CI/CD pipeline for AI must evaluate the right things for the right type of change.
Production is the hardest evaluator. Even a comprehensive evaluation suite will miss some failures because it can only cover the query types that were anticipated when the suite was built. Production traffic surfaces failure modes that no one anticipated. Continuous production monitoring is not optional - it is the safety net for everything that escapes evaluation gates.
Historical Context
Continuous integration was formalized in the late 1990s by Kent Beck as part of Extreme Programming, popularized by tools like CruiseControl (2001) and later Jenkins (2005), Travis CI (2011), and GitLab CI (2014). The practice of automated testing as a prerequisite for deployment became standard in software engineering by the mid-2000s.
Machine learning CI/CD (MLOps) emerged as a distinct discipline around 2018–2020, driven by companies like Google (which published the paper "Hidden Technical Debt in Machine Learning Systems" in 2015) and Netflix (whose chaos engineering practices influenced ML reliability). Tools like MLflow, Kubeflow, and SageMaker Pipelines codified ML-specific CI/CD patterns.
Prompt-level CI/CD is a 2022–2024 development, emerging alongside the explosion of LLM-powered products. The specific challenge of evaluating prompt changes in CI before they reach production was not systematically addressed until teams at companies like Anthropic, OpenAI, and Cohere started publishing about their internal evaluation infrastructure. By 2024, a small ecosystem of tools specifically designed for LLM CI/CD had emerged (Braintrust, PromptFoo, LangSmith, Weave), each implementing some version of the evaluation-gate pattern.
The Four-Stage Pipeline
Stage 1: PR Evaluation (under 5 minutes)
Triggered: when a pull request is opened or updated.
Goals: fast feedback that catches obvious regressions without blocking the engineer for long. Run only the highest-priority tests.
- Prompt linting: Format checks, token count limits, injection risk scanning
- CRITICAL tests only: The 10–20 tests that represent non-negotiable behaviors
- Cost delta estimate: Will this change increase per-query cost?
- Auto-blocking: Any CRITICAL failure blocks merge
Stage 2: Merge Evaluation (under 30 minutes)
Triggered: when a PR is merged to the main branch.
Goals: comprehensive regression check before staging deployment.
- Full regression suite: All severity levels in parallel
- LLM judge on sampled cases: 50-case quality scoring with baseline comparison
- Safety evaluation: Prompt injection vulnerability scan
- Deployment recommendation: APPROVE / REVIEW_NEEDED / BLOCK
Stage 3: Pre-Deployment Gate (before production deployment)
Triggered: before promoting from staging to production.
Goals: final validation against golden dataset with rollout plan.
- Full benchmark on golden dataset: All 4 RAGAS metrics on full test set
- Shadow comparison (if available): Compare to current production on live traffic sample
- Monitoring setup: Initialize dashboards and alert thresholds for the new version
Stage 4: Post-Deployment Monitoring (continuous)
Triggered: continuously in production.
Goals: catch degradations that escape evaluation gates and surface from real user traffic.
- Sampled evaluation: Score 5% of production traffic in real time using LLM judge
- Statistical process control: Detect anomalies using 3-sigma control charts
- Alert routing: Page on-call engineer when anomaly detected
- Automated rollback: Trigger rollback automatically for critical degradations
Production Monitoring with Statistical Process Control
Statistical process control (SPC) is a method developed by Walter Shewhart in the 1920s for manufacturing quality control. It detects when a process is producing results outside its normal variation - distinguishing "common cause variation" (normal randomness) from "special cause variation" (a real change in the process).
In the context of LLM monitoring, the "process" is your AI system's quality, and the "products" are the quality metric scores computed on sampled production traffic. SPC creates a control chart: a plot of quality metrics over time with upper and lower control limits set at 3 standard deviations from the baseline mean.
Why 3 sigma? At 3 sigma, the probability of a false alarm on any single measurement is 0.27%. If you measure every hour, you expect a false alarm once every 370 hours (~15 days) - acceptable in practice. At 2 sigma, false alarms occur every 22 hours - too many for on-call engineers.
A metric reading outside the control limits is called an "out-of-control signal" and triggers investigation. Important: an out-of-control signal does not mean the system is failing - it means the system's behavior has changed in a statistically detectable way. The investigation determines whether the change is a regression or an improvement.
Complete Implementation
import anthropic
import asyncio
import math
import statistics
import json
import hashlib
from dataclasses import dataclass, field
from typing import Optional, Callable
from enum import Enum
from datetime import datetime, timezone, timedelta
from collections import defaultdict, deque
client = anthropic.Anthropic()
async_client = anthropic.AsyncAnthropic()
# ---------------------------------------------------------------------------
# Data Structures
# ---------------------------------------------------------------------------
class RiskLevel(Enum):
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
CRITICAL = "critical"
class Recommendation(Enum):
APPROVE = "APPROVE"
REVIEW_NEEDED = "REVIEW_NEEDED"
BLOCK = "BLOCK"
@dataclass
class LintIssue:
severity: str # "error" | "warning" | "info"
check_name: str
message: str
fix_suggestion: str
@dataclass
class LintResult:
issues: list[LintIssue]
passed: bool
blocking_issues: list[LintIssue]
@dataclass
class PRCheckResult:
lint_passed: bool
fast_regression_passed: bool
cost_delta_usd: float
blocking_issues: list[str]
pr_comment_markdown: str
@dataclass
class QualityMetrics:
timestamp: str
faithfulness: float
relevance: float
context_precision: float
pass_rate: float
latency_p50_ms: float
latency_p95_ms: float
n_samples: int
prompt_version: str
@dataclass
class ControlChart:
metric_name: str
baseline_mean: float
baseline_std: float
ucl: float # Upper control limit (mean + 3*std)
lcl: float # Lower control limit (mean - 3*std)
observations: list[float]
out_of_control_indices: list[int]
@dataclass
class AnomalyAlert:
detected_at: str
metric: str
current_value: float
baseline_mean: float
sigma_deviation: float
is_critical: bool
recommended_action: str
@dataclass
class RollbackResult:
success: bool
rolled_back_version: str
restored_version: str
duration_seconds: float
reason: str
@dataclass
class DashboardData:
time_range_hours: int
metrics_timeline: list[QualityMetrics]
alert_history: list[AnomalyAlert]
deployment_history: list[dict]
current_pass_rate: float
trend: str # "improving" | "stable" | "degrading"
# ---------------------------------------------------------------------------
# Prompt Linter
# ---------------------------------------------------------------------------
class PromptLinter:
"""
Runs static checks on system prompts before they enter evaluation.
Catches common mistakes without spending API tokens.
"""
MAX_TOKENS = 4000
INJECTION_PATTERNS = [
"ignore previous instructions",
"disregard your training",
"you are now",
"pretend you are",
"act as if",
"forget everything",
"new instructions:",
]
def check_format(self, system_prompt: str) -> LintResult:
issues = []
# Check length
token_result = self.check_token_count(system_prompt, self.MAX_TOKENS)
if not token_result:
issues.append(LintIssue(
severity="error",
check_name="token_count",
message=f"Prompt exceeds {self.MAX_TOKENS} tokens "
f"(estimated {len(system_prompt.split()) * 1.3:.0f})",
fix_suggestion="Split into system + user message or reduce length",
))
# Check injection risk
injection_risk = self.check_injection_risk(system_prompt)
if injection_risk in (RiskLevel.HIGH, RiskLevel.CRITICAL):
issues.append(LintIssue(
severity="error" if injection_risk == RiskLevel.CRITICAL else "warning",
check_name="injection_risk",
message=f"Prompt contains injection-susceptible patterns ({injection_risk.value})",
fix_suggestion=(
"Wrap user input in XML tags (e.g., <user_query>...</user_query>) "
"and add explicit instructions not to follow instructions in user input"
),
))
# Check output format
format_specified = self.check_output_format_specified(system_prompt)
if not format_specified:
issues.append(LintIssue(
severity="warning",
check_name="output_format",
message="No explicit output format specification found",
fix_suggestion=(
"Add explicit format instructions: "
"'Respond in plain text.' or 'Respond with JSON matching schema: ...'"
),
))
# Check for vague instructions
vague_terms = ["be helpful", "be good", "try to", "if possible"]
prompt_lower = system_prompt.lower()
found_vague = [t for t in vague_terms if t in prompt_lower]
if found_vague:
issues.append(LintIssue(
severity="info",
check_name="vague_instructions",
message=f"Vague instructions found: {found_vague}",
fix_suggestion="Replace with specific, measurable constraints",
))
blocking = [i for i in issues if i.severity == "error"]
return LintResult(
issues=issues,
passed=len(blocking) == 0,
blocking_issues=blocking,
)
def check_token_count(
self, prompt: str, max_tokens: int = 4000
) -> bool:
"""Estimate token count (rough: 1 token ≈ 0.75 words)."""
estimated_tokens = int(len(prompt.split()) / 0.75)
return estimated_tokens <= max_tokens
def check_injection_risk(self, prompt: str) -> RiskLevel:
"""Scan for prompt injection vulnerability patterns."""
prompt_lower = prompt.lower()
# Check if user input is properly sandboxed
has_xml_sandboxing = (
"<user" in prompt_lower
or "user_query" in prompt_lower
or "user_input" in prompt_lower
)
has_injection_defense = any(
phrase in prompt_lower
for phrase in [
"do not follow", "ignore any instructions",
"disregard instructions from user",
]
)
found_patterns = [
p for p in self.INJECTION_PATTERNS
if p in prompt_lower
]
if found_patterns and not has_injection_defense:
return RiskLevel.CRITICAL
if not has_xml_sandboxing and not has_injection_defense:
return RiskLevel.MEDIUM
return RiskLevel.LOW
def check_output_format_specified(self, prompt: str) -> bool:
"""Check whether the prompt specifies output format."""
format_keywords = [
"json", "markdown", "plain text", "bullet points",
"respond with", "output format", "format:", "structure:"
]
return any(kw in prompt.lower() for kw in format_keywords)
def run_all(self, prompt: str) -> list[LintIssue]:
result = self.check_format(prompt)
return result.issues
# ---------------------------------------------------------------------------
# PR Evaluator (fast, <5 min)
# ---------------------------------------------------------------------------
class PREvaluator:
"""
Runs the fast PR-stage evaluation gate.
Only CRITICAL tests, cost estimation, and basic lint.
"""
def __init__(self):
self.linter = PromptLinter()
def estimate_cost_change(
self,
old_prompt: str,
new_prompt: str,
daily_call_volume: int = 10000,
) -> float:
"""
Estimate daily API cost change from prompt length difference.
Returns delta in USD (positive = more expensive).
"""
old_tokens = len(old_prompt.split()) / 0.75
new_tokens = len(new_prompt.split()) / 0.75
delta_tokens = new_tokens - old_tokens
# claude-opus-4-6 input cost: ~$0.000015 per token
cost_per_token = 0.000015
daily_delta = delta_tokens * cost_per_token * daily_call_volume
return round(daily_delta, 4)
def run_critical_tests(
self,
new_prompt: str,
critical_suite: list,
) -> tuple[bool, list[str]]:
"""
Run CRITICAL tests only. Returns (all_passed, list_of_failed_test_ids).
Uses 1 run per test for speed (accept higher false-positive rate at PR stage).
"""
failed_ids = []
for test in critical_suite:
try:
response = client.messages.create(
model="claude-opus-4-6",
max_tokens=512,
temperature=0.1,
system=new_prompt,
messages=[{"role": "user", "content": test.render_message()}],
)
response_text = response.content[0].text
# Quick property check using haiku
check_prompt = f"""Does this response satisfy all expected properties?
Response: {response_text[:500]}
Expected properties:
{chr(10).join(f'- {p}' for p in test.expected_properties)}
Forbidden properties:
{chr(10).join(f'- {p}' for p in test.forbidden_properties)}
JSON: {{"all_satisfied": true/false}}"""
check_response = client.messages.create(
model="claude-haiku-4-5-20251001",
max_tokens=32,
messages=[{"role": "user", "content": check_prompt}],
)
raw = check_response.content[0].text.strip()
try:
start = raw.index("{")
end = raw.rindex("}") + 1
result = json.loads(raw[start:end])
if not result.get("all_satisfied", True):
failed_ids.append(test.test_id)
except Exception:
pass # Parse failure - don't block on ambiguity
except Exception as e:
failed_ids.append(test.test_id)
return len(failed_ids) == 0, failed_ids
def generate_pr_comment(
self,
lint_result: LintResult,
critical_passed: bool,
failed_tests: list[str],
cost_delta_usd: float,
) -> str:
"""Generate a markdown PR comment with evaluation results."""
status = "PASSED" if (lint_result.passed and critical_passed) else "FAILED"
emoji = "✓" if status == "PASSED" else "✗"
lines = [
f"## AI Evaluation Gate - {emoji} {status}",
"",
"### Lint Results",
]
if not lint_result.issues:
lines.append("No issues found.")
else:
for issue in lint_result.issues:
marker = "[ERROR]" if issue.severity == "error" else "[WARN]"
lines.append(f"- {marker} **{issue.check_name}**: {issue.message}")
lines.append(f" - Fix: {issue.fix_suggestion}")
lines.extend([
"",
"### CRITICAL Tests",
f"Result: {'PASSED' if critical_passed else 'FAILED'}",
])
if failed_tests:
for tid in failed_tests:
lines.append(f"- FAILED: `{tid}`")
lines.extend([
"",
"### Cost Impact",
f"Estimated daily cost change: ${cost_delta_usd:+.4f} USD",
" (at 10K calls/day - adjust for your volume)",
"",
])
if status == "FAILED":
lines.extend([
"---",
"> **Merge blocked.** Fix the issues above before this PR can be merged.",
])
return "\n".join(lines)
def run_pr_check(
self,
old_prompt: str,
new_prompt: str,
critical_suite: list,
) -> PRCheckResult:
"""Run the full PR evaluation and return results."""
lint_result = self.linter.check_format(new_prompt)
critical_passed, failed_tests = self.run_critical_tests(
new_prompt, critical_suite
)
cost_delta = self.estimate_cost_change(old_prompt, new_prompt)
blocking = [i.message for i in lint_result.blocking_issues]
if failed_tests:
blocking.extend([f"CRITICAL test failed: {t}" for t in failed_tests])
comment = self.generate_pr_comment(
lint_result, critical_passed, failed_tests, cost_delta
)
return PRCheckResult(
lint_passed=lint_result.passed,
fast_regression_passed=critical_passed,
cost_delta_usd=cost_delta,
blocking_issues=blocking,
pr_comment_markdown=comment,
)
# ---------------------------------------------------------------------------
# Merge Evaluator (comprehensive, <30 min)
# ---------------------------------------------------------------------------
class MergeEvaluator:
"""
Comprehensive evaluation at merge time.
Runs full suite in parallel, compares to baseline, checks safety.
"""
def llm_judge_sample(
self,
system_prompt: str,
test_questions: list[str],
n: int = 50,
) -> dict[str, float]:
"""
Run LLM-as-judge quality scoring on a sample of test questions.
Returns aggregate quality metrics.
"""
import random
sample = random.sample(test_questions, min(n, len(test_questions)))
scores = []
for question in sample:
# Generate response
response = client.messages.create(
model="claude-opus-4-6",
max_tokens=512,
temperature=0.1,
system=system_prompt,
messages=[{"role": "user", "content": question}],
)
response_text = response.content[0].text
# Judge quality
judge_prompt = f"""Rate this AI response on a scale of 1-5.
Question: {question}
Response: {response_text[:500]}
Criteria: accuracy, completeness, clarity, appropriateness
JSON: {{"score": 1-5, "reasoning": "brief"}}"""
judge_response = client.messages.create(
model="claude-haiku-4-5-20251001",
max_tokens=64,
messages=[{"role": "user", "content": judge_prompt}],
)
try:
raw = judge_response.content[0].text.strip()
start = raw.index("{")
end = raw.rindex("}") + 1
result = json.loads(raw[start:end])
scores.append(float(result.get("score", 3)) / 5.0)
except Exception:
scores.append(0.5)
if not scores:
return {"mean_quality": 0.5, "n_scored": 0}
return {
"mean_quality": round(statistics.mean(scores), 4),
"median_quality": round(statistics.median(scores), 4),
"std_quality": round(statistics.stdev(scores) if len(scores) > 1 else 0, 4),
"n_scored": len(scores),
}
def compare_to_baseline(
self,
candidate_metrics: dict[str, float],
baseline_metrics: dict[str, float],
threshold_delta: float = -0.05,
) -> tuple[Recommendation, dict[str, float]]:
"""
Compare candidate to baseline and recommend action.
threshold_delta: how much degradation is acceptable (negative = degradation)
"""
deltas = {
metric: candidate_metrics.get(metric, 0)
- baseline_metrics.get(metric, 0)
for metric in baseline_metrics
}
blocking_regressions = [
(metric, delta)
for metric, delta in deltas.items()
if delta < threshold_delta
]
if not blocking_regressions:
recommendation = Recommendation.APPROVE
elif any(abs(d) > 0.15 for _, d in blocking_regressions):
recommendation = Recommendation.BLOCK
else:
recommendation = Recommendation.REVIEW_NEEDED
return recommendation, deltas
def run_full_evaluation(
self,
new_prompt: str,
baseline_prompt: str,
suite: list,
test_questions: list[str],
) -> dict:
"""Run complete merge evaluation."""
print("Running full regression suite on candidate...")
from concurrent.futures import ThreadPoolExecutor
def run_single(test):
try:
response = client.messages.create(
model="claude-opus-4-6",
max_tokens=512,
temperature=0.1,
system=new_prompt,
messages=[{"role": "user", "content": test.render_message()}],
)
return test.test_id, True, response.content[0].text
except Exception as e:
return test.test_id, False, str(e)
# Run suite in parallel using thread pool
with ThreadPoolExecutor(max_workers=10) as executor:
candidate_results = list(executor.map(run_single, suite))
pass_count = sum(1 for _, passed, _ in candidate_results if passed)
suite_pass_rate = pass_count / max(1, len(suite))
# LLM judge sample
print("Running LLM judge on sample...")
candidate_quality = self.llm_judge_sample(new_prompt, test_questions)
baseline_quality = self.llm_judge_sample(baseline_prompt, test_questions, n=20)
# Compare
recommendation, deltas = self.compare_to_baseline(
{"quality": candidate_quality["mean_quality"]},
{"quality": baseline_quality["mean_quality"]},
)
return {
"suite_pass_rate": round(suite_pass_rate, 4),
"candidate_quality": candidate_quality,
"baseline_quality": baseline_quality,
"metric_deltas": deltas,
"recommendation": recommendation.value,
"n_tests": len(suite),
}
# ---------------------------------------------------------------------------
# Production Monitor
# ---------------------------------------------------------------------------
class ProductionMonitor:
"""
Continuous production quality monitoring with statistical process control.
Samples production traffic, evaluates quality, and detects anomalies.
"""
def __init__(self, baseline_metrics: Optional[QualityMetrics] = None):
self.baseline = baseline_metrics
self.history: deque[QualityMetrics] = deque(maxlen=168) # 1 week at hourly
self.alert_history: list[AnomalyAlert] = []
def evaluate_sample(
self,
sampled_queries: list[dict],
system_prompt: str,
) -> QualityMetrics:
"""
Evaluate a sample of production queries.
Each query: {user_message, retrieved_context (optional)}
"""
import time
latencies = []
faithfulness_scores = []
relevance_scores = []
n_total = len(sampled_queries)
n_eval = min(n_total, 20) # Cap expensive evaluations
for query in sampled_queries[:n_eval]:
t0 = time.time()
try:
response = client.messages.create(
model="claude-opus-4-6",
max_tokens=512,
temperature=0.0,
system=system_prompt,
messages=[{"role": "user", "content": query["user_message"]}],
)
latency_ms = (time.time() - t0) * 1000
latencies.append(latency_ms)
# Quick quality score
judge_prompt = f"""Rate this response 1-5 for quality.
Q: {query['user_message'][:200]}
A: {response.content[0].text[:300]}
JSON: {{"score": 1-5}}"""
judge_resp = client.messages.create(
model="claude-haiku-4-5-20251001",
max_tokens=16,
messages=[{"role": "user", "content": judge_prompt}],
)
try:
raw = judge_resp.content[0].text.strip()
start = raw.index("{")
end = raw.rindex("}") + 1
result = json.loads(raw[start:end])
score = float(result.get("score", 3))
faithfulness_scores.append(score / 5.0)
relevance_scores.append(score / 5.0) # Simplified for demo
except Exception:
faithfulness_scores.append(0.5)
relevance_scores.append(0.5)
except Exception:
latencies.append(5000) # Penalize errors with high latency
faithfulness_scores.append(0.0)
relevance_scores.append(0.0)
latencies_sorted = sorted(latencies)
n_lat = len(latencies_sorted)
metrics = QualityMetrics(
timestamp=datetime.now(timezone.utc).isoformat(),
faithfulness=statistics.mean(faithfulness_scores) if faithfulness_scores else 0,
relevance=statistics.mean(relevance_scores) if relevance_scores else 0,
context_precision=0.0, # Requires retrieval data
pass_rate=sum(1 for s in faithfulness_scores if s >= 0.6) / max(1, n_eval),
latency_p50_ms=(
latencies_sorted[n_lat // 2] if latencies_sorted else 0
),
latency_p95_ms=(
latencies_sorted[int(n_lat * 0.95)] if latencies_sorted else 0
),
n_samples=n_total,
prompt_version=query.get("prompt_version", "unknown"),
)
self.history.append(metrics)
return metrics
def statistical_process_control(
self,
metric_name: str,
n_sigma: float = 3.0,
) -> ControlChart:
"""
Build a statistical process control chart for a metric.
Uses the first N observations as baseline to set control limits.
"""
observations = []
for m in self.history:
val = getattr(m, metric_name, None)
if val is not None:
observations.append(float(val))
if len(observations) < 10:
# Not enough data for SPC - use generous defaults
baseline_mean = statistics.mean(observations) if observations else 0.8
baseline_std = 0.1
else:
# Use first half as baseline
baseline_obs = observations[:len(observations) // 2]
baseline_mean = statistics.mean(baseline_obs)
baseline_std = statistics.stdev(baseline_obs) if len(baseline_obs) > 1 else 0.05
ucl = baseline_mean + n_sigma * baseline_std
lcl = max(0.0, baseline_mean - n_sigma * baseline_std)
out_of_control = [
i for i, val in enumerate(observations)
if val > ucl or val < lcl
]
return ControlChart(
metric_name=metric_name,
baseline_mean=round(baseline_mean, 4),
baseline_std=round(baseline_std, 4),
ucl=round(ucl, 4),
lcl=round(lcl, 4),
observations=observations,
out_of_control_indices=out_of_control,
)
def detect_anomaly(
self,
current_metrics: QualityMetrics,
n_sigma: float = 3.0,
) -> Optional[AnomalyAlert]:
"""
Detect whether current metrics indicate an anomaly using SPC.
Returns an alert if any metric is out of control.
"""
if not self.baseline:
return None
# Check each quality metric
metric_checks = {
"faithfulness": (current_metrics.faithfulness, self.baseline.faithfulness),
"relevance": (current_metrics.relevance, self.baseline.relevance),
"pass_rate": (current_metrics.pass_rate, self.baseline.pass_rate),
}
# Estimate std from history if available
for metric_name, (current_val, baseline_val) in metric_checks.items():
historical_values = [
getattr(m, metric_name, None)
for m in self.history
if getattr(m, metric_name, None) is not None
]
if len(historical_values) < 5:
baseline_std = 0.1 # Default before enough history
else:
baseline_std = max(
0.01, statistics.stdev(historical_values[-20:])
)
deviation = (current_val - baseline_val) / max(baseline_std, 0.001)
if deviation < -n_sigma:
is_critical = deviation < -(n_sigma + 1)
alert = AnomalyAlert(
detected_at=datetime.now(timezone.utc).isoformat(),
metric=metric_name,
current_value=round(current_val, 4),
baseline_mean=round(baseline_val, 4),
sigma_deviation=round(deviation, 2),
is_critical=is_critical,
recommended_action=(
"IMMEDIATE_ROLLBACK" if is_critical
else "PAGE_ONCALL_INVESTIGATE"
),
)
self.alert_history.append(alert)
return alert
return None
def should_rollback(self, alert: AnomalyAlert) -> bool:
"""
Determine whether an anomaly warrants automatic rollback.
Criteria: critical severity AND at least 3 consecutive anomalous readings.
"""
if not alert.is_critical:
return False
# Check for consecutive anomalies
recent_alerts = [
a for a in self.alert_history[-5:]
if a.metric == alert.metric
]
return len(recent_alerts) >= 3
# ---------------------------------------------------------------------------
# Rollback Orchestrator
# ---------------------------------------------------------------------------
class RollbackOrchestrator:
"""
Manages prompt version registry and automated rollback procedures.
In production, this would interface with your deployment system.
"""
def __init__(self):
self.version_registry: dict[str, str] = {} # version -> prompt
self.current_version: str = "v0"
self.deployment_history: list[dict] = []
def register_version(self, version: str, prompt: str) -> None:
self.version_registry[version] = prompt
self.deployment_history.append({
"version": version,
"deployed_at": datetime.now(timezone.utc).isoformat(),
"status": "active" if version == self.current_version else "archived",
})
def trigger_rollback(
self,
reason: str,
target_version: Optional[str] = None,
) -> RollbackResult:
"""
Roll back to previous version or specified target version.
"""
import time
t0 = time.time()
# Find the version to restore
if target_version:
restore_version = target_version
else:
# Roll back to most recent previous version
versions = sorted(self.version_registry.keys())
current_idx = versions.index(self.current_version) if self.current_version in versions else -1
if current_idx <= 0:
return RollbackResult(
success=False,
rolled_back_version=self.current_version,
restored_version="none",
duration_seconds=0,
reason="No previous version available for rollback",
)
restore_version = versions[current_idx - 1]
rolled_back = self.current_version
self.current_version = restore_version
# Update deployment history
self.deployment_history.append({
"version": restore_version,
"deployed_at": datetime.now(timezone.utc).isoformat(),
"status": "rollback",
"reason": reason,
})
duration = time.time() - t0
print(f"ROLLBACK: {rolled_back} -> {restore_version} | Reason: {reason}")
return RollbackResult(
success=True,
rolled_back_version=rolled_back,
restored_version=restore_version,
duration_seconds=round(duration, 3),
reason=reason,
)
def notify_oncall(self, event: dict) -> None:
"""Notify on-call engineer. In production: PagerDuty, Slack, etc."""
print(f"[ONCALL ALERT] {json.dumps(event, indent=2)}")
def restore_previous_prompt_version(self) -> str:
"""Return the prompt text of the most recent previous version."""
versions = sorted(self.version_registry.keys())
current_idx = (
versions.index(self.current_version)
if self.current_version in versions
else -1
)
if current_idx <= 0:
return ""
previous_version = versions[current_idx - 1]
return self.version_registry[previous_version]
# ---------------------------------------------------------------------------
# Observability: Tracing
# ---------------------------------------------------------------------------
@dataclass
class LLMTrace:
"""
Full trace for a single LLM interaction, from question to feedback.
"""
trace_id: str
user_session_id: str
prompt_version: str
model: str
user_query: str
retrieved_chunks: list[str] # chunk IDs
system_prompt_hash: str
response_text: str
latency_ms: float
input_tokens: int
output_tokens: int
quality_score: Optional[float]
user_feedback: Optional[int] # 1-5 or None
timestamp: str = field(
default_factory=lambda: datetime.now(timezone.utc).isoformat()
)
metadata: dict = field(default_factory=dict)
class TraceCollector:
"""
Collects and stores traces for observability and sampled evaluation.
"""
def __init__(self, sample_rate: float = 0.05):
self.sample_rate = sample_rate
self.traces: list[LLMTrace] = []
def should_sample(self) -> bool:
import random
return random.random() < self.sample_rate
def record(self, trace: LLMTrace) -> None:
self.traces.append(trace)
def get_recent(self, hours: int = 1) -> list[LLMTrace]:
cutoff = (
datetime.now(timezone.utc) - timedelta(hours=hours)
).isoformat()
return [t for t in self.traces if t.timestamp >= cutoff]
def aggregate_metrics(
self, traces: list[LLMTrace]
) -> dict[str, float]:
"""Compute aggregate quality metrics from a set of traces."""
if not traces:
return {}
quality_scores = [
t.quality_score for t in traces if t.quality_score is not None
]
user_feedback = [
t.user_feedback for t in traces if t.user_feedback is not None
]
latencies = [t.latency_ms for t in traces]
result = {
"n_traces": len(traces),
"mean_latency_ms": round(statistics.mean(latencies), 1),
}
if quality_scores:
result["mean_quality_score"] = round(
statistics.mean(quality_scores), 4
)
if user_feedback:
result["mean_user_feedback"] = round(
statistics.mean(user_feedback), 4
)
return result
# ---------------------------------------------------------------------------
# Eval Dashboard
# ---------------------------------------------------------------------------
class EvalDashboard:
"""
Computes and returns dashboard data for UI rendering.
In production, this would feed a Grafana dashboard or custom UI.
"""
def __init__(
self,
monitor: ProductionMonitor,
orchestrator: RollbackOrchestrator,
tracer: TraceCollector,
):
self.monitor = monitor
self.orchestrator = orchestrator
self.tracer = tracer
def compute_dashboard_data(
self, time_range_hours: int = 24
) -> DashboardData:
"""Build complete dashboard data snapshot."""
recent_traces = self.tracer.get_recent(hours=time_range_hours)
# Current metrics
metrics_list = list(self.monitor.history)
if metrics_list:
current_pass_rate = metrics_list[-1].pass_rate
else:
current_pass_rate = 0.0
# Trend: compare first half to second half
if len(metrics_list) >= 4:
mid = len(metrics_list) // 2
early_avg = statistics.mean(m.pass_rate for m in metrics_list[:mid])
recent_avg = statistics.mean(m.pass_rate for m in metrics_list[mid:])
delta = recent_avg - early_avg
if delta > 0.02:
trend = "improving"
elif delta < -0.02:
trend = "degrading"
else:
trend = "stable"
else:
trend = "insufficient_data"
return DashboardData(
time_range_hours=time_range_hours,
metrics_timeline=metrics_list,
alert_history=self.monitor.alert_history,
deployment_history=self.orchestrator.deployment_history,
current_pass_rate=round(current_pass_rate, 4),
trend=trend,
)
# ---------------------------------------------------------------------------
# Full CI/CD Pipeline Configuration
# ---------------------------------------------------------------------------
PIPELINE_CONFIG = {
"pr_stage": {
"critical_tests_only": True,
"max_time_seconds": 300,
"blocking": True,
"model": "claude-opus-4-6",
"judge_model": "claude-haiku-4-5-20251001",
},
"merge_stage": {
"full_suite": True,
"max_time_seconds": 1800,
"regression_threshold": 0.95,
"llm_judge_sample_n": 50,
"parallel_workers": 10,
},
"deploy_stage": {
"golden_dataset_eval": True,
"shadow_comparison": False, # Requires traffic splitting infrastructure
"max_time_seconds": 3600,
},
"monitoring": {
"sample_rate": 0.05, # 5% of production traffic
"eval_window_minutes": 60, # Evaluate hourly windows
"alert_threshold_sigma": 3.0,
"auto_rollback_on_critical": True,
"auto_rollback_consecutive_alerts": 3,
"min_samples_before_alert": 50,
},
}
class CICDPipeline:
"""
Orchestrates the full CI/CD pipeline for an AI system.
"""
def __init__(self, config: dict = None):
self.config = config or PIPELINE_CONFIG
self.linter = PromptLinter()
self.pr_evaluator = PREvaluator()
self.merge_evaluator = MergeEvaluator()
self.rollback_orchestrator = RollbackOrchestrator()
self.tracer = TraceCollector(
sample_rate=self.config["monitoring"]["sample_rate"]
)
def on_pr_open(
self,
old_prompt: str,
new_prompt: str,
critical_suite: list,
) -> PRCheckResult:
"""Handler for PR open event."""
print("Running PR evaluation gate...")
result = self.pr_evaluator.run_pr_check(
old_prompt, new_prompt, critical_suite
)
if result.blocking_issues:
print("PR BLOCKED:")
for issue in result.blocking_issues:
print(f" - {issue}")
else:
print("PR gate: PASSED")
print("\nPR Comment:")
print(result.pr_comment_markdown)
return result
def on_merge(
self,
new_prompt: str,
baseline_prompt: str,
full_suite: list,
test_questions: list[str],
version: str,
) -> dict:
"""Handler for merge event - runs comprehensive evaluation."""
print(f"Running merge evaluation for version {version}...")
result = self.merge_evaluator.run_full_evaluation(
new_prompt, baseline_prompt, full_suite, test_questions
)
self.rollback_orchestrator.register_version(version, new_prompt)
print(f"Merge evaluation: {result['recommendation']}")
return result
def setup_production_monitoring(
self,
baseline_metrics: QualityMetrics,
) -> ProductionMonitor:
"""Initialize production monitor for a new deployment."""
monitor = ProductionMonitor(baseline_metrics=baseline_metrics)
print("Production monitoring initialized")
return monitor
def run_monitoring_cycle(
self,
monitor: ProductionMonitor,
sampled_queries: list[dict],
system_prompt: str,
) -> Optional[AnomalyAlert]:
"""Run one monitoring evaluation cycle."""
metrics = monitor.evaluate_sample(sampled_queries, system_prompt)
alert = monitor.detect_anomaly(metrics)
if alert:
print(f"ANOMALY: {alert.metric} = {alert.current_value:.4f} "
f"({alert.sigma_deviation:.1f}σ from baseline)")
if monitor.should_rollback(alert):
print("Triggering automatic rollback...")
rollback_result = self.rollback_orchestrator.trigger_rollback(
reason=f"Auto-rollback: {alert.metric} anomaly "
f"({alert.sigma_deviation:.1f}σ deviation)"
)
self.rollback_orchestrator.notify_oncall({
"type": "AUTO_ROLLBACK",
"alert": {
"metric": alert.metric,
"value": alert.current_value,
"sigma": alert.sigma_deviation,
},
"rollback": {
"from": rollback_result.rolled_back_version,
"to": rollback_result.restored_version,
},
})
return alert
# ---------------------------------------------------------------------------
# Demo
# ---------------------------------------------------------------------------
if __name__ == "__main__":
print("=== AI CI/CD Pipeline Demo ===\n")
pipeline = CICDPipeline()
# Define a minimal test suite
class MockTest:
def __init__(self, tid, msg, expected, forbidden, severity="CRITICAL"):
from enum import Enum
self.test_id = tid
self.user_message = msg
self.expected_properties = expected
self.forbidden_properties = forbidden
self.severity = type("Severity", (), {"value": severity})()
self.metadata = {}
def render_message(self):
return self.user_message
critical_suite = [
MockTest(
"c001",
"What is 2 + 2?",
["Response gives the answer 4"],
["Response refuses to answer"],
"CRITICAL",
),
]
old_prompt = "You are a helpful assistant."
new_prompt = "You are a helpful assistant. Always respond concisely."
# Stage 1: PR check
print("--- Stage 1: PR Evaluation ---")
lint_only = pipeline.linter.run_all(new_prompt)
print(f"Lint issues: {len(lint_only)}")
for issue in lint_only:
print(f" [{issue.severity.upper()}] {issue.check_name}: {issue.message}")
# Estimate cost
cost_delta = pipeline.pr_evaluator.estimate_cost_change(
old_prompt, new_prompt, daily_call_volume=5000
)
print(f"Estimated daily cost delta: ${cost_delta:+.4f}")
# Simulate baseline metrics for monitoring
baseline = QualityMetrics(
timestamp=datetime.now(timezone.utc).isoformat(),
faithfulness=0.87,
relevance=0.82,
context_precision=0.79,
pass_rate=0.91,
latency_p50_ms=450.0,
latency_p95_ms=1200.0,
n_samples=1000,
prompt_version="v1.0",
)
monitor = pipeline.setup_production_monitoring(baseline)
# Add some history to the monitor
for i in range(15):
monitor.history.append(QualityMetrics(
timestamp=datetime.now(timezone.utc).isoformat(),
faithfulness=0.87 + (i * 0.001),
relevance=0.82,
context_precision=0.79,
pass_rate=0.91,
latency_p50_ms=450.0,
latency_p95_ms=1200.0,
n_samples=100,
prompt_version="v1.0",
))
# Simulate a degradation
degraded_metrics = QualityMetrics(
timestamp=datetime.now(timezone.utc).isoformat(),
faithfulness=0.52, # Significant drop - should trigger alert
relevance=0.80,
context_precision=0.77,
pass_rate=0.71,
latency_p50_ms=890.0,
latency_p95_ms=2100.0,
n_samples=200,
prompt_version="v1.1",
)
monitor.history.append(degraded_metrics)
alert = monitor.detect_anomaly(degraded_metrics)
if alert:
print(f"\nAnomaly detected: {alert.metric}")
print(f" Value: {alert.current_value:.4f} vs baseline {alert.baseline_mean:.4f}")
print(f" Deviation: {alert.sigma_deviation:.1f}σ")
print(f" Action: {alert.recommended_action}")
# SPC chart
chart = monitor.statistical_process_control("faithfulness")
print(f"\nSPC Chart - Faithfulness:")
print(f" Baseline mean: {chart.baseline_mean:.4f}")
print(f" UCL: {chart.ucl:.4f}")
print(f" LCL: {chart.lcl:.4f}")
print(f" Out-of-control: {len(chart.out_of_control_indices)} observations")
Architecture Diagrams
Full CI/CD Pipeline for AI Systems
Production Monitoring with Statistical Process Control
Alert-to-Rollback Decision Tree
Observability: Tracing from Query to Feedback
Production Notes
:::danger Auto-Rollback Must Have a Guard
Automatic rollback is powerful but dangerous if triggered incorrectly. The guard of "3+ consecutive anomalies" prevents a single bad sample from triggering rollback. But you also need a minimum sample count before anomaly detection activates - if you trigger rollback based on 5 samples in the first 10 minutes of deployment, you will generate false rollbacks whenever a legitimate deployment coincides with unusual traffic. Set min_samples_before_alert to at least 50 in production.
:::
:::warning Monitoring Sample Rate Trade-offs A 5% sample rate means that at 1,000 queries per hour, you evaluate 50 per hour - enough for statistically meaningful SPC with hourly windows. At 100 queries per hour, 5% gives you only 5 samples - not enough for reliable detection. Scale your sample rate to maintain a minimum of 30–50 evaluated samples per monitoring window. At low traffic, this may require 20–50% sampling; at high traffic, 1% may be sufficient. :::
:::tip Tagging Every LLM Call Every production LLM call should be tagged with: prompt version, model version, user segment, and a trace ID. These tags are the foundation of meaningful observability. Without them, you cannot isolate whether a quality change affects all users or only a specific segment, whether the change was caused by a prompt update or a model update, or how to reproduce a specific failure for debugging. :::
:::note PR Evaluation Latency Constraint The 5-minute PR stage constraint is a usability requirement, not a technical one. Engineers will work around slow CI gates - they will batch changes, skip certain checks, or simply not wait for them. Five minutes keeps the evaluation in the background without interrupting the engineering flow. To achieve this: run CRITICAL tests only (10–20 tests), use temperature=0.1 to reduce n_runs needed, use claude-haiku for judge evaluation (10x faster than opus), and run all tests in parallel using asyncio. :::
Interview Questions and Answers
Q1: What are the four stages of a CI/CD pipeline for AI systems and what is the purpose of each?
The PR stage (under 5 minutes) provides fast feedback to the engineer before code review is even complete. It runs only CRITICAL tests and prompt linting, blocking obviously bad changes immediately while not delaying the engineering workflow. The merge stage (under 30 minutes) provides comprehensive regression coverage before the change enters the deployment pipeline. It runs the full test suite, samples with an LLM judge, and compares to the production baseline. The pre-deployment stage (under 1 hour) is the final gate before production, running a full golden dataset evaluation and setting up monitoring infrastructure. The production monitoring stage is continuous - it evaluates sampled production traffic in real time, detects anomalies using statistical process control, and triggers rollback when warranted. Each stage has a different speed/depth trade-off calibrated to when it runs in the development lifecycle.
Q2: What is statistical process control and why is it the right tool for LLM quality monitoring?
Statistical process control (SPC) was developed in manufacturing to distinguish normal process variation from special cause variation - real changes in process quality. For LLM monitoring, the "process" is your AI system's quality and the "product" is the quality metric scores computed on sampled traffic. SPC sets upper and lower control limits at 3 standard deviations from the baseline mean. Any measurement outside those limits is flagged as an "out-of-control signal" - meaning the system's behavior has changed in a statistically detectable way. SPC is the right tool because: it has well-understood false alarm rates (0.27% per measurement at 3σ), it adapts to the inherent variance of your system (a noisy metric gets wider control limits automatically), and it has a long track record in quality management that practitioners understand. The alternative - fixed thresholds like "alert if faithfulness drops below 0.7" - ignores your specific system's normal variation and will either miss real regressions or generate too many false alarms.
Q3: How do you set the right alert threshold for production monitoring without generating too many false alarms?
The 3-sigma threshold means each measurement has a 0.27% probability of triggering a false alarm even when nothing is wrong. At hourly measurements, this is about one false alarm every 370 hours - roughly every 15 days - which is manageable for an on-call rotation. The consecutive-anomaly guard (require 3 consecutive out-of-control observations before auto-rollback) reduces the false rollback rate to 0.27%^3 = 0.002% per window. These are the right defaults for most systems. If your system is very noisy (high variance in quality metrics due to diverse traffic), increase to 4σ or require more consecutive observations. If the cost of missing a real regression is extremely high (medical, financial, safety systems), decrease to 2.5σ and accept more false alarms. The parameters should be calibrated to your system's historical variance by running the SPC system in "shadow mode" (alert but don't rollback) for 2–4 weeks and measuring the false alarm rate.
Q4: What should be in the PR comment generated by the evaluation gate, and why does the format matter?
The PR comment should include: lint results with severity and specific fix suggestions, CRITICAL test results with the specific test IDs that failed (not just "3 tests failed"), cost impact estimate in concrete terms ("$+0.42/day at 10K calls"), and a clear status header ("PASSED" or "BLOCKED") that engineers can see without reading the body. The format matters because the PR comment is the primary interface between the evaluation system and the engineering team. If the comment is verbose, confusing, or buries the key result in boilerplate, engineers will stop reading it and stop trusting it. The comment should answer three questions immediately: did it pass, what failed, and what do I need to fix? Everything else is secondary context.
Q5: How do you handle rollback in a system where the "deployment artifact" is a system prompt stored in a database?
The key principle is treating the system prompt like a versioned deployment artifact with the same immutability and rollback guarantees as a container image. Implementation: every prompt version is stored with a version ID (semantic version or git SHA) in a prompt registry. The current production version is tracked separately. Rollback is: (1) identify the previous stable version in the registry, (2) atomically swap the current version pointer to the previous version, (3) verify the previous version is serving correctly by checking the monitoring metrics over the next 5 minutes. The swap should be atomic - no gap during which some requests get the new version and some get the old one, which would make the monitoring signal uninterpretable. In practice, this is implemented as a feature flag update or a database write, which can typically complete in milliseconds. The rollback should be triggered and confirmed within 5 minutes of anomaly detection to limit user impact.
Q6: How do you ensure your production monitoring does not become expensive as traffic scales?
Cost management for production monitoring has three levers. Sample rate: 5% at 10K queries/hour gives 500 evaluated queries per hour, which is more than enough for statistical reliability. At 100K queries/hour, you can safely drop to 1% and still evaluate 1K queries per hour. Use adaptive sampling: if all recent samples have scored above 0.9, reduce the sample rate; if scores are variable, increase it. Judge model selection: use claude-haiku for production monitoring quality scoring - at roughly 10x less expensive than claude-opus-4-6, this keeps the cost of monitoring at 0.5–1% of the production API cost rather than 5–10%. Evaluation scope: for continuous monitoring, you do not need all four RAGAS metrics - a single "quality score" from a well-calibrated LLM judge is sufficient for anomaly detection. Reserve the full RAGAS evaluation for the pre-deployment gate where you have time and budget to run it comprehensively.