Implementing RAG (Retrieval Augmented Generation): Best Practices and Patterns

Retrieval Augmented Generation (RAG) has emerged as a powerful approach for enhancing Large Language Models (LLMs) with external knowledge. This guide will walk you through implementing production-ready RAG systems, from basic concepts to advanced optimization techniques.

$1

$1

1. Document Processing

- Text extraction

- Chunking strategies

- Metadata management

- Document storage

2. Embedding Generation

- Model selection

- Embedding optimization

- Batch processing

- Quality assurance

3. Vector Storage

- Database selection

- Index optimization

- Query performance

- Scaling strategies

4. LLM Integration

- Context injection

- Prompt engineering

- Response generation

- Output validation

$1

$1

``python

from typing import List, Dict

import tiktoken

from langchain.text_splitter import RecursiveCharacterTextSplitter

class DocumentProcessor:

def __init__(self, chunk_size: int = 1000, chunk_overlap: int = 200):

self.text_splitter = RecursiveCharacterTextSplitter(

chunk_size=chunk_size,

chunk_overlap=chunk_overlap,

length_function=len,

separators=["\n\n", "\n", " ", ""]

)

def process_document(self, content: str, metadata: Dict) -> List[Dict]:

chunks = self.text_splitter.split_text(content)

return [

{

"text": chunk,

"metadata": {

metadata,

"chunk_index": i

}

}

for i, chunk in enumerate(chunks)

]

`

$1

`python

from sentence_transformers import SentenceTransformer

import numpy as np

class EmbeddingPipeline:

def __init__(self, model_name: str = "all-MiniLM-L6-v2"):

self.model = SentenceTransformer(model_name)

def generate_embeddings(self, texts: List[str], batch_size: int = 32):

embeddings = []

for i in range(0, len(texts), batch_size):

batch = texts[i:i + batch_size]

batch_embeddings = self.model.encode(batch)

embeddings.extend(batch_embeddings)

return np.array(embeddings)

`

$1

$1

`python

import weaviate

from weaviate.util import generate_uuid5

class VectorStore:

def __init__(self, url: str):

self.client = weaviate.Client(url)

self.setup_schema()

def setup_schema(self):

class_obj = {

"class": "Document",

"vectorizer": "none",

"properties": [

{"name": "text", "dataType": ["text"]},

{"name": "metadata", "dataType": ["object"]}

]

}

self.client.schema.create_class(class_obj)

def add_documents(self, documents: List[Dict], embeddings: np.ndarray):

with self.client.batch as batch:

for doc, embedding in zip(documents, embeddings):

batch.add_data_object(

data_object={

"text": doc["text"],

"metadata": doc["metadata"]

},

class_name="Document",

vector=embedding

)

`

$1

$1

`python

class QueryEngine:

def __init__(self, vector_store: VectorStore, embedding_pipeline: EmbeddingPipeline):

self.vector_store = vector_store

self.embedding_pipeline = embedding_pipeline

def search(self, query: str, k: int = 5):

query_embedding = self.embedding_pipeline.generate_embeddings([query])[0]

result = (

self.vector_store.client.query

.get("Document", ["text", "metadata"])

.with_near_vector({

"vector": query_embedding,

"certainty": 0.7

})

.with_limit(k)

.do()

)

return result["data"]["Get"]["Document"]

`

$1

`python

from langchain.chat_models import ChatOpenAI

from langchain.prompts import ChatPromptTemplate

class RAGSystem:

def __init__(self, query_engine: QueryEngine):

self.query_engine = query_engine

self.llm = ChatOpenAI(temperature=0.7)

def generate_response(self, query: str):

# Retrieve relevant context

context = self.query_engine.search(query)

# Prepare prompt

prompt = ChatPromptTemplate.from_template("""

Answer the question based on the following context:

Context: {context}

Question: {question}

Answer:""")

# Generate response

response = self.llm(prompt.format(

context="\n\n".join([doc["text"] for doc in context]),

question=query

))

return response

`

$1

$1

`mermaid

graph TD

A[Raw Text] --> B[Preprocessing]

B --> C[Chunking]

C --> D[Embedding Generation]

D --> E[Dimensionality Reduction]

E --> F[Index Building]

`

$1

`python

class RAGMonitor:

def __init__(self):

self.metrics = {

"latency": [],

"relevance_scores": [],

"token_usage": []

}

def log_query(self, query_time: float, relevance: float, tokens: int):

self.metrics["latency"].append(query_time)

self.metrics["relevance_scores"].append(relevance)

self.metrics["token_usage"].append(tokens)

`

$1

`python

class RAGErrorHandler:

def handle_retrieval_error(self, error):

# Implement fallback strategy

pass

def handle_generation_error(self, error):

# Implement retry logic

pass

def handle_embedding_error(self, error):

# Implement backup model

pass

``

$1

Implementing a production-ready RAG system requires careful consideration of various components and their integration. By following the patterns and practices outlined in this guide, you can build robust and efficient RAG systems that provide accurate and relevant responses to user queries.