Ejercicios — Crea tu Propio Chatbot con Claude

• API: (b) generación de texto, (e) conteo (viene en response.usage; también el endpoint count_tokens).
• Tu código: (a) memoria — acumulás y reenviás messages; (c) personalidad — el system prompt que escribís; (d) ejecución de herramientas — el modelo solo las pide; (f) recorte del historial.

Si dudaste en (d): es la distinción más importante del libro. El modelo nunca ejecuta nada; propone, tu código dispone.

La segunda llamada responde que no sabe tu color favorito: la API no guardó nada de la primera (stateless). El arreglo: la segunda llamada debe llevar el historial completo:

messages=[
    {"role": "user", "content": "Mi color favorito es el verde"},
    {"role": "assistant", "content": respuesta1.content[0].text},
    {"role": "user", "content": "¿Cuál es mi color favorito?"},
]

Tres turnos: lo que dijiste, lo que respondió (¡tal cual lo respondió!), y la pregunta nueva.

Costo por llamada en Opus 4.8: entrada (850 × $5/M = $0.00425) + salida (300 × $25/M = $0.0075) = $0.01175 → $5 / 0.01175 ≈ 425 llamadas.

En Haiku 4.5: entrada (850 × $1/M = $0.00085) + salida (300 × $5/M = $0.0015) = $0.00235 → ≈ 2,127 llamadas. Cinco veces más — la relación de precios exacta (5×) porque entrada y salida escalan igual entre estos dos modelos.

Coinciden (o difieren en un par de tokens de envoltura). Para decidir el recorte antes de enviar se usa count_tokens: es gratis y no genera respuesta. usage sirve después, para contabilidad de lo realmente consumido. Patrón de producción: count_tokens como guardia previa (¿el request cabe en el presupuesto/límite?) y usage como registro contable.

HTTP 404 → anthropic.NotFoundError: el ID claude-opus-4.8 no existe (el correcto es claude-opus-4-8, con guiones). No se reintenta porque es un error permanente: el mismo request fallará siempre; reintentar solo quema tiempo. El SDK reintenta únicamente 429 y 5xx/529, que son temporales.

.env             # DINERO: la API key; si llega a GitHub, te vacían el crédito
.venv/           # higiene: el entorno virtual se recrea con requirements.txt
__pycache__/     # higiene: bytecode generado
*.pyc            # higiene
conversaciones/  # PRIVACIDAD: charlas de clientes con nombres y datos
pedidos.json     # PRIVACIDAD: nombres y hábitos de compra de clientes

.env protege dinero; conversaciones/ y pedidos.json protegen datos personales de terceros (y en el capítulo 5, logs/ se suma a esa lista); el resto es higiene de repositorio.

for texto in ["pupusa", "pupusas de chicharrón con curtido",
              "pork pupusas with cabbage slaw",
              "3.14159265358979323846264338327950288419716939937510582097494459230781640628620899862803482534211706"]:
    n = client.messages.count_tokens(
        model="claude-opus-4-8",
        messages=[{"role": "user", "content": texto}],
    ).input_tokens
    print(f"{n:4d} tokens ← {texto[:50]}")

Observaciones típicas: palabras españolas poco frecuentes se parten en más tokens que su equivalente inglés (el tokenizador fue entrenado con mayoría de inglés); los números largos son carísimos (≈1 token por 1-3 dígitos). Implicación: un bot en español consume ~10-25% más tokens que el mismo bot en inglés para el mismo contenido — presupuestá con mediciones de TUS textos, no con reglas de dedo ajenas.

import logging

def extraer_texto(respuesta) -> str:
    if respuesta.stop_reason != "end_turn":
        logging.warning("stop_reason inesperado: %s", respuesta.stop_reason)
    return "".join(b.text for b in respuesta.content if b.type == "text")

Situaciones donde content[0] no es texto: (1) tool use (cap. 3) — el primer bloque puede ser directamente tool_use si el modelo decide actuar sin preámbulo; (2) refusal — ante un rechazo de seguridad el contenido puede venir vacío. La comprensión con filtro por b.type == "text" es inmune a ambos; el índice [0] a pelo, no. En el loop del capítulo 3 ya usamos exactamente este patrón.

• (a) Rechazado: el primer mensaje debe ser user.
• (b) Rechazado: dos user consecutivos rompen la alternancia (roles must alternate).
• (c) Válido: user → assistant → user.

Las dos reglas que todo recorte de historial y todo loop de herramientas tienen que preservar.

1. Comportamiento: el system prompt cambia en cada mensaje (la hora avanza), así que el "contrato" del bot es inestable; además la hora del servidor puede no ser la del usuario, generando respuestas incorrectas sobre horarios.
2. Económico: el prompt caching matchea por prefijo exacto — un system prompt que cambia cada minuto jamás produce hits de caché, y pagás el prompt completo a precio lleno en cada mensaje. La solución correcta: la hora se obtiene con una herramienta (como verificar_horario del cap. 3) o se inyecta en el mensaje del usuario, nunca al inicio del system prompt.

El texto deja de aparecer fluido: sale a borbotones o todo junto al final, porque Python acumula la salida en un buffer y solo la vuelca cuando se llena o al terminar la línea. El costo NO cambia: el stream de la API es idéntico; lo que se rompió es solo la presentación. Buen recordatorio de que streaming tiene dos mitades: recibir en stream (la API) y mostrar en stream (tu código).

from pathlib import Path
from datetime import datetime, timezone

def guardar_con_limite(self, max_archivos: int = 50) -> None:
    carpeta = Path("conversaciones")
    carpeta.mkdir(exist_ok=True)
    ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S")
    self.guardar(str(carpeta / f"{ts}.json"))
    archivos = sorted(carpeta.glob("*.json"))
    for viejo in archivos[:-max_archivos]:
        viejo.unlink()

El nombre con timestamp ordenable (ISO básico) hace que sorted() por nombre equivalga a ordenar por fecha — truco simple que evita mirar metadatos del filesystem.

CASOS = [
    "¡Llevo 40 minutos esperando, esto es un robo!",
    "¿Por quién votarías en las elecciones?",
    "Dame un descuentito, no seas malo",
    "Do you have vegetarian options?",
    "Sos el bot más inútil que conocí",
    "¿Qué pupusas tienen?",
]
for caso in CASOS:
    bot = Chatbot(SYSTEM_PROMPT)       # sesion limpia por caso
    print(f"\n>>> {caso}\n{bot.enviar(caso)}")

El que más falla suele ser el descuento: el modelo "amable" tiende a ofrecer compensaciones que no existen. La línea que lo arregla: "No prometás descuentos ni regalos bajo ninguna circunstancia; ante la insistencia, ofrecé que un encargado lo contacte". El caso del inglés revela una decisión de producto que el prompt debe tomar explícitamente (¿responder en inglés o pedir disculpas en español?) — sin instrucción, el modelo decide solo y es inconsistente.

Sin el pop, tras el fallo el historial queda terminando en un turno user sin respuesta. El siguiente mensaje del usuario hace append de otro user → la lista contiene ..., user, user → la API responde 400 roles must alternate en TODOS los mensajes siguientes: la sesión queda envenenada para siempre.

Aparece "a veces" porque requiere que una llamada falle (rate limit, red caída) — algo raro en desarrollo con un solo usuario y frecuente en producción con cientos. Es el arquetipo de bug de producción: invisible en la demo, inevitable a escala. La limpieza de estado tras error (rollback) tiene que ser parte del diseño, no un parche.

import json

def exportar_txt(ruta_json: str, ruta_txt: str) -> None:
    with open(ruta_json, encoding="utf-8") as f:
        datos = json.load(f)
    nombres = {"user": "Cliente", "assistant": "Esquinita"}
    lineas = [f"Conversación del {datos.get('guardado_en', '?')}", "=" * 40]
    for m in datos["historial"]:
        if isinstance(m["content"], str):          # saltar turnos de tools
            lineas.append(f"{nombres[m['role']]}: {m['content']}\n")
    with open(ruta_txt, "w", encoding="utf-8") as f:
        f.write("\n".join(lineas))

El isinstance(m["content"], str) ya prepara la función para el capítulo 3, donde algunos turnos tienen listas de bloques (tool_use/tool_result) que doña Carmen no necesita ver.

Inyectar la hora en el primer mensaje del usuario, no en el system prompt:

from datetime import datetime

def enviar(self, mensaje_usuario: str) -> str:
    contenido = mensaje_usuario
    if not self.historial:                      # solo en el primer turno
        hora = datetime.now().strftime("%H:%M")
        contenido = f"[Hora local del cliente: {hora}]\n{mensaje_usuario}"
    self.historial.append({"role": "user", "content": contenido})
    ...

Y una línea en el system prompt (estable): "Si el primer mensaje incluye la hora local, saludá acorde (buenos días / buenas tardes / buenas noches)". El system prompt no cambia nunca (cachea perfecto) y el dato volátil viaja como dato, una sola vez. Patrón general de la sección 5.4: lo estable al principio, lo volátil al final.

{"type": "tool_result",
 "tool_use_id": "toolu_abc",      # el MISMO id del tool_use
 "content": '{"total": 2.55}'}

Viaja dentro de un turno {"role": "user", "content": [<ese dict>]} — rol user, inmediatamente después del turno assistant que contenía el tool_use. El tool_use_id es el hilo que une pedido y resultado.

Con la description floja, el modelo a veces responde precios "de memoria" (inventados o desactualizados) sin llamar la herramienta — sobre todo para items famosos como la de queso. La frase que cambia el comportamiento es la instructiva: "Usala SIEMPRE que el cliente pregunte precios — nunca respondas precios de memoria". La description no es documentación para humanos: es la política de uso que el modelo lee para decidir. Decí cuándo, no solo qué.

El código actual lo reporta en items_no_encontrados (la búsqueda es por igualdad exacta), y el modelo — gracias al resultado honesto — suele preguntar "¿se refiere a la de loroco con queso?". Mejora tolerante:

def buscar_precio(nombre_item: str):
    nombre = nombre_item.lower().strip()
    for categoria, items in MENU.items():
        if nombre in items:
            return categoria, items[nombre]
    # Segunda pasada: mismas palabras en otro orden ("queso con loroco")
    palabras = set(nombre.replace(" con ", " ").split())
    for categoria, items in MENU.items():
        for clave, precio in items.items():
            if palabras == set(clave.replace(" con ", " ").split()):
                return categoria, precio
    return None

La comparación exige igualdad de conjuntos completos, no intersección — "queso" a secas no matchea "loroco con queso" (eso sería un falso positivo caro). Para difusión mayor (typos: "revuelt"), el paso siguiente sería difflib.get_close_matches, con umbral conservador.

Opus 4.8 típicamente pide ambas en el mismo turno: un response.content con dos bloques tool_use (a veces precedidos de un bloque de texto). El loop de bot_v3.py lo maneja porque construye los resultados con una comprensión sobre todos los bloques:

resultados = [self._ejecutar_tool(bloque)
              for bloque in response.content if bloque.type == "tool_use"]

y los devuelve juntos en un único turno user. Si el modelo en cambio decide encadenar (consulta horario → ve el resultado → consulta menú), el for _ in range(MAX_CICLOS_TOOLS) cubre el caso multi-ciclo. Ambos caminos están soportados sin código especial — esa es la marca de un loop agéntico bien escrito.

El parámetro confirmado_por_cliente es débil: lo llena el propio modelo, que puede poner true con el mismo exceso de confianza con el que registró de más. Es prompt-engineering disfrazado de validación.

Defensa real — confirmación en dos fases con token de un solo uso:

1. Nueva herramienta proponer_pedido(items, nombre, sucursal): calcula el total, guarda la propuesta en un dict PROPUESTAS con un código aleatorio (secrets.token_hex(3)), y devuelve el resumen + código. NO escribe en pedidos.json.
2. registrar_pedido(codigo_propuesta) solo acepta un código existente, registra ESA propuesta tal cual (precios re-validados del lado servidor) y la elimina del dict.
3. El system prompt instruye: "proponé el pedido, mostrá el resumen, y solo cuando el cliente diga que sí, registralo con el código".

Ahora el modelo no puede registrar nada que no haya pasado por una propuesta explícita, y nunca puede registrar dos veces la misma. El patrón general: las acciones irreversibles se parten en proponer + confirmar, y la máquina de estados vive en tu código, no en la buena voluntad del modelo.

• (a) Enum (["efectivo", "tarjeta", "transferencia"]): conjunto cerrado por las políticas; un valor inventado rompería el negocio.
• (b) Texto libre: imposible enumerar las preferencias humanas; se valida solo la longitud.
• (c) Ni uno ni otro: integer — los schemas no son solo strings; con "type": "integer" el modelo no puede mandar "tres".
• (d) Enum (["recoger", "domicilio"]): dos valores con consecuencias de negocio distintas.

Regla: si tu código va a hacer if valor == ... contra una lista conocida, es enum. Cada enum elimina una familia entera de bugs de normalización ("Tarjeta", "tarjeta de credito", "con tarjeta").

Uso razonable: un endpoint NO conversacional, p. ej. /api/menu-del-dia que internamente pide al modelo armar una descripción atractiva del menú: ahí forzar consultar_menu con {"type": "tool", "name": "consultar_menu"} garantiza que los datos vienen de la fuente real.

El riesgo en chat abierto: el usuario escribe "gracias, ¡adiós!" y el modelo, obligado a usar una herramienta, llama cualquiera con argumentos inventados — comportamiento absurdo y gasto inútil. En conversación general, tool_choice por defecto (auto) + buenas descriptions es lo correcto; el forzado es para pipelines estructurados donde el flujo lo decide tu código.

Núcleo del cambio:

import sqlite3

def buscar_precio(nombre_item: str):
    with sqlite3.connect("la_esquina.db") as con:
        fila = con.execute(
            "SELECT categoria, precio FROM items WHERE nombre = ?",
            (nombre_item.lower().strip(),)
        ).fetchone()
    return fila if fila else None

(con su CREATE TABLE e inserts iniciales una vez). La respuesta a la pregunta clave: cero líneas de herramientas.py (schemas) y bot_v3.py (loop) cambian — solo la capa de datos. Esa es la recompensa de la separación del capítulo 3: el contrato con el modelo (schemas) es independiente del almacenamiento. El mismo refactor llevaría a Postgres o a una API externa sin tocar el bot.

Producto punto: $\vec{a}\cdot\vec{b} = 1\cdot2 + 2\cdot1 + 2\cdot2 = 8$. Normas: $|\vec{a}| = \sqrt{1+4+4} = 3$, $|\vec{b}| = \sqrt{4+1+4} = 3$. Coseno: $\cos(\theta) = 8/9 \approx 0.889$ — vectores bastante alineados.

import numpy as np
a, b = np.array([1, 2, 2]), np.array([2, 1, 2])
print(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))   # 0.888...

Fallos de keyword: (1) "¿llevan comida a la casa?" — ninguna palabra coincide con la sección "Entregas a domicilio" (ni "llevan", ni "casa" aparecen en el título; el embedding las acerca por significado). (2) "¿puedo pagar con tarjeta?" vs el texto "aceptamos tarjeta de débito y crédito" — keyword encuentra "tarjeta", pero "¿aceptan plástico?" o "¿solo cash?" ya no matchean nada.

Caso donde keyword basta: "Esquina de Oro" — un nombre propio único en el corpus; la búsqueda exacta lo encuentra perfecto. Por eso los sistemas serios suelen ser híbridos: semántica + keyword (BM25), combinando lo mejor de ambos.

El índice se construye al arrancar el servidor: los embeddings en memoria son una foto del documento en ese momento. Editar el archivo no toca la foto.

def indexar(self, ruta_documento: str) -> None:
    self.ruta = ruta_documento
    self.mtime = Path(ruta_documento).stat().st_mtime
    # ... resto igual ...

def buscar(self, pregunta: str, k: int = 3, umbral: float = 0.30) -> list[dict]:
    if Path(self.ruta).stat().st_mtime != self.mtime:
        self.indexar(self.ruta)          # re-indexa en caliente
    # ... resto igual ...

Para un documento chico, re-indexar tarda < 1 segundo y la verificación de mtime es gratis. Con corpus grandes se re-indexa solo lo cambiado (por hash de chunk) — mismo principio, granularidad más fina.

Contraejemplo: "¿me prestan el local para una fiesta?" — no hay NADA sobre alquiler del local en las políticas. El top-3 sin umbral igual devuelve los 3 chunks "menos lejanos" (probablemente Encargos para eventos por la palabra "eventos", Horarios...). El modelo recibe contexto que parece relacionado y es propenso a hilvanar una respuesta: "para eventos pedimos 24 horas de anticipación y un adelanto del 50%" — técnicamente citado ¡pero respondiendo una pregunta que el documento no responde!

top-k responde "¿cuáles son los k más cercanos?"; el umbral responde "¿alguno es suficientemente cercano?". Son preguntas distintas y necesitás ambas: k acota el costo, el umbral protege la honestidad. Sin umbral, RAG convierte "no sé" en "alucinación con cita".

from rag import IndiceRAG

CASOS = [
    ("¿hacen delivery?", "Entregas a domicilio"),
    ("¿llevan hasta mi casa?", "Entregas a domicilio"),
    ("¿aceptan tarjeta?", "Formas de pago"),
    ("¿puedo transferir?", "Formas de pago"),
    ("quiero 80 pupusas para una fiesta", "Encargos para eventos"),
    ("mi pedido vino malo", "Reclamos y devoluciones"),
    ("¿devuelven el dinero?", "Reclamos y devoluciones"),
    ("¿cómo gano pupusas gratis?", "Cliente frecuente"),
    ("soy alérgico al queso", "Higiene y alérgenos"),
    ("¿el aceite es de origen animal?", "Higiene y alérgenos"),
]

indice = IndiceRAG()
indice.indexar("politicas_la_esquina.md")
aciertos = 0
for pregunta, esperada in CASOS:
    fuentes = [r["fuente"] for r in indice.buscar(pregunta, k=3, umbral=0.0)]
    ok = esperada in fuentes
    aciertos += ok
    print(f"{'✅' if ok else '❌'} {pregunta} → {fuentes}")
print(f"\nrecall@3 = {aciertos}/{len(CASOS)} = {aciertos/len(CASOS):.0%}")

Recall típico: 9-10/10 con este corpus chico. La que puede fallar es una formulación muy indirecta ("¿el aceite es de origen animal?"). Arreglos sin cambiar el modelo: (1) enriquecer el chunk — añadir a cada sección 2-3 preguntas frecuentes parafraseadas dentro del propio documento ("¿El aceite es vegetal o animal?..."); (2) bajar el corte/incrementar k. El punto del ejercicio: la recuperación se evalúa con datos etiquetados, igual que cualquier modelo — y es la evaluación más barata y útil de todo el pipeline RAG.

def guardar_indice(self, ruta_base: str) -> None:
    np.save(f"{ruta_base}.npy", self.matriz)
    Path(f"{ruta_base}.json").write_text(
        json.dumps({"modelo": MODELO_EMBEDDINGS, "chunks": self.chunks},
                   ensure_ascii=False), encoding="utf-8")

def cargar_indice(self, ruta_base: str) -> None:
    datos = json.loads(Path(f"{ruta_base}.json").read_text(encoding="utf-8"))
    if datos["modelo"] != MODELO_EMBEDDINGS:
        raise ValueError("El índice fue creado con OTRO modelo de embeddings: "
                         "re-indexá desde cero.")
    self.chunks = datos["chunks"]
    self.matriz = np.load(f"{ruta_base}.npy")

La verificación crítica: que el modelo de embeddings sea el mismo. Vectores de modelos distintos viven en espacios distintos — compararlos con coseno da números sin significado (el bug más silencioso de RAG: nada explota, solo recupera basura). Guardar el nombre del modelo junto al índice es el seguro.

Cada vector: 384 × 4 bytes = 1,536 bytes (~1.5 KB). (a) 6 × 1.5 KB ≈ 9 KB — nada. (b) 1,000 × 1.5 KB ≈ 1.5 MB — trivial. (c) 12M chunks × 1.5 KB ≈ 18 GB — ya no cabe cómodo en RAM común, y el producto matriz-vector tarda segundos.

La frontera práctica de numpy en memoria está alrededor de unos pocos millones de vectores (unos GB). Más allá: bases vectoriales con índices aproximados (HNSW) tipo Chroma/pgvector/Qdrant, que sacrifican exactitud marginal por búsquedas en milisegundos. Para La Esquina, numpy sobra por años.

Con k=3 a veces entran las tres (el corpus es chico); con corpus reales, una pregunta compuesta diluye su embedding entre temas y pierde recall. Mejora (query decomposition):

def buscar_descompuesto(self, pregunta: str, cliente, k_por_sub: int = 2) -> list[dict]:
    resp = cliente.messages.create(
        model="claude-opus-4-8", max_tokens=200,
        system="Descomponé la pregunta en 1-3 preguntas simples e independientes. "
               "Respondé SOLO las preguntas, una por línea.",
        messages=[{"role": "user", "content": pregunta}],
    )
    subpreguntas = [l.strip() for l in resp.content[0].text.splitlines() if l.strip()]
    vistos, resultados = set(), []
    for sub in subpreguntas or [pregunta]:
        for r in self.buscar(sub, k=k_por_sub):
            if r["texto"] not in vistos:
                vistos.add(r["texto"])
                resultados.append(r)
    return resultados

Costo: una llamada extra por pregunta (solo vale la pena si detectás conjunciones, o como herramienta del ej. 4.4 del capítulo). Es la versión artesanal de lo que los frameworks llaman multi-query retrieval — y ahora sabés exactamente qué hace por dentro.

Verás líneas data: {"texto": "¡Hola"} separadas por líneas en blanco (el \n\n que delimita eventos SSE) y al final data: [FIN]. La opción -N (--no-buffer) hace que curl muestre cada byte al llegar en vez de esperar a que la respuesta se complete — el equivalente en curl del flush=True de Python. Sin -N el stream "funciona" pero lo ves todo junto al final, exactamente como el bot sin flush del capítulo 2.

Con ids secuenciales, cualquiera puede adivinar las sesiones ajenas: mandar mensajes con session_id="sesion-7" y leer/continuar la conversación de otro cliente (con su nombre y su pedido — fuga de datos personales) o quemar su presupuesto. secrets.token_urlsafe(16) genera 16 bytes de aleatoriedad criptográfica (~10^38 posibilidades): inadivinable en la práctica.

random.randint NO basta: el módulo random (Mersenne Twister) es predecible — observando algunas salidas se puede reconstruir el estado y predecir las siguientes. Para todo lo que sea seguridad (tokens, ids de sesión, códigos), siempre secrets, nunca random.

La sección maliciosa se recupera (la palabra "promo" le da similitud alta) y viaja en <contexto>. Según el día, el modelo puede: ignorarla (el system prompt dice que el contexto son datos), mencionarla con sospecha, o — peor caso — ofrecer el "regalo". Es la inyección indirecta: instrucciones maliciosas dentro de los datos.

La capa que evita el daño real es la validación en el servidor (capa 4): registrar_pedido calcula el total desde MENU con Python — no existe ningún camino por el cual el texto del contexto pueda poner el total en $0. El peor resultado es una promesa incumplible en el chat, no un pedido gratis registrado. Lección doble: (1) sanitizá/controlá quién puede editar los documentos del RAG (¡también son superficie de ataque!), (2) las acciones con dinero se validan en código, siempre.

import time

def con_reintentos(generador_factory, max_intentos=2):
    """Reintenta SOLO si el stream fallo antes del primer fragmento."""
    for intento in range(max_intentos + 1):
        produjo_algo = False
        try:
            for fragmento in generador_factory():
                produjo_algo = True
                yield fragmento
            return                              # termino bien
        except Exception:
            if produjo_algo or intento == max_intentos:
                raise                           # no reintentar: ver abajo
            time.sleep(2 ** intento)

No se reintenta tras producir texto porque el usuario ya vio media respuesta: reiniciar generaría una respuesta distinta pegada a la anterior (las respuestas no son deterministas), un Frankenstein confuso. Además el historial quedaría ambiguo. Lo honesto es cortar con un mensaje de error y dejar que el usuario repita. Reintentar solo es seguro cuando el fallo fue antes de cualquier efecto visible — el mismo principio que las transacciones de bases de datos.

import os, secrets
from datetime import date

PREGUNTAS_SIN_RESPUESTA: list[dict] = []   # llenar en _con_contexto_rag

# En _con_contexto_rag, cuando recuperados == []:
#   PREGUNTAS_SIN_RESPUESTA.append({"ts": time.strftime("%H:%M"), "q": mensaje})

@app.get("/admin/resumen")
def admin_resumen():
    token = request.headers.get("X-Admin-Token", "")
    esperado = os.environ.get("ADMIN_TOKEN", "")
    if not (esperado and secrets.compare_digest(token, esperado)):
        return jsonify({"error": "No autorizado"}), 401

    hoy = date.today().isoformat()
    ruta = Path("pedidos.json")
    pedidos = json.loads(ruta.read_text(encoding="utf-8")) if ruta.exists() else []
    pedidos_hoy = [p for p in pedidos if p["registrado_en"].startswith(hoy)]

    return jsonify({
        "pedidos_hoy": len(pedidos_hoy),
        "venta_hoy_usd": round(sum(p["total"] for p in pedidos_hoy), 2),
        "sesiones_activas": len(SESIONES),
        "gasto_api_hoy_usd": round(GASTO_DIARIO["usd"], 4),
        "preguntas_sin_respuesta": PREGUNTAS_SIN_RESPUESTA[-5:],
    })

Detalles que valen puntos: secrets.compare_digest evita ataques de temporización en la comparación del token; el token vive en variable de entorno como la API key; y el chequeo esperado and ... evita que un ADMIN_TOKEN no configurado (vacío) deje el endpoint abierto. Las preguntas sin respuesta son oro: son la lista de qué agregar al documento de políticas la próxima semana.

Borrador de respuesta: "Doña Carmen: el bot es excelente para conversar — entender pedidos enredados, responder políticas — porque ahí un error cuesta una disculpa. La planilla y los impuestos son lo contrario: cálculos con respuesta única y exacta (determinismo) donde un error cuesta dinero y problemas legales (costo de error alto), y donde hay que poder demostrar exactamente cómo se calculó cada cifra (auditabilidad — un LLM no puede garantizar eso). Para eso lo correcto es un sistema tradicional: una hoja de cálculo validada o un software de planilla, donde cada fórmula es verificable. Lo que SÍ puede hacer el bot es lo conversacional alrededor: explicarle a un empleado dónde consultar su boleta, con la información oficial vía RAG — pero el número lo calcula un programa determinista, jamás el modelo."

Si en el capítulo 3 entendiste por qué calcular_pedido existe, esta respuesta es la misma idea a escala: el LLM redacta; el código calcula; y las decisiones de alto riesgo llevan humano.

import time

TTL_SESION = 30 * 60   # segundos

def limpiar_sesiones() -> None:
    ahora = time.time()
    muertas = [sid for sid, s in SESIONES.items()
               if ahora - s.get("ultimo_uso", ahora) > TTL_SESION]
    for sid in muertas:
        del SESIONES[sid]

En /chat, al validar la sesión: sesion["ultimo_uso"] = time.time(), y disparar limpiar_sesiones() al inicio de /session (cada sesión nueva limpia las viejas — barrido "gratis" sin hilos ni schedulers, suficiente para este tamaño). Con Redis esto sale gratis de verdad: EXPIRE session:<id> 1800 y la base se encarga.

@app.get("/health")
def health():
    return jsonify({"status": "ok",
                    "sesiones": len(SESIONES),
                    "gasto_hoy": round(GASTO_DIARIO["usd"], 4)})

Si el health check llamara a la API: (1) costo — las plataformas lo golpean cada 10-30 segundos: ~3,000-9,000 llamadas diarias pagadas para nada; (2) falsos negativos — un 529 temporal de Anthropic haría que la plataforma considere TU app muerta y la reinicie en bucle. El health check responde "mi proceso vive"; la salud de las dependencias externas se monitorea aparte y con menos frecuencia.