Mini-proyecto: envuelve un run y mira adentro

Descripción

Siete lecciones construyeron, por separado, cada pieza de vocabulario que hacía falta: el problema (lección 03), las cuatro señales (lección 04), costo y latencia calculados por primera vez (lección 05), la frontera de lo que esta guía sí y no hace (lección 06), y el encargo completo que las justifica a todas (lección 07). Este mini-proyecto las junta en un solo lugar: una función, run_and_observe, que envuelve run_reservo_agent —sin tocar una línea de su código— y devuelve, junto a la respuesta de siempre, un reporte completo con las cuatro señales del módulo.

No es el logger estructurado del Módulo 2 —no hay trace_id, no hay formato JSON persistido a un archivo, no hay eventos por paso—. Es, deliberadamente, la versión más simple posible de "envolver un run con instrumentación": una función que llama a la función de siempre, mide lo que puede medir, y te lo entrega junto con la respuesta. Cuando termines esta lección, vas a tener la primera respuesta real —todavía mínima— al encargo de la lección 07.

Conexión con el módulo

Esta es la síntesis de las ocho lecciones. run_and_observe no inventa ninguna técnica nueva: reusa estimate_cost_cents y estimate_run_tokens de la lección 05, la lógica de conteo de la lección 04, y el modelo de latencia de la lección 05 — todo junto, alrededor de run_reservo_agent, ejecutado sobre un lote de runs reales.

Ejemplo trabajado: run_and_observe, sobre un lote de cuatro runs

La función que envuelve, sin tocar el agente

import itertools
import json
import logging
import statistics
from dataclasses import dataclass, asdict

import reservo_agent as ra

logging.basicConfig(level=logging.INFO, format="%(message)s")

INPUT_PRICE_CENTS_PER_MILLION_TOKENS = 300
OUTPUT_PRICE_CENTS_PER_MILLION_TOKENS = 1500

TOOL_LATENCY_MS = {
    "list_rooms": 40,
    "get_quote": 25,
    "book_room": 120,
    "cancel_booking": 90,
}

_run_ids = itertools.count(1)


def estimate_cost_cents(input_tokens, output_tokens):
    return (
        input_tokens * INPUT_PRICE_CENTS_PER_MILLION_TOKENS
        + output_tokens * OUTPUT_PRICE_CENTS_PER_MILLION_TOKENS
    ) // 1_000_000


@dataclass
class RunReport:
    """Todo lo que este módulo puede decir de UN run, en un solo lugar."""
    run_id: int
    question: str
    steps: int
    tool_calls: int
    tool_errors: int
    input_tokens: int
    output_tokens: int
    cost_cents: int
    latency_ms: int
    completed: bool

    @property
    def tool_fail_rate(self):
        return self.tool_errors / self.tool_calls if self.tool_calls else 0.0


def run_and_observe(question, model_script, max_iterations=10):
    """Envuelve run_reservo_agent (agent-fundamentals M8, SIN tocar su
    lógica) y mide lo que la lección 03 mostró que no se podía ver: pasos,
    tool calls, errores, tokens estimados, costo, latencia modelada."""
    run_id = next(_run_ids)
    completed = True
    try:
        final, history = ra.run_reservo_agent(question, model_script, max_iterations=max_iterations)
    except RuntimeError:
        completed = False
        raise

    input_chars = output_chars = 0
    tool_calls = tool_errors = 0
    latency_ms = 0
    tool_use_name = {}
    for turn in history:
        content = turn["content"]
        if isinstance(content, str):
            input_chars += len(content)
            continue
        for block in content:
            if block["type"] == "tool_use":
                tool_calls += 1
                output_chars += len(json.dumps(block["input"]))
                tool_use_name[block["id"]] = block["name"]
            elif block["type"] == "tool_result":
                input_chars += len(block["content"])
                if block.get("is_error"):
                    tool_errors += 1
                else:
                    latency_ms += TOOL_LATENCY_MS.get(tool_use_name.get(block["tool_use_id"]), 0)
            elif block["type"] == "text":
                output_chars += len(block["text"])

    input_tokens, output_tokens = input_chars // 4, output_chars // 4
    report = RunReport(
        run_id=run_id, question=question, steps=len(history),
        tool_calls=tool_calls, tool_errors=tool_errors,
        input_tokens=input_tokens, output_tokens=output_tokens,
        cost_cents=estimate_cost_cents(input_tokens, output_tokens),
        latency_ms=latency_ms, completed=completed,
    )
    logging.info("run %s completado: %s pasos, %s tool calls (%s errores), %sms, %s centavos",
                 report.run_id, report.steps, report.tool_calls, report.tool_errors,
                 report.latency_ms, report.cost_cents)
    return final, report

Lee la firma con cuidado: run_and_observe(question, model_script, max_iterations=10) — exactamente los mismos parámetros que run_reservo_agent. Adentro, la primera línea real es final, history = ra.run_reservo_agent(...) — la única llamada a la lógica del agente en toda la función. Todo lo que sigue —contar tool calls, sumar errores, estimar tokens, calcular costo, sumar latencia modelada— es la envoltura: lee lo que run_reservo_agent ya produjo, nunca cambia cómo lo produce. logging.info(...) es el único uso del módulo logging en este módulo — una sola línea por run, sin formato JSON ni trace_id todavía: eso es, deliberadamente, trabajo del Módulo 2.

Corriendo el lote: cuatro tareas, ninguna repetida a propósito

script_a = [
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_01", "name": "list_rooms", "input": {}}]},
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_02", "name": "get_quote",
         "input": {"room": "Focus", "tier": "premium", "hours": 3}}]},
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_03", "name": "get_quote",
         "input": {"room": "Focus", "tier": "pro", "hours": 3}}]},
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_04", "name": "book_room",
         "input": {"room": "Focus", "tier": "pro", "hours": 3, "member": "Ana"}}]},
    {"stop_reason": "end_turn", "content": [
        {"type": "text", "text": "Reservé Focus pro por 3 horas para Ana. Total $60.00. Confirmación #1."}]},
]

script_b = [
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_01", "name": "list_rooms", "input": {}}]},
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_02", "name": "get_quote",
         "input": {"room": "Boardroom", "tier": "pro", "hours": 1}}]},
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_03", "name": "book_room",
         "input": {"room": "Boardroom", "tier": "pro", "hours": 1, "member": "Sofía"}}]},
    {"stop_reason": "end_turn", "content": [
        {"type": "text", "text": "Reservé Boardroom pro por 1 hora para Sofía. Total $64.00. Confirmación #2."}]},
]

script_c = [
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_01", "name": "list_rooms", "input": {}}]},
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_02", "name": "get_quote",
         "input": {"room": "Focus", "tier": "premium", "hours": 3}}]},
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_03", "name": "get_quote",
         "input": {"room": "Focus", "tier": "pro", "hours": 0}}]},
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_04", "name": "get_quote",
         "input": {"room": "Focus", "tier": "pro", "hours": 3}}]},
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_05", "name": "book_room",
         "input": {"room": "Focus", "tier": "pro", "hours": 3, "member": "Ana"}}]},
    {"stop_reason": "end_turn", "content": [
        {"type": "text", "text": "Reservé Focus pro por 3 horas para Ana. Total $60.00. Confirmación #3."}]},
]

script_d = [
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_01", "name": "book_room",
         "input": {"room": "Studio", "tier": "basic", "hours": 1, "member": "Diego"}}]},
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_02", "name": "cancel_booking", "input": {"id": 4}}]},
    {"stop_reason": "end_turn", "content": [
        {"type": "text", "text": "Reservé y luego cancelé Studio basic 1h para Diego."}]},
]

reports = []
for question, script in [
    ("Reserva Focus pro 3h para Ana", script_a),
    ("Reserva Boardroom pro 1h para Sofía", script_b),
    ("Reserva Focus pro 3h para Ana", script_c),
    ("Reserva y cancela Studio basic 1h para Diego", script_d),
]:
    final, report = run_and_observe(question, script)
    reports.append(report)

print()
print("=== RunReport de cada run ===")
for r in reports:
    print(asdict(r))

print()
print("=== resumen del lote ===")
print("runs                    :", len(reports))
print("tool calls totales      :", sum(r.tool_calls for r in reports))
print("tool errors totales     :", sum(r.tool_errors for r in reports))
print("costo total (centavos)  :", sum(r.cost_cents for r in reports))
print("latencia promedio (ms)  :", round(statistics.mean(r.latency_ms for r in reports), 1))

Qué esperar:

run 1 completado: 10 pasos, 4 tool calls (1 errores), 185ms, 0 centavos
run 2 completado: 8 pasos, 3 tool calls (0 errores), 185ms, 0 centavos
run 3 completado: 12 pasos, 5 tool calls (2 errores), 185ms, 0 centavos
run 4 completado: 6 pasos, 2 tool calls (0 errores), 210ms, 0 centavos

=== RunReport de cada run ===
{'run_id': 1, 'question': 'Reserva Focus pro 3h para Ana', 'steps': 10, 'tool_calls': 4, 'tool_errors': 1, 'input_tokens': 64, 'output_tokens': 56, 'cost_cents': 0, 'latency_ms': 185, 'completed': True}
{'run_id': 2, 'question': 'Reserva Boardroom pro 1h para Sofía', 'steps': 8, 'tool_calls': 3, 'tool_errors': 0, 'input_tokens': 53, 'output_tokens': 49, 'cost_cents': 0, 'latency_ms': 185, 'completed': True}
{'run_id': 3, 'question': 'Reserva Focus pro 3h para Ana', 'steps': 12, 'tool_calls': 5, 'tool_errors': 2, 'input_tokens': 70, 'output_tokens': 67, 'cost_cents': 0, 'latency_ms': 185, 'completed': True}
{'run_id': 4, 'question': 'Reserva y cancela Studio basic 1h para Diego', 'steps': 6, 'tool_calls': 2, 'tool_errors': 0, 'input_tokens': 24, 'output_tokens': 31, 'cost_cents': 0, 'latency_ms': 210, 'completed': True}

=== resumen del lote ===
runs                    : 4
tool calls totales      : 14
tool errors totales     : 3
costo total (centavos)  : 0
latencia promedio (ms)  : 191.2

Lee el resumen del lote con las cuatro señales de la lección 04 en mente: tasa de error del lote es 0% (completed=True en los cuatro RunReport); tasa de fallo por herramienta es 3/14 = 21.4%; costo total es 0 centavos —de nuevo, la respuesta honesta para runs de este tamaño, no un error—; latencia promedio, calculada con statistics.mean sobre los cuatro valores de latency_ms, es 191.2 ms. Fíjate que el promedio no es igual a ninguno de los cuatro valores individuales (185, 185, 185, 210) — es sensible al run 4, que ejecutó book_room y cancel_booking en vez de la secuencia habitual list_rooms → get_quote → book_room, y por eso tiene una mezcla distinta de tools reales detrás.

El límite que persiste: un run que falla, ni siquiera esta envoltura lo salva

run_and_observe mejora mucho sobre no tener nada — pero hereda, a propósito, el límite exacto que la lección 03 nombró: si run_reservo_agent se agota el tope de iteraciones, la excepción se propaga antes de que run_and_observe pueda construir ningún RunReport, porque toda la lógica de conteo vive después de la llamada que puede fallar.

stuck_script = [
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": f"toolu_0{n}", "name": "list_rooms", "input": {}}]}
    for n in range(1, 4)
]
try:
    final, report = run_and_observe("Reserva algo", stuck_script, max_iterations=2)
except RuntimeError as exc:
    print("RuntimeError capturado en run_and_observe:", exc)

Qué esperar:

RuntimeError capturado en run_and_observe: max_iterations alcanzado (2)

Ni un RunReport, ni un solo logging.info, ni ningún rastro del intento — exactamente lo mismo que ya viste en la lección 03, ahora confirmado sobre la envoltura completa de este mini-proyecto. Esto no es un defecto de run_and_observe: es la prueba de que capturar información al final de un run —por más completo que sea el reporte que arma al final— nunca es suficiente para los runs que más necesitas diagnosticar, los que no llegan a un final limpio. Resolver esto de verdad —registrar cada paso según va pasando, no al final— es, con precisión, el problema que el Módulo 2 (logging estructurado y trazado de un run) construye desde su primera lección.

Errores comunes

1. Pensar que run_and_observe "ya es" el logger del Módulo 2. No lo es — no persiste nada a un archivo, no tiene trace_id, y como acabas de ver, pierde todo si el run falla. Es, deliberadamente, la envoltura más simple posible: suficiente para responder el encargo de la lección 07 de forma mínima, insuficiente para producción real.

2. Modificar run_reservo_agent para que sea más fácil de envolver. No hace falta — y no se hace en esta guía. run_and_observe demuestra, precisamente, que se puede construir una capa completa de medición sin tocar ni una línea del agente que agent-fundamentals ya entregó.

3. Calcular tool_fail_rate antes de sumar tool_calls de todo el lote. El RunReport individual expone tool_fail_rate como una @property por run — para el lote completo, hay que sumar tool_errors y tool_calls de todos los reportes primero, y dividir después (sum(r.tool_errors for r in reports) / sum(r.tool_calls for r in reports)), no promediar las tasas individuales de cada run, que da un número distinto y menos preciso cuando los runs tienen cantidades distintas de tool calls.

4. Usar statistics.mean sin tener claro qué esconde. El promedio de 191.2 ms de este lote es correcto, pero un promedio sobre solo cuatro runs puede ser engañoso — tres de los cuatro dieron exactamente 185 ms, y el cuarto (210) movió el promedio. Con un lote de miles de runs reales, mucho más variados, un solo promedio esconde aún más: por eso el Módulo 4 desarrolla percentiles (p50, p95), no solo el promedio.

5. Olvidar que completed=False en el except RuntimeError nunca llega a construir un RunReport. El código de run_and_observe marca completed = False y relanza la excepción — no la traga. Eso es intencional: la función nunca miente diciendo que un run se completó cuando no fue así, aunque el costo de esa honestidad sea no producir ningún reporte para ese caso.

Ejercicios

Ejercicio 1: Calcula la tasa de fallo por herramienta correcta del lote (Fácil)

Con los cuatro RunReport del ejemplo trabajado, calcula la tasa de fallo por herramienta del lote de la forma correcta (sumando antes de dividir) y compárala con el resultado —incorrecto— de promediar las cuatro tasas individuales.

correct_rate = sum(r.tool_errors for r in reports) / sum(r.tool_calls for r in reports)
wrong_rate = statistics.mean(r.tool_fail_rate for r in reports)
print(f"tasa correcta (suma/suma)      : {correct_rate:.1%}")
print(f"tasa incorrecta (promedio de %) : {wrong_rate:.1%}")

tasa correcta (suma/suma)      : 21.4%
tasa incorrecta (promedio de %) : 16.2%

Ejercicio 2: Agrega un quinto run limpio y observa cómo cambia el resumen (Medio)

Corre un quinto run —cualquier tarea sin ningún is_error, por ejemplo cotizar Studio pro 2h sin reservar nada— con run_and_observe, agrégalo a reports, y recalcula el resumen del lote completo (tool calls, tool errors, costo total, latencia promedio).

script_e = [
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_01", "name": "get_quote",
         "input": {"room": "Studio", "tier": "pro", "hours": 2}}]},
    {"stop_reason": "end_turn", "content": [
        {"type": "text", "text": "Studio pro 2h cuesta $64.00."}]},
]
final_e, report_e = run_and_observe("¿Cuánto cuesta Studio pro 2h?", script_e)
reports.append(report_e)

print("tool calls totales      :", sum(r.tool_calls for r in reports))
print("tool errors totales     :", sum(r.tool_errors for r in reports))
print("tool-fail rate del lote :", f"{sum(r.tool_errors for r in reports) / sum(r.tool_calls for r in reports):.1%}")
print("costo total (centavos)  :", sum(r.cost_cents for r in reports))
print("latencia promedio (ms)  :", round(statistics.mean(r.latency_ms for r in reports), 1))

tool calls totales      : 15
tool errors totales     : 3
tool-fail rate del lote : 20.0%
costo total (centavos)  : 0
latencia promedio (ms)  : 158

Ejercicio 3: Arregla run_and_observe para que sobreviva a un RuntimeError con un reporte parcial (Difícil)

El límite de la sección "El límite que persiste" es real: hoy, un RuntimeError no produce ningún RunReport. Sin construir el logger completo del Módulo 2, propón —en código— un cambio mínimo a run_and_observe que capture el RuntimeError, y devuelva un RunReport marcado con completed=False y el resto de los campos en 0, en vez de dejar que la excepción se propague sin ningún reporte. Ejecuta tu versión sobre el stuck_script del ejemplo trabajado y confirma que ahora sí obtienes un reporte.

def run_and_observe_v2(question, model_script, max_iterations=10):
    run_id = next(_run_ids)
    try:
        final, history = ra.run_reservo_agent(question, model_script, max_iterations=max_iterations)
    except RuntimeError as exc:
        report = RunReport(
            run_id=run_id, question=question, steps=0, tool_calls=0, tool_errors=0,
            input_tokens=0, output_tokens=0, cost_cents=0, latency_ms=0, completed=False,
        )
        logging.info("run %s NO completado: %s", report.run_id, exc)
        return None, report

    # ... el resto es idéntico a run_and_observe.
    input_chars = output_chars = 0
    tool_calls = tool_errors = 0
    latency_ms = 0
    tool_use_name = {}
    for turn in history:
        content = turn["content"]
        if isinstance(content, str):
            input_chars += len(content)
            continue
        for block in content:
            if block["type"] == "tool_use":
                tool_calls += 1
                output_chars += len(json.dumps(block["input"]))
                tool_use_name[block["id"]] = block["name"]
            elif block["type"] == "tool_result":
                input_chars += len(block["content"])
                if block.get("is_error"):
                    tool_errors += 1
                else:
                    latency_ms += TOOL_LATENCY_MS.get(tool_use_name.get(block["tool_use_id"]), 0)
            elif block["type"] == "text":
                output_chars += len(block["text"])

    input_tokens, output_tokens = input_chars // 4, output_chars // 4
    report = RunReport(
        run_id=run_id, question=question, steps=len(history),
        tool_calls=tool_calls, tool_errors=tool_errors,
        input_tokens=input_tokens, output_tokens=output_tokens,
        cost_cents=estimate_cost_cents(input_tokens, output_tokens),
        latency_ms=latency_ms, completed=True,
    )
    logging.info("run %s completado: %s pasos, %s tool calls (%s errores), %sms, %s centavos",
                 report.run_id, report.steps, report.tool_calls, report.tool_errors,
                 report.latency_ms, report.cost_cents)
    return final, report


stuck_script = [
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": f"toolu_0{n}", "name": "list_rooms", "input": {}}]}
    for n in range(1, 4)
]
final_stuck, report_stuck = run_and_observe_v2("Reserva algo", stuck_script, max_iterations=2)
print(report_stuck)

run N NO completado: max_iterations alcanzado (2)
RunReport(run_id=N, question='Reserva algo', steps=0, tool_calls=0, tool_errors=0, input_tokens=0, output_tokens=0, cost_cents=0, latency_ms=0, completed=False)

Resumen y siguiente paso

• Construimos run_and_observe: la primera envoltura de instrumentación de esta guía, que llama a run_reservo_agent sin tocarlo y devuelve, además de la respuesta de siempre, un RunReport con las cuatro señales del módulo.
• La corrimos sobre un lote real de cuatro tareas, ninguna repetida de una lección anterior: 14 tool calls totales, 3 errores (21.4% de tasa de fallo por herramienta), 0 centavos de costo total, 191.2 ms de latencia promedio.
• Confirmamos, ejecutado, que el límite de la lección 03 persiste incluso con esta envoltura: un RuntimeError sigue sin dejar ningún RunReport, porque toda la medición ocurre después de una llamada que puede fallar antes de terminar. Ese es, con precisión, el problema que el Módulo 2 resuelve.

Con esto se cierra el Módulo 1. Tienes el vocabulario completo —las cuatro señales—, la frontera clara —qué ya está construido, qué opera esta guía, qué es infraestructura—, el encargo que justifica los siete módulos que quedan, y tu primera envoltura de instrumentación, ejecutada sobre datos reales.

Siguiente módulo: Módulo 2 — Logging estructurado y trazado de un run. Ahí se resuelve, de fondo, el límite que esta lección dejó abierto a propósito: un logger que registra cada paso según va ocurriendo, con un trace_id determinista que correlaciona un run de punta a punta, capaz de dejar rastro incluso cuando el run entero termina en RuntimeError.

Recursos adicionales

1. Python — logging — El módulo que esta lección usa de forma mínima, y que el Módulo 2 desarrolla a fondo, con un formatter estructurado propio.
2. Python — statistics — statistics.mean, usado aquí sobre un lote de cuatro; statistics.median y statistics.quantiles son el contenido central del Módulo 4.
3. Python — dataclasses — RunReport, la estructura que junta las cuatro señales en un solo objeto por run.
4. Anthropic — Building effective agents — Sobre por qué medir un sistema agentic es una disciplina propia, separada de construirlo — el argumento completo de este módulo, cerrado.
5. Python 3.14 — What's New — La versión con la que se ejecutó cada línea de código de este módulo, incluido el RuntimeError real de esta última lección.