Construire un RAG d’entreprise fiable sur Kubernetes : architecture, sécurité et MLOps

Les systèmes RAG (Retrieval-Augmented Generation) sont devenus la voie la plus pragmatique pour exploiter les LLM sur des données privées, réduire les hallucinations et accélérer la recherche d’information. Pourtant, beaucoup de POC échouent à passer en production faute d’architecture, de sécurité et d’observabilité adaptées. Cet article propose une architecture de référence et des exemples concrets pour déployer un RAG d’entreprise fiable sur Kubernetes.

Pourquoi le RAG en entreprise

Cas d’usage courants :

• Assistance support/ITSM sur base de KB internes
• Recherche documentaire et compliance (policies, contrats)
• Copilotes métiers (achats, finance, opérations)
• Aide au développement (référentiels internes, patterns)

Enjeux clés à traiter dès la conception :

• Sécurité et gouvernance d’accès aux documents
• Évaluation continue de la qualité des réponses
• Scalabilité et contrôle des coûts
• Observabilité bout en bout (ingestion → recherche → génération)

Architecture de référence

Composants recommandés sur Kubernetes :

1. Ingestion de documents

   • Connecteurs (SharePoint, Confluence, S3/Blob, Git, bases SQL)
   • Normalisation (PDF/Office/HTML), extraction de texte, anonymisation optionnelle
   • Découpage (chunking) + embeddings
   • Vector store (pgvector, OpenSearch vector, Pinecone)
   • Stockage objets pour les sources (S3/Blob/GCS)

2. Service de requêtes

   • API Gateway (ingress + auth)
   • Orchestrateur de prompts/flux (graph d’outils, réécriture de requêtes)
   • Recherche hybride (BM25 + vector) + reranking
   • Appel LLM (OpenAI/Azure OpenAI/Bedrock ou modèle self-hosted)
   • Caches (semantic + réponse)

3. Sécurité & Observabilité

   • IAM + filtrage d’accès par document (ABAC/RBAC)
   • Secrets manager (Vault/Key Vault/Secrets Manager)
   • Traces/metrics/logs via OpenTelemetry + backend (Grafana/Prometheus/Tempo/ELK)
   • Guardrails (anti-prompt injection, PII masking, policy de sortie)

Pipeline d’ingestion: exemples concrets

Schéma de table Postgres avec pgvector (exemple minimal) :

CREATE EXTENSION IF NOT EXISTS vector;

CREATE TABLE IF NOT EXISTS documents (
  id BIGSERIAL PRIMARY KEY,
  doc_id TEXT,
  chunk_id INT,
  content TEXT,
  metadata JSONB,
  embedding VECTOR(1536)
);

-- Index vectoriel IVFFlat avec distance cosine
CREATE INDEX ON documents USING ivfflat (embedding vector_cosine_ops) WITH (lists = 100);
CREATE INDEX ON documents USING GIN (to_tsvector('simple', content));
CREATE INDEX ON documents ((metadata->>'visibility'));

Chunking et insertion en Python (simplifié) :

import os, psycopg
from typing import List, Dict

# Exemple: utilisez un service d’embedding (API fournisseur ou modèle local)
def embed(texts: List[str]) -> List[List[float]]:
    # Placeholder: appelez ici votre provider embeddings
    # return client.embed(texts, model="text-embedding")
    raise NotImplementedError

def semantic_chunks(text: str, max_tokens: int = 400, overlap: int = 60) -> List[str]:
    # Simplifié: segmente par phrases puis recompose en fenêtres glissantes
    sentences = [s.strip() for s in text.split(".") if s.strip()]
    chunks, buf = [], []
    token_est = lambda t: max(1, len(t.split()) // 0.75)  # approx.
    for s in sentences:
        if sum(token_est(x) for x in buf + [s]) > max_tokens and buf:
            chunks.append(". ".join(buf))
            buf = buf[-overlap:]
        buf.append(s)
    if buf: chunks.append(". ".join(buf))
    return chunks

def insert_chunks(conn, doc_id: str, chunks: List[str], meta: Dict):
    vectors = embed(chunks)
    with conn.cursor() as cur:
        for i, (c, v) in enumerate(zip(chunks, vectors)):
            cur.execute(
                "INSERT INTO documents (doc_id, chunk_id, content, metadata, embedding) VALUES (%s,%s,%s,%s,%s)",
                (doc_id, i, c, psycopg.types.json.Json(meta), v)
            )

with psycopg.connect(os.environ["PG_DSN"]) as conn:
    raw_text = open("policy.pdf.txt").read()  # texte déjà extrait
    chunks = semantic_chunks(raw_text)
    meta = {"visibility": "finance-only", "source": "SharePoint", "lang": "fr"}
    insert_chunks(conn, "policy-2026-04", chunks, meta)

Bonnes pratiques d’ingestion :

• Nettoyez et normalisez tôt (suppression boilerplate, tableaux, headers/footers)
• Stockez les métadonnées d’accès (entité, équipe, classification) dès l’ingestion
• Dédupliquez par hash de contenu
• Testez différentes tailles de chunk et overlaps selon vos documents

Recherche et génération: hybrid, filtres d’accès et streaming

Exemple de requête hybride Postgres (BM25 + vector) avec filtrage d’accès:

WITH q AS (
  SELECT to_tsvector('simple', $1) AS qtsv, $2::vector AS qvec
),
bm25 AS (
  SELECT id, 1.0 * ts_rank_cd(to_tsvector('simple', content), (SELECT qtsv FROM q)) AS score
  FROM documents
  WHERE metadata->>'visibility' = $3
  ORDER BY score DESC
  LIMIT 100
),
vec AS (
  SELECT id, 1.0 - (embedding <=> (SELECT qvec FROM q)) AS score
  FROM documents
  WHERE metadata->>'visibility' = $3
  ORDER BY embedding <-> (SELECT qvec FROM q)
  LIMIT 100
),
unioned AS (
  SELECT id, MAX(score) AS score FROM (
    SELECT * FROM bm25 UNION ALL SELECT * FROM vec
  ) s GROUP BY id
)
SELECT d.id, d.content, d.metadata
FROM unioned u
JOIN documents d ON d.id = u.id
ORDER BY u.score DESC
LIMIT 8;

API FastAPI en lecture, avec streaming de la réponse LLM (simplifié) :

from fastapi import FastAPI, Header
from fastapi.responses import StreamingResponse

app = FastAPI()

def llm_stream(prompt: str):
    # Itérez sur les tokens du fournisseur LLM choisi
    yield "Réponse partielle..."
    yield " suite..."
    yield " fin."

@app.post("/chat")
def chat(q: str, x_user: str = Header(...)):
    # 1) Embedding de q, 2) requête hybride filtrée par visibilité de x_user
    # 3) Construction du prompt final (citations + sources), 4) stream
    return StreamingResponse(llm_stream(q), media_type="text/plain")

Bonnes pratiques de génération :

• Citez systématiquement les sources (titres + URL + horodatage)
• Injectez un système prompt robuste (règles de non-divulgation, ton, format)
• Implémentez un reranker (cross-encoder) pour améliorer le top-k avant LLM

Sécurité et gouvernance

• Contrôles d’accès
  • À l’ingestion: étiquetez chaque chunk (ABAC). Exemple: visibility=finance-only
  • À la requête: appliquez des filtres d’accès en base (RLS Postgres, index par labels)
• Secrets et clés
  • Stockez les clés LLM et DSN dans Secrets Manager/Key Vault/Vault
• Cloisonnement réseau Kubernetes (exemple NetworkPolicy) :

apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: rag-deny-all-except-api-db
spec:
  podSelector: {}
  policyTypes: ["Ingress","Egress"]
  ingress:
  - from:
    - podSelector:
        matchLabels:
          app: api
  egress:
  - to:
    - podSelector:
        matchLabels:
          app: postgres
  - to:
    - namespaceSelector:
        matchLabels:
          kubernetes.io/metadata.name: "observability"

• Guardrails IA
  • Détection simple de prompt injection (mots-clés, heuristiques) + journalisation
  • Masquage PII en sortie (regex, entités nommées) selon politiques internes
  • Limites de taux par utilisateur et par espace métier

Observabilité et évaluation

• Traces distribuées OpenTelemetry (ingestion, retrieval, génération)
• Métriques clés: latence p95, taux de vide (no-context), ratio citations/longueur, coût par requête
• Logs d’audit: requête, documents consultés (hash/ID), décision de filtrage
• Évaluation continue
  • Jeux de questions-réponses internes, exact match + jugement LLM contrôlé
  • Alertes en cas de dérive (performance, latence, coût)
  • Boucle de feedback utilisateurs intégrée au front (thumbs up/down + commentaire)

Scalabilité et maîtrise des coûts

• Autoscaling horizontal

apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: rag-api-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: rag-api
  minReplicas: 2
  maxReplicas: 20
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 60

• Optimisations
  • Cache sémantique (Redis) pour réponses répétitives
  • Choix de modèles économiques selon le cas (petits modèles pour reformulation, grands modèles pour génération finale)
  • Batching des embeddings et ingestion asynchrone
  • Indexation incrémentale et compaction régulière

Intégration SI et APIs

• Exposez une API unique pour Chat, Search, Answer + citations
• Intégration avec bus/API management (MuleSoft, WSO2) ou GraphQL pour agrégation
• Webhooks/Events pour rafraîchissement d’index après mise à jour documentaire

Extrait OpenAPI minimal:

openapi: 3.0.3
info: { title: RAG API, version: "1.0.0" }
paths:
  /answer:
    post:
      requestBody:
        required: true
        content:
          application/json:
            schema:
              properties:
                query: { type: string }
      responses:
        "200":
          description: OK

Plan de mise en production recommandé

1. Cadrage sécurité & données (sources, labels d’accès, PII)
2. POC ciblé sur un corpus prioritaire + protocole d’évaluation
3. Industrialisation ingestion + index hybride + reranker
4. Observabilité et guardrails
5. API/Front + intégration IAM/SSO + cache
6. Tests charge, coûts et Go-Live progressif par périmètre

Conclusion

Passer d’un POC RAG à une plateforme d’entreprise exige une approche d’ingénierie complète: architecture cloud-native, sécurité stricte, observabilité et optimisation continue. Kairox accompagne ses clients sur l’ensemble du cycle — de l’idéation à l’exploitation — en combinant expertise IA, intégration SI et pratiques DevSecOps sur Kubernetes. Vous souhaitez évaluer ou industrialiser votre RAG ? Contactez nos équipes pour un atelier d’architecture et un plan de mise en production adapté à votre contexte.