Module 1: The Complete RAG Pipeline (Architecture Overview)

RAG Pipeline Components

Next →

Capsule overview

RAG isn't a magic monolith. It's a pipeline with 4 distinct components that run in sequence: Indexing (preparing documents), Retrieval (finding relevant information), Generation (creating the answer), and Evaluation (measuring quality). Each component has its own technical decisions and optimization points.

Understanding these components is critical because the advanced techniques you'll learn in modules 2-8 apply to specific components: chunking optimizes Indexing, re-ranking improves Retrieval, prompt engineering affects Generation, and RAGAS measures Evaluation. Without understanding where each technique goes, you'd be applying them at random.

This capsule gives you the detailed map of each component: what it does, what decisions you make, where the advanced techniques apply, and how the components interact. By the end, you'll be able to draw the complete pipeline and explain the data flow from raw document to final answer.

🏗️ The complete RAG architecture

High-level view:

┌─────────────┐
│  INDEXING   │  ← Offline preparation (once)
└──────┬──────┘
       │
       v
┌─────────────┐
│  RETRIEVAL  │  ← Query time (every request)
└──────┬──────┘
       │
       v
┌─────────────┐
│ GENERATION  │  ← Query time (every request)
└──────┬──────┘
       │
       v
┌─────────────┐
│ EVALUATION  │  ← Continuous (measure and improve)
└─────────────┘

Data flow:

  1. Indexing: Document → Chunks → Embeddings → Vector DB
  2. Retrieval: User query → Query embedding → Similarity search → Top-K docs
  3. Generation: Top-K docs + Query → Context injection → LLM → Response
  4. Evaluation: Response → Metrics (faithfulness, relevancy) → Feedback loop

📥 Component 1: Indexing

What does it do?

It prepares documents for semantic search: it splits them into chunks, creates embeddings, and stores them in a vector DB.

Indexing subcomponents:

1.1 Chunking (splitting documents)

Decision: how do you split long documents?

# Option A: Fixed-size chunking (naive)
chunks = [document[i:i+500] for i in range(0, len(document), 500)]

# Option B: Semantic chunking (preserves coherence)
chunks = semantic_chunker.split(document)  # Splits by topic

# Option C: Recursive chunking (LangChain)
chunks = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50
).split_text(document)

Trade-offs:

StrategyProsConsWhen to use
Fixed-sizeFast, simpleLoses semantic contextPrototypes, structured docs
SemanticPreserves coherenceSlow, variable chunksNarrative, articles
RecursiveBalanced, overlapMedium complexityProduction default

Advanced techniques (Module 2):

  • Embedding-based semantic chunking
  • Recursive with custom separators
  • Structure-aware chunking (HTML, Markdown, JSON)

1.2 Embedding model (vectorization)

Decision: which embedding model do you use?

# Option A: OpenAI (managed, high quality)
from openai import OpenAI
client = OpenAI()

embedding = client.embeddings.create(
    model="text-embedding-ada-002",  # 1536 dimensions
    input="Your text here"
).data[0].embedding

# Option B: Local (free, private)
from sentence_transformers import SentenceTransformer

model = SentenceTransformer('all-MiniLM-L6-v2')  # 384 dimensions
embedding = model.encode("Your text here")

# Option C: Cohere (multilingual, quality)
import cohere
co = cohere.Client("api_key")

embedding = co.embed(
    texts=["Your text here"],
    model="embed-multilingual-v3.0"  # 1024 dimensions
).embeddings[0]

Model comparison:

ModelDimensionsCostQualityLatencyMultilingual
OpenAI ada-0021536$0.0001/1K tokensHigh~50msYes
Cohere multilingual-v31024$0.0001/1K tokensVery high~60msExcellent
SentenceTransformers local384-768FreeMedium~20msLimited

The typical decision:

  • Prototype: Local (free, fast)
  • English production: OpenAI (quality/cost)
  • Multilingual production: Cohere (better support)

1.3 Vector storage (the database)

Decision: where do you store the embeddings?

# Option A: ChromaDB (local, free)
import chromadb

client = chromadb.Client()
collection = client.create_collection("docs")

collection.add(
    documents=["Chunk text"],
    embeddings=[[0.1, 0.2, ...]],  # 1536D vector
    ids=["chunk_001"]
)

# Option B: Pinecone (managed, production)
import pinecone

index = pinecone.Index("my-index")

index.upsert(
    vectors=[
        ("chunk_001", [0.1, 0.2, ...], {"text": "Chunk text"})
    ]
)

# Option C: Numpy (in-memory, simple)
import numpy as np

embeddings_matrix = np.array([
    [0.1, 0.2, ...],  # Chunk 1
    [0.3, 0.4, ...],  # Chunk 2
])

Vector DB comparison:

Vector DBSetupCostPerformanceScalabilityWhen to use
ChromaDBLocal, easyFree10K docs: fast100K+ docs: slowDev, prototypes
PineconeManaged cloud$70/mo+Always fastMillions of docsProduction
NumpyIn-memoryFree<1K docs: okDoesn't scaleTesting

The typical progression:

  1. Development: ChromaDB local (modules 1-6)
  2. Production: Pinecone managed (module 7)

The complete indexing pipeline:

# indexing_pipeline.py
from openai import OpenAI
import chromadb

def index_documents(documents: list[str]):
    """
    The complete indexing pipeline.
    
    Input: a list of raw documents
    Output: a vector DB populated with embeddings
    """
    
    # 1. Chunking (naive fixed-size)
    chunks = []
    for doc in documents:
        doc_chunks = [doc[i:i+500] for i in range(0, len(doc), 500)]
        chunks.extend(doc_chunks)
    
    print(f"✅ Created {len(chunks)} chunks from {len(documents)} documents")
    
    # 2. Embeddings (OpenAI)
    client = OpenAI()
    embeddings = []
    
    for chunk in chunks:
        response = client.embeddings.create(
            model="text-embedding-ada-002",
            input=chunk
        )
        embeddings.append(response.data[0].embedding)
    
    print(f"✅ Created {len(embeddings)} embeddings (1536D each)")
    
    # 3. Storage (ChromaDB)
    chroma_client = chromadb.Client()
    collection = chroma_client.create_collection("my_docs")
    
    collection.add(
        documents=chunks,
        embeddings=embeddings,
        ids=[f"chunk_{i}" for i in range(len(chunks))]
    )
    
    print(f"✅ Stored in ChromaDB")
    
    return collection

# Usage
documents = [
    "FastAPI is a modern web framework for Python...",
    "LangChain is a library for building applications with LLMs...",
    # ... more documents
]

collection = index_documents(documents)

Expected output:

✅ Created 12 chunks from 3 documents
✅ Created 12 embeddings (1536D each)
✅ Stored in ChromaDB

🔍 Component 2: Retrieval

What does it do?

It finds the chunks most relevant to the user's query using similarity search.

Retrieval subcomponents:

2.1 Query processing

Without optimization (baseline):

# A direct query, no processing
user_query = "What is FastAPI?"

# Create the query's embedding
query_embedding = client.embeddings.create(
    model="text-embedding-ada-002",
    input=user_query
).data[0].embedding

With optimization (Module 3):

# Query expansion (generate similar queries)
expanded_queries = [
    "What is FastAPI?",
    "FastAPI features",
    "FastAPI framework explained"
]

# Query rewriting (reformulate for clarity)
rewritten_query = llm.invoke(
    f"Reformulate this query for search: {user_query}"
)

Advanced techniques (Module 3):

  • Query expansion (boost recall)
  • Query rewriting (clarity)
  • Query decomposition (multi-hop reasoning)
  • HyDE (Hypothetical Document Embeddings)

2.2 Similarity search (vector search)

Baseline: cosine similarity

# ChromaDB similarity search
results = collection.query(
    query_embeddings=[query_embedding],
    n_results=5  # Top-5 documents
)

# Result
{
    'ids': [['chunk_3', 'chunk_7', 'chunk_1', 'chunk_9', 'chunk_5']],
    'distances': [[0.15, 0.18, 0.21, 0.23, 0.25]],  # Cosine distance
    'documents': [['FastAPI is...', 'FastAPI has...', ...]]
}

Distance metrics:

MetricFormulaRangeWhen to use
Cosine1 - cos(θ)[0, 2]Default (normalized)
Euclidean||a - b||[0, ∞]Non-normalized embeddings
Dot producta · b[-∞, ∞]When magnitude matters

The typical decision: cosine similarity (the default in 95% of cases)

2.3 Re-ranking (improving precision)

Without re-ranking (baseline):

# Return the top-5 directly
top_k_docs = results['documents'][0][:5]

With re-ranking (Module 4):

# Re-rank with a cross-encoder
from sentence_transformers import CrossEncoder

model = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')

# Re-rank top-20 → top-5
scores = model.predict([
    (user_query, doc) for doc in results['documents'][0][:20]
])

# Sort by the cross-encoder's score
reranked_indices = np.argsort(scores)[::-1][:5]
top_k_docs = [results['documents'][0][i] for i in reranked_indices]

Typical improvement from re-ranking:

  • Precision@5: 65% → 85% (+20%)
  • Latency: +150-200ms
  • Cost: +complexity

Trade-off: it's worth it in production if precision matters more than latency.

The complete retrieval pipeline:

# retrieval_pipeline.py

def retrieve_relevant_docs(
    user_query: str,
    collection,
    top_k: int = 5
) -> list[str]:
    """
    The complete retrieval pipeline.
    
    Input: the user's query
    Output: the top-K most relevant documents
    """
    
    # 1. Query embedding
    client = OpenAI()
    query_embedding = client.embeddings.create(
        model="text-embedding-ada-002",
        input=user_query
    ).data[0].embedding
    
    print(f"✅ Query embedding created (1536D)")
    
    # 2. Similarity search
    results = collection.query(
        query_embeddings=[query_embedding],
        n_results=top_k
    )
    
    docs = results['documents'][0]
    distances = results['distances'][0]
    
    print(f"✅ Found {len(docs)} relevant documents")
    print(f"   Distances: {[f'{d:.3f}' for d in distances]}")
    
    return docs

# Usage
docs = retrieve_relevant_docs(
    user_query="What is FastAPI?",
    collection=collection,
    top_k=5
)

for i, doc in enumerate(docs, 1):
    print(f"{i}. {doc[:100]}...")

Expected output:

✅ Query embedding created (1536D)
✅ Found 5 relevant documents
   Distances: ['0.150', '0.180', '0.210', '0.230', '0.250']

1. FastAPI is a modern, fast web framework for Python...
2. FastAPI has automatic data validation with Pydantic...
3. FastAPI generates automatic documentation with Swagger UI...
4. FastAPI is async-first, with native async/await support...
5. FastAPI is used by Microsoft, Netflix, and Uber...

🤖 Component 3: Generation

What does it do?

It generates an answer using an LLM with the retrieved documents as context.

Generation subcomponents:

3.1 Context injection

The basic pattern:

# Build the context from the retrieved documents
context = "\n\n".join([
    f"Document {i+1}: {doc}"
    for i, doc in enumerate(docs)
])

# Prompt template
prompt_template = f"""
Use the following context to answer the question.

Context:
{context}

Question: {user_query}

Answer:
"""

The advanced pattern (with metadata):

# Include metadata in the context
context_with_metadata = "\n\n".join([
    f"[Source: {doc['source']}, Date: {doc['date']}]\n{doc['text']}"
    for doc in docs_with_metadata
])

3.2 LLM prompting (generation)

Baseline prompt:

from openai import OpenAI

client = OpenAI()

response = client.chat.completions.create(
    model="gpt-3.5-turbo",
    messages=[
        {"role": "system", "content": "You are a helpful assistant that answers questions based on the provided context."},
        {"role": "user", "content": prompt_template}
    ],
    temperature=0.0  # Deterministic
)

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

Advanced prompt (with instructions):

system_prompt = """
You are an expert technical assistant.

Instructions:
1. Answer ONLY based on the provided context
2. If the context doesn't contain the information, say "I don't have enough information"
3. Cite the documents you used (Document 1, Document 2, etc.)
4. Be concise but complete
5. If there's contradictory information, mention both versions
"""

response = client.chat.completions.create(
    model="gpt-4-turbo",  # Better quality
    messages=[
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": prompt_template}
    ],
    temperature=0.0
)

3.3 Response formatting

Response structure:

# The complete response structure
response_obj = {
    "answer": answer,
    "sources": [doc['id'] for doc in docs],
    "confidence": calculate_confidence(answer, docs),
    "model_used": "gpt-3.5-turbo",
    "tokens_used": response.usage.total_tokens,
    "latency_ms": latency
}

The complete generation pipeline:

# generation_pipeline.py

def generate_answer(
    user_query: str,
    retrieved_docs: list[str]
) -> dict:
    """
    The complete generation pipeline.
    
    Input: the query + the retrieved documents
    Output: the answer generated by the LLM
    """
    
    # 1. Context injection
    context = "\n\n".join([
        f"Document {i+1}: {doc}"
        for i, doc in enumerate(retrieved_docs)
    ])
    
    prompt = f"""
Use the following context to answer the question.

Context:
{context}

Question: {user_query}

Answer:
"""
    
    # 2. LLM generation
    client = OpenAI()
    response = client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[
            {"role": "system", "content": "You are a helpful assistant. Answer based only on the provided context."},
            {"role": "user", "content": prompt}
        ],
        temperature=0.0
    )
    
    answer = response.choices[0].message.content
    tokens = response.usage.total_tokens
    
    print(f"✅ Answer generated ({tokens} tokens)")
    
    # 3. Response formatting
    return {
        "answer": answer,
        "sources": [f"Doc {i+1}" for i in range(len(retrieved_docs))],
        "model": "gpt-3.5-turbo",
        "tokens": tokens
    }

# Usage
result = generate_answer(
    user_query="What is FastAPI?",
    retrieved_docs=docs
)

print(f"Answer: {result['answer']}")
print(f"Sources: {result['sources']}")
print(f"Tokens: {result['tokens']}")

Expected output:

✅ Answer generated (234 tokens)

Answer: FastAPI is a modern, fast web framework for Python, designed for building high-performance APIs. Its main features include automatic data validation with Pydantic, automatic documentation generation with Swagger UI, and native async/await support. It's used in production by companies like Microsoft, Netflix and Uber.

Sources: ['Doc 1', 'Doc 2', 'Doc 3']
Tokens: 234

📊 Component 4: Evaluation

What does it do?

It measures the RAG system's quality with objective metrics so you can spot what to improve.

Evaluation metrics:

4.1 Retrieval metrics (search quality)

# Precision: how many of the retrieved docs are relevant?
precision = relevant_retrieved / total_retrieved

# Recall: how many of the relevant docs did we retrieve?
recall = relevant_retrieved / total_relevant

# Example
# Total relevant docs in the DB: 10
# Retrieved docs: 5
# Relevant among the retrieved: 4

precision = 4 / 5  # 0.80 (80% of what we retrieved is relevant)
recall = 4 / 10    # 0.40 (we only found 40% of the relevant docs)

4.2 Generation metrics (answer quality)

Faithfulness (groundedness):

# Is the answer grounded in the context?
# Score: 0.0 (made up) to 1.0 (fully grounded)

# Example
context = "FastAPI is a web framework."
answer_grounded = "FastAPI is a web framework."  # Faithfulness: 1.0
answer_hallucinated = "FastAPI was created in 2015."  # Faithfulness: 0.0

Answer relevancy:

# Does the answer actually answer the question?
# Score: 0.0 (irrelevant) to 1.0 (perfectly relevant)

# Example
question = "What is FastAPI?"
answer_relevant = "FastAPI is a web framework."  # Relevancy: 1.0
answer_irrelevant = "Python is a language."  # Relevancy: 0.3

Evaluation with RAGAS (Module 8):

# Automated evaluation with RAGAS
from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy

# Evaluation dataset
dataset = {
    "question": ["What is FastAPI?"],
    "answer": ["FastAPI is a web framework..."],
    "contexts": [["FastAPI is a modern web framework..."]],
    "ground_truth": ["FastAPI is a Python web framework"]
}

# Evaluate
results = evaluate(
    dataset,
    metrics=[faithfulness, answer_relevancy]
)

print(results)
# Output:
# {
#   'faithfulness': 0.95,
#   'answer_relevancy': 0.92
# }

Interpreting the scores:

ScoreInterpretationAction
0.90+ExcellentProduction-ready
0.80-0.89GoodMinor optimizations
0.70-0.79AcceptableReview chunking/prompts
<0.70Needs workRedesign the pipeline

🔄 The feedback loop (continuous improvement)

┌──────────────┐
│   Query      │
└──────┬───────┘
       │
       v
┌──────────────┐
│   Retrieval  │
└──────┬───────┘
       │
       v
┌──────────────┐
│  Generation  │
└──────┬───────┘
       │
       v
┌──────────────┐
│  Evaluation  │ ← Measure faithfulness, relevancy
└──────┬───────┘
       │
       │ (If score <0.80)
       v
┌──────────────┐
│ Improvements │ ← Tune chunking, prompts, re-ranking
└──────┬───────┘
       │
       └───────> Loop back

🎯 Summary

Key concepts:

  • ✅ RAG has 4 components: Indexing, Retrieval, Generation, Evaluation
  • ✅ Indexing: Chunking → Embeddings → Vector DB (offline, once)
  • ✅ Retrieval: Query embedding → Similarity search → Top-K docs (query time)
  • ✅ Generation: Context injection → LLM prompting → Response (query time)
  • ✅ Evaluation: Retrieval metrics + Generation metrics → Feedback loop (continuous)
  • ✅ Advanced techniques apply to specific components (chunking in Indexing, re-ranking in Retrieval)
  • ✅ The complete pipeline: document → answer in 4 clear steps

What's next:

Capsule 03 teaches you the architecture decisions: which chunking strategy to use, which embeddings to choose, which vector DB to select, and when to apply re-ranking. Technical decisions grounded in real trade-offs.

📚 Additional resources

  1. LangChain Components Docs - Official component documentation
  2. ChromaDB Architecture - Vector DB architecture
  3. OpenAI Embeddings Best Practices - The official guide
  4. RAGAS Evaluation Framework - Evaluation metrics for RAG
  5. Building Production RAG (Video) - Architecture in production
  6. RAG Components Deep Dive - A technical article from Pinecone

Created: February 6, 2026
Version: 1.0

← PreviousModule Introduction: The Complete RAG PipelineNext →RAG Architecture Decisions