--- title: "RAG Pipeline" description: "RAG is the single most-asked-about AI Engineer skill in job descriptions today — this is the project that lets you speak to it with specifics, not buzzwords." source: "https://confidentprep.com/paths/ai-engineering-fundamentals/rag-pipeline/" path: "AI Engineering Fundamentals" repository: "https://github.com/confidentprep14-web/ai-engineering-projects/tree/main/rag-pipeline" token_estimate: 5305 --- # RAG Pipeline ## Overview Document Chat proved an LLM can answer questions about your data — as long as everything fits in one request's context window. This project removes that ceiling: instead of resending a whole document on every question, you build a real retrieval pipeline that finds and injects only the relevant pieces of a much larger document set, on demand. ### What you're building `rag-pipeline` ingests a folder of documents, chunks and embeds them into a persistent vector index, and at query time retrieves only the chunks above a similarity-score threshold — not the whole corpus. It ships with an **A/B mode** that runs the same question through the model with and without retrieved context, so you can see exactly what retrieval is buying you on a question-by-question basis, instead of assuming it's working because the answers sound reasonable. ### The RAG pipeline ![RAG pipeline flow: documents are chunked with overlap and embedded into a persistent index, while queries are embedded and matched against that index by top-k cosine similarity above a score threshold, then either augmented into a grounded generation or answered honestly as not found, with an A/B comparison against a no-retrieval answer](/diagrams/rag-pipeline-flow.svg) *Ingestion (top) and querying (bottom) are two separate paths that share one index. A stale index — documents changed without re-running the top path — is one of the most common ways this pipeline quietly starts giving wrong answers.* ### Core concepts, three levels deep #### 1. Embeddings - **Definition:** a list of numbers (a vector) that encodes the *meaning* of a piece of text, not its exact wording. Texts with similar meaning produce numerically close vectors, which is what makes searching by meaning — instead of by keyword — possible at all. - **In this project:** every chunk you ingest, and every query a user sends, gets converted to a vector before anything else happens. Retrieval is entirely a numeric comparison over these vectors — the model never re-reads your raw documents at query time, it reads whatever chunks the vector comparison surfaced. - **Practical consequence:** two phrases that share zero words ("reset password" and "account recovery") can still retrieve correctly, because embeddings capture intent, not literal text overlap. That's the entire reason this beats a keyword search. #### 2. Chunking and chunk overlap - **Definition:** splitting a document into smaller pieces so each one covers roughly one topic and fits comfortably in a retrieval result — typically 300-500 tokens, with a 50-token overlap buffer at each boundary. - **In this project:** the chunk size and overlap you choose directly determine what's retrievable. A fact that straddles two chunks with no overlap can end up incomplete in both of them — retrievable in neither. - **Practical consequence:** when retrieval looks wrong, check chunking before you touch anything else. Bad chunk boundaries are one of the most common — and most fixable — causes of a RAG system returning incomplete or off-topic results. #### 3. Retrieval: cosine similarity, top-k, and score thresholds - **Definition:** cosine similarity ranks chunks by the *angle* between their vector and the query's vector (direction, not magnitude); top-k returns the k highest-scoring chunks; a score threshold discards anything below a "similar enough" cutoff even if it's in the top-k. - **In this project:** these three settings together decide what actually reaches the model. Top-k alone can return weak matches if nothing in the index is truly relevant — the score threshold is what lets the pipeline say "nothing in my documents answers this" instead of forcing a bad match into the prompt. - **Practical consequence:** cosine similarity — not Euclidean distance — is the right metric here specifically because it isn't skewed by chunk length, so a short chunk and a long chunk about the same topic can still score as equally relevant. #### 4. Grounding and index freshness - **Definition:** grounding is instructing the model to answer only from retrieved context (and say so when it can't); index freshness is whether the vector store reflects the current state of your source documents. - **In this project:** without an explicit "answer only from the provided context" instruction, the model can quietly fall back on training memory instead of your actual documents — which defeats the entire point of building this pipeline. Separately, if a source document changes and you don't re-embed it, the old chunk stays in the index and keeps getting served as if it were current. - **Practical consequence:** treat re-indexing after a document update as part of running this pipeline, not an optional maintenance step — a stale embedding is a wrong answer delivered with full confidence, which is exactly the failure mode citations and grounding are supposed to prevent. ### Decision rules | If... | Then... | |---|---| | A document set comfortably fits in one context window and rarely changes | Stuff it into context directly — that's Project 1's approach; you don't need retrieval infrastructure for this | | The document set is large, spans many files, or needs to scale past one context window | Build retrieval (this project) — chunk, embed, index, and retrieve only what's relevant per query | | Source data changes frequently and answers need a traceable source | Choose RAG over fine-tuning — no retraining needed, and the index can be updated independently of the model | | You need to change the model's writing style, tone, or domain vocabulary, not its knowledge | Choose fine-tuning over RAG — that's a behavior change, not a knowledge-injection problem | | Retrieval quality degrades as the corpus scales up | Add metadata filtering before similarity search — don't just raise top-k, which gets noisier, not better, at scale | ### Common mistakes - **Zero or too-small chunk overlap**, silently splitting the sentence that contains the answer across two incomplete chunks. - **Skipping the "answer only from context" instruction**, so the model quietly falls back to training memory instead of the chunks you actually retrieved — defeating the point of retrieval. - **Never re-indexing after a document update**, so the pipeline keeps confidently citing outdated content indefinitely. - **Reaching for a bigger top-k instead of metadata filters** when retrieval quality drops at scale — a larger candidate pool without narrowing it first just adds more noise. ### Key concepts at a glance | Concept | One-line definition | Why it matters for `rag-pipeline` | |---|---|---| | Embedding | A vector that encodes the meaning of a piece of text | The unit everything in this pipeline — chunks and queries alike — is compared as | | Chunking + overlap | Splitting documents into retrievable pieces with a boundary buffer | Determines whether a fact is actually retrievable at all | | Cosine similarity / top-k / score threshold | Angle-based ranking, capped at k results, filtered by a relevance cutoff | The exact mechanism this project's retrieval step runs on | | Index freshness | Whether the vector store reflects the current source documents | Why re-indexing on update is part of operating this pipeline, not optional maintenance | ## What to do **Path:** 1 **Position:** 3 of 12 **Difficulty:** 🟢 Local — one API key (or Ollama), no Docker, no cloud **Estimated time:** 4–5 hours **AWS cost:** None --- ### Agent Pickup Instructions **This spec is self-contained. Build this project without reading any other spec file.** ```bash mkdir -p path-1/p1-03-rag-pipeline cd path-1/p1-03-rag-pipeline # Create folder structure, paste .env.example and requirements.txt from this spec # Implement src/ modules as described below # Verify: pytest tests/ -v python src/main.py --index sample_docs/ # builds FAISS index python src/main.py --query "What is X?" # answers with RAG python src/main.py --ab "What is X?" # shows RAG vs no-RAG side by side ``` **Done when:** - [ ] `pytest tests/ -v` → minimum 5 tests, all green - [ ] Index persists to disk (re-run without `--index` should not re-embed) - [ ] `--ab` mode prints two answers side by side with retrieval scores - [ ] "I don't know" response when best score is below threshold - [ ] Retrieval latency and generation latency logged separately - [ ] No file in `src/` exceeds 200 lines --- ### What this project is A retrieval-augmented generation pipeline that embeds a folder of documents into a FAISS vector index, retrieves semantically similar chunks for each question, and augments the LLM prompt with those chunks. The learner builds this from scratch — embeddings, index persistence, score thresholding, and an A/B mode that places RAG and no-RAG answers side by side so the improvement is visible. This is the correct solution to the keyword retrieval limitation felt in p1-01. --- ### What the learner achieves "I built a RAG pipeline with a persistent FAISS index, score-threshold filtering that returns 'I don't know' rather than hallucinating, and an A/B mode that shows measurable quality improvement over keyword retrieval — with retrieval latency under 50ms on a 1,000-document corpus." --- ### Folder structure ``` p1-03-rag-pipeline/ ├── README.md ├── GUIDE.md ├── .env.example ├── requirements.txt ├── sample_docs/ │ ├── doc1.txt ← 500-word doc about topic A (include in project) │ ├── doc2.txt ← 500-word doc about topic B │ └── doc3.txt ← 500-word doc about topic C ├── src/ │ ├── main.py ← CLI entry point │ ├── llm.py ← provider-agnostic LLM wrapper │ ├── embedder.py ← sentence-transformers embedding │ ├── indexer.py ← FAISS index build/load/search │ └── rag.py ← RAG pipeline orchestration ├── tests/ │ └── test_rag.py └── .index/ ← created at runtime, gitignored ├── index.faiss └── metadata.json ``` --- ### .env.example ```bash # LLM provider: anthropic | openai | ollama LLM_PROVIDER=anthropic # API key (blank for ollama) LLM_API_KEY= # Model # anthropic → configured Anthropic model # openai → current OpenAI model # ollama → llama3.2 LLM_MODEL= # Embedding model (runs locally, no API key needed) EMBEDDING_MODEL=all-MiniLM-L6-v2 # Number of chunks to retrieve per query TOP_K=3 # Minimum similarity score (0.0–1.0) — below this, return "I don't know" SCORE_THRESHOLD=0.3 # Chunk size in characters for document splitting CHUNK_SIZE=800 # Chunk overlap in characters CHUNK_OVERLAP=150 ``` --- ### requirements.txt ``` anthropic==0.40.0 openai==1.58.0 ollama==0.4.4 sentence-transformers==3.3.1 faiss-cpu==1.9.0 python-dotenv==1.0.1 ``` --- ### src/ — what to implement #### src/llm.py Non-streaming completion. Same pattern as p1-01: **`get_completion(prompt: str, system: str = "") -> str`** - Routes to anthropic / openai / ollama based on `LLM_PROVIDER` env var - Raises `ValueError` for unknown provider #### src/embedder.py **`load_embedding_model() -> SentenceTransformer`** - Loads model from `EMBEDDING_MODEL` env var (default: `all-MiniLM-L6-v2`) - Model downloads to local cache on first run, loads from cache on subsequent runs - Print `Loading embedding model: {model_name}...` only if not already cached **`embed_texts(texts: list[str], model: SentenceTransformer) -> np.ndarray`** - Returns float32 numpy array of shape `(len(texts), embedding_dim)` - Batch encode — pass all texts at once, not one by one **`embed_query(query: str, model: SentenceTransformer) -> np.ndarray`** - Returns float32 numpy array of shape `(1, embedding_dim)` #### src/indexer.py **`chunk_text(text: str, source: str, chunk_size: int, overlap: int) -> list[dict]`** - Splits text into overlapping chunks - Each chunk: `{"text": "...", "source": "filename.txt", "chunk_index": N}` **`build_index(doc_dir: str, index_dir: str, model: SentenceTransformer) -> tuple[faiss.Index, list[dict]]`** - Walk `doc_dir`, read all `.txt` and `.md` files - Chunk each file, embed all chunks in one batch - Build `faiss.IndexFlatIP` (inner product — cosine if vectors are normalized) - Normalize all embedding vectors before adding to index - Save index: `faiss.write_index(index, f"{index_dir}/index.faiss")` - Save metadata: `json.dump(metadata_list, open(f"{index_dir}/metadata.json", "w"))` - Print: `Indexed N chunks from M files` **`load_index(index_dir: str) -> tuple[faiss.Index, list[dict]]`** - Load index.faiss and metadata.json from `index_dir` - Raises `FileNotFoundError` with message "No index found at {index_dir} — run with --index first" if missing **`search(query_embedding: np.ndarray, index: faiss.Index, metadata: list[dict], top_k: int, score_threshold: float) -> list[dict]`** - Normalize query embedding; call `index.search()` - Filter: keep only results where score >= threshold - Return list of dicts: `[{"text": "...", "source": "...", "chunk_index": N, "score": 0.85}, ...]` - Return empty list if nothing passes threshold #### src/rag.py **`build_rag_prompt(query: str, retrieved_chunks: list[dict]) -> str`** - If chunks empty: return prompt that tells LLM "no relevant context — say I don't know" - Else: format context as `[Source: {source}, Chunk {N}, Score: {score:.2f}]\n{text}\n` - Prompt instructs: cite sources in answers, say "Not found in documents" if context insufficient **`answer_with_rag(query: str, index: faiss.Index, metadata: list[dict], model: SentenceTransformer) -> dict`** - Returns `{"answer": "...", "sources": [...], "retrieval_ms": N, "generation_ms": N, "chunks_retrieved": N}` - Time retrieval separately from generation **`answer_without_rag(query: str) -> dict`** - Returns `{"answer": "...", "generation_ms": N}` - Simple LLM call with no context, same question #### src/main.py CLI: `python src/main.py [--index ] [--query "..."] [--ab "..."] [--reindex]` - `--index `: build or rebuild index from dir, save to `.index/` - `--query "..."`: load existing index, answer with RAG, print answer + sources + latency - `--ab "..."`: answer same query with RAG and without RAG, print side-by-side comparison - `--reindex`: force rebuild even if index exists - Interactive mode (no flags): loop asking for queries using loaded index Index directory defaults to `.index/` relative to cwd. --- ### tests/ — what to test **File:** `tests/test_rag.py` — use small in-memory data, avoid file I/O where possible. No LLM calls. **Test 1 — chunk_text produces correct overlap:** Call `chunk_text("A" * 2000, "test.txt", chunk_size=500, overlap=100)`. Assert adjacent chunks share 100 characters. **Test 2 — embed_texts returns correct shape:** Use a real `SentenceTransformer("all-MiniLM-L6-v2")` (loaded once via fixture). Embed 3 texts. Assert output shape is `(3, 384)`. **Test 3 — search returns empty list below threshold:** Build a tiny FAISS index with 2 chunks. Search with threshold=0.99 (unreachably high). Assert result is `[]`. **Test 4 — search returns results above threshold:** Build index, embed a query that closely matches one chunk. Search with threshold=0.1. Assert at least one result returned with score >= 0.1. **Test 5 — build_rag_prompt handles empty chunks:** Call `build_rag_prompt("What is X?", [])`. Assert returned string contains "don't know" or "no context" (case-insensitive). **Test 6 — load_index raises FileNotFoundError when missing:** Call `load_index("/nonexistent/path")`. Assert raises `FileNotFoundError`. --- ### README.md content ```markdown # RAG Pipeline Embeds a folder of documents into a persistent FAISS index and answers questions by retrieving semantically relevant chunks — with score-threshold filtering and an A/B mode to compare RAG vs no retrieval. ## Setup ```bash cd p1-03-rag-pipeline cp .env.example .env ## Edit .env: set LLM_PROVIDER and LLM_API_KEY pip install -r requirements.txt ## First run downloads the embedding model (~90MB) — cached after that ``` ## Run ```bash ## Step 1: Build the index python src/main.py --index sample_docs/ ## Step 2: Ask a question python src/main.py --query "What is the main difference between X and Y?" ## Step 3: Compare RAG vs no-RAG python src/main.py --ab "Explain the authentication flow" ``` Expected output: ``` Indexed 24 chunks from 3 files Retrieving... 12ms Retrieved 3 chunks (scores: 0.81, 0.74, 0.61) Answer: The authentication flow uses JWT tokens issued at /auth/login [Source: doc2.txt, Chunk 4, Score: 0.81]. Tokens expire after 24 hours [Source: doc2.txt, Chunk 5, Score: 0.74]. ⏱ Retrieval: 12ms | Generation: 1,840ms ``` ## Tests ```bash pytest tests/ -v ``` Tests verify chunking, embedding shape, FAISS search, and threshold filtering. Embedding model is downloaded once and cached — subsequent test runs are fast. ## What to try next - Add PDF support (see p1-01 for the pdfplumber loader pattern) - Try a code-tuned embedding model (microsoft/codebert-base) on a code corpus - Increase TOP_K and see if answer quality improves or degrades ``` --- ### GUIDE.md content ```markdown # Build guide: RAG Pipeline ## What you're building and why it matters RAG is the pattern that made LLMs useful for enterprise applications. Instead of retraining a model on your data (expensive, slow), you embed your documents into a vector store and inject the relevant pieces at query time. Every enterprise search product, every "chat with your docs" feature, every AI customer support agent uses this pattern. The key innovation over keyword search is semantic similarity: a query about "login problems" retrieves a chunk about "authentication failures" even though the words don't overlap. ## The decision that matters in this build **Score thresholding.** Without a minimum score, FAISS always returns top-k results even when none of them are relevant. A question about "climate policy" will retrieve chunks from a software documentation corpus — they're just the least-bad matches. The score threshold prevents confident wrong answers. Set it too high and you get "I don't know" on questions you can answer. Set it too low and you hallucinate. Start at 0.3 for cosine similarity with all-MiniLM-L6-v2 and tune by testing 10 questions you know the answers to. ## What will break **Embeddings must be normalized before indexing.** If you use `IndexFlatIP` (inner product) without normalizing, the scores won't represent cosine similarity and will be meaningless. Normalize with `faiss.normalize_L2(embeddings)` before adding to the index, and normalize the query vector before searching. **Chunking cuts mid-sentence.** The answer to a question often spans a sentence boundary that falls at a chunk boundary. Add overlap. Without it, you'll notice that the retrieved chunk "almost" has the answer but the critical part is in the next chunk. **Re-embedding on every run is slow.** Build the index once, persist it. The index check should be: if `.index/index.faiss` exists and no `--reindex` flag, load it — don't rebuild. This becomes critical when indexing hundreds of documents. ## How to talk about this in an interview "I built a RAG pipeline that embeds documents with sentence-transformers locally, stores them in a FAISS index, and retrieves using cosine similarity with a tunable score threshold. The threshold is what separates a real RAG system from a naive one: below 0.3, I return 'I don't know' rather than hallucinating an answer from irrelevant chunks. I measured retrieval latency at under 20ms for a 500-chunk corpus — generation was always the bottleneck." ``` --- ### Failure modes to handle | Failure | Where | How to handle | |---------|-------|---------------| | Index not found | `indexer.load_index()` | `raise FileNotFoundError("No index at {dir} — run with --index first")` | | Empty document directory | `indexer.build_index()` | `raise ValueError("No .txt or .md files found in {doc_dir}")` | | Score below threshold | `indexer.search()` | Return empty list; `rag.py` handles with "I don't know" prompt | | Embedding model download fails | `embedder.load_embedding_model()` | Let exception propagate; user sees HuggingFace error | | LLM timeout | `llm.get_completion()` | Print "LLM timeout — try again or check API status", re-raise | --- ### The metric this project measures **Retrieval latency (ms) and generation latency (ms)** — printed separately after every answer. **Retrieval score of top chunk** — printed so learner can tune `SCORE_THRESHOLD`. Format: `⏱ Retrieval: Xms (top score: 0.81) | Generation: Yms` Target: retrieval < 50ms for <1,000 chunks on a laptop CPU. ### Model source currency - Anthropic model list: https://docs.anthropic.com/en/docs/about-claude/models/overview - OpenAI model list: https://platform.openai.com/docs/models - verifiedOn: 2026-08-07 - `LLM_MODEL` is required and has no implicit default; choose a supported model for the configured provider at setup time. ## Source code A reference implementation of RAG Pipeline — README, guide, source and tests — is at https://github.com/confidentprep14-web/ai-engineering-projects/tree/main/rag-pipeline.