Mini-proyecto: un reporte de costo para los runs de Reservo

Descripción

Siete lecciones construyeron, por separado, cada pieza: los tokens como unidad (02), estimate_tokens con sus límites confirmados (03), el pricing fijo de claude-sonnet-5 (04), estimate_cost_cents y cost_for_run con desglose por tool call (05), la agregación de un lote y el escalado a miles de runs (06), y el costo como señal operacional con su propia frontera (07). Este mini-proyecto las junta todas en observability/cost_calculator.py completo, y lo ejecuta sobre el mismo lote de cuatro runs de Reservo que acompañó las últimas dos lecciones —Ana, Sofía, Diego, Carla—, cada uno correlacionado por el trace_id determinista que traced_run (Módulo 2) ya construyó.

El resultado es un reporte de costo integral: por run, con su desglose de tool calls; por lote, agregado correctamente; escalado a la magnitud en la que Reservo realmente opera; y con el criterio de señal de la lección 07 aplicado sobre el lote completo. Cuando termines esta lección, vas a tener el segundo artefacto completo de esta guía —el primero fue observability/run_logger.py del Módulo 2—, listo para que el Módulo 4 construya al lado el tercero.

Conexión con el módulo

Esta es la síntesis de las ocho lecciones. No hay ninguna pieza nueva de mecanismo —cost_for_run, aggregate_reports, flag_expensive_runs son exactamente las de las lecciones 05, 06 y 07—; el trabajo de este mini-proyecto es ensamblarlas en un solo archivo y ejecutarlas juntas, de punta a punta, sobre datos reales.

observability/cost_calculator.py, completo

Este es el archivo completo, con las piezas de cada lección en el orden en que se construyeron:

# observability/cost_calculator.py
"""Costo por run del agente de Reservo (Módulo 3): tokens estimados con
len(texto) // 4 (L03), pricing fijo de claude-sonnet-5 (L04), costo en
centavos con desglose por tool call (L05), agregación de lote y escalado
(L06), y el costo como señal operacional (L07). Reusa run_logger.py
(Módulo 2) para el trace_id -- no construye ninguna correlación nueva."""
import itertools
import json
import statistics
from dataclasses import dataclass, field

INPUT_PRICE_CENTS_PER_MILLION_TOKENS = 300    # $3.00 / 1M tokens -- claude-sonnet-5, precio de lista
OUTPUT_PRICE_CENTS_PER_MILLION_TOKENS = 1500  # $15.00 / 1M tokens -- claude-sonnet-5, precio de lista


def estimate_tokens(text):
    """L03: estimación de ORDEN DE MAGNITUD. Nunca un conteo exacto de un
    tokenizer real."""
    return len(text) // 4


def estimate_cost_cents(input_tokens, output_tokens):
    """L04+L05: la fórmula central del módulo, en centavos int."""
    return (
        input_tokens * INPUT_PRICE_CENTS_PER_MILLION_TOKENS
        + output_tokens * OUTPUT_PRICE_CENTS_PER_MILLION_TOKENS
    ) // 1_000_000


@dataclass
class StepCost:
    """L05: el costo estimado de UN tool call dentro de un run."""
    step: int
    tool: str
    input_tokens: int
    output_tokens: int
    cost_cents: int


@dataclass
class CostReport:
    """L05: el costo estimado de un run completo, identificado por el
    mismo trace_id de run_logger.py (Módulo 2)."""
    trace_id: str
    question: str
    steps: list = field(default_factory=list)
    input_tokens: int = 0
    output_tokens: int = 0
    cost_cents: int = 0


def cost_for_run(trace_id, question, history):
    """L05: recorre history (de run_reservo_agent, SIN tocarlo) y calcula
    el costo del run completo, con desglose por tool call."""
    steps = []
    total_input_text = question
    total_output_text = ""
    pending = {}
    step_counter = itertools.count(1)

    for turn in history:
        content = turn["content"]
        if isinstance(content, str):
            continue
        for block in content:
            if block["type"] == "tool_use":
                step = next(step_counter)
                args_text = json.dumps(block["input"])
                total_output_text += args_text
                pending[block["id"]] = {"step": step, "tool": block["name"], "args_text": args_text}
            elif block["type"] == "tool_result":
                entry = pending[block["tool_use_id"]]
                result_text = block["content"]
                total_input_text += result_text
                step_in = estimate_tokens(result_text)
                step_out = estimate_tokens(entry["args_text"])
                steps.append(StepCost(
                    step=entry["step"], tool=entry["tool"],
                    input_tokens=step_in, output_tokens=step_out,
                    cost_cents=estimate_cost_cents(step_in, step_out),
                ))
            elif block["type"] == "text":
                total_output_text += block["text"]

    input_tokens = estimate_tokens(total_input_text)
    output_tokens = estimate_tokens(total_output_text)
    return CostReport(
        trace_id=trace_id, question=question, steps=steps,
        input_tokens=input_tokens, output_tokens=output_tokens,
        cost_cents=estimate_cost_cents(input_tokens, output_tokens),
    )


def aggregate_reports(reports):
    """L06: suma TOKENS de todos los reports primero, aplica
    estimate_cost_cents UNA SOLA VEZ -- nunca suma cost_cents ya
    redondeados."""
    total_input = sum(r.input_tokens for r in reports)
    total_output = sum(r.output_tokens for r in reports)
    return total_input, total_output, estimate_cost_cents(total_input, total_output)


def project_cost_cents(input_tokens_per_run, output_tokens_per_run, n_runs):
    """L06: proyecta el costo de n_runs runs con un perfil de tokens dado
    -- multiplica tokens primero, redondea al final."""
    return estimate_cost_cents(input_tokens_per_run * n_runs, output_tokens_per_run * n_runs)


def flag_expensive_runs(reports, threshold_ratio=1.5):
    """L07: marca los reports cuyo total de tokens supera threshold_ratio
    veces el promedio del lote -- un criterio de FORMA, determinista."""
    totals = [r.input_tokens + r.output_tokens for r in reports]
    avg = statistics.mean(totals)
    flagged = []
    for r, total in zip(reports, totals):
        ratio = total / avg
        if ratio >= threshold_ratio:
            flagged.append((r, ratio))
    return flagged

Diez años de disciplina de esta guía resumidos en dos líneas: cada número de aquí es int, y cada estimación está rotulada, en su propio comentario, como lo que es —una aproximación de orden de magnitud, nunca un conteo exacto—.

Ejemplo trabajado: el lote completo, de punta a punta

Corre las mismas cuatro tareas de las lecciones 06 y 07, esta vez con el reporte completo: por run, agregado del lote, señal de costo, y proyección a escala.

import logging
import reservo_agent as ra
import run_logger as rl
import cost_calculator as cc

rl.logger.setLevel(logging.CRITICAL)  # silenciamos traced_run para este reporte final

# Los mismos cuatro guiones de las lecciones 06 y 07: Ana (con un error corregido),
# Sofía (limpio), Diego (reserva y cancela), Carla (compara seis combinaciones).
tasks = [
    ("Reserva Focus pro 3h para Ana", script_a),
    ("Reserva Boardroom pro 1h para Sofía", script_b),
    ("Reserva y cancela Studio basic 1h para Diego", script_d),
    ("Compara todas las salas antes de reservar la mejor opción para Carla", script_compare),
]

reports = []
for i, (question, script) in enumerate(tasks, start=1):
    with rl.traced_run(question, i) as trace_id:
        final, history = ra.run_reservo_agent(question, script)
    reports.append(cc.cost_for_run(trace_id, question, history))

print("=== reporte de costo por run ===")
for r in reports:
    print(f"{r.trace_id}  in={r.input_tokens:>4} out={r.output_tokens:>4} cost_cents={r.cost_cents}  {r.question!r}")

print()
print("=== agregado del lote ===")
total_in, total_out, total_cost = cc.aggregate_reports(reports)
print("input_tokens totales :", total_in)
print("output_tokens totales:", total_out)
print("costo total del lote :", total_cost, "centavos")

print()
print("=== señales: runs anómalamente caros (>= 1.5x el promedio) ===")
flagged = cc.flag_expensive_runs(reports, threshold_ratio=1.5)
for r, ratio in flagged:
    print(f"  {r.trace_id}: {ratio:.2f}x -- {r.question!r}")

print()
print("=== proyección a escala, usando el perfil promedio del lote ===")
avg_in, avg_out = total_in / len(reports), total_out / len(reports)
for n in (1_000, 10_000, 100_000):
    cost_n = cc.project_cost_cents(int(avg_in), int(avg_out), n)
    print(f"  {n:>7} runs -> {cost_n:>6} centavos = ${cost_n / 100:.2f}")

Qué esperar:

=== reporte de costo por run ===
run-8487582448eb  in=  64 out=  56 cost_cents=0  'Reserva Focus pro 3h para Ana'
run-ae6ff85cf0b0  in=  53 out=  49 cost_cents=0  'Reserva Boardroom pro 1h para Sofía'
run-2c27934d8a39  in=  27 out=  31 cost_cents=0  'Reserva y cancela Studio basic 1h para Diego'
run-cecde864aa84  in=  88 out= 118 cost_cents=0  'Compara todas las salas antes de reservar la mejor opción para Carla'

=== agregado del lote ===
input_tokens totales : 232
output_tokens totales: 254
costo total del lote : 0 centavos

=== señales: runs anómalamente caros (>= 1.5x el promedio) ===
  run-cecde864aa84: 1.70x -- 'Compara todas las salas antes de reservar la mejor opción para Carla'

=== proyección a escala, usando el perfil promedio del lote ===
     1000 runs ->    112 centavos = $1.12
    10000 runs ->   1126 centavos = $11.26
   100000 runs ->  11265 centavos = $112.65

Este es el reporte completo que la lección 01 prometió al abrir el módulo: cada CostReport correlacionado por su trace_id —el mismo que aparecería en RUN_LOG.jsonl si traced_run estuviera logueando a INFO en vez de CRITICAL—, el lote agregado con la aritmética correcta (0 centavos, honesto para cuatro runs de este tamaño), la señal de costo identificando exactamente el run de Carla como el atípico, y la proyección mostrando que ese mismo perfil de lote, a 100.000 runs, cuesta $112.65 reales. Cuatro líneas de "por run", tres bloques de análisis — todo desde un solo archivo, sin tocar ni una línea de reservo_agent.py ni de run_logger.py.

Errores comunes

1. Pensar que este mini-proyecto "ya optimiza" el costo del agente. No — como insistió la lección 07, todo lo que hace este archivo es medir y clasificar. Ninguna línea de cost_calculator.py cachea un tool_result, cambia de modelo, ni agrupa preguntas — esa es, con precisión, la frontera hacia cost-optimization-caching-guide.

2. Calcular el reporte de costo sin envolver el run en traced_run. Es posible llamar a cost_for_run directamente sobre un history obtenido sin traced_run —pasando cualquier string como trace_id—, pero se pierde la correlación real con RUN_LOG.jsonl del Módulo 2. El patrón correcto, usado en todo este mini-proyecto, es obtener el trace_id de traced_run y pasárselo a cost_for_run sin modificarlo.

3. Reordenar aggregate_reports y flag_expensive_runs esperando el mismo resultado. No importa el orden en que se llamen —cada una opera sobre reports, una lista ya calculada—, pero sí importa que ambas reciban la lista completa del lote: llamar a flag_expensive_runs sobre un subconjunto cambiaría el promedio contra el que se compara cada run, y por lo tanto qué runs se marcan.

4. Usar project_cost_cents con el perfil de un solo run cuando el lote tiene runs muy distintos entre sí. El ejemplo trabajado usa el promedio del lote (avg_in, avg_out) para proyectar, no el run de Ana en particular — con un lote donde el run más caro es 1.70x el promedio, proyectar con el run equivocado puede sobrestimar o subestimar el costo real a escala.

5. Olvidar que cost_calculator.py depende de history completo, no de RUN_LOG.jsonl. El Ejercicio 3 de esta lección confirma, con números reales, por qué intentar reconstruir el costo de un run solo a partir de las líneas persistidas del Módulo 2 subestima el resultado — el texto final de la respuesta del agente nunca se logueó como un evento propio.

Ejercicios

Ejercicio 1: Agrega un quinto run y recalcula el reporte completo (Fácil)

Agrega una quinta tarea al lote: Luis cancela la reserva 999, que no existe (is_error: True, sin ninguna reserva real de por medio). Corre cost_for_run sobre ese run, agrégalo a reports, y recalcula el agregado del lote con aggregate_reports.

script_luis = [
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_01", "name": "cancel_booking", "input": {"id": 999}}]},
    {"stop_reason": "end_turn", "content": [
        {"type": "text", "text": "No encontré esa reserva."}]},
]
with rl.traced_run("Cancela la reserva 999 para Luis", 5) as trace_id_luis:
    final_luis, history_luis = ra.run_reservo_agent("Cancela la reserva 999 para Luis", script_luis)

report_luis = cc.cost_for_run(trace_id_luis, "Cancela la reserva 999 para Luis", history_luis)
reports.append(report_luis)

print(f"{report_luis.trace_id}  in={report_luis.input_tokens:>4} out={report_luis.output_tokens:>4} "
      f"cost_cents={report_luis.cost_cents}  {report_luis.question!r}")

total_in, total_out, total_cost = cc.aggregate_reports(reports)
print("input_tokens totales (5 runs) :", total_in)
print("output_tokens totales (5 runs):", total_out)
print("costo total del lote (5 runs) :", total_cost, "centavos")

run-65388909596a  in=  16 out=   8 cost_cents=0  'Cancela la reserva 999 para Luis'

input_tokens totales (5 runs) : 248
output_tokens totales (5 runs): 262
costo total del lote (5 runs) : 0 centavos

Ejercicio 2: Encuentra el run que más contribuye al costo del lote (Medio)

Con los cinco CostReport del Ejercicio 1, encuentra el run con el mayor total de tokens (input_tokens + output_tokens), sin asumir de antemano cuál es.

busiest = max(reports, key=lambda r: r.input_tokens + r.output_tokens)
total_busiest = busiest.input_tokens + busiest.output_tokens
print(f"run que más contribuye: {busiest.trace_id} ({total_busiest} tokens) -- {busiest.question!r}")

run que más contribuye: run-cecde864aa84 (206 tokens) -- 'Compara todas las salas antes de reservar la mejor opción para Carla'

Ejercicio 3: Confirma por qué RUN_LOG.jsonl solo, sin history, subestima el costo (Difícil)

RUN_LOG.jsonl (Módulo 2) registra cada tool_use y tool_result, pero nunca registra el texto final de la respuesta del agente —el bloque text del turno con stop_reason: "end_turn"— como un evento propio de run_logger.py. Reconstruye el output_tokens del run de Ana de dos formas: (a) sumando solo los argumentos de cada tool_use (lo que sí se podría reconstruir desde RUN_LOG.jsonl), y (b) sumando esos mismos argumentos más el texto final de la respuesta (lo que cost_for_run sí calcula, porque tiene acceso al history completo). Compara ambos resultados.

import json

args_del_run_de_ana = [
    json.dumps({}),                                                          # list_rooms
    json.dumps({"room": "Focus", "tier": "premium", "hours": 3}),             # get_quote (rechazado)
    json.dumps({"room": "Focus", "tier": "pro", "hours": 3}),                 # get_quote (corregido)
    json.dumps({"room": "Focus", "tier": "pro", "hours": 3, "member": "Ana"}), # book_room
]
texto_final = "Reservé Focus pro por 3 horas para Ana. Total $60.00. Confirmación #1."

solo_args = "".join(args_del_run_de_ana)
args_mas_texto_final = solo_args + texto_final

print("output_tokens SOLO con args de tool_use (lo que RUN_LOG.jsonl podría dar):",
      cc.estimate_tokens(solo_args))
print("output_tokens CON el texto final (lo que cost_for_run SÍ calcula)      :",
      cc.estimate_tokens(args_mas_texto_final))
print("tokens que se pierden si solo se usa RUN_LOG.jsonl:",
      cc.estimate_tokens(args_mas_texto_final) - cc.estimate_tokens(solo_args))

output_tokens SOLO con args de tool_use (lo que RUN_LOG.jsonl podría dar): 38
output_tokens CON el texto final (lo que cost_for_run SÍ calcula)      : 56
tokens que se pierden si solo se usa RUN_LOG.jsonl: 18

Resumen y siguiente paso

• Ensamblamos observability/cost_calculator.py completo: estimate_tokens (L03), el pricing fijo (L04), estimate_cost_cents + cost_for_run con desglose (L05), aggregate_reports + project_cost_cents (L06), y flag_expensive_runs (L07) — ocho lecciones, un solo archivo.
• Lo ejecutamos sobre el lote completo de cuatro runs de Reservo, correlacionados por trace_id: un reporte por run, un agregado del lote (0 centavos, honesto), una señal de costo (el run de Carla, 1.70x el promedio), y una proyección a escala ($112.65 a 100.000 runs).
• Confirmamos, con números reales, un límite genuino de esta guía: el costo de un run se calcula sobre su history completo, disponible solo mientras el proceso que lo generó sigue vivo — reconstruirlo únicamente desde RUN_LOG.jsonl subestimaría el resultado, porque el texto final de la respuesta nunca quedó registrado como un evento propio.

Con esto se cierra el Módulo 3. Tienes observability/cost_calculator.py completo, y la evidencia ejecutada de que responde, con precisión, la pregunta que el Módulo 1 dejó abierta: "¿cuánto costó este run, y por qué?".

Siguiente módulo: Módulo 4 — Medir latencia con honestidad. Con el costo ya resuelto, este módulo completa la segunda mitad de la capa de "medir" del Módulo 1: cuánto tardó un run, por herramienta y en total, con la misma honestidad sobre qué se mide y qué se modela que ya viste en el costo — nunca con un cronómetro real dentro de un bloque "Qué esperar".

Recursos adicionales

1. Anthropic — Pricing — La fuente del precio de lista de claude-sonnet-5, la constante que este módulo fijó en la lección 04 y reusó, sin cambios, en cada lección posterior.
2. Anthropic — Token counting — El conteo real de tokens, la referencia contra la que toda la estimación de este módulo se confronta con honestidad.
3. Python — dataclasses — StepCost y CostReport, las estructuras que organizan cada resultado de este módulo.
4. Python — statistics — statistics.mean, la base de flag_expensive_runs; statistics.median y statistics.quantiles son el contenido central del Módulo 4.
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 reporte final de este mini-proyecto.