Capsule 11: End-to-end RAG pipeline with ChromaDB

Capsule description

Up to here you built pieces. ChromaDB with metadata filtering (capsules 04-08), embeddings with OpenAI (capsule 09), chunking with RecursiveCharacterTextSplitter (capsule 10). You tested each piece separately.

Now you put it all together into the complete RAG pipeline: you take a document, process it with chunking, embed it with OpenAI, store it in ChromaDB with metadata, retrieve the relevant chunks for a question, and generate the answer with GPT by passing those chunks as context. It's the minimal functional version — no REST API, no Docker, no testing — but it's the first time in this guide that you see the complete end-to-end system.

This capsule has a specific purpose: to make the "RAG mental model" stop being abstract. When you finish, you'll have run a system that receives a question in Spanish and returns an answer grounded in your documents, citing the sources. From there, everything you build in the rest of the guide (Landscape, Decision Matrix, Production, and the Module 8 project) has a concrete system to refer to.

By the end of this capsule, you'll be able to:

• ✅ Explain the complete RAG flow (ingestion → retrieval → generation) with each step executable
• ✅ Build a minimal RAG pipeline with ChromaDB + OpenAI in ~150 lines of code
• ✅ Design a RAG prompt that avoids hallucinations and requires citing sources
• ✅ Identify the five most common failure points in a RAG pipeline
• ✅ Differentiate which responsibility belongs to retrieval vs generation when an answer is bad

Estimated time: 45-60 minutes

The complete RAG flow in a single mental image

Before the code, the mental model:

INGESTION (once per document, or when it's updated):
  Original document
       ↓
  Chunking (M4/10)        → ["chunk1", "chunk2", "chunk3", ...]
       ↓
  OpenAI Embeddings       → [[v1...], [v2...], [v3...], ...]
       ↓
  ChromaDB.add(           → Storage + HNSW indexing
    documents=chunks,
    embeddings=vectors,
    metadatas=[{doc_id, chunk_index, source}, ...]
  )

──────────────────────────────────────────────────────────

RETRIEVAL + GENERATION (every time the user asks):
  User question
       ↓
  OpenAI embeddings of the query
       ↓
  ChromaDB.query(top_k=5) → 5 most similar chunks + metadata
       ↓
  Build prompt:            "Context: {chunks}\n\nQuestion: {query}"
       ↓
  GPT-4 with that prompt  → Answer + source citation
       ↓
  Answer to the user

The critical insight: RAG is not an "intelligent" system. It's a two-step system where each step is relatively simple:

1. Retrieval is just similarity search: "give me the 5 chunks most similar to this question".
2. Generation is just prompt engineering: "given this context and this question, write an answer".

There's no magic. If the system fails, it's failing at either retrieval (the right chunks weren't retrieved) or generation (the right chunks were retrieved but the LLM ignored information, hallucinated, or wrote poorly). Knowing how to distinguish which of the two fails is the central skill for debugging RAG in production.

Implementation: minimal viable pipeline

You'll build a RAG system over a small dataset (3-4 technical documents) so that each step is traceable. The concepts are the same when scaling to thousands of documents.

Step 1: Project setup

# Project structure
rag_pipeline/
├── .env                      # OPENAI_API_KEY=sk-...
├── .gitignore                # includes .env and chroma_db/
├── requirements.txt
├── data/
│   ├── chromadb_intro.md
│   ├── pinecone_overview.md
│   └── hnsw_explained.md
├── ingestion.py
├── rag.py
└── main.py

# requirements.txt
chromadb>=0.5.0
openai>=1.40.0
langchain-text-splitters>=0.3.0
python-dotenv>=1.0.0

pip install -r requirements.txt

Step 2: Ingestion — from document to ChromaDB

# ingestion.py
import os
from pathlib import Path
from dotenv import load_dotenv
from langchain_text_splitters import RecursiveCharacterTextSplitter
import chromadb
from chromadb.utils import embedding_functions

load_dotenv()

# Configuration (in production, this goes in config.py)
CHROMA_PATH = "./chroma_db"
COLLECTION_NAME = "rag_demo"
EMBEDDING_MODEL = "text-embedding-3-small"
CHUNK_SIZE = 500
CHUNK_OVERLAP = 50
DATA_DIR = Path("./data")


def get_collection():
    """Create or retrieve the collection with OpenAI embeddings."""
    openai_ef = embedding_functions.OpenAIEmbeddingFunction(
        api_key=os.getenv("OPENAI_API_KEY"),
        model_name=EMBEDDING_MODEL
    )
    client = chromadb.PersistentClient(path=CHROMA_PATH)
    return client.get_or_create_collection(
        name=COLLECTION_NAME,
        embedding_function=openai_ef,
        metadata={"hnsw:space": "cosine"}
    )


def ingest_file(file_path: Path, collection):
    """Process a file: chunking + metadata + insert."""
    content = file_path.read_text(encoding="utf-8")

    splitter = RecursiveCharacterTextSplitter(
        chunk_size=CHUNK_SIZE,
        chunk_overlap=CHUNK_OVERLAP,
        length_function=len,
        separators=["\n\n", "\n", ". ", " ", ""]
    )
    chunks = splitter.split_text(content)

    doc_id = file_path.stem  # name without extension
    chunk_ids = [f"{doc_id}_chunk_{i:03d}" for i in range(len(chunks))]
    chunk_metadatas = [
        {
            "doc_id": doc_id,
            "chunk_index": i,
            "total_chunks": len(chunks),
            "source": str(file_path),
            "filename": file_path.name,
        }
        for i in range(len(chunks))
    ]

    collection.add(
        documents=chunks,
        ids=chunk_ids,
        metadatas=chunk_metadatas
    )

    return len(chunks)


def ingest_directory(directory: Path = DATA_DIR):
    """Process all the .md files in the directory."""
    collection = get_collection()

    total_chunks = 0
    for file_path in directory.glob("*.md"):
        n_chunks = ingest_file(file_path, collection)
        print(f"  {file_path.name}: {n_chunks} chunks")
        total_chunks += n_chunks

    print(f"\nTotal: {total_chunks} chunks in collection '{COLLECTION_NAME}'")
    print(f"Collection size: {collection.count()} documents")


if __name__ == "__main__":
    ingest_directory()

Execution:

$ python ingestion.py
  chromadb_intro.md: 8 chunks
  pinecone_overview.md: 12 chunks
  hnsw_explained.md: 14 chunks

Total: 34 chunks in collection 'rag_demo'
Collection size: 34 documents

Step 3: Retrieval — find relevant chunks

# rag.py
import os
from dataclasses import dataclass
from openai import OpenAI
import chromadb
from chromadb.utils import embedding_functions

CHROMA_PATH = "./chroma_db"
COLLECTION_NAME = "rag_demo"
EMBEDDING_MODEL = "text-embedding-3-small"
LLM_MODEL = "gpt-4o-mini"
TOP_K = 5

openai_client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))


@dataclass
class RetrievedChunk:
    text: str
    source: str
    doc_id: str
    chunk_index: int
    distance: float


def get_collection():
    """Retrieve the already-populated collection."""
    openai_ef = embedding_functions.OpenAIEmbeddingFunction(
        api_key=os.getenv("OPENAI_API_KEY"),
        model_name=EMBEDDING_MODEL
    )
    client = chromadb.PersistentClient(path=CHROMA_PATH)
    return client.get_collection(
        name=COLLECTION_NAME,
        embedding_function=openai_ef
    )


def retrieve(query: str, top_k: int = TOP_K) -> list[RetrievedChunk]:
    """Find the top_k chunks most similar to the query."""
    collection = get_collection()
    results = collection.query(
        query_texts=[query],
        n_results=top_k,
        include=['documents', 'metadatas', 'distances']
    )

    chunks = []
    for doc, meta, dist in zip(
        results['documents'][0],
        results['metadatas'][0],
        results['distances'][0]
    ):
        chunks.append(RetrievedChunk(
            text=doc,
            source=meta['source'],
            doc_id=meta['doc_id'],
            chunk_index=meta['chunk_index'],
            distance=dist
        ))
    return chunks

Step 4: Generation — build the prompt and call GPT

Here's the part that most tutorials get wrong. The RAG prompt isn't "here's context, answer". It has three specific requirements:

1. Anti-hallucination: an explicit instruction not to invent information that isn't in the context.
2. Source citation: the answer must identify which chunk each statement came from.
3. Explicit fallback: if the context doesn't contain the answer, say so instead of inventing.

# rag.py (continued)
SYSTEM_PROMPT = """You are a technical assistant that answers questions based ONLY on the provided context.

STRICT RULES:
1. If the answer is not in the context, reply: "I don't have enough information in the documents to answer this."
2. Do NOT invent information. Do NOT use your prior knowledge.
3. Cite the sources using the format [Source: doc_id, chunk N] at the end of each important statement.
4. If there's contradictory information between chunks, mention it explicitly.
5. Be concise. The answer should be 2-4 paragraphs maximum."""


def build_user_prompt(query: str, chunks: list[RetrievedChunk]) -> str:
    """Build the prompt with the query and the retrieved chunks."""
    context_parts = []
    for i, chunk in enumerate(chunks, 1):
        context_parts.append(
            f"[Chunk {i} | doc_id: {chunk.doc_id} | chunk_index: {chunk.chunk_index}]\n"
            f"{chunk.text}\n"
        )
    context = "\n---\n".join(context_parts)

    return f"""CONTEXT:
{context}

USER QUESTION:
{query}

Answer based ONLY on the context. Cite the sources."""


@dataclass
class RagResponse:
    answer: str
    sources: list[dict]
    chunks_used: list[RetrievedChunk]
    fallback: bool


def generate(query: str, chunks: list[RetrievedChunk]) -> RagResponse:
    """Call the LLM with the built prompt and return a structured response."""
    user_prompt = build_user_prompt(query, chunks)

    response = openai_client.chat.completions.create(
        model=LLM_MODEL,
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": user_prompt}
        ],
        temperature=0.0,  # Deterministic — RAG shouldn't be "creative"
        max_tokens=600
    )

    answer = response.choices[0].message.content

    # Fallback detection (the LLM said it doesn't know)
    fallback = "i don't have enough information" in answer.lower()

    sources = [
        {
            "doc_id": chunk.doc_id,
            "source": chunk.source,
            "chunk_index": chunk.chunk_index,
            "distance": round(chunk.distance, 3)
        }
        for chunk in chunks
    ]

    return RagResponse(
        answer=answer,
        sources=sources,
        chunks_used=chunks,
        fallback=fallback
    )


def ask(query: str, top_k: int = TOP_K) -> RagResponse:
    """Complete pipeline: query → retrieve → generate → answer."""
    chunks = retrieve(query, top_k=top_k)
    return generate(query, chunks)

Step 5: Test the complete pipeline

# main.py
from rag import ask

queries = [
    "What are the recommended HNSW parameters for production?",
    "¿Cuál es la diferencia entre Pinecone y ChromaDB?",
    "How do I configure HTTPS for ChromaDB?",  # Not in the docs → should fall back
]

for query in queries:
    print(f"\n{'='*70}")
    print(f"Q: {query}")
    print('='*70)

    response = ask(query)

    print(f"\nA: {response.answer}\n")

    if response.fallback:
        print("⚠️ FALLBACK: The system didn't find enough information.")
    else:
        print(f"Sources consulted:")
        for src in response.sources[:3]:
            print(f"  - {src['doc_id']} (chunk {src['chunk_index']}, dist {src['distance']})")

Expected output:

======================================================================
Q: What are the recommended HNSW parameters for production?
======================================================================

A: For production deployments with ChromaDB, the recommended HNSW parameters
are M=32 and construction_ef=200 [Source: hnsw_explained, chunk 4]. The default
values (M=16, construction_ef=100) work for development and prototypes, but
production benefits from the higher accuracy that comes from increased graph
connectivity [Source: hnsw_explained, chunk 5].

Note that increasing M proportionally increases memory usage, so this is a
trade-off between recall and resources.

Sources consulted:
  - hnsw_explained (chunk 4, dist 0.198)
  - hnsw_explained (chunk 5, dist 0.241)
  - chromadb_intro (chunk 3, dist 0.456)


======================================================================
Q: ¿Cuál es la diferencia entre Pinecone y ChromaDB?
======================================================================

A: Pinecone is a managed service (Pinecone Inc. operates the infrastructure)
while ChromaDB is open-source and self-hosted [Source: pinecone_overview, chunk 1].
Pinecone is typically used for large-scale production with a guaranteed SLA,
while ChromaDB is ideal for local development and smaller-scale projects
[Source: chromadb_intro, chunk 2].

Sources consulted:
  - pinecone_overview (chunk 1, dist 0.234)
  - chromadb_intro (chunk 2, dist 0.298)
  - pinecone_overview (chunk 5, dist 0.412)


======================================================================
Q: How do I configure HTTPS for ChromaDB?
======================================================================

A: I don't have enough information in the documents to answer this.

⚠️ FALLBACK: The system didn't find enough information.

What this execution demonstrates:

1. Cross-language retrieval works: query 2 is in Spanish, the documents in English. OpenAI embeddings make the match correctly (this would be impossible with ChromaDB's default — see M4/09).

2. The system cites the correct sources: "[Source: hnsw_explained, chunk 4]" lets you verify the answer by opening that file and reading that chunk.

3. The fallback works: query 3 about HTTPS isn't in the docs, the system says "I don't know" instead of inventing. This is what separates a professional RAG from one that hallucinates.

How to distinguish whether retrieval or generation fails

When the system gives a bad answer, there are only two suspects. Knowing which one tells you what to fix.

Case A: retrieval fails

Symptom: the answer is generic or incorrect. The retrieved chunks (response.sources) don't contain the information the query needs.

How to detect it: look at the raw chunks:

response = ask("What's the difference between cosine and L2 distance?")
for chunk in response.chunks_used:
    print(f"--- dist {chunk.distance:.3f} ---")
    print(chunk.text[:200])

If the chunks don't talk about cosine vs L2, retrieval failed. Possible causes:

• chunk_size too large (diluted information)
• query too generic (doesn't match specific concepts)
• the document doesn't contain that information (not a retrieval problem, a coverage one)
• inadequate embeddings (the default when you should use OpenAI)

Solutions: adjust chunk_size/overlap, improve the query (query expansion), add missing documents, change the embedding model.

Case B: generation fails

Symptom: the retrieved chunks DO contain the correct answer, but the LLM's answer ignores it, hallucinates, or writes it poorly.

How to detect it: you read the chunks and the answer should be there, but it isn't.

Possible causes:

• badly designed prompt (no instruction not to hallucinate)
• high temperature (a "creative" LLM when it should be literal)
• model too small for the domain (gpt-3.5 with complex technical text)
• context too long (the LLM loses information in the middle — "lost in the middle")

Solutions: stricter prompt engineering, temperature=0, a more capable model, trim the context.

The 30-second test

When the system fails, before rewriting the prompt or adjusting chunking, ask yourself this question:

"If I, manually, read the 5 retrieved chunks and had to answer the question, could I?"

• Yes, clearly → generation fails. Fix the prompt.
• No, information is missing → retrieval fails. Fix the search.
• Maybe, it's subtly there → both. Start with retrieval (it's easier to fix and usually has more impact).

Traps and common mistakes

Trap 1: temperature greater than 0 in RAG

The mistake:

response = openai_client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[...],
    temperature=0.7  # ❌ "So the answers are natural"
)

Symptom: the answers vary between executions, the LLM adds information that isn't in the context, the source citations are sometimes invented.

Why it happens: temperature controls the randomness of the sampling. In creative generation (writing a story), high temperature is good. In RAG, where we want the LLM to stick to the context, any randomness introduces hallucinations.

How to fix it: temperature=0.0 always in RAG. You provide the "naturalness" with prompt design, not stochastic creativity.

Trap 2: context too long

The mistake:

chunks = retrieve(query, top_k=20)  # ❌ "more context = better answer"

Symptom: the answers become vaguer or ignore information that's clearly in chunks 12-15. Latency and cost increase.

Why it happens: LLMs suffer from "lost in the middle" — information in the middle of the context gets less attention than at the start or the end. Passing 20 chunks of 500 chars each = 10,000 chars of context, much of which the LLM will treat as noise.

How to fix it:

• top_k=3-5 is the optimal range for most cases.
• If you need more coverage, use re-ranking (covered in guide #8 Advanced RAG).
• Measure the quality: top_k=5 with re-ranking usually beats top_k=20 without re-ranking.

Trap 3: a prompt without an anti-hallucination instruction

The mistake: a minimalist prompt like "Context: {chunks}. Question: {query}. Answer."

Symptom: the LLM responds with general knowledge when the context doesn't have the answer. It doesn't cite sources. It mixes context information with its training data.

Why it happens: without an explicit instruction to stick to the context, GPT by default uses everything it knows.

How to fix it: an explicit instruction in the system prompt — see SYSTEM_PROMPT above with the 5 rules.

Trap 4: ignoring fallback feedback

The mistake: the system responds "I don't have enough information" and it's treated as an error in logs. The team adds try/except to "recover" from the fallback.

Symptom: you lose the system's most valuable signal — knowing which questions it can't answer.

Why it matters: frequent fallbacks in production tell you exactly which documents you're missing. It's gold-mine information for improving the dataset.

How to fix it: log fallbacks as a positive metric ("fallback rate"). When it exceeds a threshold (e.g.: >15%), review what type of queries are failing and add the corresponding documents.

Trap 5: not measuring before optimizing

The mistake: you sense the system could be better with re-ranking, hybrid search, query expansion, semantic chunking, a large model, etc. You start adding everything simultaneously.

Symptom: you don't know what improved what. Latency went up, cost went up, quality improved marginally. You roll back some things but aren't sure which.

How to fix it: build a fixed eval set (20-50 queries with annotated answers) BEFORE optimizing. Each change is measured against the eval set. If it doesn't improve, it's discarded.

# Eval set structure
EVAL_SET = [
    {
        "query": "What are the recommended HNSW parameters for production?",
        "expected_keywords": ["M=32", "construction_ef=200"],
        "expected_doc": "hnsw_explained",
    },
    # ... 30+ entries
]

def evaluate(eval_set):
    """Calculate recall, source precision, fallback rate."""
    correct_retrieval = 0
    fallback_count = 0
    for item in eval_set:
        response = ask(item["query"])
        if response.fallback:
            fallback_count += 1
        elif item["expected_doc"] in [s["doc_id"] for s in response.sources]:
            correct_retrieval += 1
    return {
        "retrieval_accuracy": correct_retrieval / len(eval_set),
        "fallback_rate": fallback_count / len(eval_set),
    }

Trap 6: badly managed API keys

The mistake: multiple files load OPENAI_API_KEY in different ways. Some hardcoded for "temporary" debugging. Others using defaults.

Symptom: intermittent "Invalid API key" errors, unexpected bills (someone used a production key for tests), keys leaked in commits.

How to fix it:

1. A single source of truth: an environment variable loaded with dotenv at one point in the program.
2. .env always in .gitignore.
3. A pre-commit hook (gitleaks or detect-secrets) that detects the sk- pattern.
4. Different keys for dev / staging / prod, periodic rotation.

Applied exercise

Scenario: the RAG pipeline you built works, but a beta tester reports the following:

"I asked 'what's the best M for HNSW if I have 100K vectors?' and it answered that the optimal M is 16, citing hnsw_explained chunk 4 as the source. But I opened that chunk and it clearly says that for production M=32 is recommended. The system is hallucinating."

Your task: investigate whether retrieval or generation fails, and propose a fix. You have access to the code and the populated ChromaDB.

response = ask("what's the best M for HNSW if I have 100K vectors?")
for i, chunk in enumerate(response.chunks_used):
    print(f"\n--- Chunk {i+1} | dist {chunk.distance:.3f} ---")
    print(f"doc: {chunk.doc_id} chunk {chunk.chunk_index}")
    print(chunk.text)

SYSTEM_PROMPT = """You are a technical assistant that answers questions based ONLY on the provided context.

ANSWER PROCESS (follow this order):
1. Identify all the data relevant to the question in the context.
2. If there are multiple possible values (e.g.: default vs production), identify which one applies to the case asked.
3. If the question has a specific context (e.g.: "for production", "with 100K vectors"), prioritize the answer for THAT case.
4. Only then write the final answer, citing sources.

STRICT RULES:
1. If the answer is not in the context, reply: "I don't have enough information in the documents to answer this."
2. Do NOT invent information. Do NOT use your prior knowledge.
3. Cite the sources using [Source: doc_id, chunk N].
4. If there are different values depending on the scenario (default vs production), mention both and which applies."""

def build_user_prompt(query: str, chunks: list[RetrievedChunk]) -> str:
    context = "\n---\n".join([
        f"[Chunk {i} | doc_id: {c.doc_id} | chunk_index: {c.chunk_index}]\n{c.text}"
        for i, c in enumerate(chunks, 1)
    ])

    return f"""CONTEXT:
{context}

USER QUESTION:
{query}

Before answering, reason out loud:
1. What data in the context is relevant to this question?
2. Is there different data for different scenarios?
3. Which scenario does the question describe?

After reasoning, give the final answer with citations."""

Summary and next step

What you learned:

• RAG is a two-step system: retrieval (searches for similar chunks) + generation (the LLM writes using those chunks as context). There's no magic.
• The minimal viable pipeline is ~150 lines of code: chunking → embeddings → ChromaDB → retrieve top-k → prompt with strict instructions → answer with citations.
• The professional RAG prompt has three requirements: anti-hallucination, source citation, an explicit fallback when there's no information.
• temperature=0 is non-negotiable in RAG. Randomness introduces hallucinations.
• When the system fails, distinguish retrieval vs generation with the "30-second test": could you answer by reading the retrieved chunks?

Checkpoint: before closing the module, you should be able to:

• Draw the complete RAG pipeline from memory, indicating what happens in ingestion vs runtime.
• Explain the three requirements of a professional RAG prompt and why each one exists.
• Diagnose a bad answer by distinguishing whether retrieval or generation fails.

Closing Module 4.

You just closed the complete Vector Databases cycle with ChromaDB. You know:

• Why you need vector databases (M1)
• How they work internally (M2)
• What features matter for RAG (M3)
• How to implement ChromaDB with CRUD, metadata, batch ingestion, optimization (M4/01-08)
• How to generate quality embeddings with OpenAI (M4/09)
• How to split long documents without losing resolution (M4/10)
• How to build the minimal end-to-end RAG pipeline (M4/11)

Next module: 5 — Vector Databases Landscape.

You implemented everything with ChromaDB. The inevitable question is: is it the right choice for your next project, or just the first one you learned? Module 5 opens the complete landscape: Pinecone, Weaviate, Qdrant, Milvus, and ChromaDB side by side. You won't install five vector databases — you'll build the criteria to choose between them with data, not preference.

Resources

1. OpenAI Chat Completions API — Official generation reference
2. Anthropic — How to make a LLM RAG more accurate — Advanced prompt techniques for RAG
3. Lost in the Middle: How Language Models Use Long Contexts (paper) — Why top_k=20 isn't always better
4. LangChain RAG documentation — Implementation with an alternative framework
5. Ragas — RAG evaluation framework — For building systematic eval sets (covered in guide #12 Evaluation Frameworks)
6. Prompt Engineering Guide — RAG — Effective prompt patterns for RAG

Estimated time: 45-60 minutes Next module: 01-module-introduction-5.md