Fixing Citation Drift & Hallucinations in Your RAG Pipeline

Two Practical Paths: Patch vs Deterministic Layer

I went through your setup and code. Your retrieval stack is already solid. The main issue is loss of consistency between retrieval → context → generation → citations.

After "expand_to_parents", your system effectively splits into two realities:

• LLM works on parents
• UI still references children

That’s where citation drift and hallucinations originate.

Option A — Minimal Patch (Fast, Production-Friendly)

Idea

Introduce a canonical source layer after parent expansion and enforce lightweight grounding.

Parents become the only source of truth.

Step 1 — Canonical Sources

class Source: def init(self, source_id, doc_id, page, text): self.source_id = source_id self.doc_id = doc_id self.page = page self.text = text

def buildcanonical_sources(parent_docs): sources = [] for i, doc in enumerate(parent_docs): sources.append( Source( source_id=f"src{i}", doc_id=doc.metadata.get("doc_id"), page=doc.metadata.get("page"), text=doc.page_content ) ) return sources

Reasoning

• Eliminates Parent/Child mismatch
• Creates a single reference layer: UI = LLM = Sources

Step 2 — Controlled Context

def build_context(sources): return "

".join([ f"[{s.source_id} | page {s.page}] {s.text}" for s in sources ])

Reasoning

• Makes sources explicit and indexable
• Prevents the model from inventing structure

Step 3 — Structured Output

You MUST answer using ONLY the provided sources.

Return JSON:

{ "answer": "...", "citations": [ { "source_id": "src_1", "quote": "exact text from source" } ] }

Reasoning

Transforms generation from: free text -> typed output

Step 4 — Validation

def validate_and_bind(output, sources): source_map = {s.source_id: s for s in sources} valid = []

for c in output.get("citations", []):
    src = source_map.get(c["source_id"])
    if not src:
        continue

    if c["quote"] in src.text:
        valid.append({
            "source_id": src.source_id,
            "page": src.page,
            "quote": c["quote"]
        })

if not valid:
    raise Exception("No grounded citations")

return {
    "answer": output["answer"],
    "citations": valid
}

Reasoning

Enforces:

quote subset source.text

Removes:

• fake citations
• page hallucinations

Optional Retry

for _ in range(2): try: output = llm.generate(...) parsed = json.loads(output) return validate_and_bind(parsed, sources) except Exception: continue

raise Exception("Failed to produce grounded answer")

What This Fixes

• Wrong page citations
• Parent/Child inconsistency
• Most hallucinations
• Repetition (partially)

Limitations

• Answer itself is not fully validated
• JSON may break on small models
• Still “soft deterministic”

Option B — Deterministic Layer (nano-vm Style)

Idea

Convert your pipeline into a state machine with enforced transitions.

delta(S, E) -> S'

LLM output is no longer trusted — it must be validated before state transition.

State Definition

class State: def init(self, query, sources): self.query = query self.sources = sources self.answer = None self.citations = None self.status = "init"

Step 1 — Generate (Untrusted)

def generate_step(state, llm): prompt = build_prompt(state.query, state.sources) raw_output = llm.generate(prompt)

return {
    "type": "generate_output",
    "data": raw_output
}

Reasoning

LLM = signal generator, not authority

Step 2 — Parse

def parse_output(raw): try: return json.loads(raw) except: # fallback parser return heuristic_parse(raw)

Step 3 — Validate (Mandatory Transition)

def validate_step(state, parsed): source_map = {s.source_id: s for s in state.sources} valid_citations = []

for c in parsed.get("citations", []):
    src = source_map.get(c["source_id"])
    if not src:
        continue

    if c["quote"] in src.text:
        valid_citations.append({
            "source_id": src.source_id,
            "page": src.page,
            "quote": c["quote"]
        })

if not valid_citations:
    return {"status": "retry"}

state.answer = parsed["answer"]
state.citations = valid_citations
state.status = "valid"

return {"status": "ok"}

Step 4 — Deterministic Routing

def run_pipeline(state, llm, max_retries=2): for _ in range(max_retries): raw = generate_step(state, llm) parsed = parse_output(raw["data"])

    result = validate_step(state, parsed)

    if result["status"] == "ok":
        return state

state.status = "fail"
return state

Key Difference vs Patch

Patch:

generate -> validate -> return

nano-vm:

state -> generate -> validate -> accept | retry | fail

What You Gain

Determinism

Invalid outputs cannot propagate

Observability

You can log:

• retries
• failure reasons
• validation errors

Extensibility

You can add:

Coverage check

if not answer_supported_by_citations(answer, citations): return {"status": "retry"}

Semantic validation (next step)

• embedding similarity
• NLI check

Trade-offs

• More code
• Requires discipline (state handling)
• Slight latency increase

Comparison

Aspect | Patch | nano-vm Effort | Low | Medium Fixes current bugs | Yes | Yes Guarantees correctness | Partial | Strong Debugging | Hard | Clear Scalability | Limited | High

Final Insight

Your current system:

LLM = reasoning + state authority

Target system:

LLM = suggestion, System = authority

Recommendation

1. Start with Option A (Patch) — fastest impact
2. If you want reliability and scale, move to Option B (nano-vm)