Production RAG - Enterprise Systems

Building production RAG systems requires considering performance, scalability, monitoring, and security. It is like designing an entire ecosystem - every element must work harmoniously with the others!

Production RAG Architecture

1┌─────────────────────────────────────────────────────────────────────┐
2│                    Production RAG Architecture                       │
3├─────────────────────────────────────────────────────────────────────┤
4│                                                                      │
5│  ┌──────────┐    ┌──────────────┐    ┌──────────────┐              │
6│  │  Client  │───▶│ Load Balancer│───▶│  API Gateway │              │
7│  └──────────┘    └──────────────┘    └──────────────┘              │
8│                                               │                      │
9│                                               ▼                      │
10│  ┌────────────────────────────────────────────────────────────┐    │
11│  │                     RAG Service Layer                       │    │
12│  │  ┌────────────┐  ┌────────────┐  ┌────────────┐           │    │
13│  │  │  Retrieval │  │ Augment    │  │ Generation │           │    │
14│  │  │  Service   │  │ Service    │  │ Service    │           │    │
15│  │  └─────┬──────┘  └─────┬──────┘  └─────┬──────┘           │    │
16│  └────────┼───────────────┼───────────────┼──────────────────┘    │
17│           │               │               │                        │
18│  ┌────────▼───────┐  ┌────▼────┐  ┌──────▼──────┐                │
19│  │ Vector Database│  │  Cache  │  │  LLM APIs   │                │
20│  │ (Pinecone/     │  │ (Redis) │  │ (OpenAI/    │                │
21│  │  Qdrant)       │  │         │  │  Anthropic) │                │
22│  └────────────────┘  └─────────┘  └─────────────┘                │
23│                                                                      │
24│  ┌───────────────────────────────────────────────────────────────┐ │
25│  │                    Observability Layer                         │ │
26│  │  Prometheus │ Grafana │ Jaeger │ LangSmith │ Weights & Biases │ │
27│  └───────────────────────────────────────────────────────────────┘ │
28└─────────────────────────────────────────────────────────────────────┘

FastAPI RAG Service

1from fastapi import FastAPI, HTTPException, BackgroundTasks
2from pydantic import BaseModel
3from typing import Optional
4import asyncio
5from contextlib import asynccontextmanager
6import uvicorn
7
8# Models
9class QueryRequest(BaseModel):
10    question: str
11    top_k: int = 5
12    filters: Optional[dict] = None
13
14class QueryResponse(BaseModel):
15    answer: str
16    sources: list[dict]
17    latency_ms: float
18
19class RAGService:
20    """Production RAG service."""
21
22    def __init__(self):
23        self.vector_store = None
24        self.llm = None
25        self.cache = None
26
27    async def initialize(self):
28        """Initialize connections."""
29        # Vector store
30        from qdrant_client import AsyncQdrantClient
31        self.vector_store = AsyncQdrantClient(host="qdrant", port=6333)
32
33        # Cache
34        import redis.asyncio as redis
35        self.cache = redis.Redis(host="redis", port=6379)
36
37        # LLM
38        from openai import AsyncOpenAI
39        self.llm = AsyncOpenAI()
40
41    async def query(self, request: QueryRequest) -> QueryResponse:
42        """Processes a RAG query."""
43        import time
44        start = time.time()
45
46        # 1. Check cache
47        cache_key = f"rag:{hash(request.question)}"
48        cached = await self.cache.get(cache_key)
49        if cached:
50            return QueryResponse.model_validate_json(cached)
51
52        # 2. Embedding
53        embedding = await self._get_embedding(request.question)
54
55        # 3. Retrieval
56        docs = await self._retrieve(embedding, request.top_k, request.filters)
57
58        # 4. Generation
59        answer = await self._generate(request.question, docs)
60
61        # 5. Response
62        response = QueryResponse(
63            answer=answer,
64            sources=[{"text": d.text, "score": d.score} for d in docs],
65            latency_ms=(time.time() - start) * 1000
66        )
67
68        # 6. Cache
69        await self.cache.setex(cache_key, 3600, response.model_dump_json())
70
71        return response
72
73    async def _get_embedding(self, text: str) -> list[float]:
74        """Generates an embedding."""
75        response = await self.llm.embeddings.create(
76            model="text-embedding-3-small",
77            input=text
78        )
79        return response.data[0].embedding
80
81    async def _retrieve(self, embedding, top_k, filters):
82        """Retrieves documents."""
83        return await self.vector_store.search(
84            collection_name="documents",
85            query_vector=embedding,
86            limit=top_k,
87            query_filter=filters
88        )
89
90    async def _generate(self, question: str, docs) -> str:
91        """Generates an answer."""
92        context = "\n".join([d.payload["text"] for d in docs])
93
94        response = await self.llm.chat.completions.create(
95            model="gpt-4o-mini",
96            messages=[
97                {"role": "system", "content": f"Context:\n{context}"},
98                {"role": "user", "content": question}
99            ]
100        )
101        return response.choices[0].message.content
102
103# FastAPI app
104rag_service = RAGService()
105
106@asynccontextmanager
107async def lifespan(app: FastAPI):
108    await rag_service.initialize()
109    yield
110
111app = FastAPI(title="RAG API", lifespan=lifespan)
112
113@app.post("/query", response_model=QueryResponse)
114async def query(request: QueryRequest):
115    try:
116        return await rag_service.query(request)
117    except Exception as e:
118        raise HTTPException(status_code=500, detail=str(e))

Monitoring and Observability

1from prometheus_client import Counter, Histogram, Gauge
2import logging
3import structlog
4
5# Prometheus metrics
6QUERY_COUNT = Counter(
7    "rag_queries_total",
8    "Total number of RAG queries",
9    ["status"]
10)
11
12QUERY_LATENCY = Histogram(
13    "rag_query_latency_seconds",
14    "Query latency in seconds",
15    buckets=[0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0]
16)
17
18RETRIEVAL_SCORE = Gauge(
19    "rag_retrieval_score",
20    "Average retrieval score"
21)
22
23# Structured logging
24structlog.configure(
25    processors=[
26        structlog.stdlib.add_log_level,
27        structlog.processors.TimeStamper(fmt="iso"),
28        structlog.processors.JSONRenderer()
29    ]
30)
31logger = structlog.get_logger()
32
33class MonitoredRAGService:
34    """RAG with monitoring."""
35
36    async def query(self, request: QueryRequest) -> QueryResponse:
37        with QUERY_LATENCY.time():
38            try:
39                response = await self._process_query(request)
40                QUERY_COUNT.labels(status="success").inc()
41
42                # Log
43                logger.info(
44                    "query_completed",
45                    question=request.question[:50],
46                    latency_ms=response.latency_ms,
47                    num_sources=len(response.sources)
48                )
49
50                return response
51            except Exception as e:
52                QUERY_COUNT.labels(status="error").inc()
53                logger.error("query_failed", error=str(e))
54                raise

LangSmith Tracing

1from langsmith import Client, traceable
2from langsmith.wrappers import wrap_openai
3
4# Initialize
5client = Client()
6
7# Wrap OpenAI client
8from openai import OpenAI
9openai_client = wrap_openai(OpenAI())
10
11@traceable(name="RAG Query")
12async def traced_query(question: str) -> str:
13    """RAG query with tracing."""
14
15    # Embedding
16    with client.trace("embedding") as span:
17        embedding = get_embedding(question)
18        span.metadata = {"model": "text-embedding-3-small"}
19
20    # Retrieval
21    with client.trace("retrieval") as span:
22        docs = retrieve(embedding)
23        span.metadata = {"num_docs": len(docs)}
24
25    # Generation
26    with client.trace("generation") as span:
27        response = generate(question, docs)
28        span.metadata = {"model": "gpt-4o-mini"}
29
30    return response
31
32# Evaluation in LangSmith
33from langsmith.evaluation import evaluate
34
35def evaluate_rag():
36    """Evaluates the RAG system."""
37
38    # Dataset with questions and expected answers
39    dataset = client.read_dataset("rag-eval-dataset")
40
41    results = evaluate(
42        traced_query,
43        data=dataset,
44        evaluators=[
45            "qa_helpfulness",
46            "context_precision",
47            "faithfulness"
48        ]
49    )
50
51    return results

Rate Limiting and Security

1from fastapi import Depends, HTTPException
2from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
3import time
4
5# Rate limiter
6class RateLimiter:
7    """Simple rate limiter."""
8
9    def __init__(self, requests_per_minute: int = 60):
10        self.rpm = requests_per_minute
11        self.requests: dict[str, list[float]] = {}
12
13    async def check(self, user_id: str) -> bool:
14        now = time.time()
15        minute_ago = now - 60
16
17        if user_id not in self.requests:
18            self.requests[user_id] = []
19
20        # Remove old requests
21        self.requests[user_id] = [
22            t for t in self.requests[user_id] if t > minute_ago
23        ]
24
25        if len(self.requests[user_id]) >= self.rpm:
26            return False
27
28        self.requests[user_id].append(now)
29        return True
30
31rate_limiter = RateLimiter()
32security = HTTPBearer()
33
34async def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
35    """Token verification."""
36    # In production: JWT verification
37    if credentials.credentials == "invalid":
38        raise HTTPException(status_code=401, detail="Invalid token")
39    return credentials.credentials
40
41async def rate_limit(user_id: str = Depends(verify_token)):
42    """Rate limiting middleware."""
43    if not await rate_limiter.check(user_id):
44        raise HTTPException(status_code=429, detail="Too many requests")
45    return user_id
46
47@app.post("/query")
48async def query(request: QueryRequest, user: str = Depends(rate_limit)):
49    return await rag_service.query(request)

Docker Deployment

1# Dockerfile
2FROM python:3.11-slim
3
4WORKDIR /app
5
6# Install dependencies
7COPY requirements.txt .
8RUN pip install --no-cache-dir -r requirements.txt
9
10# Copy code
11COPY . .
12
13# Run
14CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

1# docker-compose.yml
2version: '3.8'
3
4services:
5  rag-api:
6    build: .
7    ports:
8      - "8000:8000"
9    environment:
10      - OPENAI_API_KEY=${OPENAI_API_KEY}
11      - QDRANT_HOST=qdrant
12      - REDIS_HOST=redis
13    depends_on:
14      - qdrant
15      - redis
16
17  qdrant:
18    image: qdrant/qdrant:latest
19    ports:
20      - "6333:6333"
21    volumes:
22      - qdrant_data:/qdrant/storage
23
24  redis:
25    image: redis:alpine
26    ports:
27      - "6379:6379"
28
29  prometheus:
30    image: prom/prometheus
31    ports:
32      - "9090:9090"
33    volumes:
34      - ./prometheus.yml:/etc/prometheus/prometheus.yml
35
36  grafana:
37    image: grafana/grafana
38    ports:
39      - "3000:3000"
40
41volumes:
42  qdrant_data:

Best Practices

1"""
2Production RAG Best Practices:
3
41. RETRIEVAL
5   - Use hybrid search (vector + keyword)
6   - Implement reranking
7   - Filter by metadata
8
92. CHUNKING
10   - Experiment with chunk sizes
11   - Use overlap
12   - Consider semantic chunking
13
143. CACHING
15   - Cache embeddings
16   - Cache frequent queries
17   - Invalidate on document updates
18
194. MONITORING
20   - Track latency at each stage
21   - Monitor retrieval quality
22   - Alert on anomalies
23
245. SECURITY
25   - Rate limiting
26   - Input sanitization
27   - PII detection and filtering
28
296. SCALABILITY
30   - Horizontal scaling API
31   - Sharding vector database
32   - Async processing
33"""

Congratulations! You have learned about advanced RAG and Multi-Agent systems. In the next lesson you will learn about LangGraph - a framework for building complex agent workflows!