Production RAG with LangChain & LangGraph: End-to-End Implementation
Build a production-ready RAG system with document ingestion, hybrid retrieval, reranking, citations, guardrails, and monitoring.
Introduction
RAG (Retrieval-Augmented Generation) is the standard for grounding LLMs in proprietary data. This guide builds a production system — not a prototype.
Production RAG Requirements
| Requirement | Prototype | Production |
|---|---|---|
| Ingestion | Manual, one-off | Automated, incremental, monitored |
| Retrieval | Vector only | Hybrid (dense + sparse + rerank) |
| Generation | Basic prompt | Structured output + citations + guardrails |
| Evaluation | Manual | Automated metrics + LLM judge + A/B |
| Monitoring | None | Latency, quality, cost, drift alerts |
| Deployment | Local notebook | Containerized, scalable, resilient |
System Architecture
┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ Ingest │────▶│ Embedding │────▶│ Retrieval │────▶│ Generation │
│ Pipeline │ │ & Index │ │ Pipeline │ │ + Guardrails
└─────────────┘ └─────────────┘ └─────────────┘ └─────────────┘
│ │ │ │
▼ ▼ ▼ ▼
Chunking Vector DB Hybrid Search Structured Output
Metadata (Qdrant) (Dense+Sparse) Citations
Deduplication Quantization Reranking Guardrails
Incremental Replication Filtering Evaluation
Part 1: Document Ingestion Pipeline
# ingestion/pipeline.py
from dataclasses import dataclass
from typing import List, Optional
from pathlib import Path
import hashlib
import asyncio
from datetime import datetime
@dataclass
class Document:
id: str
content: str
metadata: dict
source: str
created_at: str
chunk_hash: str = ""
@dataclass
class Chunk:
id: str
document_id: str
content: str
metadata: dict
token_count: int
embedding: List[float] = None
class IngestionPipeline:
def __init__(
self,
chunker,
embedder,
vector_store,
doc_store,
deduplicator
):
self.chunker = chunker
self.embedder = embedder
self.vector_store = vector_store
self.doc_store = doc_store
self.deduplicator = deduplicator
async def ingest_file(self, file_path: Path, metadata: dict = None) -> dict:
"""Ingest a single file."""
# 1. Load document
doc = await self._load_document(file_path, metadata)
# 2. Check for duplicates
if await self.deduplicator.is_duplicate(doc.chunk_hash):
return {"status": "duplicate", "document_id": doc.id}
# 3. Chunk
chunks = self.chunker.chunk(doc)
# 4. Embed
texts = [c.content for c in chunks]
embeddings = await self.embedder.embed_batch(texts)
for chunk, emb in zip(chunks, embeddings):
chunk.embedding = emb
# 5. Store
await self.doc_store.save_document(doc)
await self.vector_store.upsert_chunks(chunks)
await self.deduplicator.mark_seen(doc.chunk_hash)
return {"status": "success", "document_id": doc.id, "chunks": len(chunks)}
async def ingest_batch(self, file_paths: List[Path]) -> dict:
"""Parallel batch ingestion."""
semaphore = asyncio.Semaphore(10)
async def limited_ingest(path):
async with semaphore:
return await self.ingest_file(path)
results = await asyncio.gather(*[limited_ingest(p) for p in file_paths])
return {
"total": len(results),
"successful": sum(1 for r in results if r["status"] == "success"),
"duplicates": sum(1 for r in results if r["status"] == "duplicate"),
"details": results
}
async def _load_document(self, path: Path, metadata: dict) -> Document:
# Implement based on file type (PDF, MD, HTML, etc.)
content = path.read_text(encoding="utf-8")
chunk_hash = hashlib.sha256(content.encode()).hexdigest()[:16]
return Document(
id=hashlib.md5(str(path).encode()).hexdigest()[:12],
content=content,
metadata=metadata or {},
source=str(path),
created_at=datetime.utcnow().isoformat(),
chunk_hash=chunk_hash
)
Semantic Chunking
# ingestion/chunker.py
from dataclasses import dataclass
import re
from typing import List
@dataclass
class Chunk:
content: str
metadata: dict
token_count: int
class SemanticChunker:
"""Chunk by semantic boundaries (headers, paragraphs)."""
def __init__(self, max_tokens: int = 512, overlap: int = 50):
self.max_tokens = max_tokens
self.overlap = overlap
def chunk(self, doc: Document) -> List[Chunk]:
# Split by markdown headers
sections = self._split_by_headers(doc.content)
all_chunks = []
for section_content, section_meta in sections:
section_chunks = self._chunk_section(
section_content,
{**doc.metadata, **section_meta}
)
all_chunks.extend(section_chunks)
return all_chunks
def _split_by_headers(self, text: str) -> List[tuple[str, dict]]:
pattern = r'^(#{1,6})\s+(.+)$'
lines = text.split('\n')
sections = []
current_header = {}
current_content = []
for line in lines:
match = re.match(pattern, line)
if match:
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}))
level = len(match.group(1))
header_text = match.group(2)
current_header[f"h{level}"] = header_text
for l in range(level + 1, 7):
current_header[f"h{l}"] = ""
current_content = [line]
else:
current_content.append(line)
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, content: str, metadata: dict) -> List[Chunk]:
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:
chunk_text = "\n\n".join(current_chunk)
chunks.append(Chunk(chunk_text, metadata, current_tokens))
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
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
def _get_overlap(self, chunks: List[str]) -> str:
if not chunks:
return ""
last = chunks[-1]
words = last.split()
overlap_words = words[-(self.overlap // 4):]
return " ".join(overlap_words)
Part 2: Hybrid Retrieval Pipeline
# retrieval/pipeline.py
from dataclasses import dataclass
from typing import List, Optional, Callable
import asyncio
@dataclass
class RetrievedDoc:
content: str
metadata: dict
score: float
rank: int
source: str # "dense", "sparse", "hybrid"
class HybridRetrievalPipeline:
def __init__(
self,
dense_embedder: Callable,
sparse_embedder: Callable,
vector_store,
sparse_store,
reranker: Callable,
dense_weight: float = 0.7,
sparse_weight: float = 0.3
):
self.dense_embedder = dense_embedder
self.sparse_embedder = sparse_embedder
self.vector_store = vector_store
self.sparse_store = sparse_store
self.reranker = reranker
self.dense_weight = dense_weight
self.sparse_weight = sparse_weight
async def retrieve(
self,
query: str,
k: int = 5,
fetch_k: int = 20,
filters: dict = None,
use_reranker: bool = True
) -> List[RetrievedDoc]:
# Stage 1: Parallel dense + sparse retrieval
dense_task = self._dense_search(query, fetch_k, filters)
sparse_task = self._sparse_search(query, fetch_k, filters)
dense_results, sparse_results = await asyncio.gather(dense_task, sparse_task)
# Stage 2: Fuse results (RRF - Reciprocal Rank Fusion)
fused = self._reciprocal_rank_fusion(dense_results, sparse_results, k * 2)
# Stage 3: Rerank
if use_reranker and fused:
fused = await self._rerank(query, fused)
# Stage 4: Final selection
return fused[:k]
async def _dense_search(self, query: str, k: int, filters: dict) -> List[RetrievedDoc]:
query_vector = await self.dense_embedder(query)
results = await self.vector_store.search(query_vector, k, filters)
return [
RetrievedDoc(
content=r["content"],
metadata=r["metadata"],
score=r["score"] * self.dense_weight,
rank=i,
source="dense"
)
for i, r in enumerate(results)
]
async def _sparse_search(self, query: str, k: int, filters: dict) -> List[RetrievedDoc]:
sparse_vector = await self.sparse_embedder(query)
results = await self.sparse_store.search(sparse_vector, k, filters)
return [
RetrievedDoc(
content=r["content"],
metadata=r["metadata"],
score=r["score"] * self.sparse_weight,
rank=i,
source="sparse"
)
for i, r in enumerate(results)
]
def _reciprocal_rank_fusion(
self,
dense: List[RetrievedDoc],
sparse: List[RetrievedDoc],
k: int,
k_constant: int = 60
) -> List[RetrievedDoc]:
"""RRF fusion for hybrid search."""
# Combine by content hash
doc_scores = {}
for results, weight in [(dense, self.dense_weight), (sparse, self.sparse_weight)]:
for rank, doc in enumerate(results):
content_hash = hash(doc.content[:200])
if content_hash not in doc_scores:
doc_scores[content_hash] = {
"doc": doc,
"rrf_score": 0
}
# RRF score: 1 / (k + rank)
doc_scores[content_hash]["rrf_score"] += weight / (k_constant + rank + 1)
# Sort by RRF score
fused = sorted(
doc_scores.values(),
key=lambda x: x["rrf_score"],
reverse=True
)
return [
RetrievedDoc(
content=item["doc"].content,
metadata=item["doc"].metadata,
score=item["rrf_score"],
rank=i,
source="hybrid"
)
for i, item in enumerate(fused[:k])
]
async def _rerank(self, query: str, docs: List[RetrievedDoc]) -> List[RetrievedDoc]:
contents = [d.content for d in docs]
rerank_scores = await self.reranker(query, contents)
for doc, score in zip(docs, rerank_scores):
doc.score = score
doc.source = "reranked"
docs.sort(key=lambda x: x.score, reverse=True)
return docs
Part 3: Generation with Citations & Guardrails
# 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="Complete answer with inline citations [doc_id]")
citations: List[dict] = Field(description="Citations with 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)
return prompt | llm | parser
def format_context(docs: List[RetrievedDoc]) -> str:
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
# 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]:
query_lower = query.lower()
for trigger in self.refusal_triggers:
if trigger in query_lower:
return False, "I can't help with that request."
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]:
# Hallucination check: numbers in answer must 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}"
# PII leakage check
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:
sanitized = answer
for pattern, _ in self.pii_patterns:
sanitized = re.sub(pattern, "[REDACTED]", sanitized)
return sanitized
Part 4: LangGraph Orchestration
# orchestration/rag_graph.py
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.sqlite import SqliteSaver
from typing import TypedDict, Annotated, List
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage
class RAGState(TypedDict):
messages: Annotated[List[BaseMessage], add_messages]
question: str
retrieved_docs: List[RetrievedDoc]
answer: CitedAnswer
guardrail_passed: bool
retry_count: int
async def retrieve_node(state: RAGState):
pipeline = get_retrieval_pipeline()
docs = await pipeline.retrieve(state["question"])
return {"retrieved_docs": docs}
async def generate_node(state: RAGState):
chain = get_rag_chain()
context = format_context(state["retrieved_docs"])
answer = await chain.ainvoke({"context": context, "question": state["question"]})
return {"answer": answer}
async def guardrail_node(state: RAGState):
guardrails = get_guardrails()
context = format_context(state["retrieved_docs"])
passed, reason = guardrails.check_output(state["answer"].answer, context)
if not passed and state["retry_count"] < 2:
return {
"guardrail_passed": False,
"retry_count": state["retry_count"] + 1,
"messages": [HumanMessage(content=f"Regenerate: {reason}")]
}
return {"guardrail_passed": passed}
def route_after_guardrail(state: RAGState) -> str:
if state["guardrail_passed"]:
return END
return "generate"
# Build graph
workflow = StateGraph(RAGState)
workflow.add_node("retrieve", retrieve_node)
workflow.add_node("generate", generate_node)
workflow.add_node("guardrail", guardrail_node)
workflow.add_edge(START, "retrieve")
workflow.add_edge("retrieve", "generate")
workflow.add_edge("generate", "guardrail")
workflow.add_conditional_edges("guardrail", route_after_guardrail, {
"generate": "generate",
END: END
})
rag_app = workflow.compile(checkpointer=SqliteSaver.from_conn_string("sqlite:///rag.db"))
Part 5: FastAPI Production Server
# server.py
from fastapi import FastAPI, HTTPException, BackgroundTasks
from pydantic import BaseModel
from typing import Optional
import uvicorn
app = FastAPI(title="Production RAG API", version="1.0.0")
class QueryRequest(BaseModel):
question: str
filters: Optional[dict] = None
k: int = 5
use_reranker: bool = True
stream: bool = False
class QueryResponse(BaseModel):
answer: str
citations: List[dict]
confidence: float
retrieved_count: int
latency_ms: float
@app.post("/query", response_model=QueryResponse)
async def query(request: QueryRequest):
start = time.time()
# Input guardrails
guardrails = get_guardrails()
passed, reason = guardrails.check_input(request.question)
if not passed:
raise HTTPException(400, reason)
# Execute RAG
config = {"configurable": {"thread_id": f"query-{uuid.uuid4()}"}}
result = await rag_app.ainvoke({
"question": request.question,
"retry_count": 0
}, config)
latency = (time.time() - start) * 1000
# Log for monitoring
background_tasks.add_task(log_metrics, {
"question": request.question,
"latency_ms": latency,
"retrieved": len(result["retrieved_docs"]),
"confidence": result["answer"].confidence,
"guardrail_passed": result["guardrail_passed"]
})
return QueryResponse(
answer=result["answer"].answer,
citations=result["answer"].citations,
confidence=result["answer"].confidence,
retrieved_count=len(result["retrieved_docs"]),
latency_ms=latency
)
@app.post("/ingest")
async def ingest(request: IngestRequest, background_tasks: BackgroundTasks):
background_tasks.add_task(run_ingestion, request.paths)
return {"status": "ingestion_started"}
@app.get("/health")
async def health():
return {"status": "healthy", "timestamp": datetime.utcnow().isoformat()}
@app.get("/metrics")
async def metrics():
return get_system_metrics()
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000)
Part 6: Evaluation & Monitoring
# evaluation/rag_eval.py
from dataclasses import dataclass
from typing import List
import json
@dataclass
class RAGTestCase:
question: str
expected_answer: str
relevant_doc_ids: List[str]
metadata_filters: dict = None
class RAGEvaluator:
def __init__(self, rag_system):
self.rag = rag_system
def evaluate_retrieval(self, test_cases: List[RAGTestCase]) -> dict:
metrics = {"precision@k": [], "recall@k": [], "mrr": []}
for tc in test_cases:
results = await self.rag.retrieve(tc.question, k=10)
retrieved_ids = [r.metadata.get("chunk_hash") for r in results]
relevant = set(tc.relevant_doc_ids)
for k in [1, 3, 5, 10]:
retrieved_k = set(retrieved_ids[:k])
metrics[f"precision@{k}"].append(len(retrieved_k & relevant) / k)
metrics[f"recall@{k}"].append(len(retrieved_k & relevant) / len(relevant))
for i, rid in enumerate(retrieved_ids):
if rid in relevant:
metrics["mrr"].append(1 / (i + 1))
break
else:
metrics["mrr"].append(0)
return {k: sum(v) / len(v) for k, v in metrics.items()}
async def evaluate_generation(self, test_cases: List[RAGTestCase]) -> dict:
scores = {"accuracy": [], "completeness": [], "citation_quality": []}
for tc in test_cases:
result = await self.rag.query(tc.question)
eval_result = await llm_judge(
task="Answer question using provided context",
input_text=tc.question,
response=result.answer,
criteria=["accuracy", "completeness", "citation_quality", "faithfulness"]
)
for k in scores:
scores[k].append(eval_result.scores.get(k, 0))
return {k: sum(v) / len(v) for k, v in scores.items()}
Production Monitoring
# monitoring/production_monitor.py
from dataclasses import dataclass, asdict
from datetime import datetime
import json
@dataclass
class RAGMetrics:
timestamp: str
question_hash: 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
error: Optional[str] = None
class ProductionMonitor:
def __init__(self, log_path: str = "logs/rag_metrics.jsonl"):
self.log_path = log_path
self.buffer = []
def log(self, metrics: RAGMetrics):
self.buffer.append(asdict(metrics))
if len(self.buffer) >= 100:
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:
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
)
}
Part 7: Deployment
Docker
# Dockerfile
FROM python:3.11-slim
WORKDIR /app
# System dependencies
RUN apt-get update && apt-get install -y \
gcc \
libpq-dev \
&& rm -rf /var/lib/apt/lists/*
# Python dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# Application
COPY . .
# Health check
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
CMD curl -f http://localhost:8000/health || exit 1
CMD ["uvicorn", "server:app", "--host", "0.0.0.0", "--port", "8000"]
Kubernetes
# k8s/rag-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: rag-api
spec:
replicas: 3
selector:
matchLabels:
app: rag-api
template:
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
- name: REDIS_URL
value: "redis://redis:6379"
resources:
requests:
memory: "2Gi"
cpu: "1000m"
limits:
memory: "4Gi"
cpu: "2000m"
livenessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 30
readinessProbe:
httpGet:
path: /ready
port: 8000
initialDelaySeconds: 10
---
apiVersion: v1
kind: Service
metadata:
name: rag-api
spec:
selector:
app: rag-api
ports:
- port: 80
targetPort: 8000
Complete Requirements
# requirements.txt
langchain==0.2.0
langgraph==0.1.0
langchain-openai==0.1.0
qdrant-client==1.10.0
sentence-transformers==3.0.0
fastapi==0.110.0
uvicorn==0.29.0
pydantic==2.7.0
pydantic-settings==2.3.0
python-multipart==0.0.9
redis==5.0.0
psycopg2-binary==2.9.9
pytest==8.2.0
pytest-asyncio==0.23.0
Production Checklist
Ingestion
- [ ] Semantic chunking with header preservation
- [ ] Metadata enrichment (source, date, department, confidentiality)
- [ ] Deduplication (exact + near-duplicate via MinHash)
- [ ] Incremental updates (upsert by chunk hash)
- [ ] Dead letter queue for failed documents
- [ ] Progress tracking & resumability
Retrieval
- [ ] Hybrid search (dense + sparse/BM25)
- [ ] Cross-encoder reranking (ms-marco-MiniLM-L-6-v2)
- [ ] MMR for diversity
- [ ] Query rewriting/expansion
- [ ] Metadata filtering (ACL, date range, department)
- [ ] Retrieval latency < 200ms p95
Generation
- [ ] Structured output with citations (JSON schema)
- [ ] 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 (explicit + implicit)
- [ ] Automated alerts (quality drop, latency spike, cost surge)
- [ ] Cost tracking per query
Operations
- [ ] Blue-green deployments
- [ ] Rollback capability (< 5 min)
- [ ] Vector index rebuild procedure
- [ ] Embedding model versioning
- [ ] Data retention policy
- [ ] Backup/restore tested monthly
Resources
- LangGraph Documentation
- Qdrant Documentation
- RAGAS Evaluation
- BGE Embeddings
- Cross-Encoder Models
- Production RAG Patterns
Production RAG is 10% retrieval, 90% engineering. Build observability, guardrails, and evaluation in from day one — retrofitting is 10x harder.
Top comments (0)