Evaluación automatizada de pipelines RAG: LLM-as-judge en FastAPI y OpenAI

Implementar un sistema RAG (Retrieval-Augmented Generation) no garantiza resultados útiles en producción. Lo verdaderamente crítico es medir su rendimiento de forma continua y automática, con la misma precisión que una API transaccional. Sin una evaluación cuantitativa y trazable en tiempo real, los pipelines RAG pueden degradarse silenciosamente y fallar al escalar.

Este artículo describe arquitectura y patrones para evaluar pipelines RAG en FastAPI, OpenAI API y PostgreSQL/pgvector, usando LLMs como jueces ("LLM-as-judge"). Analizamos métricas numéricas, rastreo de fallos reales y trade-offs de coste y latencia, con código real y datos comparativos.

Arquitectura: pipeline RAG evaluado "end-to-end"

• Pipeline RAG backend en FastAPI (Python 3.11)
• Orquestación de inferencia vía OpenAI SDK v1.23.3: completions, function calling
• Base vectorial pgvector (PostgreSQL 15, pgvector 0.5) + S3 documentos fuente
• Evaluador LLM-as-judge (OpenAI GPT-4o, stream, JSON schema para outputs)
• Postprocesado resultados y métricas en tabla de observabilidad, consultas SQL

"Si no tienes métricas automáticas por pregunta, no tienes control ni mejora continua. Solo anécdotas y pánico pos-lanzamiento."

Diseño del evaluador: LLM-as-judge con outputs estructurados

Motivación técnica para CTOs/ingenieros

• El juicio humano es lento (10-15 evals/hora por revisor). Necesitamos 1.000+ evals/día a coste marginal (≤1,5$/1k preguntas).
• Evitar rejugar prompts con parseo manual expuesto a errores. Validamos outputs vía JSON schema validado por el LLM y el cliente Python.
• Reducimos ambigüedad: puntuaciones [0-5], justificación, flag de insatisfacción, tracking granular.

Definición de schema y prompt para el evaluador (OpenAI, function calling)

from openai import OpenAI
import asyncio
import jsonschema

EVAL_SCHEMA = {
    "type": "object",
    "properties": {
        "score": {"type": "integer", "minimum": 0, "maximum": 5},
        "reason": {"type": "string"},
        "is_unsatisfactory": {"type": "boolean"},
        "failure_code": {"type": "string"}
    },
    "required": ["score", "reason", "is_unsatisfactory"]
}

async def eval_rag_response(question, context, answer):
    openai = OpenAI()
    function_def = [{
        "name": "evaluate_rag_answer",
        "parameters": EVAL_SCHEMA
    }]
    messages = [
        {"role": "system", "content": "Eres un experto revisando respuestas RAG. Devuelve JSON válido."},
        {"role": "user", "content": f"Pregunta: {question}\
Contexto: {context}\
Respuesta: {answer}"}
    ]
    result = await openai.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        functions=function_def,
        function_call="auto"
    )
    arguments = result.choices[0].message.function_call.arguments
    eval_obj = json.loads(arguments)
    jsonschema.validate(eval_obj, EVAL_SCHEMA)
    return eval_obj

• El LLM rellena el JSON precísamente: parseo directo en Python, errores mínimos.
• Esquema compatible con series temporales en PostgreSQL y dashboards.
• Permite filtrar outliers y prompts con flags de "failure_code" automáticos.

Métricas generadas: qué y cómo auditar

• Score numérico [0-5]: cada respuesta RAG recibe rating, usable para comparativa y benchmark A/B.
• Explicación (reason): texto corto que permite QC rápido (ver en dashboard, buscar patrones de fallo).
• Flag insatisfactorio: boolean explícito para detectar regresiones (pico de is_unsatisfactory).
• Failure code: mapeo opcional: "context_mismatch", "incomplete_answer", etc.

-- Ejemplo de SQL para calcular score medio agrupado por pipeline y día
WITH evals_day AS (
  SELECT 
    pipeline_id, 
    date_trunc('day', created_at) AS day, 
    AVG(score) AS avg_score, 
    COUNT(*) AS evals, 
    SUM(CASE WHEN is_unsatisfactory THEN 1 ELSE 0 END)::float / COUNT(*) AS unsat_ratio
  FROM rag_eval
  WHERE created_at > now() - interval '30 days'
  GROUP BY pipeline_id, day
)
SELECT * FROM evals_day WHERE unsat_ratio > 0.15 ORDER BY day DESC;

Integración FastAPI: procesamiento asíncrono y tolerancia a fallos

Para ingenieros backend/CTOs: patrón de integración robusto

• La inferencia evaluadora añade ~950ms por evaluación (latencia OpenAI GPT-4o, batch size 1, mayo 2024).
• Evitar bloquear el request principal del usuario. Patrón recomendable: endpoint principal responde rápido, metricado via task background.
• Manejo de reintentos exponenciales y queues en fallos de red (HTTP5xx, rate limit, fallos idempotentes). Usar aioboto3/SQS o simplemente asyncio.create_task + retry.

from fastapi import FastAPI, BackgroundTasks
from sqlalchemy.orm import Session
from models import RagEval
import asyncio, httpx
from retrying import retry

app = FastAPI()

@retry(wait_exponential_multiplier=200, stop_max_attempt_number=3)
async def safe_eval_store(eval_obj, db: Session):
    db_obj = RagEval(**eval_obj)
    db.add(db_obj)
    db.commit()

@app.post("/rag-answer")
async def rag_answer(data: dict, background_tasks: BackgroundTasks, db: Session):
    # Procesar la respuesta RAG (fuera de scope)
    answer = "..."
    async def process_eval():
        eval_obj = await eval_rag_response(data["question"], data["context"], answer)
        await safe_eval_store(eval_obj, db)
    background_tasks.add_task(process_eval)
    return {"answer": answer}

• Cada petición /rag-answer dispara una task de evaluación en background. No hay retardo extra para el usuario final.
• La función safe_eval_store se reintenta automáticamente ante fallos de conexión o lock de base de datos.
• Escalable y testable: más barato y simple que event buses si el throughput es <4 rps.

Rastreo, observabilidad y límites reales (coste, latencia, cuota API)

• Evaluar 10.000 preguntas/día consume ~6,2$ (OpenAI GPT-4o, 1k tokens/eval, mayo 2024).
• Latencia promedio end-to-end RAG+eval: 120 + 950 = 1070ms (sin batch evals).
• Cuotas OpenAI account-level: 90.000 tokens/min (ver docs), válida para pipelines <50 rps sin saltar rate limit.
• PostgreSQL/pgvector permite consultar >5.000 evals/segundo para dashboards y alarmística.

El mayor cuello de botella siempre es la inferencia del evaluador LLM -- nunca el almacenamiento o la red.

Impacto en la operación: métricas, ahorro y riesgo real

• Coste: 10.000 evals/día equivale a 1,24$; revisor humano costaría >250€/día (FTE).
• Latencia: con procesamiento background, respuesta rápida (<250ms) y métricas trazadas en <12m para reporting semanal.
• Degradación detectada en producción: con score y % insatisfactorio, los picos de regresión se identifican horas/días antes (no semanas después por QA manual).
• Riesgo de no implantarlo: sin este tipo de trazabilidad, un fallo de semántica en RAG puede pasar desapercibido y desbordar soporte, con coste operacional x6.

En Agente 404 desplegamos pipelines RAG auditados para clientes de sector legal, industrial y consumo, con reducción del tiempo de detección de errores del 80%. Si tu pipeline no da cifras automáticas, sólo verás los errores cuando sea demasiado tarde.

¿Dudas sobre cómo evaluar tu RAG en producción, con métricas trazables y coste acotado? Analizamos tu stack y construimos sistemas medibles en agente404.com.