A RAG chatbot augments language models with external knowledge retrieval, grounding responses in your actual data to eliminate hallucinations.
Instead of relying on pre-trained knowledge cutoffs, RAG fetches relevant context from your documents, databases, or APIs, making LLMs useful wherever accuracy matters.
If your goals include higher answer accuracy, traceable citations, and production-ready latency, this guide walks through a pragmatic build: from BM25 and GraphRAG foundations to hybrid retrieval, reranking, and low-cost deployment.
We’ll keep this tutorial skimmable and hands-on, with code blocks you can paste into your project. Where many guides leap straight to vector databases, we’ll start with what actually moves metrics in practice.
Quick mindset: Ship a simple, measurable baseline fast (BM25), then layer in semantic vectors, reranking, and caching only where they demonstrably improve outcomes on your real user queries.
Why most RAG chatbot tutorials start in the wrong place
Most RAG chatbot tutorials open with an embedding model and a vector database, then treat retrieval as solved and spend the rest of the word count on prompt formatting. Retrieval is what decides whether the answer is correct. It is also the first thing to break once real users start typing product codes, error strings, internal acronyms, and names that never appeared in the model's training data.
Keyword search is a stronger baseline than it gets credit for
BM25 finds the most relevant documents for a given query by examining one main thing: how often the query words appear in each document. That sounds primitive next to vectors, and for a large class of queries it wins anyway. Exact tokens are exactly what embeddings blur: a part number, a config flag, a stack trace line, a customer name. Ask a semantic index for ERR_CONN_RESET and it will happily hand you documents about connection problems in general. A keyword index hands you the page that contains the string.
The operational argument matters too. A keyword index needs no embedding calls, no model choice, no dimension mismatch, and nothing to re-embed when your documents change. You get a working retrieval layer on day one, and a baseline to measure every later improvement against.
Semantic search earns its place on the other half of the queries
Embeddings pay off when the user's words and the document's words differ: "how do I stop being billed" against a page titled "cancelling your subscription", or a question phrased as a symptom against documentation written as a procedure. Keyword scoring has no way to bridge that gap, because none of the query terms are in the target document.
Hybrid search is ideal for use cases where you want to enable semantic search capabilities for a more human-like search experience but also require exact phrase matching for specific terms, such as product names or serial numbers. Run both retrievers, merge the results, and let a reranker decide the final order. Adding the vector half once BM25 works also tells you whether it helped, which is information you never get if you start with vectors.
What the tutorials skip entirely
The parts that separate a demo from something people trust rarely appear in a beginner walkthrough:
- Chunking. Split size and overlap change your answers more than swapping LLMs does. A chunk that cuts a table in half retrieves badly no matter how good the index is.
- Metadata and filters. Product version, document date, and access permissions belong on every chunk. Without them you will eventually cite a deprecated page or leak a document to the wrong user.
- An evaluation set. Twenty real questions with the source document you expect back, checked after every change. Without it, tuning retrieval is guesswork.
- Behaviour when retrieval fails. If nothing relevant comes back, the system should say so rather than let the model improvise. Grounding only prevents hallucination when the retrieved context is actually relevant.
Tip: Treat vectors as an optimization. If BM25 plus a reranker nails your domain, skip the embedding pipeline and spend the time on chunking, caching, and UX.
Step 1: Skip the Vector Database Hype (Start With BM25)
Before spending money on Pinecone or wrestling with FAISS installations, implement a BM25-based retriever. It's fast, requires zero embeddings, and works surprisingly well for many use cases.
from rank_bm25 import BM25Okapi
import json
import numpy as np
class SimpleBM25Retriever:
def __init__(self, documents):
# Tokenize documents for BM25
self.documents = documents
self.tokenized_docs = [doc.lower().split() for doc in documents]
# Initialize BM25 with custom parameters
# k1 controls term frequency saturation (1.2-2.0 typical)
# b controls document length normalization (0.75 typical)
self.bm25 = BM25Okapi(
self.tokenized_docs,
k1=1.5,
b=0.75
)
def search(self, query, top_k=5):
tokenized_query = query.lower().split()
scores = self.bm25.get_scores(tokenized_query)
# Get top-k document indices
top_indices = np.argsort(scores)[::-1][:top_k]
results = []
for idx in top_indices:
results.append({
'document': self.documents[idx],
'score': scores[idx],
'index': idx
})
return results
This baseline retriever handles 80% of use cases without any ML infrastructure. Test it first - you might not need vectors at all.
Where BM25 shines
- Exact strings & IDs: SKUs, ticket numbers, error codes, API names, product model numbers.
- Long-tail queries: Niche terms with sparse semantic neighbors.
- Compliance & policy snippets: Users often paste exact clauses; BM25 nails them.
Sanity checks to run (5 minutes)
- Throw 20 real queries at BM25 and eyeball top-3 hits.
- If your top-3 are consistently relevant (≥80%), delay vectors and jump to Step 4 (Reranking).
The GraphRAG Alternative
For queries requiring relationship traversal ("Who directed the movie where the Inception actor played a thief?"), consider GraphRAG instead of vectors. GraphRAG builds a knowledge graph over your corpus so the retriever can follow relationships across multiple hops, not only match on semantic similarity.
from neo4j import GraphDatabase
import json
class GraphRAGRetriever:
def __init__(self, uri, user, password):
self.driver = GraphDatabase.driver(uri, auth=(user, password))
def extract_entities_and_relations(self, text):
"""Extract graph structure from text using NER and relation extraction"""
# In production, use spaCy or an LLM for entity extraction
# This is simplified for demonstration
with self.driver.session() as session:
# Create nodes and relationships
session.run("""
MERGE (e1:Entity {name: $entity1})
MERGE (e2:Entity {name: $entity2})
MERGE (e1)-[r:RELATES_TO {type: $relation}]->(e2)
""", entity1="Leonardo DiCaprio",
entity2="Inception",
relation="ACTED_IN")
def graph_search(self, query):
"""Traverse graph based on query patterns"""
with self.driver.session() as session:
# Example: Find movies by actor
result = session.run("""
MATCH (a:Entity)-[:ACTED_IN]->(m:Entity)
WHERE a.name CONTAINS $actor_hint
RETURN m.name as movie, a.name as actor
LIMIT 5
""", actor_hint=query)
return [record for record in result]
When to try GraphRAG: Your domain has rich entities (people, products, contracts) with explicit relationships (owns, depends_on, references, supersedes) and your users ask multi-hop questions.
Step 2: Implement Smart Chunking That Actually Works
Chunking splits documents into pieces small enough to index and retrieve, then feeds them as context to the LLM.
Most tutorials use naive fixed-size chunking, but real documents have structure: use it.
import re
from typing import List, Dict
import hashlib
class SemanticChunker:
def __init__(self,
chunk_size=512,
chunk_overlap=128,
min_chunk_size=100):
self.chunk_size = chunk_size
self.chunk_overlap = chunk_overlap
self.min_chunk_size = min_chunk_size
def chunk_by_structure(self, text: str) -> List[Dict]:
"""Smart chunking that respects document structure"""
chunks = []
# First, try to split by markdown headers
header_pattern = r'^#{1,6}\s+.+$'
sections = re.split(f'({header_pattern})', text, flags=re.MULTILINE)
current_chunk = ""
current_metadata = {}
for i, section in enumerate(sections):
if re.match(header_pattern, section):
# This is a header
if current_chunk and len(current_chunk) > self.min_chunk_size:
chunks.append({
'text': current_chunk.strip(),
'metadata': current_metadata,
'chunk_id': self._generate_chunk_id(current_chunk)
})
current_metadata = {'header': section.strip()}
current_chunk = section + "\n"
else:
# Regular content
if len(current_chunk) + len(section) < self.chunk_size:
current_chunk += section
else:
# Split large sections semantically
sentences = re.split(r'(?<=[.!?])\s+', section)
for sentence in sentences:
if len(current_chunk) + len(sentence) < self.chunk_size:
current_chunk += " " + sentence
else:
if len(current_chunk) > self.min_chunk_size:
chunks.append({
'text': current_chunk.strip(),
'metadata': current_metadata,
'chunk_id': self._generate_chunk_id(current_chunk)
})
# Start new chunk with overlap
overlap_text = self._get_overlap(current_chunk)
current_chunk = overlap_text + sentence
# Don't forget the last chunk
if current_chunk and len(current_chunk) > self.min_chunk_size:
chunks.append({
'text': current_chunk.strip(),
'metadata': current_metadata,
'chunk_id': self._generate_chunk_id(current_chunk)
})
return chunks
def _get_overlap(self, text: str) -> str:
"""Extract overlap text from the end of current chunk"""
words = text.split()
overlap_words = int(self.chunk_overlap / 5) # Rough estimate
return " ".join(words[-overlap_words:]) if len(words) > overlap_words else ""
def _generate_chunk_id(self, text: str) -> str:
"""Generate unique ID for chunk"""
return hashlib.md5(text.encode()).hexdigest()[:8]
Advanced: Contextual Chunking
The method is called "Contextual Retrieval" and uses two sub-techniques: Contextual Embeddings and Contextual BM25. This method can reduce the number of failed retrievals by 49% and, when combined with reranking, by 67%.
Add context to each chunk before embedding:
def add_chunk_context(chunks: List[Dict], document_title: str) -> List[Dict]:
"""Add document context to each chunk for better retrieval"""
for i, chunk in enumerate(chunks):
# Build context from surrounding chunks
context_before = chunks[i-1]['text'][:200] if i > 0 else ""
context_after = chunks[i+1]['text'][:200] if i < len(chunks)-1 else ""
# Prepend contextual information
chunk['contextualized_text'] = f"""
Document: {document_title}
Section: {chunk.get('metadata', {}).get('header', 'Main content')}
Context: This chunk appears after discussing '{context_before}'
and before '{context_after}'.
Content: {chunk['text']}
"""
return chunks
Practical guardrails: Keep chunks self-contained, but not isolated. Preserve headings in metadata for better reranking and for citations in your final answer.
Step 3: Build Your Hybrid Retrieval Pipeline
Now combine BM25 with semantic search, weighting them dynamically based on query type.
import numpy as np
from sentence_transformers import SentenceTransformer
from typing import List, Tuple
import re
class HybridRetriever:
def __init__(self, documents: List[str]):
self.documents = documents
# Initialize BM25
self.bm25_retriever = SimpleBM25Retriever(documents)
# Initialize embedding model (use a small, fast one)
self.encoder = SentenceTransformer('all-MiniLM-L6-v2')
# Pre-compute document embeddings
self.doc_embeddings = self.encoder.encode(documents,
convert_to_tensor=True)
def detect_query_type(self, query: str) -> Tuple[float, float]:
"""Dynamically adjust weights based on query characteristics"""
# Check for exact match indicators
has_quotes = '"' in query
has_product_code = bool(re.search(r'\b[A-Z]{2,}-\d+\b', query))
has_specific_terms = any(term in query.lower()
for term in ['exact', 'specifically', 'model number'])
# Check for semantic indicators
has_semantic_words = any(word in query.lower()
for word in ['similar', 'like', 'about', 'related'])
is_question = query.strip().endswith('?')
# Calculate weights
if has_quotes or has_product_code or has_specific_terms:
# Favor BM25 for exact matches
return 0.7, 0.3
elif has_semantic_words or is_question:
# Favor semantic search
return 0.3, 0.7
else:
# Balanced approach
return 0.5, 0.5
def hybrid_search(self, query: str, top_k: int = 5) -> List[Dict]:
# Get dynamic weights
bm25_weight, semantic_weight = self.detect_query_type(query)
# BM25 search
bm25_results = self.bm25_retriever.search(query, top_k=top_k*2)
# Semantic search
query_embedding = self.encoder.encode(query, convert_to_tensor=True)
semantic_scores = np.dot(self.doc_embeddings, query_embedding)
top_semantic_idx = np.argsort(semantic_scores)[::-1][:top_k*2]
# Normalize and combine scores
scores_dict = {}
# Add BM25 scores
max_bm25 = max([r['score'] for r in bm25_results]) + 1e-6
for result in bm25_results:
idx = result['index']
normalized_score = result['score'] / max_bm25
scores_dict[idx] = bm25_weight * normalized_score
# Add semantic scores
max_semantic = semantic_scores.max() + 1e-6
for idx in top_semantic_idx:
normalized_score = semantic_scores[idx] / max_semantic
if idx in scores_dict:
scores_dict[idx] += semantic_weight * normalized_score
else:
scores_dict[idx] = semantic_weight * normalized_score
# Sort by combined score
sorted_indices = sorted(scores_dict.keys(),
key=lambda x: scores_dict[x],
reverse=True)[:top_k]
results = []
for idx in sorted_indices:
results.append({
'document': self.documents[idx],
'score': scores_dict[idx],
'index': idx,
'retrieval_method': 'hybrid'
})
return results
Why hybrid beats either alone
- BM25 gives you high-precision anchors for exact terms.
- Semantic vectors capture paraphrases and fuzzy intent.
- Dynamic weighting prevents you from hard-coding a one-size-fits-all α/β that degrades edge cases.
Debug trick: Log the chosen weights and top candidates for 100 queries. You’ll quickly see which patterns want BM25-heavy vs. semantic-heavy mixes.
Step 4: Add a Reranker (The Missing Performance Multiplier)
Cross-encoders are the standard models used for reranking in a RAG framework. Unlike retriever functions used in the initial retrieval step, which just take into account the similarity scores of different text chunks, cross-encoders are able to perform a more in-depth comparison of each of the retrieved text chunks with the user's query.
Most RAG implementations skip this step. Don't.
from transformers import AutoTokenizer, AutoModelForSequenceClassification
import torch
class CrossEncoderReranker:
def __init__(self, model_name='cross-encoder/ms-marco-TinyBERT-L-2-v2'):
"""Initialize a lightweight cross-encoder for reranking"""
self.tokenizer = AutoTokenizer.from_pretrained(model_name)
self.model = AutoModelForSequenceClassification.from_pretrained(model_name)
self.model.eval()
def rerank(self, query: str, documents: List[str], top_k: int = 3) -> List[Dict]:
"""Rerank documents using cross-encoder scoring"""
pairs = [[query, doc] for doc in documents]
# Tokenize all pairs
with torch.no_grad():
inputs = self.tokenizer(pairs,
padding=True,
truncation=True,
max_length=512,
return_tensors='pt')
# Get relevance scores
scores = self.model(**inputs).logits.squeeze(-1)
scores = torch.sigmoid(scores).numpy()
# Sort by score
sorted_indices = scores.argsort()[::-1][:top_k]
results = []
for idx in sorted_indices:
results.append({
'document': documents[idx],
'relevance_score': float(scores[idx]),
'original_rank': idx
})
return results
class LLMReranker:
"""Alternative: Use an LLM for reranking when precision is critical"""
def __init__(self, model_name="gpt-3.5-turbo"):
self.model = model_name
def rerank_with_llm(self, query: str, documents: List[str]) -> List[Dict]:
"""Use LLM to score relevance - more accurate but slower"""
prompt = f"""Given the query: "{query}"
Score each document's relevance from 0-10:
{chr(10).join([f"Document {i+1}: {doc[:200]}..." for i, doc in enumerate(documents)])}
Return only scores as: [score1, score2, ...]"""
# In production, call your LLM API here
# This is a placeholder for the pattern
scores = [7, 3, 9, 5, 8] # Mock scores
ranked_docs = []
for i, (doc, score) in enumerate(zip(documents, scores)):
ranked_docs.append({
'document': doc,
'llm_relevance': score,
'index': i
})
return sorted(ranked_docs, key=lambda x: x['llm_relevance'], reverse=True)
Pragmatic guidance
- Start with a tiny cross-encoder for cost/latency.
- Keep top-10 from retrieval, rerank to top-3, and pass those to generation.
- If latency allows, sample an LLM-based reranker on a subset to validate improvements before switching.
Step 5: Construct the Generation Chain
Now assemble everything into a pipeline you can deploy. But here's where we diverge from typical tutorials - implement response caching and fallback strategies.
import hashlib
import json
from typing import Dict, List, Optional
import redis
class RAGPipeline:
def __init__(self,
documents: List[str],
use_cache: bool = True,
cache_ttl: int = 3600):
# Initialize components
self.hybrid_retriever = HybridRetriever(documents)
self.reranker = CrossEncoderReranker()
# Initialize cache (Redis for production, dict for development)
self.use_cache = use_cache
if use_cache:
try:
self.cache = redis.Redis(host='localhost', port=6379, db=0)
self.cache_ttl = cache_ttl
except:
# Fallback to in-memory cache
self.cache = {}
print("Redis not available, using in-memory cache")
def generate_cache_key(self, query: str, context: str) -> str:
"""Generate deterministic cache key"""
combined = f"{query}:{context[:500]}"
return hashlib.md5(combined.encode()).hexdigest()
def retrieve_and_generate(self,
query: str,
llm_function,
max_context_length: int = 2000) -> Dict:
# Check cache first
if self.use_cache:
cache_key = self.generate_cache_key(query, query)
cached = self._get_from_cache(cache_key)
if cached:
return cached
# Step 1: Hybrid retrieval
initial_results = self.hybrid_retriever.hybrid_search(query, top_k=10)
# Step 2: Reranking
documents = [r['document'] for r in initial_results]
reranked = self.reranker.rerank(query, documents, top_k=3)
# Step 3: Build context (with token limit management)
context = self._build_context(reranked, max_context_length)
# Step 4: Generate response
response = self._generate_with_fallback(query, context, llm_function)
# Cache the result
if self.use_cache:
self._cache_result(cache_key, response)
return response
def _build_context(self, documents: List[Dict], max_length: int) -> str:
"""Build context with smart truncation"""
context = ""
for i, doc in enumerate(documents):
doc_text = doc['document']
# Add source citation
formatted = f"\n[Source {i+1} - Relevance: {doc['relevance_score']:.2f}]\n{doc_text}\n"
# Check if adding this would exceed limit
if len(context) + len(formatted) > max_length:
# Truncate the last document to fit
remaining = max_length - len(context) - 50 # Buffer
if remaining > 100: # Only add if meaningful
formatted = formatted[:remaining] + "..."
context += formatted
break
context += formatted
return context
def _generate_with_fallback(self, query: str, context: str, llm_function) -> Dict:
"""Generate with fallback strategies"""
prompt = f"""Answer the question based on the context provided.
If the context doesn't contain the answer, say "I cannot find this information in the provided context."
Context:
{context}
Question: {query}
Answer:"""
try:
# Primary LLM call
response = llm_function(prompt)
# Validate response
if self._is_valid_response(response):
return {
'answer': response,
'context_used': context[:500],
'status': 'success'
}
else:
# Fallback to simpler extraction
return self._extractive_fallback(query, context)
except Exception as e:
# Ultimate fallback: extractive answer
return {
'answer': self._extractive_fallback(query, context)['answer'],
'context_used': context[:500],
'status': 'fallback',
'error': str(e)
}
def _extractive_fallback(self, query: str, context: str) -> Dict:
"""Simple extractive answer when generation fails"""
# Find the most relevant sentence
sentences = context.split('.')
query_words = set(query.lower().split())
best_sentence = ""
best_score = 0
for sentence in sentences:
sentence_words = set(sentence.lower().split())
overlap = len(query_words.intersection(sentence_words))
if overlap > best_score:
best_score = overlap
best_sentence = sentence.strip()
return {
'answer': best_sentence if best_sentence else "No relevant information found.",
'method': 'extractive_fallback'
}
def _is_valid_response(self, response: str) -> bool:
"""Validate LLM response"""
if not response or len(response) < 10:
return False
# Check for common failure patterns
failure_patterns = [
"i cannot assist",
"i don't have access",
"error occurred",
"context is unclear"
]
response_lower = response.lower()
return not any(pattern in response_lower for pattern in failure_patterns)
def _get_from_cache(self, key: str) -> Optional[Dict]:
"""Retrieve from cache"""
if isinstance(self.cache, dict):
return self.cache.get(key)
else:
cached = self.cache.get(key)
return json.loads(cached) if cached else None
def _cache_result(self, key: str, result: Dict):
"""Store in cache"""
if isinstance(self.cache, dict):
self.cache[key] = result
else:
self.cache.setex(key, self.cache_ttl, json.dumps(result))
Production notes
- Cache keys should include a hash of normalized query + top context IDs to avoid stale collisions.
- Always log which sources made it into the context; this is invaluable for debugging user complaints.
- Add a truthiness check in the prompt (“If not in context, say you can’t find it”) to curb hallucinations.
Step 6: Deploy Without the Cloud Tax
Instead of defaulting to expensive managed services, here's a production setup that runs on a $20/month VPS:
# docker-compose.yml for self-hosted RAG
"""
version: '3.8'
services:
qdrant:
image: qdrant/qdrant
ports:
- "6333:6333"
volumes:
- ./qdrant_storage:/qdrant/storage
redis:
image: redis:alpine
ports:
- "6379:6379"
volumes:
- ./redis_data:/data
api:
build: .
ports:
- "8000:8000"
environment:
- QDRANT_URL=http://qdrant:6333
- REDIS_URL=redis://redis:6379
depends_on:
- qdrant
- redis
"""
# FastAPI server with streaming responses
from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
import asyncio
from typing import AsyncGenerator
app = FastAPI()
# Initialize your RAG pipeline
rag_pipeline = RAGPipeline(documents=load_your_documents())
@app.post("/chat")
async def chat_endpoint(query: str):
"""Streaming RAG endpoint"""
async def generate_stream() -> AsyncGenerator[str, None]:
# Retrieve context
context = rag_pipeline.hybrid_retriever.hybrid_search(query, top_k=5)
# Stream the response
response = rag_pipeline.retrieve_and_generate(
query,
llm_function=your_llm_function
)
# Stream response in chunks
for i in range(0, len(response['answer']), 20):
chunk = response['answer'][i:i+20]
yield f"data: {json.dumps({'text': chunk})}\n\n"
await asyncio.sleep(0.1) # Simulate streaming
return StreamingResponse(
generate_stream(),
media_type="text/event-stream"
)
@app.get("/health")
async def health_check():
return {"status": "healthy", "cache_size": len(rag_pipeline.cache)}
Production Optimization Tricks
- Use SQLite for Small-Scale Vector Storage
Instead of complex vector databases, SQLite with the sqlite-vss extension handles millions of vectors efficiently:
import sqlite3
import sqlite_vss
import numpy as np
def setup_sqlite_vector_store():
conn = sqlite3.connect('vectors.db')
conn.enable_load_extension(True)
sqlite_vss.load(conn)
# Create vector table
conn.execute("""
CREATE VIRTUAL TABLE IF NOT EXISTS doc_vectors USING vss0(
embedding(384),
document_id INTEGER PRIMARY KEY,
document TEXT
)
""")
return conn
def search_vectors(conn, query_embedding, k=5):
results = conn.execute("""
SELECT document_id, document, distance
FROM doc_vectors
WHERE vss_search(embedding, ?)
LIMIT ?
""", (query_embedding.tobytes(), k))
return results.fetchall()
- Implement Request Batching
from collections import defaultdict
import asyncio
class BatchedRAG:
def __init__(self, rag_pipeline, batch_size=10, batch_timeout=0.1):
self.rag_pipeline = rag_pipeline
self.batch_size = batch_size
self.batch_timeout = batch_timeout
self.pending_requests = []
self.results = {}
async def process_query(self, query_id: str, query: str):
"""Add query to batch and wait for result"""
future = asyncio.Future()
self.pending_requests.append((query_id, query, future))
# Trigger batch processing if size reached
if len(self.pending_requests) >= self.batch_size:
await self._process_batch()
else:
# Schedule timeout-based processing
asyncio.create_task(self._timeout_trigger())
return await future
async def _process_batch(self):
"""Process all pending requests in batch"""
if not self.pending_requests:
return
batch = self.pending_requests[:self.batch_size]
self.pending_requests = self.pending_requests[self.batch_size:]
# Batch retrieve
queries = [q for _, q, _ in batch]
contexts = self._batch_retrieve(queries)
# Generate responses
for (query_id, query, future), context in zip(batch, contexts):
response = await self._generate_response(query, context)
future.set_result(response)
def _batch_retrieve(self, queries: List[str]) -> List[str]:
"""Efficient batch retrieval"""
# Encode all queries at once
all_results = []
for query in queries:
results = self.rag_pipeline.hybrid_retriever.hybrid_search(query)
all_results.append(results)
return all_results
Ops checklist: Add /health, /metrics, and /cache introspection endpoints. Log query → weights → candidates → reranked → final context for a 1% sample to investigate misses.
Performance Metrics That Actually Matter
Track these metrics in production:
- First Token Latency: Time to first streamed token (target: <500ms)
- Retrieval Precision@3: Are top 3 results relevant? (target: >80%)
- Cache Hit Rate: Percentage of cached responses (target: >30%)
- Fallback Rate: How often generation fails (target: <5%)
class RAGMetrics:
def __init__(self):
self.metrics = defaultdict(list)
def track_retrieval(self, query, retrieved_docs, relevant_docs):
"""Track retrieval metrics"""
# Precision@k
k = 3
top_k = retrieved_docs[:k]
relevant_in_top_k = sum(1 for doc in top_k if doc in relevant_docs)
precision_at_k = relevant_in_top_k / k
self.metrics['precision@3'].append(precision_at_k)
# Mean Reciprocal Rank (MRR)
for i, doc in enumerate(retrieved_docs):
if doc in relevant_docs:
self.metrics['mrr'].append(1 / (i + 1))
break
else:
self.metrics['mrr'].append(0)
def get_summary(self):
return {
metric: np.mean(values)
for metric, values in self.metrics.items()
}
Benchlike, not benchmark: Use your own gold labels. Ask SMEs to mark “good enough” vs. “wrong,” then optimize Precision@3 and MRR on that set. That’s how you avoid overfitting to public leaderboards that don’t represent your corpus.
Where RAG quality actually comes from
Chunking is a trade-off with no correct answer. Pack too much into a chunk and its vector averages out into something vague that matches everything and answers nothing. Pack too little and the chunk loses the context that made it meaningful, so a retrieved sentence arrives without the heading, product name, or date that gave it sense. This RAG chatbot tutorial spends more effort on retrieval than on prompt wording for exactly that reason.
Some practical ways to blunt the trade-off:
- Split on structure the document already has (headings, sections, paragraphs) instead of a blind character count that cuts sentences in half.
- Overlap adjacent chunks so a fact that straddles a boundary survives in at least one of them.
- Store the section heading and source metadata with each chunk, and prepend the heading to the embedded text so the vector carries its topic.
- Embed small units for precision, then hand the LLM the surrounding parent section for context.
Fix the retrieval layer first
A fast BM25 search with good reranking usually beats an expensive vector database fed by bad chunks. Keyword search still wins on exact strings that embeddings blur together: error codes, SKUs, version identifiers, surnames. Run both, merge the candidate lists, then let a cross-encoder reranker score the merged pool against the actual question and pass only the top results to the model.
Before blaming the LLM for a wrong answer, check whether the chunk containing the answer was in the retrieved set at all. Keep a small set of real questions with the passages that should answer them, and measure how often retrieval surfaces them. If the right chunk never arrives, no prompt tuning and no model upgrade will save the response.
Heuristic to live by: if reranking a smaller candidate pool lifts answer quality more than swapping LLMs does, put your next hour of work into retrieval.
What's Next?
- Test BM25 First: Before investing in vector infrastructure, establish a baseline with keyword search.
- Measure Real Queries: Log actual user queries and optimize for those patterns, not synthetic benchmarks.
- Consider GraphRAG: For relationship-heavy data, graph traversal beats semantic similarity.
- Cache Aggressively: Most queries follow a power law - cache the common ones.
- Monitor Fallbacks: When generation fails, your extractive fallback is your safety net.
Remember: The best RAG system is the one that ships. Start simple with BM25, add vectors when needed, and only adopt GraphRAG when relationships matter. Your users care about accuracy and speed, not your embedding model.
