El harness en 30 minutos

1Por qué este apéndice (y qué NO es)

Vienes aquí porque, a partir de N2, el curso decide con datos: ¿el agente robusto mejora al frágil? ¿el multi-agente compensa su coste? ¿sobra el framework? Esas preguntas se contestan midiendo. Si hiciste el curso Evals y Observabilidad de sistemas LLM, ya tienes el harness serio; úsalo. Si no, este apéndice te monta el instrumento mínimo.

Prerequisitos:

• Elcano corriendo como grafo (el grafo.py de N0·L4).
• Python 3.
• El backend mock de billing del lab (determinista).

Vas a capturar tres señales por run:

• Coste — tokens consumidos.
• Latencia — tiempo de pared, total y por superstep.
• Fiabilidad — ¿el run terminó y produjo el resultado esperado?

Lo que NO es esto: un sistema de evaluación serio. No hay error analysis (mirar fallos uno a uno y agruparlos por causa). No hay juez validado contra humanos. Sin esas dos cosas, los números absolutos no son de fiar. Para diagnosticar y comparar entre sí las versiones de Elcano —el uso que le da este curso— el mínimo alcanza. La herramienta de precisión es el otro curso.

2La palanca que ya tienes: stream

No necesitas una librería nueva para instrumentar a Elcano. Necesitas la API que ya usaste en L4 para auditar: stream.

stream ejecuta el grafo emitiendo cada superstep. Con stream_mode="updates" ves qué devolvió cada nodo; con "values", el estado completo tras cada paso. Esos eventos son los puntos naturales donde medir. Cada uno marca el avance de un nodo: puedes tomar el tiempo entre eventos y contar cuántos pasos dio el run.

La idea del harness mínimo: envolver stream en una función que cronometra cada superstep, acumula los tokens y comprueba el resultado final. Sin tocar la lógica de Elcano.

3El wrapper de medición

Lee el bloque entero antes de la explicación. Una función envuelve el stream del grafo, mide y devuelve un registro del run.

1# harness_min.py — instrumento mínimo sobre el grafo de Elcano
2import time
3from grafo import elcano        # tu grafo compilado de N0·L4
4
5def medir_run(peticion: str) -> dict:
6    entrada = {"messages": [{"role": "user", "content": peticion}]}
7    t0 = time.perf_counter()
8    supersteps = []
9    estado_final = None
10
11    # stream_mode="values": cada evento es el estado COMPLETO tras un superstep.
12    for snapshot in elcano.stream(entrada, stream_mode="values"):
13        supersteps.append({"t": time.perf_counter() - t0,
14                           "n_mensajes": len(snapshot["messages"])})
15        estado_final = snapshot
16
17    return {
18        "peticion": peticion,
19        "latencia_total_s": round(time.perf_counter() - t0, 3),
20        "n_supersteps": len(supersteps),
21        "tokens": _tokens_del_run(estado_final),   # ver §4
22        "ok": _exito(estado_final),                # ver §5
23        "supersteps": supersteps,
24    }

Tres cosas de este wrapper:

• stream(stream_mode="values") es el motor. Cada vuelta del for es un superstep; tomas el tiempo y el tamaño del historial en cada uno.
• No instrumenta la lógica de Elcano. El grafo no cambia. Mides desde fuera, observando el stream —igual que auditabas en L4—.
• Devuelve un registro, no imprime. Guardas ese dict por run (un JSONL, por ejemplo) y comparas runs entre sí después.

La latencia por superstep es la diferencia entre t consecutivos. Te dice qué nodo se lleva el tiempo —casi siempre el de modelo, no el de tools—.

4Coste: contar tokens

El coste real de un agente está en los tokens, no en el overhead del grafo. Tu cliente LLM ya expone el uso de tokens en cada respuesta del modelo —el mismo dato que veías en tu agente previo—. El harness mínimo lo recoge de los mensajes del estado final:

1def _tokens_del_run(estado_final: dict) -> int:
2    # Los mensajes del modelo traen su uso de tokens (como en tu cliente LLM).
3    # Suma el uso de cada mensaje AI del historial. La forma exacta del campo
4    # depende de tu cliente: ajústalo a cómo TU modelo reporta el uso.
5    total = 0
6    for m in estado_final["messages"]:
7        uso = getattr(m, "usage_metadata", None)   # depende del cliente
8        if uso:
9            total += uso.get("total_tokens", 0)
10    return total

Una nota de honestidad: la forma del campo de uso depende de tu cliente LLM y de tu proveedor. Este apéndice no fija esa firma —ajústala a lo que tu modelo reporta—. Lo que importa es el principio: el coste se mide en tokens, sumados sobre las llamadas al modelo del run.

5Fiabilidad: ¿el run hizo lo que debía?

La tercera señal es la más fácil de hacer mal. "¿Funcionó?" no es una impresión; es una comprobación determinista sobre el estado final.

Para Elcano, un run es un éxito si terminó (alcanzó END, no se quedó colgado) y produjo el resultado esperado para el caso. Con el backend mock determinista, puedes comprobar el efecto: ¿se emitió el reembolso correcto? ¿el plan cubría los pasos?

1def _exito(estado_final: dict) -> bool:
2    # Ilustrativo: adáptalo al mock que viene con tu kit de lab.
3    # `_billing_mock` es el backend falso del lab; `hubo_reembolso_exacto` es
4    # un ejemplo de aserción — usa la que exponga tu mock para comprobar el efecto.
5    ultimo = estado_final["messages"][-1]
6    respondio = not getattr(ultimo, "tool_calls", None)   # terminó: no pide más tools
7    reembolso_ok = _billing_mock.hubo_reembolso_exacto(esperado=19.0)  # efecto correcto
8    return respondio and reembolso_ok

Esta comprobación es determinista porque el mock lo es (el cuerpo de _exito es ilustrativo: el mock concreto llega con el kit de lab). Mide presencia de efecto, no calidad de prosa. Para juicios cualitativos —"¿la respuesta al cliente es clara?"— necesitas un juez, y un juez sin validar no es de fiar. Eso es territorio del curso de evals.

6Comparar versiones (el uso de verdad)

El harness mínimo no sirve para certificar un número absoluto. Sirve para lo que el curso necesita: comparar dos versiones de Elcano sobre los mismos casos.

1CASOS = ["Reembolsa el último cargo de la cuenta A-100",
2         "Baja mi plan al básico",
3         "¿Cuál fue mi último cargo?"]      # solo-lectura
4
5def comparar(casos: list[str]) -> None:
6    for c in casos:
7        r = medir_run(c)
8        estado = "ok" if r["ok"] else "FALLO"
9        print(f"{estado:5} | {r['latencia_total_s']:>6}s | "
10              f"{r['tokens']:>6} tok | {r['n_supersteps']} pasos | {c[:40]}")

Corres comparar(CASOS) sobre la versión actual, guardas la tabla, aplicas una mejora (un retry en N2, un worker en N3) y vuelves a correr. La pregunta no es "¿cuánto vale este run?" sino "¿esta versión es más fiable / más barata / más rápida que la anterior, sobre los mismos casos?". Para esa comparación, un instrumento consistente —aunque imperfecto— alcanza: el sesgo se aplica a las dos versiones por igual.

7Los límites del starter kit

Llegaste con un instrumento que mide, pero que no es de fiar en términos absolutos. Conviene saber dónde está el límite antes de apoyar una decisión en él.

• Sin error analysis. No agrupas los fallos por causa raíz. Sabes cuántos runs fallan, no por qué en categorías accionables.
• Sin juez validado. La fiabilidad aquí es una comprobación de efecto contra el mock. Funciona porque el mock es determinista; no se traslada a juzgar calidad de lenguaje.
• Coste aproximado. El conteo de tokens depende de cómo tu cliente reporta el uso; afínalo antes de fiarte de la cifra exacta.

Esto deja de alcanzar cuando necesitas el número absoluto: certificar a un cliente un umbral, o decidir un despliegue por el score sin un baseline. Ahí necesitas error analysis y un juez validado contra humanos.

Eso lo construye el curso Evals y Observabilidad de sistemas LLM. Este harness es el andamio para empezar a medir; el otro curso es la herramienta de precisión. Para instrumentar a Elcano en N0 y comparar versiones en N2–N4, el andamio cumple.

De vuelta al nivel: con Elcano instrumentado, cierras N0 con el checkpoint C0 — el grafo tipado y auditable que el resto del curso endurece y mide.