Ship RAG to production and watch it fail. Here's what works: semantic chunking, hybrid retrieval, reranking, and how to cut costs 5x.
I've watched teams implement RAG systems, and here's what I see: they build something that works locally, ship it to production, and then it falls apart. The chatbot that could answer questions perfectly on a 10-document test set starts giving garbage responses on real data. Cost spirals because they're embedding everything multiple times. Latency blows out because they're reranking with a model that runs in 5 seconds per query.
The problem isn't RAG itself. It's that most teams treat RAG as "throw documents in a vector DB and ask questions." That's not RAG at scale. That's a prototype.
Real production RAG is about making a hundred decisions: How do you chunk documents without losing context? Which embedding model gives you accuracy without eating your budget? When do you rerank and when do you skip it? How do you handle the fact that your top-10 retrieval results are garbage 40% of the time?
This post covers what I've learned from running RAG systems that actually work.
Your chunking strategy determines everything downstream. Get it wrong and no amount of reranking or retrieval magic fixes it. Most teams use fixed-size chunks (512 tokens, overlap 50 or 100). This is fine for learning. It's terrible for production.
Here's why: documents have structure. A legal contract has sections. A technical doc has code snippets and explanations. A research paper has abstract, methodology, results. Fixed-size chunking ignores all of that. You end up with chunks that split sentences mid-thought, chunks that miss context because they're too small, or chunks that duplicate content excessively.
1. Fixed-Size Chunking (Quick, Wrong)
python
from typing import List
def fixed_size_chunking(text: str, chunk_size: int = 512, overlap: int = 50) -> List[str]:
"""
Naive fixed-size chunking.
Fast but loses document structure.
"""
chunks = []
step = chunk_size - overlap
for i in range(0, len(text), step):
chunk = text[i : i + chunk_size]
if len(chunk) > 100: # Skip tiny chunks
chunks.append(chunk)
return chunks
text = "The capital of France is Paris. It's known for the Eiffel Tower..."
chunks = fixed_size_chunking(text)
2. Semantic Chunking (Better, What You Actually Need)
python
from sentence_transformers import SentenceTransformer
import numpy as np
from typing import List, Tuple
def semantic_chunking(
text: str,
model_name: str = "all-MiniLM-L6-v2",
similarity_threshold: float = 0.5,
min_chunk_size: int = 100,
) -> List[str]:
"""
Split text at semantic boundaries.
Algorithm:
1. Split text into sentences
2. Compute embeddings for each sentence
3. Calculate cosine similarity between adjacent sentences
4. Start new chunk when similarity drops below threshold
This preserves meaning and document structure.
"""
model = SentenceTransformer(model_name)
sentences = text.split(". ")
sentences = [s.strip() + "." for s in sentences if s.strip()]
if len(sentences) < 2:
return [text]
embeddings = model.encode(sentences)
chunks = []
current_chunk = [sentences[0]]
for i in range(1, len(sentences)):
similarity = np.dot(embeddings[i], embeddings[i-1]) / (
np.linalg.norm(embeddings[i]) * np.linalg.norm(embeddings[i-1])
)
if similarity < similarity_threshold:
chunk_text = " ".join(current_chunk)
if len(chunk_text) >= min_chunk_size:
chunks.append(chunk_text)
current_chunk = [sentences[i]]
else:
current_chunk.append(sentences[i])
if current_chunk:
chunk_text = " ".join(current_chunk)
if len(chunk_text) >= min_chunk_size:
chunks.append(chunk_text)
return chunks
text = """
The capital of France is Paris. It's known for the Eiffel Tower.
The Eiffel Tower was built in 1889. It stands 330 meters tall.
London is the capital of the UK. It has Big Ben and Westminster Abbey.
"""
chunks = semantic_chunking(text, similarity_threshold=0.4)
In production, I use a hybrid approach:
python
from typing import List, Dict, Any
import re
def production_chunking(
text: str,
source: str = "unknown",
max_chunk_size: int = 1000,
min_chunk_size: int = 100,
) -> List[Dict[str, Any]]:
"""
Production RAG chunking.
Strategy:
1. Preserve document structure (sections, subsections)
2. Chunk semantically within sections
3. Add metadata for filtering and ranking
4. Small overlap to catch cross-boundary information
"""
chunks = []
chunk_id = 0
sections = re.split(r'\n#{1,3} ', text)
for section_idx, section in enumerate(sections):
lines = section.split('\n')
current_chunk = []
current_size = 0
for line_idx, line in enumerate(lines):
line_tokens = len(line.split())
if current_size + line_tokens > max_chunk_size and current_chunk:
chunk_text = '\n'.join(current_chunk)
if len(chunk_text) > min_chunk_size:
chunks.append({
"content": chunk_text,
"source": source,
"section_idx": section_idx,
"chunk_id": chunk_id,
"length": len(chunk_text),
"metadata": {
"position": f"section_{section_idx}_chunk_{chunk_id}",
"type": "text"
}
})
chunk_id += 1
current_chunk = [line]
current_size = line_tokens
else:
current_chunk.append(line)
current_size += line_tokens
else:
current_chunk.append(line)
current_size += line_tokens
if current_chunk:
chunk_text = '\n'.join(current_chunk)
if len(chunk_text) > min_chunk_size:
chunks.append({
"content": chunk_text,
"source": source,
"section_idx": section_idx,
"chunk_id": chunk_id,
"length": len(chunk_text),
"metadata": {
"position": f"section_{section_idx}_chunk_{chunk_id}",
"type": "text"
}
})
return chunks
What I've learned:
This is where most teams leak money. They pick an expensive embedding model and embed everything. Twice. On updates they re-embed everything.
Embedding costs matter:
For a corpus of 1 million documents at 500 tokens each:
The local model is 1-2% worse at retrieval than the large model. That's acceptable when you have a reranker. The large model won't save you if your chunks are garbage.
python
import os
from typing import List, Dict
import time
class EmbeddingCostCalculator:
"""Track and predict embedding costs."""
COSTS = {
"text-embedding-3-small": 0.02 / 1_000_000, # per token
"text-embedding-3-large": 0.08 / 1_000_000,
"local": 0, # free but compute cost
}
def __init__(self, model: str = "text-embedding-3-small"):
self.model = model
self.total_tokens = 0
self.total_cost = 0
def calculate_cost(self, tokens: int) -> float:
"""Calculate cost for embedding N tokens."""
cost = tokens * self.COSTS.get(self.model, 0)
self.total_tokens += tokens
self.total_cost += cost
return cost
def predict_corpus_cost(self, num_documents: int, avg_tokens_per_doc: int) -> Dict:
"""Predict cost to embed entire corpus."""
total_tokens = num_documents * avg_tokens_per_doc
cost = total_tokens * self.COSTS.get(self.model, 0)
return {
"model": self.model,
"num_documents": num_documents,
"total_tokens": total_tokens,
"cost": cost,
"cost_per_document": cost / num_documents if num_documents > 0 else 0,
}
def compare_models(self, num_documents: int, avg_tokens_per_doc: int) -> Dict:
"""Compare cost across models."""
results = {}
for model in self.COSTS.keys():
self.model = model
results[model] = self.predict_corpus_cost(num_documents, avg_tokens_per_doc)
return results
calculator = EmbeddingCostCalculator()
comparison = calculator.compare_models(100_000, 500)
for model, costs in comparison.items():
print(f"{model}: ${costs['cost']:,.2f}")
My strategy:
This is where 73% of RAG failures happen. You retrieve garbage, the LLM can't fix it.
python
from typing import List, Tuple, Dict
from dataclasses import dataclass
import numpy as np
@dataclass
class RetrievalResult:
content: str
score: float
source: str
rerank_score: float = None
class HybridRetriever:
"""
Hybrid retrieval: combine vector search + BM25 keyword search.
Why? Vector search is great for semantic meaning but misses keywords.
BM25 catches keywords. Together they cover more ground.
"""
def __init__(self, vector_store, bm25_index):
self.vector_store = vector_store # Your vector DB (Pinecone, Weaviate, etc)
self.bm25_index = bm25_index # BM25 for keyword search
def retrieve(self, query: str, top_k: int = 5) -> List[RetrievalResult]:
"""
1. Retrieve top 50 with vector search (broad, semantic)
2. Retrieve top 50 with BM25 (keyword matches)
3. Merge results by reciprocal rank fusion
4. Return top K
"""
vector_results = self.vector_store.search(query, limit=50)
vector_scores = {r['id']: (51 - i) / 51 for i, r in enumerate(vector_results)}
bm25_results = self.bm25_index.search(query, limit=50)
bm25_scores = {r['id']: (51 - i) / 51 for i, r in enumerate(bm25_results)}
combined = {}
for result_id in set(list(vector_scores.keys()) + list(bm25_scores.keys())):
v_score = vector_scores.get(result_id, 0)
b_score = bm25_scores.get(result_id, 0)
combined[result_id] = 0.6 * v_score + 0.4 * b_score
sorted_results = sorted(
combined.items(),
key=lambda x: x[1],
reverse=True
)[:top_k]
return [
RetrievalResult(
content=self._get_content(result_id),
score=score,
source=self._get_source(result_id)
)
for result_id, score in sorted_results
]
def _get_content(self, result_id: str) -> str:
"""Get chunk content by ID."""
return self.vector_store.get(result_id)['content']
def _get_source(self, result_id: str) -> str:
"""Get source document."""
return self.vector_store.get(result_id).get('source', 'unknown')
Real numbers from a project we did: Hybrid retrieval got us to 82% recall on the first pass. Vector-only was 64%. BM25-only was 71%. The combination caught edge cases both missed.
Retrieval finds candidates. Reranking orders them correctly. This is where the magic happens.
python
from sentence_transformers import CrossEncoder
from typing import List, Dict
import numpy as np
class CrossEncoderReranker:
"""
Cross-encoder reranking.
What it does:
- Takes query + each candidate chunk
- Scores how well they match (0-1)
- Returns sorted by actual relevance, not just similarity
Why it works:
- Considers query AND content together
- Catches semantic mismatches that vector search misses
- Puts garbage at the bottom
Trade-off:
- Slow: 50-200ms per query depending on model
- But worth it: quality improvement is 10-30%
"""
def __init__(self, model_name: str = "cross-encoder/mmarco-MiniLMv2-L12-H384-v1"):
self.model = CrossEncoder(model_name)
def rerank(
self,
query: str,
candidates: List[Dict],
top_k: int = 3,
) -> List[Dict]:
"""
Rerank candidates by relevance to query.
Args:
query: User query
candidates: List of retrieved chunks
top_k: Return top K results
Returns:
Reranked candidates with scores
"""
pairs = [
[query, candidate['content']]
for candidate in candidates
]
scores = self.model.predict(pairs)
ranked = sorted(
zip(candidates, scores),
key=lambda x: x[1],
reverse=True
)[:top_k]
results = []
for candidate, score in ranked:
candidate['rerank_score'] = float(score)
results.append(candidate)
return results
reranker = CrossEncoderReranker()
retrieved = [
{"content": "Paris is the capital of France", "score": 0.85},
{"content": "The Eiffel Tower is in Paris", "score": 0.83},
{"content": "France produces wine", "score": 0.72},
]
query = "What is the capital of France?"
reranked = reranker.rerank(query, retrieved, top_k=2)
When to rerank:
Here's how I architect RAG systems that actually scale:
python
from typing import List, Dict, Any
from dataclasses import dataclass
import time
@dataclass
class RAGConfig:
"""Production RAG configuration."""
chunk_size: int = 800 # tokens
chunk_overlap: int = 100 # tokens
min_chunk_size: int = 50 # tokens
initial_retrieval_k: int = 20 # Get top 20 candidates
rerank_k: int = 3 # Rerank to top 3
retrieval_timeout: float = 5.0 # seconds
embedding_model: str = "all-MiniLM-L6-v2" # Local, fast
embedding_batch_size: int = 128
rerank_model: str = "cross-encoder/mmarco-MiniLMv2-L12-H384-v1"
enable_reranking: bool = True
track_costs: bool = True
class ProductionRAG:
"""Complete RAG pipeline for production."""
def __init__(self, config: RAGConfig):
self.config = config
self.retriever = None # Your vector DB + BM25
self.reranker = None # CrossEncoder if enabled
self.metrics = {
"queries_processed": 0,
"total_latency": 0,
"retrieval_latency": 0,
"reranking_latency": 0,
"tokens_used": 0,
}
def query(self, question: str, context_limit: int = 3000) -> Dict[str, Any]:
"""
Process query through full RAG pipeline.
Returns dict with:
- answer: LLM response
- context: Retrieved chunks used
- metrics: Timing and cost info
"""
start_time = time.time()
retrieval_start = time.time()
candidates = self._retrieve(question, k=self.config.initial_retrieval_k)
retrieval_latency = time.time() - retrieval_start
reranking_latency = 0
if self.config.enable_reranking and len(candidates) > self.config.rerank_k:
rerank_start = time.time()
candidates = self._rerank(question, candidates)
reranking_latency = time.time() - rerank_start
context = self._build_context(candidates, limit=context_limit)
answer = self._generate_answer(question, context)
total_latency = time.time() - start_time
self.metrics["queries_processed"] += 1
self.metrics["total_latency"] += total_latency
self.metrics["retrieval_latency"] += retrieval_latency
self.metrics["reranking_latency"] += reranking_latency
return {
"answer": answer,
"context_chunks": len(context),
"metrics": {
"total_ms": round(total_latency * 1000, 2),
"retrieval_ms": round(retrieval_latency * 1000, 2),
"reranking_ms": round(reranking_latency * 1000, 2),
"context_size": len(context),
},
"sources": [c['source'] for c in candidates[:self.config.rerank_k]]
}
def _retrieve(self, query: str, k: int) -> List[Dict]:
"""Hybrid retrieval: vector + BM25."""
return self.retriever.hybrid_search(query, limit=k)
def _rerank(self, query: str, candidates: List[Dict]) -> List[Dict]:
"""Rerank using cross-encoder."""
return self.reranker.rerank(query, candidates, top_k=self.config.rerank_k)
def _build_context(self, chunks: List[Dict], limit: int = 3000) -> List[Dict]:
"""Build context string, respecting token limit."""
context = []
token_count = 0
for chunk in chunks:
chunk_tokens = len(chunk['content'].split())
if token_count + chunk_tokens > limit:
break
context.append(chunk)
token_count += chunk_tokens
return context
def _generate_answer(self, question: str, context: List[Dict]) -> str:
"""Generate answer using LLM + context."""
context_str = "\n\n".join([c['content'] for c in context])
prompt = f"""Use the following context to answer the question.
Context:
{context_str}
Question: {question}
Answer:"""
return "Answer would go here"
def get_metrics(self) -> Dict:
"""Get performance metrics."""
avg_latency = (
self.metrics["total_latency"] / self.metrics["queries_processed"]
if self.metrics["queries_processed"] > 0
else 0
)
return {
"queries_processed": self.metrics["queries_processed"],
"avg_latency_ms": round(avg_latency * 1000, 2),
"avg_retrieval_ms": round(
(self.metrics["retrieval_latency"] / self.metrics["queries_processed"]) * 1000,
2
) if self.metrics["queries_processed"] > 0 else 0,
"avg_reranking_ms": round(
(self.metrics["reranking_latency"] / self.metrics["queries_processed"]) * 1000,
2
) if self.metrics["queries_processed"] > 0 else 0,
}
Here's what we actually spent on a 500k document RAG system:
Initial approach (wrong):
Optimized approach:
Quality didn't drop. We went from 78% recall to 81% recall with better chunking + hybrid retrieval + local reranking.
python
class CostOptimization:
"""Track and optimize RAG costs."""
MONTHLY_COSTS = {
"embeddings": {
"text-embedding-3-large": 400, # 500k docs
"text-embedding-3-small": 100,
"local": 0,
},
"retrieval": {
"pinecone_pro": 84,
"weaviate_cloud": 150,
"weaviate_selfhosted": 120,
},
"reranking": {
"cohere_rerank": 150, # 50k queries
"api_calls": 200,
"local": 0,
},
}
@staticmethod
def compare_strategies() -> Dict:
"""Compare cost of different strategies."""
strategies = {
"expensive": {
"embedding": "text-embedding-3-large",
"retrieval": "pinecone_pro",
"reranking": "cohere_rerank",
"cost": 400 + 84 + 150,
},
"optimized": {
"embedding": "local",
"retrieval": "weaviate_selfhosted",
"reranking": "local",
"cost": 0 + 120 + 0,
},
}
return strategies
@staticmethod
def quality_vs_cost() -> str:
"""What you get for your money."""
return """
Expensive ($684/month):
- 78% recall, 95% precision
- 1ms retrieval latency
- ~2ms reranking latency
- Fully managed
Optimized ($120/month):
- 81% recall, 94% precision
- 5ms retrieval latency (local)
- ~80ms reranking latency
- Self-hosted (dev time cost)
Better quality, 5.7x cheaper. Tradeoff: operational complexity.
"""
Problem 1: "Retrieval keeps returning irrelevant results"
Check in this order:
Code to diagnose:
python
def diagnose_retrieval(query: str, ground_truth_chunk_id: str):
"""Find why retrieval is failing."""
vector_results = vector_search(query, k=50)
vector_rank = next(
(i for i, r in enumerate(vector_results) if r['id'] == ground_truth_chunk_id),
None
)
print(f"Vector search rank: {vector_rank}") # Should be < 5
bm25_results = bm25_search(query, k=50)
bm25_rank = next(
(i for i, r in enumerate(bm25_results) if r['id'] == ground_truth_chunk_id),
None
)
print(f"BM25 rank: {bm25_rank}") # Should be < 10
hybrid_results = hybrid_search(query, k=20)
hybrid_rank = next(
(i for i, r in enumerate(hybrid_results) if r['id'] == ground_truth_chunk_id),
None
)
print(f"Hybrid rank: {hybrid_rank}") # Should be < 5
Problem 2: "Latency is too high (>500ms)"
Find the bottleneck:
python
def find_latency_bottleneck(query: str):
"""Profile where time is being spent."""
import time
start = time.time()
retrieved = retrieval_step(query)
retrieval_time = time.time() - start
start = time.time()
reranked = reranking_step(retrieved)
reranking_time = time.time() - start
start = time.time()
answer = llm_step(reranked, query)
llm_time = time.time() - start
total = retrieval_time + reranking_time + llm_time
print(f"Retrieval: {retrieval_time*1000:.0f}ms ({retrieval_time/total*100:.0f}%)")
print(f"Reranking: {reranking_time*1000:.0f}ms ({reranking_time/total*100:.0f}%)")
print(f"LLM: {llm_time*1000:.0f}ms ({llm_time/total*100:.0f}%)")
print(f"Total: {total*1000:.0f}ms")
Problem 3: "Cost keeps growing"
Track what's actually expensive:
python
def track_costs():
"""Monitor cost drivers."""
embeddings_cost = num_queries * avg_tokens_per_query * embedding_cost_per_token
reranking_cost = num_queries * reranking_cost_per_query
llm_cost = num_queries * context_tokens * llm_cost_per_token
print(f"Embedding cost: ${embeddings_cost}")
print(f"Reranking cost: ${reranking_cost}")
print(f"LLM cost: ${llm_cost}")
print(f"Total: ${embeddings_cost + reranking_cost + llm_cost}")
Before you ship RAG to production:
RAG isn't magic. It's straightforward once you understand the tradeoffs. Some teams get it right from day one. Others iterate for months. The difference isn't usually talent — it's priorities.
What separates working RAG from frustrating RAG:
The teams building RAG systems that work aren't using fancier models or more expensive APIs. They're being precise about what they measure, ruthless about what they optimize, and honest about what doesn't work for their problem.
That's it. No secret sauce.
source & further reading
trpevski.com — original article