AI Engineering Academy · Lección

Creación de un sistema automatizado de evaluación

Lección 4 de 413 pasos

Creación de un sistema automatizado de evaluación es una lección gratuita de AI Engineering Academy en CoddyKit. Esta es la lección 4 de 4. Puedes leer la lección completa abajo gratuitamente — luego la practicas en el navegador con un editor de código integrado y un tutor de IA 24/7. Forma parte de la ruta de aprendizaje de AI Engineering Academy, y tu progreso se sincroniza en la web y la app de CoddyKit. El curso de AI Engineering Academy incluye 4 lecciones en total.

¿Qué es un sistema de evaluación?

Un sistema de evaluación es un pipeline automatizado y repetible que ejecuta todo su sistema RAG con un conjunto de pruebas estandarizado, calcula todas las métricas y genera un informe. La palabra clave es repetible: cada vez que cambie su estrategia de división en fragmentos, el modelo de embeddings, el prompt o el LLM, ejecutará el mismo sistema de evaluación y comparará los resultados con una referencia. Esto transforma el desarrollo de RAG, que deja de consistir en ajustes subjetivos para convertirse en ingeniería basada en datos.

Arquitectura del sistema de evaluación

Un sistema de evaluación bien diseñado tiene cuatro capas: gestión de datos de prueba (cargar y versionar el conjunto de datos de referencia), ejecución del pipeline (ejecutar cada pregunta de prueba en todo el pipeline RAG), cálculo de métricas (calcular todas las métricas de recuperación y generación) y generación de informes (guardar los resultados con información de versión y producir una comparación con la referencia anterior). Cada capa debe poder probarse y configurarse de forma independiente.

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

Ejecución de cada caso de prueba

Para cada pregunta del conjunto de datos de referencia, ejecute el pipeline RAG completo y capture todas las salidas intermedias: los identificadores y las puntuaciones de los fragmentos recuperados, el contexto formateado, la respuesta generada y el número de tokens. Almacenar estos valores intermedios es esencial para depurar fallos: cuando una pregunta obtiene una puntuación baja, puede inspeccionar exactamente qué fragmentos se recuperaron y por qué la respuesta fue incorrecta, sin tener que volver a ejecutar el costoso 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)
    }

Cálculo de todas las métricas en una sola pasada

Después de recopilar las salidas de todos los casos de prueba, calcule el conjunto completo de métricas en una sola pasada sobre los resultados. Separe las métricas de recuperación (calculadas a partir de los identificadores de los fragmentos) de las métricas de generación (calculadas mediante llamadas al LLM evaluador). Agrupe las llamadas al LLM evaluador para maximizar la eficiencia: agrupe las evaluaciones de fidelidad y envíelas en paralelo con asyncio, en lugar de hacerlo secuencialmente. Registre el progreso, ya que las métricas de generación pueden tardar varios minutos en más de 100 casos de prueba.

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

Guardado de resultados con información de versión

Cada ejecución de evaluación debe guardarse con metadatos de versión para poder comparar los resultados entre configuraciones. Incluya el hash del commit de git del código, los parámetros de configuración (modelo de embeddings, tamaño de los fragmentos, K, umbral y modelo LLM), la marca de tiempo y una descripción legible de la ejecución. Guarde los resultados en un archivo JSON Lines o en una tabla de base de datos. Así creará un historial permanente de la evolución de su 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))

Comparación con la referencia

Después de cada ejecución, compare automáticamente con la referencia anterior y marque las regresiones. Una regresión es cualquier métrica que disminuya más allá de un umbral (por ejemplo, 2 puntos porcentuales). Muestre una tabla comparativa con los cambios de las métricas. Si alguna métrica empeora de forma significativa, la ejecución de evaluación debe fallar con un código de salida distinto de cero, lo que hará que un pipeline de CI/CD bloquee la implementación de ese cambio.

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

Integración en CI/CD

El sistema de evaluación resulta especialmente potente cuando se integra en su pipeline de CI/CD. Configúrelo para que se ejecute automáticamente en cada pull request que modifique la lógica de división en fragmentos, la configuración del modelo de embeddings, las plantillas de prompts o los parámetros de recuperación. El pipeline solo se aprueba si todas las métricas cumplen los umbrales mínimos y ninguna métrica empeora con respecto a la referencia de la rama principal. Esto evita que regresiones accidentales de calidad lleguen a producción.

# 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 }}

Generación de informes legibles

Además de los archivos de métricas sin procesar, genere un informe HTML o Markdown legible que su equipo pueda revisar en los comentarios del pull request. Incluya una tabla resumen con todas las métricas, una lista de los casos de prueba fallidos con la pregunta, la respuesta esperada, la respuesta generada y los fragmentos recuperados, así como un gráfico de tendencias que muestre las métricas de las 10 últimas ejecuciones. Los informes visuales ayudan a las partes interesadas no técnicas a entender si el sistema está mejorando.

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)

Seguimiento del coste de cada ejecución de evaluación

Las ejecuciones de evaluación tienen un coste: realizan llamadas a la API de embeddings, a la API del LLM y al LLM evaluador. Realice un seguimiento del coste de cada ejecución de evaluación junto con las métricas de calidad. Una evaluación completa de 100 casos de prueba suele costar entre 0,50 y 2,00 $, según los modelos utilizados. Utilice modelos más económicos para las llamadas de evaluación (GPT-4o-mini para puntuar la fidelidad) y reserve los modelos costosos para la generación. Incluya el coste estimado de la ejecución en el informe guardado para poder incluir la evaluación en el presupuesto de su ciclo de desarrollo.

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

Evaluación programada para la monitorización en producción

Además de la evaluación de CI/CD ante cambios en el código, ejecute el sistema de evaluación siguiendo un calendario en producción —diario o semanal— y realice pruebas con consultas reales de usuarios seleccionadas de sus registros. Esto detecta la deriva de datos: a medida que evoluciona el corpus de documentos y cambian los patrones de las consultas de los usuarios, la calidad del sistema puede degradarse sin que se produzca ningún cambio en el código. Programe evaluaciones semanales que seleccionen 50 consultas recientes de usuarios, las evalúen y envíen automáticamente un resumen de calidad al canal de Slack de su equipo.

# 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)

Visualización de las tendencias de las métricas a lo largo del tiempo

Los números sin procesar de un archivo JSONL son difíciles de interpretar de un vistazo. Cree una visualización sencilla de tendencias que represente cada métrica de las 20 últimas ejecuciones de evaluación en un gráfico de líneas. Utilice la marca de tiempo de la ejecución como eje x y la puntuación de la métrica como eje y. Trace una línea horizontal en el umbral mínimo aceptable. Cuando una métrica cae por debajo de la línea del umbral, el problema se hace visible de inmediato sin tener que revisar los datos sin procesar. Herramientas como Matplotlib o un dashboard web sencillo (Grafana, Streamlit) funcionan bien para esto.

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}')

Comprobación rápida

Compruebe su comprensión de los conceptos de Ingeniería de IA de esta lección.

Responde en la lección interactiva

Resumen de la lección

En esta lección ha aprendido: cómo estructurar un sistema de evaluación completo con gestión de datos de prueba, ejecución del pipeline, cálculo de métricas y generación de informes; cómo comparar las ejecuciones con una referencia y hacer que CI/CD falle ante regresiones; cómo generar informes legibles para que los revise el equipo; y cómo ejecutar una monitorización programada en producción para detectar la deriva de datos sin cambios en el código. Ahora cuenta con una base completa para crear y evaluar sistemas RAG en producción.

Gratis para empezar

Aprende Python con un tutor de IA — gratis

Escribe y ejecuta código real en tu navegador, obtén ayuda instantánea de un tutor de IA disponible 24/7 y continúa donde lo dejaste en la web o en la aplicación.

Cursos: 30
Lecciones: 120

Empezar a aprender gratis Ver el curso completo

Preguntas frecuentes

¿La lección «Creación de un sistema automatizado de evaluación» es gratis?

Sí — el texto completo de «Creación de un sistema automatizado de evaluación» es gratis para leer aquí en la web. Para practicarla de forma interactiva (editor de código integrado y tutor de IA 24/7) y desbloquear el resto del curso de AI Engineering Academy, actualiza a CoddyKit PRO. El curso de AI Engineering Academy incluye 4 lecciones en total.

¿Qué aprenderé en «Creación de un sistema automatizado de evaluación»?

Creará un pipeline de evaluación reproducible que ejecute su sistema RAG completo contra un conjunto de prueba, calcule todas las métricas y genere un informe para poder hacer seguimiento de las mejo… Practicas AI Engineering Academy con código real que ejecutas directamente en el navegador, y un tutor de IA 24/7 responde tus preguntas mientras trabajas en la lección.

¿Necesito experiencia previa para empezar AI Engineering Academy?

No se requiere experiencia previa. AI Engineering Academy en CoddyKit está estructurado para principiantes hasta estudiantes avanzados, así que puedes empezar aquí o desde el inicio y avanzar a tu ritmo. Esta es la lección 4 de 4.

¿Cuánto tiempo toma la lección «Creación de un sistema automatizado de evaluación»?

La mayoría de las lecciones de CoddyKit toman alrededor de 5–10 minutos. Cada una es compacta e interactiva, así que avanzas constantemente y retomas exactamente por donde dejaste en la web y la app.

¿Puedo escribir y ejecutar código en esta lección de AI Engineering Academy?

Sí. Cada lección de AI Engineering Academy incluye un editor de código integrado, así que escribes y ejecutas código real directamente en tu navegador y obtienes retroalimentación instantánea de IA — sin configuración local necesaria.

Todas las lecciones de este curso

Por qué es importante la evaluación en RAG
Métricas de recuperación: hit rate, MRR y NDCG
Métricas de generación: fidelidad y relevancia de la respuesta
Creación de un sistema automatizado de evaluación

← Volver a AI Engineering Academy