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)
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
# This will split sentences, lose context, and waste tokens
text = "The capital of France is Paris. It's known for the Eiffel Tower..."
chunks = fixed_size_chunking(text)
# Result: ['The capital of France is Paris. It's known for the Eiffel ', 'Tower...']
# ^ Garbage. Sentence got split.
2. Semantic Chunking (Better, What You Actually Need)
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)
# Split into sentences (in production, use nltk or spaCy)
sentences = text.split(". ")
sentences = [s.strip() + "." for s in sentences if s.strip()]
if len(sentences) < 2:
return [text]
# Get embeddings
embeddings = model.encode(sentences)
# Calculate similarity between adjacent sentences
chunks = []
current_chunk = [sentences[0]]
for i in range(1, len(sentences)):
# Cosine similarity
similarity = np.dot(embeddings[i], embeddings[i-1]) / (
np.linalg.norm(embeddings[i]) * np.linalg.norm(embeddings[i-1])
)
# If similarity drops (topic changed), start new chunk
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])
# Don't forget the last chunk
if current_chunk:
chunk_text = " ".join(current_chunk)
if len(chunk_text) >= min_chunk_size:
chunks.append(chunk_text)
return chunks
# Usage
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)
# Result: 3 coherent chunks, no split sentences, preserves meaning
In production, I use a hybrid approach:
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
# Split by markdown headers first (preserve structure)
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 adding this line exceeds max, save chunk and start new one
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
# Keep last line for overlap
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
# Save final chunk
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:
But initial embedding is just the start. Real production costs come from:
That's why local embeddings make sense. Even with GPU infrastructure cost amortized, you're usually ahead. 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.
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
# Example: 100k documents, 500 tokens each
calculator = EmbeddingCostCalculator()
comparison = calculator.compare_models(100_000, 500)
for model, costs in comparison.items():
print(f"{model}: ${costs['cost']:,.2f}")
# Output:
# text-embedding-3-small: $1.00
# text-embedding-3-large: $4.00
# local: $0.00
My strategy:
This is where 73% of RAG failures happen. You retrieve garbage, the LLM can't fix it.
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 search (semantic)
vector_results = self.vector_store.search(query, limit=50)
vector_scores = {r['id']: (51 - i) / 51 for i, r in enumerate(vector_results)}
# BM25 search (keywords)
bm25_results = self.bm25_index.search(query, limit=50)
bm25_scores = {r['id']: (51 - i) / 51 for i, r in enumerate(bm25_results)}
# Combine using reciprocal rank fusion
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)
# Weight: 60% semantic, 40% keyword
combined[result_id] = 0.6 * v_score + 0.4 * b_score
# Sort and return top K
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.
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
"""
# Prepare pairs for cross-encoder
pairs = [
[query, candidate['content']]
for candidate in candidates
]
# Score all pairs
scores = self.model.predict(pairs)
# Sort by score
ranked = sorted(
zip(candidates, scores),
key=lambda x: x[1],
reverse=True
)[:top_k]
# Add rerank scores to candidates
results = []
for candidate, score in ranked:
candidate['rerank_score'] = float(score)
results.append(candidate)
return results
# Usage
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)
# Results are perfectly ordered now
# ^ This is crucial when you have 10k documents
When to rerank:
Here's how I architect RAG systems that actually scale:
from typing import List, Dict, Any
from dataclasses import dataclass
import time
@dataclass
class RAGConfig:
"""Production RAG configuration."""
# Chunking
chunk_size: int = 800 # tokens
chunk_overlap: int = 100 # tokens
min_chunk_size: int = 50 # tokens
# Retrieval
initial_retrieval_k: int = 20 # Get top 20 candidates
rerank_k: int = 3 # Rerank to top 3
retrieval_timeout: float = 5.0 # seconds
# Embedding
embedding_model: str = "all-MiniLM-L6-v2" # Local, fast
embedding_batch_size: int = 128
# Reranking
rerank_model: str = "cross-encoder/mmarco-MiniLMv2-L12-H384-v1"
enable_reranking: bool = True
# Cost tracking
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()
# Step 1: Retrieve candidates
retrieval_start = time.time()
candidates = self._retrieve(question, k=self.config.initial_retrieval_k)
retrieval_latency = time.time() - retrieval_start
# Step 2: Rerank if enabled
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
# Step 3: Build context (respecting token limit)
context = self._build_context(candidates, limit=context_limit)
# Step 4: Generate answer with context
answer = self._generate_answer(question, context)
# Track metrics
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."""
# Implementation depends on your vector DB
# This is pseudocode
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."""
# Call your LLM here
# This is pseudocode
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:"""
# response = llm.generate(prompt)
# return response
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.
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:
def diagnose_retrieval(query: str, ground_truth_chunk_id: str):
"""Find why retrieval is failing."""
# Step 1: Vector search ranking
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
# Step 2: BM25 ranking
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
# Step 3: Hybrid ranking
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
# If still not in top-5, rerank might save it
# If not in top-20 at all, problem is in retrieval, not ranking
Problem 2: "Latency is too high (>500ms)"
Find the bottleneck:
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")
# Optimize the biggest component
Problem 3: "Cost keeps growing"
Track what's actually expensive:
def track_costs():
"""Monitor cost drivers."""
# Per query costs
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
# In production, usually LLM cost dominates (60%), not retrieval (10%)
# So reducing context size from 2000 to 1000 tokens saves more than optimizing reranking
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.