Salta al contenuto principale

87 - Memoria conversazionale: sessioni, finestra di contesto e riassunto

La memoria di un agente è una lista rimandata a ogni giro. Costo, sessioni, troncamento, riassunto e recupero, con codice eseguito su un modello finto.

Cosa ne pensi di questo articolo?

87 - Memoria conversazionale: sessioni, finestra di contesto e riassunto
Machine Learning Engineerfedericocalo.dev

In Breve

La memoria di un agente è una lista rimandata a ogni giro. Costo, sessioni, troncamento, riassunto e recupero, con codice eseguito su un modello finto.

Un modello linguistico non ricorda nulla: ogni chiamata è senza stato. La «memoria» di un assistente è una lista di messaggi che tu rimandi al modello ogni volta, per intero. Da questa unica frase derivano tre fatti che governano il progetto di qualunque agente conversazionale: il costo cresce in modo quadratico nel numero di turni, la finestra di contesto è un tetto fisico, e tutto ciò che vuoi far «ricordare» oltre quel tetto va ricostruito - per troncamento, per riassunto o andandolo a ripescare da un archivio.

Che cosa ho eseguito, e che cosa no. Codice provato il 2026-11-12 con openai-agents 0.24.0 e Python 3.14.4, in un ambiente virtuale temporaneo fuori dal repository. Nessun modello è stato chiamato, nemmeno Ollama: il modello è finto e risponde con un copione che ho scritto io, attraverso un trasporto HTTP simulato (nessuna rete, nessuna porta). Le uscite qui sotto provano quindi cosa fanno l'SDK e il tuo codice con una risposta data, non quanto bene un modello vero usi la memoria. Nei listati modello() è la funzione che restituisce il tuo modello: nella prova restituisce quello finto. Ogni listato eseguito lo è stato due volte, con uscita identica; i listati di una stessa sezione sono pezzi dello stesso file. Gli archivi (memoria, archivio) sono sostituti in RAM, dichiarati tali; il blocco SQL e la funzione di cancellazione sono schemi da adattare, non eseguiti. L'API delle sessioni è fra le parti più giovani dell'SDK: fissa la versione e verifica i nomi.

L'analogia: il collega con amnesia anterograda

Hai un collega bravissimo che però, ogni volta che esci dalla stanza, dimentica tutto. L'unico modo per lavorarci è consegnargli, a ogni ingresso, il verbale completo di quello che vi siete detti. Al terzo ingresso il verbale è di tre pagine; al trentesimo, di trenta - e lui legge tutto, ogni volta, prima di risponderti. A un certo punto il verbale non ci sta sulla scrivania: devi scegliere se portargli solo le ultime pagine, o un riassunto, o un indice con la possibilità di cercare i dettagli in archivio. Queste tre scelte sono esattamente le tre strategie di memoria che vediamo qui sotto, e nella pratica le varianti sono combinazioni di queste.

Il meccanismo minimo: la cronologia riusata

La forma più semplice di memoria non ha bisogno di librerie: prendi la cronologia dalla run precedente e la usi come input della prossima.

import asyncio
from agents import Agent, Runner
from lezioni import modello

agente = Agent(name="Assistente", instructions="Rispondi in italiano, breve.", model=modello())

async def main():
    esito = await Runner.run(agente, "Mi chiamo Federico e lavoro su un portfolio Angular.")
    print(esito.final_output)

    cronologia = esito.to_input_list()            # tutta la conversazione finora
    cronologia.append({"role": "user", "content": "Su che framework sto lavorando?"})

    esito2 = await Runner.run(agente, cronologia)  # input = lista, non stringa
    print(esito2.final_output)
    print(len(cronologia), "elementi inviati al secondo turno")

asyncio.run(main())

# Uscita:
# Piacere, Federico.
# Angular.
# 3 elementi inviati al secondo turno

Il punto chiave è che Runner.run accetta sia una stringa sia una lista di elementi. Senza il secondo caso non esisterebbe la memoria. L'uscita mostra cosa stai facendo davvero: al secondo turno mandi tre elementi (la domanda, la risposta, la nuova domanda), quindi paghi anche il primo turno. Gli elementi di to_input_list() non hanno sempre la forma {"role": ..., "content": ...}: i tool producono elementi di tipo function_call e function_call_output, e conviene saperlo prima di scrivere codice che li taglia (lo vedi nella sezione sul troncamento).

Il conto, con i numeri

Supponiamo turni di dimensione simile: 500 token fra domanda e risposta (cifra scelta per l'esempio, non misurata). Al turno n il prompt contiene la somma di tutti i turni precedenti, e il calcolo si fa in poche righe.

TOKEN_PER_TURNO = 500
TURNI = 20

inviati_per_turno = [TOKEN_PER_TURNO * n for n in range(1, TURNI + 1)]
print("turno 1:", inviati_per_turno[0], "- turno 2:", inviati_per_turno[1],
      "- turno 20:", inviati_per_turno[-1])
print("totale inviato:", sum(inviati_per_turno))
print("se la memoria fosse lineare:", TOKEN_PER_TURNO * TURNI)
print("rapporto:", round(sum(inviati_per_turno) / (TOKEN_PER_TURNO * TURNI), 2))

# Uscita:
# turno 1: 500 - turno 2: 1000 - turno 20: 10000
# totale inviato: 105000
# se la memoria fosse lineare: 10000
# rapporto: 10.5

Se la memoria fosse «lineare» pagheresti 20 × 500 = 10.000 token. Paghi 10,5 volte tanto (il rapporto è (n+1)/2), e il fattore peggiora con la lunghezza della conversazione: il totale cresce come n². Questo è il motivo per cui una chat lunga diventa costosa molto prima di esaurire la finestra di contesto, e il motivo per cui ogni token che metti in un system prompt lo paghi n volte, non una.

Cosa consegue dalla crescita quadratica

Il conto della sezione precedente - 105.000 token per venti turni invece di 10.000 - non è una curiosità aritmetica: è il vincolo che determina tre decisioni di progetto che altrimenti sembrano arbitrarie. Vale la pena derivarle, perché sono le stesse tre che tornano in ogni lezione sui costi.

Prima conseguenza: ogni token nel system prompt lo paghi una volta per turno rimanente. Un paragrafo da 200 token in cima alle istruzioni costa 200 token al primo turno e 200 a ciascuno dei successivi: in una conversazione da venti turni sono 4.000 token, non 200. Questo riordina completamente le priorità di ottimizzazione. Accorciare le istruzioni di una riga che si ripete vale venti volte più che accorciare una risposta che si dice una volta. E spiega anche perché la raccomandazione «metti i casi rari nella descrizione del tool, non nel system prompt» non è stilistica: la descrizione del tool la paghi anch'essa a ogni giro, ma almeno la paghi solo se quel tool è nel catalogo di quell' agente.

Seconda conseguenza: il risultato di un tool è la voce di costo più sottovalutata. Questo lo abbiamo già incontrato nella lezione sui tool, e qui si vede perché: un risultato di 1.500 token prodotto al turno tre viene ri-inviato ai turni 4, 5, 6 e seguenti. Non è un costo locale, è un costo che si ripete. Da qui la regola - un tool è un riduttore di informazione - con la sua giustificazione numerica.

Terza conseguenza: il punto di pareggio fra «tenere tutto» e «riassumere» si calcola. Non è una questione di gusto. Riassumere costa una chiamata extra (il riassuntore) e riduce la dimensione della cronologia da quel punto in avanti. Conviene quando il costo evitato nei turni successivi supera il costo della compressione.

def conviene_riassumere(token_cronologia: int, turni_attesi_residui: int,
                        token_riassunto: int = 300,
                        costo_chiamata_riassunto: int | None = None) -> bool:
    """Confronto grezzo in token: comprimere adesso conviene?"""
    # costo di non fare niente: la cronologia intera, per ogni turno residuo
    senza = token_cronologia * turni_attesi_residui
    # costo di comprimere: una lettura della cronologia + il riassunto ripetuto
    costo_compressione = costo_chiamata_riassunto or token_cronologia
    con = costo_compressione + token_riassunto * turni_attesi_residui
    return con < senza

print(conviene_riassumere(6000, 8))     # 6000 + 2400 contro 48000
print(conviene_riassumere(6000, 1))     # 6000 + 300 contro 6000
print(conviene_riassumere(6000, 2))     # 6000 + 600 contro 12000

# Uscita:
# True
# False
# True

Le tre righe di uscita sono tre scenari con la stessa cronologia da 6000 token: con otto turni residui comprimere conviene (8.400 contro 48.000), con due ancora (6.600 contro 12.000), con uno no (6.300 contro 6.000). Il parametro che manca in questa funzione, e che è il più difficile da stimare, è turni_attesi_residui. Nella pratica si usa una stima empirica dalla distribuzione reale delle conversazioni (l'esempio «90% entro sei turni» è ipotetico): se quasi tutte finiscono entro sei turni, riassumere al quinto è quasi sempre spreco. È un numero che hai gratis nei log, e che la maggior parte dei progetti non guarda mai.

Una nota per non prendere la formula troppo sul serio: paragona token a token, mentre in realtà token in ingresso e in uscita hanno prezzi diversi, e una chiamata al riassuntore ha anche un costo in latenza che cade proprio nel turno in cui l'utente aspetta. Serve come ordine di grandezza per capire quando la domanda è rilevante, non come decisore automatico.

Quando nella cronologia c'è un errore: il consolidamento

C'è un effetto della memoria che non riguarda il costo e che è più insidioso: una risposta sbagliata, una volta entrata nella cronologia, influenza tutti i turni successivi. Il modello la rilegge a ogni giro come parte della conversazione: nel contesto c'è, scritta con il ruolo «assistant», e nulla la distingue da una risposta giusta. È un rischio plausibile; in queste prove non ho misurato con che frequenza un modello vero ci cada, e il listato sotto mostra solo la meccanica.

L'esempio tipico: al turno due l'agente afferma che l'ordine A-7781 è stato spedito il 3 ottobre. È falso, ma nessuno lo corregge. Al turno sei l'utente chiede quanto tempo è passato dalla spedizione, e un agente che si fida della cronologia calcola i giorni dal 3 ottobre: coerenza interna perfetta, risultato sbagliato. L'errore si è consolidato e ne ha generato un secondo, che sembra il frutto di un calcolo corretto.

Le difese sono tre, in un ordine che è un mio giudizio di progetto e non una misura: prima si corregge, poi si previene, infine si taglia.

Correggere esplicitamente, non sperare. Se il tuo codice si accorge che un'affermazione era sbagliata (per esempio perché un guardrail di uscita ha confrontato un numero d'ordine con il database), non basta ignorarla: va inserito nella cronologia un messaggio che la smentisce in modo esplicito. Sperare che l'ultimo messaggio prevalga non è una strategia, perché il modello vede entrambi. L'uscita mostra la forma del rimedio: la cronologia passa da tre a quattro messaggi, la falsità resta (non la cancelli, perché il resto della conversazione la richiama) e subito dopo c'è la smentita, che dice quale affermazione ignorare.

def smentisci(cronologia: list, affermazione_errata: str, verita: str) -> list:
    """Aggiunge una correzione esplicita invece di contare sull'ultimo messaggio."""
    return [*cronologia, {"role": "user", "content":
        f"CORREZIONE: quanto detto prima su «{affermazione_errata}» è errato. "
        f"Il dato corretto è: {verita}. "
        f"Usa SOLO il dato corretto da qui in avanti e ignora l'affermazione precedente."}]

cronologia = [
    {"role": "user", "content": "L'ordine A-7781 è partito?"},
    {"role": "assistant", "content": "Sì, l'ordine A-7781 è stato spedito il 3 ottobre."},   # falso
    {"role": "user", "content": "Quanti giorni sono passati dalla spedizione?"},
]
nuova = smentisci(cronologia, "la spedizione di A-7781", "A-7781 non è ancora partito")
print(len(cronologia), "->", len(nuova), "messaggi")
for m in nuova:
    print(m["role"].ljust(9), m["content"][:70])

# Uscita:
# 3 -> 4 messaggi
# user      L'ordine A-7781 è partito?
# assistant Sì, l'ordine A-7781 è stato spedito il 3 ottobre.
# user      Quanti giorni sono passati dalla spedizione?
# user      CORREZIONE: quanto detto prima su «la spedizione di A-7781» è errato.

Non far entrare le affermazioni non verificate. È la difesa migliore e si progetta a monte: se l'agente non deve mai citare un numero che non ha letto da un tool, il modo di imporlo non è un'istruzione («non inventare numeri») ma la struttura - un tipo di output con il campo di citazione, come nella lezione sull'output strutturato, e un controllo che lo confronta con la fonte.

Tagliare la conversazione. Nei casi in cui la cronologia si è inquinata in modo irrecuperabile, ricominciare con un riassunto scritto dal tuo codice sulla base dei fatti verificati - non dalla cronologia - è più economico che tentare di correggere. È brutale e a volte è la cosa giusta.

Vale la pena notare la connessione con la lezione sulle trappole reali degli agenti, dove c'è un laboratorio di injection: il consolidamento è il meccanismo che rende l'injection indiretta così efficace. Un'istruzione ostile entrata nel contesto attraverso l'output di un tool non vale solo per il turno in cui arriva: resta lì, e viene rimandata al modello a ogni turno successivo. La memoria, che è la proprietà che rende utile un assistente, è anche il canale che amplifica un contenuto ostile.

Sessioni: la memoria delegata all'SDK

Ricopiare la cronologia a mano funziona in uno script e diventa fragile in un server. L'SDK offre il concetto di sessione: un oggetto che carica la cronologia prima della run e la salva dopo, identificato da una chiave. Il file conversazioni.db è un database SQLite locale; nella prova l'ho creato in una cartella temporanea.

import asyncio
from agents import Agent, Runner, SQLiteSession
from lezioni import modello

agente = Agent(name="Assistente", instructions="Rispondi in italiano, breve.", model=modello())

async def turno(id_conversazione: str, testo: str) -> str:
    sessione = SQLiteSession(id_conversazione, "conversazioni.db")
    esito = await Runner.run(agente, testo, session=sessione)
    return esito.final_output

async def main():
    # due conversazioni indipendenti, nello stesso processo
    print(await turno("utente-42", "Mi chiamo Federico."))
    print(await turno("utente-99", "Mi chiamo Giulia."))
    print(await turno("utente-42", "Come mi chiamo?"))
    for chiave in ("utente-42", "utente-99"):
        elementi = await SQLiteSession(chiave, "conversazioni.db").get_items()
        print(chiave, len(elementi), "elementi salvati")

asyncio.run(main())

# Uscita:
# Piacere, Federico.
# Piacere, Giulia.
# Federico.
# utente-42 4 elementi salvati
# utente-99 2 elementi salvati

L'uscita dice due cose: la terza risposta («Federico») arriva dopo una conversazione intercalata con un altro utente, e le due sessioni restano separate (4 elementi per utente-42, due domande e due risposte; 2 per utente-99). Poi tre cose da sapere prima di usarlo in produzione. Primo: la chiave di sessione è la tua frontiera di isolamento - se la sbagli, due utenti leggono la conversazione l'uno dell'altro, ed è un incidente di riservatezza, non un bug di comodità. Secondo: SQLite va bene per una macchina, non per più processi che scrivono in concorrenza; su un servizio reale serve una sessione appoggiata al tuo database. Terzo: per default la sessione non limita nulla (SessionSettings(limit=None), dalla documentazione): cresce finché non satura la finestra di contesto, e che cosa succede allora dipende dal server, che può rifiutare la richiesta o troncare; su un Ollama vero non l'ho verificato.

Il limite esiste, ed è un'impostazione dell'SDK. C'è una trappola che ho trovato provandolo: il limite conta elementi, non turni, e un turno con tool ne occupa parecchi.

import asyncio
from agents import Agent, Runner, RunConfig, SessionSettings, SQLiteSession
from lezioni import modello, RICHIESTE, leggi_giacenza, leggi_ordini_in_arrivo

agente = Agent(name="Magazzino", instructions="Usa i tool.", model=modello(),
               tools=[leggi_giacenza, leggi_ordini_in_arrivo])

def tipo(m: dict) -> str:
    return m.get("type") or m.get("role")

async def main():
    sessione = SQLiteSession("utente-42", "conversazioni.db")
    await Runner.run(agente, "Lo SKU-11 va riordinato?", session=sessione)
    print("salvati   :", [tipo(m) for m in await sessione.get_items()])
    print("limit=4   :", [tipo(m) for m in await sessione.get_items(limit=4)])
    RICHIESTE.clear()
    await Runner.run(agente, "E lo SKU-42?", session=sessione,
                     run_config=RunConfig(session_settings=SessionSettings(limit=4)))
    inviati = RICHIESTE[-1]["messages"]
    print("inviati   :", [m["role"] for m in inviati])

asyncio.run(main())

# Uscita:
# salvati   : ['user', 'function_call', 'function_call_output', 'function_call', 'function_call_output', 'message']
# limit=4   : ['function_call_output', 'function_call', 'function_call_output', 'message']
# inviati   : ['system', 'assistant', 'tool', 'assistant', 'user']

L'uscita si legge riga per riga. La sessione salva sei elementi per un solo turno con due tool. Dalla richiesta realmente inviata si ricava ciò che il modello riceve: per leggerla uso il registro del trasporto finto (RICHIESTE), che con un server vero non hai. Chiedere gli ultimi quattro, direttamente alla sessione, restituisce una lista che inizia con un risultato di tool senza la sua richiesta. Nella richiesta realmente inviata, invece, i ruoli sono system, assistant, tool, assistant, user: l'SDK ha scartato il risultato orfano. È un comportamento di questa versione con questo tipo di modello, che ho osservato e che la documentazione non promette; se gestisci tu la lista (con to_input_list()), nessuno ti ripulisce niente, come vedi nella sezione sul troncamento.

Stimare i token senza dipendere da un tokenizzatore

Per decidere quando troncare o riassumere serve un numero, e serve prima di fare la chiamata. Il conteggio esatto dipende dal tokenizzatore del modello specifico, che su un'installazione locale potresti non avere a portata di mano. Una stima basata sui caratteri è sempre disponibile; il rapporto caratteri per token che uso (3,2) è un'ipotesi prudente, non una misura: scegline uno più basso del vero per stare dal lato sicuro, e verificalo con il tokenizzatore del tuo modello quando puoi. La funzione lavora sul testo di qualunque elemento, annidato o no, perché gli elementi dell'SDK non hanno sempre un campo content.

import json

# Ipotesi prudente, NON misurata: il tokenizzatore del tuo modello può dare un altro rapporto.
CARATTERI_PER_TOKEN = 3.2
OVERHEAD_MESSAGGIO = 4        # ruolo e delimitatori, stimato

def _caratteri(x) -> int:
    """Caratteri di testo in un messaggio, in qualunque formato: stringhe annidate incluse."""
    if isinstance(x, str):
        return len(x)
    if isinstance(x, dict):
        return sum(_caratteri(v) for v in x.values())
    if isinstance(x, list):
        return sum(_caratteri(v) for v in x)
    return 0

def stima_token(messaggi: list[dict]) -> int:
    return int(sum(_caratteri(m) / CARATTERI_PER_TOKEN + OVERHEAD_MESSAGGIO for m in messaggi))

def budget_cronologia(finestra_modello: int, riservati_uscita: int = 1500,
                      margine: float = 0.15) -> int:
    """Quanti token può occupare la cronologia, lasciando spazio alla risposta."""
    return int((finestra_modello - riservati_uscita) * (1 - margine))

chat_classica = [{"role": "user", "content": "x" * 320}]
formato_sdk = [{"type": "function_call", "name": "leggi_giacenza", "call_id": "c1",
                "arguments": '{"sku": "SKU-11"}'},
               {"type": "function_call_output", "call_id": "c1",
                "output": "{'sku': 'SKU-11', 'giacenza': 4}"}]
print(stima_token(chat_classica))        # (320 + 4 caratteri) / 3,2 + 4: anche il ruolo conta
print(stima_token(formato_sdk))
print(budget_cronologia(8192))           # (8192 - 1500) * 0,85

# Uscita:
# 105
# 39
# 5688

Le tre uscite si leggono così. La prima è una chat classica da 320 caratteri: 105 token stimati, perché anche la parola «user» del ruolo conta come testo (scelta conservativa) e si aggiungono 4 token di overhead per messaggio. La seconda è una coppia richiesta/risultato nel formato dell'SDK: 39 token, che la versione ingenua (quella che guarda solo content e tool_calls) avrebbe stimato in 8 token, cioè il solo overhead. La terza è il budget per una finestra di 8192 token: 5688, dopo aver riservato 1500 token alla risposta e il 15% di margine. Dimenticare lo spazio per la risposta è il modo classico di ottenere un errore di contesto pieno proprio dopo aver pagato tutto il prompt.

Le tre strategie, in un disegno

Quando la cronologia non ci sta più nella finestra, le opzioni sono tre e si distinguono per cosa buttano. Tenerle in mente come immagine aiuta a scegliere in fretta.

Troncamento, riassunto e recupero esterno a confronto la cronologia completa: 20 turni. Nella finestra ne stanno 6. 1. troncamento turni 1-14: buttati, persi per sempre turni 15-20 costo minimo sintomo: «come ti ho detto prima» → l'agente non sa di cosa parli 2. riassunto riassunto 1-14 dettagli scartati turni 15-20 costo lineare sintomo: perde il codice d'ordine esatto; l'errore si accumula a ogni compressione 3. recupero turni 1-14 in archivio, indicizzati, integri turni 15-20 costo basso tool cerca_nello_storico(domanda) sintomo: il recupero restituisce il frammento sbagliato e il modello gli crede Il criterio di scelta: il passato serve spesso e per intero → riassunto. Serve raramente ma esatto → recupero. Non serve quasi mai → troncamento, che è la soluzione più semplice e spesso sufficiente.
Le tre strategie si distinguono per cosa sacrificano: il passato, i dettagli, o la semplicità dell'architettura.

Le tre strategie quando non ci sta più

1. Troncamento a finestra scorrevole. Tieni il system prompt e gli ultimi k turni. Economico, prevedibile, e perde il passato in modo brutale: l'utente dirà «come ti ho detto prima» e l'agente non saprà di cosa parla.

import asyncio
from agents import Agent, Runner
from lezioni import modello, leggi_giacenza, leggi_ordini_in_arrivo

def e_risultato_di_tool(m: dict) -> bool:
    # nel formato che l'SDK mette in to_input_list() il risultato è un elemento
    # di tipo function_call_output; nel formato chat classico è role == "tool"
    return m.get("type") == "function_call_output" or m.get("role") == "tool"

def tronca(cronologia: list, ultimi: int = 12) -> list:
    sistema = [m for m in cronologia if m.get("role") == "system"]
    resto = [m for m in cronologia if m.get("role") != "system"]
    coda = resto[-ultimi:]
    # NON iniziare con un risultato la cui richiesta è stata tagliata
    while coda and e_risultato_di_tool(coda[0]):
        coda = coda[1:]
    return sistema + coda

def tipo(m: dict) -> str:
    return m.get("type") or m.get("role")

async def main():
    agente = Agent(name="Magazzino", instructions="Usa i tool.", model=modello(),
                   tools=[leggi_giacenza, leggi_ordini_in_arrivo])
    cronologia = (await Runner.run(agente, "Lo SKU-11 va riordinato?")).to_input_list()
    print("intera      :", [tipo(m) for m in cronologia])
    print("ultimi 4    :", [tipo(m) for m in cronologia[-4:]])
    print("dopo tronca :", [tipo(m) for m in tronca(cronologia, 4)])

asyncio.run(main())

# Uscita:
# intera      : ['user', 'function_call', 'function_call_output', 'function_call', 'function_call_output', 'message']
# ultimi 4    : ['function_call_output', 'function_call', 'function_call_output', 'message']
# dopo tronca : ['function_call', 'function_call_output', 'message']

L'uscita mostra il difetto. La cronologia completa ha sei elementi; gli ultimi quattro iniziano con un function_call_output la cui richiesta è stata tagliata; dopo tronca l'orfano è sparito e restano tre elementi. Nella versione che avevo scritto in origine il controllo guardava role == "tool", che nel formato di to_input_list() non esiste: il while non avrebbe mai tolto nulla. Ho verificato anche che, passando all'SDK una lista che inizia con un risultato orfano, la richiesta parte così com'è (ruoli system, tool, user): se un endpoint vero la rifiuti, e quale, non l'ho provato, ma la regola prudente è la stessa. Le coppie richiesta/risultato sono atomiche.

2. Riassunto progressivo. Quando la cronologia supera una soglia, chiedi a un modello di comprimere i turni vecchi in un riassunto e sostituiscili con quello.

SOGLIA = 6000       # token stimati

async def comprimi(cronologia: list, riassuntore: Agent, soglia: int = SOGLIA) -> list:
    if stima_token(cronologia) < soglia:
        return cronologia
    taglio = len(cronologia) - 6                 # conserva circa gli ultimi 6 elementi
    while taglio > 0 and e_risultato_di_tool(cronologia[taglio]):
        taglio -= 1                              # la coppia richiesta/risultato resta intera
    testa, coda = cronologia[:taglio], cronologia[taglio:]
    esito = await Runner.run(riassuntore, [
        *testa,
        {"role": "user", "content":
            "Riassumi la conversazione sopra in massimo 200 parole. "
            "Conserva: decisioni prese, vincoli dichiarati, identificativi, numeri esatti, "
            "cose rimaste da fare. Scarta i saluti e le riformulazioni."}
    ])
    return [{"role": "system", "content": f"Riassunto della conversazione precedente:\n{esito.final_output}"},
            *coda]

Per provarlo ho costruito una conversazione di 30 elementi e un riassuntore finto, che risponde con un testo scritto da me.


riassuntore = Agent(name="Riassuntore", instructions="Comprimi senza perdere gli identificativi.",
                    model=modello())

async def main():
    cronologia = [{"role": "user" if i % 2 == 0 else "assistant",
                   "content": f"turno {i}: " + "testo " * 120} for i in range(30)]
    nuova = await comprimi(cronologia, riassuntore)
    print(len(cronologia), "elementi,", stima_token(cronologia), "token stimati")
    print(len(nuova), "elementi,", stima_token(nuova), "token stimati")
    print(nuova[0]["content"][:60])
    print(nuova[1]["content"][:8])

asyncio.run(main())

# Uscita:
# 30 elementi, 7021 token stimati
# 7 elementi, 1453 token stimati
# Riassunto della conversazione precedente:
# Riassunto: l'utent
# turno 24

L'uscita mostra la meccanica: 30 elementi e 7.021 token stimati diventano 7 elementi e 1.453 (il riassunto più gli ultimi sei, che qui valgono ciascuno circa 230 token). Il testo del riassunto è il mio copione, non una prova di quanto bene un modello comprima. Il taglio arretra se cade su un risultato di tool, per lo stesso motivo di prima. Il prompt del riassuntore è la parte che conta: «conserva identificativi e numeri esatti» è ciò che evita la perdita del codice d'ordine che servirà fra sei turni. Il difetto strutturale resta: il riassunto è lossy e l'errore si accumula, perché il riassunto n+1 si basa sul riassunto n. Dopo molte compressioni la conversazione racconta una versione approssimata di sé stessa.

3. Memoria esterna recuperabile. Nel contesto tieni solo la finestra recente; tutto il resto va in un archivio (ricerca testuale o vettoriale) e torna indietro solo quando serve, attraverso un tool.

import asyncio
from agents import Agent, Runner, function_tool
from lezioni import modello

class Archivio:
    """Stand-in per il tuo archivio: ricerca per parole in comune, nessun indice vettoriale."""
    def __init__(self, messaggi: dict[str, list[str]]):
        self.messaggi = messaggi
    async def cerca(self, id_sessione: str, domanda: str, k: int) -> list[str]:
        parole = set(domanda.lower().split())
        punteggio = lambda t: len(parole & set(t.lower().replace(".", "").split()))
        candidati = sorted(self.messaggi.get(id_sessione, []), key=punteggio, reverse=True)
        return [t for t in candidati[:k] if punteggio(t) > 0]

archivio = Archivio({"s1": ["Il mio numero d'ordine è A-7781.", "Preferisco la spedizione veloce.",
                            "Buongiorno!"]})

def sessione_corrente() -> str:
    return "s1"                       # nel tuo codice: l'identificativo della conversazione in corso

@function_tool
async def cerca_nello_storico(domanda: str, max_risultati: int = 3) -> dict:
    """Cerca nei messaggi passati di QUESTA conversazione, oltre la finestra recente.

    Usa questo tool quando l'utente fa riferimento a qualcosa detto 'prima'
    e non lo trovi nei messaggi che hai sotto gli occhi.
    """
    passi = await archivio.cerca(sessione_corrente(), domanda, k=max_risultati)
    return {"frammenti": passi, "nota": "estratti dai messaggi passati, non parafrasati"}

agente = Agent(name="Assistente", instructions="Cerca nello storico se serve.",
               tools=[cerca_nello_storico], model=modello())

async def main():
    esito = await Runner.run(agente, "Qual era il mio numero d'ordine, quello detto prima?")
    for voce in esito.new_items:
        if voce.type == "tool_call_output_item":
            print(voce.output)

asyncio.run(main())

# Uscita:
# {'frammenti': ["Il mio numero d'ordine è A-7781."], 'nota': 'estratti dai messaggi passati, non parafrasati'}

L'archivio del listato è un sostituto in RAM che confronta parole: l'uscita prova che il tool restituisce il frammento giusto per quella domanda, non che una ricerca vettoriale reale lo troverebbe. È la strategia migliore su conversazioni lunghe e l'unica che scala davvero, ma introduce un componente in più (l'archivio e il suo indice) e un modo nuovo di sbagliare: il recupero può restituire il frammento sbagliato e il modello gli crederà. Il lato architetturale - indici, grafi di conoscenza, memoria a lungo termine fra sessioni diverse - è approfondito nella serie dedicata in memory systems per agenti: contesto e knowledge graph; qui mi fermo al livello della singola conversazione.

Una sessione persistente è un archivio di dati personali

C'è un aspetto della memoria che non è tecnico e che va deciso prima di scrivere il codice, perché dopo è un lavoro di rimozione invece che di progettazione: nel momento in cui salvi le conversazioni su disco, hai creato un archivio di dati personali. Non perché l'hai voluto, ma perché gli utenti scrivono nelle chat cose che non scriverebbero in un form: indirizzi, numeri d'ordine, problemi di salute, situazioni familiari, lamentele su colleghi.

Tre decisioni da prendere esplicitamente, con la loro conseguenza tecnica.

Che cosa salvi. La scelta predefinita «tutto» è comoda e quasi mai necessaria. Spesso al turno successivo servono gli ultimi messaggi e un piccolo stato strutturato (il numero d'ordine in discussione, la lingua, la decisione già presa), non la trascrizione integrale. Salvare lo stato strutturato invece della trascrizione riduce contemporaneamente il costo, il rischio e la superficie di esposizione. Non è sempre possibile; quando lo è, è la scelta migliore su ogni asse.

Per quanto tempo. Una scadenza va decisa e implementata, non rinviata. L'implementazione è banale e il fatto che sia banale è la ragione per cui non va rimandata.

-- esempio con PostgreSQL: scadenza dichiarata sulla riga (schema da adattare, non eseguito)
CREATE TABLE conversazioni (
    id          text PRIMARY KEY,
    utente_id   text NOT NULL,
    messaggi    jsonb NOT NULL,
    creata_il   timestamptz NOT NULL DEFAULT now(),
    scade_il    timestamptz NOT NULL DEFAULT now() + interval '30 days'
);
CREATE INDEX ON conversazioni (scade_il);
CREATE INDEX ON conversazioni (utente_id);

-- cancellazione periodica: una riga, da un cron
DELETE FROM conversazioni WHERE scade_il < now();

Come si cancella su richiesta. Deve esistere una funzione che, dato un utente, rimuove tutto: sessioni, memoria durevole, e - questo è il punto che si dimentica - anche le tracce e i log. Una conversazione cancellata dal database e ancora presente in un sistema di tracing non è cancellata. Vale la pena scrivere quella funzione il primo giorno, perché scoprirla mancante quando serve è una brutta giornata.

# schema da adattare, non eseguito: db, tracing e indice_vettoriale sono i tuoi componenti
async def cancella_tutto_di(utente_id: str) -> dict:
    """Rimuove ogni traccia di un utente. Da chiamare su richiesta di cancellazione."""
    sessioni = await db.execute("DELETE FROM conversazioni WHERE utente_id = $1", utente_id)
    fatti    = await db.execute("DELETE FROM memoria_durevole WHERE utente_id = $1", utente_id)
    tracce   = await tracing.cancella_per_utente(utente_id)   # NON dimenticare questa
    archivio = await indice_vettoriale.cancella_per_utente(utente_id)   # ne' questa
    return {"sessioni": sessioni, "fatti": fatti,
            "tracce": tracce, "frammenti_indicizzati": archivio}

Quest'ultimo punto ha una conseguenza che tocca l'architettura e che vale la pena anticipare: la strategia di memoria esterna recuperabile - quella che indicizza i messaggi passati in un archivio di ricerca - moltiplica i posti in cui i dati di una persona esistono. È la strategia migliore per la qualità e la peggiore per il ciclo di vita del dato, perché un indice vettoriale va ripulito con la stessa diligenza del database. Non è una ragione per evitarla, è una ragione per scrivere la funzione di cancellazione insieme all'indicizzazione e non sei mesi dopo.

Un'osservazione che chiude il cerchio con il resto del modulo: su questo terreno un modello eseguito in locale cambia la natura del problema. Se il modello gira sulla tua macchina, i contenuti delle conversazioni non la lasciano, quindi il perimetro da proteggere è uno solo e lo controlli tu. Resta tutto il lavoro sulla conservazione e sulla cancellazione - quello non lo risolve nessuna scelta di provider - ma scompare la parte più difficile da governare, cioè il dato che hai già mandato a qualcun altro.

Memoria di sessione e memoria durevole sono due cose

Vale la pena separarle anche nel codice, perché hanno politiche di conservazione diverse:

  • Sessione. Vive quanto la conversazione, contiene tutto, si può buttare. Dura minuti o ore.
  • Durevole. Fatti stabili sull'utente («preferisce risposte brevi», «lavora su Angular», «fuso orario Europe/Rome»). Poche righe, scritte con intenzione, rilette all'inizio di ogni nuova conversazione.

Scrivere nella memoria durevole è una decisione, non un effetto collaterale: se ci finisce tutto, diventa un secondo contesto che paghi sempre e che contiene rumore. Due righe in cima al system prompt battono trenta righe di storia.

La memoria durevole scritta dall'agente stesso

Finora la memoria durevole l'abbiamo descritta come qualcosa che tu scrivi. Si può anche far decidere all'agente cosa vale la pena ricordare, ed è utile - a patto di trattarla come un'operazione vincolata, non come un diario libero.

import asyncio
from dataclasses import dataclass
from typing import Literal
from agents import Agent, Runner, function_tool, RunContextWrapper
from lezioni import modello

@dataclass
class Fatto:
    categoria: str
    contenuto: str

@dataclass
class Utente:
    id_utente: str

class MemoriaInRam:
    """Stand-in per il tuo archivio: un dizionario, nessun database."""
    def __init__(self, precaricati: int = 0):
        self.per_utente: dict[str, list[Fatto]] = {
            "u1": [Fatto("fatto_stabile", f"fatto {i}") for i in range(precaricati)]}
    async def leggi(self, id_utente: str) -> list[Fatto]:
        return list(self.per_utente.get(id_utente, []))
    async def aggiungi(self, id_utente: str, categoria: str, contenuto: str) -> None:
        self.per_utente.setdefault(id_utente, []).append(Fatto(categoria, contenuto))

memoria = MemoriaInRam()

@function_tool
async def ricorda(contesto: RunContextWrapper[Utente],
                  categoria: Literal["preferenza", "vincolo", "fatto_stabile"],
                  contenuto: str) -> dict:
    """Salva UN fatto stabile sull'utente, utile anche in conversazioni future.

    Usa questo tool solo per cose che restano vere domani: preferenze di stile,
    vincoli tecnici, dati anagrafici dichiarati dall'utente.
    NON usarlo per: lo stato della conversazione in corso, numeri che cambiano,
    cose dette una volta per caso.
    """
    if len(contenuto) > 180:
        return {"ok": False, "errore": "troppo_lungo",
                "come_rimediare": "Riscrivilo in meno di 180 caratteri, un fatto solo."}
    fatti = await memoria.leggi(contesto.context.id_utente)
    if len(fatti) >= 20:                       # tetto duro
        return {"ok": False, "errore": "memoria_piena",
                "come_rimediare": "Cancella un fatto superato (serve un tool dimentica), poi riprova."}
    await memoria.aggiungi(contesto.context.id_utente, categoria, contenuto)
    return {"ok": True, "fatti_memorizzati": len(fatti) + 1}

agente = Agent(name="Assistente", instructions="Ricorda i fatti stabili.", tools=[ricorda],
               model=modello())

async def main():
    esito = await Runner.run(agente, "Preferisco risposte brevi.", context=Utente("u1"))
    for voce in esito.new_items:
        if voce.type == "tool_call_output_item":
            print(voce.output)

asyncio.run(main())

# Uscita:
# {'ok': True, 'fatti_memorizzati': 1}
# {'ok': False, 'errore': 'troppo_lungo', 'come_rimediare': 'Riscrivilo in meno di 180 caratteri, un fatto solo.'}

Nella prova il copione chiama il tool due volte: la prima con un fatto valido (risposta ok: True), la seconda con 181 caratteri (risposta troppo_lungo, con il rimedio scritto per il modello). Rifatta con una memoria che parte da 20 fatti (MemoriaInRam(precaricati=20)), la stessa chiamata risponde:

# Uscita:
# {'ok': False, 'errore': 'memoria_piena', 'come_rimediare': 'Cancella un fatto superato (serve un tool dimentica), poi riprova.'}

Il tetto di venti fatti non è prudenza eccessiva: è la differenza fra una memoria utile e un secondo contesto che paghi a ogni turno di ogni conversazione futura. Senza limite, dopo un mese la memoria durevole contiene «l'utente ha salutato», «l'utente ha chiesto come stai» e tutto ciò che il modello ha giudicato interessante in un momento di entusiasmo.

E l'uso, all'inizio di ogni nuova conversazione (la classe Fatto e la memoria sono quelle del listato precedente, qui ridotte a un sostituto):

import asyncio
from dataclasses import dataclass

ISTRUZIONI = "Rispondi in italiano, breve."

@dataclass
class Fatto:
    categoria: str
    contenuto: str

class Memoria:
    async def leggi(self, id_utente: str) -> list[Fatto]:
        return [Fatto("preferenza", "risposte brevi"), Fatto("fatto_stabile", "lavora su Angular")]
memoria = Memoria()

async def apri_conversazione(id_utente: str) -> list[dict]:
    fatti = await memoria.leggi(id_utente)
    if not fatti:
        return [{"role": "system", "content": ISTRUZIONI}]
    righe = "\n".join(f"- [{f.categoria}] {f.contenuto}" for f in fatti)
    return [{"role": "system",
             "content": f"{ISTRUZIONI}\n\nCose che sai già su questo utente:\n{righe}\n"
                        "Usale senza annunciarle. Se una contraddice ciò che l'utente "
                        "dice adesso, vale quello che dice adesso."}]

async def main():
    cronologia = await apri_conversazione("u1")
    print(cronologia[0]["content"])

asyncio.run(main())

# Uscita:
# Rispondi in italiano, breve.
#
# Cose che sai già su questo utente:
# - [preferenza] risposte brevi
# - [fatto_stabile] lavora su Angular
# Usale senza annunciarle. Se una contraddice ciò che l'utente dice adesso, vale quello che dice adesso.

L'ultima riga delle istruzioni è quella che evita l'effetto più fastidioso di questi sistemi: l'agente che insiste su una preferenza di tre mesi fa mentre l'utente gli sta dicendo il contrario. La precedenza al presente va scritta esplicitamente: non c'è motivo di lasciare al modello l'ipotesi su quale delle due fonti valga di più. Nella prova ho verificato solo che il testo venga composto come previsto.

Perché «più contesto» non è «più memoria»

C'è un ragionamento che sembra ovvio e che è falso: se le finestre di contesto crescono, il problema della memoria si risolve da sé - basta mandare tutto. Vale la pena smontarlo, perché orienta male le scelte di progetto e perché il modo in cui è falso è istruttivo.

Il primo motivo è quello aritmetico che abbiamo già fatto: una finestra più grande non cambia il fatto che paghi la cronologia a ogni turno. Una finestra dieci volte più grande permette conversazioni dieci volte più lunghe, e il costo di quelle conversazioni cresce cento volte. Il tetto si sposta, la curva no.

Il secondo motivo è quello interessante: la capacità di usare bene l'informazione non cresce con la finestra. Il lavoro di Liu e colleghi (Lost in the Middle: How Language Models Use Long Contexts, arXiv:2307.03172) misura questo direttamente su due compiti (domande su più documenti e recupero di coppie chiave-valore) in cui l'informazione necessaria è collocata in posizioni diverse di un contesto lungo, e trova che le prestazioni sono spesso massime quando sta all'inizio o alla fine e calano quando sta in mezzo, un andamento a U: l'informazione all'inizio e alla fine viene usata meglio di quella al centro. La conseguenza pratica è concreta e si applica subito.

  • Le istruzioni vanno all'inizio (dove sono già, per costruzione del formato dei messaggi).
  • La richiesta corrente e i dati appena letti vanno alla fine, il più vicino possibile al punto in cui il modello genera.
  • Ciò che sta in mezzo è la parte che il modello userà peggio. Se c'è un'informazione critica sepolta al centro di quattromila token di risultati di tool, va ripetuta alla fine, non lasciata lì a sperare.

L'elenco si traduce in una regola di montaggio del contesto, e la terza riga è un pattern che vale la pena nominare, perché risolve una classe di problemi: la ri-asserzione. Quando una run ha accumulato molto contesto e devi essere sicuro che un vincolo venga rispettato, il modo efficace non è renderlo più enfatico nel system prompt (che è lontano) ma ripeterlo in un messaggio immediatamente prima della decisione.

def con_riasserzione(cronologia: list, vincoli: list[str]) -> list:
    """Ripete i vincoli critici alla FINE del contesto, dove vengono usati meglio."""
    if not vincoli:
        return cronologia
    righe = "\n".join(f"- {v}" for v in vincoli)
    return [*cronologia, {"role": "user", "content":
        f"Prima di rispondere, ricorda i vincoli che valgono sempre:\n{righe}"}]

cronologia = [{"role": "system", "content": ISTRUZIONI}]
cronologia = con_riasserzione(cronologia, [
    "Non superare 1000 EUR senza approvazione: usa richiedi_approvazione.",
    "Cita sempre il numero d'ordine letto dai tool, mai a memoria.",
])
print(len(cronologia), "messaggi; l'ultimo è del ruolo", cronologia[-1]["role"])
print(cronologia[-1]["content"])

# Uscita:
# 2 messaggi; l'ultimo è del ruolo user
# Prima di rispondere, ricorda i vincoli che valgono sempre:
# - Non superare 1000 EUR senza approvazione: usa richiedi_approvazione.
# - Cita sempre il numero d'ordine letto dai tool, mai a memoria.

Costa qualche decina di token. L'uscita prova che il messaggio di ri-asserzione finisce per ultimo; che nelle conversazioni lunghe cambi il comportamento di un modello vero non l'ho misurato, e lo presento come applicazione ragionevole di una proprietà documentata dal lavoro di Liu et al., non come risultato mio.

Il terzo motivo per cui «mandare tutto» non è una strategia riguarda il rumore. Una cronologia lunga contiene, insieme alle informazioni utili, tutti i tentativi andati male: il tool chiamato con l'argomento sbagliato, l'errore restituito, la domanda dell'utente riformulata tre volte. Tutto questo materiale è coerente con il compito e irrilevante per la decisione, e la sua presenza rende più difficile distinguere il segnale. Un riassunto scritto bene - che conserva decisioni, identificativi e vincoli, e butta i tentativi falliti - può produrre risposte migliori della cronologia integrale, non solo più economiche. È il caso in cui le due ottimizzazioni, costo e qualità, puntano nella stessa direzione: capita raramente, e quando capita vale la pena accorgersene.

Trade-off

  • Fedeltà contro costo. Cronologia completa: ricorda tutto, costa n². Riassunto: costo lineare, perde dettagli. Recupero: costo basso e preciso, ma dipende dalla qualità della ricerca.
  • Latenza. Riassumere costa una chiamata in più proprio nel turno in cui l'utente aspetta. Si mitiga comprimendo prima della soglia critica, in background.
  • Privacy e conservazione. Una sessione persistente è un archivio di dati personali: richiede scadenza, cancellazione su richiesta e una decisione esplicita su cosa viene salvato.
  • Qualità su contesti lunghi. Più contesto non è sempre meglio: il lavoro di Liu et al. (Lost in the Middle: How Language Models Use Long Contexts, arXiv:2307.03172) documenta che l'informazione collocata al centro di un contesto lungo viene usata peggio di quella all'inizio o alla fine. Conseguenza pratica: metti ciò che conta all'inizio o in fondo, non in mezzo.

In pratica le quattro voci si leggono insieme: se la priorità è non perdere nulla paghi in costo o in latenza, se la priorità è il costo paghi in dettagli perduti, e la privacy è un vincolo che vale qualunque strategia tu scelga.

Le trappole in cui si cade davvero

  1. Spezzare una coppia richiesta/risultato di tool troncando. Errore di validazione dall'endpoint, messaggio criptico, mezz'ora persa. Tronca per turni completi.
  2. Chiave di sessione sbagliata o riusata. Usare un identificativo prevedibile e non legato all'utente autenticato significa poter leggere la conversazione di qualcun altro cambiando un numero nell'URL.
  3. Memoria che cresce senza tetto. Nessun limite nella sessione = l'applicazione funziona per due settimane e poi inizia a rifiutare le richieste degli utenti più assidui.
  4. Riassumere troppo presto. Comprimere dopo tre turni risparmia nulla e perde dettagli che servivano. La soglia va calcolata sulla finestra reale del modello, non messa a intuito.
  5. Mettere dati volatili nel system prompt. Lo paghi a ogni turno e invecchia: la data di oggi, il carrello, la giacenza vanno in un tool.
  6. Confondere memoria e verità. Se nella cronologia c'è una risposta sbagliata del modello, quella resta nel contesto e influenza tutti i turni successivi: l'errore si consolida. Vale la pena correggere esplicitamente invece di sperare che l'ultimo messaggio prevalga.

L'ordine delle sei trappole non rappresenta una scala: sono modi indipendenti di sbagliare. La prima e la seconda si scoprono alla prima prova seria, la terza e la quarta solo dopo settimane di uso, la quinta e la sesta sono quelle che il codice non segnala mai da sé.

Che cosa sai fare adesso

Sai calcolare quanto costa una conversazione che cresce e decidere con un confronto quando comprimerla; sai scegliere fra troncare, riassumere e recuperare, e scrivere il troncamento senza spezzare una coppia richiesta/risultato nel formato che l'SDK produce davvero; sai usare le sessioni limitandone la dimensione e progettare scadenza e cancellazione dei dati; sai montare il contesto in modo che le istruzioni critiche stiano dove vengono lette meglio. Resta fuori ciò che qui non ho potuto provare: come un modello vero, su Ollama, si comporta con una cronologia lunga o riassunta. Nella prossima lezione la conversazione si sdoppia: cosa succede quando un agente passa il controllo a un altro.

Fonti

Eseguiti due volte con uscita identica (Python 3.14.4, openai-agents 0.24.0) contro un modello finto con trasporto HTTP simulato: cronologia riusata, conto dei token, confronto di convenienza, correzione, sessioni e limite, stima dei token, troncamento, compressione, ricerca nello storico, tool ricorda, composizione dell'apertura e ri-asserzione. Non eseguiti: il blocco SQL, la funzione di cancellazione, qualunque chiamata a un modello vero (nemmeno Ollama). Gli esempi 90%, 500 token per turno e A-7781 sono ipotetici.

Commenti

Caricamento commenti...

Accedi per commentare

Discuti questo articolo nella community

Unisciti al gruppo Telegram per discutere con altri sviluppatori, fare domande e condividere le tue esperienze.

Ti è piaciuto questo articolo?

Esplora altri contenuti sul blog o scopri i miei progetti