Pipeline d’évaluation complète pour agents IA instrumentés

Ce guide décrit la mise en place d’un pipeline d’évaluation bout à bout pour agents IA (LLM + outils) : formalisation des KPIs métier, schéma d’événements versionné, instrumentation (tokenizer, latence en secondes), collecte et exposition OpenMetrics, évaluation offline rigoureuse, dashboards Grafana, alerting, CI/CD, sécurité et gouvernance renforcées.

Prérequis

• Agent IA instrumentable (LangChain, Semantic Kernel ou SDK interne).
• CI (GitHub/GitLab Actions) + VM ou cluster pour monitoring.
• Python 3.10+, prometheus_client, tiktoken ou équivalent, Grafana.
• Datasets versionnés (DVC ou Git-LFS) avec vérité terrain.
• Vault pour secrets, chiffrement TLS1.2+ en transit et AES-256 au repos, RBAC, audit logging.

1 – Formaliser objectifs et KPIs

Clarifiez tâches métier et seuils :

• TaskCompletionRate ≥ 0,85
• ExactMatch / F1‐score
• Latency_P95 ≤ 0,5 s
• CostPerRequest (USD)
• HallucinationRate < 1 %
• EscalationRate, CSAT

Ces seuils pilotent l’instrumentation, l’agrégation et l’architecture de monitoring.

2 – Schéma d’événements et instrumentation

Versionnez un schéma Protobuf ou Avro :

message AgentEvent {
  string event_version = 1;
  string event_id      = 2;
  google.protobuf.Timestamp timestamp = 3;
  string model_version = 4;
  string provider      = 5;
  string session_id    = 6;
  string task_type     = 7;
  int32 tokens_in      = 8;
  int32 tokens_out     = 9;
  repeated ToolInfo tools = 10;
  OutputInfo output    = 11;
  EvalInfo evaluation  = 12;
  bool pii_redacted    = 13;
}

Exemple de décorateur Python utilisant tiktoken et latence en secondes :

import time, uuid, json
import tiktoken

tokenizer = tiktoken.get_encoding("cl100k_base")

def emit(evt):
    print(json.dumps(evt, ensure_ascii=False))

def instrument(task):
    def decorator(fn):
        def wrapper(*args, **kwargs):
            evt = {
                "event_version":"1.0",
                "event_id": uuid.uuid4().hex,
                "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
                "task_type": task
            }
            t0 = time.time()
            result = fn(*args, **kwargs)
            t1 = time.time()
            prompt = kwargs.get("prompt", "")
            answer = result.get("answer", "")
            evt.update({
                "tokens_in": len(tokenizer.encode(prompt)),
                "tokens_out": len(tokenizer.encode(answer)),
                "output": {
                  "latency_s": round(t1 - t0, 3),
                  "hallucination": result.get("hallucination", False)
                },
                "evaluation": result.get("evaluation", {}),
                "pii_redacted": True
            })
            emit(evt)
            return result
        return wrapper
    return decorator

Respectez :

• Cardinalité limitée des labels Prometheus.
• Redaction PII avant journalisation.
• Chiffrement et RBAC sur logs et schéma.

3 – Évaluation offline reproductible

Méthodologie :

• Échantillonnage stratifié : catégories métiers équilibrées, seed fixe (random.seed(42) pour reproductibilité).
• 10–20 % cas annotés par ≥ 2 annotateurs (« golden judges »), calcul du κ de Cohen pour valider l’accord inter-annotateur.
• Règles de labelling de l’hallucination : divergence factuelle ou absence de source fiable.
• Recalibrage périodique pour dérive.

# eval/run_eval.py
import json, random, statistics, glob
from sklearn.metrics import cohen_kappa_score

random.seed(42)

def load_events(path):
    events=[]
    for f in glob.glob(f"{path}/*.json"):
        with open(f, encoding="utf-8") as fp:
            events.append(json.load(fp))
    return events

def aggregate(events):
    lat = [e["output"]["latency_s"] for e in events]
    p95 = statistics.quantiles(lat, n=100)[94]
    tcr = sum(1 for e in events if e["evaluation"].get("completed"))/len(events)
    return {"task_completion_rate":tcr, "latency_p95_s":p95}

if __name__=="__main__":
    evts = load_events("events")
    agg = aggregate(evts)
    print(json.dumps(agg, indent=2))

4 – Exposition des métriques Prometheus/OpenMetrics

from prometheus_client import Counter, Histogram, Gauge

TASK_TOTAL = Counter("agent_tasks_total", "Total tâches")
TASK_OK    = Counter("agent_tasks_completed_total", "Tâches complétées")
LATENCY    = Histogram("agent_latency_s", "Latence agent en secondes",
                       buckets=[0.05, 0.1, 0.3, 0.5, 1.0, 2.0])
COST       = Histogram("agent_cost_usd", "Coût par requête", buckets=[0.001,0.01,0.1,1.0])
KPI_GAUGE  = Gauge("agent_kpi_compliance", "Conformité KPI", ["kpi"])

Scrape config Prometheus :

scrape_configs:
  - job_name: "agent"
    static_configs:
      - targets: ["agent-service:8000"]

5 – CI/CD et contrôle de régression

Exemple de check_thresholds.py :

# scripts/check_thresholds.py
import json, sys

THRESHOLDS = {
  "task_completion_rate": 0.85,
  "latency_p95_s": 0.5
}

def main():
    data = json.load(open("summary.json"))
    failed = []
    for k, thresh in THRESHOLDS.items():
        if data.get(k, 0) < thresh:
            failed.append(f"{k} below {thresh}: {data.get(k)}")
    if failed:
        print("❌ Regression détectée:")
        print("\n".join(failed))
        sys.exit(1)
    print("✅ Tous les KPI sont conformes.")
    sys.exit(0)

if __name__=="__main__":
    main()

Workflow GitHub Actions (PR bloquée si échec) :

on: [pull_request]
jobs:
  eval:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: python-version: "3.11"
      - run: pip install -r requirements.txt
      - run: python eval/run_eval.py > summary.json
      - run: python scripts/check_thresholds.py

6 – Dashboards et alerting Grafana

• Taux de complétion, histogram_quantile(0.95, agent_latency_s_bucket), coûts P50/P95.
• HallucinationRate, EscalationRate, conformité KPI.

Exemple de règle d’alerte :

groups:
- name: agent_alerts
  rules:
  - alert: TaskCompletionDrop
    expr: rate(agent_tasks_completed_total[5m]) / rate(agent_tasks_total[5m]) < 0.85
    for: 10m
    labels: {severity:"critical"}
    annotations:
      summary: "Taux de complétion < 85 % sur 10m"

7 – Sécurité et gouvernance

• Redaction PII selon patterns regex et anonymisation k-anonymity.
• Chiffrement TLS1.2+ en transit et AES-256 au repos.
• Stockage des secrets dans HashiCorp Vault, rotation automatique.
• RBAC granulaire, audit logging des accès et mutations.
• Conservation des logs signés et intégrité via checksums.