Introduction

Retrieval-Augmented Generation (RAG) has become the standard approach for building AI applications that need access to domain-specific knowledge. This article covers production-ready patterns and best practices.

Architecture Overview

A typical RAG system consists of three main components:

1. Document Ingestion Pipeline: Processes and stores documents
2. Retrieval System: Finds relevant context
3. Generation Layer: LLM generates responses using retrieved context

flowchart TD
  A[User Query] --> B[Query Embedding]
  B --> C[Vector Search]
  C --> D[Retrieve Top K Documents]
  D --> E[Context Assembly]
  E --> F[LLM Generation]
  F --> G[Response]
  
  H[Documents] --> I[Chunking]
  I --> J[Document Embedding]
  J --> K[Vector Database]
  K --> C
  
  style A fill:#3b82f6
  style G fill:#10b981
  style K fill:#8b5cf6

Vector Database Selection

Choosing the right vector database is crucial:

Popular Options

• Pinecone: Managed service, great for startups
• Weaviate: Open-source, self-hostable
• Qdrant: High performance, Rust-based
• Chroma: Simple, Python-native

Implementation Example

from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams

client = QdrantClient("localhost", port=6333)

# Create collection
client.create_collection(
    collection_name="documents",
    vectors_config=VectorParams(
        size=1536,  # OpenAI embedding dimension
        distance=Distance.COSINE
    )
)

Embedding Strategies

Chunking Documents

Effective chunking balances context preservation with retrieval precision:

def chunk_document(text: str, chunk_size: int = 1000, overlap: int = 200):
    """Split document into overlapping chunks."""
    chunks = []
    start = 0
    
    while start < len(text):
        end = start + chunk_size
        chunk = text[start:end]
        chunks.append({
            'text': chunk,
            'start': start,
            'end': end
        })
        start = end - overlap
    
    return chunks

Embedding Models

• OpenAI text-embedding-3-large: Best quality, paid
• sentence-transformers/all-MiniLM-L6-v2: Good balance, free
• BGE-large-en-v1.5: Open-source, competitive quality

Retrieval Strategies

Hybrid Search

Combine semantic and keyword search:

def hybrid_search(query: str, top_k: int = 5):
    # Semantic search
    semantic_results = vector_db.similarity_search(
        query_embedding=embed(query),
        limit=top_k
    )
    
    # Keyword search
    keyword_results = full_text_search(query, limit=top_k)
    
    # Combine and rerank
    return rerank_results(semantic_results, keyword_results)

Reranking

Use a cross-encoder model to improve retrieval quality:

from sentence_transformers import CrossEncoder

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

def rerank(query: str, documents: List[str]):
    pairs = [[query, doc] for doc in documents]
    scores = reranker.predict(pairs)
    return sorted(zip(documents, scores), key=lambda x: x[1], reverse=True)

Prompt Engineering

Context Injection Pattern

def build_prompt(query: str, context: List[str]) -> str:
    context_text = "\n\n".join([
        f"Document {i+1}:\n{doc}" 
        for i, doc in enumerate(context)
    ])
    
    return f"""You are a helpful assistant. Use the following context to answer the question.

Context:
{context_text}

Question: {query}

Answer:"""

Production Considerations

Caching

Cache embeddings and retrieval results:

from functools import lru_cache

@lru_cache(maxsize=1000)
def get_embedding(text: str):
    return embedding_model.encode(text)

Error Handling

Implement robust error handling:

async def retrieve_and_generate(query: str):
    try:
        # Retrieve context
        context = await retrieve_context(query)
        
        if not context:
            return "I couldn't find relevant information."
        
        # Generate response
        response = await llm.generate(query, context)
        return response
        
    except VectorDBError as e:
        logger.error(f"Vector DB error: {e}")
        return "Sorry, I'm experiencing technical difficulties."
    except LLMError as e:
        logger.error(f"LLM error: {e}")
        return "I'm having trouble generating a response."

Monitoring

Track key metrics:

• Retrieval latency
• Embedding generation time
• LLM response time
• User satisfaction scores

Advanced Patterns

Multi-Step Retrieval

For complex queries, use iterative retrieval:

1. Initial retrieval with broad query
2. Refine query based on initial results
3. Retrieve additional context
4. Generate final response

Query Expansion

Expand queries to improve retrieval:

def expand_query(query: str):
    # Generate related terms
    synonyms = get_synonyms(query)
    # Combine original and expanded terms
    return f"{query} {' '.join(synonyms)}"

Conclusion

Building production-ready RAG systems requires careful attention to retrieval quality, prompt engineering, and operational concerns. Start simple and iterate based on user feedback.

Key Takeaways

• Choose vector databases based on scale and requirements
• Implement hybrid search for better retrieval
• Use reranking to improve result quality
• Cache embeddings to reduce latency
• Monitor all components for reliability