Debug a broken RAG

s1_empty.py# SYMPTOM 1 — "no results / empty". User asks "How do I reset my password?"
# The corpus DOES contain the answer, but the retriever returns nothing.
CORPUS = [
    "To change your account credentials, open Settings then Security.",
    "Billing questions are handled by the finance portal.",
    "Our office hours are 9am to 5pm on weekdays.",
]

def retrieve_broken(query, k=2):
    # BUG: require EVERY query word to appear in the chunk (AND match).
    terms = query.lower().split()
    hits = [c for c in CORPUS if all(t in c.lower() for t in terms)]
    return hits[:k]

print("BROKEN:", retrieve_broken("how do I reset my password"))

# Root cause: the answer chunk shares NO words with the query
# ("reset/password" vs "change/credentials"). AND-matching guarantees zero hits.
# The lazy patch (OR every word) makes "how","do","i" match everything -> noise.
STOP = {"how", "do", "i", "the", "a", "my", "to"}

def retrieve_fixed(query, k=2):
    # FIX: drop stopwords, SCORE by term overlap, never hard-require all terms.
    terms = [t for t in query.lower().split() if t not in STOP]
    scored = []
    for c in CORPUS:
        cl = c.lower()
        score = sum(1 for t in terms if t in cl)
        if score:
            scored.append((score, c))
    scored.sort(key=lambda x: -x[0])
    return [c for _, c in scored[:k]]

print("FIXED (lexical):", retrieve_fixed("password reset settings security"))

s2_wrong.py# SYMPTOM 2 — "retrieves the WRONG thing". Query: "reset the API token".
# The embedding puts the query CLOSEST to a chunk about resetting a *password*
# (semantically adjacent but wrong), beating the chunk that actually documents
# API-token reset. Keyword search, which sees the exact words, fixes it.
import math

# Per-chunk/query embeddings assigned directly so the failure is deterministic
# and inspectable. Axis intuition: [account-security, api-credentials].
Q_VEC = (0.80, 0.60)                                             # "reset the api token"
CORPUS = [   # (text, embedding)
    ("To reset your password, open Settings > Security.",       (0.98, 0.20)),  # distractor
    ("Reset an API token via the developer portal token page.", (0.30, 0.95)),  # THE ANSWER
]

def cos(a, b):
    dot = sum(x*y for x, y in zip(a, b))
    na = math.sqrt(sum(x*x for x in a)); nb = math.sqrt(sum(y*y for y in b))
    return round(dot/(na*nb), 3)

# BUG: semantic ranking. 'password reset' sits at a SMALLER angle to the query
# than the true 'api token' chunk -> the WRONG chunk is retrieved at rank 1.
sem = sorted(((cos(Q_VEC, v), t) for t, v in CORPUS), reverse=True)
print("SEMANTIC top:", sem[0][0], "->", sem[0][1])

def keyword(query):
    terms = query.lower().split()
    N = len(CORPUS)
    df = {t: sum(1 for txt, _ in CORPUS if t in txt.lower()) for t in terms}
    scored = [(round(sum(math.log((N+1)/(df[t]+0.5)) for t in terms if t in txt.lower()), 3), txt)
              for txt, _ in CORPUS]
    return sorted(scored, reverse=True)

# FIX: keyword search matches the rare exact words 'api'+'token' (df=1), which
# only the correct chunk contains -> it wins. In production you FUSE the two (hybrid).
kw = keyword("reset the api token")
print("KEYWORD top:", kw[0][0], "->", kw[0][1])

s3_rank.py# SYMPTOM 3 — "the right chunk IS retrieved, but ranked too low to be used".
# First-stage retrieval returns a wide set; the answer-bearing chunk sits at rank 3.
# Generation only reads the top-1 chunk, so the answer is wrong. A reranker that
# scores query-chunk relevance promotes the right chunk to rank 1.
CANDIDATES = [   # (first_stage_score, text) -- as retrieval returned them
    (0.71, "General security best practices: rotate credentials regularly."),
    (0.68, "Passwords must be at least 12 characters with a symbol."),
    (0.66, "To reset your password: Settings > Security > Reset, then check email."),  # answer
]
QUESTION = "how do I reset my password"

def top1(cands):
    return max(cands, key=lambda c: c[0])[1]

# BUG: feed only the rank-1 chunk to the model. It's the vague one, so the model
# can't answer (or invents). The answer chunk WAS retrieved but never used.
print("USED (no rerank):", top1(CANDIDATES))

def rerank(question, cands):
    # Cheap cross-encoder stand-in: relevance = overlap of meaningful query terms.
    STOP = {"how", "do", "i", "the", "a", "my", "to"}
    q = {w for w in question.lower().split() if w not in STOP}
    scored = [(sum(t in text.lower() for t in q), text) for _, text in cands]
    scored.sort(key=lambda x: -x[0])
    return scored

# FIX: rerank the WHOLE candidate set by real query relevance before selecting.
ranked = rerank(QUESTION, CANDIDATES)
print("RERANKED top:", ranked[0][1])
print("USED (rerank):", ranked[0][1])

s4_prompt.py# SYMPTOM 4 — "the right chunk IS in context, but the answer ignores it / hallucinates".
# This is a GENERATION/PROMPT failure, not retrieval. We stub the LLM as a deterministic
# function so the bug is reproducible offline: a weak prompt lets the model fall back on
# its (stale) prior; a strict grounded prompt forces it to answer from context or refuse.

CONTEXT = "Refunds are processed within 5 business days."   # correct, retrieved chunk
PRIOR   = "Refunds are processed within 30 days."           # model's stale training memory

def llm(prompt, context):
    # Stub 'model': if the prompt does NOT pin it to the context, it answers from PRIOR.
    grounded = "answer only using the context" in prompt.lower()
    if grounded:
        if context.strip():
            return context                       # obeys: quotes the retrieved fact
        return "I don't have that information."   # honest refusal when context is empty
    return PRIOR                                   # weak prompt -> parametric hallucination

WEAK   = "You are a helpful assistant. Question: how long do refunds take?"
STRICT = ("You are a support assistant. Answer only using the context below; "
          "if it is not there, say you don't know.\nQuestion: how long do refunds take?")

# BUG: the correct chunk is right there in context, but the weak prompt lets the
# model answer from memory -> WRONG '30 days'.
print("WEAK prompt   ->", llm(WEAK, CONTEXT))

# FIX: a grounding prompt forces the answer to come from the retrieved context.
print("STRICT prompt ->", llm(STRICT, CONTEXT))

# VERIFY the escape hatch: strict prompt AND empty context -> honest refusal, no invention.
print("STRICT + empty->", llm(STRICT, ""))

s5_isolate.py# SYMPTOM 5 — "the answer is wrong, but WHICH layer failed?" The systematic method:
# test retrieval and generation SEPARATELY against a known expected chunk, so you
# bisect the pipeline instead of guessing. One gold (question, expected_id, expected).

GOLD = {"q": "how do I reset my password",
        "expected_id": "faq#3",
        "must_contain": "Settings > Security"}

# --- stubs you can swap for the real retriever / generator ---
def retrieve(q):
    return [("faq#3", "To reset your password: Settings > Security > Reset."),
            ("faq#1", "Passwords must be 12+ characters.")]

def generate(q, context):
    # grounded stub: answers from context if a chunk is present, else refuses
    return context[0][1] if context else "I don't have that information."

def diagnose(gold, retrieve, generate, k=2):
    hits = retrieve(gold["q"])
    ids = [i for i, _ in hits][:k]
    # LAYER 1 — retrieval: did the expected chunk make the top-k at all?
    if gold["expected_id"] not in ids:
        return f"RETRIEVAL failed: {gold['expected_id']} not in top-{k} {ids} -> fix chunking/embedding/hybrid"
    # LAYER 2 — ranking: is it at rank 1, or buried?
    rank = ids.index(gold["expected_id"]) + 1
    # LAYER 3 — generation: feed ONLY the gold chunk; does the answer use it?
    gold_chunk = [h for h in hits if h[0] == gold["expected_id"]]
    ans = generate(gold["q"], gold_chunk)
    if gold["must_contain"] not in ans:
        return f"GENERATION failed: right chunk in context but answer missing '{gold['must_contain']}' -> fix prompt"
    if rank != 1:
        return f"RANKING weak: correct chunk retrieved at rank {rank} (not 1) -> add a reranker"
    return "ALL LAYERS PASS on this case"

print(diagnose(GOLD, retrieve, generate))

# Inject a RANKING regression: same chunk, but retrieval now returns it at rank 2.
def retrieve_buried(q):
    return [("faq#1", "Passwords must be 12+ characters."),
            ("faq#3", "To reset your password: Settings > Security > Reset.")]
print(diagnose(GOLD, retrieve_buried, generate))