Compattazione

La compattazione opera su un oggetto MessageIndex, ovvero una visualizzazione strutturata dell'elenco di messaggi flat che raggruppa i messaggi in unità atomiche chiamate MessageGroup istanze. Ogni gruppo tiene traccia del numero di messaggi, del numero di byte e del numero di token stimato.

Gruppi di messaggi

Un MessageGroup oggetto rappresenta messaggi correlati logicamente che devono essere mantenuti o rimossi insieme. Ad esempio, un messaggio di assistente contenente le chiamate agli strumenti e i messaggi di risultato dello strumento corrispondenti formano un gruppo atomico, rimuovendone uno senza l'altro causerebbe errori dell'API LLM.

Ogni gruppo ha un MessageGroupKind:

Attivatori

Un CompactionTrigger è un delegato che valuta se la compattazione deve continuare in base alle metriche correnti MessageIndex :

public delegate bool CompactionTrigger(MessageIndex index);

La CompactionTriggers classe fornisce metodi factory comuni:

Combinare i trigger con l'operatore logico CompactionTriggers.All(...) (AND) o l'operatore logico CompactionTriggers.Any(...) (OR):

// Compact only when there are tool calls AND tokens exceed 2000
CompactionTrigger trigger = CompactionTriggers.All(
    CompactionTriggers.HasToolCalls(),
    CompactionTriggers.TokensExceed(2000));

Trigger e obiettivo

Ogni strategia ha due predicati:

• Trigger : controlla l'avvio della compattazione. Se il trigger restituisce false, la strategia viene ignorata completamente.
• Target : controlla quando si arresta la compattazione. Le strategie escludono in modo incrementale i gruppi e rivalutano l'obiettivo dopo ogni passaggio, interrompere non appena l'obiettivo ritorna true.

Quando non viene specificata alcuna destinazione, l'impostazione predefinita è l'inversa del trigger. La compattazione si arresta non appena la condizione del trigger non viene più attivata.

La compattazione opera su un elenco semplice di Message oggetti. I messaggi vengono annotati con metadati di gruppo leggeri e le strategie modificano tali annotazioni sul posto per contrassegnare i gruppi come esclusi prima che l'elenco dei messaggi venga proiettato nel modello.

Gruppi di messaggi

I messaggi vengono raggruppati in unità atomiche. A ogni gruppo viene assegnato un oggetto GroupKind:

Strategie di compattazione

Un CompactionStrategy è un protocollo: qualsiasi async chiamabile che accetta un oggetto list[Message] e lo modifica sul posto, restituendo True quando è cambiato qualcosa:

class CompactionStrategy(Protocol):
    async def __call__(self, messages: list[Message]) -> bool: ...

Tokenizzatore

Le strategie con riconoscimento dei token accettano un'implementazione TokenizerProtocol . L'interfaccia predefinita CharacterEstimatorTokenizer usa un'euristica a 4 caratteri per token:

from agent_framework._compaction import CharacterEstimatorTokenizer

tokenizer = CharacterEstimatorTokenizer()

Utilizza un tokenizzatore personalizzato quando hai bisogno di conteggi di token accurati per la codifica di un modello specifico.

Tutte le strategie ereditano dalla classe base astratta CompactionStrategy . Ogni strategia mantiene i messaggi di sistema e rispetta una MinimumPreserved base che protegge i gruppi non di sistema più recenti dalla rimozione.

Le strategie di compattazione vengono importate da agent_framework._compaction.

TruncationCompactionStrategy

StrategiaTroncamento

• Il valore predefinito di MinimumPreserved è 32.

// Drop oldest groups when tokens exceed 32K, keeping at least 10 recent groups
TruncationCompactionStrategy truncation = new(
    trigger: CompactionTriggers.TokensExceed(0x8000),
    minimumPreserved: 10);

• Quando viene fornito un tokenizer, la metrica è il numero di token; in caso contrario, viene incluso il conteggio dei messaggi.
• Il valore predefinito di preserve_system è True.

from agent_framework._compaction import CharacterEstimatorTokenizer, TruncationStrategy

# Exclude oldest groups when tokens exceed 32 000, trimming to 16 000
truncation = TruncationStrategy(
    max_n=32_000,
    compact_to=16_000,
    tokenizer=CharacterEstimatorTokenizer(),
)

SlidingWindowCompactionStrategy

SlidingWindowStrategy

Rimuove i turni dell'utente meno recenti e i relativi gruppi di risposta associati, operando sui limiti dei turni logici anziché sui singoli gruppi.

• Un turno inizia con un messaggio utente e include tutti i gruppi di assistente e chiamata agli strumenti successivi fino al messaggio utente successivo.
• MinimumPreserved predefinito a 1 (mantiene almeno il gruppo non di sistema più recente).

// Keep only the last 4 user turns
SlidingWindowCompactionStrategy slidingWindow = new(
    trigger: CompactionTriggers.TurnsExceed(4));

Mantiene solo i gruppi non di sistema più recenti keep_last_groups , esclusi tutti gli elementi precedenti.

• Il valore predefinito di preserve_system è True.

from agent_framework._compaction import SlidingWindowStrategy

# Keep only the last 20 non-system groups
sliding_window = SlidingWindowStrategy(keep_last_groups=20)

• Sostituisce i gruppi di chiamate dello strumento multi-messaggio (chiamata assistente + risultati dello strumento) con un breve riepilogo, ad esempio [Tool calls: get_weather, search_docs].
• MinimumPreserved è impostato su 2, assicurando che le interazioni dello strumento del turno corrente rimangano visibili.

// Collapse old tool results when tokens exceed 512
ToolResultCompactionStrategy toolCompaction = new(
    trigger: CompactionTriggers.TokensExceed(0x200));

• Comprime in messaggi di riepilogo compatti, ad esempio [Tool results: get_weather: sunny, 18°C].
• I gruppi di chiamate agli strumenti più recenti keep_last_tool_call_groups vengono lasciati invariati.

from agent_framework._compaction import ToolResultCompactionStrategy

# Collapse all but the newest tool-call group
tool_result = ToolResultCompactionStrategy(keep_last_tool_call_groups=1)

RiepilogoCompactionStrategy

Strategia di Riepilogo

• Protegge i messaggi di sistema e i gruppi non di sistema più recenti MinimumPreserved (impostazione predefinita: 4).
• Invia i messaggi meno recenti a una destinazione separata IChatClient con un prompt di riepilogo, successivamente inserisce il riepilogo come gruppo MessageGroupKind.Summary.

// Summarize older messages when tokens exceed 1280, keeping the last 4 groups
SummarizationCompactionStrategy summarization = new(
    chatClient: summarizerChatClient,
    trigger: CompactionTriggers.TokensExceed(0x500),
    minimumPreserved: 4);

È possibile fornire un prompt di riepilogo personalizzato:

SummarizationCompactionStrategy summarization = new(
    chatClient: summarizerChatClient,
    trigger: CompactionTriggers.TokensExceed(0x500),
    summarizationPrompt: "Summarize the key decisions and user preferences only.");

• Si attiva quando il conteggio dei messaggi non di sistema incluso supera target_count + threshold.
• Mantiene i messaggi più recenti target_count . Riepiloga tutti gli elementi precedenti.
• Richiede un SupportsChatGetResponse client.

from agent_framework._compaction import SummarizationStrategy

# Summarize when non-system message count exceeds 6, retaining the 4 newest
summarization = SummarizationStrategy(
    client=summarizer_client,
    target_count=4,
    threshold=2,
)

Fornire un prompt di riepilogo personalizzato:

summarization = SummarizationStrategy(
    client=summarizer_client,
    target_count=4,
    prompt="Summarize the key decisions and user preferences only.",
)

PipelineCompactionStrategy

Compone più strategie in una pipeline sequenziale. Ogni strategia opera sul risultato della precedente, consentendo la compattazione stratificata dal delicato all'aggressivo.

• Il trigger della pipeline è CompactionTriggers.Always : ogni strategia figlia valuta il proprio trigger in modo indipendente.
• Le strategie si eseguono in ordine, quindi mettere prima le strategie più delicate.

PipelineCompactionStrategy pipeline = new(
    new ToolResultCompactionStrategy(CompactionTriggers.TokensExceed(0x200)),
    new SummarizationCompactionStrategy(summarizerChatClient, CompactionTriggers.TokensExceed(0x500)),
    new SlidingWindowCompactionStrategy(CompactionTriggers.TurnsExceed(4)),
    new TruncationCompactionStrategy(CompactionTriggers.TokensExceed(0x8000)));

Questa pipeline:

1. Comprime i risultati dei vecchi strumenti (delicato).
2. Riepiloga gli intervalli di conversazione meno recenti (moderata).
3. Mantiene solo gli ultimi 4 turni utente (aggressivi).
4. Elimina i gruppi meno recenti se ancora oltre il budget (backstop di emergenza).

Strategia di Compattazione delle Chiamate di Strumenti Selettivi

Esclude i gruppi di chiamate agli strumenti meno recenti completamente, conservando solo l'ultimo oggetto keep_last_tool_call_groups.

• Non tocca i messaggi utente o assistente normale.
• Meglio quando lo strumento chatter domina l'utilizzo dei token e la cronologia completa degli strumenti non è necessaria.

from agent_framework._compaction import SelectiveToolCallCompactionStrategy

# Keep only the most recent tool-call group
selective_tool = SelectiveToolCallCompactionStrategy(keep_last_tool_call_groups=1)

TokenBudgetComposedStrategy

Compone più strategie in una pipeline sequenziale basata su un budget di token. Ogni strategia figlia viene eseguita in ordine, interrompendosi anticipatamente una volta soddisfatto il budget. Un meccanismo di fallback incorporato esclude i gruppi meno recenti se le strategie da sole non riescono a raggiungere l'obiettivo.

• Eseguire le strategie in ordine; posizionare prima le strategie più lievi.
• early_stop=True (impostazione predefinita) si arresta non appena viene soddisfatto il budget dei token.

from agent_framework._compaction import (
    CharacterEstimatorTokenizer,
    SelectiveToolCallCompactionStrategy,
    SlidingWindowStrategy,
    SummarizationStrategy,
    TokenBudgetComposedStrategy,
    ToolResultCompactionStrategy,
)

tokenizer = CharacterEstimatorTokenizer()

pipeline = TokenBudgetComposedStrategy(
    token_budget=16_000,
    tokenizer=tokenizer,
    strategies=[
        ToolResultCompactionStrategy(keep_last_tool_call_groups=1),
        SummarizationStrategy(client=summarizer_client, target_count=4, threshold=2),
        SlidingWindowStrategy(keep_last_groups=20),
    ],
)

Questa pipeline:

1. Comprime i risultati dei vecchi strumenti (in modo graduale).
2. Riepiloga gli intervalli di conversazione meno recenti (moderata).
3. Mantiene solo gli ultimi 20 gruppi (aggressivi).
4. Torna all'esclusione meno recente se ancora oltre il budget (backstop di emergenza).

Incapsulare una strategia di compattazione in un CompactionProvider e registrarla come AIContextProvider. Passa una singola strategia o un PipelineCompactionStrategy al costruttore.

Registrazione con l'API del generatore

Registrare il fornitore nell'oggetto ChatClientBuilder utilizzando UseAIContextProviders. Il provider viene eseguito all'interno del ciclo di chiamata degli strumenti, compattando i messaggi prima di ogni chiamata LLM.

IChatClient agentChatClient = openAIClient.GetChatClient(deploymentName).AsIChatClient();
IChatClient summarizerChatClient = openAIClient.GetChatClient(deploymentName).AsIChatClient();

PipelineCompactionStrategy compactionPipeline =
    new(
        new ToolResultCompactionStrategy(CompactionTriggers.TokensExceed(0x200)),
        new SummarizationCompactionStrategy(summarizerChatClient, CompactionTriggers.TokensExceed(0x500)),
        new SlidingWindowCompactionStrategy(CompactionTriggers.TurnsExceed(4)),
        new TruncationCompactionStrategy(CompactionTriggers.TokensExceed(0x8000)));

AIAgent agent =
    agentChatClient
        .AsBuilder()
        .UseAIContextProviders(new CompactionProvider(compactionPipeline))
        .BuildAIAgent(
            new ChatClientAgentOptions
            {
                Name = "ShoppingAssistant",
                ChatOptions = new()
                {
                    Instructions = "You are a helpful shopping assistant.",
                    Tools = [AIFunctionFactory.Create(LookupPrice)],
                },
            });

AgentSession session = await agent.CreateSessionAsync();
Console.WriteLine(await agent.RunAsync("What's the price of a laptop?", session));

Se è necessaria una sola strategia, passarla direttamente a CompactionProvider senza inserirla in un oggetto PipelineCompactionStrategy.

agentChatClient
    .AsBuilder()
    .UseAIContextProviders(new CompactionProvider(
        new SlidingWindowCompactionStrategy(CompactionTriggers.TurnsExceed(20))))
    .BuildAIAgent(...);

Registrazione tramite ChatClientAgentOptions

Il provider può anche essere specificato direttamente in ChatClientAgentOptions.AIContextProviders:

AIAgent agent = agentChatClient
    .AsBuilder()
    .BuildAIAgent(new ChatClientAgentOptions
    {
        AIContextProviders = [new CompactionProvider(compactionPipeline)]
    });

Compattazione ad hoc

CompactionProvider.CompactAsync applica una strategia a un elenco di messaggi arbitrari senza una sessione agente attiva:

IEnumerable<ChatMessage> compacted = await CompactionProvider.CompactAsync(
    new TruncationCompactionStrategy(CompactionTriggers.TokensExceed(8000)),
    existingMessages);

CompactionProvider è un provider di contesto che applica strategie di compattazione prima e dopo l'esecuzione di ogni agente. Aggiungerlo insieme a un provider di cronologia nell'elenco dell'agente context_providers .

• before_strategy : viene eseguito prima della chiamata del modello, compattando i messaggi già caricati nel contesto.
• after_strategy: viene eseguito dopo la chiamata al modello, compattando i messaggi archiviati dal fornitore di cronologia in modo che il turno successivo risulti più piccolo.
• history_source_id — il fornitore della cronologia i cui messaggi archiviati after_strategy devono essere compattati (valore predefinito "in_memory").

Registrazione con un agente

from agent_framework import Agent, CompactionProvider, InMemoryHistoryProvider
from agent_framework._compaction import (
    CharacterEstimatorTokenizer,
    SlidingWindowStrategy,
    SummarizationStrategy,
    TokenBudgetComposedStrategy,
    ToolResultCompactionStrategy,
)

tokenizer = CharacterEstimatorTokenizer()

pipeline = TokenBudgetComposedStrategy(
    token_budget=16_000,
    tokenizer=tokenizer,
    strategies=[
        ToolResultCompactionStrategy(keep_last_tool_call_groups=1),
        SummarizationStrategy(client=summarizer_client, target_count=4, threshold=2),
        SlidingWindowStrategy(keep_last_groups=20),
    ],
)

history = InMemoryHistoryProvider()
compaction = CompactionProvider(
    before_strategy=pipeline,
    history_source_id=history.source_id,
)

agent = Agent(
    client=client,
    name="ShoppingAssistant",
    instructions="You are a helpful shopping assistant.",
    context_providers=[history, compaction],
)

session = agent.create_session()
print(await agent.run("What's the price of a laptop?", session=session))

Se è necessaria una sola strategia, passarla direttamente come before_strategy:

compaction = CompactionProvider(
    before_strategy=SlidingWindowStrategy(keep_last_groups=20),
    history_source_id=history.source_id,
)

Compattazione della cronologia persistente dopo ogni esecuzione

Usare after_strategy per compattare i messaggi archiviati dal provider di cronologia in modo che i turni futuri inizino con un contesto ridotto:

compaction = CompactionProvider(
    before_strategy=SlidingWindowStrategy(keep_last_groups=20),
    after_strategy=ToolResultCompactionStrategy(keep_last_tool_call_groups=1),
    history_source_id=history.source_id,
)

Compattazione ad hoc

apply_compaction applica una strategia a un elenco di messaggi arbitrari all'esterno di una sessione dell'agente attivo:

from agent_framework._compaction import apply_compaction, TruncationStrategy, CharacterEstimatorTokenizer

tokenizer = CharacterEstimatorTokenizer()

compacted = await apply_compaction(
    messages,
    strategy=TruncationStrategy(
        max_n=8_000,
        compact_to=4_000,
        tokenizer=tokenizer,
    ),
    tokenizer=tokenizer,
)