Query → Retriever → Relevant Passages → [Query + Passages] → Generator → Response

Detailed Component Breakdown

1. Document Processing Pipeline

Before retrieval, documents must be prepared:

Ingestion:

• Loading documents from various sources (PDFs, web pages, databases)
• Format conversion to processable text
• Metadata extraction and preservation

Chunking:

Documents are divided into manageable segments:

*Fixed-size chunking:*

def fixed_chunk(text, chunk_size=500, overlap=50):

chunks = []

for i in range(0, len(text), chunk_size - overlap):

chunks.append(text[i:i + chunk_size])

return chunks

*Semantic chunking:*

• Split by paragraph, section, or semantic boundary
• Preserve logical units of meaning
• Maintain contextual coherence

*Recursive chunking:*

• Try natural boundaries first (paragraphs)
• Fall back to sentences, then words
• Preserve hierarchy where possible

Optimal Chunk Size:

• Too small: loses context, fragments meaning
• Too large: dilutes relevant information, wastes context window
• Typical range: 256-1024 tokens
• Depends on content type and use case

Metadata:

Attach useful information to chunks:

• Source document
• Section headers
• Page numbers
• Timestamps
• Custom tags

2. Embedding and Indexing

Chunks are converted to vector representations and indexed for efficient retrieval.

Embedding Models:

Popular choices include:

• OpenAI text-embedding-ada-002 (1536 dimensions)
• OpenAI text-embedding-3-small/large (variable dimensions)
• Cohere embed-v3
• Sentence-transformers models (all-MiniLM-L6-v2, etc.)
• Domain-specific models (legal, medical, code)

Embedding Considerations:

• Dimension vs. accuracy tradeoff
• Inference speed requirements
• Domain specificity needs
• Cost per embedding

Vector Stores:

Vector databases enable efficient similarity search:

*In-memory:*

• Faiss (Facebook AI Similarity Search)
• Annoy (Spotify)
• HNSWlib

*Managed Services:*

• Pinecone
• Weaviate Cloud
• Qdrant Cloud
• Zilliz

*Self-Hosted:*

• Milvus
• Weaviate
• Qdrant
• Chroma
• pgvector (PostgreSQL extension)

Indexing Algorithms:

Different algorithms trade accuracy for speed:

*Flat Index (Exact):*

• Computes exact distances to all vectors
• Perfect accuracy, O(n) search time
• Only practical for small datasets

*IVF (Inverted File Index):*

• Clusters vectors, searches only nearest clusters
• Configurable accuracy-speed tradeoff
• Good for medium-sized datasets

*HNSW (Hierarchical Navigable Small World):*

• Graph-based approximate nearest neighbor
• Excellent accuracy-speed balance
• Higher memory usage
• Industry standard for production

3. Retrieval Strategies

Multiple approaches exist for finding relevant context:

Dense Retrieval:

Uses learned embeddings to compute semantic similarity:

query_embedding = embed(query)

results = vector_store.similarity_search(query_embedding, k=5)

Sparse Retrieval:

Traditional keyword matching using BM25 or TF-IDF:

bm25_results = bm25_index.search(query, k=5)

Hybrid Retrieval:

Combines dense and sparse methods:

dense_results = dense_retriever.search(query, k=10)

sparse_results = sparse_retriever.search(query, k=10)

final_results = reciprocal_rank_fusion(dense_results, sparse_results)

Multi-Query Retrieval:

Generate multiple query variations:

variations = llm.generate_query_variations(query, n=3)

all_results = [retriever.search(v) for v in variations]

merged_results = merge_and_deduplicate(all_results)

Contextual Retrieval:

Include conversation history for follow-up questions:

contextualized_query = llm.rewrite_with_context(query, conversation_history)

results = retriever.search(contextualized_query)

4. Re-Ranking

Initial retrieval is fast but imprecise. Re-ranking improves quality:

Cross-Encoder Re-Ranking:

from sentence_transformers import CrossEncoder

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

pairs = [(query, doc.text) for doc in initial_results]

scores = reranker.predict(pairs)

reranked = sorted(zip(initial_results, scores), key=lambda x: x[1], reverse=True)

LLM-Based Re-Ranking:

Use a language model to score relevance:

for doc in initial_results:

score = llm.score_relevance(query, doc.text)

doc.relevance_score = score

reranked = sorted(initial_results, key=lambda x: x.relevance_score, reverse=True)

Cohere Rerank:

import cohere

co = cohere.Client(api_key)

reranked = co.rerank(query=query, documents=docs, top_n=5)

5. Context Construction

Prepare retrieved information for the generator:

Basic Context Assembly:

context = "\n\n".join([f"[{i+1}] {doc.text}" for i, doc in enumerate(docs)])

prompt = f"""Based on the following context, answer the question.

Context:

{context}

Question: {query}

Answer:"""

Structured Context:

context_parts = []

for doc in docs:

context_parts.append(f"""

Source: {doc.metadata['source']}

Section: {doc.metadata['section']}

Content: {doc.text}

---""")

context = "\n".join(context_parts)

Context Selection:

If retrieved content exceeds context window:

• Take top-k by relevance
• Truncate to fit
• Use long-context models
• Implement hierarchical summarization

6. Generation

The LLM produces the final response:

Basic Generation:

response = llm.generate(prompt, max_tokens=1000)

With Citation:

prompt = f"""Answer based on the context. Cite sources using [1], [2], etc.

If the context doesn't contain the answer, say so.

Context:

{context}

Question: {query}"""

response = llm.generate(prompt)

Streaming:

for chunk in llm.stream(prompt):

yield chunk

Advanced RAG Patterns

Query Transformation

Improve retrieval by modifying the query:

HyDE (Hypothetical Document Embeddings):

Generate a hypothetical answer, then use it for retrieval:

hypothetical_answer = llm.generate(f"Answer this question: {query}")

hyde_embedding = embed(hypothetical_answer)

results = vector_store.similarity_search(hyde_embedding)

Query Decomposition:

Break complex queries into sub-queries:

sub_queries = llm.decompose_query(query)

# ["What is the capital of France?", "What is its population?"]

all_results = [retrieve(sq) for sq in sub_queries]

Step-Back Prompting:

Generate more general queries:

general_query = llm.generate(f"What's a more general form of: {query}")

general_context = retrieve(general_query)

specific_context = retrieve(query)

Hierarchical Retrieval

Use multiple stages with different granularity:

Summary Indexing:

# Create document summaries

for doc in documents:

summary = llm.summarize(doc.text)

summary_store.add(summary, metadata={'full_doc_id': doc.id})

# Retrieve summaries first, then full docs

relevant_summaries = summary_store.search(query, k=5)

full_docs = [get_full_doc(s.metadata['full_doc_id']) for s in relevant_summaries]

Parent-Child Retrieval:

Retrieve at chunk level but return larger parent chunks:

# Index small chunks

for chunk in small_chunks:

vector_store.add(chunk, metadata={'parent_id': chunk.parent_id})

# Search finds small chunks

results = vector_store.search(query)

# Return larger parent chunks

parents = set(r.metadata['parent_id'] for r in results)

context = [parent_store.get(p) for p in parents]

Self-Correction and Reflection

Improve answers through iterative refinement:

Self-RAG:

# Generate initial response

response = generate_with_context(query, retrieved_docs)

# Check if retrieval was needed

if llm.check_retrieval_needed(query, response):

# Re-retrieve with refined query

new_docs = retrieve(refine_query(query, response))

response = generate_with_context(query, new_docs)

# Critique and refine

critique = llm.critique(query, response, retrieved_docs)

if needs_improvement(critique):

response = llm.improve(response, critique)

CRAG (Corrective RAG):

# Retrieve and evaluate

docs = retrieve(query)

evaluations = [llm.evaluate_relevance(query, doc) for doc in docs]

if all(e == 'irrelevant' for e in evaluations):

# Fall back to web search

docs = web_search(query)

elif any(e == 'ambiguous' for e in evaluations):

# Knowledge refinement

docs = refine_and_rerank(docs, query)

response = generate(query, docs)

Agentic RAG

Combine RAG with agent frameworks:

Tool-Enhanced RAG:

tools = [

retrieval_tool,

web_search_tool,

calculator_tool,

database_tool

]

agent = create_agent(llm, tools)

response = agent.run(query) # Agent decides when to retrieve

Multi-Step Reasoning:

# ReAct-style reasoning

while not complete:

thought = llm.think(query, context, history)

action = llm.decide_action(thought)

if action == "retrieve":

new_context = retrieve(action.query)

context.extend(new_context)

elif action == "answer":

response = llm.generate_answer(query, context)

complete = True

Evaluation and Metrics

Retrieval Evaluation

Precision@k:

Fraction of retrieved documents that are relevant.

Recall@k:

Fraction of relevant documents that were retrieved.

NDCG (Normalized Discounted Cumulative Gain):

Measures ranking quality, weighing higher-ranked relevant results more.

MRR (Mean Reciprocal Rank):

Measures where the first relevant result appears.

Hit Rate:

Percentage of queries where at least one relevant document was retrieved.

Generation Evaluation

Faithfulness:

Does the response accurately reflect the retrieved context?

Relevance:

Does the response answer the question asked?

Coherence:

Is the response well-structured and readable?

Answer Correctness:

Is the final answer factually correct?

Evaluation Frameworks

RAGAS (Retrieval Augmented Generation Assessment):

from ragas import evaluate

from ragas.metrics import faithfulness, answer_relevancy, context_precision

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

TruLens:

from trulens_eval import TruChain, Feedback

feedback = Feedback(provider.groundedness).on_input_output()

tru_chain = TruChain(rag_chain, feedbacks=[feedback])

LLM-as-Judge:

evaluation_prompt = f"""

Rate the following response on a scale of 1-5 for:

• Relevance to the question
• Factual accuracy based on context
• Completeness

Question: {question}

Context: {context}

Response: {response}

"""

scores = llm.evaluate(evaluation_prompt)

Production Considerations

Scalability

Horizontal Scaling:

• Replicate embedding services
• Distribute vector store across nodes
• Load balance retrieval requests
• Cache frequent queries

Batch Processing:

# Batch embeddings

embeddings = embed_batch(chunks, batch_size=100)

# Batch retrieval

results = vector_store.search_batch(queries, k=5)

Latency Optimization

Embedding Caching:

@cache(ttl=3600)

def get_embedding(text):

return model.embed(text)

Pre-computation:

• Pre-embed common queries
• Cache frequent search results
• Pre-generate common responses

Parallel Retrieval:

import asyncio

async def parallel_retrieve(queries):

tasks = [retrieve_async(q) for q in queries]

return await asyncio.gather(*tasks)

Cost Management

Embedding Costs:

• Choose appropriately sized models
• Batch embedding requests
• Cache embeddings aggressively
• Consider open-source alternatives

LLM Costs:

• Right-size context (don't retrieve more than needed)
• Use smaller models for simple tasks
• Implement semantic caching
• Monitor and set usage limits

Monitoring and Observability

Key Metrics:

• Retrieval latency
• Generation latency
• Relevance scores
• User feedback
• Error rates
• Cache hit rates

Logging:

import logging

def rag_query(query):

start = time.time()

docs = retrieve(query)

retrieval_time = time.time() - start

start = time.time()

response = generate(query, docs)

generation_time = time.time() - start

logging.info(f"Query: {query[:50]}..., Retrieval: {retrieval_time:.2f}s, Generation: {generation_time:.2f}s")

return response