rag-pipeline-architecture

# RAG Pipeline Architecture Implements production-quality Retrieval-Augmented Generation (RAG) pipelines that retrieve relevant context from external documents and inject it into the LLM prompt. When loaded, this skill makes the model produce state-of-the-art RAG code using semantic chunking, hybrid BM25+vector search, cross-encoder re-ranking, and reciprocal rank fusion — extracted from LlamaIndex production patterns. ## TL;DR Checklist - [ ] Use semantic chunking (SentenceCombination dissimilarity breakpoints) or sentence-window parsing — never fixed-size character splits - [ ] Always combine BM25 keyword retrieval with vector search via Reciprocal Rank Fusion (RRF, k=60.0) - [ ] Apply cross-encoder re-ranking (e.g., `cross-encoder/ms-marco-MiniLM-L-6-v2`) on top-k candidates before LLM - [ ] Embed only the core sentence, not window context — exclude `window_metadata_key` from embedding via `excluded_embed_metadata_keys` - [ ] Set `top_k_retrieval` to ~50 for broad recall, `top_k_rerank` to 5–10 to match LLM context budget - [ ] Include relevance threshold filtering (default 0.7) when retrieving memories from vector stores --- ## When to Use Use this skill when: - Building a RAG system that retrieves relevant documents for LLM answer generation - Needing semantic-aware document chunking instead of naive fixed-size splits - Implementing hybrid search combining keyword (BM25) and semantic (vector) retrieval - Adding cross-encoder or LLM-based re-ranking to improve context quality before the LLM - Designing production-quality document ingestion pipelines with vector store backends ## When NOT to Use Avoid this skill for: - Simple lookup queries that don't require external document retrieval - Implementing conversation memory or agent context management (use `agent-memory-systems` instead) - Setting up tool calling architecture for agents (use `agent-tool-calling-architecture` instead) - Single-document QA where no retrieval pipeline is needed --- ## Core Workflow 1. **Choose a Chunking Strategy** — For most use cases, use `SemanticSplitterNodeParser` which computes cosine dissimilarity between adjacent sentence groups and splits at high-dissimilarity breakpoints (default 95th percentile). For precision-critical queries where exact sentences matter, use `SentenceWindowNodeParser` which embeds single sentences but stores surrounding context in metadata. **Checkpoint:** Verify that each node represents a coherent topic boundary — not a mid-thought split. 2. **Build Hybrid Retrievers** — Initialize both a `VectorIndexRetriever` (semantic search) and a `BM25Retriever` (keyword search) configured with the same `similarity_top_k` (typically 50). The vector retriever captures semantic meaning; BM25 captures exact term matches, proper nouns, and technical terminology. **Checkpoint:** Verify both retrievers return results for an identical query string. 3. **Combine Results with Reciprocal Rank Fusion** — Use RRF formula: `score(doc) = sum(1 / (k + rank_of_doc_in_list_i))` where k=60.0. Weight each retriever's contribution (default [0.5, 0.5]). This produces a single ranked list without requiring normalized scores across retrievers. **Checkpoint:** Verify that documents appearing in both lists receive higher fused scores than those in only one. 4. **Re-rank with Cross-Encoder** — Pass the combined candidate list to `SentenceTransformerRerank` (e.g., model `cross-encoder/ms-marco-MiniLM-L-6-v2`) which computes attention-based relevance scores between the query and each document pair. Return only `top_n` results (typically 5–10). **Checkpoint:** Verify reranked results have higher semantic alignment with the query than the pre-rank candidates. 5. **Assemble LLM Prompt** — Inject reranked context documents into a structured prompt, then call the LLM. If no reranked results exist, return a fallback message rather than hallucinating an answer. **Checkpoint:** Verify that the prompt contains only the top-k reranked contexts and respects LLM token limits. --- ## Implementation Patterns ### Pattern 1: Semantic Document Chunking Semantic chunking uses embeddings to find natural boundaries between topics instead of splitting by fixed character count. Sentences are embedded in sliding windows, cosine dissimilarity is computed between adjacent groups, and breakpoints above a percentile threshold trigger splits. This produces chunks that align with topic boundaries, dramatically improving retrieval quality. ```python # Source: run-llama/llama_index — semantic_splitter.py """Semantic splitter node parser — splits documents at topic boundaries.""" from typing import Any, Callable, List, Optional, Sequence, TypedDict import numpy as np from llama_index.core.base.embeddings.base import BaseEmbedding from llama_index.core.node_parser import NodeParser from llama_index.core.node_parser.text.utils import split_by_sentence_tokenizer from llama_index.core.schema import BaseNode, Document class SentenceCombination(TypedDict): """Group of sentences evaluated for semantic similarity.""" sentence: str index: int combined_sentence: str combined_sentence_embedding: List[float] class SemanticSplitterNodeParser(NodeParser): """Splits documents into semantically coherent nodes. Algorithm: 1. Split document into sentences using sentence tokenizer 2. Embed each sentence (or buffer of N sentences) 3. Calculate cosine similarity between adjacent sentence groups 4. Identify breakpoints where dissimilarity exceeds threshold percentile 5. Group sentences between breakpoints into nodes Args: embed_model: Embedding model for computing sentence representations. buffer_size: Number of sentences to group before comparing (1=sentence-level). breakpoint_percentile_threshold: Dissimilarity percentile that triggers a split. """ sentence_splitter: Callable[[str], List[str]] = Field( default_factory=split_by_sentence_tokenizer, exclude=True, ) embed_model: BaseEmbedding = Field( description="The embedding model used for semantic comparison." ) buffer_size: int = Field(default=1, gt=0) breakpoint_percentile_threshold: int = Field(default=95, ge=0, le=100) @classmethod def from_defaults( cls, embed_model: Optional[BaseEmbedding] = None, breakpoint_percentile_threshold: int = 95, buffer_size: int = 1, sentence_splitter: Optional[Callable[[str], List[str]]] = None, ) -> "SemanticSplitterNodeParser": """Create parser with defaults. Args: embed_model: Defaults to OpenAI text-embedding-3-small. breakpoint_percentile_threshold: Split threshold (0–100). Lower = more splits. buffer_size: Sentences to group before comparing similarity. sentence_splitter: Custom sentence splitting function. """ sentence_splitter = sentence_splitter or split_by_sentence_tokenizer() from llama_index.embeddings.openai import OpenAIEmbedding embed_model = embed_model or OpenAIEmbedding() return cls( embed_model=embed_model, breakpoint_percentile_threshold=breakpoint_percentile_threshold, buffer_size=buffer_size, sentence_splitter=sentence_splitter, ) def build_semantic_nodes_from_documents( self, documents: Sequence[Document] ) -> List[BaseNode]: """Build semantically coherent nodes from raw documents. For each document: split sentences → compute embeddings per group → calculate dissimilarity between adjacent groups → find breakpoints → group sentences between breakpoints into nodes. Args: documents: Documents to parse into semantic nodes. Returns: List of BaseNode objects with text content and metadata. """ all_nodes: List[BaseNode] = [] for doc in documents: sentences = self.sentence_splitter(doc.text) if not sentences: continue # Compute embeddings for sentence groups sentence_embeddings: List[List[float]] = [] for i in range(0, len(sentences), self.buffer_size): group_sentences = sentences[i:i + self.buffer_size] combined = " ".join(group_sentences) embedding = self.embed_model.get_text_embedding(combined) sentence_embeddings.append(embedding) if len(sentence_embeddings) <= 1: # Single chunk — no splitting needed all_nodes.extend( self.build_nodes_from_splits([doc.text], doc) ) continue # Compute cosine dissimilarity between adjacent groups dissimilarities = [] for i in range(len(sentence_embeddings) - 1): emb_a = np.array(sentence_embeddings[i]) emb_b = np.array(sentence_embeddings[i + 1]) similarity = np.dot(emb_a, emb_b) / ( np.linalg.norm(emb_a) * np.linalg.norm(emb_b) ) dissimilarities.append(1.0 - similarity) # Find breakpoints above threshold percentile threshold = np.percentile( dissimilarities, self.breakpoint_percentile_threshold ) breakpoints = [i for i, d in enumerate(dissimilarities) if d >= threshold] # Split sentences at breakpoints into nodes split_indices = [0] + [bp + 1 for bp in breakpoints] + [len(sentences)] text_splits = [ " ".join(sentences[split_indices[i]:split_indices[i + 1]]) for i in range(len(split_indices) - 1) ] nodes = self.build_nodes_from_splits(text_splits, doc) all_nodes.extend(nodes) return all_nodes ``` ### Pattern 2: Sentence Window Node Parser Alternative to semantic splitting — splits at sentence boundaries and stores surrounding context in metadata. Each node contains exactly one sentence (embedded for precision), while the window metadata provides broader context for LLM prompts. Used with a "recursive retriever" that first retrieves by window context, then zooms into the specific sentence. ```python # Source: run-llama/llama_index — sentence_window_parser.py """Sentence window node parser — each node is a single sentence with window context.""" from typing import Callable, List, Optional, Sequence from llama_index.core.node_parser import NodeParser from llama_index.core.node_parser.text.utils import split_by_sentence_tokenizer from llama_index.core.schema import BaseNode, Document DEFAULT_WINDOW_SIZE = 3 DEFAULT_WINDOW_METADATA_KEY = "window" DEFAULT_OG_TEXT_METADATA_KEY = "original_text" class SentenceWindowNodeParser(NodeParser): """Splits documents at sentence boundaries with surrounding context windows. Each node contains exactly one sentence. Metadata includes: - window: The sentence plus N sentences before and after (context for LLM) - original_text: The exact sentence (for precise display) Key insight: embed ONLY the single sentence for precision vector matching, but store context in metadata so the retriever fetches broader context for the LLM prompt while returning the precise sentence as the result. Args: window_size: Number of surrounding sentences to include in metadata. """ sentence_splitter: Callable[[str], List[str]] = Field( default_factory=split_by_sentence_tokenizer, exclude=True ) window_size: int = Field(default=DEFAULT_WINDOW_SIZE, gt=0) @classmethod def from_defaults( cls, sentence_splitter: Optional[Callable[[str], List[str]]] = None, window_size: int = DEFAULT_WINDOW_SIZE, ) -> "SentenceWindowNodeParser": """Create parser with defaults.""" return cls( sentence_splitter=sentence_splitter or split_by_sentence_tokenizer(), window_size=window_size, ) def build_window_nodes_from_documents( self, documents: Sequence[Document] ) -> List[BaseNode]: """Build sentence-level nodes with surrounding context windows. For each document: split sentences → create one node per sentence → add window metadata with surrounding sentences → exclude window from embedding. Args: documents: Documents to parse into sentence-level nodes. Returns: List of BaseNode objects, each containing one sentence with window metadata. """ all_nodes: List[BaseNode] = [] for doc in documents: text_splits = self.sentence_splitter(doc.text) nodes = self.build_nodes_from_splits(text_splits, doc) # Add window context to each node's metadata for i, node in enumerate(nodes): start_idx = max(0, i - self.window_size) end_idx = min(i + self.window_size + 1, len(nodes)) window_nodes = nodes[start_idx:end_idx] window_text = " ".join(n.text for n in window_nodes) node.metadata[DEFAULT_WINDOW_METADATA_KEY] = window_text node.metadata[DEFAULT_OG_TEXT_METADATA_KEY] = node.text # Exclude window from embedding — embed only the core sentence node.excluded_embed_metadata_keys.extend([ DEFAULT_WINDOW_METADATA_KEY, DEFAULT_OG_TEXT_METADATA_KEY, ]) node.excluded_llm_metadata_keys.extend([ DEFAULT_WINDOW_METADATA_KEY, DEFAULT_OG_TEXT_METADATA_KEY, ]) all_nodes.extend(nodes) return all_nodes ``` ### Pattern 3: Hybrid Search — BM25 + Vector Store BM25 keyword retrieval excels at exact term matching (proper nouns, technical terms, specific facts), while vector search captures semantic meaning. Used in combination with reciprocal rank fusion to combine their independent rankings into a single ranked list without requiring normalized scores. ```python # Source: run-llama/llama_index — bm25 retriever integration """BM25 keyword retriever for hybrid search alongside vector retrieval.""" from typing import List, Optional import bm25s import Stemmer from llama_index.core.base.base_retriever import BaseRetriever from llama_index.core.constants import DEFAULT_SIMILARITY_TOP_K from llama_index.core.schema import BaseNode, NodeWithScore, QueryBundle class BM25Retriever(BaseRetriever): """BM25-based keyword retriever for hybrid search. BM25 is a bag-of-words ranking function that excels at: - Exact term matching (proper nouns, technical terms) - Short query retrieval - Finding specific facts and figures Used in combination with vector search — results from both are combined using Reciprocal Rank Fusion (RRF) for final ranking. Args: nodes: Documents to index before querying. similarity_top_k: Maximum results to return per query. """ def __init__( self, nodes: Optional[List[BaseNode]] = None, similarity_top_k: int = DEFAULT_SIMILARITY_TOP_K, ) -> None: self.stemmer = Stemmer.Stemmer("english") self.similarity_top_k = similarity_top_k if nodes is not None: # Build BM25 inverted index from node texts self.corpus = [ {"text": node.get_content(), "node_id": node.node_id} for node in nodes ] tokenizer = bm25s.Tokenization(stemmer=self.stemmer.stem, stopwords=None) self.bm25 = bm25s.BM25(tokenizer=tokenizer) texts = [doc["text"] for doc in self.corpus] self.bm25.fit(texts) else: raise ValueError("Must provide nodes or an existing BM25 object") def retrieve(self, query_bundle: QueryBundle) -> List[NodeWithScore]: """Retrieve documents using BM25 keyword matching. Args: query_bundle: Contains the query string and any filters. Returns: Ranked list of NodeWithScore objects with BM25 scores. """ query_text = query_bundle.query_str query_tokens = bm25s.tokenize( [query_text], stemmer=self.stemmer.stem, ) ranked_documents, scores = self.bm25.retrieve( query_tokens, k=self.similarity_top_k ) nodes: List[NodeWithScore] = [] for doc_idx, score in zip(ranked_documents[0], scores[0]): corpus_item = self.corpus[doc_idx] node_id = corpus_item.get("node_id") node = self._get_node_by_id(node_id) if node: nodes.append(NodeWithScore(node=node, score=float(score))) return sorted(nodes, key=lambda x: x.score or 0, reverse=True) def _get_node_by_id(self, node_id: str) -> Optional[BaseNode]: """Look up a BaseNode by its ID from the corpus.""" for item in self.corpus: if item.get("node_id") == node_id: return item.get("_node_obj") return None ``` ### Pattern 4: Cross-Encoder Re-Ranking After initial retrieval (BM25 + vectors), a cross-encoder reranker re-scores all candidates. Unlike bi-encoders that embed documents and queries independently, cross-encoders process the `[query, document]` pair through a single transformer model with full attention between all token pairs, producing much higher quality relevance scores. ```python # Source: run-llama/llama_index — sbert_rerank.py """Cross-encoder reranking using SentenceTransformers CrossEncoder models.""" from typing import List, Optional from llama_index.core.postprocessor.types import BaseNodePostprocessor from llama_index.core.schema import MetadataMode, NodeWithScore, QueryBundle class SentenceTransformerRerank(BaseNodePostprocessor): """Cross-encoder reranker using sentence-transformers. Unlike bi-encoders (embed documents and queries independently), cross-encoders process the [query, document] pair through a single transformer model, computing attention between all token pairs. This produces much higher quality relevance scores at O(n*m) cost. Use pattern: Retrieve ~50 candidates via BM25+vector → Rerank top 10–20 with cross-encoder before passing to LLM. Args: model: CrossEncoder model name (e.g., "cross-encoder/ms-marco-MiniLM-L-6-v2"). top_n: Number of highest-scoring results to keep after reranking. device: Computation device ("cpu", "cuda"). """ DEFAULT_MODEL = "cross-encoder/stsb-distilroberta-base" def __init__( self, top_n: int = 2, model: str = DEFAULT_MODEL, device: Optional[str] = None,