Building Production-Ready RAG Systems: A Complete Guide

from typing import TypedDict


class Chunk(TypedDict):
    content: str
    metadata: dict


def chunk_document(
    content: str,
    chunk_size: int = 1000,
    chunk_overlap: int = 200,
) -> list[Chunk]:
    """Split document into overlapping chunks."""
    chunks: list[Chunk] = []
    start = 0

    while start < len(content):
        end = start + chunk_size
        chunk_text = content[start:end]

        chunks.append({
            "content": chunk_text,
            "metadata": {
                "chunk_index": len(chunks),
                "start_char": start,
                "end_char": end,
            },
        })

        start += chunk_size - chunk_overlap

    # Add total chunks to metadata
    for chunk in chunks:
        chunk["metadata"]["total_chunks"] = len(chunks)

    return chunks

from fastembed import (
    TextEmbedding,
    SparseTextEmbedding,
    LateInteractionTextEmbedding,
)

# Initialize embedding models
dense_model = TextEmbedding("sentence-transformers/all-MiniLM-L6-v2")
sparse_model = SparseTextEmbedding("Qdrant/bm25")
late_interaction_model = LateInteractionTextEmbedding("colbert-ir/colbertv2.0")


def generate_embeddings(documents: list[str]) -> dict:
    """Generate all three types of embeddings for documents."""
    return {
        "dense": list(dense_model.embed(documents)),
        "sparse": list(sparse_model.embed(documents)),
        "late_interaction": list(late_interaction_model.embed(documents)),
    }


def generate_query_embeddings(query: str) -> dict:
    """Generate embeddings for a query."""
    return {
        "dense": next(dense_model.query_embed(query)),
        "sparse": next(sparse_model.query_embed(query)),
        "late_interaction": next(late_interaction_model.query_embed(query)),
    }

from qdrant_client import QdrantClient
from qdrant_client.models import (
    Distance,
    VectorParams,
    SparseVectorParams,
    MultiVectorConfig,
    MultiVectorComparator,
    HnswConfigDiff,
    Modifier,
)

client = QdrantClient(url="http://localhost:6333")

# Create collection with multi-vector support
client.create_collection(
    collection_name="rag-documents",
    vectors_config={
        "dense": VectorParams(
            size=384,  # all-MiniLM-L6-v2 dimension
            distance=Distance.COSINE,
        ),
        "late_interaction": VectorParams(
            size=128,  # ColBERT v2 dimension
            distance=Distance.COSINE,
            multivector_config=MultiVectorConfig(
                comparator=MultiVectorComparator.MAX_SIM,
            ),
            hnsw_config=HnswConfigDiff(m=0),  # Disable HNSW for reranking
        ),
    },
    sparse_vectors_config={
        "sparse": SparseVectorParams(
            modifier=Modifier.IDF,
        ),
    },
)

from qdrant_client.models import PointStruct

def upsert_documents(
    documents: list[str],
    embeddings: dict,
) -> None:
    """Insert documents with their embeddings into Qdrant."""
    points = []

    for idx, doc in enumerate(documents):
        point = PointStruct(
            id=idx,
            vector={
                "dense": embeddings["dense"][idx],
                "sparse": embeddings["sparse"][idx].as_object(),
                "late_interaction": embeddings["late_interaction"][idx],
            },
            payload={"document": doc},
        )
        points.append(point)

    client.upsert(
        collection_name="rag-documents",
        points=points,
    )

from qdrant_client.models import Prefetch, SparseVector


def hybrid_search(
    query: str,
    limit: int = 10,
    prefetch_limit: int = 20,
) -> list[dict]:
    """Perform hybrid search with reranking."""
    # Generate query embeddings
    query_embeddings = generate_query_embeddings(query)

    # Set up prefetch for hybrid search
    prefetch = [
        Prefetch(
            query=query_embeddings["dense"],
            using="dense",
            limit=prefetch_limit,
        ),
        Prefetch(
            query=SparseVector(**query_embeddings["sparse"].as_object()),
            using="sparse",
            limit=prefetch_limit,
        ),
    ]

    # Execute hybrid search with ColBERT reranking
    results = client.query_points(
        collection_name="rag-documents",
        prefetch=prefetch,
        query=query_embeddings["late_interaction"],
        using="late_interaction",
        with_payload=True,
        limit=limit,
    )

    return [
        {
            "id": point.id,
            "score": point.score,
            "document": point.payload["document"],
        }
        for point in results.points
    ]

from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()

openai_client = OpenAI()

SYSTEM_PROMPT = """You are a helpful assistant that answers questions based on the provided context.
If the context doesn't contain relevant information, say so honestly.

Context:
{context}"""


def generate_response(query: str, context: list[str]) -> str:
    """Generate a response using retrieved context."""
    formatted_context = "\n\n---\n\n".join(context)

    response = openai_client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": SYSTEM_PROMPT.format(context=formatted_context),
            },
            {"role": "user", "content": query},
        ],
        temperature=0.7,
    )

    return response.choices[0].message.content

def rag_query(query: str) -> str:
    """Complete RAG pipeline: search and generate."""
    # 1. Hybrid search with reranking
    results = hybrid_search(query, limit=5)

    # 2. Extract document content
    context = [r["document"] for r in results]

    # 3. Generate response
    response = generate_response(query, context)

    return response


# Example usage
if __name__ == "__main__":
    query = "How does RAG improve LLM accuracy?"
    answer = rag_query(query)
    print(answer)

from qdrant_client.models import Filter, FieldCondition, MatchValue

def filtered_search(
    query: str,
    category: str,
    limit: int = 10,
) -> list[dict]:
    """Search with metadata filtering."""
    query_embeddings = generate_query_embeddings(query)

    results = client.query_points(
        collection_name="rag-documents",
        query=query_embeddings["dense"],
        using="dense",
        query_filter=Filter(
            must=[
                FieldCondition(
                    key="category",
                    match=MatchValue(value=category),
                ),
            ],
        ),
        with_payload=True,
        limit=limit,
    )

    return results.points

import logging
from dataclasses import dataclass

logger = logging.getLogger(__name__)


@dataclass
class RAGError(Exception):
    message: str
    code: str
    retriable: bool = False


def safe_generate(query: str, context: list[str]) -> str:
    """Generate with error handling and retry."""
    try:
        return generate_response(query, context)
    except RAGError as e:
        if e.retriable:
            logger.warning("Retrying failed generation", extra={"error": str(e)})
            return generate_response(query, context)

        logger.error("Generation failed", extra={"error": str(e), "query": query})
        raise RAGError(
            message="Failed to generate response",
            code="GENERATION_FAILED",
            retriable=False,
        )