Implementing Cross-Document Reasoning: Building Multi-Hop RAG Pipelines That Synthesize Evidence Across Source Boundaries

Introduction

Here's a question that breaks most RAG systems: "How has the competitive pressure from Company X changed the strategic direction described in our Q3 board memo, given the market analysis in the industry report we commissioned last month?"

A standard RAG pipeline retrieves the top-k chunks most semantically similar to that question, hands them to an LLM, and hopes for the best. But the answer to this question doesn't live in any single chunk. It requires locating the board memo's strategic claims, finding the relevant sections of the industry report, identifying Company X's competitive moves across multiple source documents, and then reasoning across all three to synthesize a coherent answer. Each hop through the evidence chain depends on what the previous hop found. That's multi-hop reasoning — and it's where commodity RAG pipelines fall apart.

By the end of this lesson, you'll understand exactly why naive RAG fails at cross-document reasoning, and you'll have built a complete multi-hop RAG pipeline that plans retrieval iteratively, maintains an evidence buffer, synthesizes across source boundaries, and produces answers with proper attribution. This is genuinely hard engineering. We're going to cover the theory, the data structures, the prompting strategies, and the edge cases that only show up in production.

What you'll learn:

• Why single-pass retrieval fails for multi-hop questions and how to diagnose the failure modes
• How to decompose complex questions into retrieval sub-queries using an LLM planner
• How to implement an iterative retrieval loop with an evidence buffer that accumulates context across hops
• How to design the synthesis prompt so the LLM reasons over evidence rather than pattern-matches to individual chunks
• How to track provenance across hops and surface confidence-weighted citations in the final answer
• How to avoid the common traps: over-retrieval, circular reasoning, and evidence hallucination

---

Prerequisites

This lesson assumes you're comfortable with the following:

• Building basic RAG pipelines with vector embeddings (if not, start with Building a Production Document Q&A System with Vector Embeddings)
• Using the OpenAI or Anthropic API to make chat completions calls (Using the OpenAI API with Python is a good refresher)
• Python async/await patterns and basic data class usage
• Familiarity with chunking, embedding, and vector similarity search (Chunking Strategies for RAG covers this well)

You should also understand hybrid search for RAG — multi-hop pipelines benefit enormously from precise keyword matching at each hop, not just semantic similarity.

---

Why Single-Pass RAG Fails at Multi-Hop Questions

Before we build anything, let's be precise about the failure mode. When you submit a multi-hop question to a standard RAG pipeline, several things go wrong simultaneously.

The semantic averaging problem. Your embedding model produces a single vector representing the entire query. A question like "How did the FDA's 2023 guidance on GLP-1 agonists affect the R&D budget reallocation mentioned in Novo Nordisk's Q4 earnings call?" gets compressed into one point in embedding space. That point might not be close to either the FDA guidance chunks or the earnings call chunks, because no single chunk contains the intersection of both topics. You retrieve chunks that are individually relevant but collectively insufficient.

The missing bridge problem. Even if you retrieve some chunks from each source, the reasoning bridge between them — the causal chain that explains how one thing affects another — lives in the LLM's reasoning process, not in any document. If the LLM doesn't have the right context from both sources simultaneously, it will either hallucinate the connection or refuse to answer.

The fixed-context problem. Standard RAG retrieves a fixed set of chunks before the LLM sees anything. But for multi-hop questions, the right chunks to retrieve in step 2 depend on what you learned in step 1. You can't know in advance what the second hop should look for.

The solution to all three problems is the same: make retrieval iterative and conditional. Each retrieval step is informed by the results of the previous one, and the system accumulates evidence in a buffer until it has enough to synthesize an answer.

---

Designing the Architecture

Let's establish the components we're building before writing any code:

1. Query Decomposer: An LLM call that takes the original question and produces an ordered list of sub-queries, each targeting a specific piece of evidence
2. Evidence Buffer: A data structure that accumulates retrieved chunks across hops, tagged with their source document and hop number
3. Retrieval Engine: A hybrid search layer that accepts a sub-query (and optionally, evidence from prior hops as context) and returns relevant chunks
4. Hop Controller: The loop logic that decides whether to continue retrieving or move to synthesis
5. Synthesis Engine: An LLM call that reasons over the complete evidence buffer to produce the final answer with citations

Here's the core data model:

from dataclasses import dataclass, field
from typing import Optional
import time

@dataclass
class EvidenceChunk:
    """A single retrieved piece of evidence from a specific source."""
    chunk_id: str
    document_id: str
    document_title: str
    content: str
    relevance_score: float
    hop_number: int
    sub_query_that_retrieved_it: str
    retrieved_at: float = field(default_factory=time.time)

@dataclass
class HopResult:
    """The outcome of a single retrieval hop."""
    hop_number: int
    sub_query: str
    chunks: list[EvidenceChunk]
    reasoning: str  # Why the planner issued this sub-query
    sufficient_for_answer: bool = False

@dataclass
class MultiHopResult:
    """The final output of the full multi-hop pipeline."""
    original_question: str
    answer: str
    hops: list[HopResult]
    evidence_buffer: list[EvidenceChunk]
    confidence: float
    citations: list[dict]

This structure is deliberate. By tracking hop_number and sub_query_that_retrieved_it on every chunk, you maintain a complete audit trail of how the pipeline arrived at each piece of evidence. This isn't just nice-to-have — it's essential for debugging when the pipeline produces wrong answers, and it's what enables proper citation attribution.

---

Building the Query Decomposer

The decomposer is the brain of the pipeline. It takes a complex question and produces a retrieval plan: an ordered sequence of sub-queries where each one might depend on results from previous ones.

import openai
import json
from typing import Any

client = openai.OpenAI()

DECOMPOSER_SYSTEM_PROMPT = """You are a retrieval planning expert for a multi-hop question answering system.

Your job is to decompose complex questions into a sequence of targeted retrieval sub-queries.
Each sub-query should:
1. Target ONE specific piece of evidence from ONE type of source
2. Be phrased as a precise search query (not a question to answer)
3. Indicate what prior evidence it depends on, if any

Return JSON with this structure:
{
  "analysis": "Brief analysis of why this question requires multiple hops",
  "sub_queries": [
    {
      "order": 1,
      "query": "specific search phrase",
      "rationale": "what evidence this retrieves and why we need it",
      "depends_on": [],
      "target_document_type": "e.g., financial report, regulatory document, internal memo"
    }
  ],
  "synthesis_hint": "How the sub-query results should be combined to answer the original question"
}

IMPORTANT: Order sub-queries so that each one could refine subsequent ones.
Limit to 4 sub-queries maximum. If a question needs more, it should be broken into multiple questions."""

def decompose_question(question: str, available_document_types: list[str]) -> dict:
    """
    Decompose a complex question into an ordered retrieval plan.
    """
    doc_context = f"Available document types: {', '.join(available_document_types)}"
    
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": DECOMPOSER_SYSTEM_PROMPT},
            {"role": "user", "content": f"{doc_context}\n\nQuestion: {question}"}
        ],
        response_format={"type": "json_object"},
        temperature=0.1  # Low temperature for consistent planning
    )
    
    return json.loads(response.choices[0].message.content)

Notice the temperature=0.1 setting — you want the decomposer to be deterministic and precise. This is planning work, not creative work. Using structured output via response_format ensures you always get parseable JSON back.

Let's test the decomposer with a realistic example:

question = """How did the European Central Bank's interest rate decisions in 2023 
affect the mortgage portfolio stress metrics described in our Q4 2023 risk 
committee report, and what does that imply for our 2024 capital allocation 
strategy outlined in the board presentation?"""

doc_types = [
    "ECB monetary policy statements",
    "internal risk committee reports",
    "board strategy presentations",
    "regulatory filings"
]

plan = decompose_question(question, doc_types)
print(json.dumps(plan, indent=2))

A well-functioning decomposer will produce something like:

{
  "analysis": "This question requires connecting ECB rate decisions (external) to internal mortgage stress metrics, then reasoning about capital implications. Three distinct evidence sources needed.",
  "sub_queries": [
    {
      "order": 1,
      "query": "ECB interest rate decisions increases 2023 cumulative basis points",
      "rationale": "Establishes the factual rate change data that defines the stress scenario",
      "depends_on": [],
      "target_document_type": "ECB monetary policy statements"
    },
    {
      "order": 2,
      "query": "mortgage portfolio stress test results duration risk Q4 2023",
      "rationale": "Retrieves the internal stress metrics that were affected by the rate changes",
      "depends_on": [1],
      "target_document_type": "internal risk committee reports"
    },
    {
      "order": 3,
      "query": "2024 capital allocation framework risk-weighted assets buffer requirements",
      "rationale": "Retrieves the strategic capital context to assess implications",
      "depends_on": [1, 2],
      "target_document_type": "board strategy presentations"
    }
  ],
  "synthesis_hint": "Connect the rate magnitude (hop 1) to specific stress metric impacts (hop 2), then evaluate whether the capital strategy (hop 3) adequately accounts for those impacts."
}

This is exactly the kind of explicit reasoning chain we need. The planner has correctly identified that the answer requires three distinct evidence sources and that they need to be retrieved in dependency order.

---

Building the Iterative Retrieval Engine

Now we need retrieval that can incorporate prior evidence as context. This is the key architectural difference from standard RAG — each hop's retrieval can be conditioned on what previous hops found.

import numpy as np
from typing import Callable

# Assume you have a vector store client and an embedding function
# These would be your actual infrastructure (Pinecone, Weaviate, pgvector, etc.)

def embed_text(text: str) -> list[float]:
    """Embed a single string using OpenAI embeddings."""
    response = client.embeddings.create(
        input=text,
        model="text-embedding-3-large"
    )
    return response.data[0].embedding

def build_conditioned_query(
    sub_query: str, 
    prior_evidence: list[EvidenceChunk],
    max_prior_tokens: int = 500
) -> str:
    """
    Build a retrieval query that incorporates context from prior hops.
    
    Rather than just searching for the sub-query, we expand it with
    specific entities, dates, and values discovered in prior hops.
    This dramatically improves precision on subsequent hops.
    """
    if not prior_evidence:
        return sub_query
    
    # Extract key facts from prior evidence to condition this hop
    prior_context = "\n".join([
        f"[From {e.document_title}]: {e.content[:200]}"
        for e in prior_evidence[-3:]  # Most recent evidence only
    ])
    
    expansion_prompt = f"""Given this sub-query and prior evidence, write an expanded search query 
that incorporates specific entities, values, or terms from the prior evidence.

Sub-query: {sub_query}

Prior evidence summary:
{prior_context}

Write ONLY the expanded search query. Be specific. Include entity names, dates, numbers found in prior evidence."""
    
    response = client.chat.completions.create(
        model="gpt-4o-mini",  # Cheap model for query expansion
        messages=[{"role": "user", "content": expansion_prompt}],
        max_tokens=150,
        temperature=0.0
    )
    
    return response.choices[0].message.content.strip()


def retrieve_for_hop(
    sub_query: str,
    hop_number: int,
    prior_evidence: list[EvidenceChunk],
    vector_store,  # Your actual vector store client
    top_k: int = 5,
    score_threshold: float = 0.65
) -> list[EvidenceChunk]:
    """
    Execute one retrieval hop, conditioned on prior evidence.
    """
    # Build the conditioned query incorporating prior evidence
    conditioned_query = build_conditioned_query(sub_query, prior_evidence)
    
    # Embed and search
    query_embedding = embed_text(conditioned_query)
    
    # This is your actual vector store call - adapt to your infrastructure
    raw_results = vector_store.query(
        vector=query_embedding,
        top_k=top_k,
        include_metadata=True
    )
    
    # Convert to EvidenceChunk objects, filtering by score threshold
    chunks = []
    for result in raw_results:
        if result.score >= score_threshold:
            chunk = EvidenceChunk(
                chunk_id=result.id,
                document_id=result.metadata.get("document_id", "unknown"),
                document_title=result.metadata.get("title", "Unknown Document"),
                content=result.metadata.get("text", ""),
                relevance_score=result.score,
                hop_number=hop_number,
                sub_query_that_retrieved_it=sub_query
            )
            chunks.append(chunk)
    
    return chunks

---

The Hop Controller: The Loop That Drives Everything

This is where it all comes together. The hop controller executes the retrieval plan, manages the evidence buffer, and decides when we have enough to synthesize.

class MultiHopController:
    def __init__(
        self, 
        vector_store,
        max_hops: int = 4,
        min_evidence_per_hop: int = 2,
        score_threshold: float = 0.65
    ):
        self.vector_store = vector_store
        self.max_hops = max_hops
        self.min_evidence_per_hop = min_evidence_per_hop
        self.score_threshold = score_threshold
    
    def check_sufficiency(
        self, 
        question: str, 
        evidence_buffer: list[EvidenceChunk],
        synthesis_hint: str
    ) -> tuple[bool, str]:
        """
        Ask an LLM whether the current evidence is sufficient to answer the question.
        Returns (is_sufficient, reasoning).
        """
        evidence_summary = self._format_evidence_for_review(evidence_buffer)
        
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {
                    "role": "system",
                    "content": """You assess whether retrieved evidence is sufficient to answer a complex question.
                    
Return JSON: {"sufficient": true/false, "reasoning": "brief explanation", "missing": "what's still needed if not sufficient"}"""
                },
                {
                    "role": "user",
                    "content": f"""Question: {question}

Synthesis approach: {synthesis_hint}

Evidence collected so far:
{evidence_summary}

Is this evidence sufficient to answer the question accurately?"""
                }
            ],
            response_format={"type": "json_object"},
            temperature=0.0
        )
        
        result = json.loads(response.choices[0].message.content)
        return result["sufficient"], result.get("reasoning", "")
    
    def _format_evidence_for_review(self, evidence_buffer: list[EvidenceChunk]) -> str:
        """Format evidence buffer for LLM review."""
        formatted = []
        for i, chunk in enumerate(evidence_buffer):
            formatted.append(
                f"[Evidence {i+1} | Hop {chunk.hop_number} | "
                f"Source: {chunk.document_title} | "
                f"Score: {chunk.relevance_score:.2f}]\n"
                f"{chunk.content[:400]}"
            )
        return "\n\n---\n\n".join(formatted)
    
    def run(
        self, 
        question: str, 
        available_document_types: list[str]
    ) -> MultiHopResult:
        """
        Execute the full multi-hop retrieval pipeline.
        """
        # Step 1: Decompose the question into a retrieval plan
        plan = decompose_question(question, available_document_types)
        sub_queries = plan["sub_queries"]
        synthesis_hint = plan.get("synthesis_hint", "")
        
        evidence_buffer: list[EvidenceChunk] = []
        hops: list[HopResult] = []
        
        # Step 2: Execute each hop in order
        for sq in sub_queries[:self.max_hops]:
            hop_num = sq["order"]
            query_text = sq["query"]
            rationale = sq.get("rationale", "")
            
            print(f"Hop {hop_num}: {query_text}")
            
            # Retrieve chunks for this hop, conditioned on prior evidence
            chunks = retrieve_for_hop(
                sub_query=query_text,
                hop_number=hop_num,
                prior_evidence=evidence_buffer,
                vector_store=self.vector_store,
                score_threshold=self.score_threshold
            )
            
            hop_result = HopResult(
                hop_number=hop_num,
                sub_query=query_text,
                chunks=chunks,
                reasoning=rationale,
                sufficient_for_answer=False
            )
            hops.append(hop_result)
            evidence_buffer.extend(chunks)
            
            # Step 3: Check if we have enough evidence to stop early
            if len(evidence_buffer) >= self.min_evidence_per_hop * hop_num:
                is_sufficient, sufficiency_reasoning = self.check_sufficiency(
                    question, evidence_buffer, synthesis_hint
                )
                if is_sufficient:
                    hop_result.sufficient_for_answer = True
                    print(f"Evidence sufficient after hop {hop_num}. Stopping early.")
                    break
        
        # Step 4: Deduplicate evidence buffer
        evidence_buffer = self._deduplicate_evidence(evidence_buffer)
        
        # Step 5: Synthesize the answer
        answer, citations, confidence = self.synthesize(
            question=question,
            evidence_buffer=evidence_buffer,
            synthesis_hint=synthesis_hint,
            hop_chain=hops
        )
        
        return MultiHopResult(
            original_question=question,
            answer=answer,
            hops=hops,
            evidence_buffer=evidence_buffer,
            confidence=confidence,
            citations=citations
        )
    
    def _deduplicate_evidence(
        self, 
        evidence_buffer: list[EvidenceChunk]
    ) -> list[EvidenceChunk]:
        """
        Remove near-duplicate chunks that were retrieved in multiple hops.
        Uses chunk_id for exact deduplication, then content overlap for near-dedup.
        """
        seen_ids = set()
        unique_chunks = []
        
        for chunk in evidence_buffer:
            if chunk.chunk_id not in seen_ids:
                seen_ids.add(chunk.chunk_id)
                unique_chunks.append(chunk)
        
        return unique_chunks

The early stopping logic deserves attention. Without it, you'll execute all planned hops even when you found everything you need on hop 1 or 2. This wastes money and time, and paradoxically can hurt answer quality by introducing irrelevant evidence that confuses the synthesis step.

---

The Synthesis Engine: Reasoning Across Evidence

This is the hardest prompt engineering challenge in the pipeline. You need the LLM to:

1. Read evidence from multiple different documents
2. Identify which pieces are relevant to which sub-questions
3. Reason about the relationships between evidence from different sources
4. Produce a coherent answer that synthesizes rather than summarizes
5. Cite sources with hop-level attribution

The synthesis prompt is doing heavy lifting. Let's be deliberate about it:

SYNTHESIS_SYSTEM_PROMPT = """You are an expert analyst synthesizing evidence from multiple documents to answer complex questions.

Your task:
1. Read ALL evidence carefully, noting which document each piece comes from
2. Identify the KEY CLAIMS in each evidence piece that are relevant to the question
3. Reason about how claims from DIFFERENT documents connect or conflict
4. Synthesize a coherent answer that EXPLICITLY SHOWS the reasoning chain
5. Cite sources using [Doc: {document_title}, Hop {hop_number}] format

CRITICAL RULES:
- Only make claims supported by the provided evidence
- If evidence pieces conflict, acknowledge the conflict explicitly  
- Show your reasoning chain: how finding X led you to look for Y, which revealed Z
- If evidence is insufficient for part of the question, say so explicitly
- Do NOT pad with generic statements. Every sentence should advance the answer.

Format your response as:
## Reasoning Chain
[Show step-by-step how you connected evidence across sources]

## Answer
[The synthesized answer with inline citations]

## Confidence Assessment
[High/Medium/Low] - [Brief explanation of what might be missing or uncertain]"""


def synthesize(
    self,
    question: str,
    evidence_buffer: list[EvidenceChunk],
    synthesis_hint: str,
    hop_chain: list[HopResult]
) -> tuple[str, list[dict], float]:
    """
    Synthesize a final answer from the accumulated evidence buffer.
    Returns (answer_text, citations, confidence_score).
    """
    # Format the evidence with full metadata for the LLM
    evidence_text = self._format_evidence_for_synthesis(evidence_buffer)
    
    # Format the hop chain so the LLM understands how evidence was collected
    hop_summary = self._format_hop_chain(hop_chain)
    
    response = client.chat.completions.create(
        model="gpt-4o",  # Use the best model for synthesis
        messages=[
            {"role": "system", "content": SYNTHESIS_SYSTEM_PROMPT},
            {
                "role": "user",
                "content": f"""Question: {question}

Retrieval approach used:
{hop_summary}

Synthesis guidance: {synthesis_hint}

Evidence collected:
{evidence_text}

Synthesize a comprehensive answer."""
            }
        ],
        temperature=0.1,
        max_tokens=2000
    )
    
    answer_text = response.choices[0].message.content
    
    # Extract citations from the evidence buffer
    citations = self._build_citation_list(evidence_buffer)
    
    # Parse confidence from the answer (simple heuristic)
    confidence = self._extract_confidence(answer_text)
    
    return answer_text, citations, confidence


def _format_evidence_for_synthesis(self, evidence_buffer: list[EvidenceChunk]) -> str:
    """Format evidence with full provenance for synthesis."""
    sections = []
    
    # Group by document for cleaner presentation
    by_doc: dict[str, list[EvidenceChunk]] = {}
    for chunk in evidence_buffer:
        key = chunk.document_title
        if key not in by_doc:
            by_doc[key] = []
        by_doc[key].append(chunk)
    
    for doc_title, chunks in by_doc.items():
        section_parts = [f"### Source: {doc_title}"]
        for chunk in chunks:
            section_parts.append(
                f"[Hop {chunk.hop_number} | Retrieved for: '{chunk.sub_query_that_retrieved_it}' | "
                f"Relevance: {chunk.relevance_score:.2f}]\n{chunk.content}"
            )
        sections.append("\n\n".join(section_parts))
    
    return "\n\n" + "="*60 + "\n\n".join(sections)


def _build_citation_list(self, evidence_buffer: list[EvidenceChunk]) -> list[dict]:
    """Build a structured citation list from the evidence buffer."""
    citations = []
    seen_docs = set()
    
    for chunk in evidence_buffer:
        if chunk.document_id not in seen_docs:
            seen_docs.add(chunk.document_id)
            citations.append({
                "document_id": chunk.document_id,
                "title": chunk.document_title,
                "first_retrieved_at_hop": chunk.hop_number,
                "relevance_score": chunk.relevance_score,
                "chunk_ids_used": [
                    c.chunk_id for c in evidence_buffer 
                    if c.document_id == chunk.document_id
                ]
            })
    
    return sorted(citations, key=lambda x: x["first_retrieved_at_hop"])

---

Handling the Hard Cases

When Retrieval Finds Nothing Useful

Sometimes a hop returns no chunks above the score threshold. This is a signal, not a failure. The pipeline should handle this gracefully:

def handle_empty_hop(
    sub_query: str, 
    hop_number: int,
    evidence_buffer: list[EvidenceChunk]
) -> Optional[str]:
    """
    When a hop returns nothing, generate an alternative query.
    Returns a reformulated query or None if we should skip this hop.
    """
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "user",
                "content": f"""The following search query returned no useful results in our document corpus.
                
Failed query: {sub_query}

Prior evidence found: {len(evidence_buffer)} chunks from these sources: 
{list(set(c.document_title for c in evidence_buffer))}

Suggest ONE alternative search query that might find relevant information.
If no alternative seems viable, respond with: SKIP

Alternative query:"""
            }
        ],
        max_tokens=100,
        temperature=0.2
    )
    
    result = response.choices[0].message.content.strip()
    if result.upper() == "SKIP":
        return None
    return result

Detecting and Breaking Circular Retrieval

A subtle failure mode: hop 3's conditioned query retrieves the same chunks as hop 1, because the entity linking causes the search to collapse back on itself. You need to detect and break this loop:

def detect_retrieval_loop(
    new_chunks: list[EvidenceChunk],
    evidence_buffer: list[EvidenceChunk],
    overlap_threshold: float = 0.7
) -> bool:
    """
    Detect if we're retrieving the same evidence repeatedly.
    Returns True if we're in a loop.
    """
    if not evidence_buffer or not new_chunks:
        return False
    
    existing_ids = set(c.chunk_id for c in evidence_buffer)
    new_ids = set(c.chunk_id for c in new_chunks)
    
    if not new_ids:
        return True
    
    overlap = len(existing_ids & new_ids) / len(new_ids)
    return overlap >= overlap_threshold

If detect_retrieval_loop returns True, you have two options: reformulate the sub-query with more specificity, or skip the hop entirely and proceed to synthesis with what you have.

---

Evidence Quality Filtering and Reranking

Not all retrieved chunks are equally useful. Before synthesis, you should filter evidence using a cross-encoder reranker or LLM-based scoring — especially important when hops return many chunks.

For a deep dive on reranking, the Building a Reranking Layer for RAG article covers cross-encoders and LLM scoring in detail. Here's a minimal LLM-based filter that works well for multi-hop:

def filter_evidence_by_relevance(
    question: str,
    evidence_buffer: list[EvidenceChunk],
    max_chunks: int = 12
) -> list[EvidenceChunk]:
    """
    Filter evidence buffer to the most relevant chunks using LLM scoring.
    Also prunes chunks that are redundant given others in the buffer.
    """
    if len(evidence_buffer) <= max_chunks:
        return evidence_buffer
    
    # Format all chunks for batch scoring
    chunks_text = "\n\n".join([
        f"[{i}] Source: {c.document_title} | Hop: {c.hop_number}\n{c.content[:300]}"
        for i, c in enumerate(evidence_buffer)
    ])
    
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "user",
                "content": f"""Question: {question}

Rate each evidence chunk's relevance to answering this question (1-10).
Also flag chunks that are REDUNDANT given other chunks.

Chunks:
{chunks_text}

Return JSON: {{"scores": [{{"index": 0, "score": 8, "redundant": false}}, ...]}}"""
            }
        ],
        response_format={"type": "json_object"},
        temperature=0.0
    )
    
    scores_data = json.loads(response.choices[0].message.content)
    
    # Filter and sort
    scored_chunks = []
    for item in scores_data["scores"]:
        if not item.get("redundant", False) and item["score"] >= 6:
            idx = item["index"]
            if idx < len(evidence_buffer):
                chunk = evidence_buffer[idx]
                chunk.relevance_score = item["score"] / 10.0
                scored_chunks.append(chunk)
    
    # Sort by score descending, keep top max_chunks
    scored_chunks.sort(key=lambda x: x.relevance_score, reverse=True)
    return scored_chunks[:max_chunks]

---

Provenance Tracking and Citation Attribution

Multi-hop RAG is only useful if users can trust the answers. That requires citations that show not just what source was used, but how it entered the reasoning chain. The Building a Citation and Source Attribution System for RAG article covers the full citation infrastructure in depth. For multi-hop, the key addition is hop-level attribution:

def format_answer_with_citations(result: MultiHopResult) -> str:
    """
    Format the final answer with a full citation trail including hop metadata.
    """
    output = []
    output.append(f"## Answer\n\n{result.answer}\n")
    output.append("---\n")
    output.append("## Evidence Trail\n")
    
    for hop in result.hops:
        output.append(f"\n### Hop {hop.hop_number}: {hop.sub_query}")
        output.append(f"*Rationale: {hop.reasoning}*\n")
        
        if hop.chunks:
            for chunk in hop.chunks:
                output.append(
                    f"- **{chunk.document_title}** "
                    f"(relevance: {chunk.relevance_score:.2f}): "
                    f"{chunk.content[:150]}..."
                )
        else:
            output.append("- *No evidence retrieved in this hop*")
    
    output.append("\n---\n")
    output.append(f"**Confidence:** {result.confidence:.0%}")
    output.append(f"**Sources consulted:** {len(result.citations)}")
    output.append(f"**Total hops:** {len(result.hops)}")
    
    return "\n".join(output)

---

Performance and Cost Optimization

Multi-hop RAG is inherently more expensive than single-pass RAG. Here's how to control the cost without gutting quality:

Use smaller models for planning, larger for synthesis. The query decomposer and sufficiency checks work well with gpt-4o-mini or claude-3-haiku. Reserve your most capable (and expensive) model for the final synthesis step. This alone typically reduces total cost by 40-60%.

Parallelize independent hops. Look at the depends_on field in the decomposition plan. Sub-queries with no dependencies can execute in parallel. For a 4-hop plan, often hops 1 and 2 are independent and can run concurrently, cutting wall-clock time roughly in half. The orchestrating parallel LLM calls article covers the async patterns you need.

Cache retrieval results by query hash. Conditioned queries that produce the same expanded query string can reuse cached vector search results. Since the vector store is often the latency bottleneck, this is high-impact.

import hashlib
import json
from functools import lru_cache

def query_hash(sub_query: str, prior_evidence_ids: list[str]) -> str:
    """Generate a cache key for a retrieval hop."""
    key_data = {
        "query": sub_query,
        "prior_ids": sorted(prior_evidence_ids)
    }
    return hashlib.sha256(
        json.dumps(key_data, sort_keys=True).encode()
    ).hexdigest()[:16]

Set aggressive hop limits. In production, most multi-hop questions are answerable in 2-3 hops. Setting max_hops=4 provides a safety ceiling but the early stopping logic means you rarely hit it. Monitor your average hop count in production — if it's consistently above 3, your decomposer is over-splitting, which means your document types list is too granular or your chunks are too small.

For monitoring production pipelines, you'll want proper LLM observability to track hop counts, retrieval scores, and synthesis latency across your query population.

---

Hands-On Exercise

Build a complete multi-hop RAG system for a financial research scenario. Here's the setup:

Scenario: You work at an investment firm. Your document corpus includes:

• 10-K filings for 5 technology companies (2021-2023)
• Quarterly earnings call transcripts
• Internal investment committee memos
• Industry analyst reports

Exercise tasks:

1. Build the corpus: Create a small test corpus by taking at least 3 different document types (you can use publicly available 10-K excerpts and analyst report summaries). Chunk them appropriately using the strategies from Chunking Strategies for RAG and embed them into a vector store of your choice.

2. Test the decomposer: Run these questions through your decomposer and verify the sub-queries make sense:

   • "How has Microsoft's cloud revenue growth trajectory compared to AWS, and what does that imply for competitive dynamics in the enterprise AI platform market?"
   • "Did the Fed rate hike cycle affect how tech companies discussed their capital expenditure plans in their most recent earnings calls?"

3. Instrument the hop controller: Add logging so you can see exactly which chunks were retrieved at each hop, their scores, and whether early stopping triggered.

4. Evaluate synthesis quality: For at least two questions, manually evaluate whether the synthesized answer correctly attributes claims to the right sources and shows a genuine reasoning chain rather than just summarizing the retrieved chunks.

5. Break it deliberately: Find a question that causes circular retrieval and verify your detect_retrieval_loop function catches it. What types of questions tend to cause loops?

---

Common Mistakes & Troubleshooting

The synthesis is just summarizing chunks, not reasoning across them. This usually means your synthesis prompt isn't explicit enough about connecting evidence from different sources. Add a "Connections" section requirement to the prompt: "Before writing the answer, explicitly state what you learned from Source A that made you interpret Source B differently." This forces the LLM to articulate the cross-document reasoning.

The decomposer produces overly specific sub-queries that retrieve nothing. This happens when the LLM generates queries that assume a level of vocabulary precision your documents don't have. Fix this by providing a few example document excerpts in the decomposer prompt, so it calibrates its query language to match your corpus vocabulary.

Confidence scores are unreliable. LLM-generated confidence is notoriously miscalibrated. Don't display raw LLM confidence scores to end users. Instead, compute a composite signal: (average retrieval score across evidence buffer) × (1 - fraction of hops that returned nothing) × (1 if all planned hops found evidence, 0.7 if some were skipped). This mechanical score is less intelligent but far more consistent.

Multi-hop runs are inconsistent — the same question sometimes gives different answers. Three causes: (1) The decomposer generates different sub-queries on different runs, despite low temperature. Fix this by caching the decomposition plan keyed on the question hash. (2) The vector store returns slightly different top-k results between calls due to approximate nearest neighbor variation. Fix this by using deterministic retrieval (most vector stores have an exact search mode). (3) The synthesis LLM is sampling creatively. Fix this by setting temperature to 0 for synthesis.

The pipeline is too slow for interactive use. Typical multi-hop latency is 8-15 seconds for a 3-hop query. For interactive applications, use streaming responses to show the synthesis output token-by-token as soon as the retrieval phase completes. Show hop progress indicators in the UI ("Searching regulatory documents... Searching earnings transcripts... Synthesizing..."). Users tolerate 12 seconds much better when they see progress.

Evidence from later hops is overwhelming evidence from earlier hops. If the evidence buffer is sorted by hop number, the synthesis LLM tends to anchor heavily on the most recent evidence. Sort by relevance score instead, or explicitly tell the synthesis prompt: "Earlier hops established foundational context; later hops refine or extend it. Weight all evidence relative to how directly it answers the specific question asked."

---

Summary & Next Steps

Multi-hop RAG is the architecture that separates systems capable of genuine knowledge work from systems that can only pattern-match to explicit text. The key ideas:

1. Plan before you retrieve. Decompose complex questions into ordered, dependency-aware sub-queries. Retrieval without a plan is expensive wandering.

2. Condition each hop on prior evidence. Expand sub-queries using entities and values discovered in previous hops. This is what makes retrieval convergent rather than scattered.

3. Maintain a typed evidence buffer. Track every chunk's provenance: which hop, which sub-query, which source. You'll need this for debugging, citation, and confidence scoring.

4. Deduplicate and filter before synthesis. More evidence is not always better. The synthesis LLM performs best with 8-15 highly relevant, non-redundant chunks from clearly identified sources.

5. Make synthesis reason, not summarize. Your synthesis prompt must explicitly require cross-document reasoning chains, not just per-source summaries.

6. Use cheap models for planning, expensive models for reasoning. The planner, sufficiency checker, and query expander can all run on smaller models. The synthesis step is where you pay for capability.

From here, natural extensions include:

• Knowledge graph augmentation: Instead of purely vector-based hops, use entity extraction to build a graph over your corpus, then do graph traversal between hops. This is especially powerful for corporate relationship networks and citation graphs. The Knowledge Graph-Augmented RAG article covers this architecture in depth.

• Self-improving retrieval: Log which hops succeed, which fail, and what reformulations worked. Feed this data back into a retrieval feedback loop. The Self-Improving RAG Pipeline article shows how to operationalize this.

• Agentic multi-hop: Replace the fixed decomposition plan with a fully agentic loop where the LLM decides at each step whether to retrieve more, reformulate a query, or synthesize. This is more powerful but requires careful agentic loop design to prevent runaway execution.

Multi-hop RAG is, fundamentally, teaching a retrieval system to think in sequences of questions rather than a single query. Master that, and you've moved from building search engines to building reasoning engines.