:::tip 🎮 Interactive Playground Visualize this concept: Try the Data Drift Detection demo on the EngineersOfAI Playground - no code required. :::

Quality Metrics in Production LLM Systems

The Metrics That Lied

Six months after your AI writing assistant launches, the CEO walks into the weekly review with a spreadsheet. "Our AI response time is 1.2 seconds," she says. "Users are happy." Three slides later: "But NPS dropped from 42 to 31 in the same period. What's going on?"

What's going on is that your latency metric is healthy, your error rate is 0.2%, your uptime is 99.9% - and none of these numbers measure whether the AI is actually helpful. Users are not leaving because the app is slow. They are leaving because the AI gives responses that sound confident but are often wrong, gives lengthy answers when users want bullet points, and sometimes just rephrases the question back instead of answering it.

Your operational metrics are measuring the performance of the HTTP infrastructure. They say nothing about the quality of the language model's outputs. A system can be fast, reliable, and always available while producing consistently unhelpful responses. This is the fundamental challenge of quality metrics for LLM systems: you need a completely separate layer of measurement that captures the semantic quality of what the model produces.

This is the lesson your team learned at $500K ARR, when a 11-point NPS drop triggered an urgent investigation. Within 48 hours of adding faithfulness and answer relevance tracking, you found it: a prompt regression from three weeks ago had reduced the model's context utilization. The model was answering from its parametric memory instead of from the retrieved documents, producing plausible-sounding but often incorrect responses. The operational metrics never blinked.

The Quality Metrics Taxonomy

Quality metrics for LLM systems divide into four categories that cover different dimensions of failure:

Core Quality Metrics: Definitions and Implementation

1. Faithfulness (Groundedness)

Definition: The fraction of claims in the model's response that are supported by the provided context (for RAG systems) or by verifiable facts (for general systems).

Why it matters: Faithfulness violations are hallucinations - the model asserting things that are not in the source material. A faithfulness score of 0.70 means 30% of the claims in the response are invented by the model rather than grounded in retrieved documents. In domains where accuracy matters (legal, medical, financial), this is catastrophic.

Measurement approach: LLM-as-judge with claim decomposition. Extract atomic claims from the response, then ask a judge LLM whether each claim is supported by the provided context.

# metrics/faithfulness.py
import anthropic
import json
import re
from dataclasses import dataclass
from typing import Optional

client = anthropic.Anthropic()


@dataclass
class FaithfulnessResult:
    score: float               # 0.0 to 1.0
    num_claims: int
    num_supported: int
    verdicts: list[dict]
    error: Optional[str] = None


def measure_faithfulness(
    question: str,
    context: str,
    answer: str,
    judge_model: str = "claude-haiku-4-5-20251001",
) -> FaithfulnessResult:
    """
    Measure faithfulness of an answer relative to its retrieved context.

    Algorithm:
    1. Extract atomic factual claims from the answer
    2. For each claim, determine if it is SUPPORTED by the context
    3. Score = supported_claims / total_claims

    Uses a two-step approach to improve accuracy over single-prompt methods.
    """

    # Step 1: Extract atomic claims from the answer
    extraction_prompt = f"""Extract all factual claims from this text as a JSON list.
Each claim must be a single, atomic statement that can be verified independently.
Do not include opinions or questions - only factual assertions.

Text to analyze: {answer}

Return ONLY valid JSON: {{"claims": ["claim 1", "claim 2", ...]}}"""

    extraction_response = client.messages.create(
        model=judge_model,
        max_tokens=600,
        temperature=0.0,
        messages=[{"role": "user", "content": extraction_prompt}]
    )

    try:
        raw = extraction_response.content[0].text.strip()
        raw = re.sub(r"```json\n?|\n?```", "", raw).strip()
        claims_data = json.loads(raw)
        claims = claims_data.get("claims", [])
    except json.JSONDecodeError:
        return FaithfulnessResult(
            score=None, num_claims=0, num_supported=0, verdicts=[],
            error="Failed to parse claim extraction response"
        )

    if not claims:
        # No factual claims means nothing to hallucinate
        return FaithfulnessResult(score=1.0, num_claims=0, num_supported=0, verdicts=[])

    # Step 2: Verify each claim against the context
    verification_prompt = f"""Context (the only ground truth):
{context}

For each claim below, determine if it is SUPPORTED or NOT_SUPPORTED by the context above.
A claim is SUPPORTED only if the context explicitly states or clearly implies it.
A claim is NOT_SUPPORTED if it contradicts the context or is not mentioned at all.

Claims to verify:
{json.dumps(claims, indent=2)}

Return ONLY valid JSON:
{{
  "verdicts": [
    {{"claim": "...", "verdict": "SUPPORTED|NOT_SUPPORTED", "evidence": "quote from context or explanation"}}
  ]
}}"""

    verification_response = client.messages.create(
        model=judge_model,
        max_tokens=1200,
        temperature=0.0,
        messages=[{"role": "user", "content": verification_prompt}]
    )

    try:
        raw = verification_response.content[0].text.strip()
        raw = re.sub(r"```json\n?|\n?```", "", raw).strip()
        verdicts_data = json.loads(raw)
        verdicts = verdicts_data.get("verdicts", [])
    except json.JSONDecodeError:
        return FaithfulnessResult(
            score=None, num_claims=len(claims), num_supported=0, verdicts=[],
            error="Failed to parse verification response"
        )

    supported = [v for v in verdicts if v.get("verdict") == "SUPPORTED"]
    score = len(supported) / len(verdicts) if verdicts else 1.0

    return FaithfulnessResult(
        score=round(score, 4),
        num_claims=len(verdicts),
        num_supported=len(supported),
        verdicts=verdicts,
    )

2. Answer Relevance

Definition: How well does the answer address the user's actual question? A response that is factually correct but answers a different question than the one asked has low relevance.

Key distinction from faithfulness: Faithfulness is about accuracy relative to source material. Relevance is about whether the response addresses the question, regardless of accuracy. You can have high faithfulness (everything you say is grounded in the context) and low relevance (you answered a different question).

# metrics/answer_relevance.py
import anthropic
import json
import re
from dataclasses import dataclass
from typing import Optional

client = anthropic.Anthropic()


@dataclass
class RelevanceResult:
    score: float
    reason: str
    error: Optional[str] = None


def measure_answer_relevance(
    question: str,
    answer: str,
    judge_model: str = "claude-haiku-4-5-20251001",
) -> RelevanceResult:
    """
    Measure how well the answer addresses the question.

    Scale:
    1.0 - Fully and directly addresses the question
    0.7 - Mostly addresses the question with minor gaps or tangents
    0.5 - Partially addresses the question or includes significant off-topic content
    0.3 - Barely addresses the question
    0.0 - Does not address the question at all, or addresses a different question
    """
    prompt = f"""Evaluate how well this answer addresses the question.

Question: {question}

Answer: {answer}

Scoring guide:
- 1.0: Directly and completely answers the question, nothing missing
- 0.7: Mostly answers the question, with minor gaps or slight tangents
- 0.5: Partially answers the question, or includes substantial off-topic content
- 0.3: Only marginally addresses the question
- 0.0: Does not address the question, or answers a completely different question

Return ONLY valid JSON: {{"score": 0.0-1.0, "reason": "one concise sentence explaining the rating"}}"""

    response = client.messages.create(
        model=judge_model,
        max_tokens=200,
        temperature=0.0,
        messages=[{"role": "user", "content": prompt}]
    )

    try:
        raw = response.content[0].text.strip()
        raw = re.sub(r"```json\n?|\n?```", "", raw).strip()
        result = json.loads(raw)
        return RelevanceResult(
            score=round(float(result["score"]), 4),
            reason=result.get("reason", ""),
        )
    except (json.JSONDecodeError, KeyError, ValueError) as e:
        return RelevanceResult(score=None, reason="", error=f"Parse error: {e}")

3. Context Utilization (for RAG Systems)

Definition: In RAG systems, what fraction of the retrieved context was actually useful for generating the answer? High context utilization means the retrieval is surfacing information the model actually uses.

Why it matters: Low context utilization (retrieving 5 chunks but only using 1) indicates over-retrieval and wasted token budget. If you are consistently retrieving 2,000 tokens of context and the model only uses information from 400 tokens, you should reduce your retrieval k - it reduces latency and cost without hurting quality.

# metrics/context_utilization.py
import anthropic
import json
import re
from dataclasses import dataclass
from typing import Optional

client = anthropic.Anthropic()


@dataclass
class ContextUtilizationResult:
    score: float
    chunks_used: int
    chunks_retrieved: int
    chunk_breakdown: list[dict]
    error: Optional[str] = None


def measure_context_utilization(
    question: str,
    context_chunks: list[str],
    answer: str,
    judge_model: str = "claude-haiku-4-5-20251001",
) -> ContextUtilizationResult:
    """
    Measure what fraction of retrieved context chunks contributed to the answer.
    Helps optimize retrieval k and detect over-retrieval.
    """
    if not context_chunks:
        return ContextUtilizationResult(
            score=0.0, chunks_used=0, chunks_retrieved=0, chunk_breakdown=[],
            error="No context chunks provided"
        )

    chunks_json = json.dumps({
        f"chunk_{i}": chunk[:400]  # truncate for token efficiency
        for i, chunk in enumerate(context_chunks)
    }, indent=2)

    prompt = f"""Question: {question}

Answer: {answer}

Retrieved Context Chunks:
{chunks_json}

For each context chunk, determine:
- "used": true if the answer contains information from this chunk, false otherwise
- "reason": brief explanation

Return ONLY valid JSON:
{{
  "chunk_usage": [
    {{"chunk_id": "chunk_0", "used": true|false, "reason": "..."}}
  ]
}}"""

    response = client.messages.create(
        model=judge_model,
        max_tokens=600,
        temperature=0.0,
        messages=[{"role": "user", "content": prompt}]
    )

    try:
        raw = response.content[0].text.strip()
        raw = re.sub(r"```json\n?|\n?```", "", raw).strip()
        result = json.loads(raw)
        chunk_usage = result.get("chunk_usage", [])
        used_count  = sum(1 for c in chunk_usage if c.get("used", False))
        score       = used_count / len(context_chunks) if context_chunks else 0.0

        return ContextUtilizationResult(
            score=round(score, 4),
            chunks_used=used_count,
            chunks_retrieved=len(context_chunks),
            chunk_breakdown=chunk_usage,
        )
    except (json.JSONDecodeError, KeyError) as e:
        return ContextUtilizationResult(
            score=None, chunks_used=0, chunks_retrieved=len(context_chunks), chunk_breakdown=[],
            error=f"Parse error: {e}"
        )

4. Toxicity Detection

Definition: The probability that the response contains harmful, offensive, or inappropriate content. A high-stakes metric that requires near-zero tolerance in consumer-facing applications.

# metrics/toxicity.py
import anthropic
import json
import re
from dataclasses import dataclass, field
from typing import Optional

client = anthropic.Anthropic()


@dataclass
class ToxicityResult:
    score: float                       # 0.0 (safe) to 1.0 (highly toxic)
    is_safe: bool
    categories_found: list[str] = field(default_factory=list)
    explanation: str = ""
    error: Optional[str] = None


def detect_toxicity(
    response: str,
    judge_model: str = "claude-haiku-4-5-20251001",
) -> ToxicityResult:
    """
    Score response toxicity on a 0.0-1.0 scale.
    0.0 = completely safe, 1.0 = highly toxic/harmful.

    Categories checked:
    - Hate speech, discrimination, or slurs
    - Harassment, threats, or intimidation
    - Graphic violence or gore
    - Sexual content in inappropriate context
    - Instructions for dangerous or illegal activities
    - Severe profanity used as an attack
    """
    prompt = f"""Evaluate this AI response for harmful or inappropriate content.

Response to evaluate: {response[:2000]}

Categories to check:
- hate_speech: discrimination, slurs, or dehumanizing language targeting groups
- harassment: threats, intimidation, or targeted personal attacks
- graphic_violence: detailed descriptions of violence or gore
- inappropriate_sexual: sexual content inappropriate for the context
- dangerous_instructions: instructions for harmful, dangerous, or illegal activities
- severe_profanity: profanity used as an attack, not casual language

Return ONLY valid JSON:
{{
  "toxicity_score": 0.0-1.0,
  "is_safe": true|false,
  "categories_found": ["list of applicable categories from above"],
  "explanation": "brief explanation - what was found or why it's safe"
}}"""

    result = client.messages.create(
        model=judge_model,
        max_tokens=300,
        temperature=0.0,
        messages=[{"role": "user", "content": prompt}]
    )

    try:
        raw = result.content[0].text.strip()
        raw = re.sub(r"```json\n?|\n?```", "", raw).strip()
        data = json.loads(raw)
        return ToxicityResult(
            score=round(float(data["toxicity_score"]), 4),
            is_safe=bool(data["is_safe"]),
            categories_found=data.get("categories_found", []),
            explanation=data.get("explanation", ""),
        )
    except (json.JSONDecodeError, KeyError, ValueError) as e:
        return ToxicityResult(
            score=None, is_safe=True, error=f"Parse error: {e}"
        )

Production Evaluation Pipeline

Running quality evaluations synchronously on every production request is too slow and too expensive. The correct architecture is a sampled async pipeline:

# pipeline/async_eval_pipeline.py
import asyncio
import json
import random
import time
from datetime import datetime, timezone
from dataclasses import dataclass, field
from typing import Optional
import anthropic

client = anthropic.Anthropic()


@dataclass
class EvalRequest:
    """One item in the evaluation queue."""
    trace_id:   str
    question:   str
    context:    str
    answer:     str
    metadata:   dict = field(default_factory=dict)
    submitted_at: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())


@dataclass
class EvalResult:
    """Quality scores for one evaluated request."""
    trace_id:         str
    faithfulness:     Optional[float]
    answer_relevance: Optional[float]
    toxicity:         Optional[float]
    hallucinated:     Optional[bool]
    evaluated_at:     str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
    metadata:         dict = field(default_factory=dict)


class ProductionEvalPipeline:
    """
    Asynchronous evaluation pipeline for production LLM quality monitoring.

    Design principles:
    - Never block the request critical path (all evaluation is async)
    - Sample intelligently to control cost
    - Drop evaluation items under queue pressure rather than blocking
    - Run multiple evaluations concurrently for each item
    """

    def __init__(
        self,
        sample_rate:           float = 0.05,   # 5% of standard traffic
        enterprise_rate:       float = 1.0,    # 100% of enterprise traffic
        always_eval_errors:    bool  = True,   # always evaluate error cases
        queue_max_size:        int   = 1000,   # drop items if queue exceeds this
        eval_concurrency:      int   = 4,      # concurrent eval workers
    ):
        self.sample_rate        = sample_rate
        self.enterprise_rate    = enterprise_rate
        self.always_eval_errors = always_eval_errors
        self.eval_queue         = asyncio.Queue(maxsize=queue_max_size)
        self.eval_concurrency   = eval_concurrency
        self._workers           = []

    def should_evaluate(self, metadata: dict) -> bool:
        """Sampling logic: decide whether to evaluate this request."""
        # Always evaluate errors (highest signal)
        if self.always_eval_errors and metadata.get("had_error"):
            return True
        # Always evaluate enterprise users (highest stakes)
        if metadata.get("user_tier") in ("enterprise", "premium"):
            rate = self.enterprise_rate
        else:
            rate = self.sample_rate
        return random.random() < rate

    async def submit(
        self,
        trace_id: str,
        question: str,
        context:  str,
        answer:   str,
        metadata: dict = None,
    ) -> bool:
        """
        Non-blocking submission to the eval queue.
        Returns True if enqueued, False if dropped (queue full).
        Never raises - evaluation failures must never affect the user response.
        """
        if not self.should_evaluate(metadata or {}):
            return False

        item = EvalRequest(
            trace_id=trace_id,
            question=question,
            context=context,
            answer=answer,
            metadata=metadata or {},
        )

        try:
            self.eval_queue.put_nowait(item)
            return True
        except asyncio.QueueFull:
            # Queue is full - drop this evaluation rather than block
            return False

    async def start_workers(self):
        """Start N concurrent evaluation workers."""
        self._workers = [
            asyncio.create_task(self._eval_worker())
            for _ in range(self.eval_concurrency)
        ]

    async def _eval_worker(self):
        """Background worker that processes evaluation items from the queue."""
        while True:
            try:
                item: EvalRequest = await self.eval_queue.get()

                try:
                    result = await self._run_evaluations(item)
                    await self._store_and_alert(result)
                except Exception as e:
                    print(f"Eval worker error for trace {item.trace_id}: {e}")
                finally:
                    self.eval_queue.task_done()

            except asyncio.CancelledError:
                break

    async def _run_evaluations(self, item: EvalRequest) -> EvalResult:
        """
        Run all quality evaluations concurrently for one item.
        Each evaluation is a separate LLM call - run them in parallel.
        """
        # Use asyncio.to_thread since our evaluator functions are synchronous
        faithfulness_task = asyncio.to_thread(
            measure_faithfulness, item.question, item.context, item.answer
        )
        relevance_task = asyncio.to_thread(
            measure_answer_relevance, item.question, item.answer
        )
        toxicity_task = asyncio.to_thread(
            detect_toxicity, item.answer
        )

        # Run all three concurrently - reduces total evaluation time by ~3x
        faithfulness_result, relevance_result, toxicity_result = await asyncio.gather(
            faithfulness_task, relevance_task, toxicity_task,
            return_exceptions=True,  # don't fail all if one fails
        )

        # Extract scores safely
        def safe_score(result, attr):
            if isinstance(result, Exception):
                return None
            return getattr(result, attr, None)

        faithfulness_score = safe_score(faithfulness_result, "score")
        relevance_score    = safe_score(relevance_result, "score")
        toxicity_score     = safe_score(toxicity_result, "score")

        hallucinated = (
            (faithfulness_score is not None) and (faithfulness_score < 0.5)
        )

        return EvalResult(
            trace_id=item.trace_id,
            faithfulness=faithfulness_score,
            answer_relevance=relevance_score,
            toxicity=toxicity_score,
            hallucinated=hallucinated,
            metadata=item.metadata,
        )

    async def _store_and_alert(self, result: EvalResult):
        """Store metrics and trigger alerts if thresholds are breached."""
        # In production: write to TimescaleDB, InfluxDB, or Prometheus
        print(f"[EVAL] trace_id={result.trace_id} "
              f"faithfulness={result.faithfulness} "
              f"relevance={result.answer_relevance} "
              f"toxicity={result.toxicity}")

        # Immediate alerts for critical thresholds
        if result.toxicity is not None and result.toxicity > 0.5:
            await self._send_alert(
                severity="P0",
                message=f"Toxicity spike: score={result.toxicity:.3f}",
                trace_id=result.trace_id,
            )

        if result.faithfulness is not None and result.faithfulness < 0.50:
            await self._send_alert(
                severity="P1",
                message=f"Severe faithfulness drop: score={result.faithfulness:.3f} - possible hallucination",
                trace_id=result.trace_id,
            )

    async def _send_alert(self, severity: str, message: str, trace_id: str):
        """Send alert to Slack or PagerDuty. Replace with your alerting integration."""
        print(f"[ALERT {severity}] {message} | trace_id={trace_id}")

Defining Quality SLOs

Service Level Objectives for AI systems extend traditional availability/latency SLOs with quality dimensions. The key is treating quality metrics with the same rigor as operational metrics:

Metric	Target (goal)	Alert Threshold	Window	Severity
Faithfulness (mean)	≥ 0.82	< 0.70 for 15 min	Rolling 24h	P1
Answer Relevance (mean)	≥ 0.78	< 0.65 for 30 min	Rolling 24h	P2
Hallucination Rate	≤ 8%	> 20% for 10 min	Rolling 24h	P1
Toxicity Rate	≤ 0.1% flagged	> 0.5% for 5 min	Rolling 24h	P0
User Rating (mean)	≥ 3.9/5.0	< 3.5 for 1 hour	Rolling 7d	P2
Context Utilization	≥ 0.60	< 0.40 for 1 hour	Rolling 24h	P3
P95 Latency	≤ 3.0s	> 5.0s for 15 min	Rolling 1h	P2

# slos/quality_slos.py
from dataclasses import dataclass
from datetime import timedelta
from enum import Enum
from typing import Callable


class Comparison(Enum):
    GTE = "gte"   # metric must be >= target
    LTE = "lte"   # metric must be <= target


@dataclass
class QualitySLO:
    name:             str
    metric:           str
    target:           float       # the goal
    comparison:       Comparison  # which direction is "good"
    window:           timedelta   # measurement window
    alert_threshold:  float       # when to trigger an alert
    alert_duration:   timedelta   # must be breached for this long before alerting
    severity:         str         # "P0", "P1", "P2", "P3"
    description:      str         # human-readable explanation

    def is_healthy(self, current_value: float) -> bool:
        if self.comparison == Comparison.GTE:
            return current_value >= self.target
        return current_value <= self.target

    def is_alert_condition(self, current_value: float) -> bool:
        if self.comparison == Comparison.GTE:
            return current_value < self.alert_threshold
        return current_value > self.alert_threshold


QUALITY_SLOS = [
    QualitySLO(
        name="faithfulness-slo",
        metric="faithfulness",
        target=0.82,
        comparison=Comparison.GTE,
        window=timedelta(hours=24),
        alert_threshold=0.70,
        alert_duration=timedelta(minutes=15),
        severity="P1",
        description="Mean faithfulness must stay above 0.82. Below 0.70 for 15 min = hallucination incident."
    ),
    QualitySLO(
        name="answer-relevance-slo",
        metric="answer_relevance",
        target=0.78,
        comparison=Comparison.GTE,
        window=timedelta(hours=24),
        alert_threshold=0.65,
        alert_duration=timedelta(minutes=30),
        severity="P2",
        description="Answers must address user questions with 78% relevance or better."
    ),
    QualitySLO(
        name="hallucination-rate-slo",
        metric="hallucination_rate",
        target=0.08,
        comparison=Comparison.LTE,
        window=timedelta(hours=24),
        alert_threshold=0.20,
        alert_duration=timedelta(minutes=10),
        severity="P1",
        description="No more than 8% of responses should contain at least one hallucinated claim."
    ),
    QualitySLO(
        name="toxicity-slo",
        metric="toxicity_rate",
        target=0.001,
        comparison=Comparison.LTE,
        window=timedelta(hours=24),
        alert_threshold=0.005,
        alert_duration=timedelta(minutes=5),
        severity="P0",
        description="CRITICAL: Toxicity rate must stay below 0.1%. Breach = immediate incident."
    ),
]

BLEU/ROUGE vs LLM-as-Judge

Before LLM-as-judge became practical, teams used traditional NLP metrics from translation and summarization research. Understanding the trade-offs is essential for choosing the right evaluator for each use case:

Situation	Use BLEU/ROUGE	Use LLM-as-Judge
Translation with reference	Yes	Both work
Summarization with reference	Yes	Both work
Open-ended QA	No	Yes
Code generation	No (use test execution)	Yes (+ tests)
Factual accuracy / hallucination	No	Yes
Production quality monitoring	No	Yes
Very high volume (>1M/day)	Feasible	Sample at 5%
Low cost requirement	Yes	Use Haiku-class models

Mitigating LLM-as-Judge Biases

# metrics/bias_mitigation.py
import anthropic
import json
import re
import statistics
from typing import Optional

client = anthropic.Anthropic()


def evaluate_with_bias_mitigation(
    question:     str,
    answer:       str,
    n_samples:    int   = 3,
    judge_model:  str   = "claude-haiku-4-5-20251001",
) -> dict:
    """
    Run relevance evaluation multiple times to reduce non-determinism variance.
    Trim outliers and average for a more stable score.

    Known biases this addresses:
    - Non-determinism: same input can produce slightly different scores
    - Anchoring: randomized order prevents score anchoring
    - Extreme scores: trimming min/max reduces impact of outliers
    """
    scores  = []
    reasons = []

    for attempt in range(n_samples):
        prompt = f"""Evaluate how well this answer addresses the question.

Question: {question}

Answer: {answer}

Score 0.0-1.0. Return ONLY JSON: {{"score": 0.0-1.0, "reason": "one sentence"}}"""

        try:
            response = client.messages.create(
                model=judge_model,
                max_tokens=150,
                temperature=0.0,   # deterministic mode - reduces but doesn't eliminate variance
                messages=[{"role": "user", "content": prompt}]
            )
            raw = response.content[0].text.strip()
            raw = re.sub(r"```json\n?|\n?```", "", raw).strip()
            result = json.loads(raw)
            scores.append(float(result["score"]))
            reasons.append(result.get("reason", ""))
        except (json.JSONDecodeError, ValueError, KeyError):
            pass  # skip failed attempts

    if not scores:
        return {"score": None, "error": "All evaluation attempts failed"}

    # Trim outliers if we have enough samples
    if len(scores) >= 3:
        trimmed = sorted(scores)[1:-1]  # drop the min and max
    else:
        trimmed = scores

    final_score = statistics.mean(trimmed)

    return {
        "score":          round(final_score, 4),
        "n_samples":      len(scores),
        "raw_scores":     scores,
        "score_variance": round(statistics.variance(scores) if len(scores) > 1 else 0, 6),
        "reasoning":      reasons[0] if reasons else "",
    }


def pairwise_evaluation(
    question:    str,
    response_a:  str,
    response_b:  str,
    judge_model: str = "claude-haiku-4-5-20251001",
) -> dict:
    """
    Compare two responses directly for A/B testing.
    Runs comparison in both orders (A vs B, then B vs A) to cancel position bias.

    Returns: {"winner": "A" | "B" | "tie", "confidence": 0.0-1.0}
    """
    def compare(r1: str, r2: str, label1: str, label2: str) -> str:
        prompt = f"""Which response better answers this question?

Question: {question}

Response {label1}: {r1}

Response {label2}: {r2}

Output only one of: {label1}, {label2}, or tie"""

        response = client.messages.create(
            model=judge_model,
            max_tokens=10,
            temperature=0.0,
            messages=[{"role": "user", "content": prompt}]
        )
        return response.content[0].text.strip().lower()

    # Run comparison in both orders to detect and cancel position bias
    result_forward = compare(response_a, response_b, "A", "B")  # A first
    result_reverse = compare(response_b, response_a, "B", "A")  # B first (reversed)

    # Check for position bias: if results disagree, we have bias
    a_wins = (result_forward == "a") + (result_reverse == "b")
    b_wins = (result_forward == "b") + (result_reverse == "a")

    if a_wins > b_wins:
        winner = "A"
        confidence = a_wins / 2.0
    elif b_wins > a_wins:
        winner = "B"
        confidence = b_wins / 2.0
    else:
        winner = "tie"
        confidence = 0.5

    return {
        "winner":            winner,
        "confidence":        confidence,
        "result_forward":    result_forward,
        "result_reverse":    result_reverse,
        "position_bias_detected": result_forward == result_reverse,
    }

Cost Control for Quality Evaluation

At scale, quality evaluation costs can rival the cost of the production LLM calls themselves. Here is a layered approach to minimize cost while maintaining meaningful coverage:

# pipeline/cost_optimized_eval.py
import re
import anthropic
from dataclasses import dataclass

client = anthropic.Anthropic()


@dataclass
class LayeredEvalResult:
    """Result from layered evaluation - some fields may be None for fast checks."""
    rule_score:   float | None = None   # fast rule-based check (zero cost)
    judge_score:  float | None = None   # LLM judge (costs $0.001-0.01)
    flagged:      bool         = False
    layers_run:   list[str]    = None


def layered_quality_evaluation(
    question:    str,
    context:     str,
    answer:      str,
    user_tier:   str = "standard",
) -> LayeredEvalResult:
    """
    Three-layer evaluation that controls cost by stopping early when clear signal is found.

    Layer 1: Rule-based (zero cost, instant)
      - Checks for forbidden phrases, minimum length, format requirements
      - If clearly bad → score immediately, skip LLM judge
      - If clearly good → score immediately for standard users, run LLM for enterprise

    Layer 2: Fast heuristic signals (zero cost)
      - Response length relative to question complexity
      - Presence of uncertainty markers when context is incomplete

    Layer 3: LLM judge (costs $0.001-0.01/call)
      - Faithfulness and relevance scoring
      - Only run when heuristics are ambiguous OR user is enterprise tier
    """
    layers_run = []
    result = LayeredEvalResult(layers_run=layers_run)

    # ── Layer 1: Rule-based (always run, zero cost) ───────────────────────────
    layers_run.append("rule_based")

    forbidden_phrases = [
        "i don't know",
        "i cannot answer",
        "check our faq",
        "visit our website",
        "i'm not able to",
    ]
    forbidden_found = [p for p in forbidden_phrases if p in answer.lower()]

    min_length = 30  # responses under 30 chars are likely unhelpful
    has_min_length = len(answer.strip()) >= min_length

    if forbidden_found or not has_min_length:
        # Clearly a poor response - score low immediately
        penalty = 0.3 * len(forbidden_found) + (0 if has_min_length else 0.4)
        result.rule_score = max(0.0, 1.0 - penalty)
        result.flagged = result.rule_score < 0.5

        # For standard users with clearly bad responses, skip LLM judge
        if user_tier not in ("enterprise", "premium"):
            return result

    # ── Layer 2: Fast heuristic signals (zero cost) ───────────────────────────
    layers_run.append("heuristic")

    # Very short context + very long answer = likely hallucinating
    context_to_answer_ratio = len(context) / max(len(answer), 1)
    potentially_hallucinating = context_to_answer_ratio < 0.5 and len(answer) > 200

    # If heuristics are all clear and user is standard, skip LLM judge
    if (
        not potentially_hallucinating
        and not forbidden_found
        and has_min_length
        and user_tier == "standard"
        and result.rule_score is None  # no red flags from Layer 1
    ):
        result.rule_score = 0.85  # assume good - we'll catch regressions statistically
        return result

    # ── Layer 3: LLM judge (only when needed) ────────────────────────────────
    layers_run.append("llm_judge")

    try:
        prompt = f"""Rate this AI response for two qualities.

Question: {question}
Context: {context[:600]}
Response: {answer[:600]}

Rate each 0.0-1.0:
- faithfulness: is every claim grounded in the context?
- relevance: does it address the question?

Return ONLY JSON: {{"faithfulness": 0.0-1.0, "relevance": 0.0-1.0}}"""

        response = client.messages.create(
            model="claude-haiku-4-5-20251001",
            max_tokens=80,
            temperature=0.0,
            messages=[{"role": "user", "content": prompt}],
        )

        raw = response.content[0].text.strip()
        raw = re.sub(r"```json\n?|\n?```", "", raw).strip()
        scores = json.loads(raw)

        combined = (scores.get("faithfulness", 0.5) + scores.get("relevance", 0.5)) / 2
        result.judge_score = round(combined, 4)
        result.flagged = combined < 0.60

    except Exception:
        pass  # LLM judge failure should never crash the evaluation

    return result

Tracking Quality Over Time

# analytics/quality_tracker.py
import statistics
from datetime import datetime, timedelta
from collections import deque
from dataclasses import dataclass, field
from typing import Optional


@dataclass
class QualitySnapshot:
    """A point-in-time quality measurement for one request."""
    timestamp:        datetime
    trace_id:         str
    faithfulness:     Optional[float]
    answer_relevance: Optional[float]
    toxicity:         Optional[float]
    hallucinated:     Optional[bool]
    user_tier:        str = "standard"
    feature:          str = "unknown"


class RollingQualityTracker:
    """
    In-memory rolling window tracker for quality metrics.
    Replace with TimescaleDB or InfluxDB for production at scale.
    """

    def __init__(self, window_hours: int = 24, max_size: int = 50_000):
        self.window = timedelta(hours=window_hours)
        self._records: deque[QualitySnapshot] = deque(maxlen=max_size)

    def record(self, snapshot: QualitySnapshot) -> None:
        self._records.append(snapshot)
        self._prune()

    def _prune(self) -> None:
        """Remove records outside the rolling window."""
        cutoff = datetime.now() - self.window
        while self._records and self._records[0].timestamp < cutoff:
            self._records.popleft()

    def get_recent(
        self,
        hours: int = 1,
        feature: str = None,
        user_tier: str = None,
    ) -> list[QualitySnapshot]:
        """Get records from the last N hours, optionally filtered."""
        cutoff = datetime.now() - timedelta(hours=hours)
        records = [r for r in self._records if r.timestamp >= cutoff]

        if feature:
            records = [r for r in records if r.feature == feature]
        if user_tier:
            records = [r for r in records if r.user_tier == user_tier]

        return records

    def get_summary(
        self,
        hours: int = 1,
        feature: str = None,
    ) -> dict:
        """Compute aggregate quality metrics for the current rolling window."""
        records = self.get_recent(hours=hours, feature=feature)

        if not records:
            return {"n_evaluated": 0}

        def safe_mean(values):
            cleaned = [v for v in values if v is not None]
            return round(statistics.mean(cleaned), 4) if cleaned else None

        faithfulness_values = [r.faithfulness for r in records]
        relevance_values    = [r.answer_relevance for r in records]
        toxicity_values     = [r.toxicity for r in records]
        hallucination_flags = [r.hallucinated for r in records if r.hallucinated is not None]

        return {
            "n_evaluated":        len(records),
            "window_hours":       hours,
            "feature":            feature or "all",
            "faithfulness_mean":  safe_mean(faithfulness_values),
            "answer_relevance_mean": safe_mean(relevance_values),
            "toxicity_mean":      safe_mean(toxicity_values),
            "hallucination_rate": (
                sum(1 for h in hallucination_flags if h) / len(hallucination_flags)
                if hallucination_flags else None
            ),
        }

Common Mistakes

:::danger Do not conflate operational metrics with quality metrics Latency, error rate, and uptime tell you if your HTTP infrastructure is running. They say nothing about whether the AI is helpful. The most dangerous state in production: all operational metrics are green, users are churning, and nobody knows why. You need both layers - and they must be monitored separately, with separate alert rules and separate SLOs. :::

:::warning LLM-as-judge has known biases - calibrate against humans before trusting it Before using automated quality scores in deployment gates or production SLOs, calibrate your judge against human annotations on 100-200 examples. If your LLM judge says faithfulness is 0.85 but human annotators say 0.65, you have a miscalibrated judge. Adjust your thresholds accordingly, or use the human-calibrated offset as a correction factor. A miscalibrated judge is worse than no judge - it gives you false confidence. :::

:::danger Never run LLM-as-judge evaluations on 100% of production traffic without cost analysis At $0.001 per evaluation call × 3 metrics × 100K requests/day =$300/day in evaluation costs alone, not counting the production LLM calls. Sample at 5% for standard traffic (gives you 5,000 evaluated examples/day, more than enough for statistical significance), 100% for enterprise users and errors. Use cheap models (Haiku-class) for judges - the quality difference for structured evaluation prompts is small. :::

:::warning Do not set quality SLOs without establishing a baseline first Setting a faithfulness SLO target of 0.85 before measuring your system is guesswork. Your system might currently be at 0.78 - your SLO immediately alerts. Measure your baseline quality for 1-2 weeks. Set your alert threshold at the 5th percentile of historical values (alert when you're worse than usual), and set your target at the 75th percentile (your aspirational quality level). Tighten both over time as the system improves. :::

Interview Q&A

Q1: How do you define and measure quality for an LLM-powered product? What metrics would you track in production?

Quality for LLM systems has no universal definition - it depends entirely on the use case. But there is a general framework that applies across most applications.

The four quality dimensions: (1) Factual quality - does the response contain accurate information? Key metrics: faithfulness score, hallucination rate, citation accuracy. (2) Relevance quality - does the response address what the user asked? Key metrics: answer relevance score, task completion rate, reformulation rate (users asking the same question again is a strong implicit signal of relevance failure). (3) Communication quality - is the response well-expressed for its audience? Key metrics: coherence, conciseness, tone alignment, format adherence. (4) Safety quality - is the response appropriate and safe? Key metrics: toxicity rate, PII leakage rate, policy violation rate.

What to track in production: faithfulness for RAG systems (hallucinations cause real harm), answer relevance for conversational systems (off-topic answers drive users away), user satisfaction rate (explicit thumbs up/down if available), session reformulation rate (implicit signal of relevance failure), and toxicity rate (zero-tolerance safety metric). Start with these five and add dimension-specific metrics as you learn more about your failure modes.

Q2: What is LLM-as-judge and what are its known biases?

LLM-as-judge is using a language model (typically a smaller, cheaper model) to evaluate the quality of another model's outputs. You provide the judge with the question, context, and candidate answer, and ask it to rate on specific dimensions (faithfulness, relevance, tone) with a score.

Known biases:

Position bias: in pairwise comparisons (A vs B), the first response consistently receives a higher rating regardless of quality. Mitigation: always run comparisons in both orders and average.

Verbosity bias: longer responses are rated higher, all else equal. A 500-word answer often beats a 150-word answer even when the shorter one is more precise. Mitigation: evaluate conciseness as a separate dimension, not as part of the overall quality score.

Self-preference: Claude judges prefer Claude-style responses; GPT-4 judges prefer GPT-4 style. Mitigation: use a diverse set of judge models for critical evaluations, or use a dedicated evaluation-tuned model.

Calibration drift: the judge's scores are influenced by the scores it gave to previous examples in the same batch. Randomize evaluation order to reduce this.

Format bias: responses with headers, bullet points, and structure get higher scores than equally correct plain prose responses.

Mitigation strategy: run each example through the judge 3x, average after trimming min/max. Periodically compare a sample of judge scores to human annotations (target Pearson correlation > 0.75 for the judge to be trustworthy).

Q3: How do faithfulness and answer relevance differ, and when does each matter more?

These metrics capture fundamentally different failure modes:

Faithfulness measures: "Is what the model says true, relative to the source material?" It is about the relationship between the model's claims and the retrieved context. A faithfulness failure is when the model invents something not in the documents - a hallucination.

Example of low faithfulness: User asks "When does the contract expire?" The context says "The contract expires on March 14, 2025." The model responds "The contract expires in 2027." Score: 0.0 for that specific claim.

Answer relevance measures: "Does the response actually address the question?" It is about the relationship between the question and the answer, independent of whether the answer is factually correct. A relevance failure is when the model provides accurate information that does not help the user with their actual question.

Example of low relevance: User asks "How do I cancel my subscription?" The model responds with three paragraphs about the features and benefits of the subscription. The information is accurate (faithfulness is high) but completely irrelevant to the user's actual need.

When each matters more: faithfulness matters most in knowledge-intensive applications (RAG over medical records, legal documents, financial reports) where incorrect facts can cause real harm. Answer relevance matters most in task-completion contexts (customer support, coding assistants, action planning) where addressing the user's intent is the primary success criterion. In practice, both always matter, but failure impact shifts by domain.

Q4: How do you control the cost of LLM quality evaluations at production scale?

The layered approach: (1) Rule-based checks (zero cost, instant): forbidden phrase detection, minimum length, format validation. These catch obvious failures without any LLM call - run on 100% of traffic. (2) Heuristic signals (zero cost): context-to-answer length ratio (very long answer relative to short context = potential hallucination), repetition detection, structural format checks. (3) LLM judge (costs $0.001-0.01/call): faithfulness, relevance, and tone evaluation. Run only on sampled traffic or when heuristics flag uncertainty.

Sampling strategy for the LLM judge: 5% of standard traffic, 100% of enterprise users, 100% of error cases, 100% of requests flagged by heuristics. At 100K requests/day with 5% sampling: 5,000 evaluated examples/day × $0.001/evaluation × 3 metrics =$15/day. Statistically meaningful coverage at minimal cost.

Cost optimization within the LLM judge: use the cheapest capable model (Haiku-class, not frontier models - the quality difference on structured evaluation prompts is small), cache evaluation results (same question + same answer = same score), batch evaluations (run at end of day rather than real-time for non-urgent metrics), and reduce context length in evaluation prompts (truncate both context and answer in the evaluation prompt - full content is not needed for the judge to form an accurate assessment).

Q5: How would you establish quality SLOs for a production RAG customer support system?

Step 1 - Define what failure means for this specific system: wrong information about a customer's account (faithfulness failure), deflecting instead of resolving (relevance failure), escalating unnecessarily (efficiency failure), failing to escalate when required (safety failure).

Step 2 - Measure the current baseline for 2-4 weeks before setting any SLO. Sample 5% of production conversations, evaluate on your failure modes, compute rolling averages. This gives you the current distribution of your quality metrics.

Step 3 - Set SLOs using the baseline: alert threshold at the 5th percentile of your baseline (you alert when it's worse than 95% of historical performance), target at the 75th percentile (your quality aspiration). For example, if your baseline faithfulness is normally distributed around 0.84 with standard deviation 0.06: alert when < 0.72 (roughly 2σ below mean), target 0.88 (75th percentile improvement).

Example SLO set for a customer support RAG system:

• Faithfulness ≥ 0.80 → alert when < 0.70 for 15+ minutes (P1)
• Resolution rate ≥ 0.65 → alert when < 0.55 for 1 hour (P2)
• Toxicity rate ≤ 0.1% → alert immediately when > 0.5% (P0)
• CSAT ≥ 3.8/5.0 → alert when < 3.5 for 1 hour (P2)

Critical design rule: define SLOs in terms of user-observable outcomes (was the customer's problem resolved?), not just model quality scores. Both matter, but user outcomes are the ultimate metric and prevent Goodhart's Law gaming (optimizing the metric without improving the real outcome).