Condiciones de parada y presupuesto

Al terminar esta lección tu agente sabrá parar por sí mismo antes de gastar demasiado, tardar demasiado o hacer algo que no debe. Añadirás un presupuesto que se comprueba en cada vuelta del bucle y una puerta de aprobación para las acciones con consecuencias.

La idea

En la lección anterior el único freno era max_pasos. En producción no basta. Un agente puede gastar muchos tokens en pocos pasos si una herramienta devuelve mucho texto, quedarse atascado repitiendo la misma búsqueda o intentar una acción irreversible. Un agente bien hecho tiene varias condiciones de parada, cada una para un riesgo distinto:

flowchart TD
  P{¿Queda presupuesto?} -- no --> C[Cierre con resultado parcial]
  P -- sí --> M[Llamar al modelo]
  M --> S{stop_reason}
  S -- end_turn --> F[Final]
  S -- tool_use --> A{¿Requiere aprobación?}
  A -- sí --> H[Preguntar al humano]
  A -- no --> E[Ejecutar]
  H --> E
  E --> P

Por qué los tokens crecen tan rápido

Cada llamada reenvía todo el historial. Si el paso 1 envía 2.000 tokens y cada herramienta añade 1.500, el paso 10 envía más de 15.000 tokens solo de entrada, y la suma de todos los pasos es mucho mayor que la última llamada. El coste de un agente crece más deprisa que su número de pasos. Por eso conviene limitar tokens acumulados, no solo pasos, y recortar las salidas de las herramientas. El prompt caching abarata la parte repetida del historial; lo verás en Control de costes y prompt caching.

resp.usage trae lo que ha costado cada llamada. Ojo: input_tokens no incluye los tokens leídos o escritos en caché, que vienen aparte en cache_read_input_tokens y cache_creation_input_tokens. Para saber cuánto contexto procesaste, súmalos.

Un presupuesto en código

import json
import time
from dataclasses import dataclass, field

@dataclass
class Presupuesto:
    max_pasos: int = 15
    max_tokens_totales: int = 150_000
    max_segundos: float = 120.0
    max_repeticiones: int = 2
    pasos: int = 0
    tokens: int = 0
    inicio: float = field(default_factory=time.monotonic)
    vistas: dict = field(default_factory=dict)

    def registrar(self, usage) -> None:
        self.pasos += 1
        self.tokens += (
            usage.input_tokens + usage.output_tokens
            + (getattr(usage, "cache_read_input_tokens", 0) or 0)
            + (getattr(usage, "cache_creation_input_tokens", 0) or 0)
        )

    def repetida(self, nombre: str, args: dict) -> bool:
        clave = (nombre, json.dumps(args, sort_keys=True))
        self.vistas[clave] = self.vistas.get(clave, 0) + 1
        return self.vistas[clave] > self.max_repeticiones

    def agotado(self) -> str | None:
        if self.pasos >= self.max_pasos:
            return f"límite de pasos ({self.max_pasos})"
        if self.tokens >= self.max_tokens_totales:
            return f"límite de tokens ({self.tokens} de {self.max_tokens_totales})"
        if time.monotonic() - self.inicio >= self.max_segundos:
            return f"límite de tiempo ({self.max_segundos:.0f} s)"
        return None

El tiempo se comprueba entre pasos, así que una llamada lenta puede pasarse del límite. Configura además el tiempo de espera de red de tu cliente HTTP; el SDK tiene opciones para ello, consúltalas en su documentación.

Aprobación humana sin romper el historial

Algunas herramientas no deberían ejecutarse sin que alguien lo confirme: enviar un correo, emitir un reembolso, borrar datos. La clave es que si el humano dice que no, el modelo igualmente recibe un tool_result. Si no, el historial queda con un tool_use huérfano y la siguiente llamada falla.

REQUIERE_APROBACION = {"abrir_incidencia", "emitir_reembolso"}

def ejecutar_con_control(bloque, presupuesto) -> dict:
    base = {"type": "tool_result", "tool_use_id": bloque.id}
    if presupuesto.repetida(bloque.name, bloque.input):
        return base | {"is_error": True, "content":
            "Ya has hecho esta misma llamada varias veces con los mismos argumentos. "
            "Cambia de enfoque o responde con lo que tienes."}
    if bloque.name in REQUIERE_APROBACION:
        print(f"El agente quiere ejecutar {bloque.name} con {bloque.input}")
        if input("¿Apruebas? [s/N] ").strip().lower() != "s":
            return base | {"is_error": True, "content":
                "El usuario ha rechazado esta acción. No la repitas; explica qué harías en su lugar."}
    try:
        return base | {"content": ejecutar(bloque.name, bloque.input), "is_error": False}
    except Exception as e:
        return base | {"content": f"Error: {e}", "is_error": True}

En una aplicación real, la pregunta no será un input(): guardarás el estado, notificarás a alguien y reanudarás cuando conteste. Lo desarrolla Aprobación humana antes de actuar.

El bucle con presupuesto

def agente(pregunta: str, presupuesto: Presupuesto) -> dict:
    messages = [{"role": "user", "content": pregunta}]
    while True:
        motivo = presupuesto.agotado()
        if motivo:
            return {"estado": "incompleto", "motivo": motivo, "messages": messages}
        resp = client.messages.create(model=MODEL, max_tokens=2048, system=SYSTEM,
                                      tools=TOOLS, messages=messages)
        presupuesto.registrar(resp.usage)
        messages.append({"role": "assistant", "content": resp.content})
        if resp.stop_reason == "end_turn":
            texto = "".join(b.text for b in resp.content if b.type == "text")
            return {"estado": "completo", "respuesta": texto, "messages": messages}
        if resp.stop_reason == "tool_use":
            resultados = [ejecutar_con_control(b, presupuesto) for b in resp.content if b.type == "tool_use"]
            messages.append({"role": "user", "content": resultados})
            continue
        if resp.stop_reason == "pause_turn":
            continue
        return {"estado": "error", "motivo": resp.stop_reason, "messages": messages}

Fíjate en que el agente devuelve un estado en lugar de lanzar excepciones. Quien lo llama decide qué hacer con un resultado incompleto: mostrarlo, reintentar con más presupuesto o escalar a una persona.

Qué hacer con max_tokens

max_tokens es un límite duro por llamada. Si se alcanza, el stop_reason es "max_tokens" y la respuesta está cortada. Si lo último era un bloque tool_use, sus argumentos pueden estar incompletos: no lo ejecutes. Las opciones razonables son repetir la llamada con un max_tokens mayor o parar con error. Si usas razonamiento, recuerda que el esfuerzo es orientativo y max_tokens es lo que de verdad limita la salida.

Cerrar con un resumen

Cuando el presupuesto se agota, a veces quieres que el modelo resuma lo que encontró. Puedes hacer una última llamada con tool_choice={"type": "none"} y un mensaje pidiendo el resumen. Antes, asegúrate de que el último mensaje del asistente no tiene tool_use sin responder: si lo tiene, añade sus tool_result (por ejemplo, con is_error: true y «Presupuesto agotado, no se ha ejecutado») y el texto de la petición de resumen después de ellos, en el mismo mensaje.

Errores frecuentes

• Contar solo pasos y descubrir el coste en la factura.
• Lanzar una excepción al agotar el presupuesto y perder todo lo que el agente encontró.
• Dejar un tool_use sin tool_result cuando el humano rechaza la acción.
• Poner límites tan bajos que el agente nunca termina tareas normales. Ajusta con datos de ejecuciones reales.

Ejercicio

Resumen

• Un agente necesita varias condiciones de parada: pasos, tokens, tiempo, repeticiones y aprobación.
• Los tokens acumulados crecen más deprisa que los pasos porque cada llamada reenvía el historial.
• La aprobación humana va en el código y siempre devuelve un tool_result, también si se rechaza.
• Un tool_use cortado por max_tokens no se ejecuta.
• Devuelve un estado con motivo en lugar de lanzar excepciones.

Para seguir

• Planificar antes de actuar
• Humano en el bucle
• Costes y límites
• Humano en el bucle