0Pricing
AI Engineering Academy · Lezione

Creare un sistema automatizzato di valutazione

Creerà una pipeline di valutazione ripetibile che esegua l'intero sistema RAG su un test set, calcoli tutte le metriche e generi un report per monitorare i miglioramenti nel tempo.

Creare un sistema automatizzato di valutazione è una lezione AI Engineering Academy gratuita su CoddyKit. Questa è la lezione 4 di 4. Puoi leggere la lezione completa qui gratuitamente — poi esercitati direttamente nel browser con un editor di codice integrato e un tutor IA disponibile 24/7. Fa parte del percorso di apprendimento AI Engineering Academy, e i tuoi progressi si sincronizzano tra il web e l'app CoddyKit. Il corso AI Engineering Academy include 4 lezioni in totale.

Che cos'è un harness di valutazione?

Un harness di valutazione è una pipeline automatizzata e ripetibile che esegue l'intero sistema RAG su un set di test standardizzato, calcola tutte le metriche e produce un report. La parola chiave è ripetibile: ogni volta che modifica la strategia di chunking, il modello di embedding, il prompt o l'LLM, esegue lo stesso harness e confronta i risultati con una baseline. Questo trasforma lo sviluppo RAG da una sperimentazione soggettiva in un'attività di ingegneria basata sui dati.

Architettura dell'harness

Un harness di valutazione ben progettato comprende quattro livelli: gestione dei dati di test (caricare e versionare il dataset di riferimento), esecuzione della pipeline (eseguire ogni domanda di test attraverso l'intera pipeline RAG), calcolo delle metriche (calcolare tutte le metriche di recupero e generazione) e generazione dei report (salvare i risultati con le informazioni sulla versione e produrre un confronto con la baseline precedente). Ogni livello deve poter essere testato e configurato in modo indipendente.

class RAGEvaluationHarness:
    def __init__(self, retriever, llm_client, config):
        self.retriever = retriever
        self.llm_client = llm_client
        self.config = config  # chunk_size, top_k, model, threshold, etc.
        self.results = []

    def run(self, golden_dataset):
        for item in golden_dataset:
            result = self._evaluate_single(item)
            self.results.append(result)
        metrics = self._compute_metrics()
        self._save_report(metrics)
        return metrics

Eseguire ogni caso di test

Per ogni domanda del dataset di riferimento, esegua l'intera pipeline RAG e acquisisca tutti gli output intermedi: gli ID e i punteggi dei segmenti recuperati, il contesto formattato, la risposta generata e il conteggio dei token. Salvare questi valori intermedi è essenziale per il debug degli errori: quando una domanda ottiene un punteggio basso, può esaminare esattamente quali segmenti sono stati recuperati e perché la risposta era errata, senza rieseguire la costosa pipeline.

import time

def _evaluate_single(self, item):
    start = time.perf_counter()
    query_vector = embed_query(item['question'])
    chunks = self.retriever.retrieve(query_vector, top_k=self.config['top_k'])
    filtered_chunks = filter_by_score(chunks, self.config['threshold'])
    context = format_context(filtered_chunks)
    answer_result = generate_answer(item['question'], context, self.llm_client)
    latency_ms = (time.perf_counter() - start) * 1000

    return {
        'question': item['question'],
        'expected_answer': item['answer'],
        'generated_answer': answer_result['answer'],
        'retrieved_chunk_ids': [c['id'] for c in filtered_chunks],
        'retrieved_scores': [c['score'] for c in filtered_chunks],
        'relevant_chunk_ids': item['relevant_chunk_ids'],
        'context_texts': [c['text'] for c in filtered_chunks],
        'tokens_used': answer_result['tokens_used'],
        'latency_ms': round(latency_ms)
    }

Calcolare tutte le metriche in un unico passaggio

Dopo aver raccolto gli output di tutti i casi di test, calcoli l'intera serie di metriche in un unico passaggio sui risultati. Separi le metriche di recupero (calcolate dagli ID dei segmenti) dalle metriche di generazione (calcolate chiamando l'LLM giudice). Raggruppi le chiamate all'LLM giudice per massimizzare l'efficienza: raggruppi le valutazioni della fedeltà e le invii in parallelo con asyncio anziché in sequenza. Registri l'avanzamento, poiché le metriche di generazione possono richiedere diversi minuti per oltre 100 casi di test.

def _compute_metrics(self):
    # Retrieval metrics (no LLM calls needed)
    hit_rates = []
    mrr_scores = []
    for r in self.results:
        retrieved = r['retrieved_chunk_ids']
        relevant = set(r['relevant_chunk_ids'])
        hit = any(rid in relevant for rid in retrieved)
        hit_rates.append(1.0 if hit else 0.0)
        for rank, rid in enumerate(retrieved, 1):
            if rid in relevant:
                mrr_scores.append(1.0 / rank)
                break
        else:
            mrr_scores.append(0.0)

    metrics = {
        'hit_rate_at_5': sum(hit_rates) / len(hit_rates),
        'mrr': sum(mrr_scores) / len(mrr_scores),
        'mean_latency_ms': sum(r['latency_ms'] for r in self.results) / len(self.results),
        'mean_tokens': sum(r['tokens_used'] for r in self.results) / len(self.results)
    }
    return metrics

Salvare i risultati con le informazioni sulla versione

Ogni esecuzione della valutazione deve essere salvata con metadati della versione, in modo da poter confrontare i risultati tra diverse configurazioni. Includa l'hash del commit Git del codice, i parametri di configurazione (modello di embedding, dimensione dei segmenti, K, soglia, modello LLM), la marca temporale e una descrizione dell'esecuzione comprensibile alle persone. Salvi i risultati in un file JSON Lines o in una tabella di database. In questo modo crea una cronologia permanente dell'evoluzione del sistema.

import json
import subprocess
from datetime import datetime

def _save_report(self, metrics):
    git_hash = subprocess.check_output(
        ['git', 'rev-parse', '--short', 'HEAD']
    ).decode().strip()

    report = {
        'run_id': datetime.utcnow().strftime('%Y%m%d_%H%M%S'),
        'git_commit': git_hash,
        'config': self.config,
        'metrics': metrics,
        'n_test_cases': len(self.results),
        'timestamp': datetime.utcnow().isoformat()
    }

    with open('eval_history.jsonl', 'a') as f:
        f.write(json.dumps(report) + '\n')
    print(f'Saved evaluation run: {report["run_id"]}')
    print(json.dumps(metrics, indent=2))

Confrontare con la baseline

Dopo ogni esecuzione, confronti automaticamente i risultati con la baseline precedente e segnali le regressioni. Una regressione è qualsiasi metrica che diminuisca oltre una determinata soglia (ad esempio, di 2 punti percentuali). Stampi una tabella comparativa che mostri le variazioni delle metriche. Se una metrica regredisce in modo significativo, l'esecuzione della valutazione deve fallire con un codice di uscita diverso da zero; ciò farà sì che una pipeline CI/CD blocchi la distribuzione della modifica.

def compare_to_baseline(current_metrics, baseline_file='best_eval.json'):
    import json
    from pathlib import Path
    if not Path(baseline_file).exists():
        print('No baseline yet. Saving current as baseline.')
        Path(baseline_file).write_text(json.dumps(current_metrics, indent=2))
        return True

    baseline = json.loads(Path(baseline_file).read_text())
    regressions = []
    print('\nMetric comparison (current vs baseline):')
    for metric, current_val in current_metrics.items():
        baseline_val = baseline.get(metric, 0)
        delta = current_val - baseline_val
        status = 'OK' if delta >= -0.02 else 'REGRESSION'
        print(f'  {metric}: {current_val:.3f} vs {baseline_val:.3f} ({delta:+.3f}) {status}')
        if status == 'REGRESSION':
            regressions.append(metric)
    return len(regressions) == 0

Integrare nella CI/CD

L'harness di valutazione è più efficace quando viene integrato nella Sua pipeline CI/CD. Lo configuri affinché venga eseguito automaticamente a ogni pull request che modifica la logica di chunking, la configurazione del modello di embedding, i template dei prompt o i parametri di recupero. La pipeline passa solo se tutte le metriche rispettano le soglie minime e nessuna metrica regredisce rispetto alla baseline del branch principale. In questo modo si impedisce che regressioni accidentali della qualità arrivino in produzione.

# GitHub Actions workflow (eval.yml)
# on:
#   pull_request:
#     paths:
#       - 'rag/**'
#       - 'prompts/**'
#       - 'config/**'
# jobs:
#   evaluate:
#     runs-on: ubuntu-latest
#     steps:
#       - uses: actions/checkout@v3
#       - name: Install dependencies
#         run: pip install -r requirements.txt
#       - name: Run evaluation harness
#         run: |
#           python eval/run_harness.py \
#             --test-set eval/golden_dataset.json \
#             --config config/rag_config.yaml \
#             --fail-on-regression
#         env:
#           OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}

Generare report leggibili

Oltre ai file contenenti le metriche grezze, generi un report HTML o Markdown leggibile che il team possa esaminare nei commenti delle pull request. Includa una tabella riepilogativa di tutte le metriche, un elenco dei casi di test non riusciti con la domanda, la risposta prevista, la risposta generata e i segmenti recuperati, nonché un grafico dell'andamento delle metriche nelle ultime 10 esecuzioni. I report visivi aiutano i responsabili non tecnici a capire se il sistema sta migliorando.

def generate_markdown_report(metrics, failed_cases, run_id):
    lines = [
        f'# RAG Evaluation Report — {run_id}\n',
        '## Summary Metrics',
        '| Metric | Score | Target |',
        '|--------|-------|--------|',
        f'| Hit Rate@5 | {metrics["hit_rate_at_5"]:.1%} | > 80% |',
        f'| MRR | {metrics["mrr"]:.3f} | > 0.70 |',
        f'| Mean Latency | {metrics["mean_latency_ms"]:.0f}ms | < 500ms |',
        '',
        f'## Failed Cases ({len(failed_cases)} failures)'
    ]
    for case in failed_cases[:10]:  # show first 10
        lines += [
            f'**Q:** {case["question"]}',
            f'**Expected:** {case["expected_answer"]}',
            f'**Generated:** {case["generated_answer"]}\n'
        ]
    return '\n'.join(lines)

Monitorare il costo di ogni esecuzione della valutazione

Le esecuzioni delle valutazioni hanno un costo: chiamano l'API di embedding, l'API LLM e l'LLM giudice. Monitori il costo di ogni esecuzione della valutazione insieme alle metriche di qualità. Una valutazione completa di 100 casi di test costa in genere da 0,50 a 2,00 $ a seconda dei modelli utilizzati. Utilizzi modelli più economici per le chiamate al giudice (GPT-4o-mini per la valutazione della fedeltà) e riservi i modelli costosi alla generazione. Includa il costo stimato dell'esecuzione nel report salvato, così da poter inserire il budget delle valutazioni nel ciclo di sviluppo.

def estimate_run_cost(results, config):
    # Embedding cost
    embed_tokens = sum(len(r['question'].split()) * 1.3 for r in results)
    embed_cost = (embed_tokens / 1_000_000) * 0.02  # $0.02/1M tokens

    # Generation cost
    total_gen_tokens = sum(r['tokens_used'] for r in results)
    gen_cost = (total_gen_tokens / 1_000_000) * 5.0  # gpt-4o approx

    # Judge cost (faithfulness evals)
    judge_cost = len(results) * 0.001  # ~$0.001 per eval with gpt-4o-mini

    total = embed_cost + gen_cost + judge_cost
    print(f'Evaluation cost estimate: ${total:.2f}')
    print(f'  Embedding: ${embed_cost:.3f}')
    print(f'  Generation: ${gen_cost:.3f}')
    print(f'  Judgment: ${judge_cost:.3f}')
    return total

Valutazione pianificata per il monitoraggio in produzione

Oltre alla valutazione CI/CD in seguito alle modifiche al codice, esegua l'harness secondo una pianificazione in produzione — ogni giorno o ogni settimana — testandolo su query reali degli utenti campionate dai log. Questo rileva il data drift: con l'evoluzione del corpus di documenti e il cambiamento dei modelli delle query degli utenti, la qualità del sistema può diminuire anche senza modifiche al codice. Pianifichi esecuzioni settimanali della valutazione che campionino 50 query recenti degli utenti, le valutino e inviino automaticamente un riepilogo della qualità al canale Slack del team.

# Example scheduled evaluation (cron job or scheduled cloud function)
import random

def sample_production_queries(query_log_file, n=50):
    with open(query_log_file) as f:
        all_queries = [json.loads(line) for line in f]
    sample = random.sample(all_queries, min(n, len(all_queries)))
    # Convert to golden dataset format (without expected answers — use LLM judge)
    return [
        {'question': q['user_question'], 'relevant_chunk_ids': []}
        for q in sample
    ]

# Run weekly evaluation against production queries
if __name__ == '__main__':
    prod_queries = sample_production_queries('/var/log/rag_queries.jsonl')
    harness = RAGEvaluationHarness(retriever, llm_client, config)
    metrics = harness.run(prod_queries)
    send_slack_digest(metrics)

Visualizzare l'andamento delle metriche nel tempo

I numeri grezzi in un file JSONL sono difficili da interpretare a colpo d'occhio. Crei una semplice visualizzazione dell'andamento che riporti ogni metrica delle ultime 20 esecuzioni della valutazione in un grafico a linee. Utilizzi la marca temporale dell'esecuzione come asse x e il punteggio della metrica come asse y. Tracci una linea orizzontale in corrispondenza della soglia minima accettabile. Quando una metrica scende al di sotto della linea della soglia, il problema è immediatamente visibile senza dover leggere i dati grezzi. Strumenti come Matplotlib o una semplice dashboard web (Grafana, Streamlit) sono adatti a questo scopo.

import json
import matplotlib.pyplot as plt
from pathlib import Path

def plot_metric_trends(history_file='eval_history.jsonl', metric='hit_rate_at_5'):
    records = [
        json.loads(line)
        for line in Path(history_file).read_text().strip().split('\n')
    ]
    timestamps = [r['timestamp'][:10] for r in records[-20:]]
    scores = [r['metrics'].get(metric, 0) for r in records[-20:]]
    plt.figure(figsize=(10, 4))
    plt.plot(timestamps, scores, marker='o', label=metric)
    plt.axhline(y=0.80, color='r', linestyle='--', label='Min threshold')
    plt.title(f'{metric} over last 20 evaluations')
    plt.xticks(rotation=45)
    plt.tight_layout()
    plt.savefig(f'eval_trend_{metric}.png')
    print(f'Saved trend chart for {metric}')

Verifica rapida

Verifichi la Sua comprensione dei concetti di AI Engineering trattati in questa lezione.

Riepilogo della lezione

In questa lezione ha imparato: come strutturare un harness di valutazione completo con gestione dei dati di test, esecuzione della pipeline, calcolo delle metriche e generazione dei report, come confrontare le esecuzioni con una baseline e interrompere la CI/CD in caso di regressioni, come generare report leggibili per la revisione del team e come eseguire il monitoraggio pianificato in produzione per rilevare il data drift senza modifiche al codice. Ora dispone di una base completa per creare e valutare sistemi RAG di produzione.

Domande Frequenti

La lezione «Creare un sistema automatizzato di valutazione» è gratuita?

Sì — il testo completo di «Creare un sistema automatizzato di valutazione» è gratuito qui sul web. Per esercitarvi in modo interattivo (un editor di codice integrato e un tutor IA 24/7) e sbloccare il resto del corso AI Engineering Academy, passa a CoddyKit PRO. Il corso AI Engineering Academy include 4 lezioni in totale.

Cosa imparerò in «Creare un sistema automatizzato di valutazione»?

Creerà una pipeline di valutazione ripetibile che esegua l'intero sistema RAG su un test set, calcoli tutte le metriche e generi un report per monitorare i miglioramenti nel tempo. Eserciti AI Engineering Academy con codice pratico che esegui direttamente nel browser, e un tutor IA 24/7 risponde alle tue domande mentre lavori sulla lezione.

Ho bisogno di esperienza per iniziare AI Engineering Academy?

Non è richiesta alcuna esperienza precedente. AI Engineering Academy su CoddyKit è strutturato per principianti e studenti avanzati, quindi puoi iniziare da qui o dall'inizio e procedere al tuo ritmo. Questa è la lezione 4 di 4.

Quanto tempo richiede la lezione «Creare un sistema automatizzato di valutazione»?

La maggior parte delle lezioni CoddyKit richiede circa 5–10 minuti. Ogni lezione è breve e interattiva, quindi fai progressi costanti e riprendi esattamente da dove hai lasciato su web e app.

Posso scrivere ed eseguire codice in questa lezione AI Engineering Academy?

Sì. Ogni lezione AI Engineering Academy include un editor di codice integrato, quindi scrivi ed esegui codice reale direttamente nel tuo browser e ricevi feedback istantaneo dall'IA — nessuna configurazione locale necessaria.

Tutte le lezioni di questo corso

  1. Perché la valutazione è importante nel RAG
  2. Metriche di retrieval: hit rate, MRR e NDCG
  3. Metriche di generazione: faithfulness e rilevanza della risposta
  4. Creare un sistema automatizzato di valutazione
← Torna a AI Engineering Academy