Agente de larga duración con manejo de context window

Un agente real no termina en el turno 3. Corre por horas, acumulando decisiones y descubrimientos. Esta receta hace correr un agente Claude Sonnet a través de una sesión larga y muestra los tres knobs que iLAB Memory te da para mantener el context window sano.

Lo que necesitás

Python 3.11+

asyncio de la stdlib para el loop del agente.

ilab-memory

pip install ilab-memory

anthropic

pip install anthropic — SDK oficial con cliente async.

ANTHROPIC_API_KEY

Exportada en la shell o tu cargador de .env favorito.

Arquitectura

 ┌──────────────┐
 │ loop agente  │  N turnos
 │ (50+)        │
 └──────┬───────┘
        │
        │  cada turno:                cada 20 turnos:
        │  cargá top-K decisions/     mem_session_summary
        │  patterns + contexto recent → observación "summary"
        │
        ▼
 ┌────────────────┐    ┌────────────────────────────┐
 │  ILabMemory    │ ─▶ │  Claude Sonnet (anthropic) │
 │  (./agent.db)  │    └────────────────────────────┘
 └────────────────┘

Tres knobs:

1. mem_search con types=[...] — cargá solo memoria arquitectónica para el prompt.
2. mem_session_summary cada N turnos — colapsá el historial en una observación summary.
3. topic_key — deduplicá hechos repetidos (sin explosión de filas).

Implementación

1. Abrí UNA sesión para todo el run

No abras/cerres por turno. Una sesión representa la "tarea lógica del usuario", no una llamada individual al LLM. La misma sesión se reusa en cada mem_session_start hasta que pase el timeout.

mem = ILabMemory.from_path("./agent.db")
session = mem.mem_session_start(user_id="agent-001")
print(session.is_new, session.session_id)

2. Cargá solo contexto de alta señal

Usá mem_search con types=['decision', 'pattern'] para saltarte chitchat (discovery) cuando armás el prompt. Esto mantiene el system prompt chico incluso después de miles de turnos.

architectural = mem.mem_search(
    user_id="agent-001",
    query="decisiones y patrones recientes",
    limit=10,
    types=["decision", "pattern"],
)

nota

El score que devuelve es SearchScore — combinando ranking FTS, recency y revisión. No es comparable con ContextScore de mem_session_start.memories. Ver Scoring para las fórmulas.

3. Corré el turno del agente

Componer, llamar a Claude, capturar la salida. Patrón async estándar.

from anthropic import AsyncAnthropic

claude = AsyncAnthropic()

msg = await claude.messages.create(
    model="claude-sonnet-4-5",
    max_tokens=1024,
    system=system_prompt,
    messages=[{"role": "user", "content": user_input}],
)
reply = "".join(b.text for b in msg.content if b.type == "text")

4. Guardá con topic_key para dedup natural

Si el usuario repite "prefiero modo oscuro", topic_key="user/agent-001/preferences/theme" asegura que se actualiza la misma fila, no se duplica. El campo outcome te dice qué pasó.

result = mem.mem_save(
    user_id="agent-001",
    type="preference",
    title="Preferencia de tema",
    content="El usuario prefiere modo oscuro.",
    topic_key="user/agent-001/preferences/theme",
)
print(result.outcome)  # 'created' la primera vez, 'deduped' o 'updated' después

5. Resumí cada N turnos

La summarization periódica comprime historial largo en una observación summary que sesiones futuras hidratan barato. Notá la API dedicada mem_session_summary — mem_save rechaza type='summary' (reservado).

if turn_idx % 20 == 0 and turn_idx > 0:
    mem.mem_session_summary(
        user_id="agent-001",
        summary=f"Últimos 20 turnos: {brief_recap}",
    )

tip

mem_session_summary no cierra la sesión. Solo agrega una observación summary y toca last_activity_at. La sesión sigue.

6. Cerrá la sesión cuando termines de verdad

mem_session_end escribe el resumen final autorado por humano. El próximo mem_session_start para el mismo user_id creará una sesión nueva.

closed = mem.mem_session_end(
    user_id="agent-001",
    summary="Tarea de onboarding completada: 50 turnos, 12 decisiones capturadas.",
)
print(closed.is_auto_generated)  # False — resumen humano

aviso

Si te olvidás de cerrar, el próximo mem_session_start después del timeout de 24 hs auto-cerrará la sesión vieja con un resumen sintético (is_auto_generated=True). Eso es un fallback, no una feature — cerrá manualmente cuando puedas.

Archivo completo

Código completo (listo para copiar y pegar)

from __future__ import annotations

import asyncio
import os
from typing import Iterable

from anthropic import AsyncAnthropic
from ilab_memory import ILabMemory

DB_PATH = os.environ.get("ILAB_MEMORY_DB_PATH", "./agent.db")
USER_ID = "agent-001"
MODEL = "claude-sonnet-4-5"
SUMMARY_EVERY = 20
TOTAL_TURNS = 50


def compose_prompt(memories: Iterable, architectural: Iterable) -> str:
    recent = "\n".join(f"- ({m.type}) {m.title}" for m in memories) or "(ninguna)"
    decisions = "\n".join(f"* {d.title}" for d in architectural) or "(ninguna)"
    return (
        "Sos un agente de planning de larga duración. Sé conciso.\n"
        f"## Contexto reciente\n{recent}\n\n"
        f"## Decisiones / patrones arquitectónicos\n{decisions}"
    )


async def run_turn(
    claude: AsyncAnthropic,
    mem: ILabMemory,
    user_input: str,
) -> str:
    session = mem.mem_session_start(user_id=USER_ID)
    architectural = mem.mem_search(
        user_id=USER_ID,
        query="decisiones patrones",
        limit=10,
        types=["decision", "pattern"],
    )
    system_prompt = compose_prompt(session.memories, architectural)

    msg = await claude.messages.create(
        model=MODEL,
        max_tokens=1024,
        system=system_prompt,
        messages=[{"role": "user", "content": user_input}],
    )
    reply = "".join(b.text for b in msg.content if b.type == "text")

    # Persistir el turno — type='discovery', sin topic_key (una fila por turno)
    mem.mem_save(
        user_id=USER_ID,
        type="discovery",
        title=f"Turno: {user_input[:60]}",
        content=f"Usuario: {user_input}\nAsistente: {reply}",
    )
    return reply


async def main() -> None:
    claude = AsyncAnthropic()
    with ILabMemory.from_path(DB_PATH) as mem:
        for turn_idx in range(TOTAL_TURNS):
            user_input = f"(turno simulado {turn_idx}) planificá próximo paso"
            reply = await run_turn(claude, mem, user_input)
            print(f"[{turn_idx:02d}] {reply[:80]}")

            # Compresión periódica — mantiene barata la hidratación de la próxima sesión
            if turn_idx % SUMMARY_EVERY == 0 and turn_idx > 0:
                mem.mem_session_summary(
                    user_id=USER_ID,
                    summary=f"Checkpoint en turno {turn_idx}: agente en track.",
                )

        mem.mem_session_end(
            user_id=USER_ID,
            summary=f"Run completado: {TOTAL_TURNS} turnos.",
        )


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

¿Y ahora?

Aislamiento multi-usuario

El mismo loop de agente, muchos tenants — user_id es la isolation key.

Scoring

Por qué SearchScore y ContextScore no son comparables, y qué optimiza cada uno.

Topic keys

El primitivo de dedup — cuándo setear uno y cuándo saltarlo.

FastAPI + OpenAI

Los mismos primitivos de memoria envueltos en un servicio HTTP.