10 · Lo que el modelo ve en cada turno, y lo que cuesta

Esto es un fichero que se ejecuta: examples/agentes/10_economia_del_contexto.py ↗ Esta página lo transcribe y enseña lo que imprime. Si dejan de coincidir, falla un test.

Dominio: Argus — depurar un incidente mirando logs. La herramienta devuelve lo que devuelve: 150.000 caracteres de una ventana de cinco minutos. El modelo necesita saber qué hay dentro; no necesita releerlo en cada turno posterior, que es lo que ocurre si nadie lo impide.

Tres piezas que son la misma idea vista desde tres sitios:

Limits(max_tool_chars=…)   qué parte del resultado entra en el contexto
prefijo estable            qué **no** debe cambiar, para que haya caché
economy(estado)            qué costó el run y **por qué**

Ninguna de las tres inventa nada: si nadie midió la caché, el informe lo dice en vez de escribir un cero.

Cómo correrlo

uv run python examples/agentes/10_economia_del_contexto.py

No hace falta configurar nada: sin modelo, las respuestas van guionizadas y todo lo demás es real — las herramientas se ejecutan, el journal se escribe, el consumo se mide. Con AXONIUM_CLIENT_ID o SYNAPTUM_BASE_URL en el entorno, el mismo fichero sin tocar habla con un modelo de verdad; lo que cambia entonces es lo que diga el modelo, no el código. Ver Modelos.

Lo que imprime

10 · Economía del contexto
──────────────────────────
sin inferencia · respuestas guionizadas (exporta SYNAPTUM_BASE_URL para usar un modelo real)

  la herramienta devolvió      45.472 caracteres
  al modelo le llegan          16.000 caracteres

  y el recorte lo dice, en vez de terminar en mitad de una línea:

    [… recortado: 29.472 de 45.472 caracteres. Pide un rango concreto si necesitas lo que falta …]

  Cabeza **y** cola, no los primeros 16.000 caracteres: en un volcado
  de logs el final es donde está el fallo, y cortar por delante lo tira.
  El journal guarda el resultado entero pase lo que pase — esto solo
  decide qué se le **reenvía** al modelo en cada turno posterior.

  ── el mismo run, con el tope y sin él ──

    turno 1: sin tope       188  ·  con tope       188
    turno 2: sin tope    45.660  ·  con tope    16.188

  El turno 1 es idéntico: el volcado todavía no ha vuelto. La
  diferencia aparece en el 2 — y se pagaría **otra vez** en el 3, en el
  4 y en todos los demás, porque el historial se reenvía entero.

  ── el informe del run ──

Run con-tope · 2 turnos
  caché      0% de la entrada servida de caché
  crecimiento +0 tokens de entrada por turno
  prefijo    estable durante todo el run
  total      entrada=200 salida=40 caché=0

  Con el doble los números son los del guion —declara `cache_read=0`,
  un cero **medido**— así que lo que enseña el informe es su forma. Un
  proveedor que no reporte caché saldría «sin medir», que es distinto de
  cero y el informe no los confunde. Contra la plataforma real estas
  cifras son reales: exporta las credenciales y vuelve a correrlo.

  ── y el mismo agente, con la hora en las instrucciones ──

  huella del prefijo: 507a5687c0def2fb → 43f597f9f1e535b5
  qué cambió:         instrucciones de sistema: cambiaron

  Eso no dice «la huella no coincide», que no sirve para nada: dice
  qué arreglar. Y así es como sale en el informe de un run:

Run con-reloj · 2 turnos
  caché      0% de la entrada servida de caché
  crecimiento +0 tokens de entrada por turno
  ⚠ prefijo  reescrito 1 vez/veces — cada una tira toda la caché posterior
      · 000002-model: instrucciones de sistema: cambiaron
  total      entrada=200 salida=40 caché=0

  Nadie escribió un error y nada falló. Lo único que pasó es que la
  factura es más alta de lo que debería, y eso no se nota mirando un run.

from __future__ import annotations

import asyncio
import sys
from datetime import datetime
from pathlib import Path
from typing import Annotated

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from synaptum import (
    Agent,
    Limits,
    MemoryCheckpointer,
    Request,
    Role,
    Session,
    Text,
    ToolResult,
    cap_tool_output,
    economy,
    tool,
)
from synaptum.context import describe_prefix_change, prefix_fingerprint
from synaptum.testing import calls, says

from comun import encabezado, gateway, nombre_del_modelo

La herramienta que devuelve demasiado

No es un caso rebuscado. Un kubectl logs, un SELECT *, la respuesta de una API paginada que no pagina: lo normal es que una herramienta devuelva muchísimo más de lo que hacía falta.

LINEA = (
    "2026-09-18T02:14:{s:02d}.{ms:03d}Z  prometheus-gateway  INFO  "
    "req_id=8f3a{n:04d} POST /v1/chat/completions upstream=llama-4 "
    "model_load_ms={carga} total_ms={total} status=200\n"
)

#: Trescientas líneas son ~45.000 caracteres. Un `kubectl logs` de cinco
#: minutos devuelve bastante más; aquí se queda corto a propósito, para que el
#: ejemplo se pueda correr contra un modelo de verdad sin ocupar la plataforma
#: un minuto entero. La proporción es lo que importa, y no cambia.
LINEAS = 300

def _volcado() -> str:
    lineas = [
        LINEA.format(
            s=n % 60, ms=n % 1000, n=n,
            carga=5900 if n % 7 == 0 else 3,
            total=6400 if n % 7 == 0 else 510,
        )
        for n in range(LINEAS)
    ]
    return "".join(lineas)

@tool(idempotent=True)
async def leer_logs(
    servicio: Annotated[str, "Servicio del que leer"],
    ventana: Annotated[str, "Ventana temporal, por ejemplo '02:14-02:19'"],
) -> str:
    """Logs crudos de un servicio en una ventana de tiempo."""
    return _volcado()

1 · Qué entra en el contexto

def parte_uno() -> None:
    entero = ToolResult.of("call-1", _volcado())
    recortado = cap_tool_output(entero, max_chars=16_000)

    largo = len("".join(p.text for p in entero.content if isinstance(p, Text)))
    corto = len("".join(p.text for p in recortado.content if isinstance(p, Text)))

    print(f"  la herramienta devolvió   {largo:>9,} caracteres".replace(",", "."))
    print(f"  al modelo le llegan       {corto:>9,} caracteres".replace(",", "."))
    print("\n  y el recorte lo dice, en vez de terminar en mitad de una línea:\n")

    texto = "".join(p.text for p in recortado.content if isinstance(p, Text))
    marca = next(linea for linea in texto.splitlines() if linea.startswith("["))
    print(f"    {marca}")
    print(
        "\n  Cabeza **y** cola, no los primeros 16.000 caracteres: en un volcado\n"
        "  de logs el final es donde está el fallo, y cortar por delante lo tira.\n"
        "  El journal guarda el resultado entero pase lo que pase — esto solo\n"
        "  decide qué se le **reenvía** al modelo en cada turno posterior.\n"
    )

2 · Lo mismo, medido sobre un run

class Medido:
    """Cuenta los caracteres que salen hacia el modelo en cada turno.

    Envuelve el gateway que haya —el doble o el real— porque la medida tiene que
    ser la misma en los dos casos. Medir solo contra el doble sería medir el
    doble.
    """

    def __init__(self, interior) -> None:
        self._interior = interior
        self.por_turno: list[int] = []

    async def invoke_model(self, request, ctx):
        self.por_turno.append(_caracteres(request))
        return await self._interior.invoke_model(request, ctx)

    def stream_model(self, request, ctx):
        self.por_turno.append(_caracteres(request))
        return self._interior.stream_model(request, ctx)

    def __getattr__(self, nombre):
        return getattr(self._interior, nombre)

def _caracteres(request) -> int:
    """Todo el texto de una petición: sistema, mensajes y resultados."""
    total = len(request.system or "")
    for mensaje in request.messages:
        for parte in mensaje.content:
            if isinstance(parte, Text):
                total += len(parte.text)
            elif isinstance(parte, ToolResult):
                total += sum(
                    len(p.text) for p in parte.content if isinstance(p, Text)
                )
    return total

async def correr(*, tope: int | None, run_id: str) -> tuple[list[int], object]:
    agente = Agent(
        "depurador",
        model=nombre_del_modelo(),
        instructions=(
            "Depuras incidentes de una plataforma de inferencia. Lee los logs "
            "del servicio afectado y di en una frase dónde se va el tiempo."
        ),
        tools=[leer_logs],
        limits=Limits(max_tool_chars=tope),
    )

    puerta = Medido(gateway([_guion] * 4, tools=[leer_logs]))
    store = MemoryCheckpointer()
    sesion = Session(run_id, puerta, store)

    async for _ in agente.run(
        "¿Por qué el p95 del gateway pasó de 820ms a 6.4s a las 02:14?",
        session=sesion,
    ):
        pass
    return puerta.por_turno, await store.load(run_id)

def _guion(peticion):
    if any(m.role is Role.TOOL for m in peticion.messages):
        return says(
            "Una de cada siete peticiones recarga el modelo: 5.9s de los 6.4s "
            "son model_load. El resto responde en ~510ms. No es el gateway, es "
            "el desalojo de la caché de pesos."
        )
    return calls("leer_logs", servicio="prometheus-gateway", ventana="02:14-02:19")

3 · El prefijo que se rompe solo

class ConLaHoraDentro(Agent):
    """Un agente cuyas instrucciones llevan el reloj.

    Parece inofensivo y es de lo más común: dar la hora al modelo para que sepa
    qué es «ayer». El problema es que el prompt de sistema **abre** el prefijo,
    así que cambiarlo en cada turno invalida todo lo que viene detrás — y el
    síntoma es una caché que nunca arranca, no un error.
    """

    @property
    def instructions(self) -> str:
        return (
            "Depuras incidentes de una plataforma de inferencia. "
            f"Ahora son las {datetime.now():%H:%M:%S.%f}."
        )

    @instructions.setter
    def instructions(self, _valor) -> None:
        pass          # el constructor intenta asignarlas; aquí se ignoran

async def parte_tres() -> None:
    agente = ConLaHoraDentro(
        "depurador", model=nombre_del_modelo(), tools=[leer_logs]
    )

    # El mecanismo, sin depender de que el modelo haga nada: dos turnos del
    # mismo agente, y sus huellas.
    #
    # Las dos horas van escritas en vez de leer el reloj **para que este ejemplo
    # imprima siempre lo mismo**. En un run de verdad las pone `datetime.now()`
    # —como hace `ConLaHoraDentro` unas líneas más abajo— y cambian solas en
    # cada turno, que es justamente el problema.
    def turno(hora: str) -> Request:
        return Request(
            model=agente.model,
            system=f"Depuras incidentes de una plataforma de inferencia. Ahora son las {hora}.",
            tools=agente.tools,
        )

    antes, ahora = turno("02:14:07.100"), turno("02:14:07.842")

    print(f"  huella del prefijo: {prefix_fingerprint(antes)} → {prefix_fingerprint(ahora)}")
    print(f"  qué cambió:         {describe_prefix_change(antes, ahora)}")
    print(
        "\n  Eso no dice «la huella no coincide», que no sirve para nada: dice\n"
        "  qué arreglar. Y así es como sale en el informe de un run:\n"
    )

    store = MemoryCheckpointer()
    sesion = Session("con-reloj", gateway([_guion] * 4, tools=[leer_logs]), store)
    async for _ in agente.run("¿Qué pasó a las 02:14?", session=sesion):
        pass

    informe = economy(await store.load("con-reloj"))
    print(informe.report())

    if informe.prefix_rewrites:
        print(
            "\n  Nadie escribió un error y nada falló. Lo único que pasó es que la\n"
            "  factura es más alta de lo que debería, y eso no se nota mirando un run.\n"
        )
    else:
        print(
            "\n  Este run se resolvió en un solo turno —el modelo contestó sin usar\n"
            "  la herramienta— y con uno no hay nada con lo que comparar: el\n"
            "  informe no cuenta el estreno como una regresión. El daño aparece\n"
            "  cuando el run es largo, que es justo cuando la caché importaría.\n"
        )

Todo junto

async def main() -> None:
    encabezado("10 · Economía del contexto")

    parte_uno()

    print("  ── el mismo run, con el tope y sin él ──\n")
    sin_tope, _ = await correr(tope=None, run_id="sin-tope")
    con_tope, estado = await correr(tope=16_000, run_id="con-tope")

    for turno, (a, b) in enumerate(zip(sin_tope, con_tope), start=1):
        print(f"    turno {turno}: sin tope {a:>9,}  ·  con tope {b:>9,}".replace(",", "."))
    print(
        "\n  El turno 1 es idéntico: el volcado todavía no ha vuelto. La\n"
        "  diferencia aparece en el 2 — y se pagaría **otra vez** en el 3, en el\n"
        "  4 y en todos los demás, porque el historial se reenvía entero.\n"
    )

    print("  ── el informe del run ──\n")
    print(economy(estado).report())
    print(
        "\n  Con el doble los números son los del guion —declara `cache_read=0`,\n"
        "  un cero **medido**— así que lo que enseña el informe es su forma. Un\n"
        "  proveedor que no reporte caché saldría «sin medir», que es distinto de\n"
        "  cero y el informe no los confunde. Contra la plataforma real estas\n"
        "  cifras son reales: exporta las credenciales y vuelve a correrlo.\n"
    )

    print("  ── y el mismo agente, con la hora en las instrucciones ──\n")
    await parte_tres()

if __name__ == "__main__":
    asyncio.run(main())

Lo que esto enseña

Por qué esto no es una optimización, dicho una vez:

No recortar falla en silencio. Con una ventana pequeña el run revienta y te enteras; con una grande solo cuesta dinero en cada turno posterior, que es peor, porque nadie lo mira. Por eso max_tool_chars viene activado.

Reescribir el prefijo tampoco avisa. economy() existe porque el síntoma —una caché que no arranca— solo se ve agregando los turnos, y porque «la huella no coincide» no sirve para nada: lo accionable es «las instrucciones de sistema cambiaron».

Y la medida: cache_read es lo que hace correcta la decisión de compactar. Sin saber cuánto se está sirviendo de caché, conservar el historial parece caro cuando a menudo es casi gratis, y resumirlo parece barato cuando cuesta una inferencia y pierde información.

El fichero entero, para clonarlo y tocarlo: examples/agentes/10_economia_del_contexto.py ↗

Está en examples/ con los otros quince, y todos corren igual.