Building a Production-Ready RAG System: Architecture, Optimization & Monitoring
A complete guide to building Retrieval-Augmented Generation systems that work reliably in production.
Introduction
RAG is the dominant pattern for grounding LLMs in proprietary data. But prototype ≠ production.
This guide covers the full lifecycle: ingestion → retrieval → generation → evaluation → monitoring.
Architecture Overview
┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ Ingest │────▶│ Embedding │────▶│ Retrieval │────▶│ Generation │
│ (Documents)│ │ (Vectors) │ │ (Top-K) │ │ (Answer) │
└─────────────┘ └─────────────┘ └─────────────┘ └─────────────┘
│ │ │ │
▼ ▼ ▼ ▼
Chunking Vector DB Reranking LLM + Context
Strategy (Qdrant/ (Cross- (Citations,
(Semantic/ Pinecone/ encoder/ Guardrails)
Fixed-size) Weaviate) MMR)
Phase 1: Document Ingestion & Chunking
Smart Chunking Strategy
# ingestion/chunker.py
from typing import List
from dataclasses import dataclass
import re
@dataclass
class Chunk:
content: str
metadata: dict
token_count: int
class SemanticChunker:
"""Chunk by semantic boundaries (headers, paragraphs) not fixed size."""
def __init__(self, max_tokens: int = 512, overlap: int = 50):
self.max_tokens = max_tokens
self.overlap = overlap
def chunk(self, text: str, metadata: dict) -> List[Chunk]:
# Split by markdown headers first
sections = self._split_by_headers(text)
chunks = []
for section in sections:
section_chunks = self._chunk_section(section, metadata)
chunks.extend(section_chunks)
return chunks
def _split_by_headers(self, text: str) -> List[tuple[str, dict]]:
"""Split by # ## ### headers, preserve hierarchy."""
pattern = r'^(#{1,6})\s+(.+)$'
lines = text.split('\n')
sections = []
current_header = {"h1": "", "h2": "", "h3": ""}
current_content = []
for line in lines:
match = re.match(pattern, line)
if match:
# Save previous section
if current_content:
header_path = " > ".join([v for v in current_header.values() if v])
sections.append(("\n".join(current_content), {"header_path": header_path}))
# Update header hierarchy
level = len(match.group(1))
header_text = match.group(2)
current_header[f"h{level}"] = header_text
# Clear deeper levels
for l in range(level + 1, 7):
current_header[f"h{l}"] = ""
current_content = [line]
else:
current_content.append(line)
# Last section
if current_content:
header_path = " > ".join([v for v in current_header.values() if v])
sections.append(("\n".join(current_content), {"header_path": header_path}))
return sections
def _chunk_section(self, section: tuple[str, dict], base_metadata: dict) -> List[Chunk]:
content, section_meta = section
metadata = {**base_metadata, **section_meta}
# Split by paragraphs
paragraphs = [p.strip() for p in content.split('\n\n') if p.strip()]
chunks = []
current_chunk = []
current_tokens = 0
for para in paragraphs:
para_tokens = self._estimate_tokens(para)
if current_tokens + para_tokens > self.max_tokens and current_chunk:
# Finalize current chunk
chunk_text = "\n\n".join(current_chunk)
chunks.append(Chunk(chunk_text, metadata, current_tokens))
# Start new chunk with overlap
overlap_text = self._get_overlap(current_chunk)
current_chunk = [overlap_text, para] if overlap_text else [para]
current_tokens = self._estimate_tokens("\n\n".join(current_chunk))
else:
current_chunk.append(para)
current_tokens += para_tokens
# Last chunk
if current_chunk:
chunk_text = "\n\n".join(current_chunk)
chunks.append(Chunk(chunk_text, metadata, current_tokens))
return chunks
def _estimate_tokens(self, text: str) -> int:
return len(text) // 4 # Rough estimate
def _get_overlap(self, chunks: List[str]) -> str:
if not chunks:
return ""
last = chunks[-1]
words = last.split()
overlap_words = words[-(self.overlap // 4):] # Rough token->word
return " ".join(overlap_words)
Metadata Enrichment
# ingestion/enrich.py
from datetime import datetime
import hashlib
def enrich_metadata(doc: dict, chunk: Chunk) -> dict:
"""Add rich metadata for filtering and debugging."""
return {
**chunk.metadata,
"doc_id": doc["id"],
"doc_title": doc.get("title", ""),
"doc_source": doc.get("source", ""),
"doc_created": doc.get("created_at", ""),
"chunk_hash": hashlib.md5(chunk.content.encode()).hexdigest()[:12],
"ingested_at": datetime.utcnow().isoformat(),
"token_count": chunk.token_count,
# Custom fields for your domain
"department": doc.get("department", "general"),
"confidentiality": doc.get("confidentiality", "public"),
"version": doc.get("version", "1.0"),
}
Phase 2: Embedding & Vector Storage
Embedding Model Selection
| Model | Dimensions | Speed | Quality | Cost | Best For |
|---|---|---|---|---|---|
text-embedding-3-small |
1536 | Fast | Good | $0.02/M | General purpose |
text-embedding-3-large |
3072 | Medium | Best | $0.13/M | High accuracy needs |
bge-base-en-v1.5 |
768 | Fast | Good | Free (local) | Privacy/offline |
bge-large-en-v1.5 |
1024 | Medium | Better | Free (local) | Quality + privacy |
e5-base-v2 |
768 | Fast | Good | Free (local) | Multilingual |
mxbai-embed-large-v1 |
1024 | Fast | Excellent | Free (local) | Best open source |
Vector Database Comparison
# vectorstore/qdrant_client.py
from qdrant_client import QdrantClient
from qdrant_client.models import (
Distance, VectorParams, PointStruct,
Filter, FieldCondition, MatchValue,
SearchParams, HnswConfigDiff
)
from typing import List, Optional
import uuid
class QdrantVectorStore:
def __init__(self, url: str, api_key: str, collection_name: str):
self.client = QdrantClient(url=url, api_key=api_key)
self.collection_name = collection_name
def create_collection(self, vector_size: int, distance: Distance = Distance.COSINE):
self.client.create_collection(
collection_name=self.collection_name,
vectors_config=VectorParams(
size=vector_size,
distance=distance,
hnsw_config=HnswConfigDiff(
m=16, # Connections per layer
ef_construct=100 # Build-time accuracy
)
)
)
def upsert_chunks(self, chunks: List[Chunk], embeddings: List[List[float]]):
points = [
PointStruct(
id=str(uuid.uuid4()),
vector=emb,
payload={
"content": chunk.content,
"metadata": chunk.metadata
}
)
for chunk, emb in zip(chunks, embeddings)
]
self.client.upsert(
collection_name=self.collection_name,
points=points
)
def search(
self,
query_vector: List[float],
limit: int = 10,
filter_conditions: Optional[dict] = None,
score_threshold: float = 0.7
) -> List[dict]:
query_filter = None
if filter_conditions:
conditions = [
FieldCondition(key=f"metadata.{k}", match=MatchValue(value=v))
for k, v in filter_conditions.items()
]
query_filter = Filter(must=conditions)
results = self.client.search(
collection_name=self.collection_name,
query_vector=query_vector,
limit=limit,
query_filter=query_filter,
score_threshold=score_threshold,
search_params=SearchParams(
hnsw_ef=128, # Search-time accuracy
exact=False
)
)
return [
{
"content": r.payload["content"],
"metadata": r.payload["metadata"],
"score": r.score
}
for r in results
]
def hybrid_search(
self,
query_vector: List[float],
query_text: str,
limit: int = 10,
sparse_weight: float = 0.3
) -> List[dict]:
"""Combine dense + sparse (BM25) search."""
# Requires Qdrant hybrid index setup
# This is a simplified version
dense_results = self.search(query_vector, limit * 2)
# In production, use Qdrant's native hybrid search
# or combine with separate BM25 index
return dense_results[:limit]
Phase 3: Retrieval Optimization
Multi-Stage Retrieval Pipeline
# retrieval/pipeline.py
from typing import List, Callable
from dataclasses import dataclass
@dataclass
class RetrievedDoc:
content: str
metadata: dict
score: float
rank: int
class RetrievalPipeline:
def __init__(
self,
embedder: Callable[[str], List[float]],
vector_store,
reranker: Callable[[str, List[str]], List[float]] = None
):
self.embedder = embedder
self.vector_store = vector_store
self.reranker = reranker
def retrieve(
self,
query: str,
k: int = 5,
fetch_k: int = 20,
filters: dict = None,
use_reranker: bool = True
) -> List[RetrievedDoc]:
# Stage 1: Embed query
query_vector = self.embedder(query)
# Stage 2: Vector search (fetch more for reranking)
raw_results = self.vector_store.search(
query_vector=query_vector,
limit=fetch_k,
filter_conditions=filters
)
# Stage 3: Rerank (optional but recommended)
if use_reranker and self.reranker and raw_results:
contents = [r["content"] for r in raw_results]
rerank_scores = self.reranker(query, contents)
for r, score in zip(raw_results, rerank_scores):
r["rerank_score"] = score
# Sort by rerank score
raw_results.sort(key=lambda x: x.get("rerank_score", 0), reverse=True)
# Stage 4: Final selection
final = raw_results[:k]
return [
RetrievedDoc(
content=r["content"],
metadata=r["metadata"],
score=r.get("rerank_score", r["score"]),
rank=i
)
for i, r in enumerate(final)
]
# Cross-encoder reranker (best quality)
from sentence_transformers import CrossEncoder
reranker_model = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")
def rerank(query: str, documents: List[str]) -> List[float]:
pairs = [(query, doc) for doc in documents]
scores = reranker_model.predict(pairs)
return scores.tolist()
# MMR (Maximal Marginal Relevance) for diversity
def mmr_retrieve(
query_vector: List[float],
candidate_vectors: List[List[float]],
candidate_docs: List[RetrievedDoc],
k: int = 5,
lambda_mult: float = 0.5
) -> List[RetrievedDoc]:
"""Select diverse results using MMR."""
import numpy as np
selected = []
remaining = list(range(len(candidate_docs)))
# First: highest score
best_idx = max(remaining, key=lambda i: candidate_docs[i].score)
selected.append(best_idx)
remaining.remove(best_idx)
while len(selected) < k and remaining:
mmr_scores = []
for idx in remaining:
# Relevance to query
relevance = candidate_docs[idx].score
# Max similarity to already selected
max_sim = max(
cosine_similarity(
candidate_vectors[idx],
candidate_vectors[sel]
)
for sel in selected
)
# MMR score
mmr_score = lambda_mult * relevance - (1 - lambda_mult) * max_sim
mmr_scores.append((mmr_score, idx))
best = max(mmr_scores, key=lambda x: x[0])
selected.append(best[1])
remaining.remove(best[1])
return [candidate_docs[i] for i in selected]
def cosine_similarity(a: List[float], b: List[float]) -> float:
import numpy as np
a, b = np.array(a), np.array(b)
return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))
Query Expansion & Rewrite
# retrieval/query_rewrite.py
from langchain_openai import ChatOpenAI
from langchain_core.prompts import PromptTemplate
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
QUERY_REWRITE_PROMPT = """Rewrite the user query to be more effective for vector search.
Original query: {query}
Guidelines:
- Expand acronyms and abbreviations
- Add synonyms and related terms
- Make implicit context explicit
- Break complex questions into sub-questions
- Keep the core intent
Return 3-5 rewritten queries, one per line:"""
def rewrite_query(query: str) -> List[str]:
prompt = PromptTemplate.from_template(QUERY_REWRITE_PROMPT)
response = llm.invoke(prompt.format(query=query))
rewrites = [q.strip() for q in response.content.split('\n') if q.strip()]
return [query] + rewrites # Include original
def multi_query_retrieve(
pipeline: RetrievalPipeline,
query: str,
k: int = 5
) -> List[RetrievedDoc]:
"""Retrieve for multiple query variations, merge results."""
queries = rewrite_query(query)
all_results = []
for q in queries:
results = pipeline.retrieve(q, k=k, fetch_k=k*2)
all_results.extend(results)
# Deduplicate by content hash
seen = set()
unique = []
for r in all_results:
content_hash = hash(r.content[:200])
if content_hash not in seen:
seen.add(content_hash)
unique.append(r)
# Rescore and sort
unique.sort(key=lambda x: x.score, reverse=True)
return unique[:k]
Phase 4: Generation with Citations & Guardrails
Structured Generation
# generation/rag_chain.py
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser
from pydantic import BaseModel, Field
from typing import List, Optional
class CitedAnswer(BaseModel):
answer: str = Field(description="The complete answer")
citations: List[dict] = Field(
description="List of citations: [{doc_id, snippet, metadata}]"
)
confidence: float = Field(ge=0, le=1, description="Confidence 0-1")
needs_clarification: bool = Field(default=False)
RAG_PROMPT = """You are a precise assistant that answers questions using ONLY the provided context.
**Context Documents:**
{context}
**Question:** {question}
**Instructions:**
1. Answer ONLY from the provided context
2. If context is insufficient, say "I don't have enough information to answer this"
3. Cite sources inline using [doc_id] format
4. Be concise but complete
5. Include confidence score (0-1) based on context coverage
**Output Format (JSON):**
{{
"answer": "Your answer with [doc_id] citations",
"citations": [
{{"doc_id": "abc123", "snippet": "relevant excerpt", "metadata": {{...}}}}
],
"confidence": 0.9,
"needs_clarification": false
}}"""
def build_rag_chain(model_name: str = "gpt-4o"):
llm = ChatOpenAI(model=model_name, temperature=0)
prompt = ChatPromptTemplate.from_template(RAG_PROMPT)
parser = JsonOutputParser(pydantic_object=CitedAnswer)
chain = prompt | llm | parser
return chain
def format_context(docs: List[RetrievedDoc]) -> str:
"""Format retrieved docs for prompt."""
formatted = []
for i, doc in enumerate(docs):
doc_id = doc.metadata.get("chunk_hash", f"doc_{i}")
formatted.append(
f"[Document {doc_id}]\n"
f"Source: {doc.metadata.get('doc_title', 'Unknown')}\n"
f"Section: {doc.metadata.get('header_path', 'N/A')}\n"
f"Content: {doc.content}\n"
)
return "\n---\n".join(formatted)
Guardrails & Safety
# generation/guardrails.py
import re
from typing import Tuple
class RAGGuardrails:
def __init__(self):
self.pii_patterns = [
(r'\b\d{3}-\d{2}-\d{4}\b', 'SSN'),
(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', 'EMAIL'),
(r'\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b', 'CREDIT_CARD'),
(r'\b\d{10}\b', 'PHONE'),
]
self.refusal_triggers = [
"ignore previous instructions",
"system prompt",
"reveal your prompt",
"bypass",
"jailbreak",
]
def check_input(self, query: str) -> Tuple[bool, str]:
"""Validate user input."""
query_lower = query.lower()
for trigger in self.refusal_triggers:
if trigger in query_lower:
return False, "I can't help with that request."
# Check for PII in query
for pattern, pii_type in self.pii_patterns:
if re.search(pattern, query):
return False, f"Please don't share {pii_type} in your question."
return True, ""
def check_output(self, answer: str, context: str) -> Tuple[bool, str]:
"""Validate generated answer."""
# Check for hallucination (claims not in context)
# Simplified: check if numbers/entities in answer exist in context
answer_numbers = re.findall(r'\b\d+\.?\d*\b', answer)
context_numbers = set(re.findall(r'\b\d+\.?\d*\b', context))
for num in answer_numbers:
if num not in context_numbers and len(num) > 2:
return False, f"Unverified claim detected: {num}"
# Check for PII leakage
for pattern, pii_type in self.pii_patterns:
if re.search(pattern, answer):
return False, f"Potential {pii_type} leakage in response."
return True, ""
def sanitize_output(self, answer: str) -> str:
"""Remove any PII from output."""
sanitized = answer
for pattern, _ in self.pii_patterns:
sanitized = re.sub(pattern, "[REDACTED]", sanitized)
return sanitized
Phase 5: Evaluation & Monitoring
Offline Evaluation
# evaluation/rag_eval.py
from dataclasses import dataclass
from typing import List, Callable
import json
@dataclass
class RAGTestCase:
question: str
expected_answer: str
relevant_doc_ids: List[str] # Ground truth relevant docs
metadata_filters: dict = None
class RAGEvaluator:
def __init__(self, rag_system):
self.rag = rag_system
def evaluate_retrieval(self, test_cases: List[RAGTestCase]) -> dict:
"""Measure retrieval quality."""
metrics = {"precision@k": [], "recall@k": [], "mrr": [], "ndcg": []}
for tc in test_cases:
results = self.rag.retrieve(tc.question, k=10)
retrieved_ids = [r.metadata.get("chunk_hash") for r in results]
relevant = set(tc.relevant_doc_ids)
# Precision@K
for k in [1, 3, 5, 10]:
retrieved_k = set(retrieved_ids[:k])
prec = len(retrieved_k & relevant) / k if k > 0 else 0
metrics[f"precision@{k}"].append(prec)
# Recall@K
for k in [1, 3, 5, 10]:
retrieved_k = set(retrieved_ids[:k])
rec = len(retrieved_k & relevant) / len(relevant) if relevant else 1
metrics[f"recall@{k}"].append(rec)
# MRR (Mean Reciprocal Rank)
for i, rid in enumerate(retrieved_ids):
if rid in relevant:
metrics["mrr"].append(1 / (i + 1))
break
else:
metrics["mrr"].append(0)
# Average
return {k: sum(v) / len(v) for k, v in metrics.items()}
def evaluate_generation(self, test_cases: List[RAGTestCase]) -> dict:
"""Measure answer quality using LLM judge."""
from evaluation.judge import evaluate_response
scores = {"accuracy": [], "completeness": [], "citation_quality": []}
for tc in test_cases:
result = self.rag.query(tc.question)
eval_result = evaluate_response(
task="Answer question using provided context",
input_text=tc.question,
response=result.answer,
criteria=[
"Accuracy: Does the answer correctly address the question?",
"Completeness: Does it cover all relevant aspects?",
"Citation quality: Are citations accurate and relevant?",
"Faithfulness: Does it only use information from context?"
]
)
scores["accuracy"].append(eval_result.scores.get("accuracy", 0))
scores["completeness"].append(eval_result.scores.get("completeness", 0))
scores["citation_quality"].append(eval_result.scores.get("citation_quality", 0))
return {k: sum(v) / len(v) for k, v in scores.items()}
Online Monitoring (Production)
# monitoring/rag_monitor.py
from dataclasses import dataclass, asdict
from datetime import datetime
import json
import asyncio
from collections import defaultdict
@dataclass
class RAGMetrics:
timestamp: str
query: str
latency_ms: float
retrieval_latency_ms: float
generation_latency_ms: float
num_retrieved: int
num_cited: int
confidence: float
user_feedback: Optional[str] = None # thumbs up/down
error: Optional[str] = None
class RAGMonitor:
def __init__(self, log_path: str = "logs/rag_metrics.jsonl"):
self.log_path = log_path
self.buffer = []
self.buffer_size = 100
def log(self, metrics: RAGMetrics):
self.buffer.append(asdict(metrics))
if len(self.buffer) >= self.buffer_size:
self.flush()
def flush(self):
with open(self.log_path, "a") as f:
for m in self.buffer:
f.write(json.dumps(m) + "\n")
self.buffer.clear()
def get_stats(self, hours: int = 24) -> dict:
"""Compute rolling statistics."""
import pandas as pd
df = pd.read_json(self.log_path, lines=True)
df["timestamp"] = pd.to_datetime(df["timestamp"])
cutoff = pd.Timestamp.utcnow() - pd.Timedelta(hours=hours)
df = df[df["timestamp"] > cutoff]
return {
"total_queries": len(df),
"avg_latency_ms": df["latency_ms"].mean(),
"p95_latency_ms": df["latency_ms"].quantile(0.95),
"avg_confidence": df["confidence"].mean(),
"error_rate": df["error"].notna().mean(),
"avg_retrieved": df["num_retrieved"].mean(),
"feedback_positive_rate": (
df["user_feedback"].eq("positive").mean()
if "user_feedback" in df.columns else 0
)
}
# Auto-alerting
def check_alerts(stats: dict, thresholds: dict) -> List[str]:
alerts = []
if stats["error_rate"] > thresholds.get("error_rate", 0.05):
alerts.append(f"High error rate: {stats['error_rate']:.1%}")
if stats["avg_latency_ms"] > thresholds.get("latency_ms", 5000):
alerts.append(f"High latency: {stats['avg_latency_ms']:.0f}ms")
if stats["avg_confidence"] < thresholds.get("confidence", 0.7):
alerts.append(f"Low confidence: {stats['avg_confidence']:.2f}")
return alerts
Phase 6: Deployment & Scaling
Docker Compose for Local Dev
# docker-compose.yml
version: '3.8'
services:
qdrant:
image: qdrant/qdrant:v1.10
ports:
- "6333:6333"
- "6334:6334"
volumes:
- qdrant_data:/qdrant/storage
environment:
- QDRANT__SERVICE__HTTP_PORT=6333
ollama:
image: ollama/ollama:latest
ports:
- "11434:11434"
volumes:
- ollama_data:/root/.ollama
deploy:
resources:
reservations:
devices:
- driver: nvidia
count: 1
capabilities: [gpu]
rag-api:
build: .
ports:
- "8000:8000"
environment:
- QDRANT_URL=http://qdrant:6333
- OLLAMA_URL=http://ollama:11434
- EMBEDDING_MODEL=bge-large-en-v1.5
- GENERATION_MODEL=llama3.1:8b
depends_on:
- qdrant
- ollama
volumes:
- ./data:/app/data
volumes:
qdrant_data:
ollama_data:
Kubernetes Deployment
# k8s/rag-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: rag-api
spec:
replicas: 3
selector:
matchLabels:
app: rag-api
template:
metadata:
labels:
app: rag-api
spec:
containers:
- name: rag-api
image: your-registry/rag-api:v1.0
ports:
- containerPort: 8000
env:
- name: QDRANT_URL
value: "http://qdrant:6333"
- name: OPENAI_API_KEY
valueFrom:
secretKeyRef:
name: rag-secrets
key: openai-key
resources:
requests:
memory: "2Gi"
cpu: "1000m"
limits:
memory: "4Gi"
cpu: "2000m"
livenessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 30
periodSeconds: 10
readinessProbe:
httpGet:
path: /ready
port: 8000
initialDelaySeconds: 10
periodSeconds: 5
---
apiVersion: v1
kind: Service
metadata:
name: rag-api
spec:
selector:
app: rag-api
ports:
- port: 80
targetPort: 8000
type: ClusterIP
Cost Optimization
| Component | Optimization | Savings |
|---|---|---|
| Embeddings | Use local models (BGE, E5) | 90%+ vs OpenAI |
| Vector DB | Self-hosted Qdrant/Milvus | 80% vs Pinecone |
| LLM | Route simple queries to small models | 50-80% |
| Retrieval | Cache frequent queries (Redis) | 30-60% |
| Chunking | Optimal chunk size (512 tokens) | Better quality, less tokens |
# Smart model routing
def route_query(query: str, complexity: str) -> str:
"""Route to appropriate model based on query complexity."""
if complexity == "simple": # Factual, short answer
return "llama3.1:8b" # Local, free
elif complexity == "medium": # Analysis, synthesis
return "gpt-4o-mini" # Cheap cloud
else: # Complex reasoning
return "gpt-4o" # Best quality
def estimate_complexity(query: str) -> str:
"""Heuristic complexity estimation."""
word_count = len(query.split())
has_multiple_questions = query.count("?") > 1
needs_reasoning = any(w in query.lower() for w in
["why", "how", "compare", "analyze", "explain", "pros", "cons"])
if word_count < 10 and not needs_reasoning:
return "simple"
elif needs_reasoning or has_multiple_questions or word_count > 30:
return "complex"
return "medium"
Complete Production Checklist
Ingestion
- [ ] Semantic chunking with header preservation
- [ ] Metadata enrichment (source, date, department, confidentiality)
- [ ] Deduplication (exact + near-duplicate)
- [ ] Incremental updates (upsert by chunk hash)
- [ ] Dead letter queue for failed documents
Retrieval
- [ ] Hybrid search (dense + sparse)
- [ ] Cross-encoder reranking
- [ ] MMR for diversity
- [ ] Query rewriting/expansion
- [ ] Metadata filtering (ACL, date range, department)
- [ ] Retrieval latency < 200ms p95
Generation
- [ ] Structured output with citations
- [ ] Input/output guardrails (PII, injection, hallucination)
- [ ] Confidence scoring
- [ ] Fallback for low-confidence answers
- [ ] Generation latency < 2s p95
Evaluation
- [ ] Golden test set (100+ cases)
- [ ] Retrieval metrics (Precision@K, Recall@K, MRR)
- [ ] Generation metrics (Accuracy, Faithfulness, Citation quality)
- [ ] Regression testing in CI/CD
- [ ] A/B testing framework
Monitoring
- [ ] Latency percentiles (p50, p95, p99)
- [ ] Error rates by type
- [ ] Confidence distribution
- [ ] User feedback collection
- [ ] Automated alerts
- [ ] Cost tracking per query
Operations
- [ ] Blue-green deployments
- [ ] Rollback capability
- [ ] Vector index rebuild procedure
- [ ] Embedding model versioning
- [ ] Data retention policy
- [ ] Backup/restore tested
Resources
- Qdrant Documentation
- LangChain RAG Guide
- RAGAS Evaluation
- BGE Embeddings
- Cross-Encoder Models
- Production RAG Patterns
Build RAG right the first time. Retrofitting observability and guardrails into a working prototype is 10x harder than including them from day one.
Top comments (0)