Tu agente IA olvida todo despues de 50 mensajes. Asi se soluciona.

Ejecuta un agente de Pydantic AI durante 50 mensajes y preguntale sobre algo del mensaje #3. No lo recordara. No porque el modelo sea malo - sino porque la conversacion se hizo demasiado larga, la ventana de contexto se lleno y los mensajes tempranos se perdieron.

Este no es un problema del modelo. Es un problema de infraestructura. Y empeora con agentes que usan herramientas, porque cada llamada a herramienta y respuesta es un mensaje. Una conversacion de 20 turnos con 3 llamadas a herramientas por turno son en realidad 80+ mensajes. A ~4 caracteres por token, son decenas de miles de tokens consumidos antes de que te des cuenta.

TL;DR

• El desbordamiento de contexto es un problema de infraestructura, no del modelo. Todo agente de larga ejecucion necesita una estrategia de gestion de contexto.
• SummarizationProcessor usa un LLM para comprimir inteligentemente los mensajes antiguos en un resumen - mayor calidad, pero cuesta una llamada API por trigger.
• SlidingWindowProcessor simplemente recorta los mensajes antiguos sin costo - rapido pero pierde el contexto completamente.
• La preservacion de pares de tool call es critica. Ambos procesadores garantizan que las llamadas a herramientas y sus respuestas nunca se separen.
• Los triggers basados en fraccion (("fraction", 0.8)) son la opcion mas portable - se adaptan automaticamente a la ventana de contexto de cualquier modelo.

Nos topamos con este muro en cada despliegue de agentes de larga ejecucion. Bots de soporte al cliente que olvidan el nombre del cliente. Asistentes de codigo que pierden la pista de que archivos ya han modificado. Agentes de investigacion que reinvestigan temas que cubrieron hace 30 mensajes.

Por eso construimos summarization-pydantic-ai - dos procesadores que gestionan el historial de conversacion de tu agente: uno que resume inteligentemente usando un LLM, y otro que simplemente recorta mensajes antiguos sin costo.

Dos procesadores, dos compromisos

Ambos funcionan como history processors de Pydantic AI - funciones drop-in que transforman el historial de mensajes antes de cada ejecucion del agente.

SummarizationProcessor: Compresion inteligente

from pydantic_ai import Agent
from pydantic_ai_summarization import create_summarization_processor

processor = create_summarization_processor(
    model="openai:gpt-4.1",
    trigger=("tokens", 100000),  # Trigger at 100k tokens
    keep=("messages", 20),       # Keep last 20 messages
)

agent = Agent(
    "openai:gpt-4.1",
    history_processors=[processor],
)

Cuando el conteo de tokens excede el trigger, el procesador:

1. Calcula un punto de corte seguro (nunca separando pares de tool call)
2. Envia los mensajes mas antiguos a un LLM para resumen
3. Reemplaza los mensajes antiguos con un resumen compacto
4. Mantiene los ultimos 20 mensajes intactos

El resultado: tu agente mantiene el contexto del inicio de la conversacion sin consumir todo el presupuesto de tokens.

Tres tipos de trigger

Puedes activar la sumarizacion basandote en mensajes, tokens o fraccion de contexto:

Basado en mensajes - simple y predecible:

processor = SummarizationProcessor(
    model="openai:gpt-4.1",
    trigger=("messages", 50),  # After 50 messages
)

Basado en tokens - tiene en cuenta la longitud del mensaje:

processor = SummarizationProcessor(
    model="openai:gpt-4.1",
    trigger=("tokens", 100000),  # After 100k tokens
)

Basado en fraccion - se adapta al contexto de cualquier modelo:

processor = SummarizationProcessor(
    model="openai:gpt-4.1",
    trigger=("fraction", 0.8),    # At 80% of context
    max_input_tokens=128000,       # GPT-4o's context window
)

Multiples triggers - logica OR, el primero gana:

processor = SummarizationProcessor(
    model="openai:gpt-4.1",
    trigger=[
        ("messages", 50),     # OR
        ("tokens", 100000),   # OR
        ("fraction", 0.8),
    ],
    max_input_tokens=128000,
)

Los triggers basados en fraccion son los mas robustos para produccion - se adaptan automaticamente cuando cambias entre modelos con diferentes ventanas de contexto.

Preservacion de pares de Tool Call

Este es el detalle que la mayoria de las soluciones de gestion de contexto hacen mal. Considera esta secuencia de mensajes:

User: "Search for Python tutorials"
Assistant: [tool_call: search("Python tutorials"), id=call_1]
Tool: [tool_return: "Found 5 results...", id=call_1]
Assistant: "Here are the top results..."
User: "Tell me more about the first one"

Si cortas entre la llamada a herramienta y su retorno, el modelo ve un tool call huerfano - y se rompe. Nuestros procesadores manejan esto:

def _is_safe_cutoff_point(self, messages, cutoff_index):
    """Check if cutting at index would separate tool call/response pairs."""
    search_start = max(0, cutoff_index - 5)
    search_end = min(len(messages), cutoff_index + 5)

    for i in range(search_start, search_end):
        msg = messages[i]
        if not isinstance(msg, ModelResponse):
            continue

        tool_call_ids = set()
        for part in msg.parts:
            if isinstance(part, ToolCallPart) and part.tool_call_id:
                tool_call_ids.add(part.tool_call_id)

        if not tool_call_ids:
            continue

        # Check if cutoff separates this tool call from its response
        for j in range(i + 1, len(messages)):
            check_msg = messages[j]
            if isinstance(check_msg, ModelRequest):
                for part in check_msg.parts:
                    if (isinstance(part, ToolReturnPart)
                        and part.tool_call_id in tool_call_ids):
                        tool_before = i < cutoff_index
                        response_before = j < cutoff_index
                        if tool_before != response_before:
                            return False  # Unsafe - would split pair
    return True

El procesador busca 5 mensajes en cada direccion desde el punto de corte, encuentra todos los pares tool call/return y se asegura de que permanezcan en el mismo lado del corte.

SlidingWindowProcessor: Recorte sin costo

Cuando no necesitas sumarizacion inteligente - cuando la velocidad y el costo importan mas que la calidad del contexto:

from pydantic_ai_summarization import create_sliding_window_processor

processor = create_sliding_window_processor(
    trigger=("messages", 100),  # Trim at 100 messages
    keep=("messages", 50),      # Keep last 50
)

agent = Agent(
    "openai:gpt-4.1",
    history_processors=[processor],
)

Sin llamada LLM. Sin latencia. Los mensajes antiguos simplemente se descartan. La misma preservacion de pares de tool call aplica - la ventana deslizante no cortara en medio de una secuencia de tool call.

Conteo de tokens: Aproximado por defecto

El contador de tokens por defecto usa una heuristica simple - ~4 caracteres por token:

def count_tokens_approximately(messages):
    total_chars = 0
    for msg in messages:
        if isinstance(msg, ModelRequest):
            for part in msg.parts:
                if isinstance(part, UserPromptPart):
                    if isinstance(part.content, str):
                        total_chars += len(part.content)
                elif isinstance(part, SystemPromptPart):
                    total_chars += len(part.content)
                elif isinstance(part, ToolReturnPart):
                    total_chars += len(str(part.content))
        elif isinstance(msg, ModelResponse):
            for part in msg.parts:
                if isinstance(part, TextPart):
                    total_chars += len(part.content)
                elif isinstance(part, ToolCallPart):
                    total_chars += len(part.tool_name)
                    total_chars += len(str(part.args))
    return total_chars // 4

Esto es rapido y suficientemente bueno para la mayoria de los casos. Para precision, conecta tiktoken:

import tiktoken

def accurate_counter(messages):
    encoding = tiktoken.encoding_for_model("gpt-4")
    total = 0
    for msg in messages:
        total += len(encoding.encode(str(msg)))
    return total

processor = create_summarization_processor(
    token_counter=accurate_counter,
    trigger=("tokens", 100000),
)

Prompts de resumen personalizados

El prompt de resumen por defecto extrae contexto clave. Puedes personalizarlo:

processor = create_summarization_processor(
    summary_prompt="""
    You are summarizing an agent conversation. Extract:

    1. **Key Decisions**: What was decided?
    2. **Code Changes**: What code was written/modified?
    3. **Pending Tasks**: What still needs to be done?
    4. **Important Context**: What context is crucial to preserve?

    Conversation to summarize:
    {messages}

    Provide a concise summary that preserves essential information.
    """,
)

El placeholder {messages} se reemplaza con el historial de mensajes formateado. Personaliza esto para tu dominio - un agente de soporte al cliente podria extraer el nombre del cliente y el problema, mientras que un asistente de codigo podria enfocarse en los cambios de archivos y resultados de tests.

Patron de bucle de conversacion

Aqui esta el patron de produccion tipico:

from pydantic_ai import Agent
from pydantic_ai_summarization import create_summarization_processor

processor = create_summarization_processor(
    trigger=("messages", 20),
    keep=("messages", 5),
)

agent = Agent(
    "openai:gpt-4.1",
    history_processors=[processor],
)

async def chat():
    message_history = []

    while True:
        user_input = input("You: ")
        if user_input.lower() == "quit":
            break

        result = await agent.run(
            user_input,
            message_history=message_history,
        )

        print(f"Assistant: {result.output}")
        message_history = result.all_messages()

El history processor se ejecuta automaticamente antes de cada llamada a agent.run(). Pasas el historial completo de mensajes, el procesador verifica si se necesita sumarizacion y devuelve un historial (posiblemente comprimido). Sin conteo manual de tokens.

Conclusiones clave

• El desbordamiento de contexto es un problema de infraestructura, no del modelo. Todo agente de larga ejecucion necesita una estrategia de gestion de contexto. Elige entre sumarizacion inteligente (mayor calidad, costo LLM) y ventana deslizante (cero costo, perdida de contexto).
• La preservacion de pares de tool call es critica. Nunca cortes entre una llamada a herramienta y su respuesta. Ambos procesadores manejan esto automaticamente - buscan 5 mensajes en cada direccion para pares huerfanos.
• Los triggers basados en fraccion son los mas portables. ("fraction", 0.8) funciona independientemente del modelo que uses o de lo largos que sean tus mensajes. Los conteos de tokens y mensajes son especificos del modelo y del contenido.
• La heuristica de ~4 caracteres/token es suficientemente buena. Para precision en produccion, usa tiktoken. Para todo lo demas, el contador por defecto mantiene las cosas rapidas sin agregar una dependencia.
• Personaliza el prompt de resumen para tu dominio. El predeterminado extrae contexto generico. Un prompt especifico del dominio (“extraer nombre del cliente, ID del ticket y estado de resolucion”) produce resumenes mucho mejores.

Pruebalo tu mismo

summarization-pydantic-ai - Sumarizacion automatica de conversaciones y gestion de contexto para agentes de Pydantic AI.

pip install summarization-pydantic-ai