Tutorial 8: Capture and Replay Failures

Contributor
4 days ago
3 min read

A production failure is a free, real-world test case. Capture it, replay it, fix it, add it to your eval set forever.

Step 1: Identify Failure Signals (5 min)

A "failure" is:

User thumbs-down or negative feedback
User retries / rephrases (signal of bad first answer)
Escalation to human
Detected hallucination (didn't match retrieved source)
Crashed/empty response

Each is a signal. Log it.

Step 2: Capture Full Context (10 min)

When a failure happens, save everything needed to reproduce:

def log_failure(event):
    record = {
        "id": event.id,
        "timestamp": event.timestamp,
        "user_input": event.input,
        "system_prompt": event.system_prompt,
        "model": event.model,
        "model_version": event.model_version,
        "rag_chunks": event.rag_chunks,  # if applicable
        "tool_calls": event.tool_calls,
        "output": event.output,
        "failure_signal": event.failure_signal,
        "user_feedback": event.feedback_text,
    }
    save_to_failure_log(record)

Without all this, you can't reproduce.

Step 3: Build a Replay Tool (15 min)

def replay_failure(failure_id):
    record = load_failure(failure_id)
    
    # Re-run with same inputs
    response = call_llm(
        model=record["model"],
        system=record["system_prompt"],
        input=record["user_input"],
        context=record["rag_chunks"],
    )
    
    print("Original output:", record["output"])
    print("Replayed output:", response)
    return response

Does the failure reproduce? Sometimes yes (deterministic failure), sometimes no (noise).

Step 4: Test Mitigations (10 min)

def test_fix(failure_id, new_prompt):
    record = load_failure(failure_id)
    response = call_llm(
        model=record["model"],
        system=new_prompt,
        input=record["user_input"],
        context=record["rag_chunks"],
    )
    return response

Run the failure through a new prompt. Does it work now?

Step 5: Categorize Failures (10 min)

categories = {
    "hallucination": "Output contains false information",
    "format_violation": "Output doesn't match schema",
    "off_topic": "Output unrelated to question",
    "incomplete": "Output truncated or partial",
    "wrong_tone": "Output rude or inappropriate",
    "refusal": "LLM refused to answer (false positive)",
    "tool_misuse": "Wrong tool used",
}

def categorize(failure):
    # Manual or LLM-based categorization
    ...

Categorize so you can address patterns, not just individual cases.

Step 6: Promote to Eval Set (10 min)

For confirmed-reproducible failures:

def promote_to_eval(failure_id):
    failure = load_failure(failure_id)
    
    eval_case = {
        "id": f"prod-{failure_id}",
        "input": failure["user_input"],
        "category": categorize(failure),
        "expected": {
            "must_not_contain": [extract_hallucination(failure)],
            # Or "must_contain", "no_pii", etc.
        },
        "source": "production",
        "added_at": now(),
    }
    
    eval_set.append(eval_case)
    save_eval_set(eval_set)

Eval set grows organically from real failures. Each one prevented forever.

Step 7: Anonymize Personal Info (10 min)

def anonymize(failure):
    # Replace PII before adding to eval
    text = failure["user_input"]
    text = re.sub(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', 'EMAIL', text)
    text = re.sub(r'\b\d{3}-\d{2}-\d{4}\b', 'SSN', text)
    # ... more patterns
    failure["user_input"] = text
    return failure

Don't put customer data in eval set. Anonymize first.

Step 8: Bulk Replay (10 min)

When a new prompt is proposed, replay all known failures:

def bulk_replay(prompt_version):
    failures = load_all_failures(status="open")
    results = []
    for failure in failures:
        new_output = replay_failure_with_prompt(failure["id"], prompt_version)
        fixed = check_if_fixed(failure, new_output)
        results.append({"id": failure["id"], "fixed": fixed})
    
    fixed_count = sum(r["fixed"] for r in results)
    print(f"{fixed_count}/{len(failures)} failures fixed")

Quantify the fix.

Step 9: Track Recurrence (5 min)

When a failure recurs in production, increment its count:

def track_recurrence(failure_event):
    # Match against known failures (semantic similarity or rules)
    similar = find_similar_failures(failure_event)
    if similar:
        similar.count += 1
        similar.last_seen = now()
    else:
        log_new_failure(failure_event)

Frequent failures get priority. Rare/once-only may not be worth chasing.

Step 10: Close the Loop (5 min)

After fixing:

Run the failure replay through new prompt → confirms fix
Add to eval set → prevents regression
Mark failure "fixed" in log
Deploy
Monitor for recurrence

If it recurs: fix again, this time more thoroughly.

What You Just Did

Production failure capture and replay. Real failures become eval cases. Real fixes get validated before deploy.

Common Failure Modes

Lost context. Can't reproduce because logs lack the system prompt or RAG chunks.

No anonymization. PII in eval set; legal risk.

Replay non-determinism. Failure happened with old model version; new version behaves differently.

No prioritization. Treat all failures equally; miss the frequent ones.

Closed loop never run. Fixes deployed without replay verification.

ShiftQuality