How to Build a RAG System in Python: From Chunking to Hybrid Search in Production

Introduction

Imagine: you're deploying a corporate chatbot based on an LLM, but it gives outdated data or hallucinates. The classic approach—fine-tuning the model—is expensive and inflexible. The solution is RAG (Retrieval-Augmented Generation). However, building a RAG that works in production, not in a Jupyter Notebook, is a non-trivial task. In this article, we'll break down a real case: how we replaced a static knowledge base with a production-ready RAG pipeline in Python for a fintech startup. You'll learn how to choose a chunking strategy, set up hybrid search (dense + sparse), and add reranking for accuracy.

The Problem: Static Knowledge Base vs Dynamic Queries

The startup FinFlow managed a knowledge base of 10,000 documents (regulations, FAQs, API documentation). Support answered queries manually, with an average response time of 4 hours. An attempt to implement an LLM without RAG failed: the GPT-4 model produced 30% hallucinations on specific questions. A system was needed that:
- Handles 500+ queries per day with >90% accuracy.
- Works with both Russian and English texts.
- Scales to 100,000 documents.

The Solution: Step-by-Step RAG Pipeline

We broke the project into 4 stages: chunking, embeddings, hybrid search, reranking. Each stage comes with benchmarks and code.

1. Chunking Strategies: Not All Pieces Are Equally Useful

The first stage is to split documents into fragments (chunks). A mistake here kills the entire pipeline. We tested 3 strategies on a corpus of 500 PDFs:

Strategy Chunk Size Overlap Recall@10
Fixed 512 tokens 128 tokens 0.72
Semantic splitting (NLTK) Variable Sentences 0.81
RecursiveCharacterTextSplitter (LangChain) 1024 tokens 256 tokens 0.85

Conclusion: RecursiveCharacterTextSplitter with a size of 1024 and overlap of 256 gave the best recall (0.85). Important: for Russian, we used a tokenizer from BERT (wordpiece) instead of spaces—this reduced meaning loss by 15%.

Example code (Python):

from langchain.text_splitter import RecursiveCharacterTextSplitter

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1024,
    chunk_overlap=256,
    separators=["\n\n", "\n", ".", " ", ""]
)
chunks = text_splitter.split_text(document)

2. Embedding Generation: Dense and Sparse

For indexing, we used two approaches:
- Dense (dense embeddings)—model intfloat/multilingual-e5-large (support for 100+ languages). Dimensionality—1024, inference speed—50 ms per chunk.
- Sparse (sparse embeddings)—BM25 via Elasticsearch. This compensates for the weakness of dense models on rare terms (e.g., "TIN 7707083893").

Benchmark on a test set (1000 questions):

Method MAP@10 Search Time (ms)
Dense only 0.78 45
BM25 only 0.65 12
Hybrid (dense + sparse) 0.89 60

The hybrid approach improved accuracy by 14% compared to dense-only.

3. Hybrid Search: How to Combine Dense and Sparse

We implemented hybrid search via weighted sum. Weights were chosen empirically: α=0.7 for dense, β=0.3 for BM25. Python code using FAISS and Elasticsearch:

from sentence_transformers import SentenceTransformer
from elasticsearch import Elasticsearch
from sklearn.preprocessing import normalize
import numpy as np

# Dense
model = SentenceTransformer('intfloat/multilingual-e5-large')
dense_embed = model.encode(query)

# Sparse (BM25)
es = Elasticsearch()
sparse_scores = es.search(index="docs", query={"match": {"text": query}})

# Fusion
scores = 0.7 * dense_similarities + 0.3 * sparse_scores
top_k = np.argsort(scores)[-10:][::-1]

Important: for production, we added query caching (Redis) so repeated questions are processed in 5 ms instead of 150 ms.

4. Reranking: The Final Filter

Even hybrid search returned 2-3 irrelevant chunks in the top-10. We added a reranker—model cross-encoder/ms-marco-MiniLM-L-6-v2. It re-ranks results in 100 ms per query, boosting NDCG@10 from 0.85 to 0.93.

Example integration:

from sentence_transformers import CrossEncoder

reranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')
pairs = [(query, chunk) for chunk in top_chunks]
scores = reranker.predict(pairs)
reranked = [chunk for _, chunk in sorted(zip(scores, top_chunks), reverse=True)]

Results

After implementing the production RAG pipeline:
- Answer accuracy (accuracy@1) increased from 70% to 93%.
- Response time—200 ms (with cache—50 ms).
- The system handles 1000 queries/day with 99.9% uptime.
- Hallucinations reduced to 3% (verified via LLM-as-a-judge).

Why This Works in Production

Key success factors:
- Chunking with overlap—preserves context between fragments.
- Hybrid search—dense for semantics, BM25 for exact matches.
- Reranker—filters out noise.
- Monitoring—we added metric logging (recall, latency) via Prometheus + Grafana.

On the ASI Biont platform, there is a full course where we dive deeper into each stage: from choosing a vector DB (Qdrant vs Milvus) to deployment with Docker and Kubernetes. You'll learn to build RAG that handles enterprise load.

Conclusion

Building a RAG system in Python is feasible if you approach it systematically. Start with chunking (RecursiveCharacterTextSplitter), add hybrid search (dense + BM25), and a reranker. Don't forget caching and monitoring. If you want to master all the nuances—from Graph RAG to quality evaluation—check out the specialized course at asibiont.com.

Ready to take your knowledge base to the next level? Start small: split one document and test hybrid search. And when you need production—you'll already know what to do.

← All posts

Comments