QuivrHQ

PDFをAIが読める形に変換するのに何時間もかけていないか。 答えはMegaParseだった。 Word・PPTも含め、情報損失ゼロで一発構造化できる。 QuivrHQ/MegaParse teikoku-signal.netlify.app

Production RAG with LangChain & LangGraph: Retrieval, Reranking & Agentic Workflows 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. --- 📦 Want the production-ready version? This article covers the fundamentals. The RAG Pipeline Patterns pack goes further — production chunking, reranking, evaluation and observability patterns, plus full source code and deployment configs. → Get RAG Pipeline Patterns Extracted from real production systems, not toy examples.

Production RAG with LangChain & LangGraph: End-to-End Implementation 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. --- 📦 Want the production-ready version? This article covers the fundamentals. The RAG Pipeline Patterns pack goes further — production chunking, reranking, evaluation and observability patterns, plus full source code and deployment configs. → Get RAG Pipeline Patterns Extracted from real production systems, not toy examples.

Building a Production-Ready RAG System: Architecture, Optimization & Monitoring 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. --- 📦 Want the production-ready version? This article covers the fundamentals. The RAG Pipeline Patterns pack goes further — production chunking, reranking, evaluation and observability patterns, plus full source code and deployment configs. → Get RAG Pipeline Patterns Extracted from real production systems, not toy examples.