API RAG : 10 Design Patterns que les Meilleurs Systemes Utilisent Tous

TL;DR

Les meilleures API RAG ne se contentent pas de renvoyer du texte. Elles implementent 10 design patterns critiques : streaming SSE, threads de conversation, attribution des sources, scores de confiance, fallback handling, rate limiting, authentification, versioning, webhooks et batch processing. Ce guide detaille chaque pattern avec des exemples FastAPI prets pour la production.

Pourquoi le design d'API RAG est critique

Une API RAG mal concue coute cher :

Les 10 patterns ci-dessous sont utilises par OpenAI, Anthropic, Cohere et les meilleurs produits RAG du marche.

Pattern 1 : Streaming SSE (Server-Sent Events)

Le streaming est non-negociable pour une UX RAG moderne. L'utilisateur voit la reponse token par token au lieu d'attendre 3-5 secondes.

Implementation FastAPI

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 (envoyer un event de statut)
        yield f"data: {json.dumps({'type': 'status', 'content': 'searching'})}\n\n"

        contexts = await retrieve_documents(query)

        # Phase 2 : Envoyer les sources AVANT la reponse
        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 : Streaming de la reponse
        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 : Metadata finale
        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 : desactiver le buffering
        }
    )

Client-side (JavaScript)

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

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 : Threads de conversation

Chaque conversation doit avoir un identifiant unique pour maintenir le contexte.

Schema de donnees

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 endpoint
@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")

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

    # Generer la reponse RAG avec contexte de conversation
    response = await rag_pipeline.query(
        query=body["message"],
        history=conversation.messages[-10:]  # Derniers 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 : Attribution des sources

Chaque reponse doit citer ses sources avec des references cliquables.

Format de reponse avec sources

DEVELOPERpython
class SourceReference(BaseModel):
    id: str
    title: str
    url: str | None = None
    relevance_score: float  # 0.0 - 1.0
    snippet: str  # Extrait du passage utilise
    page: int | None = None
    section: str | None = None

class RAGResponse(BaseModel):
    answer: str
    sources: list[SourceReference]
    confidence: float  # Score de confiance global
    model: str
    tokens_used: int
    latency_ms: int

# Exemple de reponse
{
    "answer": "Les delais de livraison en France metropolitaine sont de 2 a 5 jours ouvrables [1]. Pour la livraison express, comptez 24h [1]. Les DOM-TOM necessitent 7 a 14 jours [2].",
    "sources": [
        {
            "id": "src_001",
            "title": "Politique de livraison",
            "url": "/docs/livraison",
            "relevance_score": 0.94,
            "snippet": "Delais standard : 2-5 jours ouvrables. Express : 24h.",
            "section": "Delais metropolitains"
        },
        {
            "id": "src_002",
            "title": "FAQ Expedition",
            "url": "/faq/expedition",
            "relevance_score": 0.87,
            "snippet": "DOM-TOM : 7 a 14 jours selon la destination.",
            "section": "Outre-mer"
        }
    ],
    "confidence": 0.91,
    "model": "gpt-5.1",
    "tokens_used": 342,
    "latency_ms": 1850
}

Inline citations

DEVELOPERpython
def add_inline_citations(answer: str, sources: list[dict]) -> str:
    """Ajoute des references numerotees dans la reponse."""
    # Le LLM genere deja les [1], [2], etc.
    # On ajoute la correspondance source <-> numero
    citation_map = {}
    for i, source in enumerate(sources, 1):
        citation_map[f"[{i}]"] = source["id"]

    return answer, citation_map

Pattern 4 : Scores de confiance

Indiquer a l'utilisateur a quel point la reponse est fiable.

Calcul du score de confiance

DEVELOPERpython
def compute_confidence(
    retrieval_scores: list[float],
    answer: str,
    contexts: list[str]
) -> dict:
    """Calcule un score de confiance multi-facteurs."""

    # Facteur 1 : Qualite du retrieval (score max du top document)
    retrieval_confidence = max(retrieval_scores) if retrieval_scores else 0.0

    # Facteur 2 : Couverture (combien de docs contribuent)
    coverage = len([s for s in retrieval_scores if s > 0.7]) / max(len(retrieval_scores), 1)

    # Facteur 3 : Longueur de reponse (trop court = suspect)
    length_factor = min(len(answer.split()) / 20, 1.0)

    # Score composite
    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())
        }
    }

Seuils de confiance et actions

Pattern 5 : Fallback handling

Quand le RAG ne peut pas repondre, il faut gerer gracieusement.

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

        # Strategie de fallback en cascade
        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):
        """Elargir la recherche (moins de filtres)."""
        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):
        """Chercher dans les FAQ pre-indexees."""
        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):
        """Decliner poliment avec suggestions."""
        return {
            "answer": "Je n'ai pas trouve d'information suffisamment fiable pour repondre a cette question. Voici ce que je peux suggerer :",
            "suggestions": [
                "Reformulez votre question avec des termes differents",
                "Consultez notre centre d'aide",
                "Contactez notre support"
            ],
            "confidence": {"overall": 0.0},
            "fallback": "declined"
        }

Pattern 6 : Rate limiting intelligent

Proteger votre API ET votre facture LLM.

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:
        """Rate limiting multi-niveaux."""
        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"])

        # Verifier RPM (requetes par 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"}
            )

        # Verifier RPD (requetes par jour)
        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
        }

# Usage dans l'endpoint
@app.post("/api/v1/query")
async def query_rag(
    body: dict,
    rate_info: dict = Depends(rate_limiter.check_rate_limit)
):
    # Headers de rate limiting dans la reponse
    response = await process_query(body)
    return JSONResponse(
        content=response,
        headers={
            "X-RateLimit-Remaining": str(rate_info["remaining_rpm"]),
            "X-RateLimit-Limit": "60",
            "X-RateLimit-Reset": "60"
        }
    )

Pattern 7 : Authentification multi-niveaux

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:
    """Authentification flexible : Bearer token OU API key."""

    if api_key:
        # Authentification par 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:
        # Authentification par JWT
        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 : Versioning d'API

DEVELOPERpython
from fastapi import APIRouter

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

# V1 : format de reponse original
@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 : format enrichi avec 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"
    }

# Header de deprecation pour 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

Pour les traitements longs (ingestion de documents, batch processing).

DEVELOPERpython
from fastapi import BackgroundTasks

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

    # Demarrer le traitement en arriere-plan
    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):
    """Traitement en arriere-plan avec notification webhook."""
    try:
        # Telecharger et traiter le document
        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)
        }

    # Envoyer le webhook
    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

Traiter plusieurs requetes en une seule API call pour reduire la latence et les couts.

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):
    """Traitement batch avec concurrence controlee."""
    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 pour RAG

Notre recommandation : REST + SSE pour 90% des cas d'usage RAG. WebSocket uniquement pour le chat interactif temps reel avec typing indicators.

Specification OpenAPI type

DEVELOPERyaml
openapi: 3.1.0
info:
  title: RAG API
  version: 2.0.0
  description: API RAG avec streaming, sources et confiance

paths:
  /api/v2/query:
    post:
      summary: Query the RAG system
      requestBody:
        required: true
        content:
          application/json:
            schema:
              type: object
              required: [message]
              properties:
                message:
                  type: string
                  maxLength: 4000
                conversation_id:
                  type: string
                  format: uuid
                options:
                  type: object
                  properties:
                    stream:
                      type: boolean
                      default: true
                    max_sources:
                      type: integer
                      default: 5
                    language:
                      type: string
                      enum: [fr, en, de]
      responses:
        '200':
          description: Successful response
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/RAGResponse'
            text/event-stream:
              schema:
                type: string
        '429':
          description: Rate limit exceeded

components:
  schemas:
    RAGResponse:
      type: object
      properties:
        data:
          type: object
          properties:
            answer:
              type: string
            sources:
              type: array
              items:
                $ref: '#/components/schemas/Source'
            confidence:
              type: number
              minimum: 0
              maximum: 1
        usage:
          type: object
          properties:
            tokens_input:
              type: integer
            tokens_output:
              type: integer
            cost_usd:
              type: number

Gestion d'erreurs

Structure d'erreur standardisee

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

# Mapping des erreurs
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,  # Pas une erreur technique
        "message": "No relevant information found for your question."
    }
}

@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    request_id = request.headers.get("X-Request-ID", str(uuid4()))
    logger.error(f"[{request_id}] {exc}", exc_info=True)

    return JSONResponse(
        status_code=500,
        content={
            "error": "internal_error",
            "code": "INTERNAL_ERROR",
            "message": "An unexpected error occurred.",
            "request_id": request_id
        }
    )

Notre API chez Ailog

Chez Ailog, notre API RAG implemente les 10 patterns decrits dans ce guide. Voici ce que nos clients obtiennent :

• Streaming SSE avec sources en amont
• Multi-canal : widget, API, team chat
• Rate limiting intelligent par plan
• Webhooks pour l'ingestion de documents
• SDK Python et TypeScript pour une integration facile

Decouvrez notre guide sur le streaming RAG et le deploiement en production.

Conclusion

Les 10 design patterns decrits ici ne sont pas optionnels pour une API RAG en production. Ils constituent le socle minimum de qualite que vos utilisateurs et clients attendent :

1. Streaming SSE - UX instantanee
2. Conversation threads - Contexte maintenu
3. Sources - Confiance et transparence
4. Scores de confiance - Decisions eclairees
5. Fallbacks - Resilience
6. Rate limiting - Protection couts
7. Authentification - Securite
8. Versioning - Stabilite
9. Webhooks - Asynchrone
10. Batch - Efficacite

Commencez par les patterns 1 a 5, puis ajoutez les suivants selon vos besoins.

Vous voulez une API RAG qui implemente tous ces patterns sans effort ? Essayez l'API Ailog - tout est pret, il suffit de connecter vos documents.