Answer with Citations

Generation is where RAG demos cheat. They retrieve well and then let the model talk. This part writes a generator that is structurally incapable of using non-retrieved text: the fake model may only copy/summarize strings from the retrieved list. When you later swap an LLM, you keep the same output schema and a grounding check.

Output schema

`` { "answer": str, "citations": list[str], # chunk ids "refused": bool, "retrieved": list[{id, score}] }

Rules:

- `citations` ⊆ retrieved ids for this query.
- If not refused, answer sentences should have an **evidence span** in the concatenation of retrieved texts (same idea as the research agent).
- Mention of numbers: any integer in the answer should appear in retrieved text (`5-7`, `60`, `14`, `15`, `90`, `429`). This kills "refunds in 30 days" when only 5-7 is in context — unless the mug chunk was retrieved.

## Fake generator

If retrieved is empty or best score < tau: refuse.

Else: pick the top chunk and **extract the first sentence** as the answer (extractive QA). Cheap, grounded, slightly ugly. Ugly is honest. An LLM abstractive summary can come later **with the same evidence check**.

For questions that need two facts (rate limit + 429), k=2 extractive can concatenate two sentences. Still only from retrieved text.

import json import math import re

CHUNKS = [ {"id": "chunk-01", "text": "Refunds. Refunds are issued in 5-7 business days to the original payment method. Digital goods are refundable within 14 days of purchase if unused."}, {"id": "chunk-02", "text": "Rate limits. Each API key may make 60 requests per minute. Burst above that returns HTTP 429."}, {"id": "chunk-03", "text": "Privacy. Support will never ask for your password or full API key."}, {"id": "chunk-05", "text": "Mug shop. Acme mugs ship in 30 days. Mug refunds take 90 days."}, ] STOP = set("the a an of to in is for your or".split()) TAU = 0.2

def tokenize(text): return [t for t in re.findall(r"[a-z0-9]+", text.lower()) if len(t) > 1 and t not in STOP]

VOCAB = {tok: i for i, tok in enumerate(sorted({t for c in CHUNKS for t in tokenize(c["text"])}))}

def vectorize(text): v = [0.0] * len(VOCAB) for tok in tokenize(text): if tok in VOCAB: v[VOCAB[tok]] += 1.0 return v

def cosine(a, b): na = math.sqrt(sum(x x for x in a)) nb = math.sqrt(sum(x x for x in b)) if na == 0 or nb == 0: return 0.0 return sum(x y for x, y in zip(a, b)) / (na nb)

INDEX = [(c, vectorize(c["text"])) for c in CHUNKS]

def retrieve(query, k=2): qv = vectorize(query) scored = sorted(((cosine(qv, vec), c) for c, vec in INDEX), key=lambda r: -r[0]) return [{"id": c["id"], "score": round(s, 4), "text": c["text"]} for s, c in scored[:k]]

def first_sentence(text): parts = re.split(r"(?<=\.)\s+", text, maxsplit=1) return parts[0].strip()

def integers_in(text): return set(re.findall(r"\d+", text))

def generate(query, tau=TAU): hits = retrieve(query) if not hits or hits[0]["score"] < tau: return { "answer": "cannot: not in handbook", "citations": [], "refused": True, "retrieved": [{"id": h["id"], "score": h["score"]} for h in hits], } top = hits[0] answer = first_sentence(top["text"]) return { "answer": answer, "citations": [top["id"]], "refused": False, "retrieved": [{"id": h["id"], "score": h["score"]} for h in hits], }

def grounding_ok(payload, hits_by_id): if payload["refused"]: return payload["answer"].startswith("cannot:") blob = " ".join(hits_by_id[i]["text"] for i in payload["citations"] if i in hits_by_id) if not payload["citations"] or payload["answer"] not in blob and payload["answer"] not in blob.replace(" ", " "): # extractive: answer should be a substring of cited chunk(s) cited = " ".join(hits_by_id[i]["text"] for i in payload["citations"] if i in hits_by_id) if payload["answer"] not in cited: return False if not all(n in "".join(hits_by_id[i]["text"] for i in payload["citations"]) for n in integers_in(payload["answer"])): return False return True

for q in ["How long do API refunds take?", "What HTTP code is a rate limit burst?", "equine dental"]: payload = generate(q) hits = retrieve(q) by_id = {h["id"]: h for h in hits} print(json.dumps({"q": q, "payload": payload, "grounded": grounding_ok(payload, by_id)}, indent=2)) print("---")

## Step-by-step generator contract

1. Retrieve first. Never generate with empty context unless refusing.
2. Apply tau on the **best** score.
3. Produce extractive text (or later, LLM text).
4. Cite the chunk you used, not the runner-up, unless you truly merged.
5. Run `grounding_ok` **in the loop**, not in a notebook you forgot.

When you add a real LLM, the prompt says: `Answer only from these chunks. If missing, refuse.` Then **code** still runs `grounding_ok`. Prompts are not enforcement.

## Two-chunk answers

If the user asks "rate limit and what status code?", retrieve k=2, both from the rate-limit section if your chunker split them. Concatenate two extractive sentences. Citations become two ids. Grounding checks both.

> **Note:** Extractive answers can include the heading ("Refunds. Refunds are issued..."). Strip the duplicated heading in a polish pass if you want prettier UX.

## Exercise

Add a generator path: if two hits are both ≥ tau and the query contains `and`, join both first sentences and cite both ids. Test the rate-limit question. Confirm equine still refuses.

Why run grounding_ok in code instead of only prompting "don't hallucinate"?

• Code is slower
• *The model is untrusted; prompts are not enforcement
• Extractive QA is illegal
• Cosine already guarantees the answer text

explain: Retrieval selects context. The generator can still ignore it. Schema and substring checks catch that.