02 Infrastructure & MLOps 6 min read 1,347 words
CI/CD for LLM Applications
Deploying LLM applications requires adapting traditional CI/CD practices for AI-specific concerns like model evaluation, prompt testing, and quality gates.
LLM CI/CD Challenges
What Makes LLM Deployments Different
| Traditional CI/CD | LLM CI/CD |
|---|---|
| Binary tests (pass/fail) | Probabilistic evaluation |
| Fast tests | Slow, expensive evaluations |
| Deterministic outputs | Non-deterministic outputs |
| Code changes only | Prompt + model + data changes |
| Version control obvious | Prompt versioning complex |
Change Types
| Change Type | Risk | Testing Required |
|---|---|---|
| Prompt text | Medium | Regression + quality eval |
| System prompt | High | Full evaluation suite |
| Model version | High | Comprehensive benchmark |
| RAG index | Medium | Retrieval + quality eval |
| Parameters (temp, etc) | Low-Medium | Quality sampling |
Pipeline Architecture
Full Pipeline
An LLM CI/CD pipeline is an ordinary pipeline until the quality gate, which is the only stage that can send a build to production or to nothing
Text version of this diagram
┌─────────────────────────────────────────────────────────────────┐
│ LLM CI/CD PIPELINE │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ │
│ │ Commit │ │
│ │ Trigger │ │
│ └──────┬───────┘ │
│ │ │
│ ▼ │
│ ┌──────────────┐ │
│ │ Validate │ ─── Prompt syntax, config validation │
│ └──────┬───────┘ │
│ │ │
│ ▼ │
│ ┌──────────────┐ │
│ │ Unit Tests │ ─── Fast, deterministic tests │
│ └──────┬───────┘ │
│ │ │
│ ▼ │
│ ┌──────────────┐ │
│ │ Golden Set │ ─── Known input/output pairs │
│ │ Tests │ │
│ └──────┬───────┘ │
│ │ │
│ ▼ │
│ ┌──────────────┐ │
│ │ LLM Eval │ ─── Quality scoring, regression detection │
│ │ (Sampled) │ │
│ └──────┬───────┘ │
│ │ │
│ ▼ │
│ ┌──────────────┐ │
│ │ Quality Gate │ ─── Pass/fail based on thresholds │
│ └──────┬───────┘ │
│ │ │
│ ┌────┴────┐ │
│ ▼ ▼ │
│ ┌──────┐ ┌───────┐ │
│ │Canary│ │Blocked│ │
│ │Deploy│ │ │ │
│ └──┬───┘ └───────┘ │
│ │ │
│ ▼ │
│ ┌──────────────┐ │
│ │ Production │ │
│ │ Monitoring │ │
│ └──────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘
Testing Stages
Stage 1: Static Validation
class PromptValidator:
def validate(self, prompt_config: dict) -> ValidationResult:
errors = []
# Required fields
if not prompt_config.get("system_prompt"):
errors.append("Missing system_prompt")
# Template syntax
try:
Template(prompt_config["user_template"]).substitute({})
except KeyError:
pass # Expected for templates with variables
except ValueError as e:
errors.append(f"Invalid template syntax: {e}")
# Token limits
system_tokens = count_tokens(prompt_config.get("system_prompt", ""))
if system_tokens > 4000:
errors.append(f"System prompt too long: {system_tokens} tokens")
return ValidationResult(
valid=len(errors) == 0,
errors=errors
)
Stage 2: Unit Tests
class PromptUnitTests:
def test_template_rendering(self):
prompt = PromptTemplate(SYSTEM_PROMPT, USER_TEMPLATE)
rendered = prompt.render(
query="test query",
context="test context"
)
assert "test query" in rendered
assert "test context" in rendered
assert len(rendered) < 10000 # Token limit
def test_output_parsing(self):
parser = OutputParser()
valid_output = '{"answer": "test", "confidence": 0.9}'
result = parser.parse(valid_output)
assert result["answer"] == "test"
invalid_output = "not json"
with pytest.raises(ParseError):
parser.parse(invalid_output)
Stage 3: Golden Set Tests
class GoldenSetRunner:
def __init__(self, golden_set: list[dict]):
self.golden_set = golden_set
async def run(self, llm_client) -> TestResults:
results = []
for example in self.golden_set:
response = await llm_client.generate(example["input"])
# Exact match for deterministic outputs
if example.get("exact_match"):
passed = response == example["expected"]
# Contains check for flexible outputs
elif example.get("must_contain"):
passed = all(
phrase in response
for phrase in example["must_contain"]
)
# LLM judge for quality
else:
passed = await self.judge_quality(
response, example["expected"]
)
results.append(TestResult(
input=example["input"],
expected=example["expected"],
actual=response,
passed=passed
))
return TestResults(
total=len(results),
passed=sum(1 for r in results if r.passed),
failed=[r for r in results if not r.passed]
)
Stage 4: LLM Evaluation
class LLMEvaluationStage:
def __init__(self, eval_set: list[dict], sample_rate: float = 0.1):
self.eval_set = eval_set
self.sample_rate = sample_rate
self.evaluator = LLMEvaluator()
async def run(self, llm_client) -> EvalResults:
# Sample for cost efficiency
sample = random.sample(
self.eval_set,
int(len(self.eval_set) * self.sample_rate)
)
scores = []
for example in sample:
response = await llm_client.generate(example["input"])
score = await self.evaluator.evaluate(
query=example["input"],
response=response,
reference=example.get("reference"),
criteria=["relevance", "accuracy", "helpfulness"]
)
scores.append(score)
return EvalResults(
sample_size=len(sample),
avg_relevance=np.mean([s["relevance"] for s in scores]),
avg_accuracy=np.mean([s["accuracy"] for s in scores]),
avg_helpfulness=np.mean([s["helpfulness"] for s in scores])
)
Quality Gates
Gate Configuration
class QualityGate:
def __init__(self, thresholds: dict):
self.thresholds = thresholds
def evaluate(self, results: dict) -> GateResult:
failures = []
# Golden set pass rate
if results["golden_pass_rate"] < self.thresholds["golden_pass_rate"]:
failures.append({
"metric": "golden_pass_rate",
"actual": results["golden_pass_rate"],
"threshold": self.thresholds["golden_pass_rate"]
})
# Quality scores
for metric in ["relevance", "accuracy", "helpfulness"]:
if results.get(f"avg_{metric}", 0) < self.thresholds.get(metric, 0):
failures.append({
"metric": metric,
"actual": results.get(f"avg_{metric}"),
"threshold": self.thresholds[metric]
})
# Regression detection
if results.get("regression_detected"):
failures.append({
"metric": "regression",
"details": results["regression_details"]
})
return GateResult(
passed=len(failures) == 0,
failures=failures
)
# Example thresholds
QUALITY_THRESHOLDS = {
"golden_pass_rate": 0.95, # 95% of golden tests must pass
"relevance": 4.0, # Average score >= 4.0/5.0
"accuracy": 4.0,
"helpfulness": 3.5
}
Deployment Strategies
Canary Deployment
class CanaryDeployer:
def __init__(
self,
initial_percentage: int = 5,
increment: int = 10,
bake_time_minutes: int = 30
):
self.initial_percentage = initial_percentage
self.increment = increment
self.bake_time = bake_time_minutes
async def deploy(self, new_version: str):
# Start canary
await self.router.set_canary(new_version, self.initial_percentage)
percentage = self.initial_percentage
while percentage < 100:
# Wait for bake time
await asyncio.sleep(self.bake_time * 60)
# Check canary health
metrics = await self.get_canary_metrics(new_version)
if not self.is_healthy(metrics):
await self.rollback(new_version)
raise CanaryFailedError(metrics)
# Increment traffic
percentage = min(100, percentage + self.increment)
await self.router.set_canary(new_version, percentage)
# Full rollout
await self.router.promote_canary(new_version)
Shadow Deployment
class ShadowDeployer:
async def shadow_test(
self,
new_version: str,
duration_hours: int = 24
):
# Run new version in shadow mode
await self.enable_shadow(new_version)
# Collect comparison data
start = datetime.now()
while datetime.now() - start < timedelta(hours=duration_hours):
await asyncio.sleep(60)
comparison = await self.compare_outputs()
if comparison["divergence_rate"] > 0.1:
await self.alert("High divergence in shadow test", comparison)
# Analyze results
return await self.generate_comparison_report(new_version)
Rollback Procedures
Automated Rollback
class AutoRollback:
def __init__(self, rollback_thresholds: dict):
self.thresholds = rollback_thresholds
async def monitor_and_rollback(self, version: str):
while True:
metrics = await self.get_live_metrics(version)
# Check error rate
if metrics["error_rate"] > self.thresholds["error_rate"]:
await self.trigger_rollback(version, "error_rate_exceeded")
return
# Check latency
if metrics["p99_latency"] > self.thresholds["p99_latency"]:
await self.trigger_rollback(version, "latency_exceeded")
return
# Check quality (sampled)
if metrics.get("quality_score", 5) < self.thresholds["quality_score"]:
await self.trigger_rollback(version, "quality_degradation")
return
await asyncio.sleep(60)
async def trigger_rollback(self, version: str, reason: str):
previous = await self.get_previous_version()
await self.router.rollback_to(previous)
await self.alert(f"Auto-rollback from {version}: {reason}")
Interview Questions
Q: How do you test prompt changes before production?
Strong answer:
"I use a multi-stage testing pipeline:
Stage 1: Static validation. Syntax check, token limits, template errors. Fast and cheap.
Stage 2: Unit tests. Template rendering, output parsing, deterministic behavior. Still fast.
Stage 3: Golden set tests. Known input/output pairs that must pass. Catches obvious regressions.
Stage 4: LLM evaluation. Sampled evaluation using LLM-as-judge. Measures quality dimensions (relevance, accuracy). More expensive but catches subtle issues.
Quality gates: All stages must pass thresholds. Golden set > 95% pass rate, quality scores > 4.0/5.0.
Deployment: Canary at 5% traffic, bake for 30 minutes, monitor metrics, gradually increase.
The key insight is that LLM outputs are non-deterministic, so testing must be statistical. I cannot guarantee 100% correctness, but I can ensure quality stays within acceptable bounds."
Q: What triggers should cause automatic rollback?
Strong answer:
"I configure multiple rollback triggers:
Error rate
If errors exceed 5% for 5 consecutive minutes, rollback. This catches outright failures.
Latency
If P99 latency exceeds SLA (e.g., 10s) for 10 minutes, rollback. This catches performance regressions.
Quality score
If sampled quality score drops below 3.5/5.0, rollback. This catches subtle quality degradation.
User signals
If negative feedback rate spikes 2x baseline, investigate and potentially rollback.
Implementation:
- Prometheus alerts trigger rollback script
- Automatic notification to team
- Rollback to last known good version
- Block further deploys until investigated
The key is fast detection and action. A bad prompt in production for 10 minutes is acceptable. For 10 hours is not."
References
- ML Ops: https://ml-ops.org/ ml-ops.org
- LangSmith: https://docs.smith.langchain.com/ docs.smith.langchain.com
Previous: LLM Infrastructure
Key takeaways
01
LLM tests are statistical, not binary
Non-deterministic outputs mean the pipeline asks whether quality stayed inside bounds rather than whether every case passed, and prompt, model and index changes each carry different risk.
02
Four stages, cheapest first
Static validation, then unit tests, then golden-set pass/fail, then sampled LLM-as-judge — each stage costs more, so cheap checks catch failures before the expensive one runs.
03
Gates need numbers, not judgment
The worked thresholds are a 95 percent golden-set pass rate with average relevance and accuracy at or above 4.0 out of 5.0, and helpfulness at 3.5.
04
Canary in increments with a bake time
Start at 5 percent of traffic, bake 30 minutes, check health, then add 10 points at a time — any unhealthy reading rolls back before the next increment.
05
Automate the rollback triggers
Error rate over 5 percent for five minutes, P99 past SLA for ten, or sampled quality under 3.5 out of 5.0 should revert without waiting for a human.