RAG-Prod 01 Challenge: a leaderboard that scores your model the way a company would

┌──────────────────────┐   POST /v1/infer    ┌─────────────────────────┐
│ Evaluator            │ ──────────────────▶ │ FastAPI                 │
│ 8 concurrent requests│                     │ Pydantic validation     │
│                      │ ◀────────────────── │   │                     │
│                      │ answer + sources    │   ▼                     │
│                      │ + usage (tokens)    │ TF-IDF retrieval        │
│                      │                     │ top-k documents         │
│                      │   GET /metrics      │   │                     │
│                      │ ──────────────────▶ │   ▼                     │
└──────────┬───────────┘ peak RAM            │ Answer sentence extract.│
           ▼                                 └─────────────────────────┘
  Composite score: F1 · P99 · memory · $

rag-prod-01/
├── Dockerfile
├── docker-compose.yml
├── requirements.txt
├── app/main.py          # inference service (baseline)
├── eval/evaluate.py     # multi-dimensional evaluator
└── data/
    ├── corpus.jsonl     # {"id": "doc-001", "text": "..."}
    └── questions.jsonl  # {"id": "q-001", "question": "...", "answer": "..."}

docker compose up --build --abort-on-container-exit --exit-code-from eval

fastapi==0.115.0
uvicorn==0.30.6
pydantic==2.9.2
scikit-learn==1.5.2
httpx==0.27.2

FROM python:3.11-slim

ENV PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1
WORKDIR /srv

# Dépendances d'abord : couche mise en cache tant que requirements.txt ne change pas
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY app/ app/
COPY eval/ eval/

# Jamais root en production
RUN useradd --uid 10001 --no-create-home runner
USER runner

EXPOSE 8000
# /health ne répond 200 qu'une fois l'index chargé
HEALTHCHECK --interval=10s --timeout=3s --retries=5 \
  CMD python -c "import urllib.request,sys; sys.exit(0 if urllib.request.urlopen('http://127.0.0.1:8000/health', timeout=2).status == 200 else 1)"

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "1"]

services:
  rag:
    build: .
    expose: ["8000"]     # réseau interne Compose : aucun port publié sur l hôte
    environment:
      CORPUS_PATH: /data/corpus.jsonl
      TOP_K: "3"
    volumes: ["./data:/data:ro"]
    mem_limit: 4g        # plafond officiel du challenge
    cpus: 2

  eval:
    build: .
    command: ["python", "eval/evaluate.py"]
    environment:
      TARGET_URL: http://rag:8000
      QUESTIONS_PATH: /data/questions.jsonl
      CONCURRENCY: "8"
      PRICE_IN_PER_MTOK: "0"      # renseigner si votre service appelle un LLM payant
      PRICE_OUT_PER_MTOK: "0"
      INFRA_USD_PER_HOUR: "0.10"
    volumes: ["./data:/data:ro"]
    depends_on:
      rag:
        condition: service_healthy
    healthcheck:
      disable: true

"""Service d'inférence de référence — challenge RAG-Prod 01.

Baseline volontairement simple (TF-IDF + extraction de phrase), à battre.
Endpoints : GET /health, GET /metrics, POST /v1/infer.
"""
from __future__ import annotations

import json
import logging
import os
import re
import resource
import time
import uuid
from contextlib import asynccontextmanager
from pathlib import Path
from typing import Any

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field, field_validator
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import linear_kernel

CORPUS_PATH = Path(os.getenv("CORPUS_PATH", "/data/corpus.jsonl"))
TOP_K = int(os.getenv("TOP_K", "3"))
MAX_QUERY_CHARS = 2000
SENTENCE_SPLIT = re.compile(r"(?<=[.!?])\s+")

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("rag-prod-01")
state: dict[str, Any] = {}


def load_corpus(path: Path) -> list[dict[str, str]]:
    """Charge et valide le corpus JSONL ; échoue tôt (fail-fast) si invalide."""
    if not path.is_file():
        raise RuntimeError(f"Corpus introuvable : {path}")
    docs: list[dict[str, str]] = []
    with path.open(encoding="utf-8") as fh:
        for line_no, line in enumerate(fh, start=1):
            if not line.strip():
                continue
            try:
                rec = json.loads(line)
            except json.JSONDecodeError as exc:
                raise RuntimeError(f"JSON invalide ligne {line_no}") from exc
            if not isinstance(rec.get("id"), str) or not isinstance(rec.get("text"), str) \
                    or not rec["text"].strip():
                raise RuntimeError(f"Champs 'id'/'text' manquants ligne {line_no}")
            docs.append({"id": rec["id"], "text": rec["text"]})
    if not docs:
        raise RuntimeError("Corpus vide")
    return docs


@asynccontextmanager
async def lifespan(_: FastAPI):
    docs = load_corpus(CORPUS_PATH)
    vectorizer = TfidfVectorizer(ngram_range=(1, 2), sublinear_tf=True)
    matrix = vectorizer.fit_transform([d["text"] for d in docs])
    state.update(docs=docs, vectorizer=vectorizer, matrix=matrix)
    logger.info("Index prêt : %d documents", len(docs))
    yield
    state.clear()


app = FastAPI(title="RAG-Prod 01 baseline", lifespan=lifespan)


class InferRequest(BaseModel):
    question: str = Field(min_length=3, max_length=MAX_QUERY_CHARS)

    @field_validator("question")
    @classmethod
    def not_blank(cls, value: str) -> str:
        value = value.strip()
        if len(value) < 3:
            raise ValueError("question vide ou trop courte")
        return value


class Usage(BaseModel):
    input_tokens: int = Field(ge=0)
    output_tokens: int = Field(ge=0)


class InferResponse(BaseModel):
    answer: str | None
    sources: list[str]
    latency_ms: float
    usage: Usage


@app.exception_handler(Exception)
async def unhandled(_: Request, exc: Exception) -> JSONResponse:
    """Aucune trace interne n'est renvoyée au client : seulement un identifiant d'erreur."""
    error_id = uuid.uuid4().hex[:12]
    logger.exception("Erreur %s : %s", error_id, type(exc).__name__)
    return JSONResponse(status_code=500, content={"error": "internal", "error_id": error_id})


@app.get("/health")
def health() -> dict[str, str]:
    if "matrix" not in state:
        raise HTTPException(status_code=503, detail="index non chargé")
    return {"status": "ok"}


@app.get("/metrics")
def metrics() -> dict[str, float]:
    # ru_maxrss est en Ko sous Linux : mémoire résidente de POINTE du processus
    return {"peak_rss_mb": resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024}


@app.post("/v1/infer", response_model=InferResponse)
def infer(req: InferRequest) -> InferResponse:
    # Fonction synchrone : FastAPI l'exécute dans un pool de threads, la boucle reste libre
    t0 = time.perf_counter()
    if "matrix" not in state:
        raise HTTPException(status_code=503, detail="index non chargé")
    docs, vec = state["docs"], state["vectorizer"]

    q = vec.transform([req.question])
    scores = linear_kernel(q, state["matrix"]).ravel()
    top = [int(i) for i in scores.argsort()[::-1][:TOP_K] if scores[i] > 0]

    answer: str | None = None
    if top:
        sentences = [s for i in top for s in SENTENCE_SPLIT.split(docs[i]["text"]) if s.strip()]
        s_scores = linear_kernel(q, vec.transform(sentences)).ravel()
        answer = sentences[int(s_scores.argmax())]

    return InferResponse(
        answer=answer,
        sources=[docs[i]["id"] for i in top],
        latency_ms=round((time.perf_counter() - t0) * 1000, 2),
        # Baseline sans LLM : estimation par mots. Une soumission LLM DOIT
        # renvoyer les compteurs exacts fournis par son fournisseur.
        usage=Usage(input_tokens=len(req.question.split()),
                    output_tokens=len(answer.split()) if answer else 0),
    )

"""Évaluateur multidimensionnel — RAG-Prod 01. Affiche un rapport JSON."""
from __future__ import annotations

import json
import math
import os
import string
import sys
import time
import unicodedata
from collections import Counter
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path

import httpx

BASE_URL = os.getenv("TARGET_URL", "http://rag:8000")
QUESTIONS = Path(os.getenv("QUESTIONS_PATH", "/data/questions.jsonl"))
CONCURRENCY = int(os.getenv("CONCURRENCY", "8"))
PRICE_IN = float(os.getenv("PRICE_IN_PER_MTOK", "0"))
PRICE_OUT = float(os.getenv("PRICE_OUT_PER_MTOK", "0"))
INFRA_PER_HOUR = float(os.getenv("INFRA_USD_PER_HOUR", "0.10"))
L_MAX_MS, M_MAX_MB, C_MAX_USD, MAX_ERROR_RATE = 1500.0, 4096.0, 2.0, 0.01
W_F1, W_LAT, W_MEM, W_COST = 0.4, 0.2, 0.2, 0.2


def normalize(text: str) -> list[str]:
    text = unicodedata.normalize("NFKD", text.lower())
    text = "".join(c for c in text if not unicodedata.combining(c))
    return text.translate(str.maketrans("", "", string.punctuation)).split()


def token_f1(pred: str, gold: str) -> float:
    p, g = normalize(pred), normalize(gold)
    if not p or not g:
        return float(p == g)
    common = sum((Counter(p) & Counter(g)).values())
    if common == 0:
        return 0.0
    precision, recall = common / len(p), common / len(g)
    return 2 * precision * recall / (precision + recall)


def p99(values: list[float]) -> float:
    ordered = sorted(values)
    return ordered[math.ceil(0.99 * len(ordered)) - 1]


def load_questions(path: Path) -> list[dict[str, str]]:
    items = []
    for n, line in enumerate(path.read_text(encoding="utf-8").splitlines(), start=1):
        if not line.strip():
            continue
        rec = json.loads(line)
        if not all(isinstance(rec.get(k), str) for k in ("id", "question", "answer")):
            raise ValueError(f"questions.jsonl ligne {n} : champs id/question/answer requis")
        items.append(rec)
    if not items:
        raise ValueError("Aucune question à évaluer")
    return items


def ask(client: httpx.Client, item: dict[str, str]) -> tuple[bool, float, dict]:
    t0 = time.perf_counter()
    try:
        resp = client.post(f"{BASE_URL}/v1/infer", json={"question": item["question"]}, timeout=10)
        resp.raise_for_status()
        body, ok = resp.json(), True
    except (httpx.HTTPError, ValueError):
        body, ok = {}, False
    return ok, (time.perf_counter() - t0) * 1000, body


def main() -> int:
    items = load_questions(QUESTIONS)
    n = len(items)
    with httpx.Client() as client:
        start = time.perf_counter()
        with ThreadPoolExecutor(max_workers=CONCURRENCY) as pool:
            results = list(pool.map(lambda it: ask(client, it), items))
        duration_s = time.perf_counter() - start
        memory_mb = float(client.get(f"{BASE_URL}/metrics", timeout=5).json()["peak_rss_mb"])

    errors = sum(1 for ok, _, _ in results if not ok)
    f1 = sum(token_f1((b.get("answer") or ""), it["answer"])
             for (_, _, b), it in zip(results, items)) / n
    lat_p99 = p99([ms for _, ms, _ in results])
    t_in = sum(b.get("usage", {}).get("input_tokens", 0) for _, _, b in results)
    t_out = sum(b.get("usage", {}).get("output_tokens", 0) for _, _, b in results)
    cost_1k = (1000 / n) * ((t_in * PRICE_IN + t_out * PRICE_OUT) / 1e6
                            + INFRA_PER_HOUR * duration_s / 3600)

    error_rate = errors / n
    disqualified = lat_p99 > L_MAX_MS or memory_mb > M_MAX_MB or error_rate > MAX_ERROR_RATE
    score = 0.0 if disqualified else 100 * (
        W_F1 * f1
        + W_LAT * (1 - lat_p99 / L_MAX_MS)
        + W_MEM * (1 - memory_mb / M_MAX_MB)
        + W_COST * (1 - min(cost_1k / C_MAX_USD, 1.0))
    )
    print(json.dumps({
        "questions": n, "error_rate": round(error_rate, 4),
        "f1": round(f1, 4), "p99_ms": round(lat_p99, 1),
        "throughput_rps": round(n / duration_s, 2), "peak_memory_mb": round(memory_mb, 1),
        "cost_per_1k_usd": round(cost_1k, 4),
        "disqualified": disqualified, "score": round(score, 2),
    }, indent=2))
    return 0


if __name__ == "__main__":
    sys.exit(main())