Costo por run, en centavos

Descripción

Las dos lecciones anteriores construyeron las piezas por separado: estimate_tokens (lección 03) convierte texto en una cantidad de tokens; el pricing fijo de claude-sonnet-5 (lección 04) convierte tokens en centavos. Esta lección las combina en estimate_cost_cents —la fórmula central de todo el módulo— y construye cost_for_run, la función que recorre el history de un run real de Reservo y calcula, con desglose por cada tool call, cuánto costó ese run completo.

Esta es la lección donde el trace_id del Módulo 2 y el costo de este módulo se encuentran por primera vez: cada CostReport que produce esta lección se identifica por el mismo trace_id determinista que ya viste en RUN_LOG.jsonl — la misma correlación de siempre, aplicada ahora al dinero.

Conexión con el módulo

Esta lección entrega la pieza central de observability/cost_calculator.py: estimate_cost_cents, StepCost, CostReport, y cost_for_run. Es, con precisión, el punto donde este módulo dejó de ser preparación y se convierte en la respuesta real a la pregunta que abrió la lección 01: "¿cuánto costó este run?".

estimate_cost_cents: la fórmula completa, ejecutada

Con las dos constantes de la lección 04 ya fijas, la fórmula combina tokens de entrada y de salida, cada uno con su propio precio, y redondea hacia abajo al final —la misma disciplina de aritmética entera de siempre—:

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):
    return len(text) // 4


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

Fíjate en el orden de las operaciones: primero se multiplica cada cantidad de tokens por su precio (dos números potencialmente grandes), después se suman ambos productos, y al final se divide entre 1_000_000. Invertir ese orden —dividir antes de sumar, por ejemplo— produciría resultados distintos e incorrectos por el redondeo hacia abajo de cada división parcial. Esta es la misma fórmula que el DISEÑO de esta guía fijó desde el principio; esta lección es, simplemente, la primera vez que se ejecuta de verdad sobre un run completo.

StepCost y CostReport: el desglose, estructurado

Un solo número de centavos por run es útil, pero no dice dónde se gastó ese costo. StepCost guarda el costo de un tool call individual; CostReport junta todos los StepCost de un run, más sus totales:

from dataclasses import dataclass, field


@dataclass
class StepCost:
    """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:
    """El costo estimado de un run completo, con su desglose por paso."""
    trace_id: str
    question: str
    steps: list = field(default_factory=list)
    input_tokens: int = 0
    output_tokens: int = 0
    cost_cents: int = 0

trace_id es, deliberadamente, el primer campo de CostReport — el mismo identificador determinista de run_logger.py (Módulo 2), nunca un id nuevo inventado para este módulo. Cualquier CostReport se puede cruzar, por ese campo, contra las líneas de RUN_LOG.jsonl del mismo run.

cost_for_run: recorriendo history, sin tocar run_reservo_agent

cost_for_run recibe el trace_id (de traced_run, Módulo 2), la pregunta, y el history que devuelve run_reservo_agent (agent-fundamentals M8, sin tocar su lógica) — la misma técnica de "envolver, no reconstruir" que ya usaste en el Módulo 1 con run_and_observe.

import itertools
import json


def cost_for_run(trace_id, question, history):
    """Recorre history (de run_reservo_agent, SIN tocarlo) y calcula el
    costo estimado del run completo, con desglose por tool call. Input =
    texto que el agente LEE (la pregunta + cada tool_result); output =
    texto que el agente PRODUCE (el input de cada tool_use + el texto
    final)."""
    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  # ya se contó como `question`, arriba
        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),
    )

Lee la clasificación con cuidado, porque es la misma que ya viste en el Módulo 1, lección 08 (run_and_observe), aplicada aquí con más detalle: el texto que el agente lee —la pregunta original y cada tool_result— cuenta como entrada; el texto que el agente produce —los argumentos de cada tool_use (serializados como el modelo los "escribió") y la respuesta final— cuenta como salida. pending, un diccionario indexado por el id del tool_use, es lo que permite emparejar cada tool_use con su tool_result correspondiente para construir un StepCost completo por cada paso, sin asumir que llegan en un orden particular.

Nota una decisión de diseño real: los totales del run (input_tokens, output_tokens de CostReport) se calculan sobre el texto concatenado completo (total_input_text, total_output_text), aplicando estimate_tokens una sola vez al final — no sumando los input_tokens/output_tokens de cada StepCost individual. La lección 03 ya adelantó por qué: sumar estimados ya redondeados pierde precisión frente a concatenar primero y redondear una sola vez. El desglose por paso (steps) sigue siendo útil para ver dónde se concentró el costo, pero el total oficial del run nunca se calcula sumando esas cifras ya redondeadas.

Ejemplo trabajado: el costo del run canónico de Ana

Corre el guion de siempre —list_rooms → get_quote (rechazado) → get_quote (corregido) → book_room—, esta vez envuelto en traced_run del Módulo 2, y calcula su costo inmediatamente después:

import logging
import reservo_agent as ra
import run_logger as rl

rl.logger.setLevel(logging.INFO)

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."}]},
]
with rl.traced_run("Reserva Focus pro 3h para Ana", 1) as trace_id:
    final, history = ra.run_reservo_agent("Reserva Focus pro 3h para Ana", script_a)

print()
print("=== costo del run", trace_id, "===")
report = cost_for_run(trace_id, "Reserva Focus pro 3h para Ana", history)
print("input_tokens :", report.input_tokens)
print("output_tokens:", report.output_tokens)
print("cost_cents   :", report.cost_cents)
print()
print("--- desglose por tool call ---")
for s in report.steps:
    print(f"  paso {s.step}: {s.tool:<12} in={s.input_tokens:>3} out={s.output_tokens:>3} cost_cents={s.cost_cents}")

Qué esperar:

{"seq": 1, "trace_id": "run-8487582448eb", "event": "run_started", "question": "Reserva Focus pro 3h para Ana", "tool_errors": 0, "error": ""}
{"seq": 2, "trace_id": "run-8487582448eb", "event": "tool_use", "step": 1, "tool": "list_rooms", "is_error": false, "content": ""}
{"seq": 4, "trace_id": "run-8487582448eb", "event": "tool_result", "step": 1, "tool": "list_rooms", "is_error": false, "content": "[{\"room\": \"Focus\", \"rate_cents\": 2500}, {\"room\": \"Studio\", \"rate_cents\": 4000}, {\"room\": \"Boardroom\", \"rate_cents\": 8000}]"}
{"seq": 6, "trace_id": "run-8487582448eb", "event": "tool_use", "step": 2, "tool": "get_quote", "is_error": false, "content": ""}
{"seq": 8, "trace_id": "run-8487582448eb", "event": "tool_result", "step": 2, "tool": "get_quote", "is_error": true, "content": "'tier'='premium' no está en enum ['basic', 'pro']"}
{"seq": 10, "trace_id": "run-8487582448eb", "event": "tool_use", "step": 3, "tool": "get_quote", "is_error": false, "content": ""}
{"seq": 12, "trace_id": "run-8487582448eb", "event": "tool_result", "step": 3, "tool": "get_quote", "is_error": false, "content": "{\"price_cents\": 6000}"}
{"seq": 14, "trace_id": "run-8487582448eb", "event": "tool_use", "step": 4, "tool": "book_room", "is_error": false, "content": ""}
{"seq": 16, "trace_id": "run-8487582448eb", "event": "tool_result", "step": 4, "tool": "book_room", "is_error": false, "content": "{\"booking_id\": 1, \"confirmed\": true}"}
{"seq": 18, "trace_id": "run-8487582448eb", "event": "run_finished", "question": "Reserva Focus pro 3h para Ana", "tool_errors": 1, "error": ""}

=== costo del run run-8487582448eb ===
input_tokens : 64
output_tokens: 56
cost_cents   : 0

--- desglose por tool call ---
  paso 1: list_rooms   in= 30 out=  0 cost_cents=0
  paso 2: get_quote    in= 12 out= 12 cost_cents=0
  paso 3: get_quote    in=  5 out= 11 cost_cents=0
  paso 4: book_room    in=  9 out= 15 cost_cents=0

Lee el desglose con atención, porque revela algo que no es obvio antes de calcularlo: el paso 2 —el get_quote con el tier inválido, rechazado por check_input_v2— costó tokens igual que cualquier otro paso (12 de entrada, 12 de salida). El error no fue gratis. El modelo (concepto) gastó tokens de salida proponiendo tier="premium", y el sistema gastó tokens de entrada leyendo el mensaje de error que le llegó de vuelta. cost_cents del run completo sigue siendo 0 —la respuesta honesta para un run de este tamaño, la misma que ya viste en el Módulo 1—, pero el desglose por paso ya te dice, con precisión, que un cuarto del costo total de este run se gastó en un intento que terminó rechazado.

Errores comunes

1. Sumar los cost_cents de cada StepCost para obtener el costo total del run. Como ya advirtió la lección 03, sumar valores ya redondeados por separado pierde precisión frente a concatenar el texto primero y aplicar estimate_tokens/estimate_cost_cents una sola vez. CostReport.cost_cents se calcula sobre los totales del run, nunca sumando steps.

2. Pensar que un tool_result con is_error: True no cuesta nada. El ejemplo trabajado lo desmiente con números: el paso 2, el intento rechazado, costó exactamente lo mismo en tokens que un paso exitoso de tamaño similar. Un agente que se equivoca y se auto-corrige (M7 de agent-fundamentals) paga el costo de ambos intentos, no solo del que funcionó.

3. Olvidar inicializar total_input_text con question. El primer turno de history es {"role": "user", "content": question} —un string, no una lista de bloques—, así que el for turn in history de cost_for_run lo salta explícitamente (if isinstance(content, str): continue) porque ya se contó al inicializar total_input_text = question. Olvidar esa inicialización dejaría la pregunta original fuera del costo del run.

4. Confundir entry["args_text"] (los argumentos del tool_use, ya serializados) con result_text (el content del tool_result). El primero cuenta como salida (lo que el modelo generó al pedir la tool); el segundo cuenta como entrada (lo que el sistema le devuelve para que el modelo lo lea). Invertir esta clasificación invertiría, también, qué parte del costo se atribuye a cada dirección.

5. Ejecutar cost_for_run sobre un history de un run que lanzó RuntimeError. Como ya viste en el Módulo 1 con run_and_observe, si run_reservo_agent lanza una excepción, nunca llega a return, y no hay ningún history que capturar desde afuera de la llamada. cost_for_run asume, como precondición, que history es el resultado de un run que sí terminó —bien o mal, pero terminó—; no está diseñada para un run que ni siquiera llegó a producir un history completo.

Ejercicios

Ejercicio 1: Calcula el costo del run de Sofía (Fácil)

Corre cost_for_run sobre el guion de Sofía —list_rooms → get_quote (Boardroom, pro, 1h) → book_room, sin ningún error—, envuelto en traced_run con sequence_number=2. Confirma input_tokens, output_tokens y cost_cents.

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."}]},
]
with rl.traced_run("Reserva Boardroom pro 1h para Sofía", 2) as trace_id_b:
    final_b, history_b = ra.run_reservo_agent("Reserva Boardroom pro 1h para Sofía", script_b)

report_b = cost_for_run(trace_id_b, "Reserva Boardroom pro 1h para Sofía", history_b)
print("input_tokens :", report_b.input_tokens)
print("output_tokens:", report_b.output_tokens)
print("cost_cents   :", report_b.cost_cents)

input_tokens : 53
output_tokens: 49
cost_cents   : 0

Ejercicio 2: Encuentra el paso más costoso del run de Ana (Medio)

Usando el report del ejemplo trabajado (el run de Ana), encuentra el StepCost con el mayor total de tokens (input_tokens + output_tokens), sin asumir de antemano cuál es.

busiest = max(report.steps, key=lambda s: s.input_tokens + s.output_tokens)
print(f"paso más costoso: paso {busiest.step} ({busiest.tool}), "
      f"{busiest.input_tokens + busiest.output_tokens} tokens totales")

paso más costoso: paso 1 (list_rooms), 30 tokens totales

Ejercicio 3: Cuantifica el costo del error y su corrección (Difícil)

Corre cost_for_run sobre una versión sin ningún error de la tarea de Ana —list_rooms → get_quote (tier="pro" directo, sin el intento rechazado) → book_room—. Compara sus input_tokens + output_tokens contra los del run del ejemplo trabajado (con el error), y confirma que la diferencia coincide, exactamente, con los tokens del paso 2 del run con error.

script_clean = [
    {"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": "pro", "hours": 3}}]},
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_03", "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."}]},
]
final_clean, history_clean = ra.run_reservo_agent("Reserva Focus pro 3h para Ana", script_clean)
report_clean = cost_for_run("run-clean", "Reserva Focus pro 3h para Ana", history_clean)

total_con_error = report.input_tokens + report.output_tokens          # del ejemplo trabajado
total_sin_error = report_clean.input_tokens + report_clean.output_tokens
diferencia = total_con_error - total_sin_error

paso_2 = next(s for s in report.steps if s.step == 2)
tokens_paso_2 = paso_2.input_tokens + paso_2.output_tokens

print("tokens CON error (run original)   :", total_con_error)
print("tokens SIN error (run limpio)     :", total_sin_error)
print("diferencia                        :", diferencia)
print("tokens del paso 2 (el rechazado)  :", tokens_paso_2)
print("coinciden exactamente             :", diferencia == tokens_paso_2)

tokens CON error (run original)   : 120
tokens SIN error (run limpio)     : 96
diferencia                        : 24
tokens del paso 2 (el rechazado)  : 24
coinciden exactamente             : True

Resumen y siguiente paso

• Construimos estimate_cost_cents, StepCost, CostReport, y cost_for_run — el corazón de observability/cost_calculator.py, combinando estimate_tokens (L03) con el pricing fijo (L04).
• Ejecutamos cost_for_run sobre el run canónico de Ana, envuelto en traced_run del Módulo 2: 64 tokens de entrada, 56 de salida, 0 centavos —con un desglose de cuatro pasos que muestra, con precisión, dónde se concentró cada token—.
• Confirmamos, ejecutado, que un paso rechazado por validación (is_error: True) cuesta tokens igual que uno exitoso — el error no es gratis, y el Ejercicio 3 lo cuantificó con exactitud: 24 tokens, el costo íntegro de un intento fallido y su corrección.
• Cada CostReport se identifica por el mismo trace_id determinista de run_logger.py — la correlación del Módulo 2, aplicada ahora al costo.

Siguiente lección: 06 — Escalando el costo a miles de runs. Con el costo de un run individual ya calculado —y frecuentemente 0 centavos, la respuesta honesta para runs de este tamaño—, escalamos esa cifra a lotes de 1.000, 10.000 y 100.000 runs, y confirmamos por qué sumar tokens antes de redondear es la única forma correcta de hacerlo.

Recursos adicionales

1. Anthropic — Token counting — El conteo real de tokens de entrada y salida, la referencia contra la que esta lección confronta su propia estimación.
2. Python — dataclasses — StepCost y CostReport, y field(default_factory=list) para evitar el error clásico de un valor por defecto mutable compartido.
3. Python — itertools.count — El contador usado para numerar cada paso dentro de cost_for_run, la misma técnica que ya viste en run_logger.py.
4. Anthropic — Tool use (function calling) overview — El protocolo tool_use/tool_result que cost_for_run recorre sin alterarlo, exactamente como lo hizo run_and_observe en el Módulo 1.
5. Python 3.14 — What's New — La versión con la que se ejecutó cada línea de código de esta lección.