How to detect AI hallucinations in production

Detecting AI hallucinations in production requires more than vibe checks. You need a layered approach: define your failure modes, build ground truth, run reference-based and reference-free checks offline, then sample live traffic with automated alerts. This guide walks through each layer with code you can adapt.

Step 1: Define what counts as a hallucination for your use case

Hallucination is not a single phenomenon. A medical summarizer that invents a dosage is catastrophic; a creative writing assistant that invents a character backstory is expected. Before instrumenting, write down the specific claim types your system must get right.

Create a taxonomy document (Markdown or YAML) that your evaluation pipeline can reference:

# hallucination_taxonomy.yaml
claim_types:
  - name: factual_entity
    severity: critical
    examples:
      - "Patient prescribed 50mg lisinopril"  # must match source
      - "Revenue was $4.2B in Q3"              # must match source
  - name: logical_inference
    severity: high
    examples:
      - "Since revenue grew, profitability improved"  # verify from context
  - name: style_tone
    severity: low
    examples:
      - "Response uses bullet points as requested"
  - name: creative_extrapolation
    severity: none
    examples:
      - "The dragon's scales shimmered like amethyst"

Load this in your evaluation code to tag and weight findings:

# taxonomy.py
from dataclasses import dataclass
from enum import Enum
import yaml

class Severity(Enum):
    CRITICAL = 3
    HIGH = 2
    LOW = 1
    NONE = 0

@dataclass
class ClaimType:
    name: str
    severity: Severity
    examples: list[str]

def load_taxonomy(path: str) -> dict[str, ClaimType]:
    with open(path) as f:
        raw = yaml.safe_load(f)
    return {
        ct["name"]: ClaimType(
            name=ct["name"],
            description=ct["description"],
            severity=Severity[ct["severity"].upper()],
            examples=ct.get("examples", []),
        )
        for ct in raw["claim_types"]
    }

Verify: Your taxonomy covers every output field that downstream systems consume. Review it with product and legal stakeholders quarterly.

Step 2: Build a ground truth evaluation set

You cannot measure what you have not labeled. Curate a representative dataset of (input, expected_output, context) triples. Aim for 200–500 examples spanning your claim types, including edge cases and known failure modes.

Store as JSONL for pipeline compatibility:

{"id": "med_001", "input": "Summarize: Patient John Doe, 54, prescribed lisinopril 20mg daily for hypertension.", "context": "Patient John Doe, 54, prescribed lisinopril 20mg daily for hypertension.", "expected": "Patient John Doe, 54, takes lisinopril 20mg daily for hypertension.", "claim_types": ["factual_entity"]}
{"id": "med_002", "input": "Summarize: Patient Jane Smith, 62, prescribed metformin 500mg BID for type 2 diabetes.", "context": "Patient Jane Smith, 62, prescribed metformin 500mg BID for type 2 diabetes.", "expected": "Patient Jane Smith, 62, takes metformin 500mg twice daily for type 2 diabetes.", "claim_types": ["factual_entity"]}
{"id": "fin_001", "input": "Extract revenue from: Q3 revenue was $4.2B, up 12% YoY.", "context": "Q3 revenue was $4.2B, up 12% YoY.", "expected": "Revenue: $4.2B", "claim_types": ["factual_entity"]}

Version this dataset with your model artifacts. Tag each example with the claim types from Step 1 so you can slice metrics later.

# eval_dataset.py
from dataclasses import dataclass
from typing import Optional
import json

@dataclass
class EvalExample:
    id: str
    input: str
    context: str
    expected: str
    claim_types: list[str]

def load_eval_set(path: str) -> list[EvalExample]:
    examples = []
    with open(path) as f:
        for line in f:
            data = json.loads(line)
            examples.append(EvalExample(**data))
    return examples

Verify: Run your current model against this set. You should see non-zero failure rates on at least two claim types. If everything passes, your set is too easy — add adversarial examples.

Step 3: Implement reference-based detection

When ground truth or a trusted knowledge base exists, compare model claims against it. This is the highest-signal layer. Two practical patterns: retrieval-augmented verification and structured extraction matching.

3A: Retrieval-augmented verification

For RAG systems, verify each generated claim against the retrieved chunks that fed the prompt. Use an entailment model (e.g., microsoft/deberta-v3-large-mnli or a smaller distilled variant) to classify each claim as supported, contradicted, or neutral.

# reference_check.py
from typing import Literal
from transformers import pipeline
import re

EntailmentLabel = Literal["supported", "contradicted", "neutral"]

class ReferenceChecker:
    def __init__(self, model_name: str = "microsoft/deberta-v3-large-mnli"):
        self.nli = pipeline("text-classification", model=model_name, tokenizer=model_name, device=0)
    
    def split_claims(self, text: str) -> list[str]:
        """Naive sentence splitter — replace with spaCy or your segmenter."""
        return [s.strip() for s in re.split(r'(?<=[.!?])\s+', text) if s.strip()]
    
    def check_claim(self, claim: str, evidence: str) -> EntailmentLabel:
        result = self.nli(f"{evidence} [SEP] {claim}", top_k=3)
        # Map NLI labels to our taxonomy
        label_map = {"ENTAILMENT": "supported", "CONTRADICTION": "contradicted", "NEUTRAL": "neutral"}
        return label_map[result[0]["label"]]
    
    def verify(self, generated: str, context: str) -> dict:
        claims = self.split_claims(generated)
        results = []
        for claim in claims:
            label = self.check_claim(claim, context)
            results.append({"claim": claim, "label": label})
        return {
            "claims": results,
            "supported_ratio": sum(1 for r in results if r["label"] == "supported") / len(results) if results else 1.0,
            "has_contradiction": any(r["label"] == "contradicted" for r in results),
        }

3B: Structured extraction matching

When outputs follow a schema (JSON, function calls, tables), parse both expected and generated outputs and compare field-by-field. This catches entity-level hallucinations that sentence-level NLI misses.

# structured_check.py
from typing import Any
from dataclasses import dataclass
from difflib import SequenceMatcher

@dataclass
class FieldResult:
    field: str
    expected: Any
    generated: Any
    match: bool
    similarity: float

def compare_structured(expected: dict, generated: dict, fields: list[str]) -> list[FieldResult]:
    results = []
    for field in fields:
        exp_val = expected.get(field)
        gen_val = generated.get(field)
        if exp_val is None and gen_val is None:
            match, sim = True, 1.0
        elif exp_val is None or gen_val is None:
            match, sim = False, 0.0
        elif isinstance(exp_val, str) and isinstance(gen_val, str):
            sim = SequenceMatcher(None, exp_val.lower(), gen_val.lower()).ratio()
            match = sim > 0.9  # threshold per field type
        else:
            match = exp_val == gen_val
            sim = 1.0 if match else 0.0
        results.append(FieldResult(field, exp_val, gen_val, match, sim))
    return results

Verify: Run your reference checks against the eval set from Step 2. You should achieve >95% recall on factual_entity claim types (i.e., catch nearly all injected errors). If recall is low, your evidence retrieval or claim splitter needs work.

Step 4: Implement reference-free detection

Reference-based checks require ground truth or retrieved context. In production, you often have neither. Reference-free methods trade precision for coverage. Deploy three complementary signals:

4A: Self-consistency sampling

Generate multiple responses at temperature > 0. If claims diverge, the model is uncertain. This is expensive — run on a sampled subset.

# consistency.py
from collections import Counter
from typing import Literal
import numpy as np

ConsistencyLabel = Literal["consistent", "divergent", "uncertain"]

def self_consistency_check(
    prompt: str,
    generate_fn,  # callable(prompt, temperature) -> str
    n_samples: int = 5,
    temperature: float = 0.7,
    claim_splitter=None,
) -> dict:
    """Returns consistency score per claim."""
    if claim_splitter is None:
        claim_splitter = lambda t: [s.strip() for s in t.split(". ") if s.strip()]
    
    samples = [generate_fn(prompt, temperature) for _ in range(n_samples)]
    all_claims = [claim_splitter(s) for s in samples]
    
    # Align claims by position (naive; improve with semantic matching)
    max_claims = max(len(c) for c in all_claims)
    claim_votes = []
    for i in range(max_claims):
        variants = [c[i] for c in all_claims if i < len(c)]
        if not variants:
            continue
        # Cluster similar claims
        clusters = []
        for v in variants:
            matched = False
            for cluster in clusters:
                if SequenceMatcher(None, v.lower(), cluster[0].lower()).ratio() > 0.85:
                    cluster.append(v)
                    matched = True
                    break
            if not matched:
                clusters.append([v])
        # Score: size of largest cluster / total samples
        largest = max(len(c) for c in clusters)
        claim_votes.append(largest / n_samples)
    
    avg_consistency = np.mean(claim_votes) if claim_votes else 1.0
    return {
        "per_claim_consistency": claim_votes,
        "avg_consistency": avg_consistency,
        "label": "consistent" if avg_consistency > 0.8 else "divergent" if avg_consistency > 0.5 else "uncertain",
    }

4B: Token-level uncertainty via logprobs

Models assign low probability to hallucinated tokens. If your provider returns logprobs (OpenAI, Anthropic, most open models via vLLM/TGI), compute per-token and aggregate uncertainty.

# uncertainty.py
import math
from dataclasses import dataclass

@dataclass
class UncertaintyResult:
    token_entropies: list[float]
    mean_entropy: float
    max_entropy: float
    high_entropy_spans: list[tuple[int, int]]  # (start, end) token indices

def compute_uncertainty(logprobs: list[dict], entropy_threshold: float = 2.0) -> UncertaintyResult:
    """
    logprobs: list of {token: str, logprob: float, top_logprobs: list[{token, logprob}]}
    """
    entropies = []
    for lp in logprobs:
        # Approximate entropy from top-k logprobs
        probs = [math.exp(p["logprob"]) for p in lp.get("top_logprobs", [])]
        # Renormalize (top-k may not sum to 1)
        total = sum(probs)
        if total > 0:
            probs = [p / total for p in probs]
            entropy = -sum(p * math.log(p + 1e-10) for p in probs)
        else:
            entropy = 0.0
        entropies.append(entropy)
    
    # Find contiguous high-entropy spans
    spans = []
    in_span = False
    start = 0
    for i, e in enumerate(entropies):
        if e > entropy_threshold and not in_span:
            in_span = True
            start = i
        elif e <= entropy_threshold and in_span:
            in_span = False
            spans.append((start, i))
    if in_span:
        spans.append((start, len(entropies)))
    
    return UncertaintyResult(
        token_entropies=entropies,
        mean_entropy=sum(entropies) / len(entropies) if entropies else 0.0,
        max_entropy=max(entropies) if entropies else 0.0,
        high_entropy_spans=spans,
    )

4C: Self-verification prompt

Ask the model to critique its own output. This catches different errors than consistency or logprobs. Use a structured prompt that forces a verdict.

# self_verify.py
from typing import Literal

VerificationLabel = Literal["accurate", "inaccurate", "unsure"]

SELF_VERIFY_PROMPT = """You are a careful fact-checker. Given a CONTEXT and a CLAIM, determine if the claim is fully supported by the context.

CONTEXT:
{context}

CLAIM:
{claim}

Respond with ONLY one word: ACCURATE, INACCURATE, or UNSURE.
"""

async def self_verify(claim: str, context: str, generate_fn) -> VerificationLabel:
    prompt = SELF_VERIFY_PROMPT.format(context=context, claim=claim)
    response = await generate_fn(prompt, temperature=0.0, max_tokens=3)
    label = response.strip().upper()
    if label in ("ACCURATE", "INACCURATE", "UNSURE"):
        return label.lower()  # type: ignore
    return "unsure"

Verify: On your eval set, measure precision/ inject known hallucinations and confirm:

• Self-consistency flags >80% of divergent claims
• Mean entropy correlates with error (AUC > 0.75)
• Self-verification catches errors the other two miss (complementary recall)

Step 5: Deploy online monitoring with sampling

You cannot run full reference checks on 100% of traffic. Sample strategically: 100% of high-severity claim types (via fast heuristic filters), 1–5% of everything else.

5A: Fast heuristic pre-filter

Before expensive checks, route traffic through cheap classifiers that flag likely hallucinations.

# sampler.py
import random
import hashlib
from dataclasses import dataclass
from typing import Callable

@dataclass
class SamplingDecision:
    sample: bool
    reason: str
    checks: list[str]  # which detectors to run

class AdaptiveSampler:
    def __init__(
        self,
        base_rate: float = 0.02,
        high_severity_rate: float = 1.0,
        heuristic_filters: dict[str, Callable[[str], bool]] | None = None,
    ):
        self.base_rate = base_rate
        self.high_severity_rate = high_severity_rate
        self.heuristic_filters = heuristic_filters or {}
    
    def decide(self, request_id: str, output: str, metadata: dict) -> SamplingDecision:
        # Deterministic sampling by request_id for reproducibility
        hash_val = int(hashlib.md5(request_id.encode()).hexdigest(), 16)
        base_sample = (hash_val % 10000) / 10000 < self.base_rate
        
        # Check heuristic filters (e.g., contains number, cites source, medical terms)
        triggered_filters = [
            name for name, fn in self.heuristic_filters.items()
            if fn(output)
        ]
        
        if triggered_filters:
            return SamplingDecision(
                sample=True,
                reason=f"heuristic:{','.join(triggered_filters)}",
                checks=["reference", "consistency", "uncertainty", "self_verify"],
            )
        
        if base_sample:
            return SamplingDecision(
                sample=True,
                reason="base_rate",
                checks=["consistency", "uncertainty"],  # cheaper subset
            )
        
        return SamplingDecision(sample=False, reason="not_sampled", checks=[])

Example heuristic filters:

# heuristics.py
import re

HEURISTIC_FILTERS = {
    "contains_number": lambda t: bool(re.search(r'\b\d+(\.\d+)?\b', t)),
    "contains_citation": lambda t: bool(re.search(r'\[\d+\]|\(\w+ et al\.', t)),
    "medical_terms": lambda t: bool(re.search(r'\b(mg|ml|dosage|prescribed|diagnosis)\b', t, re.I)),
    "financial_terms": lambda t: bool(re.search(r'\b(\$|revenue|profit|EBITDA|YoY)\b', t, re.I)),
    "high_entropy_proxy": lambda t: len(set(t.split())) / max(len(t.split()), 1) > 0.7,  # diverse vocab
}

5B: Async detection pipeline

Run selected checks asynchronously. Write results to a time-series DB (ClickHouse, TimescaleDB, or even Postgres with partitioning) for dashboards and alerting.

# pipeline.py
import asyncio
from dataclasses import dataclass, asdict
from typing import Awaitable
import time

@dataclass
class DetectionResult:
    request_id: str
    timestamp: float
    model: str
    claim_type: str
    reference_check: dict | None = None
    consistency_check: dict | None = None
    uncertainty_check: dict | None = None
    self_verify_check: dict | None = None
    overall_label: str = "unknown"
    severity_score: float = 0.0

class DetectionPipeline:
    def __init__(
        self,
        reference_checker,
        consistency_checker,
        uncertainty_checker,
        self_verify_checker,
        claim_classifier,  # maps output -> claim_type from taxonomy
        db_writer,         # async write function
    ):
        self.ref = reference_checker
        self.cons = consistency_checker
        self.unc = uncertainty_checker
        self.sv = self_verify_checker
        self.classify = claim_classifier
        self.write = db_writer
    
    async def process(
        self,
        request_id: str,
        prompt: str,
        output: str,
        context: str | None,
        model: str,
        checks: list[str],
    ):
        claim_type = self.classify(output)
        start = time.time()
        
        tasks = {}
        if "reference" in checks and context:
            tasks["reference"] = asyncio.create_task(self._run_ref(output, context))
        if "consistency" in checks:
            tasks["consistency"] = asyncio.create_task(self._run_cons(prompt, output))
        if "uncertainty" in checks:
            tasks["uncertainty"] = asyncio.create_task(self._run_unc(output))
        if "self_verify" in checks and context:
            tasks["self_verify"] = asyncio.create_task(self._run_sv(output, context))
        
        results = {}
        for name, task in tasks.items():
            try:
                results[name] = await task
            except Exception as e:
                results[name] = {"error": str(e)}
        
        overall, severity = self._aggregate(results, claim_type)
        
        record = DetectionResult(
            request_id=request_id,
            timestamp=start,
            model=model,
            claim_type=claim_type,
            reference_check=results.get("reference"),
            consistency_check=results.get("consistency"),
            uncertainty_check=results.get("uncertainty"),
            self_verify_check=results.get("self_verify"),
            overall_label=overall,
            severity_score=severity,
        )
        await self.write(asdict(record))
    
    async def _run_ref(self, output: str, context: str):
        return self.ref.verify(output, context)
    
    async def _run_cons(self, prompt: str, output: str):
        # Requires a generate_fn — pass via closure or partial
        return {"placeholder": "wire your generate_fn"}
    
    async def _run_unc(self, output: str):
        return {"placeholder": "requires logprobs from provider"}
    
    async def _run_sv(self, output: str, context: str):
        claims = self.ref.split_claims(output)
        verdicts = await asyncio.gather(*[self.sv(c, context) for c in claims])
        return {"claims": list(zip(claims, verdicts))}
    
    def _aggregate(self, results: dict, claim_type: str) -> tuple[str, float]:
        # Simple weighted voting — replace with your calibrated model
        score = 0.0
        weight = 0.0
        if "reference" in results and results["reference"]:
            r = results["reference"]
            if r.get("has_contradiction"):
                score += 3.0
            weight += 3.0
        if "consistency" in results and results["consistency"]:
            c = results["consistency"]
            score += (1.0 - c.get("avg_consistency", 1.0)) * 2.0
            weight += 2.0
        if "uncertainty" in results and results["uncertainty"]:
            u = results["uncertainty"]
            score += min(u.get("mean_entropy", 0.0) / 3.0, 1.0) * 1.5
            weight += 1.5
        if "self_verify" in results and results["self_verify"]:
            sv = results["self_verify"]
            inaccurate = sum(1 for _, v in sv.get("claims", []) if v == "inaccurate")
            score += min(inaccurate / max(len(sv.get("claims", [])), 1), 1.0) * 2.0
            weight += 2.0
        
        normalized = score / weight if weight > 0 else 0.0
        if normalized > 0.7:
            return "hallucinated", normalized
        elif normalized > 0.3:
            return "suspect", normalized
        return "clean", normalized

Verify: Deploy to a canary (1–5% of traffic). Confirm:

• Pipeline latency p99 < 500ms added
• Write throughput handles your peak QPS
• Dashboard shows non-zero suspect and hallucinated rates within 24h

Step 6: Close the loop with alerting and feedback

Detection without response is noise. Wire three response paths:

6A: Real-time alerting for critical claim types

# alerts.py
from dataclasses import dataclass
from enum import Enum

class AlertChannel(Enum):
    PAGERDUTY = "pagerduty"
    SLACK = "slack"
    EMAIL = "email"

@dataclass
class AlertRule:
    name: str
    condition: str  # SQL-like or PromQL
    severity: str
    channels: list[AlertChannel]
    cooldown_minutes: int = 60

CRITICAL_RULES = [
    AlertRule(
        name="medical_dosage_hallucination",
        condition="claim_type='factual_entity' AND overall_label='hallucinated' AND output LIKE '%mg%'",
        severity="critical",
        channels=[AlertChannel.PAGERDUTY],
        cooldown_minutes=5,
    ),
    AlertRule(
        name="financial_figure_hallucination",
        condition="claim_type='factual_entity' AND overall_label='hallucinated' AND output LIKE '%$%'",
        severity="high",
        channels=[AlertChannel.SLACK],
        cooldown_minutes=15,
    ),
    AlertRule(
        name="hallucination_rate_spike",
        condition="rate(hallucinated_total[5m]) > 0.05",  # >5% of sampled traffic
        severity="high",
        channels=[AlertChannel.SLACK, AlertChannel.EMAIL],
        cooldown_minutes=30,
    ),
]

6B: Automated fallback for user-facing requests

If you detect a hallucination before returning to the user (async detection is too slow for this; use the fast heuristic + a lightweight reference check), trigger a fallback: retry with higher context, switch to a larger model, or return a refusal.

# fallback.py
from typing import Literal

FallbackAction = Literal["retry_with_context", "escalate_model", "refuse", "original"]

async def maybe_fallback(
    detection: DetectionResult,
    original_output: str,
    generate_fn,
    context: str,
    max_retries: int = 1,
) -> tuple[str, FallbackAction]:
    if detection.overall_label != "hallucinated":
        return original_output, "original"
    
    if detection.claim_type == "factual_entity" and detection.severity_score > 0.8:
        # Try once with explicit "cite your sources" instruction
        retry_prompt = f"{context}\n\nAnswer the question. Cite specific sentences from the context for every claim."
        retry_output = await generate_fn(retry_prompt, temperature=0.0)
        # Quick re-check
        recheck = await reference_checker.verify(retry_output, context)
        if not recheck["has_contradiction"]:
            return retry_output, "retry_with_context"
    
    return "I cannot confidently answer based on the provided information.", "refuse"

6C: Continuous improvement loop

Feed confirmed hallucinations back into your eval set (Step 2) and fine-tuning data. Tag each with the detector that caught it.

# feedback_loop.py
import json
from datetime import datetime

def export_for_retraining(detections: list[DetectionResult], output_path: str):
    """Write confirmed hallucinations as training examples."""
    with open(output_path, "a") as f:
        for d in detections:
            if d.overall_label == "hallucinated" and d.severity_score > 0.6:
                example = {
                    "prompt": d.prompt,  # you'll need to store this
                    "completion": d.expected_correction,  # requires human label or model-generated correction
                    "metadata": {
                        "detected_by": [k for k, v in {
                            "reference": d.reference_check,
                            "consistency": d.consistency_check,
                            "uncertainty": d.uncertainty_check,
                            "self_verify": d.self_verify_check,
                        }.items() if v and not v.get("error")],
                        "claim_type": d.claim_type,
                        "severity": d.severity_score,
                        "timestamp": d.timestamp,
                    }
                }
                f.write(json.dumps(example) + "\n")

Verify: After one month:

• Alert volume is actionable (not paging on noise)
• Fallback triggers on <0.1% of requests but catches >50% of critical hallucinations
• Retraining set grows by 50–200 examples/month per model
• Hallucination rate on eval set drops 20%+ quarter over quarter

Summary checklist

Start with Steps 1–2 this week. Ship the heuristic sampler and one reference check to canary next week. Add reference-free layers once you have baseline metrics. The goal is not zero hallucinations — it is knowing exactly when and where they happen, and having a mechanical response ready.