DEV Community: Phil IPS

Production RAG with LangChain & LangGraph: Retrieval, Reranking & Agentic Workflows

Phil IPS — Mon, 12 Oct 2026 21:49:29 +0000

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

Production RAG is 10% retrieval, 90% engineering. Build observability, guardrails, and evaluation in from day one — retrofitting is 10x harder.

Fine-Tuning LLMs with LoRA & QLoRA: Complete Production Guide

Phil IPS — Mon, 12 Oct 2026 21:48:24 +0000

Fine-Tuning LLMs with LoRA & QLoRA: Complete Practical Guide

Master parameter-efficient fine-tuning to adapt any open-source LLM to your specific domain, style, or task.

Introduction

Full fine-tuning requires massive compute. LoRA (Low-Rank Adaptation) and QLoRA (Quantized LoRA) change the game — fine-tune 7B-70B models on consumer GPUs.

Why LoRA/QLoRA?

Method	Trainable Params	VRAM (7B)	Quality	Hardware
Full FT	100% (7B)	~48 GB	Best	A100/H100
LoRA (r=64)	~0.5% (35M)	~16 GB	Excellent	RTX 3090/4090
QLoRA (r=64)	~0.5% (35M)	~8 GB	Excellent	RTX 3060 12GB+
LoRA (r=16)	~0.1% (7M)	~12 GB	Good	RTX 3060 12GB
QLoRA (r=16)	~0.1% (7M)	~6 GB	Good	RTX 3060 8GB

QLoRA = LoRA + 4-bit quantization + paged optimizers + gradient checkpointing

Prerequisites

# Core dependencies
pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu121
pip install transformers accelerate peft bitsandbytes trl datasets

# For quantization
pip install auto-gptq optimum

# Optional: Weights & Biases for logging
pip install wandb

Understanding LoRA

The Math

Instead of updating all weights $W \in \mathbb{R}^{d \times d}$, LoRA learns two small matrices:

$$
W' = W + \Delta W = W + BA
$$

Where:

$B \in \mathbb{R}^{d \times r}$ (down-projection)
$A \in \mathbb{R}^{r \times d}$ (up-projection)
$r \ll d$ (rank, typically 8-64)

Only $A$ and $B$ are trained. $W$ stays frozen.

Target Modules

Which layers to adapt? Research shows:

# For LLaMA/Mistral/Gemma architecture
TARGET_MODULES = [
    "q_proj", "k_proj", "v_proj", "o_proj",  # Attention
    "gate_proj", "up_proj", "down_proj"       # MLP
]

# For smaller models or less VRAM, use attention only
TARGET_MODULES_ATTENTION = ["q_proj", "k_proj", "v_proj", "o_proj"]

Data Preparation

Format: Instruction Tuning (Alpaca/ShareGPT)

// alpaca_format.json
[
  {
    "instruction": "Write a Python function to calculate fibonacci numbers",
    "input": "",
    "output": "def fibonacci(n):\n    if n <= 1:\n        return n\n    return fibonacci(n-1) + fibonacci(n-2)"
  },
  {
    "instruction": "Explain the concept of recursion",
    "input": "In the context of computer science",
    "output": "Recursion is a programming technique where a function calls itself..."
  }
]

Format: Conversation (ChatML)

// chatml_format.json
[
  {
    "messages": [
      {"role": "system", "content": "You are a helpful coding assistant."},
      {"role": "user", "content": "How do I handle errors in Go?"},
      {"role": "assistant", "content": "In Go, errors are values..."}
    ]
  }
]

Data Processing Pipeline

# data_prep.py
from datasets import Dataset, load_dataset
from transformers import AutoTokenizer
import json

def load_and_format_data(file_path: str, format: str = "alpaca") -> Dataset:
    with open(file_path) as f:
        data = json.load(f)

    if format == "alpaca":
        def format_alpaca(example):
            if example["input"]:
                prompt = f"### Instruction:\n{example['instruction']}\n\n### Input:\n{example['input']}\n\n### Response:\n{example['output']}"
            else:
                prompt = f"### Instruction:\n{example['instruction']}\n\n### Response:\n{example['output']}"
            return {"text": prompt}

        dataset = Dataset.from_list(data).map(format_alpaca)

    elif format == "chatml":
        def format_chatml(example):
            messages = example["messages"]
            # Apply chat template
            text = tokenizer.apply_chat_template(messages, tokenize=False)
            return {"text": text}

        dataset = Dataset.from_list(data).map(format_chatml)

    return dataset

def tokenize_dataset(dataset: Dataset, tokenizer, max_length: int = 2048):
    def tokenize(example):
        return tokenizer(
            example["text"],
            truncation=True,
            max_length=max_length,
            padding="max_length",
            return_tensors="pt"
        )

    return dataset.map(tokenize, batched=True, remove_columns=dataset.column_names)

# Train/val split
dataset = load_and_format_data("train.json")
dataset = dataset.train_test_split(test_size=0.1, seed=42)
train_dataset = dataset["train"]
eval_dataset = dataset["test"]

LoRA Fine-Tuning with Hugging Face TRL

Complete Training Script

# train_lora.py
import torch
from transformers import (
    AutoModelForCausalLM,
    AutoTokenizer,
    TrainingArguments,
    BitsAndBytesConfig
)
from peft import LoraConfig, get_peft_model, prepare_model_for_kbit_training
from trl import SFTTrainer
from datasets import load_dataset
import os

# ============ CONFIGURATION ============
MODEL_ID = "meta-llama/Llama-3.2-3B-Instruct"  # Or mistralai/Mistral-7B-Instruct-v0.3
DATASET_PATH = "train.json"
OUTPUT_DIR = "./llama-3.2-3b-lora-finetuned"
MAX_LENGTH = 2048
BATCH_SIZE = 2
GRAD_ACCUM = 8  # Effective batch = 16
LEARNING_RATE = 2e-4
NUM_EPOCHS = 3
LORA_R = 64
LORA_ALPHA = 16
LORA_DROPOUT = 0.1

# ============ QUANTIZATION CONFIG (QLoRA) ============
bnb_config = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_quant_type="nf4",        # NormalFloat 4-bit
    bnb_4bit_compute_dtype=torch.bfloat16,
    bnb_4bit_use_double_quant=True,   # Double quantization
)

# ============ LoRA CONFIG ============
lora_config = LoraConfig(
    r=LORA_R,
    lora_alpha=LORA_ALPHA,
    target_modules=[
        "q_proj", "k_proj", "v_proj", "o_proj",
        "gate_proj", "up_proj", "down_proj"
    ],
    lora_dropout=LORA_DROPOUT,
    bias="none",
    task_type="CAUSAL_LM",
    # For QLoRA: modules_to_save=["embed_tokens", "lm_head"]  # If extending vocab
)

# ============ LOAD MODEL & TOKENIZER ============
tokenizer = AutoTokenizer.from_pretrained(MODEL_ID, trust_remote_code=True)
tokenizer.pad_token = tokenizer.eos_token
tokenizer.padding_side = "right"

model = AutoModelForCausalLM.from_pretrained(
    MODEL_ID,
    quantization_config=bnb_config,
    device_map="auto",
    trust_remote_code=True,
    torch_dtype=torch.bfloat16,
    attn_implementation="flash_attention_2",  # Requires flash-attn
)

# Prepare for k-bit training
model = prepare_model_for_kbit_training(model)
model = get_peft_model(model, lora_config)
model.print_trainable_parameters()
# Example output: trainable params: 35,651,584 || all params: 3,200,000,000 || trainable%: 1.11%

# ============ LOAD DATA ============
def formatting_func(example):
    return example["text"]

dataset = load_dataset("json", data_files=DATASET_PATH, split="train")
dataset = dataset.train_test_split(test_size=0.1, seed=42)

# ============ TRAINING ARGS ============
training_args = TrainingArguments(
    output_dir=OUTPUT_DIR,
    num_train_epochs=NUM_EPOCHS,
    per_device_train_batch_size=BATCH_SIZE,
    per_device_eval_batch_size=BATCH_SIZE,
    gradient_accumulation_steps=GRAD_ACCUM,
    learning_rate=LEARNING_RATE,
    lr_scheduler_type="cosine",
    warmup_ratio=0.03,
    logging_steps=10,
    eval_strategy="steps",
    eval_steps=100,
    save_strategy="steps",
    save_steps=100,
    save_total_limit=3,
    load_best_model_at_end=True,
    metric_for_best_model="eval_loss",
    bf16=True,
    tf32=True,
    gradient_checkpointing=True,
    gradient_checkpointing_kwargs={"use_reentrant": False},
    optim="paged_adamw_32bit",  # QLoRA optimizer
    report_to="wandb",  # or "tensorboard" or "none"
    run_name=f"llama-3.2-3b-lora-r{LORA_R}",
    seed=42,
    ddp_find_unused_parameters=False,
)

# ============ TRAINER ============
trainer = SFTTrainer(
    model=model,
    args=training_args,
    train_dataset=dataset["train"],
    eval_dataset=dataset["test"],
    formatting_func=formatting_func,
    max_seq_length=MAX_LENGTH,
    tokenizer=tokenizer,
    packing=True,  # Efficient packing of sequences
)

# ============ TRAIN ============
trainer.train()

# Save final model
trainer.save_model()
tokenizer.save_pretrained(OUTPUT_DIR)

print(f"Model saved to {OUTPUT_DIR}")

Launch Training

# Single GPU
python train_lora.py

# Multi-GPU with Accelerate
accelerate config  # First time setup
accelerate launch train_lora.py

# With DeepSpeed ZeRO-3 (for larger models)
accelerate launch --config_file ds_zero3.yaml train_lora.py

QLoRA Specific Optimizations

Paged Optimizers

# Automatically used with optim="paged_adamw_32bit"
# Moves optimizer states to CPU when not in use
# Enables larger models on same VRAM

Gradient Checkpointing

# In TrainingArguments:
gradient_checkpointing=True,
gradient_checkpointing_kwargs={"use_reentrant": False},

# Saves ~30% VRAM at cost of ~20% slower training

Flash Attention 2

# Install
pip install flash-attn --no-build-isolation

# Use in model loading
attn_implementation="flash_attention_2"

# 2-3x faster, less memory

Monitoring Training

Weights & Biases

# In TrainingArguments
report_to="wandb",
run_name="my-lora-run",

# Before training
import wandb
wandb.init(project="llm-finetuning", name="llama-3b-lora-r64")

TensorBoard

# Start tensorboard
tensorboard --logdir ./llama-3.2-3b-lora-finetuned/runs

# In TrainingArguments
report_to="tensorboard",

Key Metrics to Watch

Metric	Healthy Range	Warning Signs
train_loss	Decreasing steadily	Plateauing early, diverging
eval_loss	Decreasing, close to train	Increasing (overfitting)
learning_rate	Cosine decay to 0	Stuck, not decaying
grad_norm	< 1.0	> 10 (exploding gradients)
GPU memory	Stable	Growing (memory leak)

Merging & Exporting

Merge LoRA Weights

# merge_lora.py
from peft import PeftModel
from transformers import AutoModelForCausalLM, AutoTokenizer
import torch

BASE_MODEL = "meta-llama/Llama-3.2-3B-Instruct"
LORA_PATH = "./llama-3.2-3b-lora-finetuned"
OUTPUT_PATH = "./llama-3.2-3b-merged"

# Load base model in fp16/bf16 (not 4-bit)
base_model = AutoModelForCausalLM.from_pretrained(
    BASE_MODEL,
    torch_dtype=torch.bfloat16,
    device_map="auto",
    trust_remote_code=True,
)

# Load LoRA adapter
model = PeftModel.from_pretrained(base_model, LORA_PATH)

# Merge and unload
merged_model = model.merge_and_unload()

# Save merged model
merged_model.save_pretrained(OUTPUT_PATH)
tokenizer = AutoTokenizer.from_pretrained(BASE_MODEL)
tokenizer.save_pretrained(OUTPUT_PATH)

print(f"Merged model saved to {OUTPUT_PATH}")

Export to GGUF (for llama.cpp/Ollama)

# Install llama.cpp with Python bindings
pip install llama-cpp-python

# Convert using llama.cpp convert script
python llama.cpp/convert_hf_to_gguf.py \
  ./llama-3.2-3b-merged \
  --outfile ./llama-3.2-3b-finetuned.Q4_K_M.gguf \
  --outtype q4_k_m

# Or use llama.cpp quantize directly
./llama.cpp/quantize ./llama-3.2-3b-merged/model.fp16.gguf \
  ./llama-3.2-3b-finetuned.Q4_K_M.gguf q4_k_m

Create Ollama Modelfile

# Modelfile
FROM ./llama-3.2-3b-finetuned.Q4_K_M.gguf

TEMPLATE "{{ .System }}\n{{ .Prompt }}\n{{ .Response }}"

PARAMETER temperature 0.7
PARAMETER top_p 0.9
PARAMETER num_ctx 8192

# Create Ollama model
ollama create my-finetuned-model -f Modelfile

# Test
ollama run my-finetuned-model "Write a Python function for binary search"

Advanced: LoRA Hyperparameter Tuning

Rank (r) Selection

# Experiment with different ranks
RANK_CONFIGS = [
    {"r": 8, "alpha": 16},    # Minimal, fast, less capacity
    {"r": 16, "alpha": 32},   # Good balance
    {"r": 32, "alpha": 64},   # More capacity
    {"r": 64, "alpha": 16},   # High capacity (alpha often = r/4 to r/2)
    {"r": 128, "alpha": 32},  # Maximum capacity
]

Target Module Ablation

# Test different target module sets
TARGET_CONFIGS = {
    "attention_only": ["q_proj", "k_proj", "v_proj", "o_proj"],
    "mlp_only": ["gate_proj", "up_proj", "down_proj"],
    "all_linear": ["q_proj", "k_proj", "v_proj", "o_proj", 
                   "gate_proj", "up_proj", "down_proj"],
    "attention_mlp": ["q_proj", "v_proj", "gate_proj", "up_proj"],  # Key ones
}

Learning Rate Schedule

# For LoRA, higher LR often works (2e-4 to 1e-3)
# But depends on dataset size:
# Small dataset (<1k): 1e-4 to 5e-4
# Medium (1k-10k): 2e-4
# Large (>10k): 1e-4 to 2e-4

Evaluation & Testing

Quantitative Evaluation

# eval_model.py
from transformers import AutoModelForCausalLM, AutoTokenizer
from peft import PeftModel
import torch
from datasets import load_dataset
import json

def evaluate_model(model_path, test_data_path, base_model_id=None):
    if base_model_id:
        base = AutoModelForCausalLM.from_pretrained(
            base_model_id, torch_dtype=torch.bfloat16, device_map="auto"
        )
        model = PeftModel.from_pretrained(base, model_path)
    else:
        model = AutoModelForCausalLM.from_pretrained(
            model_path, torch_dtype=torch.bfloat16, device_map="auto"
        )

    tokenizer = AutoTokenizer.from_pretrained(model_path)

    # Load test data
    with open(test_data_path) as f:
        test_data = json.load(f)

    results = []
    for item in test_data[:50]:  # Sample 50
        prompt = f"### Instruction:\n{item['instruction']}\n\n### Response:\n"
        inputs = tokenizer(prompt, return_tensors="pt").to("cuda")

        with torch.no_grad():
            outputs = model.generate(
                **inputs,
                max_new_tokens=256,
                temperature=0.7,
                do_sample=True,
                pad_token_id=tokenizer.eos_token_id
            )

        response = tokenizer.decode(outputs[0][inputs.input_ids.shape[1]:], skip_special_tokens=True)
        results.append({
            "instruction": item["instruction"],
            "expected": item["output"],
            "generated": response
        })

    # Save for human evaluation
    with open("eval_results.json", "w") as f:
        json.dump(results, f, indent=2)

    return results

Qualitative Evaluation (Human/LLM Judge)

# judge_eval.py
from openai import OpenAI
import json

client = OpenAI()

def llm_judge(instruction, expected, generated):
    prompt = f"""Rate the quality of the generated response compared to expected.

Instruction: {instruction}
Expected: {expected}
Generated: {generated}

Rate 1-5 on:
1. Accuracy
2. Completeness  
3. Clarity
4. Relevance

Return JSON: {{"accuracy": 1-5, "completeness": 1-5, "clarity": 1-5, "relevance": 1-5, "reasoning": "..."}}"""

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0
    )
    return json.loads(response.content)

# Run evaluation
with open("eval_results.json") as f:
    results = json.load(f)

scores = {"accuracy": [], "completeness": [], "clarity": [], "relevance": []}
for r in results:
    eval = llm_judge(r["instruction"], r["expected"], r["generated"])
    for k in scores:
        scores[k].append(eval[k])

for k, v in scores.items():
    print(f"{k}: {sum(v)/len(v):.2f}")

Common Issues & Solutions

Issue	Cause	Solution
OOM	Model too large	Increase gradient_accumulation, reduce batch_size, use QLoRA, gradient_checkpointing
Loss not decreasing	LR too low/high	Try 2e-4, check data format, verify tokenization
Overfitting	Too many epochs, small data	Reduce epochs, add dropout, more data, weight decay
Gibberish output	Wrong chat template	Match training format exactly
NaN loss	Unstable training	Reduce LR, add gradient clipping, check data
Slow training	No Flash Attention	Install flash-attn, use bf16

Cost Analysis

Training Cost (Estimates)

Model	Method	GPU	Time (1k samples, 3 epochs)	Cost (Cloud)
3B	LoRA	RTX 3090 (24GB)	~30 min	$0.50
3B	QLoRA	RTX 3060 (12GB)	~45 min	$0.30
7B	LoRA	RTX 3090 (24GB)	~1 hr	$1.00
7B	QLoRA	RTX 3090 (24GB)	~1.5 hr	$0.80
13B	QLoRA	2× RTX 3090	~3 hr	$3.00
70B	QLoRA	4× A100 (80GB)	~8 hr	$40+

Local training on owned hardware = $0 marginal cost

Best Practices Summary

Data

[ ] Quality > Quantity: 1k high-quality > 100k noisy
[ ] Deduplicate: Remove exact/near duplicates
[ ] Balance: Cover all target domains evenly
[ ] Format consistency: Match training format exactly at inference

Training

[ ] Start small: Test with r=16, 1 epoch first
[ ] Monitor eval_loss: Stop when it plateaus/increases
[ ] Save checkpoints: Every 100-500 steps
[ ] Use cosine LR: With warmup (3-5% of steps)
[ ] Pack sequences: packing=True in SFTTrainer

LoRA Config

[ ] r=64, alpha=16 for most cases
[ ] Target all linear layers for best quality
[ ] Dropout 0.05-0.1 for regularization
[ ] Don't quantize LoRA weights (keep in bf16/fp16)

Deployment

[ ] Merge weights for inference speed
[ ] Quantize merged model to GGUF Q4_K_M
[ ] Test inference with same prompt format
[ ] Benchmark latency before production

Complete Example: Domain Adaptation

# Complete workflow for adapting LLaMA-3.2-3B to medical QA
# 1. Collect 2k high-quality medical QA pairs
# 2. Format as Alpaca
# 3. Train with QLoRA r=64, 3 epochs
# 4. Evaluate on held-out medical questions
# 5. Merge, quantize to GGUF, deploy via Ollama

# Expected: 90%+ accuracy on medical specialty vs 60% base model

Resources

Fine-tuning is no longer reserved for big tech. With QLoRA, a $300 GPU can customize a 7B model to beat GPT-3.5 on your specific task. Start small, iterate fast, deploy locally.

Deploying LLMs Locally: Ollama, LM Studio, llama.cpp & vLLM Compared

Phil IPS — Mon, 12 Oct 2026 21:43:37 +0000

Liquid syntax error: Unknown tag 'endraw'

Building AI Agents with LangGraph: State, Tools & Multi-Agent Workflows

Phil IPS — Mon, 12 Oct 2026 21:42:29 +0000

Building AI Agents with LangGraph: From Fundamentals to Production

A comprehensive guide to building reliable, stateful AI agents using LangGraph — the orchestration framework for agentic workflows.

Introduction

AI agents are LLMs equipped with tools, memory, and reasoning loops. LangGraph provides the orchestration layer to make them production-ready.

Why LangGraph?

Feature	LangChain Agents	LangGraph
State Management	Implicit, limited	Explicit, full control
Cycles/Loops	Difficult	Native support
Human-in-the-Loop	Basic	First-class
Debugging	Hard	Time-travel debugging
Persistence	Manual	Built-in checkpoints
Multi-Agent	Ad-hoc	Structured graphs

Core Concepts

1. State

The single source of truth for your agent:

# state.py
from typing import TypedDict, List, Annotated
from langgraph.graph import add_messages
from pydantic import BaseModel

class AgentState(TypedDict):
    # Messages with automatic reducer (appends)
    messages: Annotated[List[BaseModel], add_messages]

    # Custom fields
    user_id: str
    session_id: str
    tool_call_count: int
    current_plan: str
    retrieved_docs: List[dict]
    error_count: int

2. Nodes

Functions that process state and return updates:

# nodes.py
from langgraph.graph import END
from langchain_core.messages import AIMessage, ToolMessage
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
llm_with_tools = llm.bind_tools(tools)

async def agent_node(state: AgentState):
    """Main reasoning node."""
    response = await llm_with_tools.ainvoke(state["messages"])

    return {
        "messages": [response],
        "tool_call_count": state.get("tool_call_count", 0) + 
                          len(response.tool_calls) if response.tool_calls else 0
    }

async def tool_node(state: AgentState):
    """Execute tool calls."""
    last_message = state["messages"][-1]
    tool_outputs = []

    for tool_call in last_message.tool_calls:
        tool = get_tool_by_name(tool_call["name"])
        try:
            result = await tool.ainvoke(tool_call["args"])
            tool_outputs.append(ToolMessage(
                content=str(result),
                tool_call_id=tool_call["id"]
            ))
        except Exception as e:
            tool_outputs.append(ToolMessage(
                content=f"Error: {str(e)}",
                tool_call_id=tool_call["id"]
            ))

    return {"messages": tool_outputs}

def should_continue(state: AgentState) -> str:
    """Route: continue to tools or end."""
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tools"
    return END

3. Edges & Conditional Routing

# graph.py
from langgraph.graph import StateGraph, START

def build_agent_graph():
    workflow = StateGraph(AgentState)

    # Add nodes
    workflow.add_node("agent", agent_node)
    workflow.add_node("tools", tool_node)
    workflow.add_node("validate", validation_node)

    # Edges
    workflow.add_edge(START, "agent")
    workflow.add_conditional_edges(
        "agent",
        should_continue,
        {"tools": "tools", END: END}
    )
    workflow.add_edge("tools", "agent")

    # Compile with checkpointing
    return workflow.compile(
        checkpointer=MemorySaver(),  # Or PostgresSaver for production
        interrupt_before=["tools"]  # Human-in-the-loop
    )

Building a Production Agent

Complete Example: Research Assistant

# research_agent.py
from typing import TypedDict, List, Annotated, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.sqlite import SqliteSaver
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, ToolMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
import json
import sqlite3
from datetime import datetime

# ============ STATE ============
class ResearchState(TypedDict):
    messages: Annotated[List[BaseMessage], add_messages]
    research_topic: str
    sub_questions: List[str]
    gathered_info: List[dict]
    current_step: str
    iteration: int
    max_iterations: int
    final_report: str

# ============ TOOLS ============
@tool
def web_search(query: str, num_results: int = 5) -> str:
    """Search the web for information."""
    # Implementation using your preferred search API
    # Return formatted results
    pass

@tool
def save_research_note(topic: str, content: str, source: str) -> str:
    """Save a research finding."""
    conn = sqlite3.connect("research.db")
    conn.execute("""
        CREATE TABLE IF NOT EXISTS notes (
            id INTEGER PRIMARY KEY,
            topic TEXT, content TEXT, source TEXT, created_at TEXT
        )
    """)
    conn.execute(
        "INSERT INTO notes (topic, content, source, created_at) VALUES (?, ?, ?, ?)",
        (topic, content, source, datetime.utcnow().isoformat())
    )
    conn.commit()
    conn.close()
    return f"Saved note for '{topic}'"

tools = [web_search, save_research_note]

# ============ NODES ============
llm = ChatOpenAI(model="gpt-4o", temperature=0)
planner_llm = llm.bind_tools(tools)
researcher_llm = llm.bind_tools([web_search])
writer_llm = llm.bind_tools([save_research_note])

async def planner(state: ResearchState):
    """Plan the research approach."""
    topic = state["research_topic"]

    prompt = f"""Break down this research topic into 3-5 specific sub-questions:
    Topic: {topic}

    Return as JSON: {{"sub_questions": ["q1", "q2", ...]}}"""

    response = await planner_llm.ainvoke([HumanMessage(content=prompt)])
    plan = json.loads(response.content)

    return {
        "sub_questions": plan["sub_questions"],
        "current_step": "researching",
        "iteration": 0,
        "gathered_info": []
    }

async def researcher(state: ResearchState):
    """Research each sub-question."""
    if state["iteration"] >= len(state["sub_questions"]):
        return {"current_step": "writing"}

    question = state["sub_questions"][state["iteration"]]

    prompt = f"""Research this question thoroughly using web search:
    Question: {question}

    Search for multiple sources and synthesize findings."""

    response = await researcher_llm.ainvoke([
        HumanMessage(content=prompt)
    ])

    # Execute tool calls
    tool_results = []
    for tool_call in response.tool_calls:
        if tool_call["name"] == "web_search":
            result = await web_search.ainvoke(tool_call["args"])
            tool_results.append({
                "question": question,
                "result": result,
                "source": "web_search"
            })

    return {
        "gathered_info": state["gathered_info"] + tool_results,
        "iteration": state["iteration"] + 1
    }

async def writer(state: ResearchState):
    """Compile final report."""
    prompt = f"""Write a comprehensive research report on: {state['research_topic']}

    Sub-questions and findings:
    {json.dumps(state['gathered_info'], indent=2)}

    Structure:
    1. Executive Summary
    2. Detailed Findings (per sub-question)
    3. Key Insights
    4. Sources
    5. Limitations"""

    response = await writer_llm.ainvoke([HumanMessage(content=prompt)])

    # Save final report
    await save_research_note.ainvoke({
        "topic": state["research_topic"],
        "content": response.content,
        "source": "final_report"
    })

    return {
        "final_report": response.content,
        "current_step": "complete"
    }

def route_research(state: ResearchState) -> Literal["researcher", "writer"]:
    return state["current_step"]

# ============ GRAPH ============
def build_research_agent():
    workflow = StateGraph(ResearchState)

    workflow.add_node("planner", planner)
    workflow.add_node("researcher", researcher)
    workflow.add_node("writer", writer)

    workflow.add_edge(START, "planner")
    workflow.add_edge("planner", "researcher")
    workflow.add_conditional_edges(
        "researcher",
        route_research,
        {"researcher": "researcher", "writer": "writer"}
    )
    workflow.add_edge("writer", END)

    # Persistence
    checkpointer = SqliteSaver.from_conn_string("sqlite:///research_agent.db")

    return workflow.compile(checkpointer=checkpointer)

# ============ USAGE ============
async def run_research(topic: str, thread_id: str):
    agent = build_research_agent()

    config = {"configurable": {"thread_id": thread_id}}

    initial_state = {
        "messages": [HumanMessage(content=f"Research: {topic}")],
        "research_topic": topic,
        "max_iterations": 5
    }

    # Stream for progress
    async for chunk in agent.astream(initial_state, config):
        for node, update in chunk.items():
            print(f"[{node}] {update.get('current_step', '')}")
            if update.get("final_report"):
                print("\n=== FINAL REPORT ===")
                print(update["final_report"])

# Run
# asyncio.run(run_research("Impact of AI on software development jobs", "thread-123"))

Advanced Patterns

1. Multi-Agent Collaboration

# multi_agent.py
from langgraph.graph import StateGraph
from typing import Literal

class MultiAgentState(TypedDict):
    messages: Annotated[List[BaseMessage], add_messages]
    next_agent: str
    task: str
    results: dict

def supervisor(state: MultiAgentState) -> Literal["researcher", "coder", "reviewer", END]:
    """Route to appropriate specialist."""
    last_msg = state["messages"][-1].content.lower()

    if "research" in last_msg or "find" in last_msg:
        return "researcher"
    elif "code" in last_msg or "implement" in last_msg:
        return "coder"
    elif "review" in last_msg or "check" in last_msg:
        return "reviewer"
    return END

# Each specialist is a subgraph
researcher_graph = build_research_subgraph()
coder_graph = build_coder_subgraph()
reviewer_graph = build_reviewer_subgraph()

workflow = StateGraph(MultiAgentState)
workflow.add_node("supervisor", supervisor_node)
workflow.add_node("researcher", researcher_graph)
workflow.add_node("coder", coder_graph)
workflow.add_node("reviewer", reviewer_graph)

workflow.add_edge(START, "supervisor")
workflow.add_conditional_edges("supervisor", supervisor)
workflow.add_edge("researcher", "supervisor")
workflow.add_edge("coder", "supervisor")
workflow.add_edge("reviewer", "supervisor")

2. Human-in-the-Loop

# human_in_loop.py
from langgraph.types import interrupt, Command

async def sensitive_action(state: AgentState):
    """Action requiring human approval."""
    action = state.get("pending_action")

    # Interrupt and wait for human
    human_response = interrupt({
        "action": action,
        "question": f"Approve {action['name']} with params {action['params']}?"
    })

    if human_response.get("approved"):
        result = await execute_action(action)
        return {"messages": [ToolMessage(content=result, tool_call_id=action["id"])]}
    else:
        return {"messages": [ToolMessage(
            content=f"Rejected: {human_response.get('reason')}",
            tool_call_id=action["id"]
        )]}

# Resume after human input
config = {"configurable": {"thread_id": "thread-1"}}
agent.invoke(Command(resume={"approved": True, "reason": "Looks good"}), config)

3. Time-Travel Debugging

# debugging.py
# Get state history
history = list(agent.get_state_history(config))

for state_snapshot in history:
    print(f"Step: {state_snapshot.metadata['step']}")
    print(f"Messages: {len(state_snapshot.values['messages'])}")
    print(f"Next: {state_snapshot.next}")
    print("---")

# Replay from specific point
past_config = agent.get_state({"configurable": {"thread_id": "thread-1", "checkpoint_id": "checkpoint-5"}})
agent.invoke(new_input, past_config)  # Branches from that point

4. Structured Output with Validation

# structured_output.py
from langchain_core.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field, field_validator

class CodeReview(BaseModel):
    file: str
    issues: List[dict] = Field(default_factory=list)
    suggestions: List[str] = Field(default_factory=list)
    score: int = Field(ge=0, le=100)

    @field_validator("score")
    @classmethod
    def validate_score(cls, v):
        if not 0 <= v <= 100:
            raise ValueError("Score must be 0-100")
        return v

parser = PydanticOutputParser(pydantic_object=CodeReview)

review_prompt = f"""Review this code and output JSON:
{parser.get_format_instructions()}

Code:
{{code}}"""

async def code_reviewer(state):
    code = state["code_to_review"]
    response = await llm.ainvoke(review_prompt.format(code=code))

    try:
        review = parser.parse(response.content)
        return {"review": review.model_dump()}
    except Exception as e:
        # Retry with error feedback
        return {"error": str(e), "retry": True}

Production Checklist

Reliability

[ ] Checkpointing: Use PostgresSaver or RedisSaver for production
[ ] Error handling: Wrap nodes in try/catch, add retry logic
[ ] Timeouts: Set timeout on LLM calls and tool executions
[ ] Rate limiting: Implement token bucket for external APIs
[ ] Circuit breakers: Fail fast when dependencies are down

Observability

[ ] LangSmith tracing: os.environ["LANGCHAIN_TRACING_V2"] = "true"
[ ] Custom metrics: Log latency, token usage, error rates
[ ] State logging: Periodic state snapshots for debugging

# monitoring.py
from langsmith import traceable

@traceable(name="agent_step")
async def monitored_node(state: AgentState):
    start = time.time()
    result = await agent_node(state)
    duration = time.time() - start

    # Custom metrics
    log_metric("node_latency", duration, tags={"node": "agent"})
    log_metric("token_usage", result.get("token_count", 0))

    return result

Scaling

[ ] Horizontal scaling: Stateless nodes + external checkpointer
[ ] Queue-based: Use Celery/RQ for async node execution
[ ] Caching: Cache LLM responses for repeated queries

# caching.py
import hashlib
import redis

redis_client = redis.Redis(decode_responses=True)

async def cached_llm_call(prompt: str, model: str) -> str:
    cache_key = f"llm:{model}:{hashlib.md5(prompt.encode()).hexdigest()}"

    cached = redis_client.get(cache_key)
    if cached:
        return cached

    response = await llm.ainvoke(prompt)
    redis_client.setex(cache_key, 3600, response.content)  # 1hr TTL
    return response.content

Deployment

Docker

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Install LangGraph
RUN pip install langgraph langchain-openai langchain-core

# Copy app
COPY . .

# Run
CMD ["python", "server.py"]

FastAPI Server

# server.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from langgraph.checkpoint.sqlite import SqliteSaver
import uvicorn

app = FastAPI()
agent = build_research_agent()
checkpointer = SqliteSaver.from_conn_string("sqlite:///checkpoints.db")

class ChatRequest(BaseModel):
    message: str
    thread_id: str

class ChatResponse(BaseModel):
    response: str
    thread_id: str

@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
    config = {"configurable": {"thread_id": request.thread_id}}

    try:
        result = await agent.ainvoke(
            {"messages": [HumanMessage(content=request.message)]},
            config
        )

        last_msg = result["messages"][-1]
        return ChatResponse(
            response=last_msg.content,
            thread_id=request.thread_id
        )
    except Exception as e:
        raise HTTPException(500, str(e))

@app.get("/threads/{thread_id}/history")
async def get_history(thread_id: str):
    config = {"configurable": {"thread_id": thread_id}}
    state = agent.get_state(config)
    return {"messages": [m.content for m in state.values["messages"]]}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

Kubernetes

# k8s/langgraph-agent.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: langgraph-agent
spec:
  replicas: 3
  selector:
    matchLabels:
      app: langgraph-agent
  template:
    spec:
      containers:
      - name: agent
        image: your-registry/langgraph-agent:v1
        ports:
        - containerPort: 8000
        env:
        - name: OPENAI_API_KEY
          valueFrom:
            secretKeyRef:
              name: agent-secrets
              key: openai-key
        - name: DATABASE_URL
          value: "postgresql://user:pass@postgres:5432/agent"
        resources:
          requests:
            memory: "1Gi"
            cpu: "500m"
          limits:
            memory: "2Gi"
            cpu: "1000m"
---
apiVersion: v1
kind: Service
metadata:
  name: langgraph-agent
spec:
  selector:
    app: langgraph-agent
  ports:
  - port: 80
    targetPort: 8000

Testing Strategies

# test_agent.py
import pytest
from langgraph.graph import StateGraph

# Unit test nodes
@pytest.mark.asyncio
async def test_planner_node():
    state = {"research_topic": "AI agents", "messages": []}
    result = await planner(state)

    assert "sub_questions" in result
    assert len(result["sub_questions"]) >= 3
    assert result["current_step"] == "researching"

# Integration test with mocked LLM
@pytest.mark.asyncio
async def test_full_graph(mock_llm):
    agent = build_research_agent()

    config = {"configurable": {"thread_id": "test-1"}}
    result = await agent.ainvoke(
        {"messages": [HumanMessage(content="Test topic")], "research_topic": "Test"},
        config
    )

    assert "final_report" in result
    assert len(result["final_report"]) > 100

# Property-based testing
from hypothesis import given, strategies as st

@given(st.text(min_size=10, max_size=200))
@pytest.mark.asyncio
async def test_agent_handles_any_input(topic):
    agent = build_research_agent()
    config = {"configurable": {"thread_id": f"test-{hash(topic)}"}}

    # Should not crash
    result = await agent.ainvoke(
        {"messages": [HumanMessage(content=topic)], "research_topic": topic},
        config
    )
    assert "messages" in result

Migration from LangChain Agents

# Old LangChain Agent
from langchain.agents import AgentExecutor, create_openai_tools_agent

agent = create_openai_tools_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
result = executor.invoke({"input": "Research AI trends"})

# New LangGraph (drop-in replacement)
from langgraph.prebuilt import create_react_agent

agent = create_react_agent(llm, tools, checkpointer=MemorySaver())
result = agent.invoke(
    {"messages": [HumanMessage(content="Research AI trends")]},
    {"configurable": {"thread_id": "1"}}
)

Resources

LangGraph transforms agents from fragile prototypes into reliable, debuggable, production systems. Start simple, add persistence, then scale.

RAG Evaluation & Monitoring: Metrics, Frameworks & Production Observability

Phil IPS — Mon, 12 Oct 2026 20:19:21 +0000

RAG Evaluation & Monitoring: Metrics, Frameworks & Production Observability

Comprehensive guide to evaluating and monitoring Retrieval-Augmented Generation systems in production — from offline benchmarks to real-time observability.

Introduction

RAG systems combine retrieval + generation. Evaluation must cover both components plus the end-to-end pipeline. This guide covers metrics, frameworks, and production monitoring.

RAG Evaluation Dimensions

RAG PIPELINE:
Query → Retrieval → Rerank → Context → Generation → Answer

EVALUATION LAYERS:
1. Retrieval Quality    (precision@k, recall@k, MRR, nDCG)
2. Generation Quality   (faithfulness, relevance, hallucination)
3. End-to-End          (answer correctness, user satisfaction)
4. System Metrics       (latency, cost, throughput, errors)

Part 1: Retrieval Metrics

Offline Retrieval Evaluation

# rag_eval/retrieval_metrics.py
from typing import List, Dict, Set
import numpy as np
from rank_bm25 import BM25Okapi

def precision_at_k(retrieved: List[str], relevant: Set[str], k: int) -> float:
    """Precision@K: fraction of top-K retrieved that are relevant."""
    if k == 0:
        return 0.0
    top_k = retrieved[:k]
    relevant_retrieved = sum(1 for doc in top_k if doc in relevant)
    return relevant_retrieved / k

def recall_at_k(retrieved: List[str], relevant: Set[str], k: int) -> float:
    """Recall@K: fraction of relevant docs retrieved in top-K."""
    if len(relevant) == 0:
        return 0.0
    top_k = retrieved[:k]
    relevant_retrieved = sum(1 for doc in top_k if doc in relevant)
    return relevant_retrieved / len(relevant)

def mrr(retrieved: List[str], relevant: Set[str]) -> float:
    """Mean Reciprocal Rank: 1/rank of first relevant doc."""
    for i, doc in enumerate(retrieved, 1):
        if doc in relevant:
            return 1.0 / i
    return 0.0

def ndcg_at_k(retrieved: List[str], relevant: Dict[str, float], k: int) -> float:
    """nDCG@K: normalized discounted cumulative gain with graded relevance."""
    def dcg(retrieved_docs: List[str], rel_scores: Dict[str, float], k: int) -> float:
        score = 0.0
        for i, doc in enumerate(retrieved_docs[:k], 1):
            rel = rel_scores.get(doc, 0.0)
            score += (2**rel - 1) / np.log2(i + 1)
        return score

    ideal_dcg = dcg(sorted(relevant.keys(), key=relevant.get, reverse=True), relevant, k)
    if ideal_dcg == 0:
        return 0.0
    return dcg(retrieved, relevant, k) / ideal_dcg

def hit_rate(retrieved: List[str], relevant: Set[str], k: int) -> float:
    """Hit Rate@K: whether at least one relevant doc in top-K."""
    top_k = set(retrieved[:k])
    return 1.0 if top_k & relevant else 0.0

# Batch evaluation
def evaluate_retrieval(
    queries: List[str],
    retrieval_fn,  # query -> List[doc_ids]
    ground_truth: Dict[str, Set[str]],  # query -> relevant doc_ids
    k_values: List[int] = [1, 3, 5, 10, 20]
) -> Dict:
    """Evaluate retrieval across multiple queries."""
    results = {k: {"precision": [], "recall": [], "mrr": [], "hit_rate": []} for k in k_values}

    for query in queries:
        retrieved = retrieval_fn(query)
        relevant = ground_truth.get(query, set())

        for k in k_values:
            results[k]["precision"].append(precision_at_k(retrieved, relevant, k))
            results[k]["recall"].append(recall_at_k(retrieved, relevant, k))
            results[k]["hit_rate"].append(hit_rate(retrieved, relevant, k))

        results[k_values[0]]["mrr"].append(mrr(retrieved, relevant))  # MRR uses full list

    # Aggregate
    aggregated = {}
    for k in k_values:
        aggregated[f"precision@{k}"] = np.mean(results[k]["precision"])
        aggregated[f"recall@{k}"] = np.mean(results[k]["recall"])
        aggregated[f"hit_rate@{k}"] = np.mean(results[k]["hit_rate"])
    aggregated["mrr"] = np.mean(results[k_values[0]]["mrr"])

    return aggregated

Retrieval Ablation Studies

# rag_eval/retrieval_ablation.py
def ablation_study(
    base_retriever,
    query_set: List[str],
    ground_truth: Dict[str, Set[str]],
    variants: Dict[str, dict]
) -> Dict:
    """Compare retrieval variants: chunk size, embedding model, reranker, etc."""

    results = {}
    for name, config in variants.items():
        retriever = base_retriever.clone_with_config(config)
        metrics = evaluate_retrieval(query_set, retriever.retrieve, ground_truth)
        results[name] = metrics
        print(f"{name}: {metrics}")

    return results

# Example variants
ABLATION_VARIANTS = {
    "baseline": {"chunk_size": 512, "overlap": 50, "embed_model": "bge-small", "rerank": False},
    "larger_chunks": {"chunk_size": 1024, "overlap": 100, "embed_model": "bge-small", "rerank": False},
    "better_embedder": {"chunk_size": 512, "overlap": 50, "embed_model": "bge-large", "rerank": False},
    "with_reranker": {"chunk_size": 512, "overlap": 50, "embed_model": "bge-small", "rerank": "cross-encoder"},
    "full_upgrade": {"chunk_size": 1024, "overlap": 100, "embed_model": "bge-large", "rerank": "cross-encoder"},
}

Part 2: Generation Metrics

Faithfulness (Groundedness)

# rag_eval/generation_metrics.py
from transformers import AutoModelForSequenceClassification, AutoTokenizer
import torch

class FaithfulnessEvaluator:
    """NLI-based faithfulness: does answer entail from context?"""

    def __init__(self, model_name: str = "microsoft/deberta-v3-large-mnli"):
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.model = AutoModelForSequenceClassification.from_pretrained(model_name)
        self.model.eval()

    def score(self, context: str, answer: str) -> float:
        """Entailment probability: context → answer."""
        # Premise: context, Hypothesis: answer
        inputs = self.tokenizer(
            context, answer,
            return_tensors="pt",
            truncation=True,
            max_length=512,
            padding=True
        )
        with torch.no_grad():
            logits = self.model(**inputs).logits
            probs = torch.softmax(logits, dim=-1)
            # MNLI: 0=entailment, 1=neutral, 2=contradiction
            entailment_prob = probs[0, 0].item()
        return entailment_prob

class HallucinationDetector:
    """Detect unsupported claims in generated answer."""

    def __init__(self):
        self.nli = FaithfulnessEvaluator()

    def detect(self, context: str, answer: str, threshold: float = 0.5) -> Dict:
        """Return hallucination score and flagged spans."""
        sentences = self._split_sentences(answer)
        results = []

        for sent in sentences:
            score = self.nli.score(context, sent)
            results.append({
                "sentence": sent,
                "faithfulness": score,
                "hallucinated": score < threshold
            })

        overall = np.mean([r["faithfulness"] for r in results])
        hallucinated_spans = [r["sentence"] for r in results if r["hallucinated"]]

        return {
            "overall_faithfulness": overall,
            "hallucination_rate": len(hallucinated_spans) / len(sentences),
            "flagged_spans": hallucinated_spans,
            "per_sentence": results
        }

Relevance & Answer Quality

# rag_eval/relevance_metrics.py
from sentence_transformers import SentenceTransformer, util

class RelevanceEvaluator:
    """Semantic similarity between query and answer."""

    def __init__(self, model_name: str = "sentence-transformers/all-MiniLM-L6-v2"):
        self.model = SentenceTransformer(model_name)

    def query_answer_relevance(self, query: str, answer: str) -> float:
        """Cosine similarity between query and answer embeddings."""
        q_emb = self.model.encode(query, convert_to_tensor=True)
        a_emb = self.model.encode(answer, convert_to_tensor=True)
        return util.cos_sim(q_emb, a_emb).item()

    def context_answer_relevance(self, context: str, answer: str) -> float:
        """How well answer addresses context."""
        c_emb = self.model.encode(context, convert_to_tensor=True)
        a_emb = self.model.encode(answer, convert_to_tensor=True)
        return util.cos_sim(c_emb, a_emb).item()

class AnswerCorrectnessEvaluator:
    """LLM-as-judge for answer correctness."""

    def __init__(self, judge_model: str = "gpt-4o-mini"):
        self.judge_model = judge_model
        # Use OpenAI API or local model

    def evaluate(
        self,
        query: str,
        ground_truth: str,
        generated: str,
        context: str = None
    ) -> Dict:
        """Score 1-5 on correctness, completeness, conciseness."""
        prompt = f"""Rate the generated answer (1-5) on:
1. Correctness: factually accurate vs ground truth
2. Completeness: covers all key points from ground truth
3. Conciseness: no fluff, direct answer

Query: {query}
Ground Truth: {ground_truth}
Generated: {generated}
Context: {context or 'N/A'}

Return JSON: {{"correctness": 1-5, "completeness": 1-5, "conciseness": 1-5, "reasoning": "..."}}"""

        # Call judge model
        return self._call_judge(prompt)

Part 3: End-to-End Frameworks

Ragas — Reference-Free Evaluation

# rag_eval/ragas_integration.py
from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
    context_relevancy,
    answer_correctness,
    answer_similarity,
)
from datasets import Dataset

def evaluate_rag_with_ragas(
    questions: List[str],
    answers: List[str],
    contexts: List[List[str]],
    ground_truths: List[str] = None
) -> Dict:
    """Evaluate RAG using Ragas metrics."""

    data = {
        "question": questions,
        "answer": answers,
        "contexts": contexts,
    }
    if ground_truths:
        data["ground_truth"] = ground_truths

    dataset = Dataset.from_dict(data)

    # Select metrics based on available data
    metrics = [
        faithfulness,
        answer_relevancy,
        context_precision,
        context_recall,
    ]
    if ground_truths:
        metrics.extend([answer_correctness, answer_similarity])

    result = evaluate(dataset, metrics=metrics)
    return result.to_pandas().to_dict(orient="records")[0]

# Ragas metric explanations
RAGAS_METRICS = {
    "faithfulness": "Answer grounded in context? (0-1, higher better)",
    "answer_relevancy": "Answer relevant to question? (0-1)",
    "context_precision": "Retrieved contexts ranked by relevance? (0-1)",
    "context_recall": "All needed info retrieved? (0-1)",
    "context_relevancy": "Context relevant to question? (0-1)",
    "answer_correctness": "Answer matches ground truth? (0-1, needs GT)",
    "answer_similarity": "Semantic similarity to ground truth? (0-1, needs GT)",
}

DeepEval — Unit Testing for LLMs

# rag_eval/deepeval_integration.py
from deepeval import evaluate
from deepeval.metrics import (
    AnswerRelevancyMetric,
    FaithfulnessMetric,
    ContextualPrecisionMetric,
    ContextualRecallMetric,
    ContextualRelevancyMetric,
    HallucinationMetric,
    ToxicityMetric,
    BiasMetric,
)
from deepeval.test_case import LLMTestCase

def evaluate_rag_with_deepeval(
    test_cases: List[LLMTestCase],
    threshold: float = 0.7
) -> Dict:
    """Evaluate with DeepEval's comprehensive metrics."""

    metrics = [
        AnswerRelevancyMetric(threshold=threshold),
        FaithfulnessMetric(threshold=threshold),
        ContextualPrecisionMetric(threshold=threshold),
        ContextualRecallMetric(threshold=threshold),
        ContextualRelevancyMetric(threshold=threshold),
        HallucinationMetric(threshold=threshold),
        ToxicityMetric(threshold=0.5),
    ]

    results = evaluate(test_cases, metrics)
    return {
        "passed": all(r.success for r in results),
        "scores": {m.__class__.__name__: r.score for r in results for m in r.metrics},
        "details": results
    }

# Test case creation
def create_test_case(
    query: str,
    answer: str,
    retrieval_context: List[str],
    expected_output: str = None
) -> LLMTestCase:
    return LLMTestCase(
        input=query,
        actual_output=answer,
        retrieval_context=retrieval_context,
        expected_output=expected_output,
    )

TruLens — Instrumentation & Feedback

# rag_eval/trulens_integration.py
from trulens_eval import Tru, Feedback, OpenAI as fOpenAI
from trulens_eval.feedback import Groundedness
import numpy as np

def setup_trulens():
    """Configure TruLens feedback functions."""
    provider = fOpenAI()

    # Groundedness: answer supported by context
    grounded = Groundedness(groundedness_provider=provider)
    f_groundedness = Feedback(
        grounded.groundedness_measure_with_cot_reasons,
        name="Groundedness"
    ).on_input_output()

    # Answer relevance
    f_answer_relevance = Feedback(
        provider.relevance,
        name="Answer Relevance"
    ).on_input_output()

    # Context relevance
    f_context_relevance = Feedback(
        provider.context_relevance,
        name="Context Relevance"
    ).on_input()

    return [f_groundedness, f_answer_relevance, f_context_relevance]

def instrument_rag_chain(rag_chain, feedback_funcs):
    """Wrap RAG chain for TruLens logging."""
    from trulens_eval import TruChain

    tru_chain = TruChain(
        rag_chain,
        app_id="rag_qa",
        feedbacks=feedback_funcs
    )
    return tru_chain

Part 4: Production Monitoring

Real-Time Metrics Pipeline

# rag_monitoring/production_metrics.py
from dataclasses import dataclass, field
from typing import List, Optional
from datetime import datetime
import time
import asyncio
from prometheus_client import Counter, Histogram, Gauge, Summary

# Prometheus metrics
RAG_REQUESTS_TOTAL = Counter(
    "rag_requests_total",
    "Total RAG requests",
    ["status"]  # success, error, timeout
)

RAG_LATENCY = Histogram(
    "rag_latency_seconds",
    "RAG request latency",
    buckets=[0.5, 1, 2, 5, 10, 30, 60]
)

RETRIEVAL_LATENCY = Histogram(
    "rag_retrieval_latency_seconds",
    "Retrieval step latency"
)

GENERATION_LATENCY = Histogram(
    "rag_generation_latency_seconds",
    "Generation step latency"
)

RAG_TOKENS = Counter(
    "rag_tokens_total",
    "Total tokens used",
    ["type"]  # prompt, completion
)

RAG_COST = Counter(
    "rag_cost_usd_total",
    "Total cost in USD"
)

ACTIVE_REQUESTS = Gauge(
    "rag_active_requests",
    "Currently processing requests"
)

QUALITY_SCORE = Histogram(
    "rag_quality_score",
    "Quality scores from evaluators",
    ["metric"],  # faithfulness, relevance, etc.
    buckets=[0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0]
)

@dataclass
class RAGRequestLog:
    request_id: str
    timestamp: datetime
    query: str
    retrieved_docs: List[str]
    answer: str
    latency_ms: float
    retrieval_latency_ms: float
    generation_latency_ms: float
    prompt_tokens: int
    completion_tokens: int
    cost_usd: float
    quality_scores: dict = field(default_factory=dict)
    error: Optional[str] = None

class ProductionMonitor:
    """Production RAG monitoring with alerting."""

    def __init__(
        self,
        alert_webhook: str = None,
        quality_threshold: float = 0.6,
        latency_p99_threshold: float = 30.0,
        error_rate_threshold: float = 0.05
    ):
        self.alert_webhook = alert_webhook
        self.quality_threshold = quality_threshold
        self.latency_threshold = latency_p99_threshold
        self.error_rate_threshold = error_rate_threshold
        self.request_logs: List[RAGRequestLog] = []
        self.window_logs: List[RAGRequestLog] = []

    async def log_request(self, log: RAGRequestLog):
        """Log request and update metrics."""
        self.request_logs.append(log)
        self.window_logs.append(log)

        # Update Prometheus
        RAG_REQUESTS_TOTAL.labels(
            status="error" if log.error else "success"
        ).inc()

        RAG_LATENCY.observe(log.latency_ms / 1000)
        RETRIEVAL_LATENCY.observe(log.retrieval_latency_ms / 1000)
        GENERATION_LATENCY.observe(log.generation_latency_ms / 1000)

        RAG_TOKENS.labels(type="prompt").inc(log.prompt_tokens)
        RAG_TOKENS.labels(type="completion").inc(log.completion_tokens)
        RAG_COST.inc(log.cost_usd)

        for metric, score in log.quality_scores.items():
            QUALITY_SCORE.labels(metric=metric).observe(score)

        # Check alerts
        await self._check_alerts(log)

    async def _check_alerts(self, log: RAGRequestLog):
        """Check and fire alerts."""
        alerts = []

        # Quality alert
        for metric, score in log.quality_scores.items():
            if score < self.quality_threshold:
                alerts.append(f"Quality alert: {metric}={score:.2f} < {self.quality_threshold}")

        # Latency alert
        if log.latency_ms > self.latency_threshold * 1000:
            alerts.append(f"Latency alert: {log.latency_ms}ms > {self.latency_threshold}s")

        # Error rate alert (rolling window)
        await self._check_error_rate()

        for alert in alerts:
            await self._send_alert(alert)

    async def _check_error_rate(self):
        """Check error rate in last 100 requests."""
        if len(self.window_logs) >= 100:
            recent = self.window_logs[-100:]
            errors = sum(1 for r in recent if r.error)
            rate = errors / 100
            if rate > self.error_rate_threshold:
                await self._send_alert(f"Error rate alert: {rate:.1%} > {self.error_rate_threshold:.1%}")

    async def _send_alert(self, message: str):
        """Send alert to webhook."""
        if self.alert_webhook:
            # Implement webhook call
            print(f"ALERT: {message}")

    def get_dashboard_data(self, hours: int = 24) -> Dict:
        """Get data for Grafana dashboard."""
        cutoff = datetime.now() - timedelta(hours=hours)
        recent = [r for r in self.request_logs if r.timestamp > cutoff]

        if not recent:
            return {}

        return {
            "total_requests": len(recent),
            "success_rate": sum(1 for r in recent if not r.error) / len(recent),
            "avg_latency_ms": np.mean([r.latency_ms for r in recent]),
            "p99_latency_ms": np.percentile([r.latency_ms for r in recent], 99),
            "avg_cost_per_request": np.mean([r.cost_usd for r in recent]),
            "total_cost": sum(r.cost_usd for r in recent),
            "avg_quality": {
                metric: np.mean([r.quality_scores.get(metric, 0) for r in recent])
                for metric in ["faithfulness", "relevance", "correctness"]
            },
        }

Distributed Tracing

# rag_monitoring/tracing.py
from opentelemetry import trace
from opentelemetry.exporter.jaeger import JaegerExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor

def setup_tracing(service_name: str = "rag-service"):
    """Setup OpenTelemetry tracing."""

    trace.set_tracer_provider(TracerProvider())
    jaeger_exporter = JaegerExporter(
        agent_host_name="jaeger",
        agent_port=6831,
    )
    trace.get_tracer_provider().add_span_processor(
        BatchSpanProcessor(jaeger_exporter)
    )

    return trace.get_tracer(__name__)

class TracedRAGChain:
    """RAG chain with full distributed tracing."""

    def __init__(self, retriever, generator, tracer):
        self.retriever = retriever
        self.generator = generator
        self.tracer = tracer

    async def answer(self, query: str) -> dict:
        with self.tracer.start_as_current_span("rag_query") as span:
            span.set_attribute("query", query)

            # Retrieval span
            with self.tracer.start_as_current_span("retrieval") as retr_span:
                start = time.time()
                docs = await self.retriever.retrieve(query)
                retr_span.set_attribute("num_docs", len(docs))
                retr_span.set_attribute("latency_ms", (time.time() - start) * 1000)

            # Generation span
            with self.tracer.start_as_current_span("generation") as gen_span:
                start = time.time()
                answer = await self.generator.generate(query, docs)
                gen_span.set_attribute("answer_length", len(answer))
                gen_span.set_attribute("latency_ms", (time.time() - start) * 1000)

            span.set_attribute("total_latency_ms", (time.time() - start) * 1000)
            return {"answer": answer, "sources": docs}

Part 5: Automated Evaluation Pipeline

CI/CD Integration

# rag_eval/ci_pipeline.py
import pytest
from ragas import evaluate
from datasets import Dataset

class RAGEvalTestSuite:
    """Pytest-style RAG evaluation tests."""

    @pytest.fixture(scope="class")
    def test_data(self):
        return [
            {
                "question": "What is LoRA?",
                "ground_truth": "LoRA is Low-Rank Adaptation for efficient fine-tuning...",
                "contexts": ["LoRA freezes base weights and adds low-rank adapters..."],
            },
            # ... more test cases
        ]

    def test_faithfulness_threshold(self, rag_chain, test_data):
        """Faithfulness must exceed threshold."""
        results = []
        for case in test_data:
            answer = rag_chain.answer(case["question"])
            score = self._compute_faithfulness(case["contexts"][0], answer)
            results.append(score)
            assert score >= 0.7, f"Faithfulness {score:.2f} below threshold for: {case['question']}"

    def test_retrieval_precision(self, rag_chain, test_data):
        """Retrieval precision@5 must exceed threshold."""
        for case in test_data:
            retrieved = rag_chain.retrieve(case["question"])
            relevant = self._get_relevant_docs(case["question"])
            p5 = precision_at_k(retrieved, relevant, 5)
            assert p5 >= 0.6, f"Precision@5 {p5:.2f} below threshold for: {case['question']}"

    def test_latency_sla(self, rag_chain, test_data):
        """P99 latency must be under 10 seconds."""
        latencies = []
        for case in test_data:
            start = time.time()
            rag_chain.answer(case["question"])
            latencies.append(time.time() - start)

        p99 = np.percentile(latencies, 99)
        assert p99 <= 10.0, f"P99 latency {p99:.1f}s exceeds 10s SLA"

    def test_cost_per_query(self, rag_chain, test_data):
        """Average cost per query under $0.01."""
        costs = [rag_chain.get_last_cost() for _ in test_data]
        avg_cost = np.mean(costs)
        assert avg_cost <= 0.01, f"Avg cost ${avg_cost:.4f} exceeds $0.01"

# Regression detection
def detect_regression(
    current_scores: Dict,
    baseline_scores: Dict,
    threshold: float = 0.05
) -> List[str]:
    """Detect metric regressions vs baseline."""
    regressions = []
    for metric, current in current_scores.items():
        baseline = baseline_scores.get(metric)
        if baseline and current < baseline - threshold:
            regressions.append(
                f"REGRESSION: {metric} dropped from {baseline:.3f} to {current:.3f} "
                f"(delta: {current-baseline:.3f})"
            )
    return regressions

A/B Testing Framework

# rag_eval/ab_testing.py
import random
from dataclasses import dataclass
from typing: Callable, Dict, List
import numpy as np
from scipy import stats

@dataclass
class ABTestResult:
    variant_a: str
    variant_b: str
    metric: str
    a_mean: float
    b_mean: float
    p_value: float
    significant: bool
    winner: str

class RAGABTester:
    """A/B test RAG variants in production."""

    def __init__(self, traffic_split: float = 0.5):
        self.traffic_split = traffic_split
        self.variants: Dict[str, Callable] = {}
        self.results: Dict[str, List[float]] = {}

    def register_variant(self, name: str, chain_factory: Callable):
        self.variants[name] = chain_factory
        self.results[name] = []

    def route_request(self, query: str) -> tuple:
        """Route to variant based on traffic split."""
        variant_name = "A" if random.random() < self.traffic_split else "B"
        chain = self.variants[variant_name]()
        return variant_name, chain

    def record_result(self, variant: str, score: float):
        self.results[variant].append(score)

    def analyze(self, metric: str = "quality") -> ABTestResult:
        """Statistical significance test."""
        a_scores = self.results["A"]
        b_scores = self.results["B"]

        if len(a_scores) < 30 or len(b_scores) < 30:
            return ABTestResult(
                "A", "B", metric,
                np.mean(a_scores), np.mean(b_scores),
                1.0, False, "insufficient_data"
            )

        # Welch's t-test (unequal variance)
        t_stat, p_value = stats.ttest_ind(a_scores, b_scores, equal_var=False)

        a_mean, b_mean = np.mean(a_scores), np.mean(b_scores)
        winner = "A" if a_mean > b_mean else "B"

        return ABTestResult(
            "A", "B", metric,
            a_mean, b_mean,
            p_value, p_value < 0.05, winner
        )

Part 6: Data Quality & Drift Detection

Embedding Drift Monitoring

# rag_monitoring/drift_detection.py
from sklearn.decomposition import PCA
from sklearn.manifold import TSNE
import numpy as np
from scipy.spatial.distance import jensenshannon

class EmbeddingDriftDetector:
    """Detect drift in query/document embeddings."""

    def __init__(self, reference_embeddings: np.ndarray, window_size: int = 1000):
        self.reference = reference_embeddings
        self.window_size = window_size
        self.recent_embeddings: List[np.ndarray] = []

        # Fit PCA on reference for visualization
        self.pca = PCA(n_components=2)
        self.pca.fit(reference_embeddings)

    def add_embeddings(self, embeddings: np.ndarray):
        self.recent_embeddings.append(embeddings)
        if len(self.recent_embeddings) > self.window_size:
            self.recent_embeddings.pop(0)

    def compute_drift_score(self) -> Dict:
        """Compute multiple drift metrics."""
        if not self.recent_embeddings:
            return {"drift_detected": False}

        recent = np.vstack(self.recent_embeddings)

        # 1. Mean shift (L2 distance between centroids)
        ref_centroid = self.reference.mean(axis=0)
        recent_centroid = recent.mean(axis=0)
        mean_shift = np.linalg.norm(ref_centroid - recent_centroid)

        # 2. Covariance shift (Frobenius norm)
        ref_cov = np.cov(self.reference.T)
        recent_cov = np.cov(recent.T)
        cov_shift = np.linalg.norm(ref_cov - recent_cov, 'fro')

        # 3. JS divergence on PCA projections
        ref_pca = self.pca.transform(self.reference)
        recent_pca = self.pca.transform(recent)

        # 2D histogram
        hist_range = [[ref_pca[:,0].min(), ref_pca[:,0].max()],
                      [ref_pca[:,1].min(), ref_pca[:,1].max()]]
        ref_hist, _, _ = np.histogram2d(ref_pca[:,0], ref_pca[:,1], bins=20, range=hist_range, density=True)
        recent_hist, _, _ = np.histogram2d(recent_pca[:,0], recent_pca[:,1], bins=20, range=hist_range, density=True)

        # Flatten and add epsilon
        ref_flat = ref_hist.flatten() + 1e-10
        recent_flat = recent_hist.flatten() + 1e-10

        js_div = jensenshannon(ref_flat, recent_flat)

        return {
            "drift_detected": js_div > 0.1 or mean_shift > 0.5,
            "mean_shift": float(mean_shift),
            "cov_shift": float(cov_shift),
            "js_divergence": float(js_div),
            "ref_centroid": ref_centroid.tolist(),
            "recent_centroid": recent_centroid.tolist(),
        }

Data Quality Checks

# rag_monitoring/data_quality.py
def validate_rag_inputs(
    query: str,
    retrieved_docs: List[str],
    answer: str
) -> Dict:
    """Validate inputs for quality issues."""

    issues = []

    # Query checks
    if len(query.strip()) < 3:
        issues.append("Query too short")
    if len(query) > 2000:
        issues.append("Query extremely long")

    # Retrieval checks
    if not retrieved_docs:
        issues.append("No documents retrieved")
    else:
        avg_doc_len = np.mean([len(d) for d in retrieved_docs])
        if avg_doc_len < 50:
            issues.append("Retrieved docs very short")
        if avg_doc_len > 5000:
            issues.append("Retrieved docs very long")

        # Check for duplicate docs
        unique_docs = set(retrieved_docs)
        if len(unique_docs) < len(retrieved_docs) * 0.8:
            issues.append("High duplicate rate in retrieval")

    # Answer checks
    if not answer.strip():
        issues.append("Empty answer")
    if len(answer) < 10:
        issues.append("Answer too short")
    if answer.count("I don't know") > 0:
        issues.append("Model abstained")

    # Repetition check
    words = answer.split()
    if len(words) > 10:
        unique_ratio = len(set(words)) / len(words)
        if unique_ratio < 0.3:
            issues.append("High repetition in answer")

    return {
        "valid": len(issues) == 0,
        "issues": issues,
        "severity": "high" if any("Empty" in i or "No documents" in i for i in issues) else "low"
    }

Part 7: Dashboards & Alerting

Grafana Dashboard JSON

# rag_monitoring/grafana_dashboard.json
{
  "dashboard": {
    "title": "RAG Production Monitoring",
    "panels": [
      {
        "title": "Request Rate",
        "targets": [{"expr": "rate(rag_requests_total[5m])"}]
      },
      {
        "title": "Error Rate",
        "targets": [{"expr": "rate(rag_requests_total{status=\"error\"}[5m]) / rate(rag_requests_total[5m])"}]
      },
      {
        "title": "Latency P50/P95/P99",
        "targets": [
          {"expr": "histogram_quantile(0.50, rag_latency_seconds)"},
          {"expr": "histogram_quantile(0.95, rag_latency_seconds)"},
          {"expr": "histogram_quantile(0.99, rag_latency_seconds)"}
        ]
      },
      {
        "title": "Quality Scores",
        "targets": [
          {"expr": "avg by (metric) (rag_quality_score)"}
        ]
      },
      {
        "title": "Cost per Request",
        "targets": [{"expr": "rate(rag_cost_usd_total[1h]) / rate(rag_requests_total[1h])"}]
      },
      {
        "title": "Tokens per Request",
        "targets": [
          {"expr": "rate(rag_tokens_total{type=\"prompt\"}[5m]) / rate(rag_requests_total[5m])"},
          {"expr": "rate(rag_tokens_total{type=\"completion\"}[5m]) / rate(rag_requests_total[5m])"}
        ]
      }
    ]
  }
}

Alert Rules

# rag_monitoring/alerts.yaml
groups:
- name: rag_alerts
  rules:
  - alert: RAGHighErrorRate
    expr: rate(rag_requests_total{status="error"}[5m]) / rate(rag_requests_total[5m]) > 0.05
    for: 5m
    labels:
      severity: critical
    annotations:
      summary: "RAG error rate > 5%"

  - alert: RAGHighLatency
    expr: histogram_quantile(0.99, rag_latency_seconds) > 30
    for: 5m
    labels:
      severity: warning
    annotations:
      summary: "RAG P99 latency > 30s"

  - alert: RAGLowFaithfulness
    expr: avg(rag_quality_score{metric="faithfulness"}) < 0.6
    for: 10m
    labels:
      severity: warning
    annotations:
      summary: "RAG faithfulness dropped below 0.6"

  - alert: RAGHighCost
    expr: rate(rag_cost_usd_total[1h]) > 10
    for: 15m
    labels:
      severity: warning
    annotations:
      summary: "RAG cost > $10/hour"

Part 8: Continuous Improvement Loop

# rag_eval/improvement_loop.py
class RAGImprovementLoop:
    """Automated RAG improvement via evaluation feedback."""

    def __init__(self, rag_system, evaluator, optimizer):
        self.rag = rag_system
        self.evaluator = evaluator
        self.optimizer = optimizer
        self.history = []

    async def run_iteration(self, eval_queries: List[str]):
        """One improvement cycle."""

        # 1. Evaluate current system
        current_metrics = await self.evaluator.evaluate(self.rag, eval_queries)

        # 2. Identify weak spots
        weak_spots = self._identify_weaknesses(current_metrics)

        # 3. Generate optimization hypotheses
        hypotheses = self.optimizer.generate_hypotheses(weak_spots)

        # 4. Test hypotheses (A/B or shadow)
        best_variant = await self._test_hypotheses(hypotheses, eval_queries)

        # 5. Deploy if significant improvement
        if best_variant and best_variant.improvement > 0.02:
            await self._deploy_variant(best_variant)
            self.history.append({
                "iteration": len(self.history),
                "metrics": current_metrics,
                "deployed": best_variant.name,
                "improvement": best_variant.improvement
            })

        return current_metrics

    def _identify_weaknesses(self, metrics: Dict) -> List[str]:
        """Identify which metrics need improvement."""
        thresholds = {
            "faithfulness": 0.8,
            "answer_relevancy": 0.75,
            "context_precision": 0.7,
            "context_recall": 0.7,
        }

        return [
            metric for metric, threshold in thresholds.items()
            if metrics.get(metric, 0) < threshold
        ]

Part 9: Cost Optimization

Token & Cost Tracking

# rag_monitoring/cost_optimization.py
class CostOptimizer:
    """Optimize RAG costs while maintaining quality."""

    def __init__(self, pricing: Dict):
        self.pricing = pricing  # {"gpt-4o": {"input": 5, "output": 15}, ...}

    def estimate_cost(self, prompt_tokens: int, completion_tokens: int, model: str) -> float:
        """Estimate cost in USD."""
        rates = self.pricing.get(model, {"input": 0, "output": 0})
        return (prompt_tokens * rates["input"] + completion_tokens * rates["output"]) / 1_000_000

    def optimize_model_selection(
        self,
        query: str,
        complexity_score: float,  # 0-1
        quality_requirement: float  # 0-1
    ) -> str:
        """Route to cheapest model meeting quality."""

        models = [
            ("gpt-4o-mini", 0.7, 0.15),   # quality, cost_factor
            ("gpt-4o", 0.9, 1.0),
            ("gpt-4-turbo", 0.95, 2.0),
        ]

        for model, quality, cost_factor in models:
            if quality >= quality_requirement:
                # Check if we can use cheaper for simple queries
                if complexity_score < 0.3 and model != "gpt-4o-mini":
                    continue
                return model

        return "gpt-4o"  # fallback

    def optimize_retrieval(self, retrieval_config: Dict) -> Dict:
        """Find optimal retrieval config for cost/quality."""
        # Fewer docs = lower prompt tokens = lower cost
        # But too few = lower quality
        # Use Bayesian optimization
        pass

Resources

RAG evaluation is not one-time — it's continuous. Automate metrics, alert on regressions, and close the loop with A/B testing. The best RAG system is the one you can measure and improve.

Vector Databases Deep Dive: Qdrant, Pinecone, Weaviate & Milvus Compared

Phil IPS — Mon, 12 Oct 2026 20:14:51 +0000

Vector Databases Deep Dive: Qdrant, Pinecone, Weaviate & Milvus Compared

Complete comparison of the four major vector databases for AI applications — architecture, indexing, performance, scaling, and production deployment.

Introduction

Vector databases are the backbone of modern AI applications — RAG, semantic search, recommendations, and embeddings storage. This guide compares the four major players with production insights.

Vector Database Landscape

Database	Type	License	Best For
Qdrant	Purpose-built	Apache 2.0	High performance, filtering, Rust
Pinecone	Managed service	Proprietary	Zero-ops, enterprise
Weaviate	Purpose-built	BSD-3	GraphQL, modules, hybrid search
Milvus	Purpose-built	Apache 2.0	Massive scale, Kubernetes native

Part 1: Core Concepts

Vector Index Types

# vectordb/indexes.py
from enum import Enum
from dataclasses import dataclass
from typing: List

class IndexType(Enum):
    FLAT = "flat"                    # Exact search, O(n)
    IVF_FLAT = "ivf_flat"           # Inverted file, balanced
    IVF_PQ = "ivf_pq"               # Product quantization, compressed
    HNSW = "hnsw"                   # Hierarchical NSW, fast approximate
    ANNOY = "annoy"                 # Trees, good for static data
    SCANN = "scann"                 # Google's ScaNN, high recall
    DISKANN = "diskann"             # Disk-based, large scale

@dataclass
class IndexConfig:
    index_type: IndexType
    params: dict

    # HNSW params
    m: int = 16              # Connections per layer
    ef_construct: int = 100  # Build-time search width
    ef_search: int = 50      # Query-time search width

    # IVF params
    nlist: int = 100         # Number of clusters
    nprobe: int = 10         # Clusters to search

    # PQ params
    m_pq: int = 8            # Subquantizers
    nbits: int = 8           # Bits per subquantizer

# Recommendations by use case
INDEX_RECOMMENDATIONS = {
    "small_exact": IndexConfig(IndexType.FLAT, {}),
    "medium_fast": IndexConfig(IndexType.HNSW, {"m": 16, "ef_construct": 200}),
    "large_scale": IndexConfig(IndexType.IVF_PQ, {"nlist": 4096, "m_pq": 16}),
    "massive": IndexConfig(IndexType.DISKANN, {}),
}

Distance Metrics

# vectordb/metrics.py
class DistanceMetric(Enum):
    COSINE = "cosine"           # Normalized dot product, most common
    EUCLIDEAN = "euclidean"     # L2 distance
    DOT = "dot"                 # Inner product (requires normalized vectors)
    MANHATTAN = "manhattan"     # L1 distance
    HAMMING = "hamming"         # Binary vectors
    JACCARD = "jaccard"         # Set similarity

# Choose metric based on embedding model
METRIC_BY_MODEL = {
    "text-embedding-3-large": DistanceMetric.COSINE,
    "text-embedding-3-small": DistanceMetric.COSINE,
    "bge-large-en-v1.5": DistanceMetric.COSINE,
    "e5-large-v2": DistanceMetric.COSINE,
    "instructor-xl": DistanceMetric.COSINE,
    # Some models work with dot product (vectors pre-normalized)
}

def normalize_vectors(vectors: List[List[float]]) -> List[List[float]]:
    """Normalize for cosine similarity with dot product index."""
    import numpy as np
    arr = np.array(vectors)
    norms = np.linalg.norm(arr, axis=1, keepdims=True)
    return (arr / (norms + 1e-8)).tolist()

Part 2: Qdrant Deep Dive

Architecture & Features

# vectordb/qdrant_client.py
from qdrant_client import QdrantClient
from qdrant_client.models import *
from typing: List, Dict, Optional
import numpy as np

class QdrantManager:
    def __init__(self, url: str = "http://localhost:6333", api_key: str = None):
        self.client = QdrantClient(url=url, api_key=api_key)

    def create_collection(
        self,
        name: str,
        vector_size: int,
        distance: Distance = Distance.COSINE,
        hnsw_config: HnswConfigDiff = None,
        quantization_config: QuantizationConfig = None,
        on_disk_payload: bool = True
    ):
        """Create optimized collection."""

        vectors_config = VectorParams(
            size=vector_size,
            distance=distance,
            hnsw_config=hnsw_config or HnswConfigDiff(
                m=16,
                ef_construct=200,
                full_scan_threshold=10000
            ),
            quantization_config=quantization_config or ScalarQuantizationConfig(
                type=ScalarQuantizationType.INT8,
                quantile=0.99,
                always_ram=True
            ),
            on_disk=on_disk_payload
        )

        self.client.create_collection(
            collection_name=name,
            vectors_config=vectors_config,
            optimizers_config=OptimizersConfigDiff(
                default_segment_number=2,
                max_segment_size=20000,
                memmap_threshold=50000,
                indexing_threshold=20000,
                flush_interval_sec=5
            ),
            wal_config=WalConfigDiff(wal_capacity_mb=32),
        )

    def upsert_batch(
        self,
        collection_name: str,
        points: List[PointStruct],
        batch_size: int = 1000
    ) -> UpdateResult:
        """Efficient batch upsert."""
        for i in range(0, len(points), batch_size):
            batch = points[i:i + batch_size]
            self.client.upsert(
                collection_name=collection_name,
                points=batch,
                wait=True
            )
        return UpdateResult(status=UpdateStatus.COMPLETED)

    def search(
        self,
        collection_name: str,
        query_vector: List[float],
        limit: int = 10,
        score_threshold: float = None,
        query_filter: Filter = None,
        search_params: SearchParams = None
    ) -> List[ScoredPoint]:
        """High-performance search with filtering."""
        return self.client.search(
            collection_name=collection_name,
            query_vector=query_vector,
            limit=limit,
            score_threshold=score_threshold,
            query_filter=query_filter,
            search_params=search_params or SearchParams(
                hnsw_ef=128,
                exact=False
            )
        )

    def hybrid_search(
        self,
        collection_name: str,
        query_vector: List[float],
        query_text: str,
        limit: int = 10,
        vector_weight: float = 0.7,
        text_weight: float = 0.3
    ) -> List[ScoredPoint]:
        """Hybrid vector + keyword search (requires sparse vectors)."""
        # Requires sparse vector index setup
        return self.client.query_points(
            collection_name=collection_name,
            prefetch=[
                Prefetch(
                    query=query_vector,
                    using="dense",
                    limit=limit * 2
                ),
                Prefetch(
                    query=query_text,
                    using="sparse",
                    limit=limit * 2
                )
            ],
            query=FusionQuery(fusion=Fusion.RRF),
            limit=limit
        )

Qdrant Performance Tuning

# vectordb/qdrant_tuning.py
QDRANT_TUNING_GUIDE = {
    "memory_optimization": {
        "quantization": "Use scalar (INT8) or binary quantization for 4-8x memory reduction",
        "on_disk_payload": "Store payload on disk, keep only vectors in RAM",
        "memmap_threshold": "Enable memmap for segments > 50k vectors",
        "hnsw_m": "Lower m (8-16) for memory, higher (32-64) for recall",
    },
    "search_speed": {
        "hnsw_ef_search": "Increase for recall (50-256), decrease for speed",
        "exact_search": "Use exact=True for small collections (<10k)",
        "parallel_queries": "Batch multiple queries in single request",
        "filter_optimization": "Put selective filters first, use payload indexes",
    },
    "indexing_speed": {
        "max_segment_size": "Larger segments = faster indexing, more memory",
        "indexing_threshold": "Delay indexing until segment reaches threshold",
        "default_segment_number": "Match CPU cores for parallel indexing",
        "flush_interval_sec": "Balance durability vs performance",
    },
    "scaling": {
        "sharding": "Auto-shard by setting shard_number > 1",
        "replication": "Set replication_factor >= 2 for HA",
        "distributed": "Use Qdrant Cluster for multi-node",
    }
}

Part 3: Pinecone Deep Dive

Serverless Architecture

# vectordb/pinecone_client.py
from pinecone import Pinecone, ServerlessSpec
from typing: List, Dict, Optional
import time

class PineconeManager:
    def __init__(self, api_key: str):
        self.pc = Pinecone(api_key=api_key)

    def create_index(
        self,
        name: str,
        dimension: int,
        metric: str = "cosine",
        spec: ServerlessSpec = None,
        deletion_protection: str = "disabled"
    ):
        """Create serverless index."""
        if spec is None:
            spec = ServerlessSpec(cloud="aws", region="us-east-1")

        self.pc.create_index(
            name=name,
            dimension=dimension,
            metric=metric,
            spec=spec,
            deletion_protection=deletion_protection
        )

        # Wait for ready
        while not self.pc.describe_index(name).status["ready"]:
            time.sleep(1)

    def upsert(
        self,
        index_name: str,
        vectors: List[Dict],
        namespace: str = "",
        batch_size: int = 100
    ):
        """Batch upsert with automatic retry."""
        index = self.pc.Index(index_name)

        for i in range(0, len(vectors), batch_size):
            batch = vectors[i:i + batch_size]
            index.upsert(vectors=batch, namespace=namespace)

    def query(
        self,
        index_name: str,
        vector: List[float],
        top_k: int = 10,
        namespace: str = "",
        filter: Dict = None,
        include_metadata: bool = True,
        include_values: bool = False
    ):
        index = self.pc.Index(index_name)
        return index.query(
            vector=vector,
            top_k=top_k,
            namespace=namespace,
            filter=filter,
            include_metadata=include_metadata,
            include_values=include_values
        )

    def query_by_id(
        self,
        index_name: str,
        id: str,
        top_k: int = 10,
        namespace: str = ""
    ):
        """Query using existing vector ID (efficient for recommendations)."""
        index = self.pc.Index(index_name)
        return index.query(
            id=id,
            top_k=top_k,
            namespace=namespace
        )

Pinecone Cost Optimization

# vectordb/pinecone_costs.py
PINECONE_COSTS = {
    "serverless": {
        "storage_gb_month": 0.33,      # $0.33/GB/month
        "read_units": 0.000065,        # Per 1000 read units
        "write_units": 0.000013,       # Per 1000 write units
    },
    "pod_based": {
        "p1_x1": 0.096,                # $0.096/hour
        "p1_x2": 0.192,
        "p1_x4": 0.384,
        "p1_x8": 0.768,
        "p2_x1": 0.346,
        "p2_x2": 0.692,
        "p2_x4": 1.384,
        "p2_x8": 2.768,
    }
}

def estimate_monthly_cost(
    index_type: str,
    vector_count: int,
    dimension: int,
    queries_per_month: int,
    writes_per_month: int,
    pod_type: str = "p1_x1"
) -> dict:
    """Estimate Pinecone costs."""

    # Storage: ~4 bytes per dimension per vector (float32)
    storage_gb = (vector_count * dimension * 4) / (1024 ** 3)

    if index_type == "serverless":
        storage_cost = storage_gb * PINECONE_COSTS["serverless"]["storage_gb_month"]
        read_cost = (queries_per_month / 1000) * PINECONE_COSTS["serverless"]["read_units"]
        write_cost = (writes_per_month / 1000) * PINECONE_COSTS["serverless"]["write_units"]
        total = storage_cost + read_cost + write_cost
    else:
        pod_cost = PINECONE_COSTS["pod_based"][pod_type] * 24 * 30
        total = pod_cost

    return {
        "storage_gb": storage_gb,
        "monthly_cost_usd": total,
        "breakdown": {
            "storage": storage_cost if index_type == "serverless" else pod_cost,
            "reads": read_cost if index_type == "serverless" else 0,
            "writes": write_cost if index_type == "serverless" else 0
        }
    }

Part 4: Weaviate Deep Dive

GraphQL & Modules

# vectordb/weaviate_client.py
import weaviate
from weaviate.classes.init import Auth
from weaviate.classes.config import Configure, Property, DataType
from typing: List, Dict, Optional

class WeaviateManager:
    def __init__(
        self,
        url: str = "http://localhost:8080",
        auth_credentials: Auth = None
    ):
        self.client = weaviate.connect_to_custom(
            http_host=url.replace("http://", "").replace("https://", ""),
            http_port=8080,
            http_secure=False,
            auth_credentials=auth_credentials
        )

    def create_collection(
        self,
        name: str,
        properties: List[Property],
        vectorizer_config: Configure.Vectorizer = None,
        vector_index_config: Configure.VectorIndex = None
    ):
        """Create collection with vectorizer module."""

        if vectorizer_config is None:
            vectorizer_config = Configure.Vectorizer.text2vec_openai(
                model="text-embedding-3-small"
            )

        if vector_index_config is None:
            vector_index_config = Configure.VectorIndex.hnsw(
                distance_metric=VectorDistances.COSINE,
                quantizer=Configure.VectorIndex.Quantizer.bq()
            )

        self.client.collections.create(
            name=name,
            properties=properties,
            vectorizer_config=vectorizer_config,
            vector_index_config=vector_index_config
        )

    def hybrid_search(
        self,
        collection_name: str,
        query: str,
        vector: List[float] = None,
        alpha: float = 0.5,  # 0 = keyword, 1 = vector
        limit: int = 10,
        filters = None
    ):
        """Hybrid search with GraphQL-like flexibility."""
        collection = self.client.collections.get(collection_name)

        return collection.query.hybrid(
            query=query,
            vector=vector,
            alpha=alpha,
            limit=limit,
            filters=filters,
            return_metadata=MetadataQuery(
                score=True,
                explain_score=True
            )
        )

    def bm25_search(
        self,
        collection_name: str,
        query: str,
        limit: int = 10,
        properties: List[str] = None
    ):
        """Pure keyword search (BM25)."""
        collection = self.client.collections.get(collection_name)
        return collection.query.bm25(
            query=query,
            limit=limit,
            query_properties=properties
        )

Weaviate Modules

# vectordb/weaviate_modules.py
WEAVIATE_MODULES = {
    "text2vec_openai": {
        "description": "OpenAI embeddings",
        "models": ["text-embedding-3-small", "text-embedding-3-large", "ada-002"],
        "params": ["model", "dimensions", "base_url"]
    },
    "text2vec_cohere": {
        "description": "Cohere embeddings",
        "models": ["embed-english-v3.0", "embed-multilingual-v3.0"],
        "params": ["model", "truncate"]
    },
    "text2vec_huggingface": {
        "description": "Local HuggingFace models",
        "models": ["sentence-transformers/all-MiniLM-L6-v2", "BAAI/bge-large-en-v1.5"],
        "params": ["model", "endpoint"]
    },
    "generative_openai": {
        "description": "RAG generation with OpenAI",
        "models": ["gpt-4o", "gpt-4o-mini", "gpt-3.5-turbo"],
        "params": ["model", "temperature"]
    },
    "generative_cohere": {
        "description": "RAG generation with Cohere",
        "models": ["command-r-plus", "command-r"],
        "params": ["model", "temperature"]
    },
    "reranker_cohere": {
        "description": "Cohere reranking",
        "models": ["rerank-english-v3.0", "rerank-multilingual-v3.0"]
    },
}

# Multi-vector collections
MULTI_VECTOR_CONFIG = {
    "dense": Configure.Vectorizer.text2vec_openai(model="text-embedding-3-small"),
    "sparse": Configure.Vectorizer.text2vec_contextionary()  # BM25-like
}

Part 5: Milvus Deep Dive

Kubernetes-Native Architecture

# vectordb/milvus_client.py
from pymilvus import (
    MilvusClient, DataType, Function, FunctionType,
    AnnSearchRequest, RRFRanker, WeightedRanker
)
from typing: List, Dict, Optional

class MilvusManager:
    def __init__(self, uri: str = "http://localhost:19530", token: str = ""):
        self.client = MilvusClient(uri=uri, token=token)

    def create_collection(
        self,
        name: str,
        dimension: int,
        primary_field: str = "id",
        vector_field: str = "vector",
        enable_dynamic_field: bool = True,
        auto_id: bool = True
    ):
        schema = self.client.create_schema(
            auto_id=auto_id,
            enable_dynamic_field=enable_dynamic_field
        )

        schema.add_field(
            field_name=primary_field,
            datatype=DataType.INT64,
            is_primary=True
        )
        schema.add_field(
            field_name=vector_field,
            datatype=DataType.FLOAT_VECTOR,
            dim=dimension
        )
        schema.add_field(
            field_name="text",
            datatype=DataType.VARCHAR,
            max_length=65535
        )
        schema.add_field(
            field_name="metadata",
            datatype=DataType.JSON
        )

        # Add sparse vector field for hybrid search
        schema.add_field(
            field_name="sparse_vector",
            datatype=DataType.SPARSE_FLOAT_VECTOR
        )

        # Add BM25 function for sparse vectors
        bm25_function = Function(
            name="text_bm25",
            function_type=FunctionType.BM25,
            input_field_names=["text"],
            output_field_names=["sparse_vector"]
        )
        schema.add_function(bm25_function)

        index_params = self.client.prepare_index_params()
        index_params.add_index(
            field_name=vector_field,
            index_type="HNSW",
            metric_type="COSINE",
            params={"M": 16, "efConstruction": 200}
        )
        index_params.add_index(
            field_name="sparse_vector",
            index_type="SPARSE_INVERTED_INDEX",
            metric_type="BM25"
        )

        self.client.create_collection(
            collection_name=name,
            schema=schema,
            index_params=index_params
        )

    def hybrid_search(
        self,
        collection_name: str,
        dense_vector: List[float],
        sparse_query: str,
        limit: int = 10,
        dense_weight: float = 0.7,
        sparse_weight: float = 0.3,
        output_fields: List[str] = None
    ):
        """Hybrid dense + sparse search with re-ranking."""

        dense_req = AnnSearchRequest(
            data=[dense_vector],
            anns_field="vector",
            param={"metric_type": "COSINE", "params": {"ef": 128}},
            limit=limit * 2
        )

        sparse_req = AnnSearchRequest(
            data=[sparse_query],
            anns_field="sparse_vector",
            param={"metric_type": "BM25"},
            limit=limit * 2
        )

        # RRF fusion (default)
        rerank = RRFRanker(k=60)

        return self.client.hybrid_search(
            collection_name=collection_name,
            reqs=[dense_req, sparse_req],
            rerank=rerank,
            limit=limit,
            output_fields=output_fields or ["text", "metadata"]
        )

Milvus Scaling

# vectordb/milvus_scaling.py
MILVUS_SCALING = {
    "components": {
        "proxy": "Stateless, handles requests, auto-scale",
        "query_node": "Executes searches, scale with query load",
        "data_node": "Handles ingestion, scale with write load",
        "index_node": "Builds indexes, batch scale",
        "root_coord": "Metadata, single active",
        "query_coord": "Load balancing, single active",
        "data_coord": "Segment management, single active",
    },
    "deployment": {
        "standalone": "All-in-one, dev/test only",
        "cluster": "Separate components, production",
        "k8s_operator": "Milvus Operator for Kubernetes",
        "helm_chart": "Official Helm chart with all components",
    },
    "partitioning": {
        "partition_key": "Hash-based partitioning for multi-tenancy",
        "shard_num": "Number of shards per collection (default: 1-64)",
    }
}

Part 6: Production Comparison

Benchmark Results (Typical)

# vectordb/benchmarks.py
BENCHMARK_RESULTS = {
    "dataset": "1M vectors, 768-dim, COSINE",
    "queries": 1000,
    "results": {
        "Qdrant": {
            "p50_latency_ms": 12,
            "p99_latency_ms": 45,
            "recall_at_10": 0.98,
            "memory_gb": 3.2,
            "index_build_min": 8
        },
        "Pinecone": {
            "p50_latency_ms": 18,
            "p99_latency_ms": 55,
            "recall_at_10": 0.97,
            "memory_gb": "managed",
            "index_build_min": "managed"
        },
        "Weaviate": {
            "p50_latency_ms": 22,
            "p99_latency_ms": 68,
            "recall_at_10": 0.96,
            "memory_gb": 4.1,
            "index_build_min": 12
        },
        "Milvus": {
            "p50_latency_ms": 15,
            "p99_latency_ms": 52,
            "recall_at_10": 0.98,
            "memory_gb": 3.8,
            "index_build_min": 10
        }
    }
}

Feature Matrix

# vectordb/feature_matrix.py
FEATURE_MATRIX = {
    "Feature": ["Qdrant", "Pinecone", "Weaviate", "Milvus"],
    "Deployment": ["Self-hosted/Cloud", "Managed only", "Self-hosted/Cloud", "Self-hosted/K8s"],
    "License": ["Apache 2.0", "Proprietary", "BSD-3", "Apache 2.0"],
    "Hybrid Search": ["✅ Native", "✅ Sparse+Dense", "✅ Native", "✅ Native"],
    "Filtering": ["✅ Payload indexes", "✅ Metadata filter", "✅ GraphQL where", "✅ Scalar index"],
    "Multi-tenancy": ["✅ Sharding", "✅ Namespaces", "✅ Tenants", "✅ Partition keys"],
    "Backups": ["✅ Snapshots", "✅ Collections", "✅ Backups", "✅ Backups"],
    "Rollback": ["✅ Point-in-time", "❌", "✅", "✅"],
    "GraphQL": ["❌", "❌", "✅ Native", "❌"],
    "REST API": ["✅", "✅", "✅", "✅"],
    "gRPC": ["✅", "✅", "✅", "✅"],
    "Python SDK": ["✅ Excellent", "✅ Good", "✅ Good", "✅ Good"],
    "Rust Core": ["✅", "❌", "❌ (Go)", "❌ (C++)"],
    "Quantization": ["✅ Scalar/Binary", "❌", "✅ BQ/PQ", "✅ PQ/SQ"],
    "DiskANN": ["✅", "❌", "❌", "✅"],
    "RBAC": ["✅ Cloud only", "✅", "✅", "✅"],
    "Cost (1M vecs)": ["~$50/mo self-host", "~$200/mo serverless", "~$80/mo self-host", "~$100/mo K8s"],
}

Part 7: Migration & Interoperability

Universal Client Interface

# vectordb/universal.py
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing: List, Dict, Optional

@dataclass
class VectorRecord:
    id: str
    vector: List[float]
    payload: Dict
    sparse_vector: Dict[int, float] = None

@dataclass
class SearchResult:
    id: str
    score: float
    payload: Dict
    vector: List[float] = None

class VectorDBClient(ABC):
    @abstractmethod
    def create_collection(self, name: str, dimension: int, config: dict): pass

    @abstractmethod
    def upsert(self, collection: str, records: List[VectorRecord]): pass

    @abstractmethod
    def search(
        self,
        collection: str,
        query_vector: List[float],
        limit: int = 10,
        filter: Dict = None
    ) -> List[SearchResult]: pass

    @abstractmethod
    def hybrid_search(
        self,
        collection: str,
        dense_vector: List[float],
        sparse_query: str,
        limit: int = 10
    ) -> List[SearchResult]: pass

    @abstractmethod
    def delete(self, collection: str, ids: List[str]): pass

    @abstractmethod
    def get_stats(self, collection: str) -> dict: pass

# Factory
def create_vector_db(provider: str, **kwargs) -> VectorDBClient:
    providers = {
        "qdrant": QdrantClient,
        "pinecone": PineconeClient,
        "weaviate": WeaviateClient,
        "milvus": MilvusClient,
    }
    return providers[provider.lower()](**kwargs)

Migration Strategy

# vectordb/migration.py
async def migrate_collection(
    source: VectorDBClient,
    target: VectorDBClient,
    collection_name: str,
    batch_size: int = 1000
):
    """Migrate collection between vector databases."""

    # 1. Get source schema/stats
    stats = source.get_stats(collection_name)
    dimension = stats["dimension"]

    # 2. Create target collection
    target.create_collection(collection_name, dimension, {})

    # 3. Stream migrate in batches
    offset = 0
    total = stats["vector_count"]

    while offset < total:
        # Scroll source
        records = source.scroll(collection_name, offset, batch_size)

        # Transform if needed (dimension change, etc.)
        transformed = transform_records(records, dimension)

        # Upsert to target
        target.upsert(collection_name, transformed)

        offset += batch_size
        print(f"Migrated {min(offset, total)}/{total}")

    # 4. Verify
    target_stats = target.get_stats(collection_name)
    assert target_stats["vector_count"] == total

Part 8: Production Checklist

Pre-Production

# vectordb/production_checklist.py
PRODUCTION_CHECKLIST = {
    "capacity_planning": [
        "Estimate vector count growth (3-5x current)",
        "Calculate memory: vectors * dim * 4 bytes * 1.5 (overhead)",
        "Plan for quantization (INT8 = 4x savings)",
        "Reserve 30% headroom for indexing",
    ],
    "index_tuning": [
        "HNSW: m=16, ef_construct=200 for balanced",
        "ef_search=128 for production queries",
        "Enable quantization for >100k vectors",
        "Set appropriate nlist/nprobe for IVF",
    ],
    "reliability": [
        "Enable WAL (write-ahead log)",
        "Configure replication factor >= 2",
        "Set up automated backups (daily + incremental)",
        "Test restore procedure monthly",
        "Monitor disk usage, set alerts at 70%",
    ],
    "security": [
        "Enable TLS for all connections",
        "Use API keys, rotate quarterly",
        "Network policies: restrict to app subnets",
        "Audit logs enabled",
    ],
    "monitoring": [
        "Latency p50/p95/p99 per collection",
        "Query throughput (QPS)",
        "Index build queue depth",
        "Memory/CPU/disk per node",
        "Recall@K sampling (weekly)",
        "Error rates by type",
    ],
    "scaling": [
        "Horizontal pod autoscaler (K8s)",
        "Read replicas for query-heavy workloads",
        "Sharding strategy for >10M vectors",
        "Separate indexing nodes from query nodes",
    ]
}

Decision Framework

VECTOR DATABASE SELECTION:

Team has DevOps capacity?
├─→ No → Pinecone (managed) or Qdrant Cloud
└─→ Yes
    ├─→ Need GraphQL / semantic modules?
    │   └─→ Yes → Weaviate
    │
    ├─→ Kubernetes-native, massive scale (>100M)?
    │   └─→ Yes → Milvus
    │
    ├─→ Rust performance, filtering, hybrid search?
    │   └─→ Yes → Qdrant
    │
    ├─→ Budget sensitive, self-hosted?
    │   └─→ Yes → Qdrant or Milvus
    │
    └─→ Enterprise support required?
        └─→ Yes → Pinecone or Weaviate Cloud

Migration Paths

CURRENT → RECOMMENDED MIGRATION

Elasticsearch/OpenSearch → Qdrant (similar filter model)
pgvector → Qdrant/Milvus (better scaling)
Chroma → Qdrant (compatible API, better perf)
Pinecone → Qdrant Cloud (API compatible, cost savings)
Weaviate → Qdrant (if GraphQL not needed)
Custom FAISS → Milvus (managed FAISS at scale)

Resources

Vector databases are not interchangeable — each has distinct strengths. Benchmark with YOUR data, YOUR queries, and YOUR scale before committing. The "best" database is the one your team can operate reliably.

LLM Evaluation & Benchmarking: Metrics, Frameworks & Production Testing

Phil IPS — Mon, 12 Oct 2026 20:13:41 +0000

Liquid syntax error: Unknown tag 'endraw'

Building AI Agents with Function Calling: Tools, Patterns & Production Systems

Phil IPS — Mon, 12 Oct 2026 20:00:55 +0000

Building AI Agents with Function Calling: Complete Implementation Guide

Build production-ready AI agents that can execute code, call APIs, query databases, and interact with external systems through function calling.

Introduction

Function calling transforms LLMs from text generators into action-taking agents. This guide covers building robust function-calling agents with proper validation, error handling, and production patterns.

Function Calling Architecture

USER REQUEST
     │
     ▼
┌─────────────┐
│   LLM       │ ──▶ Analyzes request, decides function calls
│  Reasoning  │
└─────────────┘
     │
     ▼ (if function needed)
┌─────────────┐
│  Function   │ ──▶ Validates, executes, returns result
│  Executor   │
└─────────────┘
     │
     ▼
┌─────────────┐
│   LLM       │ ──▶ Synthesizes final response
│  Response   │
└─────────────┘

Part 1: Function Definition & Schema

JSON Schema for Functions

# functions/schemas.py
from typing: Dict, List, Any, Optional, get_type_hints
from pydantic import BaseModel, Field, create_model
from dataclasses import dataclass
import inspect
import json

@dataclass
class FunctionSpec:
    name: str
    description: str
    parameters: Dict  # JSON Schema
    required: List[str] = None
    returns: Dict = None
    category: str = "general"
    version: str = "1.0"

def function_to_schema(func: callable) -> FunctionSpec:
    """Auto-generate function schema from Python function."""
    sig = inspect.signature(func)
    type_hints = get_type_hints(func)
    docstring = inspect.getdoc(func) or ""

    # Parse docstring for param descriptions
    param_descriptions = _parse_docstring_params(docstring)

    properties = {}
    required = []

    for name, param in sig.parameters.items():
        if name in ("self", "cls"):
            continue

        param_type = type_hints.get(name, Any)
        json_schema = _python_type_to_json_schema(param_type)
        json_schema["description"] = param_descriptions.get(name, "")

        properties[name] = json_schema

        if param.default == inspect.Parameter.empty:
            required.append(name)

    return FunctionSpec(
        name=func.__name__,
        description=docstring.split("\n")[0] if docstring else "",
        parameters={
            "type": "object",
            "properties": properties,
            "required": required
        },
        required=required,
        returns=_python_type_to_json_schema(sig.return_annotation) if sig.return_annotation != inspect.Signature.empty else {}
    )

def _python_type_to_json_schema(py_type: type) -> Dict:
    """Convert Python type to JSON Schema."""
    origin = getattr(py_type, "__origin__", None)
    args = getattr(py_type, "__args__", ())

    if py_type is str:
        return {"type": "string"}
    elif py_type is int:
        return {"type": "integer"}
    elif py_type is float:
        return {"type": "number"}
    elif py_type is bool:
        return {"type": "boolean"}
    elif origin is list or origin is List:
        return {"type": "array", "items": _python_type_to_json_schema(args[0]) if args else {}}
    elif origin is dict or origin is Dict:
        return {"type": "object", "additionalProperties": _python_type_to_json_schema(args[1]) if len(args) > 1 else {}}
    elif origin is Optional or origin is Union:
        # Handle Optional[T] = Union[T, None]
        non_none = [a for a in args if a is not type(None)]
        if non_none:
            return _python_type_to_json_schema(non_none[0])
        return {"type": "null"}
    elif hasattr(py_type, "__annotations__"):  # Pydantic model
        return py_type.model_json_schema()
    else:
        return {"type": "string"}  # fallback

# Example functions with rich schemas
class FunctionLibrary:
    @staticmethod
    def execute_python(code: str, timeout: int = 30) -> Dict:
        """Execute Python code in sandboxed environment.

        Args:
            code: Python code to execute
            timeout: Maximum execution time in seconds
        """
        pass

    @staticmethod
    def http_request(
        method: str,
        url: str,
        headers: Dict = None,
        body: str = None,
        timeout: int = 30
    ) -> Dict:
        """Make HTTP request to external API.

        Args:
            method: HTTP method (GET, POST, PUT, DELETE, etc.)
            url: Target URL
            headers: Optional request headers
            body: Optional request body
            timeout: Request timeout in seconds
        """
        pass

    @staticmethod
    def query_database(sql: str, params: List = None) -> Dict:
        """Execute read-only SQL query.

        Args:
            sql: SELECT query to execute
            params: Query parameters for prepared statement
        """
        pass

    @staticmethod
    def search_web(query: str, max_results: int = 10) -> Dict:
        """Search the web for information.

        Args:
            query: Search query
            max_results: Maximum number of results to return
        """
        pass

    @staticmethod
    def file_operation(
        operation: str,
        path: str,
        content: str = None
    ) -> Dict:
        """Perform file operations.

        Args:
            operation: read, write, append, delete, list
            path: File or directory path
            content: Content for write/append operations
        """
        pass

Function Registry

# functions/registry.py
from typing: Dict, List, Callable, Any
from dataclasses import dataclass, field
import json

@dataclass
class RegisteredFunction:
    spec: FunctionSpec
    handler: Callable
    permissions: List[str] = field(default_factory=list)  # required permissions
    rate_limit: int = 100  # calls per minute
    timeout: int = 30

class FunctionRegistry:
    def __init__(self):
        self.functions: Dict[str, RegisteredFunction] = {}
        self.categories: Dict[str, List[str]] = {}

    def register(
        self,
        func: Callable,
        spec: FunctionSpec = None,
        permissions: List[str] = None,
        rate_limit: int = 100,
        timeout: int = 30
    ):
        """Register a function for LLM use."""
        if spec is None:
            spec = function_to_schema(func)

        registered = RegisteredFunction(
            spec=spec,
            handler=func,
            permissions=permissions or [],
            rate_limit=rate_limit,
            timeout=timeout
        )

        self.functions[spec.name] = registered

        if spec.category not in self.categories:
            self.categories[spec.category] = []
        self.categories[spec.category].append(spec.name)

    def get_schema(self, name: str) -> Dict:
        """Get OpenAI-compatible function schema."""
        if name not in self.functions:
            raise ValueError(f"Function {name} not registered")

        spec = self.functions[name].spec
        return {
            "name": spec.name,
            "description": spec.description,
            "parameters": spec.parameters
        }

    def get_all_schemas(self, category: str = None) -> List[Dict]:
        """Get all function schemas, optionally filtered by category."""
        names = self.categories.get(category, []) if category else list(self.functions.keys())
        return [self.get_schema(name) for name in names]

    def call(self, name: str, arguments: Dict) -> Any:
        """Execute a registered function."""
        if name not in self.functions:
            raise ValueError(f"Function {name} not registered")

        registered = self.functions[name]

        # Validate arguments against schema
        # (Would use jsonschema.validate in production)

        # Execute
        return registered.handler(**arguments)

    async def acall(self, name: str, arguments: Dict) -> Any:
        """Async execute a registered function."""
        if name not in self.functions:
            raise ValueError(f"Function {name} not registered")

        registered = self.functions[name]

        if asyncio.iscoroutinefunction(registered.handler):
            return await registered.handler(**arguments)
        else:
            # Run sync function in executor
            loop = asyncio.get_event_loop()
            return await loop.run_in_executor(None, registered.handler, **arguments)

# Global registry
registry = FunctionRegistry()

# Register built-in functions
registry.register(FunctionLibrary.execute_python, category="code")
registry.register(FunctionLibrary.http_request, category="network")
registry.register(FunctionLibrary.query_database, category="data")
registry.register(FunctionLibrary.search_web, category="research")
registry.register(FunctionLibrary.file_operation, category="filesystem")

Part 2: Agent Implementation

Core Agent Loop

# agents/function_agent.py
from typing: Dict, List, Any, Optional
from dataclasses import dataclass, field
from openai import AsyncOpenAI
import asyncio
import json

@dataclass
class FunctionCall:
    name: str
    arguments: Dict
    call_id: str

@dataclass
class AgentResponse:
    content: str
    function_calls: List[FunctionCall] = field(default_factory=list)
    finish_reason: str = "stop"
    usage: Dict = field(default_factory=dict)

class FunctionCallingAgent:
    def __init__(
        self,
        client: AsyncOpenAI,
        model: str = "gpt-4o",
        registry: FunctionRegistry = None,
        system_prompt: str = None,
        max_iterations: int = 10,
        temperature: float = 0.1
    ):
        self.client = client
        self.model = model
        self.registry = registry or FunctionRegistry()
        self.system_prompt = system_prompt or self._default_system_prompt()
        self.max_iterations = max_iterations
        self.temperature = temperature
        self.conversation_history: List[Dict] = []

    def _default_system_prompt(self) -> str:
        return """You are an AI agent with access to various tools. 
        Use tools when needed to answer questions or complete tasks.
        Think step by step. If you need to use a tool, call it.
        Always provide a final answer to the user."""

    def _build_messages(self, user_message: str) -> List[Dict]:
        messages = [{"role": "system", "content": self.system_prompt}]
        messages.extend(self.conversation_history)
        messages.append({"role": "user", "content": user_message})
        return messages

    def _get_function_schemas(self) -> List[Dict]:
        return self.registry.get_all_schemas()

    async def run(self, user_message: str) -> AgentResponse:
        """Run agent loop until completion."""
        messages = self._build_messages(user_message)

        for iteration in range(self.max_iterations):
            # Call LLM with function definitions
            response = await self.client.chat.completions.create(
                model=self.model,
                messages=messages,
                tools=[{"type": "function", "function": schema} for schema in self._get_function_schemas()],
                tool_choice="auto",
                temperature=self.temperature,
                max_tokens=4000
            )

            message = response.choices[0].message

            # Add assistant message to history
            messages.append({
                "role": "assistant",
                "content": message.content,
                "tool_calls": message.tool_calls
            })

            # Check for function calls
            if not message.tool_calls:
                # No function calls - we're done
                self.conversation_history.append({"role": "user", "content": user_message})
                self.conversation_history.append({"role": "assistant", "content": message.content})

                return AgentResponse(
                    content=message.content or "",
                    finish_reason=response.choices[0].finish_reason,
                    usage=response.usage.model_dump() if response.usage else {}
                )

            # Execute function calls
            function_calls = []
            for tool_call in message.tool_calls:
                call = FunctionCall(
                    name=tool_call.function.name,
                    arguments=json.loads(tool_call.function.arguments),
                    call_id=tool_call.id
                )
                function_calls.append(call)

                # Execute function
                try:
                    result = await self.registry.acall(call.name, call.arguments)
                    result_str = json.dumps(result, default=str)
                except Exception as e:
                    result_str = json.dumps({"error": str(e)})

                # Add function result to messages
                messages.append({
                    "role": "tool",
                    "tool_call_id": call.call_id,
                    "content": result_str
                })

            # Continue loop for LLM to process results

        # Max iterations reached
        return AgentResponse(
            content="Max iterations reached without completion.",
            finish_reason="length",
            function_calls=function_calls
        )

    def clear_history(self):
        self.conversation_history = []

Streaming Agent

# agents/streaming_agent.py
from typing: AsyncGenerator, Dict, List, Any
from openai import AsyncOpenAI
import json

class StreamingFunctionAgent:
    def __init__(self, client: AsyncOpenAI, model: str = "gpt-4o", registry=None):
        self.client = client
        self.model = model
        self.registry = registry or FunctionRegistry()

    async def run_stream(self, user_message: str) -> AsyncGenerator[str, None]:
        """Stream agent execution with function calls."""
        messages = [
            {"role": "system", "content": self._default_system_prompt()},
            {"role": "user", "content": user_message}
        ]

        while True:
            stream = await self.client.chat.completions.create(
                model=self.model,
                messages=messages,
                tools=[{"type": "function", "function": schema} for schema in self.registry.get_all_schemas()],
                tool_choice="auto",
                temperature=0.1,
                stream=True
            )

            function_calls = []
            current_content = ""
            current_tool_call = None

            async for chunk in stream:
                delta = chunk.choices[0].delta

                if delta.content:
                    current_content += delta.content
                    yield delta.content

                if delta.tool_calls:
                    for tc in delta.tool_calls:
                        if tc.index >= len(function_calls):
                            function_calls.append({
                                "id": tc.id,
                                "name": tc.function.name,
                                "arguments": ""
                            })
                            current_tool_call = function_calls[-1]
                        else:
                            current_tool_call = function_calls[tc.index]

                        if tc.function.name:
                            current_tool_call["name"] = tc.function.name
                        if tc.function.arguments:
                            current_tool_call["arguments"] += tc.function.arguments

            # Add assistant message
            messages.append({
                "role": "assistant",
                "content": current_content,
                "tool_calls": [
                    {"id": fc["id"], "type": "function", "function": {"name": fc["name"], "arguments": fc["arguments"]}}
                    for fc in function_calls
                ] if function_calls else None
            })

            if not function_calls:
                yield "\n[Done]"
                break

            # Execute functions and yield results
            for fc in function_calls:
                yield f"\n[Calling {fc['name']}...]"
                try:
                    result = await self.registry.acall(fc["name"], json.loads(fc["arguments"]))
                    result_str = json.dumps(result, default=str)
                except Exception as e:
                    result_str = json.dumps({"error": str(e)})

                messages.append({
                    "role": "tool",
                    "tool_call_id": fc["id"],
                    "content": result_str
                })
                yield f" [Result: {result_str[:100]}...]"

Part 3: Validation & Safety

Input Validation

# functions/validation.py
from typing: Dict, Any, List
from pydantic import BaseModel, ValidationError
import jsonschema

class FunctionValidator:
    def __init__(self, registry: FunctionRegistry):
        self.registry = registry

    def validate_call(self, name: str, arguments: Dict) -> tuple[bool, List[str]]:
        """Validate function call arguments against schema."""
        if name not in self.registry.functions:
            return False, [f"Unknown function: {name}"]

        spec = self.registry.functions[name].spec
        schema = spec.parameters

        try:
            jsonschema.validate(arguments, schema)
            return True, []
        except jsonschema.ValidationError as e:
            return False, [f"Validation error: {e.message} at {'.'.join(str(x) for x in e.path)}"]

    def sanitize_arguments(self, name: str, arguments: Dict) -> Dict:
        """Sanitize arguments (remove extra fields, apply defaults)."""
        if name not in self.registry.functions:
            return arguments

        spec = self.registry.functions[name].spec
        properties = spec.parameters.get("properties", {})

        # Keep only defined properties
        sanitized = {}
        for key, value in arguments.items():
            if key in properties:
                sanitized[key] = value

        # Apply defaults for missing required fields
        for req in spec.parameters.get("required", []):
            if req not in sanitized and req in properties:
                prop = properties[req]
                if "default" in prop:
                    sanitized[req] = prop["default"]

        return sanitized

Output Validation

# functions/output_validation.py
from typing: Dict, Any, Callable
from pydantic import BaseModel, validator
import json

class OutputValidator:
    def __init__(self):
        self.validators: Dict[str, Callable] = {}

    def register_validator(self, function_name: str, validator_fn: Callable[[Any], bool]):
        """Register custom validator for function output."""
        self.validators[function_name] = validator_fn

    def validate_output(self, function_name: str, output: Any) -> tuple[bool, str]:
        """Validate function output."""
        if function_name not in self.validators:
            return True, ""

        try:
            result = self.validators[function_name](output)
            if result:
                return True, ""
            return False, f"Output validation failed for {function_name}"
        except Exception as e:
            return False, f"Validation error: {e}"

# Example validators
def validate_code_execution(output: Dict) -> bool:
    """Validate code execution output."""
    if not isinstance(output, dict):
        return False
    if "success" not in output:
        return False
    if output["success"] and "result" not in output:
        return False
    return True

def validate_http_response(output: Dict) -> bool:
    """Validate HTTP response."""
    if not isinstance(output, dict):
        return False
    if "status_code" not in output:
        return False
    return True

validator = OutputValidator()
validator.register_validator("execute_python", validate_code_execution)
validator.register_validator("http_request", validate_http_response)

Security Sandbox

# functions/sandbox.py
"""Secure function execution sandbox."""

import subprocess
import tempfile
import os
import resource
import signal
from typing: Dict, Any
from dataclasses import dataclass

@dataclass
class SandboxConfig:
    max_cpu_seconds: int = 30
    max_memory_mb: int = 512
    max_processes: int = 10
    allowed_modules: List[str] = None
    blocked_modules: List[str] = field(default_factory=lambda: [
        "os", "sys", "subprocess", "shutil", "pathlib",
        "importlib", "pkgutil", "modulefinder", "runpy",
        "socket", "urllib", "http", "ftplib", "telnetlib",
        "sqlite3", "pickle", "marshal", "shelve"
    ])

class PythonSandbox:
    def __init__(self, config: SandboxConfig = None):
        self.config = config or SandboxConfig()

    def execute(self, code: str, timeout: int = 30) -> Dict:
        """Execute Python code in sandboxed subprocess."""

        # Write code to temp file
        with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
            # Add safety wrapper
            wrapped_code = self._wrap_code(code)
            f.write(wrapped_code)
            temp_path = f.name

        try:
            # Set resource limits
            def set_limits():
                resource.setrlimit(resource.RLIMIT_CPU, (self.config.max_cpu_seconds, self.config.max_cpu_seconds))
                resource.setrlimit(resource.RLIMIT_AS, (self.config.max_memory_mb * 1024 * 1024, self.config.max_memory_mb * 1024 * 1024))
                resource.setrlimit(resource.RLIMIT_NPROC, (self.config.max_processes, self.config.max_processes))

            # Run with timeout
            result = subprocess.run(
                ["python", temp_path],
                capture_output=True,
                text=True,
                timeout=timeout,
                preexec_fn=set_limits if os.name != 'nt' else None
            )

            return {
                "success": result.returncode == 0,
                "stdout": result.stdout,
                "stderr": result.stderr,
                "return_code": result.returncode
            }

        except subprocess.TimeoutExpired:
            return {"success": False, "error": "Execution timeout"}
        except Exception as e:
            return {"success": False, "error": str(e)}
        finally:
            try:
                os.unlink(temp_path)
            except:
                pass

    def _wrap_code(self, code: str) -> str:
        """Wrap user code with safety checks."""
        return f"""
import sys
import json

# Redirect stdout/stderr
import io
old_stdout = sys.stdout
old_stderr = sys.stderr
sys.stdout = io.StringIO()
sys.stderr = io.StringIO()

result = {{"success": False, "result": None, "stdout": "", "stderr": ""}}

try:
{self._indent(code, 8)}
    result["success"] = True
    result["result"] = locals().get("result", None)
except Exception as e:
    result["success"] = False
    result["error"] = str(e)
    import traceback
    result["traceback"] = traceback.format_exc()
finally:
    result["stdout"] = sys.stdout.getvalue()
    result["stderr"] = sys.stderr.getvalue()
    sys.stdout = old_stdout
    sys.stderr = old_stderr
    print(json.dumps(result))
"""

    def _indent(self, text: str, spaces: int) -> str:
        return "\n".join(" " * spaces + line for line in text.split("\n"))

Part 4: Advanced Patterns

Parallel Function Execution

# functions/parallel.py
from typing: List, Dict, Any
import asyncio

class ParallelFunctionExecutor:
    def __init__(self, registry: FunctionRegistry, max_concurrent: int = 5):
        self.registry = registry
        self.semaphore = asyncio.Semaphore(max_concurrent)

    async def execute_batch(self, calls: List[Dict]) -> List[Any]:
        """Execute multiple function calls in parallel."""
        async def execute_one(call):
            async with self.semaphore:
                try:
                    return await self.registry.acall(call["name"], call["arguments"])
                except Exception as e:
                    return {"error": str(e), "call": call}

        tasks = [execute_one(call) for call in calls]
        return await asyncio.gather(*tasks)

    async def execute_with_dependencies(self, calls: List[Dict]) -> List[Any]:
        """Execute calls respecting dependencies."""
        # calls format: {"name": "", "arguments": {}, "depends_on": [call_ids]}
        completed = {}
        remaining = {call.get("id", i): call for i, call in enumerate(calls)}

        while remaining:
            # Find callable tasks (no unmet dependencies)
            callable_tasks = []
            for call_id, call in remaining.items():
                deps = call.get("depends_on", [])
                if all(d in completed for d in deps):
                    callable_tasks.append((call_id, call))

            if not callable_tasks:
                # Circular dependency or missing dependency
                raise ValueError("Unresolvable dependencies")

            # Execute callable tasks in parallel
            tasks = []
            for call_id, call in callable_tasks:
                task = self._execute_with_deps(call_id, call, completed)
                tasks.append(task)

            results = await asyncio.gather(*tasks)

            for (call_id, _), result in zip(callable_tasks, results):
                completed[call_id] = result
                del remaining[call_id]

        return [completed.get(call.get("id", i)) for i, call in enumerate(calls)]

    async def _execute_with_deps(self, call_id: str, call: Dict, completed: Dict) -> Any:
        # Substitute dependency results into arguments
        args = call["arguments"].copy()
        for key, value in args.items():
            if isinstance(value, str) and value.startswith("$ref:"):
                dep_id = value[5:]
                if dep_id in completed:
                    args[key] = completed[dep_id]

        async with self.semaphore:
            return await self.registry.acall(call["name"], args)

Function Composition

# functions/composition.py
from typing: Dict, Any, List, Callable
from dataclasses import dataclass

@dataclass
class FunctionPipeline:
    """Chain functions together: output of one feeds into next."""
    steps: List[Dict]  # {"name": "", "arguments": {}, "output_to": "next_arg"}

class FunctionComposer:
    def __init__(self, registry: FunctionRegistry):
        self.registry = registry

    async def compose(self, pipeline: FunctionPipeline, initial_input: Dict) -> Dict:
        """Execute function pipeline."""
        context = initial_input.copy()

        for step in pipeline.steps:
            # Merge context into arguments
            args = {**step.get("arguments", {}), **context}

            # Execute
            result = await self.registry.acall(step["name"], args)

            # Store result for next steps
            if "output_to" in step:
                context[step["output_to"]] = result
            else:
                context[step["name"]] = result

            context["last_result"] = result

        return context

# Example: Research -> Summarize -> Save
research_pipeline = FunctionPipeline(steps=[
    {"name": "search_web", "arguments": {"query": "AI agents 2024"}, "output_to": "search_results"},
    {"name": "execute_python", "arguments": {
        "code": """
# Process search results
result = [r for r in search_results if '2024' in r.get('snippet', '')]
"""
    }, "output_to": "filtered_results"},
    {"name": "file_operation", "arguments": {
        "operation": "write",
        "path": "research_summary.json",
        "content": "{{filtered_results}}"
    }}
])

Part 5: Observability

Function Call Tracing

# functions/observability.py
from dataclasses import dataclass, field
from typing: Dict, Any, List
from datetime import datetime
import time
import uuid

@dataclass
class FunctionTrace:
    trace_id: str
    function_name: str
    arguments: Dict
    start_time: float
    end_time: float = 0
    duration_ms: float = 0
    success: bool = False
    result: Any = None
    error: str = ""
    metadata: Dict = field(default_factory=dict)

class FunctionTracer:
    def __init__(self):
        self.traces: List[FunctionTrace] = []
        self.active_traces: Dict[str, FunctionTrace] = {}

    def start_trace(self, function_name: str, arguments: Dict, trace_id: str = None) -> str:
        trace_id = trace_id or str(uuid.uuid4())[:8]
        trace = FunctionTrace(
            trace_id=trace_id,
            function_name=function_name,
            arguments=arguments,
            start_time=time.time()
        )
        self.active_traces[trace_id] = trace
        return trace_id

    def end_trace(self, trace_id: str, success: bool, result: Any = None, error: str = ""):
        if trace_id not in self.active_traces:
            return

        trace = self.active_traces[trace_id]
        trace.end_time = time.time()
        trace.duration_ms = (trace.end_time - trace.start_time) * 1000
        trace.success = success
        trace.result = result
        trace.error = error

        self.traces.append(trace)
        del self.active_traces[trace_id]

    def get_stats(self, function_name: str = None, window_seconds: int = 300) -> Dict:
        cutoff = time.time() - window_seconds
        recent = [t for t in self.traces if t.end_time > cutoff]

        if function_name:
            recent = [t for t in recent if t.function_name == function_name]

        if not recent:
            return {}

        durations = [t.duration_ms for t in recent]
        successes = [t for t in recent if t.success]

        return {
            "function": function_name or "all",
            "total_calls": len(recent),
            "success_rate": len(successes) / len(recent),
            "avg_duration_ms": sum(durations) / len(durations),
            "p50_duration_ms": sorted(durations)[len(durations) // 2],
            "p95_duration_ms": sorted(durations)[int(len(durations) * 0.95)],
            "p99_duration_ms": sorted(durations)[int(len(durations) * 0.99)],
            "errors": [t.error for t in recent if not t.success]
        }

# Global tracer
tracer = FunctionTracer()

# Decorator for automatic tracing
def trace_function(func_name: str = None):
    def decorator(handler):
        async def wrapper(*args, **kwargs):
            name = func_name or handler.__name__
            trace_id = tracer.start_trace(name, {"args": str(args), "kwargs": str(kwargs)})
            try:
                result = await handler(*args, **kwargs)
                tracer.end_trace(trace_id, True, result)
                return result
            except Exception as e:
                tracer.end_trace(trace_id, False, error=str(e))
                raise
        return wrapper
    return decorator

Part 6: Testing

Unit Testing Functions

# tests/test_functions.py
import pytest
from functions.registry import FunctionRegistry, registry
from functions.schemas import function_to_schema
from functions.validation import FunctionValidator

class TestFunctionRegistry:
    def test_register_function(self):
        reg = FunctionRegistry()

        def add(a: int, b: int) -> int:
            """Add two numbers."""
            return a + b

        reg.register(add, category="math")

        assert "add" in reg.functions
        schema = reg.get_schema("add")
        assert schema["name"] == "add"
        assert "a" in schema["parameters"]["properties"]
        assert "b" in schema["parameters"]["properties"]

    def test_call_function(self):
        reg = FunctionRegistry()

        def multiply(x: int, y: int) -> int:
            return x * y

        reg.register(multiply)
        result = reg.call("multiply", {"x": 5, "y": 3})
        assert result == 15

class TestFunctionValidator:
    def test_valid_arguments(self):
        validator = FunctionValidator(registry)
        valid, errors = validator.validate_call("execute_python", {"code": "print('hello')", "timeout": 30})
        assert valid
        assert len(errors) == 0

    def test_invalid_arguments(self):
        validator = FunctionValidator(registry)
        valid, errors = validator.validate_call("execute_python", {"code": 123})  # wrong type
        assert not valid
        assert len(errors) > 0

    def test_sanitize_arguments(self):
        validator = FunctionValidator(registry)
        sanitized = validator.sanitize_arguments("execute_python", {"code": "x=1", "timeout": 30, "extra": "ignored"})
        assert "code" in sanitized
        assert "timeout" in sanitized
        assert "extra" not in sanitized

class TestSandbox:
    def test_safe_execution(self):
        from functions.sandbox import PythonSandbox
        sandbox = PythonSandbox()
        result = sandbox.execute("result = 2 + 2")
        assert result["success"]
        assert result["result"] == 4

    def test_blocked_modules(self):
        from functions.sandbox import PythonSandbox
        sandbox = PythonSandbox()
        result = sandbox.execute("import os; os.system('ls')")
        assert not result["success"]
        assert "error" in result or "blocked" in result.get("stderr", "").lower()

Part 7: Production Deployment

Kubernetes Deployment

# k8s/function-agent.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: function-calling-agent
spec:
  replicas: 3
  selector:
    matchLabels:
      app: function-agent
  template:
    metadata:
      labels:
        app: function-agent
    spec:
      containers:
      - name: agent
        image: myorg/function-agent:v1.0
        env:
        - name: OPENAI_API_KEY
          valueFrom:
            secretKeyRef:
              name: ai-secrets
              key: openai-key
        - name: FUNCTION_REGISTRY
          value: "builtin"
        - name: SANDBOX_ENABLED
          value: "true"
        - name: MAX_ITERATIONS
          value: "10"
        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: function-agent
spec:
  selector:
    app: function-agent
  ports:
  - port: 80
    targetPort: 8000

FastAPI Server

# server.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing: List, Optional
import asyncio

app = FastAPI(title="Function Calling Agent API")

class ChatRequest(BaseModel):
    message: str
    conversation_id: Optional[str] = None
    model: Optional[str] = "gpt-4o"
    temperature: Optional[float] = 0.1
    max_iterations: Optional[int] = 10

class ChatResponse(BaseModel):
    response: str
    conversation_id: str
    function_calls: List[Dict] = []
    usage: Dict = {}

# Initialize agent
agent = FunctionCallingAgent(
    client=AsyncOpenAI(api_key=os.getenv("OPENAI_API_KEY")),
    model="gpt-4o"
)

@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
    try:
        result = await agent.run(request.message)
        return ChatResponse(
            response=result.content,
            conversation_id=request.conversation_id or str(uuid.uuid4()),
            function_calls=[{"name": fc.name, "arguments": fc.arguments} for fc in result.function_calls],
            usage=result.usage
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.post("/chat/stream")
async def chat_stream(request: ChatRequest):
    from fastapi.responses import StreamingResponse

    async def generate():
        streaming_agent = StreamingFunctionAgent(
            client=agent.client,
            model=request.model,
            registry=agent.registry
        )
        async for chunk in streaming_agent.run_stream(request.message):
            yield f"data: {json.dumps({'chunk': chunk})}\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")

@app.get("/functions")
async def list_functions(category: Optional[str] = None):
    return {"functions": agent.registry.get_all_schemas(category)}

@app.get("/health")
async def health():
    return {"status": "healthy"}

@app.get("/metrics")
async def metrics():
    return {"traces": tracer.get_stats()}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Part 8: Best Practices

Function Design Guidelines

FUNCTION DESIGN PRINCIPLES:

1. SINGLE RESPONSIBILITY
   - Each function does ONE thing well
   - Clear, descriptive name
   - Minimal parameters

2. IDEMPOTENCY
   - Same inputs = same outputs
   - Safe to retry
   - No side effects for read operations

3. EXPLICIT CONTRACTS
   - Full JSON Schema with descriptions
   - Document all error cases
   - Version your functions

4. FAILURE HANDLING
   - Return structured errors
   - Include retry guidance
   - Log for debugging

5. SECURITY
   - Validate all inputs
   - Sandbox code execution
   - Rate limit sensitive operations
   - Audit logging

6. OBSERVABILITY
   - Trace every call
   - Metrics on latency, errors
   - Alert on anomalies

Anti-Patterns to Avoid

❌ OVERLOADING FUNCTIONS
   - One function with 20 parameters
   - Multiple operations via "mode" parameter

❌ IMPLICIT STATE
   - Functions that depend on hidden context
   - Global variables between calls

❌ UNBOUNDED EXECUTION
   - No timeout on external calls
   - No rate limiting
   - No circuit breakers

❌ POOR ERROR HANDLING
   - Generic "error" responses
   - No retry guidance
   - Swallowing exceptions

❌ MISSING VALIDATION
   - Trusting LLM arguments blindly
   - No schema validation
   - No output verification

Resources

Function calling is the bridge between LLMs and the real world. Design your functions like APIs: clear contracts, robust validation, comprehensive observability. The agent is only as reliable as its weakest function.

API Mocking for AI Applications: Complete Guide to APIMock-AI Patterns

Phil IPS — Mon, 12 Oct 2026 19:56:11 +0000

API Mocking for AI Applications: Complete Guide to APIMock-AI Patterns

Build robust AI applications with reliable API mocking — contract testing, chaos engineering, and production-grade mock servers.

Introduction

API mocking is critical for AI applications that depend on external LLM APIs, vector databases, and third-party services. This guide covers production mocking patterns using APIMock-AI principles.

Why API Mocking Matters for AI

WITHOUT MOCKS                    WITH MOCKS
─────────────────────            ─────────────────────
Flaky CI/CD                      Reliable tests
Rate limit failures              Controlled responses
Costly API calls                 Zero-cost development
No offline development           Full offline workflow
Slow feedback loops              Instant feedback
Vendor lock-in risk              Portable test suites

Part 1: Mock Server Architecture

Core Mock Server

# mocks/server.py
from dataclasses import dataclass, field
from typing: Dict, List, Optional, Callable, Any
from enum import Enum
import json
import asyncio
from aiohttp import web
import uuid

class MockStrategy(Enum):
    STATIC = "static"                    # Fixed response
    SEQUENCE = "sequence"                # Rotating responses
    TEMPLATE = "template"                # Jinja2 template
    CALLBACK = "callback"                # Dynamic function
    PROXY = "proxy"                      # Forward to real API
    CHAOS = "chaos"                      # Inject failures

@dataclass
class MockRule:
    id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
    method: str
    path_pattern: str
    strategy: MockStrategy = MockStrategy.STATIC
    response: Any = None
    sequence: List[Any] = field(default_factory=list)
    template: str = ""
    callback: Optional[Callable] = None
    proxy_target: str = ""
    chaos_config: Dict = field(default_factory=dict)
    conditions: Dict = field(default_factory=dict)  # headers, query, body
    priority: int = 0
    times_matched: int = 0
    active: bool = True

class MockServer:
    def __init__(self, host: str = "localhost", port: int = 8080):
        self.host = host
        self.port = port
        self.rules: List[MockRule] = []
        self.request_log: List[Dict] = []
        self.app = web.Application()
        self.app.router.add_route("*", "/{path:.*}", self.handle_request)

    def add_rule(self, rule: MockRule):
        self.rules.append(rule)
        self.rules.sort(key=lambda r: -r.priority)

    def add_static(self, method: str, path: str, response: Any, status: int = 200, headers: Dict = None):
        rule = MockRule(
            method=method.upper(),
            path_pattern=path,
            strategy=MockStrategy.STATIC,
            response={"status": status, "body": response, "headers": headers or {}}
        )
        self.add_rule(rule)

    def add_sequence(self, method: str, path: str, responses: List[Any]):
        rule = MockRule(
            method=method.upper(),
            path_pattern=path,
            strategy=MockStrategy.SEQUENCE,
            sequence=responses
        )
        self.add_rule(rule)

    def add_callback(self, method: str, path: str, callback: Callable):
        rule = MockRule(
            method=method.upper(),
            path_pattern=path,
            strategy=MockStrategy.CALLBACK,
            callback=callback
        )
        self.add_rule(rule)

    def add_chaos(self, method: str, path: str, chaos_config: Dict):
        """Inject failures: latency, errors, corruption."""
        rule = MockRule(
            method=method.upper(),
            path_pattern=path,
            strategy=MockStrategy.CHAOS,
            chaos_config=chaos_config
        )
        self.add_rule(rule)

    async def handle_request(self, request: web.Request):
        # Log request
        body = await request.read()
        request_data = {
            "id": str(uuid.uuid4())[:8],
            "method": request.method,
            "path": request.path,
            "headers": dict(request.headers),
            "query": dict(request.query),
            "body": body.decode() if body else "",
            "timestamp": asyncio.get_event_loop().time()
        }
        self.request_log.append(request_data)

        # Find matching rule
        rule = self._match_rule(request)

        if not rule:
            return web.Response(status=404, text="No mock rule matched")

        rule.times_matched += 1

        # Execute strategy
        return await self._execute_strategy(rule, request_data)

    def _match_rule(self, request: web.Request) -> Optional[MockRule]:
        for rule in self.rules:
            if not rule.active:
                continue
            if rule.method != request.method and rule.method != "*":
                continue
            if not self._match_path(rule.path_pattern, request.path):
                continue
            if not self._match_conditions(rule.conditions, request):
                continue
            return rule
        return None

    def _match_path(self, pattern: str, path: str) -> bool:
        # Simple glob matching
        import fnmatch
        return fnmatch.fnmatch(path, pattern)

    def _match_conditions(self, conditions: Dict, request: web.Request) -> bool:
        if "headers" in conditions:
            for k, v in conditions["headers"].items():
                if request.headers.get(k) != v:
                    return False
        if "query" in conditions:
            for k, v in conditions["query"].items():
                if request.query.get(k) != v:
                    return False
        return True

    async def _execute_strategy(self, rule: MockRule, request_data: Dict) -> web.Response:
        if rule.strategy == MockStrategy.STATIC:
            resp = rule.response
            return web.Response(
                status=resp.get("status", 200),
                body=json.dumps(resp["body"]) if isinstance(resp["body"], dict) else str(resp["body"]),
                headers=resp.get("headers", {"Content-Type": "application/json"})
            )

        elif rule.strategy == MockStrategy.SEQUENCE:
            idx = (rule.times_matched - 1) % len(rule.sequence)
            resp = rule.sequence[idx]
            return web.Response(
                status=resp.get("status", 200),
                body=json.dumps(resp["body"]) if isinstance(resp["body"], dict) else str(resp["body"]),
                headers=resp.get("headers", {"Content-Type": "application/json"})
            )

        elif rule.strategy == MockStrategy.CALLBACK:
            if asyncio.iscoroutinefunction(rule.callback):
                return await rule.callback(request_data)
            return rule.callback(request_data)

        elif rule.strategy == MockStrategy.CHAOS:
            return await self._apply_chaos(rule.chaos_config)

        return web.Response(status=500, text="Unknown strategy")

    async def _apply_chaos(self, config: Dict) -> web.Response:
        import random

        # Latency injection
        if "latency_ms" in config:
            await asyncio.sleep(config["latency_ms"] / 1000)

        # Error injection
        if "error_rate" in config and random.random() < config["error_rate"]:
            error_type = config.get("error_type", "500")
            if error_type == "timeout":
                raise asyncio.TimeoutError()
            return web.Response(status=int(error_type), text="Chaos error")

        # Response corruption
        if "corrupt" in config and random.random() < config["corrupt"]:
            return web.Response(
                status=200,
                body="{ invalid json",
                headers={"Content-Type": "application/json"}
            )

        # Default success
        return web.Response(
            status=200,
            body=json.dumps({"mock": True, "chaos": "applied"}),
            headers={"Content-Type": "application/json"}
        )

    async def start(self):
        runner = web.AppRunner(self.app)
        await runner.setup()
        site = web.TCPSite(runner, self.host, self.port)
        await site.start()
        print(f"Mock server running on http://{self.host}:{self.port}")
        return runner

    def get_request_log(self) -> List[Dict]:
        return self.request_log

    def reset_log(self):
        self.request_log.clear()

    def verify_call(self, method: str, path: str, times: int = 1) -> bool:
        count = sum(1 for r in self.request_log 
                   if r["method"] == method.upper() and self._match_path(path, r["path"]))
        return count == times

LLM API Mocking

# mocks/llm_mocks.py
from mocks.server import MockServer, MockStrategy, MockRule
from typing: Dict, List, Any
import json
import random

class LLMMockBuilder:
    """Build realistic LLM API mocks."""

    @staticmethod
    def openai_chat_completions(mock_server: MockServer, responses: List[Dict] = None):
        """Mock OpenAI /v1/chat/completions endpoint."""

        default_responses = [
            {
                "id": "chatcmpl-mock1",
                "object": "chat.completion",
                "created": 1699999999,
                "model": "gpt-4o-mini",
                "choices": [{
                    "index": 0,
                    "message": {
                        "role": "assistant",
                        "content": "This is a mocked response from APIMock-AI."
                    },
                    "finish_reason": "stop"
                }],
                "usage": {
                    "prompt_tokens": 10,
                    "completion_tokens": 20,
                    "total_tokens": 30
                }
            }
        ]

        mock_server.add_rule(MockRule(
            method="POST",
            path_pattern="/v1/chat/completions",
            strategy=MockStrategy.SEQUENCE,
            sequence=[{"status": 200, "body": r, "headers": {"Content-Type": "application/json"}} 
                     for r in (responses or default_responses)],
            priority=10
        ))

    @staticmethod
    def openai_embeddings(mock_server: MockServer, dimensions: int = 1536):
        """Mock OpenAI /v1/embeddings endpoint."""

        def embedding_callback(request_data):
            import json
            body = json.loads(request_data["body"]) if request_data["body"] else {}
            inputs = body.get("input", [])
            if isinstance(inputs, str):
                inputs = [inputs]

            embeddings = []
            for i, _ in enumerate(inputs):
                # Deterministic pseudo-embedding based on input hash
                import hashlib
                seed = hashlib.md5(inputs[i].encode()).hexdigest()
                random.seed(seed)
                emb = [random.uniform(-1, 1) for _ in range(dimensions)]
                # Normalize
                norm = sum(x*x for x in emb) ** 0.5
                emb = [x/norm for x in emb]
                embeddings.append({
                    "object": "embedding",
                    "embedding": emb,
                    "index": i
                })

            return web.Response(
                status=200,
                body=json.dumps({
                    "object": "list",
                    "data": embeddings,
                    "model": "text-embedding-3-small",
                    "usage": {"prompt_tokens": sum(len(x.split()) for x in inputs), "total_tokens": 100}
                }),
                headers={"Content-Type": "application/json"}
            )

        mock_server.add_rule(MockRule(
            method="POST",
            path_pattern="/v1/embeddings",
            strategy=MockStrategy.CALLBACK,
            callback=embedding_callback,
            priority=10
        ))

    @staticmethod
    def anthropic_messages(mock_server: MockServer):
        """Mock Anthropic /v1/messages endpoint."""

        def anthropic_callback(request_data):
            import json
            body = json.loads(request_data["body"]) if request_data["body"] else {}

            return web.Response(
                status=200,
                body=json.dumps({
                    "id": "msg_mock1",
                    "type": "message",
                    "role": "assistant",
                    "content": [{"type": "text", "text": "Mocked Anthropic response"}],
                    "model": body.get("model", "claude-3-sonnet-20240229"),
                    "stop_reason": "end_turn",
                    "usage": {"input_tokens": 10, "output_tokens": 20}
                }),
                headers={"Content-Type": "application/json"}
            )

        mock_server.add_rule(MockRule(
            method="POST",
            path_pattern="/v1/messages",
            strategy=MockStrategy.CALLBACK,
            callback=anthropic_callback,
            priority=10
        ))

    @staticmethod
    def add_rate_limit_mock(mock_server: MockServer, limit: int = 10, window: int = 60):
        """Add rate limiting simulation."""
        from collections import defaultdict
        import time

        request_times = defaultdict(list)

        def rate_limit_middleware(request_data):
            client_ip = request_data["headers"].get("x-forwarded-for", "unknown")
            now = time.time()

            # Clean old requests
            request_times[client_ip] = [t for t in request_times[client_ip] if now - t < window]

            if len(request_times[client_ip]) >= limit:
                return web.Response(
                    status=429,
                    body=json.dumps({
                        "error": {
                            "message": "Rate limit exceeded",
                            "type": "rate_limit_error",
                            "code": "rate_limit_exceeded"
                        }
                    }),
                    headers={
                        "Content-Type": "application/json",
                        "Retry-After": str(window)
                    }
                )

            request_times[client_ip].append(now)
            return None  # Continue to next rule

        mock_server.add_rule(MockRule(
            method="*",
            path_pattern="/v1/*",
            strategy=MockStrategy.CALLBACK,
            callback=rate_limit_middleware,
            priority=100  # Highest priority
        ))

    @staticmethod
    def add_streaming_mock(mock_server: MockServer):
        """Mock streaming responses."""

        async def streaming_callback(request_data):
            from aiohttp import web
            import asyncio

            response = web.StreamResponse(
                status=200,
                headers={
                    "Content-Type": "text/event-stream",
                    "Cache-Control": "no-cache",
                    "Connection": "keep-alive"
                }
            )
            await response.prepare(request_data.get("_aiohttp_request"))  # Would need actual request

            chunks = [
                'data: {"id":"mock","choices":[{"delta":{"content":"Hello"}}]}\n\n',
                'data: {"id":"mock","choices":[{"delta":{"content":" world"}}]}\n\n',
                'data: {"id":"mock","choices":[{"delta":{},"finish_reason":"stop"}]}\n\n',
                'data: [DONE]\n\n'
            ]

            for chunk in chunks:
                await response.write(chunk.encode())
                await asyncio.sleep(0.1)

            await response.write_eof()
            return response

        mock_server.add_rule(MockRule(
            method="POST",
            path_pattern="/v1/chat/completions",
            strategy=MockStrategy.CALLBACK,
            callback=streaming_callback,
            conditions={"headers": {"accept": "text/event-stream"}},
            priority=20
        ))

Part 2: Contract Testing

Contract Definition

# mocks/contracts.py
from dataclasses import dataclass, field
from typing: Dict, List, Optional, Any
from pydantic import BaseModel, Field
import json

class ParameterLocation(Enum):
    PATH = "path"
    QUERY = "query"
    HEADER = "header"
    BODY = "body"

@dataclass
class ContractParameter:
    name: str
    location: ParameterLocation
    required: bool = True
    schema: Dict = field(default_factory=dict)
    description: str = ""

@dataclass
class ContractResponse:
    status_code: int
    headers: Dict[str, str] = field(default_factory=dict)
    body_schema: Dict = field(default_factory=dict)
    description: str = ""

@dataclass
class APIContract:
    provider: str
    version: str
    base_url: str
    endpoints: Dict[str, Dict] = field(default_factory=dict)

    def add_endpoint(
        self,
        method: str,
        path: str,
        summary: str,
        parameters: List[ContractParameter] = None,
        responses: Dict[int, ContractResponse] = None,
        request_body: Dict = None
    ):
        key = f"{method.upper()} {path}"
        self.endpoints[key] = {
            "summary": summary,
            "parameters": parameters or [],
            "responses": responses or {},
            "request_body": request_body
        }

    def to_openapi(self) -> Dict:
        """Convert to OpenAPI 3.0 spec."""
        paths = {}
        for key, endpoint in self.endpoints.items():
            method, path = key.split(" ", 1)
            if path not in paths:
                paths[path] = {}

            parameters = []
            for p in endpoint["parameters"]:
                parameters.append({
                    "name": p.name,
                    "in": p.location.value,
                    "required": p.required,
                    "schema": p.schema,
                    "description": p.description
                })

            responses = {}
            for status, resp in endpoint["responses"].items():
                responses[str(status)] = {
                    "description": resp.description,
                    "headers": resp.headers,
                    "content": {
                        "application/json": {
                            "schema": resp.body_schema
                        }
                    }
                }

            paths[path][method.lower()] = {
                "summary": endpoint["summary"],
                "parameters": parameters,
                "responses": responses,
                "requestBody": endpoint.get("request_body", {})
            }

        return {
            "openapi": "3.0.0",
            "info": {"title": self.provider, "version": self.version},
            "servers": [{"url": self.base_url}],
            "paths": paths
        }

# OpenAI Contract
OPENAI_CONTRACT = APIContract(
    provider="OpenAI",
    version="2024-10-01-preview",
    base_url="https://api.openai.com/v1"
)

OPENAI_CONTRACT.add_endpoint(
    method="POST",
    path="/chat/completions",
    summary="Create chat completion",
    parameters=[
        ContractParameter("Authorization", ParameterLocation.HEADER, True, 
                         {"type": "string"}, "Bearer token"),
    ],
    request_body={
        "required": ["model", "messages"],
        "properties": {
            "model": {"type": "string"},
            "messages": {"type": "array", "items": {"type": "object"}},
            "temperature": {"type": "number", "minimum": 0, "maximum": 2},
            "max_tokens": {"type": "integer", "minimum": 1},
            "stream": {"type": "boolean"},
        }
    },
    responses={
        200: ContractResponse(200, {"Content-Type": "application/json"}, {
            "type": "object",
            "properties": {
                "id": {"type": "string"},
                "object": {"type": "string"},
                "created": {"type": "integer"},
                "model": {"type": "string"},
                "choices": {"type": "array"},
                "usage": {"type": "object"}
            }
        }, "Successful response"),
        400: ContractResponse(400, {}, {}, "Bad request"),
        401: ContractResponse(401, {}, {}, "Unauthorized"),
        429: ContractResponse(429, {}, {}, "Rate limited"),
    }
)

OPENAI_CONTRACT.add_endpoint(
    method="POST",
    path="/embeddings",
    summary="Create embeddings",
    request_body={
        "required": ["model", "input"],
        "properties": {
            "model": {"type": "string"},
            "input": {"type": ["string", "array"]},
            "dimensions": {"type": "integer"},
        }
    },
    responses={
        200: ContractResponse(200, {"Content-Type": "application/json"}, {
            "type": "object",
            "properties": {
                "object": {"type": "string"},
                "data": {"type": "array"},
                "model": {"type": "string"},
                "usage": {"type": "object"}
            }
        }, "Successful response"),
    }
)

Contract Testing

# mocks/contract_testing.py
from typing: Dict, List, Any
from dataclasses import dataclass
import jsonschema
import requests

@dataclass
class ContractTestResult:
    endpoint: str
    passed: bool
    errors: List[str]
    request: Dict
    response: Dict

class ContractTester:
    def __init__(self, contract: APIContract, base_url: str):
        self.contract = contract
        self.base_url = base_url.rstrip("/")

    def test_endpoint(
        self,
        method: str,
        path: str,
        request_data: Dict,
        expected_status: int = 200
    ) -> ContractTestResult:
        """Test single endpoint against contract."""
        endpoint_key = f"{method.upper()} {path}"
        contract_endpoint = self.contract.endpoints.get(endpoint_key)

        if not contract_endpoint:
            return ContractTestResult(
                endpoint=endpoint_key,
                passed=False,
                errors=[f"No contract defined for {endpoint_key}"],
                request=request_data,
                response={}
            )

        errors = []

        # Validate request
        if "request_body" in contract_endpoint:
            try:
                jsonschema.validate(request_data.get("json", {}), contract_endpoint["request_body"])
            except jsonschema.ValidationError as e:
                errors.append(f"Request validation failed: {e.message}")

        # Make request
        url = f"{self.base_url}{path}"
        headers = request_data.get("headers", {})
        json_body = request_data.get("json")

        try:
            response = requests.request(method, url, headers=headers, json=json_body, timeout=30)
        except Exception as e:
            return ContractTestResult(
                endpoint=endpoint_key,
                passed=False,
                errors=[f"Request failed: {e}"],
                request=request_data,
                response={}
            )

        # Validate response status
        expected_responses = contract_endpoint.get("responses", {})
        if expected_status not in expected_responses:
            # Check if any 2xx expected
            success_codes = [c for c in expected_responses if 200 <= c < 300]
            if response.status_code not in success_codes:
                errors.append(f"Unexpected status: {response.status_code}")

        # Validate response body
        if response.status_code in expected_responses:
            expected_schema = expected_responses[response.status_code].body_schema
            if expected_schema:
                try:
                    jsonschema.validate(response.json(), expected_schema)
                except jsonschema.ValidationError as e:
                    errors.append(f"Response validation failed: {e.message}")

        return ContractTestResult(
            endpoint=endpoint_key,
            passed=len(errors) == 0,
            errors=errors,
            request=request_data,
            response={"status": response.status_code, "body": response.json() if response.content else {}}
        )

    def run_all_tests(self, test_cases: List[Dict]) -> List[ContractTestResult]:
        """Run multiple contract tests."""
        results = []
        for tc in test_cases:
            result = self.test_endpoint(
                tc["method"],
                tc["path"],
                tc.get("request", {}),
                tc.get("expected_status", 200)
            )
            results.append(result)
        return results

# Test cases for OpenAI
OPENAI_TEST_CASES = [
    {
        "method": "POST",
        "path": "/v1/chat/completions",
        "request": {
            "headers": {"Authorization": "Bearer test-key"},
            "json": {
                "model": "gpt-4o-mini",
                "messages": [{"role": "user", "content": "Hello"}],
                "temperature": 0.7
            }
        },
        "expected_status": 200
    },
    {
        "method": "POST",
        "path": "/v1/chat/completions",
        "request": {
            "headers": {"Authorization": "Bearer test-key"},
            "json": {
                "model": "gpt-4o-mini",
                "messages": []
            }
        },
        "expected_status": 400
    },
    {
        "method": "POST",
        "path": "/v1/embeddings",
        "request": {
            "headers": {"Authorization": "Bearer test-key"},
            "json": {
                "model": "text-embedding-3-small",
                "input": "Test text"
            }
        },
        "expected_status": 200
    }
]

Part 3: Chaos Engineering

Failure Injection

# mocks/chaos.py
from dataclasses import dataclass
from typing: Dict, Any, Callable
from enum import Enum
import random
import asyncio

class ChaosType(Enum):
    LATENCY = "latency"
    ERROR = "error"
    TIMEOUT = "timeout"
    CORRUPTION = "corruption"
    PARTIAL = "partial"
    THROTTLE = "throttle"

@dataclass
class ChaosExperiment:
    name: str
    target_service: str
    chaos_type: ChaosType
    intensity: float  # 0-1
    duration_seconds: int
    conditions: Dict = field(default_factory=dict)

class ChaosEngine:
    def __init__(self, mock_server):
        self.mock_server = mock_server
        self.active_experiments: Dict[str, ChaosExperiment] = {}
        self.results: List[Dict] = []

    def inject_latency(self, path_pattern: str, min_ms: int, max_ms: int, probability: float = 1.0):
        """Inject random latency."""
        config = {
            "type": ChaosType.LATENCY.value,
            "min_ms": min_ms,
            "max_ms": max_ms,
            "probability": probability
        }
        self.mock_server.add_chaos("*", path_pattern, config)

    def inject_errors(self, path_pattern: str, error_rate: float, status_codes: List[int] = None):
        """Inject random errors."""
        config = {
            "type": ChaosType.ERROR.value,
            "error_rate": error_rate,
            "status_codes": status_codes or [500, 502, 503, 504]
        }
        self.mock_server.add_chaos("*", path_pattern, config)

    def inject_timeouts(self, path_pattern: str, timeout_rate: float):
        """Inject request timeouts."""
        config = {
            "type": ChaosType.TIMEOUT.value,
            "timeout_rate": timeout_rate
        }
        self.mock_server.add_chaos("*", path_pattern, config)

    def inject_corruption(self, path_pattern: str, corruption_rate: float):
        """Inject response corruption."""
        config = {
            "type": ChaosType.CORRUPTION.value,
            "corruption_rate": corruption_rate
        }
        self.mock_server.add_chaos("*", path_pattern, config)

    def inject_throttle(self, path_pattern: str, rate_limit: int, window_seconds: int):
        """Inject rate limiting."""
        config = {
            "type": ChaosType.THROTTLE.value,
            "rate_limit": rate_limit,
            "window_seconds": window_seconds
        }
        self.mock_server.add_chaos("*", path_pattern, config)

    async def run_experiment(self, experiment: ChaosExperiment, test_fn: Callable) -> Dict:
        """Run chaos experiment with test function."""
        self.active_experiments[experiment.name] = experiment

        start_time = asyncio.get_event_loop().time()
        errors = 0
        successes = 0
        latencies = []

        try:
            while asyncio.get_event_loop().time() - start_time < experiment.duration_seconds:
                try:
                    result = await test_fn()
                    if result.get("success"):
                        successes += 1
                    else:
                        errors += 1
                    if "latency_ms" in result:
                        latencies.append(result["latency_ms"])
                except Exception as e:
                    errors += 1

                await asyncio.sleep(0.1)
        finally:
            del self.active_experiments[experiment.name]

        result = {
            "experiment": experiment.name,
            "duration": experiment.duration_seconds,
            "total_requests": successes + errors,
            "successes": successes,
            "errors": errors,
            "error_rate": errors / max(1, successes + errors),
            "avg_latency_ms": sum(latencies) / len(latencies) if latencies else 0,
            "p95_latency_ms": sorted(latencies)[int(len(latencies) * 0.95)] if latencies else 0
        }

        self.results.append(result)
        return result

Resilience Testing Patterns

# mocks/resilience.py
"""
Resilience testing patterns for AI applications:

1. RETRY WITH BACKOFF
2. CIRCUIT BREAKER
3. FALLBACK RESPONSES
4. BULKHEAD ISOLATION
5. TIMEOUT MANAGEMENT
"""

from dataclasses import dataclass
from typing: Callable, Any, Optional
import asyncio
import time
from enum import Enum

class CircuitState(Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"

@dataclass
class RetryConfig:
    max_attempts: int = 3
    base_delay: float = 1.0
    max_delay: float = 30.0
    exponential_base: float = 2.0
    jitter: bool = True
    retry_on: tuple = (Exception,)

@dataclass
class CircuitBreakerConfig:
    failure_threshold: int = 5
    success_threshold: int = 2
    timeout: float = 30.0

class ResilientLLMClient:
    def __init__(
        self,
        base_client,
        retry_config: RetryConfig = None,
        circuit_config: CircuitBreakerConfig = None,
        fallback: Callable = None
    ):
        self.client = base_client
        self.retry_config = retry_config or RetryConfig()
        self.circuit_config = circuit_config or CircuitBreakerConfig()
        self.fallback = fallback

        self.circuit_state = CircuitState.CLOSED
        self.failure_count = 0
        self.success_count = 0
        self.last_failure_time = 0

    async def chat_completion(self, **kwargs) -> Dict:
        return await self._execute_with_resilience("chat_completion", kwargs)

    async def _execute_with_resilience(self, method: str, kwargs: Dict) -> Dict:
        # Circuit breaker check
        if self.circuit_state == CircuitState.OPEN:
            if time.time() - self.last_failure_time > self.circuit_config.timeout:
                self.circuit_state = CircuitState.HALF_OPEN
            else:
                if self.fallback:
                    return await self.fallback(method, kwargs)
                raise Exception("Circuit breaker open")

        # Retry loop
        last_exception = None
        for attempt in range(self.retry_config.max_attempts):
            try:
                result = await getattr(self.client, method)(**kwargs)

                # Success - reset circuit breaker
                self._on_success()
                return result

            except self.retry_config.retry_on as e:
                last_exception = e
                self._on_failure()

                if attempt < self.retry_config.max_attempts - 1:
                    delay = self._calculate_delay(attempt)
                    await asyncio.sleep(delay)

        # All retries failed
        if self.fallback:
            return await self.fallback(method, kwargs)
        raise last_exception

    def _calculate_delay(self, attempt: int) -> float:
        delay = min(
            self.retry_config.base_delay * (self.retry_config.exponential_base ** attempt),
            self.retry_config.max_delay
        )
        if self.retry_config.jitter:
            import random
            delay *= (0.5 + random.random())  # 50-150% of delay
        return delay

    def _on_success(self):
        self.failure_count = 0
        if self.circuit_state == CircuitState.HALF_OPEN:
            self.success_count += 1
            if self.success_count >= self.circuit_config.success_threshold:
                self.circuit_state = CircuitState.CLOSED
                self.success_count = 0

    def _on_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        self.success_count = 0

        if self.failure_count >= self.circuit_config.failure_threshold:
            self.circuit_state = CircuitState.OPEN

# Fallback implementations
async def cached_fallback(method: str, kwargs: Dict) -> Dict:
    """Return cached response."""
    return {"fallback": True, "source": "cache", "data": "cached response"}

async def static_fallback(method: str, kwargs: Dict) -> Dict:
    """Return static response."""
    return {"fallback": True, "source": "static", "data": "I'm currently experiencing issues."}

async def degraded_fallback(method: str, kwargs: Dict) -> Dict:
    """Return degraded but functional response."""
    # Use smaller model, fewer tokens, no streaming
    kwargs["model"] = "gpt-3.5-turbo"
    kwargs["max_tokens"] = 100
    kwargs["stream"] = False
    return {"fallback": True, "source": "degraded", **kwargs}

Part 4: Testing Strategies

Unit Testing with Mocks

# tests/test_with_mocks.py
import pytest
from unittest.mock import AsyncMock, patch
from mocks.server import MockServer
from mocks.llm_mocks import LLMMockBuilder

class TestAIApplication:
    @pytest.fixture
    async def mock_server(self):
        server = MockServer("localhost", 8081)
        LLMMockBuilder.openai_chat_completions(server)
        LLMMockBuilder.openai_embeddings(server)
        runner = await server.start()
        yield server
        await runner.cleanup()

    @pytest.fixture
    def ai_client(self, mock_server):
        from myapp import AIClient
        return AIClient(base_url="http://localhost:8081")

    @pytest.mark.asyncio
    async def test_chat_completion(self, ai_client, mock_server):
        response = await ai_client.chat(
            messages=[{"role": "user", "content": "Hello"}],
            model="gpt-4o-mini"
        )

        assert response["choices"][0]["message"]["content"] == "This is a mocked response from APIMock-AI."
        assert mock_server.verify_call("POST", "/v1/chat/completions", times=1)

    @pytest.mark.asyncio
    async def test_embeddings(self, ai_client, mock_server):
        embeddings = await ai_client.embed(["text1", "text2"])

        assert len(embeddings) == 2
        assert len(embeddings[0]) == 1536
        assert mock_server.verify_call("POST", "/v1/embeddings", times=1)

    @pytest.mark.asyncio
    async def test_rate_limit_handling(self, ai_client, mock_server):
        # Add rate limit mock
        LLMMockBuilder.add_rate_limit_mock(mock_server, limit=2, window=60)

        # First two should succeed
        await ai_client.chat(messages=[{"role": "user", "content": "1"}])
        await ai_client.chat(messages=[{"role": "user", "content": "2"}])

        # Third should handle rate limit
        with pytest.raises(Exception) as exc_info:
            await ai_client.chat(messages=[{"role": "user", "content": "3"}])

        assert "rate limit" in str(exc_info.value).lower()

Integration Testing

# tests/integration_test.py
import pytest
from mocks.server import MockServer
from mocks.llm_mocks import LLMMockBuilder
from mocks.chaos import ChaosEngine, ChaosType

class TestResilience:
    @pytest.fixture
    async def chaos_server(self):
        server = MockServer("localhost", 8082)
        LLMMockBuilder.openai_chat_completions(server)
        runner = await server.start()
        yield server, ChaosEngine(server)
        await runner.cleanup()

    @pytest.mark.asyncio
    async def test_latency_resilience(self, chaos_server):
        server, chaos = chaos_server

        # Inject 1-2s latency
        chaos.inject_latency("/v1/chat/completions", 1000, 2000, probability=1.0)

        from myapp import ResilientAIClient
        client = ResilientAIClient(base_url="http://localhost:8082", timeout=5.0)

        # Should succeed with retry
        response = await client.chat(messages=[{"role": "user", "content": "Test"}])
        assert "fallback" not in response or response["fallback"] == False

    @pytest.mark.asyncio
    async def test_error_resilience(self, chaos_server):
        server, chaos = chaos_server

        # Inject 50% error rate
        chaos.inject_errors("/v1/chat/completions", 0.5, [500, 503])

        from myapp import ResilientAIClient
        client = ResilientAIClient(
            base_url="http://localhost:8082",
            fallback=static_fallback
        )

        # Should eventually succeed or fallback
        response = await client.chat(messages=[{"role": "user", "content": "Test"}])
        assert response is not None

    @pytest.mark.asyncio
    async def test_circuit_breaker(self, chaos_server):
        server, chaos = chaos_server

        # Inject 100% errors
        chaos.inject_errors("/v1/chat/completions", 1.0, [500])

        from myapp import ResilientAIClient
        client = ResilientAIClient(
            base_url="http://localhost:8082",
            fallback=cached_fallback
        )

        # First few calls fail, then circuit opens
        for _ in range(3):
            try:
                await client.chat(messages=[{"role": "user", "content": "Test"}])
            except:
                pass

        # Next call should use fallback immediately
        response = await client.chat(messages=[{"role": "user", "content": "Test"}])
        assert response.get("fallback") == True
        assert response.get("source") == "cache"

Part 5: CI/CD Integration

GitHub Actions Workflow

# .github/workflows/api-mock-tests.yml
name: API Mock Tests

on:
  push:
    branches: [main, develop]
  pull_request:
  schedule:
    - cron: '0 6 * * 1'  # Weekly Monday 6 AM

jobs:
  contract-tests:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Start Mock Server
        run: |
          python -m mocks.server --port 8080 &
          sleep 3

      - name: Configure OpenAI Mocks
        run: python -m mocks.configure_openai_mocks

      - name: Run Contract Tests
        run: pytest tests/contract_tests.py -v

      - name: Run Unit Tests
        run: pytest tests/test_with_mocks.py -v

      - name: Run Resilience Tests
        run: pytest tests/integration_test.py -v

      - name: Generate Report
        run: python -m mocks.generate_report > mock-test-report.json

      - name: Upload Report
        uses: actions/upload-artifact@v4
        with:
          name: mock-test-report
          path: mock-test-report.json

  chaos-engineering:
    runs-on: ubuntu-latest
    if: github.event_name == 'schedule'
    steps:
      - uses: actions/checkout@v4

      - name: Run Chaos Experiments
        run: |
          python -m mocks.run_chaos_experiments \
            --experiment latency \
            --experiment errors \
            --experiment timeouts \
            --duration 60

      - name: Analyze Results
        run: python -m mocks.analyze_chaos_results

      - name: Alert on Degradation
        if: failure()
        uses: slackapi/slack-github-action@v1.23.0
        with:
          payload: |
            {
              "text": "Chaos engineering detected resilience degradation",
              "blocks": [...]
            }

Local Development Setup

# scripts/start_mocks.sh
#!/bin/bash

# Start mock servers for local development
echo "Starting APIMock-AI servers..."

# LLM API mock
python -m mocks.server --port 8080 --config mocks/configs/openai.yaml &

# Vector DB mock
python -m mocks.server --port 8081 --config mocks/configs/qdrant.yaml &

# Embedding service mock
python -m mocks.server --port 8082 --config mocks/configs/embeddings.yaml &

echo "Mock servers running:"
echo "  OpenAI API: http://localhost:8080"
echo "  Qdrant: http://localhost:8081"
echo "  Embeddings: http://localhost:8082"

# Save PIDs for cleanup
jobs -p > .mock_pids

# mocks/configs/openai.yaml
server:
  host: localhost
  port: 8080

rules:
  - method: POST
    path: /v1/chat/completions
    strategy: sequence
    responses:
      - status: 200
        body:
          id: "chatcmpl-dev"
          object: "chat.completion"
          choices:
            - index: 0
              message:
                role: assistant
                content: "Development mock response"
              finish_reason: stop
          usage:
            prompt_tokens: 10
            completion_tokens: 20
            total_tokens: 30

  - method: POST
    path: /v1/embeddings
    strategy: callback
    module: mocks.llm_mocks
    function: openai_embeddings_callback

Part 6: Production Patterns

Mock Recording/Playback

# mocks/recorder.py
"""Record real API calls and replay as mocks."""

from dataclasses import dataclass, field
from typing: Dict, List, Any
import json
import hashlib
from datetime import datetime

@dataclass
class RecordedInteraction:
    id: str
    timestamp: str
    request: Dict
    response: Dict
    metadata: Dict = field(default_factory=dict)

class MockRecorder:
    def __init__(self, storage_path: str = "mocks/recordings"):
        self.storage_path = storage_path
        self.recordings: List[RecordedInteraction] = []
        self.recording = False

    def start_recording(self):
        self.recording = True
        self.recordings = []

    def stop_recording(self) -> List[RecordedInteraction]:
        self.recording = False
        return self.recordings

    def record(self, request: Dict, response: Dict, metadata: Dict = None):
        if not self.recording:
            return

        interaction = RecordedInteraction(
            id=hashlib.md5(f"{request['method']}{request['path']}{datetime.now()}".encode()).hexdigest()[:12],
            timestamp=datetime.utcnow().isoformat(),
            request=request,
            response=response,
            metadata=metadata or {}
        )
        self.recordings.append(interaction)

    def save(self, name: str):
        import os
        os.makedirs(self.storage_path, exist_ok=True)

        filepath = f"{self.storage_path}/{name}.json"
        with open(filepath, "w") as f:
            json.dump([{
                "id": r.id,
                "timestamp": r.timestamp,
                "request": r.request,
                "response": r.response,
                "metadata": r.metadata
            } for r in self.recordings], f, indent=2)

    def load(self, name: str) -> List[RecordedInteraction]:
        filepath = f"{self.storage_path}/{name}.json"
        with open(filepath) as f:
            data = json.load(f)
        return [RecordedInteraction(**r) for r in data]

    def generate_mock_rules(self, recordings: List[RecordedInteraction]) -> List[MockRule]:
        """Generate mock rules from recordings."""
        rules = []

        # Group by endpoint
        by_endpoint = {}
        for r in recordings:
            key = f"{r.request['method']} {r.request['path']}"
            if key not in by_endpoint:
                by_endpoint[key] = []
            by_endpoint[key].append(r)

        for endpoint, interactions in by_endpoint.items():
            method, path = endpoint.split(" ", 1)

            if len(interactions) == 1:
                # Single response - static
                r = interactions[0]
                rules.append(MockRule(
                    method=method,
                    path_pattern=path,
                    strategy=MockStrategy.STATIC,
                    response={
                        "status": r.response["status"],
                        "body": r.response["body"],
                        "headers": r.response.get("headers", {})
                    }
                ))
            else:
                # Multiple responses - sequence
                rules.append(MockRule(
                    method=method,
                    path_pattern=path,
                    strategy=MockStrategy.SEQUENCE,
                    sequence=[{
                        "status": r.response["status"],
                        "body": r.response["body"],
                        "headers": r.response.get("headers", {})
                    } for r in interactions]
                ))

        return rules

Part 7: APIMock-AI Product Features

Key Features Checklist

CORE MOCKING
✅ Static responses
✅ Sequential responses
✅ Template-based responses
✅ Dynamic callbacks
✅ Proxy mode
✅ Chaos engineering

CONTRACT TESTING
✅ OpenAPI 3.0 support
✅ Request validation
✅ Response validation
✅ Schema evolution detection
✅ Breaking change alerts

DEVELOPER EXPERIENCE
✅ Zero-config startup
✅ Hot reload
✅ Request logging
✅ Verification API
✅ VS Code extension
✅ CLI tool

RESILIENCE TESTING
✅ Latency injection
✅ Error injection
✅ Timeout simulation
✅ Response corruption
✅ Rate limiting
✅ Circuit breaker testing

INTEGRATIONS
✅ OpenAI API
✅ Anthropic API
✅ Google Gemini API
✅ Vector databases (Qdrant, Pinecone, Weaviate, Milvus)
✅ Custom HTTP APIs
✅ GraphQL endpoints
✅ gRPC services

DEPLOYMENT
✅ Docker image
✅ Kubernetes operator
✅ Helm chart
✅ GitHub Action
✅ GitLab CI
✅ Local development

Resources

API mocking is not a testing luxury — it's a development necessity. The teams that mock external dependencies from day one ship faster, debug easier, and sleep better. APIMock-AI makes it trivial to simulate the entire AI ecosystem locally.

Vector Databases Deep Dive: Qdrant, Pinecone, Weaviate & Milvus Compared

Phil IPS — Mon, 12 Oct 2026 19:17:30 +0000

Vector Databases Deep Dive: Qdrant, Pinecone, Weaviate & Milvus Compared

Complete comparison of the four major vector databases for AI applications — architecture, indexing, performance, scaling, and production deployment.

Introduction

Vector databases are the backbone of modern AI applications — RAG, semantic search, recommendations, and embeddings storage. This guide compares the four major players with production insights.

Vector Database Landscape

Database	Type	License	Best For
Qdrant	Purpose-built	Apache 2.0	High performance, filtering, Rust
Pinecone	Managed service	Proprietary	Zero-ops, enterprise
Weaviate	Purpose-built	BSD-3	GraphQL, modules, hybrid search
Milvus	Purpose-built	Apache 2.0	Massive scale, Kubernetes native

Part 1: Core Concepts

Vector Index Types

# vectordb/indexes.py
from enum import Enum
from dataclasses import dataclass
from typing: List

class IndexType(Enum):
    FLAT = "flat"                    # Exact search, O(n)
    IVF_FLAT = "ivf_flat"           # Inverted file, balanced
    IVF_PQ = "ivf_pq"               # Product quantization, compressed
    HNSW = "hnsw"                   # Hierarchical NSW, fast approximate
    ANNOY = "annoy"                 # Trees, good for static data
    SCANN = "scann"                 # Google's ScaNN, high recall
    DISKANN = "diskann"             # Disk-based, large scale

@dataclass
class IndexConfig:
    index_type: IndexType
    params: dict

    # HNSW params
    m: int = 16              # Connections per layer
    ef_construct: int = 100  # Build-time search width
    ef_search: int = 50      # Query-time search width

    # IVF params
    nlist: int = 100         # Number of clusters
    nprobe: int = 10         # Clusters to search

    # PQ params
    m_pq: int = 8            # Subquantizers
    nbits: int = 8           # Bits per subquantizer

# Recommendations by use case
INDEX_RECOMMENDATIONS = {
    "small_exact": IndexConfig(IndexType.FLAT, {}),
    "medium_fast": IndexConfig(IndexType.HNSW, {"m": 16, "ef_construct": 200}),
    "large_scale": IndexConfig(IndexType.IVF_PQ, {"nlist": 4096, "m_pq": 16}),
    "massive": IndexConfig(IndexType.DISKANN, {}),
}

Distance Metrics

# vectordb/metrics.py
class DistanceMetric(Enum):
    COSINE = "cosine"           # Normalized dot product, most common
    EUCLIDEAN = "euclidean"     # L2 distance
    DOT = "dot"                 # Inner product (requires normalized vectors)
    MANHATTAN = "manhattan"     # L1 distance
    HAMMING = "hamming"         # Binary vectors
    JACCARD = "jaccard"         # Set similarity

# Choose metric based on embedding model
METRIC_BY_MODEL = {
    "text-embedding-3-large": DistanceMetric.COSINE,
    "text-embedding-3-small": DistanceMetric.COSINE,
    "bge-large-en-v1.5": DistanceMetric.COSINE,
    "e5-large-v2": DistanceMetric.COSINE,
    "instructor-xl": DistanceMetric.COSINE,
    # Some models work with dot product (vectors pre-normalized)
}

def normalize_vectors(vectors: List[List[float]]) -> List[List[float]]:
    """Normalize for cosine similarity with dot product index."""
    import numpy as np
    arr = np.array(vectors)
    norms = np.linalg.norm(arr, axis=1, keepdims=True)
    return (arr / (norms + 1e-8)).tolist()

Part 2: Qdrant Deep Dive

Architecture & Features

# vectordb/qdrant_client.py
from qdrant_client import QdrantClient
from qdrant_client.models import *
from typing: List, Dict, Optional
import numpy as np

class QdrantManager:
    def __init__(self, url: str = "http://localhost:6333", api_key: str = None):
        self.client = QdrantClient(url=url, api_key=api_key)

    def create_collection(
        self,
        name: str,
        vector_size: int,
        distance: Distance = Distance.COSINE,
        hnsw_config: HnswConfigDiff = None,
        quantization_config: QuantizationConfig = None,
        on_disk_payload: bool = True
    ):
        """Create optimized collection."""

        vectors_config = VectorParams(
            size=vector_size,
            distance=distance,
            hnsw_config=hnsw_config or HnswConfigDiff(
                m=16,
                ef_construct=200,
                full_scan_threshold=10000
            ),
            quantization_config=quantization_config or ScalarQuantizationConfig(
                type=ScalarQuantizationType.INT8,
                quantile=0.99,
                always_ram=True
            ),
            on_disk=on_disk_payload
        )

        self.client.create_collection(
            collection_name=name,
            vectors_config=vectors_config,
            optimizers_config=OptimizersConfigDiff(
                default_segment_number=2,
                max_segment_size=20000,
                memmap_threshold=50000,
                indexing_threshold=20000,
                flush_interval_sec=5
            ),
            wal_config=WalConfigDiff(wal_capacity_mb=32),
        )

    def upsert_batch(
        self,
        collection_name: str,
        points: List[PointStruct],
        batch_size: int = 1000
    ) -> UpdateResult:
        """Efficient batch upsert."""
        for i in range(0, len(points), batch_size):
            batch = points[i:i + batch_size]
            self.client.upsert(
                collection_name=collection_name,
                points=batch,
                wait=True
            )
        return UpdateResult(status=UpdateStatus.COMPLETED)

    def search(
        self,
        collection_name: str,
        query_vector: List[float],
        limit: int = 10,
        score_threshold: float = None,
        query_filter: Filter = None,
        search_params: SearchParams = None
    ) -> List[ScoredPoint]:
        """High-performance search with filtering."""
        return self.client.search(
            collection_name=collection_name,
            query_vector=query_vector,
            limit=limit,
            score_threshold=score_threshold,
            query_filter=query_filter,
            search_params=search_params or SearchParams(
                hnsw_ef=128,
                exact=False
            )
        )

    def hybrid_search(
        self,
        collection_name: str,
        query_vector: List[float],
        query_text: str,
        limit: int = 10,
        vector_weight: float = 0.7,
        text_weight: float = 0.3
    ) -> List[ScoredPoint]:
        """Hybrid vector + keyword search (requires sparse vectors)."""
        # Requires sparse vector index setup
        return self.client.query_points(
            collection_name=collection_name,
            prefetch=[
                Prefetch(
                    query=query_vector,
                    using="dense",
                    limit=limit * 2
                ),
                Prefetch(
                    query=query_text,
                    using="sparse",
                    limit=limit * 2
                )
            ],
            query=FusionQuery(fusion=Fusion.RRF),
            limit=limit
        )

Qdrant Performance Tuning

# vectordb/qdrant_tuning.py
QDRANT_TUNING_GUIDE = {
    "memory_optimization": {
        "quantization": "Use scalar (INT8) or binary quantization for 4-8x memory reduction",
        "on_disk_payload": "Store payload on disk, keep only vectors in RAM",
        "memmap_threshold": "Enable memmap for segments > 50k vectors",
        "hnsw_m": "Lower m (8-16) for memory, higher (32-64) for recall",
    },
    "search_speed": {
        "hnsw_ef_search": "Increase for recall (50-256), decrease for speed",
        "exact_search": "Use exact=True for small collections (<10k)",
        "parallel_queries": "Batch multiple queries in single request",
        "filter_optimization": "Put selective filters first, use payload indexes",
    },
    "indexing_speed": {
        "max_segment_size": "Larger segments = faster indexing, more memory",
        "indexing_threshold": "Delay indexing until segment reaches threshold",
        "default_segment_number": "Match CPU cores for parallel indexing",
        "flush_interval_sec": "Balance durability vs performance",
    },
    "scaling": {
        "sharding": "Auto-shard by setting shard_number > 1",
        "replication": "Set replication_factor >= 2 for HA",
        "distributed": "Use Qdrant Cluster for multi-node",
    }
}

Part 3: Pinecone Deep Dive

Serverless Architecture

# vectordb/pinecone_client.py
from pinecone import Pinecone, ServerlessSpec
from typing: List, Dict, Optional
import time

class PineconeManager:
    def __init__(self, api_key: str):
        self.pc = Pinecone(api_key=api_key)

    def create_index(
        self,
        name: str,
        dimension: int,
        metric: str = "cosine",
        spec: ServerlessSpec = None,
        deletion_protection: str = "disabled"
    ):
        """Create serverless index."""
        if spec is None:
            spec = ServerlessSpec(cloud="aws", region="us-east-1")

        self.pc.create_index(
            name=name,
            dimension=dimension,
            metric=metric,
            spec=spec,
            deletion_protection=deletion_protection
        )

        # Wait for ready
        while not self.pc.describe_index(name).status["ready"]:
            time.sleep(1)

    def upsert(
        self,
        index_name: str,
        vectors: List[Dict],
        namespace: str = "",
        batch_size: int = 100
    ):
        """Batch upsert with automatic retry."""
        index = self.pc.Index(index_name)

        for i in range(0, len(vectors), batch_size):
            batch = vectors[i:i + batch_size]
            index.upsert(vectors=batch, namespace=namespace)

    def query(
        self,
        index_name: str,
        vector: List[float],
        top_k: int = 10,
        namespace: str = "",
        filter: Dict = None,
        include_metadata: bool = True,
        include_values: bool = False
    ):
        index = self.pc.Index(index_name)
        return index.query(
            vector=vector,
            top_k=top_k,
            namespace=namespace,
            filter=filter,
            include_metadata=include_metadata,
            include_values=include_values
        )

    def query_by_id(
        self,
        index_name: str,
        id: str,
        top_k: int = 10,
        namespace: str = ""
    ):
        """Query using existing vector ID (efficient for recommendations)."""
        index = self.pc.Index(index_name)
        return index.query(
            id=id,
            top_k=top_k,
            namespace=namespace
        )

Pinecone Cost Optimization

# vectordb/pinecone_costs.py
PINECONE_COSTS = {
    "serverless": {
        "storage_gb_month": 0.33,      # $0.33/GB/month
        "read_units": 0.000065,        # Per 1000 read units
        "write_units": 0.000013,       # Per 1000 write units
    },
    "pod_based": {
        "p1_x1": 0.096,                # $0.096/hour
        "p1_x2": 0.192,
        "p1_x4": 0.384,
        "p1_x8": 0.768,
        "p2_x1": 0.346,
        "p2_x2": 0.692,
        "p2_x4": 1.384,
        "p2_x8": 2.768,
    }
}

def estimate_monthly_cost(
    index_type: str,
    vector_count: int,
    dimension: int,
    queries_per_month: int,
    writes_per_month: int,
    pod_type: str = "p1_x1"
) -> dict:
    """Estimate Pinecone costs."""

    # Storage: ~4 bytes per dimension per vector (float32)
    storage_gb = (vector_count * dimension * 4) / (1024 ** 3)

    if index_type == "serverless":
        storage_cost = storage_gb * PINECONE_COSTS["serverless"]["storage_gb_month"]
        read_cost = (queries_per_month / 1000) * PINECONE_COSTS["serverless"]["read_units"]
        write_cost = (writes_per_month / 1000) * PINECONE_COSTS["serverless"]["write_units"]
        total = storage_cost + read_cost + write_cost
    else:
        pod_cost = PINECONE_COSTS["pod_based"][pod_type] * 24 * 30
        total = pod_cost

    return {
        "storage_gb": storage_gb,
        "monthly_cost_usd": total,
        "breakdown": {
            "storage": storage_cost if index_type == "serverless" else pod_cost,
            "reads": read_cost if index_type == "serverless" else 0,
            "writes": write_cost if index_type == "serverless" else 0
        }
    }

Part 4: Weaviate Deep Dive

GraphQL & Modules

# vectordb/weaviate_client.py
import weaviate
from weaviate.classes.init import Auth
from weaviate.classes.config import Configure, Property, DataType
from typing: List, Dict, Optional

class WeaviateManager:
    def __init__(
        self,
        url: str = "http://localhost:8080",
        auth_credentials: Auth = None
    ):
        self.client = weaviate.connect_to_custom(
            http_host=url.replace("http://", "").replace("https://", ""),
            http_port=8080,
            http_secure=False,
            auth_credentials=auth_credentials
        )

    def create_collection(
        self,
        name: str,
        properties: List[Property],
        vectorizer_config: Configure.Vectorizer = None,
        vector_index_config: Configure.VectorIndex = None
    ):
        """Create collection with vectorizer module."""

        if vectorizer_config is None:
            vectorizer_config = Configure.Vectorizer.text2vec_openai(
                model="text-embedding-3-small"
            )

        if vector_index_config is None:
            vector_index_config = Configure.VectorIndex.hnsw(
                distance_metric=VectorDistances.COSINE,
                quantizer=Configure.VectorIndex.Quantizer.bq()
            )

        self.client.collections.create(
            name=name,
            properties=properties,
            vectorizer_config=vectorizer_config,
            vector_index_config=vector_index_config
        )

    def hybrid_search(
        self,
        collection_name: str,
        query: str,
        vector: List[float] = None,
        alpha: float = 0.5,  # 0 = keyword, 1 = vector
        limit: int = 10,
        filters = None
    ):
        """Hybrid search with GraphQL-like flexibility."""
        collection = self.client.collections.get(collection_name)

        return collection.query.hybrid(
            query=query,
            vector=vector,
            alpha=alpha,
            limit=limit,
            filters=filters,
            return_metadata=MetadataQuery(
                score=True,
                explain_score=True
            )
        )

    def bm25_search(
        self,
        collection_name: str,
        query: str,
        limit: int = 10,
        properties: List[str] = None
    ):
        """Pure keyword search (BM25)."""
        collection = self.client.collections.get(collection_name)
        return collection.query.bm25(
            query=query,
            limit=limit,
            query_properties=properties
        )

Weaviate Modules

# vectordb/weaviate_modules.py
WEAVIATE_MODULES = {
    "text2vec_openai": {
        "description": "OpenAI embeddings",
        "models": ["text-embedding-3-small", "text-embedding-3-large", "ada-002"],
        "params": ["model", "dimensions", "base_url"]
    },
    "text2vec_cohere": {
        "description": "Cohere embeddings",
        "models": ["embed-english-v3.0", "embed-multilingual-v3.0"],
        "params": ["model", "truncate"]
    },
    "text2vec_huggingface": {
        "description": "Local HuggingFace models",
        "models": ["sentence-transformers/all-MiniLM-L6-v2", "BAAI/bge-large-en-v1.5"],
        "params": ["model", "endpoint"]
    },
    "generative_openai": {
        "description": "RAG generation with OpenAI",
        "models": ["gpt-4o", "gpt-4o-mini", "gpt-3.5-turbo"],
        "params": ["model", "temperature"]
    },
    "generative_cohere": {
        "description": "RAG generation with Cohere",
        "models": ["command-r-plus", "command-r"],
        "params": ["model", "temperature"]
    },
    "reranker_cohere": {
        "description": "Cohere reranking",
        "models": ["rerank-english-v3.0", "rerank-multilingual-v3.0"]
    },
}

# Multi-vector collections
MULTI_VECTOR_CONFIG = {
    "dense": Configure.Vectorizer.text2vec_openai(model="text-embedding-3-small"),
    "sparse": Configure.Vectorizer.text2vec_contextionary()  # BM25-like
}

Part 5: Milvus Deep Dive

Kubernetes-Native Architecture

# vectordb/milvus_client.py
from pymilvus import (
    MilvusClient, DataType, Function, FunctionType,
    AnnSearchRequest, RRFRanker, WeightedRanker
)
from typing: List, Dict, Optional

class MilvusManager:
    def __init__(self, uri: str = "http://localhost:19530", token: str = ""):
        self.client = MilvusClient(uri=uri, token=token)

    def create_collection(
        self,
        name: str,
        dimension: int,
        primary_field: str = "id",
        vector_field: str = "vector",
        enable_dynamic_field: bool = True,
        auto_id: bool = True
    ):
        schema = self.client.create_schema(
            auto_id=auto_id,
            enable_dynamic_field=enable_dynamic_field
        )

        schema.add_field(
            field_name=primary_field,
            datatype=DataType.INT64,
            is_primary=True
        )
        schema.add_field(
            field_name=vector_field,
            datatype=DataType.FLOAT_VECTOR,
            dim=dimension
        )
        schema.add_field(
            field_name="text",
            datatype=DataType.VARCHAR,
            max_length=65535
        )
        schema.add_field(
            field_name="metadata",
            datatype=DataType.JSON
        )

        # Add sparse vector field for hybrid search
        schema.add_field(
            field_name="sparse_vector",
            datatype=DataType.SPARSE_FLOAT_VECTOR
        )

        # Add BM25 function for sparse vectors
        bm25_function = Function(
            name="text_bm25",
            function_type=FunctionType.BM25,
            input_field_names=["text"],
            output_field_names=["sparse_vector"]
        )
        schema.add_function(bm25_function)

        index_params = self.client.prepare_index_params()
        index_params.add_index(
            field_name=vector_field,
            index_type="HNSW",
            metric_type="COSINE",
            params={"M": 16, "efConstruction": 200}
        )
        index_params.add_index(
            field_name="sparse_vector",
            index_type="SPARSE_INVERTED_INDEX",
            metric_type="BM25"
        )

        self.client.create_collection(
            collection_name=name,
            schema=schema,
            index_params=index_params
        )

    def hybrid_search(
        self,
        collection_name: str,
        dense_vector: List[float],
        sparse_query: str,
        limit: int = 10,
        dense_weight: float = 0.7,
        sparse_weight: float = 0.3,
        output_fields: List[str] = None
    ):
        """Hybrid dense + sparse search with re-ranking."""

        dense_req = AnnSearchRequest(
            data=[dense_vector],
            anns_field="vector",
            param={"metric_type": "COSINE", "params": {"ef": 128}},
            limit=limit * 2
        )

        sparse_req = AnnSearchRequest(
            data=[sparse_query],
            anns_field="sparse_vector",
            param={"metric_type": "BM25"},
            limit=limit * 2
        )

        # RRF fusion (default)
        rerank = RRFRanker(k=60)

        return self.client.hybrid_search(
            collection_name=collection_name,
            reqs=[dense_req, sparse_req],
            rerank=rerank,
            limit=limit,
            output_fields=output_fields or ["text", "metadata"]
        )

Milvus Scaling

# vectordb/milvus_scaling.py
MILVUS_SCALING = {
    "components": {
        "proxy": "Stateless, handles requests, auto-scale",
        "query_node": "Executes searches, scale with query load",
        "data_node": "Handles ingestion, scale with write load",
        "index_node": "Builds indexes, batch scale",
        "root_coord": "Metadata, single active",
        "query_coord": "Load balancing, single active",
        "data_coord": "Segment management, single active",
    },
    "deployment": {
        "standalone": "All-in-one, dev/test only",
        "cluster": "Separate components, production",
        "k8s_operator": "Milvus Operator for Kubernetes",
        "helm_chart": "Official Helm chart with all components",
    },
    "partitioning": {
        "partition_key": "Hash-based partitioning for multi-tenancy",
        "shard_num": "Number of shards per collection (default: 1-64)",
    }
}

Part 6: Production Comparison

Benchmark Results (Typical)

# vectordb/benchmarks.py
BENCHMARK_RESULTS = {
    "dataset": "1M vectors, 768-dim, COSINE",
    "queries": 1000,
    "results": {
        "Qdrant": {
            "p50_latency_ms": 12,
            "p99_latency_ms": 45,
            "recall_at_10": 0.98,
            "memory_gb": 3.2,
            "index_build_min": 8
        },
        "Pinecone": {
            "p50_latency_ms": 18,
            "p99_latency_ms": 55,
            "recall_at_10": 0.97,
            "memory_gb": "managed",
            "index_build_min": "managed"
        },
        "Weaviate": {
            "p50_latency_ms": 22,
            "p99_latency_ms": 68,
            "recall_at_10": 0.96,
            "memory_gb": 4.1,
            "index_build_min": 12
        },
        "Milvus": {
            "p50_latency_ms": 15,
            "p99_latency_ms": 52,
            "recall_at_10": 0.98,
            "memory_gb": 3.8,
            "index_build_min": 10
        }
    }
}

Feature Matrix

# vectordb/feature_matrix.py
FEATURE_MATRIX = {
    "Feature": ["Qdrant", "Pinecone", "Weaviate", "Milvus"],
    "Deployment": ["Self-hosted/Cloud", "Managed only", "Self-hosted/Cloud", "Self-hosted/K8s"],
    "License": ["Apache 2.0", "Proprietary", "BSD-3", "Apache 2.0"],
    "Hybrid Search": ["✅ Native", "✅ Sparse+Dense", "✅ Native", "✅ Native"],
    "Filtering": ["✅ Payload indexes", "✅ Metadata filter", "✅ GraphQL where", "✅ Scalar index"],
    "Multi-tenancy": ["✅ Sharding", "✅ Namespaces", "✅ Tenants", "✅ Partition keys"],
    "Backups": ["✅ Snapshots", "✅ Collections", "✅ Backups", "✅ Backups"],
    "Rollback": ["✅ Point-in-time", "❌", "✅", "✅"],
    "GraphQL": ["❌", "❌", "✅ Native", "❌"],
    "REST API": ["✅", "✅", "✅", "✅"],
    "gRPC": ["✅", "✅", "✅", "✅"],
    "Python SDK": ["✅ Excellent", "✅ Good", "✅ Good", "✅ Good"],
    "Rust Core": ["✅", "❌", "❌ (Go)", "❌ (C++)"],
    "Quantization": ["✅ Scalar/Binary", "❌", "✅ BQ/PQ", "✅ PQ/SQ"],
    "DiskANN": ["✅", "❌", "❌", "✅"],
    "RBAC": ["✅ Cloud only", "✅", "✅", "✅"],
    "Cost (1M vecs)": ["~$50/mo self-host", "~$200/mo serverless", "~$80/mo self-host", "~$100/mo K8s"],
}

Part 7: Migration & Interoperability

Universal Client Interface

# vectordb/universal.py
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing: List, Dict, Optional

@dataclass
class VectorRecord:
    id: str
    vector: List[float]
    payload: Dict
    sparse_vector: Dict[int, float] = None

@dataclass
class SearchResult:
    id: str
    score: float
    payload: Dict
    vector: List[float] = None

class VectorDBClient(ABC):
    @abstractmethod
    def create_collection(self, name: str, dimension: int, config: dict): pass

    @abstractmethod
    def upsert(self, collection: str, records: List[VectorRecord]): pass

    @abstractmethod
    def search(
        self,
        collection: str,
        query_vector: List[float],
        limit: int = 10,
        filter: Dict = None
    ) -> List[SearchResult]: pass

    @abstractmethod
    def hybrid_search(
        self,
        collection: str,
        dense_vector: List[float],
        sparse_query: str,
        limit: int = 10
    ) -> List[SearchResult]: pass

    @abstractmethod
    def delete(self, collection: str, ids: List[str]): pass

    @abstractmethod
    def get_stats(self, collection: str) -> dict: pass

# Factory
def create_vector_db(provider: str, **kwargs) -> VectorDBClient:
    providers = {
        "qdrant": QdrantClient,
        "pinecone": PineconeClient,
        "weaviate": WeaviateClient,
        "milvus": MilvusClient,
    }
    return providers[provider.lower()](**kwargs)

Migration Strategy

# vectordb/migration.py
async def migrate_collection(
    source: VectorDBClient,
    target: VectorDBClient,
    collection_name: str,
    batch_size: int = 1000
):
    """Migrate collection between vector databases."""

    # 1. Get source schema/stats
    stats = source.get_stats(collection_name)
    dimension = stats["dimension"]

    # 2. Create target collection
    target.create_collection(collection_name, dimension, {})

    # 3. Stream migrate in batches
    offset = 0
    total = stats["vector_count"]

    while offset < total:
        # Scroll source
        records = source.scroll(collection_name, offset, batch_size)

        # Transform if needed (dimension change, etc.)
        transformed = transform_records(records, dimension)

        # Upsert to target
        target.upsert(collection_name, transformed)

        offset += batch_size
        print(f"Migrated {min(offset, total)}/{total}")

    # 4. Verify
    target_stats = target.get_stats(collection_name)
    assert target_stats["vector_count"] == total

Part 8: Production Checklist

Pre-Production

# vectordb/production_checklist.py
PRODUCTION_CHECKLIST = {
    "capacity_planning": [
        "Estimate vector count growth (3-5x current)",
        "Calculate memory: vectors * dim * 4 bytes * 1.5 (overhead)",
        "Plan for quantization (INT8 = 4x savings)",
        "Reserve 30% headroom for indexing",
    ],
    "index_tuning": [
        "HNSW: m=16, ef_construct=200 for balanced",
        "ef_search=128 for production queries",
        "Enable quantization for >100k vectors",
        "Set appropriate nlist/nprobe for IVF",
    ],
    "reliability": [
        "Enable WAL (write-ahead log)",
        "Configure replication factor >= 2",
        "Set up automated backups (daily + incremental)",
        "Test restore procedure monthly",
        "Monitor disk usage, set alerts at 70%",
    ],
    "security": [
        "Enable TLS for all connections",
        "Use API keys, rotate quarterly",
        "Network policies: restrict to app subnets",
        "Audit logs enabled",
    ],
    "monitoring": [
        "Latency p50/p95/p99 per collection",
        "Query throughput (QPS)",
        "Index build queue depth",
        "Memory/CPU/disk per node",
        "Recall@K sampling (weekly)",
        "Error rates by type",
    ],
    "scaling": [
        "Horizontal pod autoscaler (K8s)",
        "Read replicas for query-heavy workloads",
        "Sharding strategy for >10M vectors",
        "Separate indexing nodes from query nodes",
    ]
}

Decision Framework

VECTOR DATABASE SELECTION:

Team has DevOps capacity?
├─→ No → Pinecone (managed) or Qdrant Cloud
└─→ Yes
    ├─→ Need GraphQL / semantic modules?
    │   └─→ Yes → Weaviate
    │
    ├─→ Kubernetes-native, massive scale (>100M)?
    │   └─→ Yes → Milvus
    │
    ├─→ Rust performance, filtering, hybrid search?
    │   └─→ Yes → Qdrant
    │
    ├─→ Budget sensitive, self-hosted?
    │   └─→ Yes → Qdrant or Milvus
    │
    └─→ Enterprise support required?
        └─→ Yes → Pinecone or Weaviate Cloud

Migration Paths

CURRENT → RECOMMENDED MIGRATION

Elasticsearch/OpenSearch → Qdrant (similar filter model)
pgvector → Qdrant/Milvus (better scaling)
Chroma → Qdrant (compatible API, better perf)
Pinecone → Qdrant Cloud (API compatible, cost savings)
Weaviate → Qdrant (if GraphQL not needed)
Custom FAISS → Milvus (managed FAISS at scale)

Resources

Deploying LLMs Locally: Ollama, llama.cpp, vLLM & Text Generation Inference

Phil IPS — Mon, 12 Oct 2026 19:11:04 +0000

Deploying LLMs Locally: Ollama, llama.cpp, vLLM & Text Generation Inference

Complete guide to running LLMs locally — from quick experiments with Ollama to production-grade serving with vLLM and TGI.

Introduction

Local LLM deployment offers privacy, cost savings, and offline capability. This guide covers the four main approaches with production configurations.

Quick Comparison

Tool	Best For	Ease of Use	Performance	Hardware
Ollama	Quick start, Mac/Win/Linux	⭐⭐⭐⭐⭐	Good	CPU/GPU
llama.cpp	CPU inference, quantization	⭐⭐⭐	Excellent (CPU)	CPU/GPU
vLLM	High-throughput serving	⭐⭐⭐	Best (GPU)	GPU required
TGI	Enterprise, OpenAI-compatible	⭐⭐⭐	Excellent	GPU required

Part 1: Ollama — Easiest Start

Installation & Basics

# macOS
brew install ollama

# Linux
curl -fsSL https://ollama.com/install.sh | sh

# Windows
# Download from https://ollama.com/download

# Start service
ollama serve

# Pull model
ollama pull llama3.1:8b
ollama pull codellama:13b
ollama pull nemotron-mini:4b

Python Integration

# ollama/client.py
import ollama
from typing import List, Dict, Optional

class OllamaClient:
    def __init__(self, host: str = "http://localhost:11434"):
        self.client = ollama.Client(host=host)

    def list_models(self) -> List[Dict]:
        return self.client.list()["models"]

    def pull(self, model: str) -> None:
        self.client.pull(model)

    def generate(
        self,
        model: str,
        prompt: str,
        system: str = None,
        temperature: float = 0.7,
        top_p: float = 0.9,
        num_predict: int = -1,
        stream: bool = False
    ) -> str:
        response = self.client.generate(
            model=model,
            prompt=prompt,
            system=system,
            options={
                "temperature": temperature,
                "top_p": top_p,
                "num_predict": num_predict,
            },
            stream=stream
        )
        return response["response"]

    def chat(
        self,
        model: str,
        messages: List[Dict],
        stream: bool = False
    ):
        return self.client.chat(model=model, messages=messages, stream=stream)

    def embeddings(self, model: str, prompt: str) -> List[float]:
        return self.client.embeddings(model=model, prompt=prompt)["embedding"]

# Usage
client = OllamaClient()
response = client.generate("llama3.1:8b", "Explain LoRA in 3 sentences")
print(response)

# Streaming chat
for chunk in client.chat("llama3.1:8b", [
    {"role": "user", "content": "Write a haiku about GPUs"}
], stream=True):
    print(chunk["message"]["content"], end="", flush=True)

Custom Modelfiles

# Modelfile for custom model
FROM llama3.1:8b

# System prompt
SYSTEM """You are a senior Python developer. Write clean, typed, documented code."""

# Parameters
PARAMETER temperature 0.3
PARAMETER top_p 0.9
PARAMETER num_ctx 8192
PARAMETER num_predict 2048

# Template (optional)
TEMPLATE """{{ .System }}User: {{ .Prompt }}Assistant: """

# Save and create
# ollama create python-expert -f Modelfile

Ollama API Server

# ollama/api_server.py
from fastapi import FastAPI
from pydantic import BaseModel
import ollama

app = FastAPI()

class GenerateRequest(BaseModel):
    model: str
    prompt: str
    system: str = None
    temperature: float = 0.7
    stream: bool = False

@app.post("/generate")
async def generate(req: GenerateRequest):
    if req.stream:
        return StreamingResponse(
            stream_generate(req),
            media_type="text/event-stream"
        )

    response = ollama.generate(
        model=req.model,
        prompt=req.prompt,
        system=req.system,
        options={"temperature": req.temperature}
    )
    return {"response": response["response"]}

async def stream_generate(req: GenerateRequest):
    stream = ollama.generate(
        model=req.model,
        prompt=req.prompt,
        system=req.system,
        options={"temperature": req.temperature},
        stream=True
    )
    for chunk in stream:
        yield f"data: {chunk['response']}\n\n"

# Run: uvicorn api_server:app --host 0.0.0.0 --port 8000

Part 2: llama.cpp — CPU Optimization

Building from Source

# Ubuntu/Debian
sudo apt update && sudo apt install -y build-essential cmake git

# Clone
git clone https://github.com/ggerganov/llama.cpp
cd llama.cpp

# CPU-only build (AVX2)
make LLAMA_CUBLAS=0

# With CUDA (NVIDIA GPU)
make LLAMA_CUBLAS=1

# With Metal (Apple Silicon)
make LLAMA_METAL=1

# With Vulkan (AMD/Intel GPU)
make LLAMA_VULKAN=1

Model Conversion & Quantization

# Download HF model
huggingface-cli download meta-llama/Meta-Llama-3.1-8B --local-dir llama3.1-8b

# Convert to GGUF
python convert_hf_to_gguf.py llama3.1-8b --outfile llama3.1-8b.f16.gguf --outtype f16

# Quantize (smaller, faster, slight quality loss)
# Q4_K_M = best balance (4-bit)
./llama-quantize llama3.1-8b.f16.gguf llama3.1-8b.Q4_K_M.gguf Q4_K_M

# Other quantization levels
# Q2_K  - smallest, lowest quality
# Q3_K_M - small, decent
# Q4_K_M - recommended default
# Q5_K_M - higher quality
# Q6_K  - near original
# Q8_0  - almost lossless

Running llama.cpp Server

# Basic server
./llama-server -m llama3.1-8b.Q4_K_M.gguf -c 8192 --host 0.0.0.0 --port 8080

# With GPU offloading (Metal)
./llama-server -m llama3.1-8b.Q4_K_M.gguf -c 8192 --host 0.0.0.0 --port 8080 -ngl 99

# With CUDA
./llama-server -m llama3.1-8b.Q4_K_M.gguf -c 8192 --host 0.0.0.0 --port 8080 -ngl 99 -mgpu 0

# Performance options
./llama-server \
  -m model.gguf \
  -c 8192 \           # Context size
  -b 512 \            # Batch size
  -ub 512 \           # UBatch size
  -ngl 99 \           # GPU layers (-1 = all)
  -t 8 \              # Threads
  --flash-attn \      # Flash attention
  --mlock \           # Lock in RAM
  --no-mmap \         # Don't mmap (for containers)

Python Client

# llama_cpp/client.py
from llama_cpp import Llama
from typing: List, Dict
import json

class LlamaCppClient:
    def __init__(
        self,
        model_path: str,
        n_ctx: int = 8192,
        n_gpu_layers: int = -1,
        n_threads: int = 8,
        verbose: bool = False
    ):
        self.llm = Llama(
            model_path=model_path,
            n_ctx=n_ctx,
            n_gpu_layers=n_gpu_layers,
            n_threads=n_threads,
            verbose=verbose,
            chat_format="llama-3",  # or "chatml", "mistral-instruct"
        )

    def generate(
        self,
        prompt: str,
        max_tokens: int = 512,
        temperature: float = 0.7,
        top_p: float = 0.9,
        stop: List[str] = None,
        stream: bool = False
    ):
        return self.llm(
            prompt=prompt,
            max_tokens=max_tokens,
            temperature=temperature,
            top_p=top_p,
            stop=stop,
            stream=stream
        )

    def chat(
        self,
        messages: List[Dict],
        max_tokens: int = 512,
        temperature: float = 0.7,
        stream: bool = False
    ):
        return self.llm.create_chat_completion(
            messages=messages,
            max_tokens=max_tokens,
            temperature=temperature,
            stream=stream
        )

    def embeddings(self, texts: List[str]) -> List[List[float]]:
        return [self.llm.embed(text) for text in texts]

# Usage
client = LlamaCppClient("./llama3.1-8b.Q4_K_M.gguf", n_gpu_layers=99)
response = client.generate("Explain quantization:")
print(response["choices"][0]["text"])

Part 3: vLLM — High-Throughput Serving

Installation

# Requires NVIDIA GPU with CUDA 11.8+
pip install vllm

# Or with specific CUDA
pip install vllm --extra-index-url https://download.pytorch.org/whl/cu118

# For AMD ROCm
pip install vllm --extra-index-url https://download.pytorch.org/whl/rocm5.6

Basic Server

# OpenAI-compatible server
vllm serve meta-llama/Meta-Llama-3.1-8B-Instruct \
  --host 0.0.0.0 \
  --port 8000 \
  --dtype auto \
  --max-model-len 8192 \
  --gpu-memory-utilization 0.9 \
  --tensor-parallel-size 1 \
  --enforce-eager

# With quantization (AWQ/GPTQ)
vllm serve TheBloke/Llama-3.1-8B-Instruct-AWQ \
  --quantization awq \
  --max-model-len 8192

# Multi-GPU tensor parallel
vllm serve meta-llama/Meta-Llama-3.1-70B-Instruct \
  --tensor-parallel-size 4 \
  --gpu-memory-utilization 0.9

Python Client (OpenAI Compatible)

# vllm/client.py
from openai import OpenAI
from typing: List, Dict
import asyncio

class VLLMClient:
    def __init__(self, base_url: str = "http://localhost:8000/v1"):
        self.client = OpenAI(
            base_url=base_url,
            api_key="dummy"  # Not used by vLLM
        )

    def list_models(self):
        return self.client.models.list()

    def generate(
        self,
        model: str,
        prompt: str,
        max_tokens: int = 512,
        temperature: float = 0.7,
        top_p: float = 0.9,
        stop: List[str] = None,
        stream: bool = False
    ):
        return self.client.completions.create(
            model=model,
            prompt=prompt,
            max_tokens=max_tokens,
            temperature=temperature,
            top_p=top_p,
            stop=stop,
            stream=stream
        )

    def chat(
        self,
        model: str,
        messages: List[Dict],
        max_tokens: int = 512,
        temperature: float = 0.7,
        stream: bool = False
    ):
        return self.client.chat.completions.create(
            model=model,
            messages=messages,
            max_tokens=max_tokens,
            temperature=temperature,
            stream=stream
        )

    async def achat(self, *args, **kwargs):
        return await self.client.chat.completions.create(*args, **kwargs)

# Usage
client = VLLMClient()

# Chat completion
response = client.chat("meta-llama/Meta-Llama-3.1-8B-Instruct", [
    {"role": "system", "content": "You are a helpful assistant."},
    {"role": "user", "content": "Write a Python decorator for retry logic"}
])
print(response.choices[0].message.content)

# Streaming
for chunk in client.chat(model, messages, stream=True):
    if chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)

Advanced vLLM Features

# vllm/advanced.py
from vllm import LLM, SamplingParams
from vllm.lora.request import LoRARequest

# Offline inference (no server)
llm = LLM(
    model="meta-llama/Meta-Llama-3.1-8B-Instruct",
    tensor_parallel_size=1,
    gpu_memory_utilization=0.9,
    max_model_len=8192,
    enable_lora=True,
    max_lora_rank=64,
)

sampling_params = SamplingParams(
    temperature=0.7,
    top_p=0.9,
    max_tokens=512,
    stop=["</s>", "<|eot_id|>"],
)

# Batch generation
prompts = [
    "Write a function for: ",
    "Explain the concept of: ",
    "Debug this code: ",
]
outputs = llm.generate(prompts, sampling_params)

# With LoRA adapter
lora_outputs = llm.generate(
    prompts,
    sampling_params,
    lora_request=LoRARequest("sql_adapter", 1, "./sql-lora-adapter")
)

# Structured output (guided decoding)
from vllm import SamplingParams
from pydantic import BaseModel

class CodeResponse(BaseModel):
    explanation: str
    code: str
    language: str

guided_params = SamplingParams(
    temperature=0.3,
    max_tokens=1024,
    guided_json=CodeResponse.model_json_schema()
)

output = llm.generate(["Write a binary search in Python"], guided_params)
print(output[0].outputs[0].text)  # Valid JSON matching schema

vLLM Performance Tuning

# vllm/tuning.sh

# Key parameters:
# --max-model-len: Context window (affects KV cache memory)
# --gpu-memory-utilization: Fraction of GPU memory for KV cache (0.8-0.95)
# --block-size: KV cache block size (16 or 32)
# --max-num-seqs: Max concurrent sequences
# --max-num-batched-tokens: Max tokens per batch
# --enable-chunked-prefill: For long contexts
# --disable-log-stats: Reduce logging overhead

# High-throughput config
vllm serve meta-llama/Meta-Llama-3.1-8B-Instruct \
  --max-model-len 4096 \
  --gpu-memory-utilization 0.95 \
  --max-num-seqs 256 \
  --max-num-batched-tokens 8192 \
  --enable-chunked-prefill \
  --disable-log-stats \
  --block-size 16

# Low-latency config
vllm serve meta-llama/Meta-Llama-3.1-8B-Instruct \
  --max-model-len 8192 \
  --gpu-memory-utilization 0.8 \
  --max-num-seqs 16 \
  --enforce-eager \
  --disable-log-stats

Part 4: Text Generation Inference (TGI)

Installation

# Docker (recommended)
docker run --gpus all --shm-size 1g \
  -p 8080:80 \
  -v $PWD/data:/data \
  ghcr.io/huggingface/text-generation-inference:latest \
  --model-id meta-llama/Meta-Llama-3.1-8B-Instruct \
  --max-input-length 4096 \
  --max-total-tokens 8192 \
  --max-batch-prefill-tokens 4096

# With quantization
docker run --gpus all --shm-size 1g \
  -p 8080:80 \
  ghcr.io/huggingface/text-generation-inference:latest \
  --model-id TheBloke/Llama-3.1-8B-Instruct-AWQ \
  --quantize awq

# Multi-GPU
docker run --gpus all --shm-size 1g \
  -p 8080:80 \
  ghcr.io/huggingface/text-generation-inference:latest \
  --model-id meta-llama/Meta-Llama-3.1-70B-Instruct \
  --num-shard 4

Python Client

# tgi/client.py
from text_generation import Client
from typing: List, Dict
import asyncio

class TGIClient:
    def __init__(self, url: str = "http://localhost:8080"):
        self.client = Client(url, timeout=60)

    def generate(
        self,
        prompt: str,
        max_new_tokens: int = 512,
        temperature: float = 0.7,
        top_p: float = 0.9,
        stop_sequences: List[str] = None,
        stream: bool = False
    ):
        return self.client.generate(
            prompt,
            max_new_tokens=max_new_tokens,
            temperature=temperature,
            top_p=top_p,
            stop_sequences=stop_sequences,
            stream=stream
        )

    async def agenerate(self, *args, **kwargs):
        return await self.client.generate_async(*args, **kwargs)

# Usage
client = TGIClient()
response = client.generate("Explain RAG:")
print(response.generated_text)

Part 5: Model Selection Guide

Recommended Models by Use Case

# models/recommendations.py
RECOMMENDED_MODELS = {
    "general_chat": [
        "meta-llama/Meta-Llama-3.1-8B-Instruct",
        "mistralai/Mistral-7B-Instruct-v0.3",
        "google/gemma-2-9b-it",
        "Qwen/Qwen2.5-7B-Instruct",
    ],
    "coding": [
        "deepseek-ai/deepseek-coder-6.7b-instruct",
        "Qwen/Qwen2.5-Coder-7B-Instruct",
        "bigcode/starcoder2-7b",
        "codellama/CodeLlama-13b-Instruct-hf",
    ],
    "reasoning": [
        "deepseek-ai/DeepSeek-R1-Distill-Llama-8B",
        "Qwen/QwQ-32B-Preview",
    ],
    "multilingual": [
        "Qwen/Qwen2.5-7B-Instruct",
        "google/gemma-2-9b-it",
        "CohereForAI/aya-23-8B",
    ],
    "small_efficient": [
        "microsoft/Phi-3.5-mini-instruct",
        "google/gemma-2-2b-it",
        "Qwen/Qwen2.5-1.5B-Instruct",
    ],
    "long_context": [
        "gradientai/Llama-3-8B-Instruct-Gradient-1048k",
        "Qwen/Qwen2.5-7B-Instruct-1M",
    ],
}

# Quantization recommendations
QUANTIZATION_GUIDE = {
    "4bit": "Q4_K_M (llama.cpp) / AWQ (vLLM/TGI) — best balance",
    "8bit": "Q8_0 / GPTQ — near lossless",
    "2bit": "Q2_K — only for extreme constraints",
}

Hardware Requirements

# models/hardware.py
MODEL_REQUIREMENTS = {
    "7B_f16": {"vram_gb": 14, "ram_gb": 16, "disk_gb": 14},
    "7B_q4":  {"vram_gb": 6,  "ram_gb": 8,  "disk_gb": 4.5},
    "7B_q8":  {"vram_gb": 10, "ram_gb": 12, "disk_gb": 8},
    "13B_f16": {"vram_gb": 26, "ram_gb": 32, "disk_gb": 26},
    "13B_q4":  {"vram_gb": 10, "ram_gb": 16, "disk_gb": 8},
    "70B_f16": {"vram_gb": 140, "ram_gb": 160, "disk_gb": 140},
    "70B_q4":  {"vram_gb": 48, "ram_gb": 64, "disk_gb": 40},
}

def estimate_requirements(model_size: str, quantization: str) -> dict:
    """Estimate hardware needs."""
    key = f"{model_size}_{quantization}"
    return MODEL_REQUIREMENTS.get(key, {"error": "Unknown combination"})

# Apple Silicon unified memory
# M1/M2/M3 Max (96GB) → 70B q4
# M1/M2/M3 Pro (36GB) → 13B q4 or 7B f16
# M1/M2/M3 Base (8-24GB) → 7B q4

Part 6: Production Deployment

Docker Compose Stack

# deployment/docker-compose.yml
version: '3.8'

services:
  # vLLM for high throughput
  vllm:
    image: vllm/vllm-openai:latest
    runtime: nvidia
    environment:
      - HF_TOKEN=${HF_TOKEN}
    command: >
      --model meta-llama/Meta-Llama-3.1-8B-Instruct
      --max-model-len 8192
      --gpu-memory-utilization 0.9
      --tensor-parallel-size 1
      --host 0.0.0.0
      --port 8000
    ports:
      - "8000:8000"
    deploy:
      resources:
        reservations:
          devices:
            - driver: nvidia
              count: 1
              capabilities: [gpu]

  # TGI for OpenAI compatibility
  tgi:
    image: ghcr.io/huggingface/text-generation-inference:latest
    runtime: nvidia
    command: >
      --model-id meta-llama/Meta-Llama-3.1-8B-Instruct
      --max-input-length 4096
      --max-total-tokens 8192
    ports:
      - "8080:80"
    shm_size: 1g

  # Ollama for local development
  ollama:
    image: ollama/ollama:latest
    ports:
      - "11434:11434"
    volumes:
      - ollama_data:/root/.ollama
    deploy:
      resources:
        reservations:
          devices:
            - driver: nvidia
              count: 1
              capabilities: [gpu]

volumes:
  ollama_data:

Kubernetes Deployment

# deployment/k8s-vllm.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: vllm-llama3-8b
spec:
  replicas: 2
  selector:
    matchLabels:
      app: vllm
  template:
    metadata:
      labels:
        app: vllm
    spec:
      containers:
      - name: vllm
        image: vllm/vllm-openai:latest
        command: ["python", "-m", "vllm.entrypoints.openai.api_server"]
        args:
          - "--model=meta-llama/Meta-Llama-3.1-8B-Instruct"
          - "--max-model-len=8192"
          - "--gpu-memory-utilization=0.9"
          - "--host=0.0.0.0"
          - "--port=8000"
        ports:
        - containerPort: 8000
        resources:
          limits:
            nvidia.com/gpu: 1
          requests:
            nvidia.com/gpu: 1
            memory: "16Gi"
        env:
        - name: HF_TOKEN
          valueFrom:
            secretKeyRef:
              name: hf-token
              key: token
---
apiVersion: v1
kind: Service
metadata:
  name: vllm-service
spec:
  selector:
    app: vllm
  ports:
  - port: 8000
    targetPort: 8000
  type: LoadBalancer

Load Balancing & Autoscaling

# deployment/load_balancer.py
import httpx
from typing: List
import random

class LLMLoadBalancer:
    """Round-robin load balancer with health checks."""

    def __init__(self, endpoints: List[str]):
        self.endpoints = endpoints
        self.healthy = set(endpoints)
        self.current = 0

    async def health_check(self):
        """Periodic health check."""
        async with httpx.AsyncClient(timeout=5) as client:
            for ep in self.endpoints:
                try:
                    resp = await client.get(f"{ep}/health")
                    if resp.status_code == 200:
                        self.healthy.add(ep)
                    else:
                        self.healthy.discard(ep)
                except:
                    self.healthy.discard(ep)

    def get_endpoint(self) -> str:
        healthy = list(self.healthy)
        if not healthy:
            raise Exception("No healthy endpoints")
        ep = healthy[self.current % len(healthy)]
        self.current += 1
        return ep

    async def generate(self, *args, **kwargs):
        ep = self.get_endpoint()
        async with httpx.AsyncClient(base_url=ep, timeout=60) as client:
            return await client.post("/v1/completions", json=kwargs)

Part 7: Monitoring & Observability

Prometheus Metrics

# monitoring/prometheus_metrics.py
from prometheus_client import Counter, Histogram, Gauge
import time

LLM_REQUESTS = Counter(
    "llm_requests_total",
    "Total LLM requests",
    ["model", "status"]
)

LLM_LATENCY = Histogram(
    "llm_latency_seconds",
    "Request latency",
    ["model"],
    buckets=[0.1, 0.5, 1, 2, 5, 10, 30, 60]
)

LLM_TOKENS = Counter(
    "llm_tokens_total",
    "Tokens generated",
    ["model", "type"]  # prompt, completion
)

LLM_GPU_MEMORY = Gauge(
    "llm_gpu_memory_bytes",
    "GPU memory usage",
    ["model", "gpu_id"]
)

LLM_QUEUE_SIZE = Gauge(
    "llm_queue_size",
    "Request queue size",
    ["model"]
)

class MonitoredLLM:
    def __init__(self, client, model_name: str):
        self.client = client
        self.model_name = model_name

    def generate(self, prompt: str, **kwargs):
        start = time.time()
        try:
            result = self.client.generate(prompt, **kwargs)
            LLM_REQUESTS.labels(model=self.model_name, status="success").inc()
            return result
        except Exception as e:
            LLM_REQUESTS.labels(model=self.model_name, status="error").inc()
            raise
        finally:
            LLM_LATENCY.labels(model=self.model_name).observe(time.time() - start)

Part 8: Benchmarking

Throughput Test

# benchmark/throughput.py
import asyncio
import time
from statistics import mean, stdev
from vllm import LLM, SamplingParams

async def benchmark_throughput(
    model: str,
    num_requests: int = 100,
    prompt_len: int = 512,
    output_len: int = 256,
    concurrency: int = 10
):
    """Benchmark throughput and latency."""

    llm = LLM(model=model, gpu_memory_utilization=0.9)
    sampling_params = SamplingParams(
        max_tokens=output_len,
        temperature=0.7
    )

    prompt = "x" * prompt_len
    prompts = [prompt] * num_requests

    # Warmup
    llm.generate([prompt], sampling_params)

    # Benchmark
    start = time.time()
    outputs = llm.generate(prompts, sampling_params)
    total_time = time.time() - start

    total_tokens = sum(len(o.outputs[0].token_ids) for o in outputs)

    return {
        "total_time_sec": total_time,
        "requests_per_sec": num_requests / total_time,
        "tokens_per_sec": total_tokens / total_time,
        "avg_latency_ms": (total_time / num_requests) * 1000,
        "total_tokens": total_tokens,
    }

# Run
results = asyncio.run(benchmark_throughput("meta-llama/Meta-Llama-3.1-8B-Instruct"))
print(f"Throughput: {results['requests_per_sec']:.1f} req/s")
print(f"Tokens/sec: {results['tokens_per_sec']:.0f}")

Part 9: Cost Analysis

Local vs Cloud

# benchmark/cost_analysis.py
CLOUD_PRICING = {
    "gpt-4o": {"input": 2.50, "output": 10.00},      # per 1M tokens
    "gpt-4o-mini": {"input": 0.15, "output": 0.60},
    "claude-3.5-sonnet": {"input": 3.00, "output": 15.00},
}

GPU_COSTS = {
    "RTX 4090 (24GB)": {"purchase": 1600, "power_w": 450, "vram": 24},
    "RTX 3090 (24GB)": {"purchase": 800, "power_w": 350, "vram": 24},
    "A100 40GB": {"hourly": 1.50, "power_w": 400, "vram": 40},
    "A100 80GB": {"hourly": 3.00, "power_w": 400, "vram": 80},
    "H100 80GB": {"hourly": 8.00, "power_w": 700, "vram": 80},
}

def calculate_break_even(
    model: str,
    requests_per_day: int,
    avg_input_tokens: int,
    avg_output_tokens: int,
    gpu: str = "RTX 4090"
) -> dict:
    """Calculate when local becomes cheaper than cloud."""

    cloud = CLOUD_PRICING.get(model, {"input": 0, "output": 0})
    daily_cloud_cost = (
        requests_per_day * avg_input_tokens * cloud["input"] +
        requests_per_day * avg_output_tokens * cloud["output"]
    ) / 1_000_000

    gpu_info = GPU_COSTS[gpu]
    if "purchase" in gpu_info:
        # Electricity cost (assume $0.12/kWh)
        daily_power_cost = (gpu_info["power_w"] * 24 / 1000) * 0.12
        daily_local_cost = daily_power_cost
        break_even_days = gpu_info["purchase"] / (daily_cloud_cost - daily_local_cost)
    else:
        daily_local_cost = gpu_info["hourly"] * 24
        break_even_days = float('inf') if daily_local_cost >= daily_cloud_cost else 0

    return {
        "daily_cloud_cost": daily_cloud_cost,
        "daily_local_cost": daily_local_cost,
        "monthly_savings": (daily_cloud_cost - daily_local_cost) * 30,
        "break_even_days": max(0, break_even_days),
    }

# Example: 1000 requests/day, 1k in / 500 out tokens
print(calculate_break_even("gpt-4o-mini", 1000, 1000, 500, "RTX 4090"))
# {'daily_cloud_cost': 0.45, 'daily_local_cost': 1.30, 'monthly_savings': -25.5, 'break_even_days': 0}
# Local cheaper at scale, cloud cheaper for low volume

Part 10: Troubleshooting

Common Issues

# troubleshooting/common.py
TROUBLESHOOTING = {
    "OOM": {
        "vllm": [
            "Reduce --gpu-memory-utilization (0.8 → 0.7)",
            "Reduce --max-model-len",
            "Enable --enable-chunked-prefill",
            "Use quantization (AWQ/GPTQ)",
            "Reduce --max-num-seqs",
        ],
        "llama.cpp": [
            "Reduce -c (context size)",
            "Reduce -ngl (GPU layers)",
            "Use smaller quantization (Q4 → Q3)",
            "Add --mlock to prevent swap",
        ],
        "tgi": [
            "Reduce --max-total-tokens",
            "Reduce --max-batch-prefill-tokens",
            "Enable --quantize",
        ],
    },
    "Slow_Inference": {
        "vllm": [
            "Increase --gpu-memory-utilization",
            "Disable --enforce-eager",
            "Increase --max-num-batched-tokens",
            "Use tensor parallel for large models",
        ],
        "llama.cpp": [
            "Increase -ngl (more GPU layers)",
            "Use -ngl 99 (all layers GPU)",
            "Enable --flash-attn",
            "Increase -t (threads) for CPU",
        ],
    },
    "Quality_Loss": {
        "quantization": [
            "Try Q4_K_M instead of Q4_0",
            "Use Q5_K_M or Q6_K for better quality",
            "Avoid Q2_K/Q3_K for reasoning tasks",
            "Test with your specific prompts",
        ],
    },
}

Resources

Start with Ollama for development, graduate to vLLM/TGI for production. The best local LLM setup is the one that runs on your hardware.

Building AI Agents with LangGraph: State Machines, Workflows & Multi-Agent Systems

Phil IPS — Mon, 12 Oct 2026 19:04:56 +0000

Building AI Agents with LangGraph: State Machines, Workflows & Multi-Agent Systems

Complete guide to building production-ready AI agents using LangGraph — state management, workflows, human-in-the-loop, and multi-agent orchestration.

Introduction

LangGraph extends LangChain with stateful, cyclic graphs for building agents. Unlike linear chains, LangGraph supports:

Cycles & loops for iterative reasoning
State persistence across steps
Human-in-the-loop checkpoints
Multi-agent coordination

# langgraph/basics.py
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List
import operator

class AgentState(TypedDict):
    messages: Annotated[List[str], operator.add]  # Auto-append
    query: str
    answer: str
    iterations: int

def agent_node(state: AgentState) -> AgentState:
    # Process and return updated state
    return {"answer": "processed", "iterations": state["iterations"] + 1}

graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)
graph.set_entry_point("agent")
graph.add_edge("agent", END)

app = graph.compile()
result = app.invoke({"query": "Hello", "messages": [], "iterations": 0})

Part 1: State Management

TypedDict State Schema

# langgraph/state.py
from typing import TypedDict, Annotated, List, Optional, Dict, Any
import operator
from pydantic import BaseModel, Field

class AgentState(TypedDict):
    # Core fields
    query: str
    messages: Annotated[List[Dict], operator.add]  # Auto-merge lists

    # Retrieval
    retrieved_docs: Annotated[List[Dict], operator.add]
    retrieval_scores: Annotated[List[float], operator.add]

    # Generation
    answer: str
    citations: Annotated[List[str], operator.add]

    # Control
    iterations: int
    max_iterations: int
    should_continue: bool

    # Human-in-the-loop
    human_feedback: Optional[str]
    requires_human: bool

    # Multi-agent
    active_agent: str
    agent_outputs: Annotated[Dict[str, Any], operator.add]

# Pydantic for validation
class ValidatedState(BaseModel):
    query: str = Field(min_length=1)
    messages: List[Dict] = []
    iterations: int = Field(default=0, ge=0, le=10)

    class Config:
        arbitrary_types_allowed = True

State Reducers

# langgraph/reducers.py
from typing import Annotated, List, Dict, Any
import operator
from functools import reduce

# Custom reducers for complex merging
def merge_dicts(existing: Dict, new: Dict) -> Dict:
    """Deep merge dictionaries."""
    result = existing.copy()
    for k, v in new.items():
        if k in result and isinstance(result[k], dict) and isinstance(v, dict):
            result[k] = merge_dicts(result[k], v)
        else:
            result[k] = v
    return result

def merge_lists_unique(existing: List, new: List) -> List:
    """Merge lists, deduplicate by ID if dicts."""
    seen = set()
    result = []
    for item in existing + new:
        key = item.get("id") if isinstance(item, dict) else item
        if key not in seen:
            seen.add(key)
            result.append(item)
    return result

# Usage in state
class StateWithReducers(TypedDict):
    config: Annotated[Dict, merge_dicts]
    documents: Annotated[List[Dict], merge_lists_unique]
    metrics: Annotated[Dict[str, float], lambda old, new: {**old, **new}]

Part 2: Core Graph Patterns

Basic Agent Loop

# langgraph/agent_loop.py
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.sqlite import SqliteSaver
from langchain_core.messages import HumanMessage, AIMessage
from langchain_openai import ChatOpenAI
from typing import TypedDict, Annotated, List
import operator

class AgentState(TypedDict):
    messages: Annotated[List, operator.add]
    query: str
    answer: str
    steps: int

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

def reason(state: AgentState) -> AgentState:
    """LLM reasoning step."""
    messages = state["messages"]
    response = llm.invoke(messages)
    return {
        "messages": [AIMessage(content=response.content)],
        "steps": state["steps"] + 1
    }

def should_continue(state: AgentState) -> str:
    """Decide whether to continue or end."""
    if state["steps"] >= 5:
        return "end"
    last_msg = state["messages"][-1]
    if "FINAL ANSWER" in last_msg.content:
        return "end"
    return "continue"

graph = StateGraph(AgentState)
graph.add_node("reason", reason)
graph.set_entry_point("reason")
graph.add_conditional_edges(
    "reason",
    should_continue,
    {"continue": "reason", "end": END}
)

# With checkpointing
checkpointer = SqliteSaver.from_conn_string("checkpoints.db")
app = graph.compile(checkpointer=checkpointer)

# Run with thread ID for persistence
config = {"configurable": {"thread_id": "user-123"}}
result = app.invoke(
    {"query": "Research AI agents", "messages": [HumanMessage(content="Research AI agents")], "steps": 0},
    config=config
)

Tool-Using Agent

# langgraph/tool_agent.py
from langgraph.prebuilt import ToolNode
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List
import operator
import requests

@tool
def search_web(query: str) -> str:
    """Search the web for information."""
    # Implement with SerpAPI, Tavily, etc.
    return f"Search results for: {query}"

@tool
def calculate(expression: str) -> str:
    """Evaluate mathematical expression."""
    try:
        return str(eval(expression))
    except:
        return "Invalid expression"

@tool
def get_weather(city: str) -> str:
    """Get current weather for a city."""
    # Implement with weather API
    return f"Weather in {city}: Sunny, 22°C"

tools = [search_web, calculate, get_weather]
tool_node = ToolNode(tools)

llm = ChatOpenAI(model="gpt-4o").bind_tools(tools)

class AgentState(TypedDict):
    messages: Annotated[List, operator.add]

def agent(state: AgentState) -> AgentState:
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

def should_use_tools(state: AgentState) -> str:
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tools"
    return "end"

graph = StateGraph(AgentState)
graph.add_node("agent", agent)
graph.add_node("tools", tool_node)
graph.set_entry_point("agent")
graph.add_conditional_edges("agent", should_use_tools, {"tools": "tools", "end": END})
graph.add_edge("tools", "agent")

app = graph.compile()

# Run
result = app.invoke({
    "messages": [HumanMessage(content="What's the weather in Tokyo?")]
})

Part 3: RAG with LangGraph

Adaptive RAG Flow

# langgraph/rag_flow.py
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List, Literal
import operator

class RAGState(TypedDict):
    query: str
    messages: Annotated[List, operator.add]
    retrieved_docs: Annotated[List[Dict], operator.add]
    answer: str
    grade: Literal["relevant", "irrelevant", "hallucinated"]
    iterations: int

def retrieve(state: RAGState) -> RAGState:
    """Retrieve documents for query."""
    docs = retriever.invoke(state["query"])
    return {"retrieved_docs": docs}

def grade_documents(state: RAGState) -> RAGState:
    """Grade retrieved document relevance."""
    query = state["query"]
    docs = state["retrieved_docs"]

    grades = []
    for doc in docs:
        score = grader.invoke({"query": query, "document": doc})
        grades.append(score)

    relevant_docs = [d for d, g in zip(docs, grades) if g == "relevant"]
    return {
        "retrieved_docs": relevant_docs,
        "grade": "relevant" if relevant_docs else "irrelevant"
    }

def generate(state: RAGState) -> RAGState:
    """Generate answer from retrieved docs."""
    context = "\n".join([d["content"] for d in state["retrieved_docs"]])
    answer = generator.invoke({"query": state["query"], "context": context})
    return {"answer": answer}

def check_hallucination(state: RAGState) -> RAGState:
    """Verify answer is grounded in context."""
    score = hallucination_checker.invoke({
        "answer": state["answer"],
        "context": state["retrieved_docs"]
    })
    return {"grade": score}

def rewrite_query(state: RAGState) -> RAGState:
    """Rewrite query for better retrieval."""
    new_query = query_rewriter.invoke({"query": state["query"]})
    return {"query": new_query, "iterations": state["iterations"] + 1}

graph = StateGraph(RAGState)
graph.add_node("retrieve", retrieve)
graph.add_node("grade_docs", grade_documents)
graph.add_node("generate", generate)
graph.add_node("check_hallucination", check_hallucination)
graph.add_node("rewrite_query", rewrite_query)

graph.set_entry_point("retrieve")
graph.add_edge("retrieve", "grade_docs")

graph.add_conditional_edges(
    "grade_docs",
    lambda s: s["grade"],
    {"relevant": "generate", "irrelevant": "rewrite_query"}
)

graph.add_edge("generate", "check_hallucination")
graph.add_conditional_edges(
    "check_hallucination",
    lambda s: s["grade"] if s["iterations"] < 3 else "max_retries",
    {"relevant": END, "hallucinated": "generate", "max_retries": END}
)

graph.add_edge("rewrite_query", "retrieve")

app = graph.compile()

Part 4: Human-in-the-Loop

Checkpoint & Resume

# langgraph/human_in_loop.py
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List, Optional
import operator
import uuid

class HILState(TypedDict):
    query: str
    messages: Annotated[List, operator.add]
    answer: str
    requires_approval: bool
    human_feedback: Optional[str]
    approved: bool

def generate_draft(state: HILState) -> HILState:
    """Generate initial answer."""
    answer = llm.invoke(f"Answer: {state['query']}")
    return {
        "answer": answer.content,
        "requires_approval": True
    }

def human_review(state: HILState) -> HILState:
    """Wait for human input (pauses graph)."""
    # This node pauses execution
    # Human provides feedback via interrupt
    return state

def apply_feedback(state: HILState) -> HILState:
    """Incorporate human feedback."""
    if state["approved"]:
        return {"messages": [AIMessage(content=state["answer"])]}
    else:
        # Regenerate with feedback
        new_answer = llm.invoke(f"Revise: {state['answer']}\nFeedback: {state['human_feedback']}")
        return {"answer": new_answer.content, "requires_approval": True}

graph = StateGraph(HILState)
graph.add_node("generate", generate_draft)
graph.add_node("human_review", human_review)
graph.add_node("apply_feedback", apply_feedback)

graph.set_entry_point("generate")
graph.add_edge("generate", "human_review")
graph.add_edge("human_review", "apply_feedback")
graph.add_conditional_edges(
    "apply_feedback",
    lambda s: "approved" if s["approved"] else "revise",
    {"approved": END, "revise": "human_review"}
)

# Compile with checkpointer
checkpointer = SqliteSaver.from_conn_string("hil.db")
app = graph.compile(checkpointer=checkpointer, interrupt_before=["human_review"])

# Run - will pause at human_review
config = {"configurable": {"thread_id": str(uuid.uuid4())}}
result = app.invoke({"query": "Write a report", "messages": []}, config=config)

# Check if paused
state = app.get_state(config)
print(f"Next node: {state.next}")  # ('human_review',)

# Resume with human input
app.update_state(config, {"human_feedback": "Make it shorter", "approved": False})
result = app.invoke(None, config=config)  # Continues from interrupt

Streaming Human Feedback

# langgraph/streaming_hil.py
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List
import operator

class StreamingState(TypedDict):
    query: str
    chunks: Annotated[List[str], operator.add]
    complete: bool

def stream_generate(state: StreamingState, config) -> StreamingState:
    """Stream tokens, allow interruption."""
    full_answer = ""
    for chunk in llm.stream(state["query"]):
        # Check for interrupt signal
        if config.get("interrupt"):
            return {"chunks": [chunk], "complete": False}
        full_answer += chunk
        yield {"chunks": [chunk], "complete": False}

    return {"chunks": [full_answer], "complete": True}

# Client-side interrupt
async def run_with_interrupt(query: str):
    config = {"configurable": {"thread_id": "stream-1"}, "interrupt": False}

    async for chunk in app.astream({"query": query}, config=config):
        print(chunk, end="", flush=True)
        # User can press Ctrl+C to interrupt
        # Set config["interrupt"] = True to pause

    # Resume later with updated config

Part 5: Multi-Agent Systems

Supervisor Pattern

# langgraph/multi_agent.py
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List, Literal
import operator
from langchain_openai import ChatOpenAI

class MultiAgentState(TypedDict):
    query: str
    messages: Annotated[List, operator.add]
    active_agent: str
    agent_outputs: Annotated[Dict[str, str], operator.add]
    final_answer: str

llm = ChatOpenAI(model="gpt-4o")

# Specialist agents
def researcher(state: MultiAgentState) -> MultiAgentState:
    result = llm.invoke(f"Research: {state['query']}")
    return {
        "agent_outputs": {"researcher": result.content},
        "active_agent": "analyst"
    }

def analyst(state: MultiAgentState) -> MultiAgentState:
    research = state["agent_outputs"].get("researcher", "")
    result = llm.invoke(f"Analyze: {research}")
    return {
        "agent_outputs": {"analyst": result.content},
        "active_agent": "writer"
    }

def writer(state: MultiAgentState) -> MultiAgentState:
    research = state["agent_outputs"].get("researcher", "")
    analysis = state["agent_outputs"].get("analyst", "")
    result = llm.invoke(f"Write report based on:\nResearch: {research}\nAnalysis: {analysis}")
    return {
        "final_answer": result.content,
        "active_agent": "supervisor"
    }

def supervisor(state: MultiAgentState) -> MultiAgentState:
    """Route to next agent or end."""
    if state["active_agent"] == "supervisor":
        return {"active_agent": "end"}
    return {"active_agent": state["active_agent"]}

graph = StateGraph(MultiAgentState)
graph.add_node("researcher", researcher)
graph.add_node("analyst", analyst)
graph.add_node("writer", writer)
graph.add_node("supervisor", supervisor)

graph.set_entry_point("supervisor")
graph.add_conditional_edges(
    "supervisor",
    lambda s: s["active_agent"],
    {"researcher": "researcher", "analyst": "analyst", "writer": "writer", "end": END}
)
graph.add_edge("researcher", "supervisor")
graph.add_edge("analyst", "supervisor")
graph.add_edge("writer", "supervisor")

app = graph.compile()

Collaborative Agents with Shared State

# langgraph/collaborative.py
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List, Dict
import operator

class CollaborativeState(TypedDict):
    task: str
    messages: Annotated[List[Dict], operator.add]
    shared_context: Annotated[Dict, operator.add]  # Deep merge
    agent_notes: Annotated[Dict[str, List[str]], operator.add]
    consensus: bool

def agent_a(state: CollaborativeState) -> CollaborativeState:
    note = f"Agent A: {llm.invoke(f'A perspective on {state[\"task\"]}')}"
    return {
        "agent_notes": {"A": [note]},
        "shared_context": {"perspective_a": note}
    }

def agent_b(state: CollaborativeState) -> CollaborativeState:
    note = f"Agent B: {llm.invoke(f'B perspective on {state[\"task\"]}')}"
    return {
        "agent_notes": {"B": [note]},
        "shared_context": {"perspective_b": note}
    }

def synthesizer(state: CollaborativeState) -> CollaborativeState:
    notes = state["agent_notes"]
    consensus = llm.invoke(f"Synthesize: {notes}")
    return {
        "shared_context": {"consensus": consensus.content},
        "consensus": True
    }

graph = StateGraph(CollaborativeState)
graph.add_node("agent_a", agent_a)
graph.add_node("agent_b", agent_b)
graph.add_node("synthesizer", synthesizer)

graph.set_entry_point("agent_a")
graph.add_edge("agent_a", "agent_b")
graph.add_edge("agent_b", "synthesizer")
graph.add_edge("synthesizer", END)

# Run in parallel using subgraphs
from langgraph.graph import StateGraph

parallel_graph = StateGraph(CollaborativeState)
parallel_graph.add_node("agent_a", agent_a)
parallel_graph.add_node("agent_b", agent_b)
parallel_graph.set_entry_point("agent_a")
parallel_graph.add_edge("agent_a", "agent_b")  # Sequential for demo

# For true parallel, use subgraph with fan-out

Part 6: Advanced Patterns

Reflection & Self-Correction

# langgraph/reflection.py
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List, Literal
import operator

class ReflectionState(TypedDict):
    query: str
    draft: str
    critique: str
    revised: str
    iteration: int
    max_iterations: int
    status: Literal["drafting", "critiquing", "revising", "done"]

def generate_draft(state: ReflectionState) -> ReflectionState:
    draft = llm.invoke(f"Write: {state['query']}")
    return {"draft": draft.content, "status": "critiquing", "iteration": 1}

def critique(state: ReflectionState) -> ReflectionState:
    critique = llm.invoke(f"Critique this draft:\n{state['draft']}")
    return {"critique": critique.content, "status": "revising"}

def revise(state: ReflectionState) -> ReflectionState:
    revised = llm.invoke(f"Revise based on critique:\nDraft: {state['draft']}\nCritique: {state['critique']}")
    return {
        "revised": revised.content,
        "draft": revised.content,  # Update draft for next iteration
        "iteration": state["iteration"] + 1,
        "status": "critiquing" if state["iteration"] < state["max_iterations"] else "done"
    }

graph = StateGraph(ReflectionState)
graph.add_node("draft", generate_draft)
graph.add_node("critique", critique)
graph.add_node("revise", revise)

graph.set_entry_point("draft")
graph.add_edge("draft", "critique")
graph.add_edge("critique", "revise")
graph.add_conditional_edges(
    "revise",
    lambda s: s["status"],
    {"critiquing": "critique", "done": END}
)

app = graph.compile()

Planning & Execution

# langgraph/planning.py
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List, Dict
import operator
import json

class PlanState(TypedDict):
    goal: str
    plan: List[Dict]
    current_step: int
    results: Annotated[Dict[str, str], operator.add]
    completed: bool

def planner(state: PlanState) -> PlanState:
    """Create execution plan."""
    plan_prompt = f"""Create a step-by-step plan for: {state['goal']}
    Return JSON: [{{"step": 1, "action": "...", "tool": "..."}}]"""
    plan = json.loads(llm.invoke(plan_prompt).content)
    return {"plan": plan, "current_step": 0}

def executor(state: PlanState) -> PlanState:
    """Execute current step."""
    step = state["plan"][state["current_step"]]
    result = execute_step(step)
    return {
        "results": {f"step_{state['current_step']}": result},
        "current_step": state["current_step"] + 1
    }

def should_continue(state: PlanState) -> str:
    if state["current_step"] >= len(state["plan"]):
        return "done"
    return "continue"

graph = StateGraph(PlanState)
graph.add_node("plan", planner)
graph.add_node("execute", executor)
graph.set_entry_point("plan")
graph.add_edge("plan", "execute")
graph.add_conditional_edges("execute", should_continue, {"continue": "execute", "done": END})

app = graph.compile()

Part 7: Persistence & Scaling

Postgres Checkpointing

# langgraph/postgres_checkpoint.py
from langgraph.checkpoint.postgres import PostgresSaver
from psycopg_pool import ConnectionPool

# Connection pool
pool = ConnectionPool(
    conninfo="postgresql://user:pass@localhost/langgraph",
    max_size=20,
    kwargs={"autocommit": True}
)

checkpointer = PostgresSaver(pool)
checkpointer.setup()  # Creates tables

app = graph.compile(checkpointer=checkpointer)

# List all threads
threads = checkpointer.list(None)  # All threads
for thread in threads:
    print(f"Thread: {thread.config['configurable']['thread_id']}")
    print(f"  Steps: {len(thread.checkpoint['channel_values'].get('messages', []))}")

# Get specific thread state
state = app.get_state({"configurable": {"thread_id": "user-123"}})

Redis for High Throughput

# langgraph/redis_checkpoint.py
from langgraph.checkpoint.redis import RedisSaver
import redis

redis_client = redis.Redis(host="localhost", port=6379, decode_responses=True)
checkpointer = RedisSaver(redis_client)

app = graph.compile(checkpointer=checkpointer)

# TTL for auto-cleanup
checkpointer.ttl = 86400  # 24 hours

Part 8: Testing & Debugging

Unit Testing Graphs

# langgraph/testing.py
import pytest
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List
import operator

class TestState(TypedDict):
    value: int
    history: Annotated[List[int], operator.add]

def increment(state: TestState) -> TestState:
    return {"value": state["value"] + 1}

def test_graph():
    graph = StateGraph(TestState)
    graph.add_node("inc", increment)
    graph.set_entry_point("inc")
    graph.add_edge("inc", END)
    app = graph.compile()

    result = app.invoke({"value": 0, "history": []})
    assert result["value"] == 1
    assert result["history"] == [1]

def test_graph_with_mock(monkeypatch):
    """Mock external dependencies."""
    def mock_llm(*args, **kwargs):
        return "Mocked response"

    monkeypatch.setattr("langgraph.testing.llm", mock_llm)
    # Test with mocked LLM

Visualization & Debugging

# langgraph/visualization.py
from langgraph.graph import StateGraph
import mermaid

def visualize_graph(app):
    """Generate Mermaid diagram of graph."""
    graph = app.get_graph()

    mermaid_code = "graph TD\n"
    for node_id, node in graph.nodes.items():
        mermaid_code += f"  {node_id}[{node_id}]\n"

    for edge in graph.edges:
        mermaid_code += f"  {edge.source} --> {edge.target}\n"

    return mermaid_code

# Print graph structure
print(app.get_graph().draw_ascii())

# State inspection
state = app.get_state(config)
print(f"Current node: {state.next}")
print(f"State values: {state.values}")
print(f"Pending tasks: {state.tasks}")

Part 9: Production Deployment

FastAPI Server

# langgraph/api.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from langgraph.graph import StateGraph
import uvicorn
import uuid

app = FastAPI(title="LangGraph Agent API")

class QueryRequest(BaseModel):
    query: str
    thread_id: str = None
    config: dict = {}

class QueryResponse(BaseModel):
    thread_id: str
    answer: str
    state: dict

@app.post("/invoke", response_model=QueryResponse)
async def invoke_agent(request: QueryRequest):
    thread_id = request.thread_id or str(uuid.uuid4())
    config = {"configurable": {"thread_id": thread_id}, **request.config}

    try:
        result = await agent_app.ainvoke(
            {"query": request.query, "messages": []},
            config=config
        )
        return QueryResponse(
            thread_id=thread_id,
            answer=result.get("answer", ""),
            state=result
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.post("/stream")
async def stream_agent(request: QueryRequest):
    thread_id = request.thread_id or str(uuid.uuid4())
    config = {"configurable": {"thread_id": thread_id}}

    async def generate():
        async for chunk in agent_app.astream(
            {"query": request.query, "messages": []},
            config=config
        ):
            yield f"data: {chunk}\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")

@app.get("/state/{thread_id}")
async def get_state(thread_id: str):
    config = {"configurable": {"thread_id": thread_id}}
    state = agent_app.get_state(config)
    return state.values

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

Docker Deployment

# langgraph/Dockerfile
FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["uvicorn", "api:app", "--host", "0.0.0.0", "--port", "8000"]

# langgraph/docker-compose.yml
version: '3.8'

services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - POSTGRES_URL=postgresql://postgres:pass@db:5432/langgraph
    depends_on:
      - db
    deploy:
      replicas: 3

  db:
    image: postgres:15
    environment:
      - POSTGRES_DB=langgraph
      - POSTGRES_PASSWORD=pass
    volumes:
      - pgdata:/var/lib/postgresql/data

volumes:
  pgdata:

Part 10: Monitoring & Observability

LangSmith Integration

# langgraph/monitoring.py
from langsmith import Client
from langgraph.graph import StateGraph
from langgraph.callbacks import LangGraphTracer

client = Client()

# Trace runs
tracer = LangGraphTracer(project_name="langgraph-agents")

app = graph.compile()
app = app.with_config({"callbacks": [tracer]})

# Custom metrics
def log_metrics(state, config):
    client.create_run(
        name="agent_step",
        run_type="chain",
        inputs={"state": state},
        outputs={},
        extra={
            "metrics": {
                "tokens": state.get("token_count", 0),
                "latency_ms": state.get("latency_ms", 0),
            }
        }
    )

Resources

LangGraph transforms agents from linear chains into stateful, controllable systems. Master the graph, master the agent.