Implementing RAPTOR: Recursive Abstractive Processing for Tree-Organized Retrieval in Production RAG Pipelines

Standard RAG pipelines have a dirty secret: they're fundamentally flat. You chunk your documents, embed those chunks, and retrieve the top-k nearest neighbors to a query. When a user asks "what were the main strategic themes across all our Q4 board reports?" your retrieval system returns a handful of individual chunks — each accurate in isolation, but none of them synthesizing the broader picture. The answer requires reasoning across dozens of documents simultaneously, and no single chunk contains it.

This is the problem RAPTOR was designed to solve. Published by researchers at Stanford in 2024, RAPTOR (Recursive Abstractive Processing for Tree-Organized Retrieval) builds a hierarchical tree structure over your corpus by recursively clustering and summarizing chunks at multiple levels of abstraction. The result is a multi-granularity index where a single query can retrieve both a high-level thematic summary and the specific supporting details — from the same retrieval pass.

By the end of this lesson, you'll have built a complete RAPTOR pipeline that ingests a realistic document corpus, constructs the summarization tree, indexes every level, and retrieves intelligently based on query type. You'll also know when RAPTOR earns its indexing overhead and when simpler approaches are the better call.

What you'll learn:

• How RAPTOR's recursive clustering and summarization algorithm works, including the mathematical intuition behind UMAP dimensionality reduction and Gaussian Mixture Model clustering
• How to implement a production-grade RAPTOR indexing pipeline from scratch using Python, LangChain, and a vector store
• How to design retrieval strategies that exploit the tree structure — collapsed vs. tree-traversal retrieval
• How to manage the cost and latency tradeoffs of building and maintaining a RAPTOR index at scale
• How to evaluate whether RAPTOR actually improves answer quality for your specific workload

---

Prerequisites

This is an intermediate-to-advanced lesson. You should already be comfortable with:

• Building basic RAG pipelines — if you need a refresher, start with RAG Fundamentals: Build Your First Retrieval-Augmented Generation System
• Chunking documents and embedding them — Chunking Strategies: How to Split Documents for Better Retrieval covers this well
• A working understanding of vector stores and similarity search — reviewed in Vector Databases Compared: Pinecone vs Weaviate vs pgvector for Production RAG
• Python proficiency and familiarity with LLM APIs

You'll need: langchain, langchain-openai, scikit-learn, umap-learn, numpy, tiktoken, and a vector store (we'll use pgvector for this lesson, but the pattern transfers easily).

---

The Core Problem: Why Flat Retrieval Fails at Scale

Before building anything, it's worth being precise about why flat RAG struggles with certain query types, because that precision will guide every design decision you make.

Consider a legal team that has ingested three years of merger and acquisition agreements — roughly 800 documents, each 40–120 pages. A flat RAG system works reasonably well for questions like "what are the indemnification caps in the Meridian acquisition?" That question has a specific, localized answer. But questions like "across all acquisitions since 2021, how have our IP warranty provisions evolved?" require the system to synthesize patterns across hundreds of document sections.

With flat retrieval, you get a context window full of specific clause fragments — none of which tells the LLM what the trend is. The LLM is forced to do all the synthesis from raw text, which quickly hits token limits and produces answers that are either truncated or hallucinated.

RAPTOR addresses this by precomputing the synthesis. During indexing, it recursively:

1. Takes your base-level chunks
2. Clusters them by semantic similarity
3. Generates an LLM summary for each cluster
4. Treats those summaries as new "documents" and repeats

After a few iterations, you have a tree where the leaf nodes are your original chunks and the internal nodes are progressively more abstract summaries. A query about strategic trends hits the high-level nodes. A query about specific clauses hits the leaves. A well-designed retrieval strategy can hit both levels simultaneously.

---

Understanding the RAPTOR Algorithm

Let's walk through the indexing algorithm carefully before touching code. There are three stages: dimensionality reduction, clustering, and summarization. These repeat recursively until clusters become too small to subdivide further.

Stage 1: Dimensionality Reduction with UMAP

High-dimensional embedding vectors (typically 1536 dimensions for text-embedding-3-small) cluster poorly in high-dimensional space due to the curse of dimensionality. Distances between points become nearly uniform, which makes clustering algorithms produce garbage results.

RAPTOR uses UMAP (Uniform Manifold Approximation and Projection) to reduce embeddings to a much lower dimensionality — typically 10 dimensions for local structure-preserving reduction during clustering, and 2 dimensions for visualization. This is a nonlinear dimensionality reduction that preserves neighborhood structure better than PCA for this use case.

The key UMAP parameters that matter here:

• n_neighbors: Controls how local vs. global the structure is. Lower values (10–15) emphasize local cluster structure. Higher values (50+) emphasize global topology.
• n_components: The target dimensionality. For GMM clustering, 10 works well. For visualization, use 2.
• metric: Use cosine to match your embedding similarity metric.

Stage 2: Gaussian Mixture Model Clustering

RAPTOR uses Gaussian Mixture Models (GMM) rather than k-means for a critical reason: GMMs support soft assignment, meaning a chunk can belong to multiple clusters with fractional probabilities. This is semantically meaningful — a paragraph about "pricing strategy in competitive markets" genuinely belongs in both a "pricing" cluster and a "competitive analysis" cluster.

With soft assignment, you include a chunk in a cluster's summary if its membership probability exceeds a threshold (typically 0.5). This means the same source chunk might appear in multiple summarization branches, which improves recall at the cost of some index redundancy.

The number of clusters is determined using the Bayesian Information Criterion (BIC), which penalizes model complexity. You fit GMMs with increasing numbers of clusters and select the one with the lowest BIC. This is computationally expensive for large datasets, so in practice you often use a heuristic: roughly sqrt(n_chunks / 2) as a starting estimate.

Stage 3: LLM Summarization

For each cluster, you concatenate the member chunks and pass them to an LLM with a summarization prompt. The resulting summary becomes a new node in the tree. These summaries are then embedded and become inputs to the next iteration of the algorithm.

Recursion stops when clusters contain fewer than a minimum number of nodes (typically 10), ensuring you don't summarize individual summaries repeatedly into meaninglessness.

---

Building the RAPTOR Indexing Pipeline

Let's build this step by step. We'll use a realistic scenario: a corpus of annual reports and 10-K filings for a portfolio of companies. The goal is to support both specific financial queries ("what was Northstar Holdings' revenue growth in fiscal 2023?") and thematic queries ("which portfolio companies cited supply chain risk as a material concern?").

Project Structure

raptor_pipeline/
├── raptor/
│   ├── __init__.py
│   ├── clustering.py
│   ├── summarizer.py
│   ├── tree_builder.py
│   ├── retriever.py
│   └── models.py
├── ingest.py
├── query.py
└── config.py

Data Models

Start with clean data structures that represent your tree nodes:

# raptor/models.py
from dataclasses import dataclass, field
from typing import Optional
from uuid import uuid4

@dataclass
class RaptorNode:
    """A single node in the RAPTOR tree — either a leaf chunk or an abstract summary."""
    node_id: str = field(default_factory=lambda: str(uuid4()))
    text: str = ""
    level: int = 0  # 0 = leaf chunks, 1+ = summaries
    embedding: Optional[list[float]] = None
    children_ids: list[str] = field(default_factory=list)
    parent_id: Optional[str] = None
    source_document: Optional[str] = None
    cluster_id: Optional[int] = None
    metadata: dict = field(default_factory=dict)

    def token_count(self) -> int:
        """Rough token estimate — useful for context window management."""
        import tiktoken
        enc = tiktoken.get_encoding("cl100k_base")
        return len(enc.encode(self.text))

The Clustering Module

# raptor/clustering.py
import numpy as np
from sklearn.mixture import GaussianMixture
from umap import UMAP
from typing import Optional
import logging

logger = logging.getLogger(__name__)

def reduce_dimensions(
    embeddings: np.ndarray,
    n_components: int = 10,
    n_neighbors: int = 15,
    random_state: int = 42
) -> np.ndarray:
    """
    Reduce embedding dimensionality before clustering.
    
    For small datasets (<50 nodes), UMAP can be unstable.
    Fall back to PCA in that case.
    """
    n_samples = embeddings.shape[0]
    
    if n_samples <= n_components:
        logger.warning(
            f"Only {n_samples} samples — skipping dimensionality reduction."
        )
        return embeddings
    
    if n_samples < 50:
        logger.info("Small dataset: using PCA instead of UMAP for stability.")
        from sklearn.decomposition import PCA
        n_components = min(n_components, n_samples - 1)
        return PCA(n_components=n_components, random_state=random_state).fit_transform(embeddings)
    
    reducer = UMAP(
        n_neighbors=min(n_neighbors, n_samples - 1),
        n_components=n_components,
        metric="cosine",
        random_state=random_state,
        low_memory=True,  # Important for large corpora
    )
    return reducer.fit_transform(embeddings)


def select_optimal_clusters(
    reduced_embeddings: np.ndarray,
    max_clusters: Optional[int] = None,
    threshold: float = 0.1,
) -> int:
    """
    Use BIC to find the optimal number of GMM clusters.
    Falls back to heuristic if BIC selection is too slow.
    """
    n_samples = reduced_embeddings.shape[0]
    
    if max_clusters is None:
        max_clusters = max(2, min(int(np.sqrt(n_samples / 2)), 50))
    
    if n_samples < 4:
        return 1
    
    best_bic = np.inf
    best_n = 2
    
    for n in range(2, max_clusters + 1):
        try:
            gm = GaussianMixture(
                n_components=n,
                random_state=42,
                covariance_type="full",
                max_iter=300,
            )
            gm.fit(reduced_embeddings)
            bic = gm.bic(reduced_embeddings)
            if bic < best_bic:
                best_bic = bic
                best_n = n
        except Exception as e:
            logger.warning(f"GMM with {n} clusters failed: {e}")
            break
    
    logger.info(f"Optimal clusters: {best_n} (BIC={best_bic:.2f})")
    return best_n


def cluster_nodes(
    embeddings: np.ndarray,
    n_clusters: int,
    assignment_threshold: float = 0.5,
) -> dict[int, list[int]]:
    """
    Fit a GMM and return soft cluster assignments.
    
    Returns a dict mapping cluster_id -> list of node indices.
    A node may appear in multiple clusters (soft assignment).
    """
    gm = GaussianMixture(
        n_components=n_clusters,
        random_state=42,
        covariance_type="full",
        max_iter=300,
    )
    gm.fit(embeddings)
    
    # Soft probabilities — shape: (n_samples, n_clusters)
    probs = gm.predict_proba(embeddings)
    
    clusters: dict[int, list[int]] = {i: [] for i in range(n_clusters)}
    for node_idx, node_probs in enumerate(probs):
        for cluster_idx, prob in enumerate(node_probs):
            if prob >= assignment_threshold:
                clusters[cluster_idx].append(node_idx)
    
    # Drop empty clusters (can happen with soft thresholds)
    return {k: v for k, v in clusters.items() if v}

The Summarizer

# raptor/summarizer.py
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage
import tiktoken
import logging
from raptor.models import RaptorNode

logger = logging.getLogger(__name__)

SUMMARIZATION_SYSTEM_PROMPT = """You are an expert at synthesizing information from multiple related documents or document sections.

Your task is to produce a concise, comprehensive summary that:
1. Captures the main themes, key facts, and important details across all provided text
2. Preserves specific data points, names, and figures that might be needed for detailed questions
3. Explicitly notes patterns, trends, or relationships that span multiple sections
4. Is written as coherent prose, not bullet points

Do not add information not present in the source text. Do not speculate."""

MAX_INPUT_TOKENS = 100_000  # Stay well within model limits


def summarize_cluster(
    nodes: list[RaptorNode],
    model: str = "gpt-4o-mini",
    max_summary_tokens: int = 1000,
) -> str:
    """
    Summarize a cluster of nodes into a single coherent summary.
    
    Handles token budget by truncating long clusters gracefully.
    """
    enc = tiktoken.get_encoding("cl100k_base")
    
    # Build the cluster text, respecting token limits
    combined_parts = []
    total_tokens = 0
    
    for i, node in enumerate(nodes):
        section = f"--- Section {i+1} (Level {node.level}) ---\n{node.text}\n"
        section_tokens = len(enc.encode(section))
        
        if total_tokens + section_tokens > MAX_INPUT_TOKENS:
            logger.warning(
                f"Cluster too large — truncating at {i} of {len(nodes)} nodes. "
                f"Consider increasing tree depth or reducing chunk size."
            )
            break
        
        combined_parts.append(section)
        total_tokens += section_tokens
    
    combined_text = "\n".join(combined_parts)
    
    llm = ChatOpenAI(model=model, temperature=0, max_tokens=max_summary_tokens)
    
    response = llm.invoke([
        SystemMessage(content=SUMMARIZATION_SYSTEM_PROMPT),
        HumanMessage(content=f"Please summarize the following document sections:\n\n{combined_text}"),
    ])
    
    return response.content.strip()

The Tree Builder

This is the core orchestration logic — the recursive loop that builds the tree level by level:

# raptor/tree_builder.py
import numpy as np
from langchain_openai import OpenAIEmbeddings
from raptor.models import RaptorNode
from raptor.clustering import reduce_dimensions, select_optimal_clusters, cluster_nodes
from raptor.summarizer import summarize_cluster
from typing import Optional
import logging

logger = logging.getLogger(__name__)


class RaptorTreeBuilder:
    def __init__(
        self,
        embedding_model: str = "text-embedding-3-small",
        summarization_model: str = "gpt-4o-mini",
        max_levels: int = 5,
        min_cluster_size: int = 10,
        umap_n_components: int = 10,
        summary_max_tokens: int = 800,
    ):
        self.embedder = OpenAIEmbeddings(model=embedding_model)
        self.summarization_model = summarization_model
        self.max_levels = max_levels
        self.min_cluster_size = min_cluster_size
        self.umap_n_components = umap_n_components
        self.summary_max_tokens = summary_max_tokens
        self.all_nodes: dict[str, RaptorNode] = {}

    def _embed_nodes(self, nodes: list[RaptorNode]) -> None:
        """Batch embed nodes in place."""
        texts = [n.text for n in nodes]
        # Batch in groups of 100 to respect API rate limits
        batch_size = 100
        all_embeddings = []
        for i in range(0, len(texts), batch_size):
            batch = texts[i:i + batch_size]
            all_embeddings.extend(self.embedder.embed_documents(batch))
        
        for node, embedding in zip(nodes, all_embeddings):
            node.embedding = embedding

    def _process_level(
        self,
        current_nodes: list[RaptorNode],
        level: int,
    ) -> list[RaptorNode]:
        """
        Take the current level's nodes, cluster them, summarize each cluster,
        and return the new summary nodes for the next level.
        """
        logger.info(f"Processing level {level}: {len(current_nodes)} nodes")
        
        if len(current_nodes) < self.min_cluster_size:
            logger.info(f"Level {level}: too few nodes to cluster further. Stopping.")
            return []
        
        # Get embeddings as matrix
        embeddings = np.array([n.embedding for n in current_nodes])
        
        # Reduce dimensions for stable clustering
        reduced = reduce_dimensions(
            embeddings,
            n_components=min(self.umap_n_components, len(current_nodes) - 1),
        )
        
        # Find optimal cluster count and assign
        n_clusters = select_optimal_clusters(reduced)
        cluster_assignments = cluster_nodes(reduced, n_clusters)
        
        new_nodes = []
        for cluster_id, member_indices in cluster_assignments.items():
            member_nodes = [current_nodes[i] for i in member_indices]
            
            logger.info(
                f"  Cluster {cluster_id}: {len(member_nodes)} members — summarizing..."
            )
            
            summary_text = summarize_cluster(
                member_nodes,
                model=self.summarization_model,
                max_summary_tokens=self.summary_max_tokens,
            )
            
            summary_node = RaptorNode(
                text=summary_text,
                level=level,
                children_ids=[n.node_id for n in member_nodes],
                cluster_id=cluster_id,
                metadata={
                    "cluster_size": len(member_nodes),
                    "source_levels": list({n.level for n in member_nodes}),
                },
            )
            
            # Update parent pointers
            for member in member_nodes:
                member.parent_id = summary_node.node_id
            
            new_nodes.append(summary_node)
            self.all_nodes[summary_node.node_id] = summary_node
        
        # Embed the new summary nodes
        self._embed_nodes(new_nodes)
        
        return new_nodes

    def build(self, leaf_nodes: list[RaptorNode]) -> dict[str, RaptorNode]:
        """
        Build the full RAPTOR tree from a list of leaf (chunk) nodes.
        
        Returns all nodes at all levels, keyed by node_id.
        """
        # Register and embed leaf nodes
        logger.info(f"Building RAPTOR tree from {len(leaf_nodes)} leaf nodes...")
        for node in leaf_nodes:
            self.all_nodes[node.node_id] = node
        
        self._embed_nodes(leaf_nodes)
        
        current_level_nodes = leaf_nodes
        
        for level in range(1, self.max_levels + 1):
            next_level_nodes = self._process_level(current_level_nodes, level)
            
            if not next_level_nodes:
                logger.info(f"Tree construction complete at level {level - 1}.")
                break
            
            current_level_nodes = next_level_nodes
        
        logger.info(
            f"Tree built: {len(self.all_nodes)} total nodes across "
            f"{max(n.level for n in self.all_nodes.values()) + 1} levels."
        )
        return self.all_nodes

---

Indexing the Tree into Your Vector Store

Once the tree is built, every node at every level needs to be stored in your vector database with metadata that identifies its tree position. This metadata is essential for retrieval strategy logic.

# ingest.py
import json
import psycopg2
from pgvector.psycopg2 import register_vector
from raptor.tree_builder import RaptorTreeBuilder
from raptor.models import RaptorNode
from langchain.text_splitter import RecursiveCharacterTextSplitter
import os

def load_documents_as_leaf_nodes(document_paths: list[str]) -> list[RaptorNode]:
    """Load and chunk documents into leaf-level RaptorNodes."""
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=400,
        chunk_overlap=50,
        separators=["\n\n", "\n", ". ", " ", ""],
    )
    
    leaf_nodes = []
    for path in document_paths:
        with open(path) as f:
            content = f.read()
        
        chunks = splitter.split_text(content)
        doc_name = os.path.basename(path)
        
        for chunk in chunks:
            node = RaptorNode(
                text=chunk,
                level=0,
                source_document=doc_name,
                metadata={"file_path": path, "company": extract_company_name(path)},
            )
            leaf_nodes.append(node)
    
    return leaf_nodes


def extract_company_name(path: str) -> str:
    """Extract company name from filename like 'northstar_holdings_10k_2023.txt'."""
    return os.path.basename(path).split("_")[0].title()


def store_nodes_in_pgvector(
    nodes: dict[str, RaptorNode],
    conn_string: str,
) -> None:
    """Store all RAPTOR tree nodes in pgvector with tree metadata."""
    conn = psycopg2.connect(conn_string)
    register_vector(conn)
    cur = conn.cursor()
    
    cur.execute("""
        CREATE TABLE IF NOT EXISTS raptor_nodes (
            node_id TEXT PRIMARY KEY,
            text TEXT NOT NULL,
            level INTEGER NOT NULL,
            embedding vector(1536),
            parent_id TEXT,
            children_ids JSONB,
            source_document TEXT,
            cluster_id INTEGER,
            metadata JSONB
        );
        
        CREATE INDEX IF NOT EXISTS raptor_nodes_level_idx 
            ON raptor_nodes (level);
        CREATE INDEX IF NOT EXISTS raptor_nodes_embedding_idx 
            ON raptor_nodes USING ivfflat (embedding vector_cosine_ops)
            WITH (lists = 100);
    """)
    
    for node in nodes.values():
        cur.execute("""
            INSERT INTO raptor_nodes 
                (node_id, text, level, embedding, parent_id, children_ids, 
                 source_document, cluster_id, metadata)
            VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
            ON CONFLICT (node_id) DO UPDATE SET
                text = EXCLUDED.text,
                embedding = EXCLUDED.embedding;
        """, (
            node.node_id,
            node.text,
            node.level,
            node.embedding,
            node.parent_id,
            json.dumps(node.children_ids),
            node.source_document,
            node.cluster_id,
            json.dumps(node.metadata),
        ))
    
    conn.commit()
    cur.close()
    conn.close()
    print(f"Stored {len(nodes)} RAPTOR nodes in pgvector.")


if __name__ == "__main__":
    import glob
    
    doc_paths = glob.glob("data/annual_reports/*.txt")
    print(f"Found {len(doc_paths)} documents.")
    
    leaf_nodes = load_documents_as_leaf_nodes(doc_paths)
    print(f"Created {len(leaf_nodes)} leaf nodes.")
    
    builder = RaptorTreeBuilder(
        max_levels=4,
        min_cluster_size=10,
        summarization_model="gpt-4o-mini",
    )
    all_nodes = builder.build(leaf_nodes)
    
    store_nodes_in_pgvector(
        all_nodes,
        conn_string=os.environ["DATABASE_URL"],
    )

---

Designing RAPTOR Retrieval Strategies

This is where RAPTOR's power is actually unlocked — or wasted. The original paper describes two retrieval modes, and choosing between them (or combining them) is a judgment call you need to make based on query type.

Collapsed Tree Retrieval

In collapsed tree retrieval, you treat all nodes at all levels as a single flat index and retrieve the top-k most similar nodes regardless of their level. A semantic query will naturally surface the right granularity: abstract questions float toward high-level summary nodes, specific questions sink toward leaf chunks.

This is simple and often effective. Its weakness is that it can retrieve redundant information — a summary node and its constituent leaf nodes might both appear in the top-k, giving the LLM overlapping content.

Tree Traversal Retrieval

Tree traversal starts at the root (highest-level summaries) and works downward. At each level, it retrieves the top nodes and follows their children if those children are more relevant than siblings. This is more targeted but requires more retrieval passes.

In practice, a hybrid approach often works best: retrieve from the top two levels to get thematic context, then retrieve leaf-level nodes for supporting detail, then deduplicate by parent relationship.

# raptor/retriever.py
import numpy as np
from langchain_openai import OpenAIEmbeddings
import psycopg2
from pgvector.psycopg2 import register_vector


class RaptorRetriever:
    def __init__(self, conn_string: str, embedding_model: str = "text-embedding-3-small"):
        self.conn_string = conn_string
        self.embedder = OpenAIEmbeddings(model=embedding_model)

    def _get_query_embedding(self, query: str) -> list[float]:
        return self.embedder.embed_query(query)

    def _retrieve_from_level(
        self,
        query_embedding: list[float],
        level: int,
        top_k: int,
        metadata_filter: dict = None,
    ) -> list[dict]:
        """Retrieve top-k nodes from a specific tree level."""
        conn = psycopg2.connect(self.conn_string)
        register_vector(conn)
        cur = conn.cursor()
        
        base_query = """
            SELECT node_id, text, level, parent_id, source_document, metadata,
                   1 - (embedding <=> %s::vector) AS similarity
            FROM raptor_nodes
            WHERE level = %s
        """
        params = [query_embedding, level]
        
        if metadata_filter:
            # Simple metadata filter — extend as needed
            for key, value in metadata_filter.items():
                base_query += f" AND metadata->>'{key}' = %s"
                params.append(value)
        
        base_query += " ORDER BY similarity DESC LIMIT %s"
        params.append(top_k)
        
        cur.execute(base_query, params)
        rows = cur.fetchall()
        cur.close()
        conn.close()
        
        return [
            {
                "node_id": r[0],
                "text": r[1],
                "level": r[2],
                "parent_id": r[3],
                "source_document": r[4],
                "metadata": r[5],
                "similarity": float(r[6]),
            }
            for r in rows
        ]

    def collapsed_retrieval(
        self,
        query: str,
        top_k: int = 6,
        metadata_filter: dict = None,
    ) -> list[dict]:
        """
        Flat retrieval across all tree levels simultaneously.
        Best for queries where you're unsure of the right granularity.
        """
        query_embedding = self._get_query_embedding(query)
        
        conn = psycopg2.connect(self.conn_string)
        register_vector(conn)
        cur = conn.cursor()
        
        sql = """
            SELECT node_id, text, level, parent_id, source_document, metadata,
                   1 - (embedding <=> %s::vector) AS similarity
            FROM raptor_nodes
            ORDER BY similarity DESC
            LIMIT %s
        """
        cur.execute(sql, [query_embedding, top_k])
        rows = cur.fetchall()
        cur.close()
        conn.close()
        
        results = [
            {
                "node_id": r[0], "text": r[1], "level": r[2],
                "parent_id": r[3], "source_document": r[4],
                "metadata": r[5], "similarity": float(r[6]),
            }
            for r in rows
        ]
        
        # Deduplicate: if a summary and its child are both retrieved,
        # prefer the summary (it contains the child's content already)
        return self._deduplicate_by_ancestry(results)

    def layered_retrieval(
        self,
        query: str,
        top_k_per_level: int = 3,
        max_level: int = None,
        metadata_filter: dict = None,
    ) -> list[dict]:
        """
        Retrieve from multiple specific levels and combine.
        Best for queries needing both thematic overview and specific evidence.
        """
        query_embedding = self._get_query_embedding(query)
        
        # Get max level from db
        conn = psycopg2.connect(self.conn_string)
        cur = conn.cursor()
        cur.execute("SELECT MAX(level) FROM raptor_nodes")
        tree_max_level = cur.fetchone()[0] or 0
        cur.close()
        conn.close()
        
        if max_level is None:
            max_level = tree_max_level
        
        all_results = []
        
        # Always retrieve from top level (most abstract) and leaf level (most specific)
        priority_levels = list({max_level, 0})
        
        for level in priority_levels:
            level_results = self._retrieve_from_level(
                query_embedding, level, top_k_per_level, metadata_filter
            )
            all_results.extend(level_results)
        
        # Deduplicate and sort by similarity
        seen_ids = set()
        unique_results = []
        for r in sorted(all_results, key=lambda x: x["similarity"], reverse=True):
            if r["node_id"] not in seen_ids:
                seen_ids.add(r["node_id"])
                unique_results.append(r)
        
        return unique_results

    def _deduplicate_by_ancestry(self, results: list[dict]) -> list[dict]:
        """
        If a parent node and its child both appear, drop the child.
        The parent's summary already contains that information.
        """
        retrieved_ids = {r["node_id"] for r in results}
        filtered = []
        
        for result in results:
            parent_id = result.get("parent_id")
            if parent_id and parent_id in retrieved_ids:
                # Parent is already in results — skip this child
                continue
            filtered.append(result)
        
        return filtered

---

Handling Incremental Updates

One of RAPTOR's most awkward production properties is that it's not naturally incremental. Adding a new document to your corpus technically invalidates the clustering — the new document might belong in an existing cluster, which would change the cluster's summary, which would ripple up through the tree.

In practice, most teams handle this with a tiered strategy:

Option 1: Scheduled full rebuilds. If your corpus is relatively stable (updated weekly or monthly), rebuild the tree on a schedule. This is simple and guarantees consistency. Store the tree with a version timestamp; serve the previous version while the new one builds.

Option 2: Append-only with periodic reconciliation. New leaf nodes are indexed immediately at level 0 only (no tree construction). Periodically (nightly, weekly), re-cluster and rebuild summary nodes. Queries in the interim period use collapsed retrieval, which still works — it just misses higher-level synthesis for the newest documents.

Option 3: Mini-tree merging. Build a small RAPTOR tree for each new batch of documents. Merge the root nodes of the mini-trees into the main tree as new internal nodes. This is the most complex but avoids full rebuilds.

For the annual reports use case, Option 1 (monthly rebuild) is usually the right call. For a corpus that ingests dozens of documents daily, Option 2 is more practical.

---

Wiring It into a Query Pipeline

Here's how you connect RAPTOR retrieval to your final answer generation step:

# query.py
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage
from raptor.retriever import RaptorRetriever
import os

ANSWER_SYSTEM_PROMPT = """You are a financial analyst assistant with access to retrieved excerpts from company annual reports and 10-K filings.

Answer the user's question using ONLY the information in the provided context.
- Cite specific companies and years when referencing data points.
- If the context contains both high-level summaries and specific data, use both.
- If you cannot answer from the provided context, say so clearly.
- Do not speculate or add external knowledge."""


def classify_query_type(query: str) -> str:
    """
    Simple heuristic classification. In production, you'd use an LLM router.
    See: query routing patterns for more sophisticated approaches.
    """
    thematic_keywords = [
        "across", "all", "trend", "pattern", "compare", "overall",
        "which companies", "generally", "common", "typically",
    ]
    query_lower = query.lower()
    if any(kw in query_lower for kw in thematic_keywords):
        return "thematic"
    return "specific"


def answer_query(
    query: str,
    conn_string: str,
    metadata_filter: dict = None,
) -> dict:
    retriever = RaptorRetriever(conn_string)
    query_type = classify_query_type(query)
    
    if query_type == "thematic":
        # Layered retrieval: get high-level summaries + some leaf evidence
        context_nodes = retriever.layered_retrieval(
            query,
            top_k_per_level=4,
            metadata_filter=metadata_filter,
        )
    else:
        # Collapsed retrieval: let the vector similarity find the right level
        context_nodes = retriever.collapsed_retrieval(
            query,
            top_k=6,
            metadata_filter=metadata_filter,
        )
    
    # Format context with level indicators for transparency
    context_parts = []
    for node in context_nodes:
        level_label = "Summary" if node["level"] > 0 else "Source excerpt"
        source = node.get("source_document", "Unknown")
        context_parts.append(
            f"[{level_label} | Source: {source} | Similarity: {node['similarity']:.3f}]\n"
            f"{node['text']}"
        )
    
    context = "\n\n---\n\n".join(context_parts)
    
    llm = ChatOpenAI(model="gpt-4o", temperature=0)
    response = llm.invoke([
        SystemMessage(content=ANSWER_SYSTEM_PROMPT),
        HumanMessage(content=f"Context:\n{context}\n\nQuestion: {query}"),
    ])
    
    return {
        "answer": response.content,
        "query_type": query_type,
        "nodes_retrieved": len(context_nodes),
        "context_levels": list({n["level"] for n in context_nodes}),
    }


if __name__ == "__main__":
    conn = os.environ["DATABASE_URL"]
    
    # Test a thematic query
    result = answer_query(
        "Which portfolio companies cited supply chain risk as a material concern "
        "in their most recent annual reports?",
        conn_string=conn,
    )
    print(f"Query type: {result['query_type']}")
    print(f"Retrieved {result['nodes_retrieved']} nodes from levels {result['context_levels']}")
    print(f"\nAnswer:\n{result['answer']}")

This query type classification is intentionally simple. For production systems where routing accuracy is critical, consider implementing a proper query routing strategy that uses an LLM to classify queries across multiple dimensions.

---

Hands-On Exercise

Now it's your turn to build something meaningful. Use the following setup to practice the full RAPTOR pipeline on a real dataset.

Scenario: You work for a consulting firm. Your team has accumulated 150 client-facing reports across 12 industry verticals over the past three years (PDF/text files). Analysts frequently ask questions that span multiple reports — "what trends are we seeing in fintech M&A advisory across all our engagements?" — and your current flat RAG system is failing them.

Task:

1. Download 20–30 freely available 10-K filings from SEC EDGAR (search for filings at sec.gov/cgi-bin/browse-edgar) for companies in two or three industry sectors. Save them as text files.

2. Build the leaf node corpus using RecursiveCharacterTextSplitter with chunk size 400. Add sector as a metadata field.

3. Run RaptorTreeBuilder.build() on your corpus. Log the number of nodes at each level and inspect the level-1 and level-2 summary text quality manually. This quality check is critical — bad summaries propagate upward.

4. Store everything in pgvector (or use Chroma locally if you don't have Postgres available — the raptor_nodes table structure maps directly to a Chroma collection with metadata fields).

5. Run all four of the following queries and record which tree levels were retrieved for each:

   • "What was [Company X]'s net revenue in fiscal 2023?"
   • "Which companies in the technology sector mentioned AI investment as a strategic priority?"
   • "Across all filings, what risk factors appear most commonly?"
   • "What are the differences in capital allocation strategy between the retail and technology sectors?"

6. For each query, compare the RAPTOR answer to what you'd get from flat leaf-level retrieval only. Note cases where RAPTOR is better and cases where it's equivalent.

Stretch goal: Integrate reranking as a post-retrieval step. Apply a cross-encoder to re-score the combined RAPTOR results before passing them to the LLM. Does reranking improve results for specific queries? For thematic queries?

---

Common Mistakes & Troubleshooting

Mistake 1: Chunking Too Large Before Tree Construction

If your leaf chunks are 2,000 tokens each, your level-1 summaries will be summarizing 10–20 chunks at a time — easily 20,000+ tokens of input. This strains the summarization model and produces summaries that are too long to be useful as intermediate nodes.

Fix: Use smaller leaf chunks (300–500 tokens) for RAPTOR than you might for standard flat RAG. The tree structure compensates for the lost context by providing summary nodes.

Mistake 2: Ignoring UMAP Instability on Small Datasets

UMAP with default settings will crash or produce nonsensical results when n_samples < n_neighbors. Our reduce_dimensions function handles this, but watch your logs for the PCA fallback warning — if you're seeing it frequently, your corpus might be too small for the RAPTOR overhead to pay off.

Fix: For corpora with fewer than 200 chunks, standard RAG with parent document retrieval is likely a better choice. RAPTOR earns its complexity at scale.

Mistake 3: Applying Metadata Filters to Summary Nodes

As mentioned earlier, summary nodes aggregate across multiple source documents. Filtering summaries by company = "Northstar" will exclude summaries that discuss Northstar in the context of broader industry trends — exactly the nodes you want for thematic queries.

Fix: In your retrieval logic, apply metadata filters only when level = 0. For higher-level nodes, retrieve without filters and let the LLM determine relevance from the content.

Mistake 4: Using the Same Summarization Model for All Levels

Using gpt-4o for every level of summarization is expensive and usually unnecessary for intermediate levels. Using gpt-4o-mini for everything sometimes produces shallow root-level summaries that fail to capture cross-document themes.

Fix: Parameterize summarization model by level. Use gpt-4o-mini for levels 1–2 and gpt-4o for levels 3+. This balances cost and quality where it matters most.

Mistake 5: Not Evaluating Tree Quality Before Production

The RAPTOR tree is only useful if the summaries are accurate and complete. Bad summaries silently degrade retrieval quality in ways that are hard to diagnose after the fact.

Fix: Before deploying, manually inspect 10–15 summary nodes at each level. Ask: does this summary accurately represent its constituent chunks? Does it preserve specific data points? Run your evaluation suite — evaluating RAG precision, recall, and faithfulness — comparing RAPTOR vs. flat retrieval on a held-out question set.

---

When to Use RAPTOR vs. Alternatives

RAPTOR is not always the right tool. Let's be honest about the tradeoffs.

Use RAPTOR when:

• Your corpus has strong thematic structure across documents (annual reports, research papers, policy documents, support case histories)
• Users regularly ask cross-document synthesis questions
• You have the budget for indexing-time LLM calls and the corpus is stable enough to justify rebuilds
• Your corpus is large enough (500+ chunks) that the tree structure adds meaningful abstraction

Consider alternatives when:

• Queries are mostly specific and localized → flat RAG with contextual compression is cheaper and simpler
• Documents have complex relational structure between entities → Graph RAG handles multi-hop reasoning better
• Your corpus updates very frequently → the rebuild cost becomes prohibitive; consider agentic RAG with self-correction instead
• You need sub-100ms retrieval latency → RAPTOR's additional nodes increase index size and can push latency up, especially during tree-traversal retrieval

---

Summary & Next Steps

You've built a complete RAPTOR pipeline: from leaf node creation through recursive clustering and summarization, into a multi-level vector index, and out through intelligent retrieval strategies that match granularity to query type.

The key ideas to carry forward:

• RAPTOR is a preprocessing investment. The indexing cost is real, but you pay it once per corpus revision. The retrieval quality improvement for thematic queries is substantial and difficult to replicate with other approaches.
• Soft GMM clustering is the right fit. The probabilistic membership ensures that cross-cutting content appears in multiple branches of the tree, improving recall for queries that span topic boundaries.
• Retrieval strategy matters as much as tree structure. Collapsed retrieval is often good enough. Layered retrieval earns its complexity when you need both overview and evidence in the same answer.
• Evaluate your tree quality independently. The tree is a data artifact, not just a pipeline step. Treat it with the same rigor you'd apply to a database schema.

Where to go next:

• Explore how adaptive retrieval can automatically select the right RAPTOR retrieval strategy based on query complexity at runtime
• Consider combining RAPTOR with hybrid search — keyword matching against the summary text can improve precision for queries with specific named entities
• If your documents have complex cross-references, look at how Graph RAG can complement RAPTOR's hierarchical structure
• For production deployment, integrate RAPTOR retrieval with your monitoring stack using the patterns in Production RAG: Caching, Monitoring, and Continuous Improvement