Building a Production RAG System in Python

import os
from pydantic import BaseModel

class RAGConfig(BaseModel):
    QDRANT_HOST: str = os.getenv("QDRANT_HOST", "http://localhost:6333")
    COLLECTION_NAME: str = "enterprise_knowledge_base"
    EMBEDDING_MODEL: str = "sentence-transformers/all-MiniLM-L6-v2"
    VECTOR_SIZE: int = 384
    TOP_K: int = 5

config = RAGConfig()

from typing import List, Dict

class SemanticChunker:
    def __init__(self, chunk_size: int = 500, chunk_overlap: int = 50):
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap

    def split_text(self, document_id: str, text: str) -> List[Dict]:
        words = text.split()
        chunks = []
        start = 0
        chunk_idx = 0

        while start < len(words):
            end = min(start + self.chunk_size, len(words))
            chunk_text = " ".join(words[start:end])
            
            chunks.append({
                "chunk_id": f"{document_id}_chunk_{chunk_idx}",
                "doc_id": document_id,
                "text": chunk_text,
                "token_count": len(words[start:end])
            })
            chunk_idx += 1
            start += (self.chunk_size - self.chunk_overlap)
            
        return chunks

# Example execution
if __name__ == "__main__":
    chunker = SemanticChunker(chunk_size=100, chunk_overlap=20)
    sample_text = "Echo INC. provides production AI and cybersecurity architectures..." * 10
    results = chunker.split_text("doc_001", sample_text)
    print(f"Generated {len(results)} structured chunks.")

from qdrant_client import QdrantClient
from qdrant_client.models import VectorParams, Distance, PointStruct
from sentence_transformers import SentenceTransformer
from config import config

class VectorIndexer:
    def __init__(self):
        self.client = QdrantClient(url=config.QDRANT_HOST)
        self.model = SentenceTransformer(config.EMBEDDING_MODEL)
        self._create_collection()

    def _create_collection(self):
        collections = self.client.get_collections().collections
        exists = any(c.name == config.COLLECTION_NAME for c in collections)
        if not exists:
            self.client.create_collection(
                collection_name=config.COLLECTION_NAME,
                vectors_config=VectorParams(size=config.VECTOR_SIZE, distance=Distance.COSINE)
            )

    def index_chunks(self, chunks: list):
        texts = [c["text"] for c in chunks]
        embeddings = self.model.encode(texts, show_progress_bar=True)
        
        points = []
        for idx, (chunk, vector) in enumerate(zip(chunks, embeddings)):
            points.append(PointStruct(
                id=idx + 1,
                vector=vector.tolist(),
                payload=chunk
            ))
            
        self.client.upsert(collection_name=config.COLLECTION_NAME, points=points)
        print(f"Indexed {len(points)} vector records into Qdrant.")

from rank_bm25 import BM25Okapi
from indexer import VectorIndexer
from config import config

class HybridSearchEngine:
    def __init__(self, corpus: list):
        self.indexer = VectorIndexer()
        self.corpus = corpus
        self.tokenized_corpus = [doc["text"].lower().split() for doc in corpus]
        self.bm25 = BM25Okapi(self.tokenized_corpus)

    def search(self, query: str, top_k: int = 5) -> list:
        # 1. Sparse BM25 Search
        tokenized_query = query.lower().split()
        bm25_scores = self.bm25.get_scores(tokenized_query)
        bm25_top_indices = sorted(range(len(bm25_scores)), key=lambda i: bm25_scores[i], reverse=True)[:top_k*2]

        # 2. Dense Vector Search
        query_vector = self.indexer.model.encode(query).tolist()
        vector_hits = self.indexer.client.search(
            collection_name=config.COLLECTION_NAME,
            query_vector=query_vector,
            limit=top_k*2
        )

        # 3. Reciprocal Rank Fusion (RRF)
        rrf_scores = {}
        k_factor = 60

        for rank, idx in enumerate(bm25_top_indices):
            doc_text = self.corpus[idx]["text"]
            rrf_scores[doc_text] = rrf_scores.get(doc_text, 0) + (1.0 / (k_factor + rank + 1))

        for rank, hit in enumerate(vector_hits):
            doc_text = hit.payload["text"]
            rrf_scores[doc_text] = rrf_scores.get(doc_text, 0) + (1.0 / (k_factor + rank + 1))

        sorted_results = sorted(rrf_scores.items(), key=lambda x: x[1], reverse=True)
        return [doc for doc, score in sorted_results[:top_k]]

from openai import OpenAI
import os

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY", "mock_key"))

SYSTEM_PROMPT = """You are EchoAI Knowledge Assistant. Answer the question using ONLY the retrieved context below.
If the context does not contain the answer, reply with 'Information not found in enterprise knowledge base.'

RETRIEVED CONTEXT:
{context}"""

def generate_grounded_answer(query: str, retrieved_chunks: list) -> str:
    context_str = "\n\n".join([f"- {chunk}" for chunk in retrieved_chunks])
    full_prompt = SYSTEM_PROMPT.format(context=context_str)

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": full_prompt},
            {"role": "user", "content": query}
        ],
        temperature=0.1
    )
    return response.choices[0].message.content

import time
from chunker import SemanticChunker
from indexer import VectorIndexer
from hybrid_search import HybridSearchEngine

def run_pipeline_benchmark():
    print("=== STARTING END-TO-END PYTHON RAG BENCHMARK ===")
    t0 = time.time()

    raw_document = "EchoAgent Orchestrator features stateful tool execution and memory persistence."
    chunker = SemanticChunker(chunk_size=30, chunk_overlap=5)
    chunks = chunker.split_text("doc_catalog", raw_document)
    print(f"Step 1 Complete: Created {len(chunks)} chunks in {round(time.time()-t0, 3)}s")

    indexer = VectorIndexer()
    indexer.index_chunks(chunks)
    print("=== RAG BENCHMARK COMPLETED SUCCESSFULLY ===")

if __name__ == "__main__":
    run_pipeline_benchmark()