RAG API: 10 Design Patterns Every Top System Uses

TL;DR

The best RAG APIs do not just return text. They implement 10 critical design patterns: SSE streaming, conversation threads, source attribution, confidence scores, fallback handling, rate limiting, authentication, versioning, webhooks, and batch processing. This guide details each pattern with production-ready FastAPI examples.

Why RAG API design matters

A poorly designed RAG API is expensive:

The 10 patterns below are used by OpenAI, Anthropic, Cohere, and the best RAG products on the market.

Pattern 1: SSE Streaming (Server-Sent Events)

Streaming is non-negotiable for modern RAG UX. Users see the response token by token instead of waiting 3-5 seconds.

FastAPI implementation

DEVELOPERpython
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from openai import OpenAI
import json

app = FastAPI()
client = OpenAI()

@app.post("/api/v1/chat/stream")
async def stream_chat(request: Request):
    body = await request.json()
    query = body["message"]
    conversation_id = body.get("conversation_id")

    async def event_generator():
        # Phase 1: Retrieval (send status event)
        yield f"data: {json.dumps({'type': 'status', 'content': 'searching'})}\n\n"

        contexts = await retrieve_documents(query)

        # Phase 2: Send sources BEFORE the response
        sources = [{"title": c.title, "url": c.url, "score": c.score}
                   for c in contexts]
        yield f"data: {json.dumps({'type': 'sources', 'content': sources})}\n\n"

        # Phase 3: Stream the response
        stream = client.chat.completions.create(
            model="gpt-5.1",
            messages=build_messages(query, contexts, conversation_id),
            stream=True
        )

        full_response = ""
        for chunk in stream:
            if chunk.choices[0].delta.content:
                token = chunk.choices[0].delta.content
                full_response += token
                yield f"data: {json.dumps({'type': 'token', 'content': token})}\n\n"

        # Phase 4: Final metadata
        yield f"data: {json.dumps({'type': 'done', 'metadata': {'tokens_used': len(full_response.split()), 'model': 'gpt-5.1', 'conversation_id': conversation_id}})}\n\n"

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no"  # Nginx: disable buffering
        }
    )

Client-side (JavaScript)

DEVELOPERjavascript
const eventSource = new EventSource('/api/v1/chat/stream', {
  method: 'POST',
  headers: { 'Content-Type': 'application/json' },
  body: JSON.stringify({ message: 'How does RAG work?' })
});

eventSource.onmessage = (event) => {
  const data = JSON.parse(event.data);

  switch (data.type) {
    case 'status':
      showLoader(data.content);
      break;
    case 'sources':
      renderSources(data.content);
      break;
    case 'token':
      appendToken(data.content);
      break;
    case 'done':
      hideLoader();
      logMetadata(data.metadata);
      break;
  }
};

Pattern 2: Conversation Threads

Every conversation must have a unique identifier to maintain context.

Data schema

DEVELOPERpython
from pydantic import BaseModel, Field
from datetime import datetime
from uuid import uuid4

class Message(BaseModel):
    id: str = Field(default_factory=lambda: str(uuid4()))
    role: str  # "user" | "assistant" | "system"
    content: str
    sources: list[dict] = []
    metadata: dict = {}
    created_at: datetime = Field(default_factory=datetime.utcnow)

class Conversation(BaseModel):
    id: str = Field(default_factory=lambda: str(uuid4()))
    messages: list[Message] = []
    metadata: dict = {}
    created_at: datetime = Field(default_factory=datetime.utcnow)
    updated_at: datetime = Field(default_factory=datetime.utcnow)

# API endpoints
@app.post("/api/v1/conversations")
async def create_conversation():
    conv = Conversation()
    await db.save_conversation(conv)
    return {"conversation_id": conv.id}

@app.post("/api/v1/conversations/{conv_id}/messages")
async def send_message(conv_id: str, body: dict):
    conversation = await db.get_conversation(conv_id)
    if not conversation:
        raise HTTPException(404, "Conversation not found")

    # Add user message
    user_msg = Message(role="user", content=body["message"])
    conversation.messages.append(user_msg)

    # Generate RAG response with conversation context
    response = await rag_pipeline.query(
        query=body["message"],
        history=conversation.messages[-10:]  # Last 10 messages
    )

    assistant_msg = Message(
        role="assistant",
        content=response.answer,
        sources=response.sources,
        metadata={"model": response.model, "tokens": response.tokens}
    )
    conversation.messages.append(assistant_msg)
    await db.update_conversation(conversation)

    return assistant_msg.dict()

Pattern 3: Source Attribution

Every response must cite its sources with clickable references.

Response format with sources

DEVELOPERpython
class SourceReference(BaseModel):
    id: str
    title: str
    url: str | None = None
    relevance_score: float  # 0.0 - 1.0
    snippet: str  # Extract from the passage used
    page: int | None = None
    section: str | None = None

class RAGResponse(BaseModel):
    answer: str
    sources: list[SourceReference]
    confidence: float  # Overall confidence score
    model: str
    tokens_used: int
    latency_ms: int

# Example response
{
    "answer": "Delivery times in metropolitan France are 2-5 business days [1]. For express delivery, expect 24h [1]. Overseas territories require 7-14 days [2].",
    "sources": [
        {
            "id": "src_001",
            "title": "Shipping Policy",
            "url": "/docs/shipping",
            "relevance_score": 0.94,
            "snippet": "Standard delivery: 2-5 business days. Express: 24h.",
            "section": "Metropolitan delivery times"
        },
        {
            "id": "src_002",
            "title": "Shipping FAQ",
            "url": "/faq/shipping",
            "relevance_score": 0.87,
            "snippet": "Overseas territories: 7-14 days depending on destination.",
            "section": "Overseas"
        }
    ],
    "confidence": 0.91,
    "model": "gpt-5.1",
    "tokens_used": 342,
    "latency_ms": 1850
}

Pattern 4: Confidence Scores

Tell the user how reliable the response is.

Confidence score calculation

DEVELOPERpython
def compute_confidence(
    retrieval_scores: list[float],
    answer: str,
    contexts: list[str]
) -> dict:
    """Compute a multi-factor confidence score."""

    # Factor 1: Retrieval quality (max score of top document)
    retrieval_confidence = max(retrieval_scores) if retrieval_scores else 0.0

    # Factor 2: Coverage (how many docs contribute)
    coverage = len([s for s in retrieval_scores if s > 0.7]) / max(len(retrieval_scores), 1)

    # Factor 3: Answer length (too short = suspicious)
    length_factor = min(len(answer.split()) / 20, 1.0)

    # Composite score
    confidence = (
        retrieval_confidence * 0.5 +
        coverage * 0.3 +
        length_factor * 0.2
    )

    return {
        "overall": round(confidence, 3),
        "retrieval": round(retrieval_confidence, 3),
        "coverage": round(coverage, 3),
        "detail": {
            "top_doc_score": retrieval_scores[0] if retrieval_scores else 0,
            "docs_above_threshold": len([s for s in retrieval_scores if s > 0.7]),
            "answer_length": len(answer.split())
        }
    }

Confidence thresholds and actions

Pattern 5: Fallback Handling

When RAG cannot answer, handle it gracefully.

DEVELOPERpython
class FallbackHandler:
    def __init__(self, confidence_threshold: float = 0.4):
        self.threshold = confidence_threshold

    async def handle(self, query: str, rag_result: dict) -> dict:
        confidence = rag_result["confidence"]["overall"]

        if confidence >= self.threshold:
            return rag_result

        # Cascading fallback strategy
        fallbacks = [
            self._try_broader_search,
            self._try_faq_match,
            self._graceful_decline
        ]

        for fallback in fallbacks:
            result = await fallback(query, rag_result)
            if result:
                return result

        return self._graceful_decline(query, rag_result)

    async def _try_broader_search(self, query, original):
        """Broaden the search (fewer filters)."""
        broader_result = await rag_pipeline.query(
            query, filters=None, top_k=20
        )
        if broader_result["confidence"]["overall"] >= self.threshold:
            broader_result["fallback"] = "broader_search"
            return broader_result
        return None

    async def _try_faq_match(self, query, original):
        """Search pre-indexed FAQs."""
        faq_match = await faq_index.search(query, threshold=0.8)
        if faq_match:
            return {
                "answer": faq_match.answer,
                "sources": [{"title": "FAQ", "url": faq_match.url}],
                "confidence": {"overall": 0.85},
                "fallback": "faq_match"
            }
        return None

    def _graceful_decline(self, query, original):
        """Politely decline with suggestions."""
        return {
            "answer": "I could not find sufficiently reliable information to answer this question. Here is what I can suggest:",
            "suggestions": [
                "Rephrase your question with different terms",
                "Check our help center",
                "Contact our support team"
            ],
            "confidence": {"overall": 0.0},
            "fallback": "declined"
        }

Pattern 6: Smart Rate Limiting

Protect your API AND your LLM bill.

DEVELOPERpython
from fastapi import Depends, HTTPException
from datetime import datetime, timedelta
import redis.asyncio as redis

class RateLimiter:
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client

    async def check_rate_limit(
        self,
        api_key: str,
        plan: str = "free"
    ) -> dict:
        """Multi-tier rate limiting."""
        limits = {
            "free": {"rpm": 10, "rpd": 100, "tokens_per_day": 50_000},
            "pro": {"rpm": 60, "rpd": 5_000, "tokens_per_day": 1_000_000},
            "enterprise": {"rpm": 300, "rpd": 50_000, "tokens_per_day": 10_000_000}
        }

        plan_limits = limits.get(plan, limits["free"])

        # Check RPM (requests per minute)
        minute_key = f"rate:{api_key}:minute:{datetime.now().strftime('%Y%m%d%H%M')}"
        rpm_count = await self.redis.incr(minute_key)
        await self.redis.expire(minute_key, 60)

        if rpm_count > plan_limits["rpm"]:
            raise HTTPException(
                status_code=429,
                detail={
                    "error": "rate_limit_exceeded",
                    "limit": plan_limits["rpm"],
                    "reset_at": (datetime.now() + timedelta(minutes=1)).isoformat(),
                    "type": "requests_per_minute"
                },
                headers={"Retry-After": "60"}
            )

        # Check RPD (requests per day)
        day_key = f"rate:{api_key}:day:{datetime.now().strftime('%Y%m%d')}"
        rpd_count = await self.redis.incr(day_key)
        await self.redis.expire(day_key, 86400)

        if rpd_count > plan_limits["rpd"]:
            raise HTTPException(status_code=429, detail={
                "error": "daily_limit_exceeded",
                "limit": plan_limits["rpd"]
            })

        return {
            "remaining_rpm": plan_limits["rpm"] - rpm_count,
            "remaining_rpd": plan_limits["rpd"] - rpd_count
        }

Pattern 7: Multi-Level Authentication

DEVELOPERpython
from fastapi import Security, HTTPException
from fastapi.security import HTTPBearer, APIKeyHeader
import jwt

security = HTTPBearer()
api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)

async def authenticate(
    bearer: str = Security(security, auto_error=False),
    api_key: str = Security(api_key_header, auto_error=False)
) -> dict:
    """Flexible authentication: Bearer token OR API key."""

    if api_key:
        key_data = await db.get_api_key(api_key)
        if not key_data or not key_data.is_active:
            raise HTTPException(401, "Invalid API key")
        return {"type": "api_key", "user_id": key_data.user_id,
                "plan": key_data.plan}

    if bearer:
        try:
            payload = jwt.decode(bearer.credentials, SECRET_KEY,
                                algorithms=["HS256"])
            return {"type": "jwt", "user_id": payload["sub"],
                    "plan": payload.get("plan", "free")}
        except jwt.ExpiredSignatureError:
            raise HTTPException(401, "Token expired")
        except jwt.InvalidTokenError:
            raise HTTPException(401, "Invalid token")

    raise HTTPException(401, "Authentication required")

Pattern 8: API Versioning

DEVELOPERpython
from fastapi import APIRouter

v1_router = APIRouter(prefix="/api/v1", tags=["v1"])
v2_router = APIRouter(prefix="/api/v2", tags=["v2"])

# V1: original response format
@v1_router.post("/query")
async def query_v1(body: dict):
    result = await rag_pipeline.query(body["message"])
    return {"answer": result.answer, "sources": result.sources}

# V2: enriched format with metadata
@v2_router.post("/query")
async def query_v2(body: QueryRequestV2):
    result = await rag_pipeline.query(body.message, options=body.options)
    return {
        "data": {
            "answer": result.answer,
            "sources": result.sources,
            "confidence": result.confidence,
            "metadata": result.metadata
        },
        "usage": {
            "tokens_input": result.tokens_in,
            "tokens_output": result.tokens_out,
            "cost_usd": result.estimated_cost
        },
        "api_version": "2.0"
    }

# Deprecation header for V1
@v1_router.middleware("http")
async def add_deprecation_header(request, call_next):
    response = await call_next(request)
    response.headers["Deprecation"] = "true"
    response.headers["Sunset"] = "2026-12-31"
    response.headers["Link"] = '</api/v2/query>; rel="successor-version"'
    return response

app.include_router(v1_router)
app.include_router(v2_router)

Pattern 9: Webhook Callbacks

For long-running operations (document ingestion, batch processing).

DEVELOPERpython
from fastapi import BackgroundTasks

@app.post("/api/v1/documents/ingest")
async def ingest_document(body: dict, background_tasks: BackgroundTasks):
    """Asynchronous ingestion with webhook callback."""
    job_id = str(uuid4())

    background_tasks.add_task(
        process_ingestion,
        job_id=job_id,
        document_url=body["url"],
        webhook_url=body.get("webhook_url"),
        api_key=body.get("api_key")
    )

    return {
        "job_id": job_id,
        "status": "processing",
        "status_url": f"/api/v1/jobs/{job_id}"
    }

async def process_ingestion(job_id, document_url, webhook_url, api_key):
    """Background processing with webhook notification."""
    try:
        doc = await download_document(document_url)
        chunks = chunk_document(doc)
        embeddings = await embed_chunks(chunks)
        await store_in_qdrant(embeddings)

        result = {
            "job_id": job_id,
            "status": "completed",
            "chunks_created": len(chunks),
            "document_id": doc.id
        }
    except Exception as e:
        result = {"job_id": job_id, "status": "failed", "error": str(e)}

    if webhook_url:
        async with httpx.AsyncClient() as client:
            await client.post(
                webhook_url, json=result,
                headers={"X-Webhook-Secret": api_key}
            )

Pattern 10: Batch Processing

Process multiple queries in a single API call to reduce latency and costs.

DEVELOPERpython
from pydantic import BaseModel
from asyncio import gather

class BatchQuery(BaseModel):
    id: str
    message: str
    options: dict = {}

class BatchRequest(BaseModel):
    queries: list[BatchQuery]
    max_concurrent: int = 5

@app.post("/api/v1/batch/query")
async def batch_query(request: BatchRequest):
    """Batch processing with controlled concurrency."""
    import asyncio

    semaphore = asyncio.Semaphore(request.max_concurrent)
    results = []

    async def process_single(query: BatchQuery):
        async with semaphore:
            try:
                result = await rag_pipeline.query(query.message, **query.options)
                return {
                    "id": query.id,
                    "status": "success",
                    "answer": result.answer,
                    "sources": result.sources,
                    "confidence": result.confidence
                }
            except Exception as e:
                return {"id": query.id, "status": "error", "error": str(e)}

    tasks = [process_single(q) for q in request.queries]
    results = await gather(*tasks)

    return {
        "results": results,
        "total": len(results),
        "successful": sum(1 for r in results if r["status"] == "success"),
        "failed": sum(1 for r in results if r["status"] == "error")
    }

REST vs GraphQL vs WebSocket for RAG

Our recommendation: REST + SSE for 90% of RAG use cases. WebSocket only for interactive real-time chat with typing indicators.

Error Handling

Standardized error structure

DEVELOPERpython
class RAGError(BaseModel):
    error: str
    code: str
    message: str
    details: dict = {}
    request_id: str

ERROR_RESPONSES = {
    "retrieval_failed": {
        "status": 503,
        "message": "Unable to search the knowledge base. Please retry."
    },
    "llm_timeout": {
        "status": 504,
        "message": "Response generation timed out. Please try a shorter question."
    },
    "context_too_long": {
        "status": 400,
        "message": "Your question with conversation history exceeds the context limit."
    },
    "no_relevant_docs": {
        "status": 200,  # Not a technical error
        "message": "No relevant information found for your question."
    }
}

Our API at Ailog

At Ailog, our RAG API implements all 10 patterns described in this guide. Here is what our clients get:

• SSE streaming with sources sent upfront
• Multi-channel: widget, API, team chat
• Smart rate limiting per plan
• Webhooks for document ingestion
• Python and TypeScript SDKs for easy integration

Check out our guide on RAG streaming and production deployment.

Conclusion

The 10 design patterns described here are not optional for a production RAG API. They constitute the minimum quality foundation that your users and clients expect:

1. SSE Streaming - Instant UX
2. Conversation threads - Context maintained
3. Sources - Trust and transparency
4. Confidence scores - Informed decisions
5. Fallbacks - Resilience
6. Rate limiting - Cost protection
7. Authentication - Security
8. Versioning - Stability
9. Webhooks - Asynchronous operations
10. Batch - Efficiency

Start with patterns 1 through 5, then add the rest as needed.

Want a RAG API that implements all these patterns effortlessly? Try the Ailog API - everything is ready, just connect your documents.