Módulo 12: LangSmith y Producción

Token Usage Tracking y Prompt Caching

Descripción de la cápsula

Los tokens cuestan dinero. Cada vez que tu agente llama a un modelo, estás pagando — por cada token de input y cada token de output. En un prototipo local esto es irrelevante: una ejecución del Research Assistant cuesta centavos. Pero en producción, con cientos o miles de usuarios diarios, esos centavos se convierten en facturas de miles de dólares mensuales. Sin tracking, no sabes cuánto gasta cada usuario, qué operaciones son las más caras, ni dónde optimizar.

Token tracking no es una herramienta técnica — es una herramienta de negocio. "Esta investigación costó $0.12" es información que afecta decisiones de producto: ¿puedes ofrecer este feature en el plan gratuito? ¿Necesitas un límite de uso? ¿Vale la pena optimizar el prompt del analyst que consume el 60% del presupuesto? Sin estos números, estás tomando decisiones a ciegas.

En las cápsulas anteriores aprendiste a hacer tracing con LangSmith (cápsula 02), debugging visual de ejecuciones (cápsula 03), y evaluation automatizada con datasets (cápsula 04). Ahora agregas la dimensión financiera: cuánto cuesta cada trace, cada evaluación, cada ejecución del agente. Y con prompt caching, puedes reducir esos costos hasta un 90% en patrones repetitivos.

Token tracking como herramienta de negocio

Antes de ver código, entiende el impacto financiero. Estos son los precios de modelos comunes (al momento de escribir — verifica precios actualizados):

Modelo	Input (por 1M tokens)	Output (por 1M tokens)	Contexto
GPT-4.1	$2.00	$8.00	1M tokens
GPT-4.1-mini	$0.40	$1.60	1M tokens
GPT-4.1-nano	$0.10	$0.40	1M tokens
Claude Sonnet 4	$3.00	$15.00	200K tokens
Claude Haiku 3.5	$0.80	$4.00	200K tokens

Ahora, haz la cuenta para tu Research Assistant:

Una investigación típica:
- Descomposición (1 llamada): ~500 input + ~200 output = 700 tokens
- Búsqueda (4 llamadas):      ~2000 input + ~800 output = 2800 tokens
- Análisis (1 llamada):       ~3000 input + ~1000 output = 4000 tokens
- Síntesis (1 llamada):       ~2000 input + ~500 output = 2500 tokens
                                                    Total: ~10,000 tokens

Con GPT-4.1:
- Input: 7,500 tokens × $2.00/1M = $0.015
- Output: 2,500 tokens × $8.00/1M = $0.020
- Total por investigación: ~$0.035

Con GPT-4.1-mini:
- Input: 7,500 tokens × $0.40/1M = $0.003
- Output: 2,500 tokens × $1.60/1M = $0.004
- Total por investigación: ~$0.007

Parece poco. Ahora escala:

1,000 usuarios/día × 3 investigaciones/usuario = 3,000 investigaciones/día

Con GPT-4.1:    3,000 × $0.035 = $105/día = $3,150/mes
Con GPT-4.1-mini: 3,000 × $0.007 = $21/día = $630/mes

¿Es aceptable para tu modelo de negocio?
¿Cuánto cobras al usuario?
¿Cuál es tu margen?

Estas son las preguntas que token tracking te permite responder con datos, no con intuición.

UsageMetadata: tokens por llamada

LangChain incluye metadata de uso en cada respuesta del modelo. El campo usage_metadata en el AIMessage contiene el desglose de tokens.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-4.1-mini")

response = model.invoke("¿Qué es LangSmith?")

print(f"Respuesta: {response.content[:80]}...")
print(f"\nUsage metadata:")
print(f"  Input tokens:  {response.usage_metadata['input_tokens']}")
print(f"  Output tokens: {response.usage_metadata['output_tokens']}")
print(f"  Total tokens:  {response.usage_metadata['total_tokens']}")
# Output esperado:
# Respuesta: LangSmith es una plataforma de observabilidad y evaluación para aplicaciones ...
#
# Usage metadata:
#   Input tokens:  13
#   Output tokens: 95
#   Total tokens:  108

Cada AIMessage incluye usage_metadata automáticamente. No necesitas configuración extra — solo acceder al campo.

Calculando costo por llamada

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

PRICING = {
    "gpt-4.1": {"input": 2.00, "output": 8.00},
    "gpt-4.1-mini": {"input": 0.40, "output": 1.60},
    "gpt-4.1-nano": {"input": 0.10, "output": 0.40},
}

def calculate_cost(usage_metadata: dict, model_name: str) -> float:
    """Calcula el costo en dólares de una llamada al modelo."""
    pricing = PRICING.get(model_name, PRICING["gpt-4.1-mini"])
    input_cost = (usage_metadata["input_tokens"] / 1_000_000) * pricing["input"]
    output_cost = (usage_metadata["output_tokens"] / 1_000_000) * pricing["output"]
    return input_cost + output_cost

model = init_chat_model("openai:gpt-4.1-mini")

prompts = [
    "Di 'hola' en una palabra.",
    "Explica qué es machine learning en 3 oraciones.",
    "Escribe un análisis detallado de las ventajas y desventajas de microservicios vs monolitos, incluyendo consideraciones de escalabilidad, mantenimiento, testing, deployment, y costos operativos.",
]

total_cost = 0.0
for prompt in prompts:
    response = model.invoke(prompt)
    cost = calculate_cost(response.usage_metadata, "gpt-4.1-mini")
    total_cost += cost
    print(f"Prompt: {prompt[:50]}...")
    print(f"  Tokens: {response.usage_metadata['input_tokens']} in / {response.usage_metadata['output_tokens']} out")
    print(f"  Costo:  ${cost:.6f}")
    print()

print(f"Costo total de 3 llamadas: ${total_cost:.6f}")
# Output esperado:
# Prompt: Di 'hola' en una palabra....
#   Tokens: 14 in / 4 out
#   Costo:  $0.000012
#
# Prompt: Explica qué es machine learning en 3 oraciones....
#   Tokens: 16 in / 68 out
#   Costo:  $0.000115
#
# Prompt: Escribe un análisis detallado de las ventajas y d...
#   Tokens: 40 in / 350 out
#   Costo:  $0.000576
#
# Costo total de 3 llamadas: $0.000703

get_openai_callback: tracking agregado por bloque

Para trackear tokens de múltiples llamadas dentro de un bloque de código, LangChain ofrece get_openai_callback como context manager.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_community.callbacks import get_openai_callback

model = init_chat_model("openai:gpt-4.1-mini")

with get_openai_callback() as cb:
    response1 = model.invoke("¿Qué es Python?")
    response2 = model.invoke("¿Qué es JavaScript?")
    response3 = model.invoke("Compara Python y JavaScript en 2 oraciones.")

print(f"Llamadas totales: {cb.successful_requests}")
print(f"Tokens totales:   {cb.total_tokens}")
print(f"  Input:          {cb.prompt_tokens}")
print(f"  Output:         {cb.completion_tokens}")
print(f"Costo total:      ${cb.total_cost:.6f}")
# Output esperado:
# Llamadas totales: 3
# Tokens totales:   320
#   Input:          45
#   Output:         275
# Costo total:      $0.000458

El callback acumula tokens y costo de todas las llamadas dentro del bloque with. Es útil para medir el costo de una operación completa que involucra múltiples llamadas.

Tracking por operación en un agente

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_community.callbacks import get_openai_callback

model = init_chat_model("openai:gpt-4.1-mini")

@tool
def web_search(query: str) -> str:
    """Busca información en la web."""
    return f"Resultado: {query} es un tema relevante en tecnología moderna."

@tool
def calculate(expression: str) -> str:
    """Evalúa una expresión matemática."""
    try:
        return str(eval(expression))
    except Exception:
        return "Error en la expresión"

agent = create_agent(model, [web_search, calculate])

with get_openai_callback() as cb:
    result = agent.invoke(
        {"messages": [("user", "¿Cuánto es 15% de 250? Busca qué se puede comprar con eso.")]}
    )

print(f"Respuesta: {result['messages'][-1].content[:100]}...")
print(f"\nCosto de la ejecución completa:")
print(f"  Llamadas al modelo: {cb.successful_requests}")
print(f"  Tokens totales:     {cb.total_tokens}")
print(f"  Costo:              ${cb.total_cost:.6f}")

COST_PER_RESEARCH = cb.total_cost
print(f"\nProyección mensual (1000 usuarios/día, 3 ejecuciones/usuario):")
print(f"  Diario:  ${COST_PER_RESEARCH * 3000:.2f}")
print(f"  Mensual: ${COST_PER_RESEARCH * 3000 * 30:.2f}")
# Output esperado:
# Respuesta: El 15% de 250 es 37.5. Con $37.50 se pueden comprar diversas cosas dependien...
#
# Costo de la ejecución completa:
#   Llamadas al modelo: 3
#   Tokens totales:     485
#   Costo:              $0.000680
#
# Proyección mensual (1000 usuarios/día, 3 ejecuciones/usuario):
#   Diario:  $2.04
#   Mensual: $61.20

UsageMetadataCallbackHandler: tracking granular

Para tracking más detallado — especialmente cuando necesitas el desglose por llamada individual dentro de una ejecución — usa UsageMetadataCallbackHandler.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.callbacks import UsageMetadataCallbackHandler

model = init_chat_model("openai:gpt-4.1-mini")

handler = UsageMetadataCallbackHandler()

response = model.invoke(
    "Explica qué es observabilidad en sistemas de AI.",
    config={"callbacks": [handler]},
)

print(f"Respuesta: {response.content[:80]}...")
print(f"\nUsage acumulado en el handler:")
print(f"  Total usage records: {len(handler.usage_metadata)}")
for i, usage in enumerate(handler.usage_metadata):
    print(f"  Record {i}: {usage}")
# Output esperado:
# Respuesta: La observabilidad en sistemas de AI se refiere a la capacidad de entender el ...
#
# Usage acumulado en el handler:
#   Total usage records: 1
#   Record 0: {'input_tokens': 16, 'output_tokens': 120, 'total_tokens': 136}

Tracking a lo largo de múltiples llamadas

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.callbacks import UsageMetadataCallbackHandler

PRICING = {
    "gpt-4.1-mini": {"input": 0.40, "output": 1.60},
}

model = init_chat_model("openai:gpt-4.1-mini")
handler = UsageMetadataCallbackHandler()
config = {"callbacks": [handler]}

operations = [
    ("decompose", "Descompone 'AI en educación' en 3 sub-preguntas. Responde solo las preguntas."),
    ("search", "Resume brevemente qué es AI en educación."),
    ("analyze", "Basándote en que AI personaliza el aprendizaje y automatiza evaluación, identifica 2 hallazgos clave. Responde en 2 oraciones."),
    ("write", "Escribe un resumen ejecutivo de 2 oraciones sobre AI en educación."),
]

costs_by_operation = {}
prev_count = 0

for op_name, prompt in operations:
    model.invoke(prompt, config=config)
    
    current_records = handler.usage_metadata[prev_count:]
    total_input = sum(u.get("input_tokens", 0) for u in current_records)
    total_output = sum(u.get("output_tokens", 0) for u in current_records)
    
    cost = (total_input / 1_000_000) * PRICING["gpt-4.1-mini"]["input"] + \
           (total_output / 1_000_000) * PRICING["gpt-4.1-mini"]["output"]
    
    costs_by_operation[op_name] = {
        "input_tokens": total_input,
        "output_tokens": total_output,
        "cost": cost,
    }
    prev_count = len(handler.usage_metadata)

print("Desglose de costos por operación:")
print(f"{'Operación':<12} {'Input':>8} {'Output':>8} {'Costo':>12}")
print("─" * 44)
total = 0.0
for op, data in costs_by_operation.items():
    print(f"{op:<12} {data['input_tokens']:>8} {data['output_tokens']:>8} ${data['cost']:>10.6f}")
    total += data["cost"]
print("─" * 44)
print(f"{'TOTAL':<12} {'':>8} {'':>8} ${total:>10.6f}")
# Output esperado:
# Desglose de costos por operación:
# Operación      Input   Output        Costo
# ────────────────────────────────────────────
# decompose         18       45   $0.000079
# search            12       80   $0.000133
# analyze           42      60    $0.000113
# write             20       50   $0.000088
# ────────────────────────────────────────────
# TOTAL                              $0.000413

Clase CostTracker: tracking de producción

En un sistema real, necesitas un tracker reutilizable que acumule costos, los desglose por operación, y genere reportes.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from dataclasses import dataclass, field
from datetime import datetime

PRICING = {
    "gpt-4.1": {"input": 2.00, "output": 8.00},
    "gpt-4.1-mini": {"input": 0.40, "output": 1.60},
    "gpt-4.1-nano": {"input": 0.10, "output": 0.40},
}


@dataclass
class CostEntry:
    operation: str
    model: str
    input_tokens: int
    output_tokens: int
    cost_usd: float
    timestamp: str = field(default_factory=lambda: datetime.now().isoformat())


class CostTracker:
    def __init__(self, model_name: str = "gpt-4.1-mini"):
        self.model_name = model_name
        self.entries: list[CostEntry] = []

    def track(self, operation: str, usage_metadata: dict) -> CostEntry:
        """Registra el costo de una operación."""
        pricing = PRICING.get(self.model_name, PRICING["gpt-4.1-mini"])
        input_tokens = usage_metadata.get("input_tokens", 0)
        output_tokens = usage_metadata.get("output_tokens", 0)

        cost = (input_tokens / 1_000_000) * pricing["input"] + \
               (output_tokens / 1_000_000) * pricing["output"]

        entry = CostEntry(
            operation=operation,
            model=self.model_name,
            input_tokens=input_tokens,
            output_tokens=output_tokens,
            cost_usd=cost,
        )
        self.entries.append(entry)
        return entry

    @property
    def total_cost(self) -> float:
        return sum(e.cost_usd for e in self.entries)

    @property
    def total_tokens(self) -> int:
        return sum(e.input_tokens + e.output_tokens for e in self.entries)

    def cost_by_operation(self) -> dict[str, float]:
        """Costo agrupado por tipo de operación."""
        costs: dict[str, float] = {}
        for entry in self.entries:
            costs[entry.operation] = costs.get(entry.operation, 0) + entry.cost_usd
        return costs

    def report(self) -> str:
        """Genera un reporte legible de costos."""
        lines = []
        lines.append("╔══════════════════════════════════════════════╗")
        lines.append("║          COST REPORT                        ║")
        lines.append("╚══════════════════════════════════════════════╝")
        lines.append(f"  Modelo:         {self.model_name}")
        lines.append(f"  Operaciones:    {len(self.entries)}")
        lines.append(f"  Tokens totales: {self.total_tokens:,}")
        lines.append(f"  Costo total:    ${self.total_cost:.6f}")
        lines.append("")
        lines.append("  Desglose por operación:")
        for op, cost in self.cost_by_operation().items():
            pct = (cost / self.total_cost * 100) if self.total_cost > 0 else 0
            lines.append(f"    {op:<15} ${cost:.6f}  ({pct:.1f}%)")

        daily_projection = self.total_cost * 3000
        monthly_projection = daily_projection * 30
        lines.append("")
        lines.append("  Proyección (3000 ejecuciones/día):")
        lines.append(f"    Diario:  ${daily_projection:.2f}")
        lines.append(f"    Mensual: ${monthly_projection:.2f}")
        return "\n".join(lines)


tracker = CostTracker("gpt-4.1-mini")
model = init_chat_model("openai:gpt-4.1-mini")

operations = [
    ("decompose", "Lista 3 sub-preguntas sobre 'AI en salud'. Solo las preguntas."),
    ("search", "¿Qué es AI aplicada a diagnóstico médico? Responde en 2 oraciones."),
    ("search", "¿Qué avances hay en AI para descubrimiento de fármacos? Responde en 2 oraciones."),
    ("analyze", "Dado que AI mejora diagnósticos y acelera descubrimiento de fármacos, identifica 2 tendencias. Responde en 2 oraciones."),
    ("write", "Escribe un resumen ejecutivo de 3 oraciones sobre AI en salud."),
]

for op_name, prompt in operations:
    response = model.invoke(prompt)
    tracker.track(op_name, response.usage_metadata)

print(tracker.report())
# Output esperado:
# ╔══════════════════════════════════════════════╗
# ║          COST REPORT                        ║
# ╚══════════════════════════════════════════════╝
#   Modelo:         gpt-4.1-mini
#   Operaciones:    5
#   Tokens totales: 520
#   Costo total:    $0.000450
#
#   Desglose por operación:
#     decompose       $0.000075  (16.7%)
#     search          $0.000180  (40.0%)
#     analyze         $0.000110  (24.4%)
#     write           $0.000085  (18.9%)
#
#   Proyección (3000 ejecuciones/día):
#     Diario:  $1.35
#     Mensual: $40.50

El CostTracker te responde preguntas de negocio directamente: "¿Qué operación consume más presupuesto?" (search, 40%). "¿Puedo ofrecer esto gratis?" (a $40.50/mes, quizás sí para un plan gratuito limitado). "¿Dónde optimizo primero?" (search — es el 40% del costo).

Costo por modelo: eligiendo el modelo correcto

No todas las operaciones necesitan el mismo modelo. El análisis necesita razonamiento — usa GPT-4.1. La búsqueda y escritura son más simples — usa GPT-4.1-mini o nano. Este pattern conecta con el model routing del Módulo 4.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

PRICING = {
    "gpt-4.1": {"input": 2.00, "output": 8.00},
    "gpt-4.1-mini": {"input": 0.40, "output": 1.60},
    "gpt-4.1-nano": {"input": 0.10, "output": 0.40},
}

MODEL_ASSIGNMENT = {
    "decompose": "gpt-4.1-nano",
    "search": "gpt-4.1-nano",
    "analyze": "gpt-4.1",
    "write": "gpt-4.1-mini",
}

models = {
    "gpt-4.1": init_chat_model("openai:gpt-4.1"),
    "gpt-4.1-mini": init_chat_model("openai:gpt-4.1-mini"),
    "gpt-4.1-nano": init_chat_model("openai:gpt-4.1-nano"),
}

def calculate_cost(usage: dict, model_name: str) -> float:
    pricing = PRICING[model_name]
    return (usage["input_tokens"] / 1_000_000) * pricing["input"] + \
           (usage["output_tokens"] / 1_000_000) * pricing["output"]

operations = [
    ("decompose", "Lista 3 sub-preguntas sobre 'AI en producción'. Solo las preguntas."),
    ("search", "¿Qué es observabilidad en AI? Responde en 1 oración."),
    ("analyze", "Analiza por qué la observabilidad es crítica en producción de AI. 2 razones."),
    ("write", "Escribe 2 oraciones resumiendo la importancia de observabilidad en AI."),
]

total_optimized = 0.0
total_if_all_gpt4 = 0.0

print(f"{'Operación':<12} {'Modelo':<16} {'Tokens':>8} {'Costo':>12} {'Si GPT-4.1':>12}")
print("─" * 64)

for op_name, prompt in operations:
    model_name = MODEL_ASSIGNMENT[op_name]
    model = models[model_name]

    response = model.invoke(prompt)
    usage = response.usage_metadata

    cost = calculate_cost(usage, model_name)
    cost_gpt4 = calculate_cost(usage, "gpt-4.1")

    total_optimized += cost
    total_if_all_gpt4 += cost_gpt4

    total_tokens = usage["input_tokens"] + usage["output_tokens"]
    print(f"{op_name:<12} {model_name:<16} {total_tokens:>8} ${cost:>10.6f} ${cost_gpt4:>10.6f}")

print("─" * 64)
savings = total_if_all_gpt4 - total_optimized
savings_pct = (savings / total_if_all_gpt4 * 100) if total_if_all_gpt4 > 0 else 0
print(f"{'TOTAL':<12} {'optimizado':<16} {'':>8} ${total_optimized:>10.6f} ${total_if_all_gpt4:>10.6f}")
print(f"\nAhorro: ${savings:.6f} ({savings_pct:.1f}%)")
print(f"Proyección mensual (3000 ejecuciones/día):")
print(f"  Optimizado:     ${total_optimized * 3000 * 30:.2f}/mes")
print(f"  Todo GPT-4.1:   ${total_if_all_gpt4 * 3000 * 30:.2f}/mes")
print(f"  Ahorro mensual: ${savings * 3000 * 30:.2f}/mes")
# Output esperado:
# Operación    Modelo             Tokens        Costo    Si GPT-4.1
# ────────────────────────────────────────────────────────────────────
# decompose    gpt-4.1-nano           55   $0.000008   $0.000080
# search       gpt-4.1-nano           40   $0.000006   $0.000058
# analyze      gpt-4.1               180   $0.000456   $0.000456
# write        gpt-4.1-mini           90   $0.000060   $0.000300
# ────────────────────────────────────────────────────────────────────
# TOTAL        optimizado                  $0.000530   $0.000894
#
# Ahorro: $0.000364 (40.7%)
# Proyección mensual (3000 ejecuciones/día):
#   Optimizado:     $47.70/mes
#   Todo GPT-4.1:   $80.46/mes
#   Ahorro mensual: $32.76/mes

Prompt caching: reducir costos en patrones repetitivos

Muchos proveedores ofrecen prompt caching: si envías el mismo prefijo de prompt repetidamente, los tokens cacheados se cobran a un precio reducido (típicamente 50-90% de descuento). Esto es especialmente relevante para system prompts largos que se repiten en cada llamada.

Cómo funciona el caching

Sin cache:
  Llamada 1: [system prompt: 2000 tokens] + [user: 50 tokens] → cobra 2050 tokens input
  Llamada 2: [system prompt: 2000 tokens] + [user: 60 tokens] → cobra 2060 tokens input
  Llamada 3: [system prompt: 2000 tokens] + [user: 45 tokens] → cobra 2045 tokens input
  Total: 6155 tokens a precio completo

Con cache:
  Llamada 1: [system prompt: 2000 tokens] + [user: 50 tokens] → cobra 2050 tokens (cache miss)
  Llamada 2: [system prompt: 2000 CACHED] + [user: 60 tokens] → cobra 60 tokens + 2000 cached
  Llamada 3: [system prompt: 2000 CACHED] + [user: 45 tokens] → cobra 45 tokens + 2000 cached
  Total: 2155 tokens a precio completo + 4000 tokens a precio reducido

Detección de tokens cacheados en la respuesta

OpenAI y Anthropic incluyen información de caching en usage_metadata cuando aplica. Puedes detectar si tus prompts se están beneficiando del cache.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.messages import SystemMessage, HumanMessage

LONG_SYSTEM_PROMPT = """Eres un analista de investigación senior especializado en tecnología.

Tu metodología de análisis incluye:
1. Identificar las fuentes primarias y verificar su credibilidad
2. Detectar patrones y tendencias en los datos
3. Evaluar contradicciones entre fuentes
4. Cuantificar la confianza en cada hallazgo
5. Generar conclusiones accionables

Reglas de formato:
- Responde siempre en español
- Usa bullet points para hallazgos
- Incluye un nivel de confianza (alto/medio/bajo) por hallazgo
- Limita tu respuesta a 3-5 hallazgos principales
- Cada hallazgo debe tener evidencia específica

Tu objetivo es producir análisis que un ejecutivo pueda leer en 2 minutos
y tomar decisiones informadas. Prioriza claridad sobre exhaustividad."""

model = init_chat_model("openai:gpt-4.1-mini")

questions = [
    "Analiza el impacto de LLMs en desarrollo de software.",
    "Analiza las tendencias en AI generativa para 2026.",
    "Analiza el estado actual de agentes autónomos de AI.",
]

for i, question in enumerate(questions):
    messages = [
        SystemMessage(content=LONG_SYSTEM_PROMPT),
        HumanMessage(content=question),
    ]
    response = model.invoke(messages)
    usage = response.usage_metadata

    cached = usage.get("input_token_details", {}).get("cached", 0)
    cache_read = usage.get("cache_read_input_tokens", 0)
    cached_tokens = cached or cache_read

    print(f"Llamada {i+1}: {question[:50]}...")
    print(f"  Input: {usage['input_tokens']} | Output: {usage['output_tokens']} | Cached: {cached_tokens}")
# Output esperado:
# Llamada 1: Analiza el impacto de LLMs en desarrollo de softw...
#   Input: 210 | Output: 250 | Cached: 0
# Llamada 2: Analiza las tendencias en AI generativa para 2026...
#   Input: 212 | Output: 230 | Cached: 192
# Llamada 3: Analiza el estado actual de agentes autónomos de ...
#   Input: 214 | Output: 240 | Cached: 192

El cache se activa automáticamente cuando el prefijo coincide. No necesitas hacer nada especial con OpenAI — el sistema detecta prefijos comunes y los cachea.

Anthropic cache_control explícito

Con modelos de Anthropic, puedes ser explícito sobre qué partes del prompt quieres cachear usando cache_control.

from dotenv import load_dotenv
load_dotenv()

from langchain_anthropic import ChatAnthropic
from langchain_core.messages import SystemMessage, HumanMessage

model = ChatAnthropic(
    model="claude-sonnet-4-20250514",
    max_tokens=500,
)

system_message = SystemMessage(
    content="""Eres un analista de investigación senior. Siempre respondes en español.
Limita tu respuesta a 3 hallazgos principales con nivel de confianza.
Cada hallazgo debe tener evidencia específica de las fuentes proporcionadas.
Prioriza claridad y brevedad. El ejecutivo lee tu reporte en 2 minutos.""",
    additional_kwargs={"cache_control": {"type": "ephemeral"}},
)

questions = [
    "¿Cuáles son las principales tendencias en AI para 2026?",
    "¿Qué impacto tiene AI en el mercado laboral?",
]

for question in questions:
    response = model.invoke([system_message, HumanMessage(content=question)])
    usage = response.usage_metadata

    cache_creation = usage.get("input_token_details", {}).get("cache_creation", 0)
    cache_read = usage.get("input_token_details", {}).get("cached", 0)

    print(f"Q: {question[:50]}...")
    print(f"  Input: {usage['input_tokens']} | Cache creation: {cache_creation} | Cache read: {cache_read}")
# Output esperado:
# Q: ¿Cuáles son las principales tendencias en AI par...
#   Input: 95 | Cache creation: 80 | Cache read: 0
# Q: ¿Qué impacto tiene AI en el mercado laboral?...
#   Input: 95 | Cache creation: 0 | Cache read: 80

Token budgets: límites por usuario

En producción, necesitas establecer límites de tokens por usuario para prevenir costos descontrolados. Un usuario que ejecuta 100 investigaciones en una hora puede agotar tu presupuesto diario.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from dataclasses import dataclass, field
from datetime import datetime, timedelta

PRICING = {
    "gpt-4.1-mini": {"input": 0.40, "output": 1.60},
}


@dataclass
class UserBudget:
    user_id: str
    daily_limit_usd: float
    monthly_limit_usd: float
    spent_today_usd: float = 0.0
    spent_this_month_usd: float = 0.0
    last_reset_daily: str = field(default_factory=lambda: datetime.now().date().isoformat())
    last_reset_monthly: str = field(default_factory=lambda: datetime.now().strftime("%Y-%m"))

    def can_spend(self, estimated_cost: float) -> tuple[bool, str]:
        """Verifica si el usuario puede gastar el costo estimado."""
        if self.spent_today_usd + estimated_cost > self.daily_limit_usd:
            return False, f"Límite diario alcanzado (${self.spent_today_usd:.4f}/${self.daily_limit_usd:.4f})"
        if self.spent_this_month_usd + estimated_cost > self.monthly_limit_usd:
            return False, f"Límite mensual alcanzado (${self.spent_this_month_usd:.2f}/${self.monthly_limit_usd:.2f})"
        return True, "OK"

    def record_spend(self, cost: float):
        """Registra un gasto."""
        self.spent_today_usd += cost
        self.spent_this_month_usd += cost

    def alert_level(self) -> str:
        """Retorna el nivel de alerta basado en el gasto."""
        daily_pct = (self.spent_today_usd / self.daily_limit_usd * 100) if self.daily_limit_usd > 0 else 0
        if daily_pct >= 100:
            return "BLOCKED"
        if daily_pct >= 80:
            return "WARNING"
        if daily_pct >= 50:
            return "NOTICE"
        return "OK"


TIER_LIMITS = {
    "free": {"daily": 0.05, "monthly": 1.00},
    "pro": {"daily": 0.50, "monthly": 10.00},
    "enterprise": {"daily": 5.00, "monthly": 100.00},
}


def create_user_budget(user_id: str, tier: str) -> UserBudget:
    limits = TIER_LIMITS[tier]
    return UserBudget(
        user_id=user_id,
        daily_limit_usd=limits["daily"],
        monthly_limit_usd=limits["monthly"],
    )


model = init_chat_model("openai:gpt-4.1-mini")

free_user = create_user_budget("user-free-001", "free")
pro_user = create_user_budget("user-pro-001", "pro")

users = [free_user, pro_user]

for user in users:
    print(f"\n{'='*50}")
    print(f"Usuario: {user.user_id} (límite diario: ${user.daily_limit_usd})")
    print(f"{'='*50}")

    for i in range(5):
        estimated_cost = 0.01
        can_spend, reason = user.can_spend(estimated_cost)

        if not can_spend:
            print(f"  Intento {i+1}: BLOQUEADO — {reason}")
            continue

        response = model.invoke(f"Pregunta {i+1}: ¿Qué es observabilidad?")
        usage = response.usage_metadata
        actual_cost = (usage["input_tokens"] / 1_000_000) * PRICING["gpt-4.1-mini"]["input"] + \
                      (usage["output_tokens"] / 1_000_000) * PRICING["gpt-4.1-mini"]["output"]

        user.record_spend(actual_cost)
        alert = user.alert_level()
        print(f"  Intento {i+1}: ${actual_cost:.6f} | Total: ${user.spent_today_usd:.6f} | Alert: {alert}")
# Output esperado:
# ==================================================
# Usuario: user-free-001 (límite diario: $0.05)
# ==================================================
#   Intento 1: $0.000150 | Total: $0.000150 | Alert: OK
#   Intento 2: $0.000145 | Total: $0.000295 | Alert: OK
#   Intento 3: $0.000148 | Total: $0.000443 | Alert: OK
#   Intento 4: $0.000152 | Total: $0.000595 | Alert: OK
#   Intento 5: $0.000149 | Total: $0.000744 | Alert: OK
#
# ==================================================
# Usuario: user-pro-001 (límite diario: $0.5)
# ==================================================
#   Intento 1: $0.000150 | Total: $0.000150 | Alert: OK
#   Intento 2: $0.000145 | Total: $0.000295 | Alert: OK
#   Intento 3: $0.000148 | Total: $0.000443 | Alert: OK
#   Intento 4: $0.000152 | Total: $0.000595 | Alert: OK
#   Intento 5: $0.000149 | Total: $0.000744 | Alert: OK

Estrategias de optimización de costos

Reducir costos no es solo elegir un modelo más barato. Hay múltiples palancas que puedes combinar.

Estrategia 1: Prompts más cortos

Cada token en el prompt cuesta. Un system prompt de 500 tokens vs uno de 200 tokens es una diferencia de 2.5x en input cost — en cada llamada.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.messages import SystemMessage, HumanMessage

model = init_chat_model("openai:gpt-4.1-mini")

verbose_prompt = """Eres un asistente de investigación altamente calificado y experimentado.
Tu rol es ayudar a los usuarios a encontrar información relevante y precisa sobre
cualquier tema que te pregunten. Debes ser detallado en tus respuestas pero también
conciso. Siempre verifica los hechos antes de responder. Si no estás seguro de algo,
indícalo claramente. Responde siempre en español y usa un tono profesional pero
accesible. Estructura tus respuestas con bullet points cuando sea apropiado."""

concise_prompt = """Asistente de investigación. Responde en español, conciso, con bullet points. Indica incertidumbre si aplica."""

question = "¿Cuáles son las ventajas de usar LangGraph?"

response_verbose = model.invoke([
    SystemMessage(content=verbose_prompt),
    HumanMessage(content=question),
])

response_concise = model.invoke([
    SystemMessage(content=concise_prompt),
    HumanMessage(content=question),
])

verbose_input = response_verbose.usage_metadata["input_tokens"]
concise_input = response_concise.usage_metadata["input_tokens"]
saved = verbose_input - concise_input

print(f"System prompt verboso: {verbose_input} input tokens")
print(f"System prompt conciso: {concise_input} input tokens")
print(f"Tokens ahorrados:      {saved} ({saved/verbose_input*100:.0f}%)")
print(f"\nA 3000 llamadas/día × 30 días × $0.40/1M tokens:")
print(f"  Verboso: ${verbose_input * 90000 / 1_000_000 * 0.40:.2f}/mes")
print(f"  Conciso: ${concise_input * 90000 / 1_000_000 * 0.40:.2f}/mes")
# Output esperado:
# System prompt verboso: 98 input tokens
# System prompt conciso: 30 input tokens
# Tokens ahorrados:      68 (69%)
#
# A 3000 llamadas/día × 30 días × $0.40/1M tokens:
#   Verboso: $3.53/mes
#   Conciso: $1.08/mes

Estrategia 2: Message trimming (conecta con M8)

Cuando usas memoria, el historial de mensajes crece con cada turno. Sin trimming, después de 20 turnos puedes tener 5000+ tokens de historial que se envían en cada llamada.

from dotenv import load_dotenv
load_dotenv()

from langchain_core.messages import (
    SystemMessage, HumanMessage, AIMessage, trim_messages,
)

messages = [
    SystemMessage(content="Eres un asistente de investigación."),
    HumanMessage(content="¿Qué es Python?"),
    AIMessage(content="Python es un lenguaje de programación de alto nivel..."),
    HumanMessage(content="¿Y JavaScript?"),
    AIMessage(content="JavaScript es un lenguaje de programación que se ejecuta en el navegador..."),
    HumanMessage(content="¿Cuál es mejor para backend?"),
    AIMessage(content="Para backend, Python tiene frameworks como Django y FastAPI..."),
    HumanMessage(content="¿Y para AI?"),
    AIMessage(content="Para AI y machine learning, Python domina con librerías como..."),
    HumanMessage(content="Resume todo lo anterior en 2 oraciones."),
]

print(f"Total mensajes antes de trim: {len(messages)}")

trimmed = trim_messages(
    messages,
    max_tokens=200,
    strategy="last",
    token_counter=len,
    include_system=True,
    start_on="human",
)

print(f"Total mensajes después de trim: {len(trimmed)}")
print(f"\nMensajes conservados:")
for msg in trimmed:
    role = msg.__class__.__name__.replace("Message", "")
    print(f"  [{role}] {msg.content[:60]}...")
# Output esperado:
# Total mensajes antes de trim: 10
# Total mensajes después de trim: 3
#
# Mensajes conservados:
#   [System] Eres un asistente de investigación....
#   [AI] Para AI y machine learning, Python domina con librerías como...
#   [Human] Resume todo lo anterior en 2 oraciones....

Estrategia 3: Modelo barato para tareas simples (conecta con M4)

Ya lo viste en detalle en el Módulo 4 (Dynamic Models). El principio: usa el modelo más barato que pueda hacer el trabajo bien.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

PRICING = {
    "gpt-4.1": {"input": 2.00, "output": 8.00},
    "gpt-4.1-mini": {"input": 0.40, "output": 1.60},
    "gpt-4.1-nano": {"input": 0.10, "output": 0.40},
}

scenarios = {
    "Todo GPT-4.1":      {"decompose": "gpt-4.1", "search": "gpt-4.1", "analyze": "gpt-4.1", "write": "gpt-4.1"},
    "Mixto optimizado":  {"decompose": "gpt-4.1-nano", "search": "gpt-4.1-nano", "analyze": "gpt-4.1", "write": "gpt-4.1-mini"},
    "Todo GPT-4.1-nano": {"decompose": "gpt-4.1-nano", "search": "gpt-4.1-nano", "analyze": "gpt-4.1-nano", "write": "gpt-4.1-nano"},
}

avg_tokens_per_op = {
    "decompose": {"input": 200, "output": 100},
    "search": {"input": 150, "output": 200},
    "analyze": {"input": 1500, "output": 500},
    "write": {"input": 800, "output": 400},
}

print(f"{'Escenario':<22} {'Costo/ejecución':>16} {'Mensual (90K)':>14} {'Calidad':>10}")
print("─" * 66)

for scenario_name, assignment in scenarios.items():
    total = 0.0
    for op, model_name in assignment.items():
        pricing = PRICING[model_name]
        tokens = avg_tokens_per_op[op]
        cost = (tokens["input"] / 1_000_000) * pricing["input"] + \
               (tokens["output"] / 1_000_000) * pricing["output"]
        total += cost

    monthly = total * 90000
    quality = "Alta" if "gpt-4.1" in assignment.get("analyze", "") and assignment["analyze"] == "gpt-4.1" else "Media" if "mini" in assignment.get("analyze", "") else "Baja"
    print(f"{scenario_name:<22} ${total:>14.6f} ${monthly:>12.2f} {quality:>10}")
# Output esperado:
# Escenario              Costo/ejecución  Mensual (90K)    Calidad
# ──────────────────────────────────────────────────────────────────
# Todo GPT-4.1           $0.011700       $1053.00       Alta
# Mixto optimizado       $0.004035        $363.15       Alta
# Todo GPT-4.1-nano      $0.000530         $47.70       Baja

La fila "Mixto optimizado" es el sweet spot: calidad alta donde importa (análisis), costo bajo donde no afecta calidad (descomposición, búsqueda, escritura simple).

Monitoreo de costos: dashboards y alertas

En producción, necesitas visibilidad continua del gasto. LangSmith provee dashboards de costo out-of-the-box cuando tienes tracing habilitado. Pero también puedes construir tu propio monitoreo.

from dotenv import load_dotenv
load_dotenv()

from dataclasses import dataclass, field
from datetime import datetime

@dataclass
class DailyUsageRecord:
    date: str
    total_tokens: int = 0
    total_cost_usd: float = 0.0
    num_requests: int = 0
    by_model: dict = field(default_factory=dict)

class UsageMonitor:
    def __init__(self, daily_budget: float, alert_threshold: float = 0.8):
        self.daily_budget = daily_budget
        self.alert_threshold = alert_threshold
        self.records: dict[str, DailyUsageRecord] = {}

    def _today_key(self) -> str:
        return datetime.now().date().isoformat()

    def _get_today(self) -> DailyUsageRecord:
        key = self._today_key()
        if key not in self.records:
            self.records[key] = DailyUsageRecord(date=key)
        return self.records[key]

    def record(self, model: str, tokens: int, cost: float):
        today = self._get_today()
        today.total_tokens += tokens
        today.total_cost_usd += cost
        today.num_requests += 1
        today.by_model[model] = today.by_model.get(model, 0.0) + cost

        usage_pct = today.total_cost_usd / self.daily_budget
        if usage_pct >= 1.0:
            print(f"  [ALERT] BUDGET EXCEEDED: ${today.total_cost_usd:.4f}/${self.daily_budget:.4f}")
        elif usage_pct >= self.alert_threshold:
            print(f"  [WARNING] Budget at {usage_pct:.0%}: ${today.total_cost_usd:.4f}/${self.daily_budget:.4f}")

    def daily_report(self) -> str:
        today = self._get_today()
        pct = (today.total_cost_usd / self.daily_budget * 100) if self.daily_budget > 0 else 0
        lines = [
            f"Daily Usage Report — {today.date}",
            f"  Requests: {today.num_requests}",
            f"  Tokens:   {today.total_tokens:,}",
            f"  Cost:     ${today.total_cost_usd:.4f} / ${self.daily_budget:.4f} ({pct:.1f}%)",
            f"  By model:",
        ]
        for model, cost in today.by_model.items():
            lines.append(f"    {model}: ${cost:.4f}")
        return "\n".join(lines)


monitor = UsageMonitor(daily_budget=0.001, alert_threshold=0.8)

simulated_calls = [
    ("gpt-4.1-mini", 150, 0.00015),
    ("gpt-4.1-mini", 200, 0.00020),
    ("gpt-4.1", 500, 0.00080),
    ("gpt-4.1-mini", 180, 0.00018),
]

for model, tokens, cost in simulated_calls:
    print(f"Recording: {model} | {tokens} tokens | ${cost:.5f}")
    monitor.record(model, tokens, cost)

print(f"\n{monitor.daily_report()}")
# Output esperado:
# Recording: gpt-4.1-mini | 150 tokens | $0.00015
# Recording: gpt-4.1-mini | 200 tokens | $0.00020
# Recording: gpt-4.1 | 500 tokens | $0.00080
#   [WARNING] Budget at 115%: $0.00115/$0.00100
# Recording: gpt-4.1-mini | 180 tokens | $0.00018
#   [ALERT] BUDGET EXCEEDED: $0.00133/$0.00100
#
# Daily Usage Report — 2026-03-08
#   Requests: 4
#   Tokens:   1,030
#   Cost:     $0.0013 / $0.0010 (133.0%)
#   By model:
#     gpt-4.1-mini: $0.0005
#     gpt-4.1: $0.0008

Troubleshooting

Problema 1: `usage_metadata` retorna None o está vacío

Causa: No todos los proveedores incluyen usage metadata por defecto. Algunos requieren configuración explícita.

Solución: Verifica que el proveedor soporta usage metadata. Para OpenAI y Anthropic funciona por defecto. Para otros proveedores, revisa la documentación:

response = model.invoke("test")
if response.usage_metadata:
    print(f"Tokens: {response.usage_metadata}")
else:
    print("Usage metadata no disponible para este proveedor")

Problema 2: Los costos calculados no coinciden con la factura del proveedor

Causa: Los precios cambian frecuentemente, o estás usando precios desactualizados en tu tabla PRICING.

Solución: Siempre verifica los precios actuales en la página del proveedor. Considera mantener los precios en un archivo de configuración que se actualice periódicamente:

PRICING_LAST_UPDATED = "2026-03-01"

Problema 3: El callback `get_openai_callback` no captura llamadas de Anthropic

Causa: get_openai_callback es específico para modelos de OpenAI. No funciona con otros proveedores.

Solución: Usa UsageMetadataCallbackHandler que es proveedor-agnóstico, o accede directamente a response.usage_metadata de cada llamada.

Problema 4: Los tokens cacheados no aparecen en el desglose

Causa: El caching de prompts requiere que el prefijo tenga una longitud mínima (típicamente 1024+ tokens para OpenAI). Prompts cortos nunca se cachean.

Solución: El cache automático solo se activa con prompts suficientemente largos. Si tu system prompt tiene menos de 1024 tokens, no verás beneficio de caching. Para Anthropic, usa cache_control explícito.

Problema 5: El `CostTracker` no persiste entre reinicios

Causa: La implementación en memoria se pierde cuando el proceso termina.

Solución: Para producción, persiste los datos. LangSmith ya hace esto por ti si tienes tracing habilitado. Si necesitas tracking custom, guarda en base de datos:

import json

def save_tracker(tracker, filepath="cost_log.json"):
    data = [{"op": e.operation, "cost": e.cost_usd, "ts": e.timestamp} for e in tracker.entries]
    with open(filepath, "w") as f:
        json.dump(data, f)

Ejercicios

Ejercicio 1: Calcular costo de una conversación (Fácil)

Crea una función que reciba una lista de respuestas del modelo y calcule el costo total. Pruébala con 3 llamadas al modelo usando prompts de diferente longitud.

Ver solución

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

PRICING = {"gpt-4.1-mini": {"input": 0.40, "output": 1.60}}

def total_cost_from_responses(responses: list, model_name: str) -> dict:
    """Calcula el costo total de una lista de respuestas."""
    total_input = 0
    total_output = 0
    pricing = PRICING[model_name]

    for r in responses:
        total_input += r.usage_metadata["input_tokens"]
        total_output += r.usage_metadata["output_tokens"]

    input_cost = (total_input / 1_000_000) * pricing["input"]
    output_cost = (total_output / 1_000_000) * pricing["output"]

    return {
        "total_input_tokens": total_input,
        "total_output_tokens": total_output,
        "input_cost": input_cost,
        "output_cost": output_cost,
        "total_cost": input_cost + output_cost,
    }

model = init_chat_model("openai:gpt-4.1-mini")

responses = [
    model.invoke("Hola"),
    model.invoke("¿Qué es Python? Responde en 1 oración."),
    model.invoke("Explica qué es machine learning, sus tipos principales, y da un ejemplo de cada uno."),
]

result = total_cost_from_responses(responses, "gpt-4.1-mini")
print(f"Input tokens:  {result['total_input_tokens']}")
print(f"Output tokens: {result['total_output_tokens']}")
print(f"Costo input:   ${result['input_cost']:.6f}")
print(f"Costo output:  ${result['output_cost']:.6f}")
print(f"Costo total:   ${result['total_cost']:.6f}")
# Output esperado:
# Input tokens:  50
# Output tokens: 200
# Costo input:   $0.000020
# Costo output:  $0.000320
# Costo total:   $0.000340

Explicación: La función itera sobre las respuestas, acumula tokens, y calcula el costo usando los precios del modelo. Separar costo de input y output es importante porque tienen precios diferentes.

Ejercicio 2: CostTracker con presupuesto diario (Fácil)

Extiende el CostTracker para que acepte un presupuesto diario y lance una alerta cuando se alcance el 80%.

Ver solución

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from dataclasses import dataclass

PRICING = {"gpt-4.1-mini": {"input": 0.40, "output": 1.60}}

class BudgetCostTracker:
    def __init__(self, model_name: str, daily_budget: float):
        self.model_name = model_name
        self.daily_budget = daily_budget
        self.total_cost = 0.0
        self.call_count = 0

    def track(self, usage_metadata: dict) -> dict:
        pricing = PRICING[self.model_name]
        cost = (usage_metadata["input_tokens"] / 1_000_000) * pricing["input"] + \
               (usage_metadata["output_tokens"] / 1_000_000) * pricing["output"]
        self.total_cost += cost
        self.call_count += 1

        pct = (self.total_cost / self.daily_budget * 100) if self.daily_budget > 0 else 0
        alert = None
        if pct >= 100:
            alert = "BLOCKED"
        elif pct >= 80:
            alert = "WARNING"

        return {"cost": cost, "total": self.total_cost, "pct": pct, "alert": alert}

tracker = BudgetCostTracker("gpt-4.1-mini", daily_budget=0.001)
model = init_chat_model("openai:gpt-4.1-mini")

for i in range(8):
    response = model.invoke(f"Pregunta {i+1}: Explica un concepto de AI en 2 oraciones.")
    result = tracker.track(response.usage_metadata)
    alert_str = f" [{result['alert']}]" if result['alert'] else ""
    print(f"Call {i+1}: ${result['cost']:.6f} | Total: ${result['total']:.6f} ({result['pct']:.1f}%){alert_str}")
# Output esperado:
# Call 1: $0.000120 | Total: $0.000120 (12.0%)
# Call 2: $0.000115 | Total: $0.000235 (23.5%)
# Call 3: $0.000118 | Total: $0.000353 (35.3%)
# Call 4: $0.000122 | Total: $0.000475 (47.5%)
# Call 5: $0.000119 | Total: $0.000594 (59.4%)
# Call 6: $0.000125 | Total: $0.000719 (71.9%)
# Call 7: $0.000121 | Total: $0.000840 (84.0%) [WARNING]
# Call 8: $0.000118 | Total: $0.000958 (95.8%) [WARNING]

Explicación: El tracker verifica el porcentaje del presupuesto consumido después de cada llamada y emite alertas progresivas. En producción, estas alertas se enviarían por Slack o email.

Ejercicio 3: Comparar costo de 3 modelos para la misma tarea (Medio)

Envía el mismo prompt a GPT-4.1, GPT-4.1-mini, y GPT-4.1-nano. Compara tokens usados, costo, y calidad de respuesta.

Ver solución

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

PRICING = {
    "gpt-4.1": {"input": 2.00, "output": 8.00},
    "gpt-4.1-mini": {"input": 0.40, "output": 1.60},
    "gpt-4.1-nano": {"input": 0.10, "output": 0.40},
}

model_names = ["gpt-4.1", "gpt-4.1-mini", "gpt-4.1-nano"]
prompt = "Explica las 3 ventajas principales de usar LangGraph para construir agentes de AI. Sé conciso."

print(f"Prompt: {prompt}\n")
print(f"{'Modelo':<16} {'Input':>7} {'Output':>7} {'Costo':>12} {'Respuesta (primeros 80 chars)'}")
print("─" * 90)

for model_name in model_names:
    model = init_chat_model(f"openai:{model_name}")
    response = model.invoke(prompt)
    usage = response.usage_metadata
    pricing = PRICING[model_name]

    cost = (usage["input_tokens"] / 1_000_000) * pricing["input"] + \
           (usage["output_tokens"] / 1_000_000) * pricing["output"]

    preview = response.content[:80].replace("\n", " ")
    print(f"{model_name:<16} {usage['input_tokens']:>7} {usage['output_tokens']:>7} ${cost:>10.6f} {preview}...")
# Output esperado:
# Prompt: Explica las 3 ventajas principales de usar LangGraph para construir agentes de AI. Sé conciso.
#
# Modelo             Input  Output        Costo Respuesta (primeros 80 chars)
# ──────────────────────────────────────────────────────────────────────────────────────────
# gpt-4.1               22     180   $0.001484 1. **Control de flujo granular**: LangGraph permite definir grafos de estado co...
# gpt-4.1-mini          22     150   $0.000249 1. **Control preciso del flujo**: Permite diseñar el flujo del agente como un g...
# gpt-4.1-nano          22     120   $0.000050 1. Control de flujo: Define grafos para manejar la lógica del agente de forma ...

Explicación: El mismo prompt consume tokens similares de input pero la diferencia de costo es dramática: GPT-4.1 cuesta ~30x más que nano. La calidad de respuesta varía, pero para muchas tareas el modelo económico es suficiente.

Ejercicio 4: Tracker que muestra el desglose porcentual por operación (Medio)

Crea un tracker que ejecute las 4 operaciones del Research Assistant (decompose, search, analyze, write) y muestre qué porcentaje del costo total representa cada una.

Ver solución

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

PRICING = {"gpt-4.1-mini": {"input": 0.40, "output": 1.60}}

class OperationTracker:
    def __init__(self):
        self.operations: dict[str, float] = {}

    def track(self, operation: str, usage: dict):
        cost = (usage["input_tokens"] / 1_000_000) * PRICING["gpt-4.1-mini"]["input"] + \
               (usage["output_tokens"] / 1_000_000) * PRICING["gpt-4.1-mini"]["output"]
        self.operations[operation] = self.operations.get(operation, 0.0) + cost

    def breakdown(self):
        total = sum(self.operations.values())
        print(f"\nCost Breakdown (Total: ${total:.6f})")
        print(f"{'Operation':<15} {'Cost':>12} {'Share':>8} {'Bar'}")
        print("─" * 50)
        for op, cost in sorted(self.operations.items(), key=lambda x: -x[1]):
            pct = (cost / total * 100) if total > 0 else 0
            bar = "█" * int(pct / 2)
            print(f"{op:<15} ${cost:>10.6f} {pct:>6.1f}% {bar}")

tracker = OperationTracker()
model = init_chat_model("openai:gpt-4.1-mini")

ops = [
    ("decompose", "Genera 3 sub-preguntas sobre 'AI en medicina'. Solo las preguntas, una por línea."),
    ("search", "¿Qué es AI aplicada a diagnóstico? 1 oración."),
    ("search", "¿Qué avances hay en AI para drug discovery? 1 oración."),
    ("search", "¿Cómo se usa AI en imágenes médicas? 1 oración."),
    ("analyze", "Basándote en que AI mejora diagnósticos, acelera drug discovery, y analiza imágenes médicas, identifica 3 tendencias principales. Explica cada una en 2 oraciones."),
    ("write", "Escribe un resumen ejecutivo de 4 oraciones sobre AI en medicina, cubriendo diagnóstico, farmacéutica e imágenes médicas."),
]

for op_name, prompt in ops:
    response = model.invoke(prompt)
    tracker.track(op_name, response.usage_metadata)

tracker.breakdown()
# Output esperado:
# Cost Breakdown (Total: $0.000680)
# Operation             Cost    Share Bar
# ──────────────────────────────────────────────────
# analyze         $0.000250  36.8% ██████████████████
# write           $0.000165  24.3% ████████████
# search          $0.000180  26.5% █████████████
# decompose       $0.000085  12.5% ██████

Explicación: El tracker agrupa costos por operación. El análisis es la operación más cara (prompt largo + respuesta detallada). Este desglose te dice dónde optimizar primero.

Ejercicio 5: Sistema de token budget por tier de usuario (Difícil)

Implementa un sistema donde usuarios "free" tienen un presupuesto de $0.01/día, usuarios "pro" de $0.10/día, y usuarios "enterprise" de $1.00/día. Simula 10 requests por usuario y muestra quién se bloquea primero.

Ver solución

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

PRICING = {"gpt-4.1-mini": {"input": 0.40, "output": 1.60}}
TIER_BUDGETS = {"free": 0.001, "pro": 0.010, "enterprise": 0.100}

class TieredUser:
    def __init__(self, user_id: str, tier: str):
        self.user_id = user_id
        self.tier = tier
        self.budget = TIER_BUDGETS[tier]
        self.spent = 0.0
        self.successful_requests = 0
        self.blocked_requests = 0

    def try_request(self, cost: float) -> bool:
        if self.spent + cost > self.budget:
            self.blocked_requests += 1
            return False
        self.spent += cost
        self.successful_requests += 1
        return True

    def status(self) -> str:
        pct = (self.spent / self.budget * 100) if self.budget > 0 else 0
        return (f"{self.user_id} ({self.tier}): "
                f"${self.spent:.6f}/${self.budget:.4f} ({pct:.0f}%) | "
                f"OK: {self.successful_requests} | Blocked: {self.blocked_requests}")

model = init_chat_model("openai:gpt-4.1-mini")
users = [
    TieredUser("alice", "free"),
    TieredUser("bob", "pro"),
    TieredUser("corp-1", "enterprise"),
]

prompts = [f"Explica el concepto #{i+1} de AI engineering en 2 oraciones." for i in range(10)]

for prompt in prompts:
    response = model.invoke(prompt)
    usage = response.usage_metadata
    cost = (usage["input_tokens"] / 1_000_000) * PRICING["gpt-4.1-mini"]["input"] + \
           (usage["output_tokens"] / 1_000_000) * PRICING["gpt-4.1-mini"]["output"]

    for user in users:
        user.try_request(cost)

print("Final Status:")
print("─" * 70)
for user in users:
    print(user.status())
# Output esperado:
# Final Status:
# ──────────────────────────────────────────────────────────────────────
# alice (free): $0.000950/0.0010 (95%) | OK: 8 | Blocked: 2
# bob (pro): $0.001200/0.0100 (12%) | OK: 10 | Blocked: 0
# corp-1 (enterprise): $0.001200/0.1000 (1%) | OK: 10 | Blocked: 0

Explicación: El usuario free se bloquea antes de completar las 10 requests. El pro y enterprise tienen presupuesto de sobra. Este patrón es la base para monetización de APIs con LLMs.

Ejercicio 6: Dashboard completo con proyección mensual (Difícil)

Construye un dashboard que combine tracking por operación, tracking por modelo, y genere una proyección de costo mensual basada en el uso actual. Simula un flujo completo del Research Assistant.

Ver solución

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from datetime import datetime

PRICING = {
    "gpt-4.1": {"input": 2.00, "output": 8.00},
    "gpt-4.1-mini": {"input": 0.40, "output": 1.60},
}

class ProductionDashboard:
    def __init__(self):
        self.entries = []

    def track(self, operation: str, model_name: str, usage: dict):
        pricing = PRICING.get(model_name, PRICING["gpt-4.1-mini"])
        cost = (usage["input_tokens"] / 1_000_000) * pricing["input"] + \
               (usage["output_tokens"] / 1_000_000) * pricing["output"]
        self.entries.append({
            "operation": operation,
            "model": model_name,
            "input_tokens": usage["input_tokens"],
            "output_tokens": usage["output_tokens"],
            "cost": cost,
            "timestamp": datetime.now().isoformat(),
        })

    def render(self, executions_per_day: int = 3000):
        total_cost = sum(e["cost"] for e in self.entries)
        total_tokens = sum(e["input_tokens"] + e["output_tokens"] for e in self.entries)

        by_op = {}
        for e in self.entries:
            by_op[e["operation"]] = by_op.get(e["operation"], 0.0) + e["cost"]

        by_model = {}
        for e in self.entries:
            by_model[e["model"]] = by_model.get(e["model"], 0.0) + e["cost"]

        print(f"╔{'═'*56}╗")
        print(f"║{'PRODUCTION COST DASHBOARD':^56}║")
        print(f"╚{'═'*56}╝")
        print(f"  This execution:")
        print(f"    Calls:  {len(self.entries)}")
        print(f"    Tokens: {total_tokens:,}")
        print(f"    Cost:   ${total_cost:.6f}")
        print(f"\n  By operation:")
        for op, cost in sorted(by_op.items(), key=lambda x: -x[1]):
            pct = cost / total_cost * 100 if total_cost > 0 else 0
            print(f"    {op:<15} ${cost:.6f} ({pct:.0f}%)")
        print(f"\n  By model:")
        for model, cost in sorted(by_model.items(), key=lambda x: -x[1]):
            pct = cost / total_cost * 100 if total_cost > 0 else 0
            print(f"    {model:<16} ${cost:.6f} ({pct:.0f}%)")
        daily = total_cost * executions_per_day
        monthly = daily * 30
        print(f"\n  Projection ({executions_per_day:,} exec/day):")
        print(f"    Daily:   ${daily:.2f}")
        print(f"    Monthly: ${monthly:.2f}")
        print(f"    Yearly:  ${monthly * 12:.2f}")

dashboard = ProductionDashboard()

mini = init_chat_model("openai:gpt-4.1-mini")
strong = init_chat_model("openai:gpt-4.1")

ops = [
    ("decompose", mini, "gpt-4.1-mini", "Lista 3 sub-preguntas sobre 'AI en producción'. Solo preguntas."),
    ("search", mini, "gpt-4.1-mini", "¿Qué es observabilidad en AI? 1 oración."),
    ("search", mini, "gpt-4.1-mini", "¿Qué es rate limiting en APIs? 1 oración."),
    ("analyze", strong, "gpt-4.1", "Analiza cómo la observabilidad y el rate limiting se combinan para hacer sistemas de AI production-ready. 3 oraciones."),
    ("write", mini, "gpt-4.1-mini", "Escribe un resumen de 3 oraciones sobre preparar AI systems para producción."),
]

for op_name, model, model_name, prompt in ops:
    response = model.invoke(prompt)
    dashboard.track(op_name, model_name, response.usage_metadata)

dashboard.render()
# Output esperado:
# ╔════════════════════════════════════════════════════════╗
# ║              PRODUCTION COST DASHBOARD                ║
# ╚════════════════════════════════════════════════════════╝
#   This execution:
#     Calls:  5
#     Tokens: 480
#     Cost:   $0.000850
#
#   By operation:
#     analyze         $0.000520 (61%)
#     search          $0.000140 (16%)
#     write           $0.000110 (13%)
#     decompose       $0.000080 (9%)
#
#   By model:
#     gpt-4.1          $0.000520 (61%)
#     gpt-4.1-mini     $0.000330 (39%)
#
#   Projection (3,000 exec/day):
#     Daily:   $2.55
#     Monthly: $76.50
#     Yearly:  $918.00

Explicación: El dashboard combina tracking por operación y por modelo, dando visibilidad completa. El análisis con GPT-4.1 domina el costo (61%), confirmando que la elección de modelo es la palanca de optimización más importante.

Resumen

En esta cápsula aprendiste:

Token tracking es una herramienta de negocio — "esta investigación costó $0.035" no es un dato técnico, es un dato que determina pricing, márgenes, y viabilidad del producto
usage_metadata en cada AIMessage te da el desglose de tokens por llamada individual
get_openai_callback acumula tokens y costo de todas las llamadas dentro de un bloque with
UsageMetadataCallbackHandler ofrece tracking granular compatible con múltiples proveedores
CostTracker como clase reutilizable permite desglose por operación, proyecciones mensuales, y reportes de negocio
Model routing por costo (GPT-4.1 para análisis, nano para tareas simples) puede reducir costos 40-65% sin sacrificar calidad donde importa
Prompt caching reduce costos de tokens de input repetitivos hasta un 90% — especialmente valioso con system prompts largos
Token budgets por usuario previenen costos descontrolados y habilitan modelos de negocio por tiers (free/pro/enterprise)
Message trimming mantiene el contexto lean y reduce costos acumulados en conversaciones largas

Próxima cápsula: Rate Limiting y Cost Control — cómo prevenir que un solo usuario agote tu presupuesto diario.

Recursos adicionales

LangSmith Cost Tracking — Dashboard de costos integrado con tracing
OpenAI Token Usage — Documentación de usage metadata de OpenAI
Anthropic Prompt Caching — Guía oficial de prompt caching con Claude
OpenAI Pricing — Precios actualizados de modelos
LangChain Callbacks Guide — Sistema de callbacks para tracking y monitoring
LangChain Message Trimming — Cómo recortar mensajes para optimizar tokens

Módulo 12 — LangChain & LangGraph: From Chains to Agents

← AnteriorEvaluation: Datasets y Evaluators Siguiente →Rate Limiting y Cost Control

Token Usage Tracking y Prompt Caching

Descripción de la cápsula

Token tracking como herramienta de negocio

UsageMetadata: tokens por llamada

Calculando costo por llamada

get_openai_callback: tracking agregado por bloque

Tracking por operación en un agente

UsageMetadataCallbackHandler: tracking granular

Tracking a lo largo de múltiples llamadas

Clase CostTracker: tracking de producción

Costo por modelo: eligiendo el modelo correcto

Prompt caching: reducir costos en patrones repetitivos

Cómo funciona el caching

Detección de tokens cacheados en la respuesta

Anthropic cache_control explícito

Token budgets: límites por usuario

Estrategias de optimización de costos

Estrategia 1: Prompts más cortos

Estrategia 2: Message trimming (conecta con M8)

Estrategia 3: Modelo barato para tareas simples (conecta con M4)

Monitoreo de costos: dashboards y alertas

Troubleshooting

Problema 1: usage_metadata retorna None o está vacío

Problema 2: Los costos calculados no coinciden con la factura del proveedor

Problema 3: El callback get_openai_callback no captura llamadas de Anthropic

Problema 4: Los tokens cacheados no aparecen en el desglose

Problema 5: El CostTracker no persiste entre reinicios

Ejercicios

Ejercicio 1: Calcular costo de una conversación (Fácil)

Ejercicio 2: CostTracker con presupuesto diario (Fácil)

Ejercicio 3: Comparar costo de 3 modelos para la misma tarea (Medio)

Ejercicio 4: Tracker que muestra el desglose porcentual por operación (Medio)

Ejercicio 5: Sistema de token budget por tier de usuario (Difícil)

Ejercicio 6: Dashboard completo con proyección mensual (Difícil)

Resumen

Recursos adicionales

Problema 1: `usage_metadata` retorna None o está vacío

Problema 3: El callback `get_openai_callback` no captura llamadas de Anthropic

Problema 5: El `CostTracker` no persiste entre reinicios