Compactage

Le compactage fonctionne sur une MessageIndex — une vue structurée d'une liste plate de messages qui regroupe les messages en unités atomiques, appelées instances de MessageGroup. Chaque groupe suit son nombre de messages, son nombre d’octets et son nombre estimé de jetons.

Groupes de messages

Un MessageGroup représente les messages liés logiquement qui doivent être conservés ou supprimés ensemble. Par exemple, un message d’assistant contenant des appels d’outil et ses messages de résultat d’outil correspondants forment un groupe atomique : la suppression d’un message sans l’autre entraînerait des erreurs d’API LLM.

Chaque groupe a un MessageGroupKind:

Triggers

Un CompactionTrigger est un délégué qui évalue si le compactage devrait se poursuivre en fonction des métriques actuelles MessageIndex.

public delegate bool CompactionTrigger(MessageIndex index);

La CompactionTriggers classe fournit des méthodes de fabrique courantes :

Combiner des déclencheurs avec CompactionTriggers.All(...) (AND logique) ou CompactionTriggers.Any(...) (OR logique) :

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

Déclencheur et cible

Chaque stratégie a deux prédicats :

• Déclencheur : contrôle quand commence le compactage. Si le déclencheur retourne false, la stratégie est ignorée entièrement.
• Cible : contrôle quand le compactage s’arrête. Les stratégies excluent de façon incrémentielle les groupes et réévaluent la cible après chaque étape, s’arrêtant dès que la cible retourne true.

Lorsqu’aucun objectif n’est spécifié, il est défini par défaut sur l’inverse du déclencheur : le compactage s’arrête dès que la condition du déclencheur n’est plus remplie.

Le compactage fonctionne sur une liste plate d’objets Message. Les messages sont annotés avec des métadonnées de groupe léger et les stratégies mutent ces annotations en place pour marquer les groupes comme exclus avant que la liste des messages ne soit projetée vers le modèle.

Groupes de messages

Les messages sont regroupés en unités atomiques. Chaque groupe est affecté à un GroupKind:

Stratégies de compactage

Un CompactionStrategy est un protocole : tout async appelable qui accepte un list[Message] et le modifie sur place, en retournant True lorsqu'il a modifié quoi que ce soit.

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

Générateur de jetons

Les stratégies conscientes des jetons acceptent une TokenizerProtocol implémentation. L'outil intégré CharacterEstimatorTokenizer utilise une heuristique de 4 caractères par jeton :

from agent_framework._compaction import CharacterEstimatorTokenizer

tokenizer = CharacterEstimatorTokenizer()

Transmettez un tokenizer personnalisé lorsque vous avez besoin d’un nombre de jetons précis pour l’encodage d’un modèle spécifique.

Toutes les stratégies héritent de la classe de base abstraite CompactionStrategy . Chaque stratégie préserve les messages systèmes et respecte un MinimumPreserved seuil qui protège les groupes non-systémiques les plus récents contre la suppression.

Les stratégies de compactage sont importées à partir de agent_framework._compaction.

StratégieDeCompactageParTroncature

StratégieDeTroncation

• MinimumPreserved a la valeur par défaut 32.

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

• Lorsqu’un tokenizer est fourni, la métrique est le nombre de jetons ; sinon, la métrique est le nombre de messages.
• preserve_system a la valeur par défaut 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

Supprime les utilisateurs les plus anciens et leurs groupes de réponse associés, fonctionnant sur des limites de tour logique plutôt que sur des groupes individuels.

• Un tour commence par un message utilisateur et inclut tous les groupes d’assistants et d’appels d’outils suivants jusqu’au message utilisateur suivant.
• MinimumPreserved utilise par défaut 1 (ce qui préserve au moins le groupe non système le plus récent).

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

Conserve uniquement les groupes non système les plus récents keep_last_groups , à l’exclusion de tout ce qui est plus ancien.

• preserve_system a la valeur par défaut True.

from agent_framework._compaction import SlidingWindowStrategy

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

• Remplace les groupes d’appels d’outils multi-messages (appel d’assistant + résultats de l’outil) par un résumé court comme [Tool calls: get_weather, search_docs].
• MinimumPreserved par défaut à 2, ce qui garantit que les interactions de l'outil actuel restent visibles.

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

• Se réduit à des messages récapitulatifs compacts tels que [Tool results: get_weather: sunny, 18°C].
• Les groupes d’appels d’outils les plus récents keep_last_tool_call_groups sont laissés inchangés.

from agent_framework._compaction import ToolResultCompactionStrategy

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

StratégieDeCompactageParRésumé

StratégieDeRésumé

• Protège les messages système et les groupes non système les plus récents MinimumPreserved (par défaut : 4).
• Envoie les anciens messages à un autre IChatClient avec une instruction de synthèse, puis insère le résumé en tant que groupe 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);

Vous pouvez fournir une demande de synthèse personnalisée :

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

• Se déclenche quand le nombre de messages non système inclus dépasse target_count + threshold.
• Conserve les messages les plus target_count récents ; récapitule tout ce qui est plus ancien.
• Nécessite 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,
)

Fournissez une invite de synthèse personnalisée :

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

StratégieDeCompactageDuPipeline

Compose plusieurs stratégies dans un pipeline séquentiel. Chaque stratégie opère sur le résultat de la précédente, permettant une compaction progressive par couches, allant de légère à agressive.

• Le propre déclencheur du pipeline est CompactionTriggers.Always : chaque stratégie enfant évalue son propre déclencheur indépendamment.
• Les stratégies s’exécutent dans l’ordre, donc mettre les stratégies les plus douces en premier.

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

Ce pipeline :

1. Décroît les anciens résultats de l’outil (léger).
2. Résume les anciennes étendues de conversation (modérées).
3. Conserve uniquement les 4 dernières interactions utilisateur (mode agressif).
4. Supprime les groupes les plus anciens s’ils dépassent encore le budget (mesure de secours d’urgence).

SelectiveToolCallCompactionStrategy

Exclut entièrement les anciens groupes d'appels d'outils, en conservant uniquement le dernier keep_last_tool_call_groups.

• Ne touche pas aux messages d'utilisateur ou de l'assistant basique.
• Convient mieux lorsque le phénomène de vibration de l'outil prédomine l'utilisation des jetons et que l'historique complet de l'outil n'est pas nécessaire.

from agent_framework._compaction import SelectiveToolCallCompactionStrategy

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

StratégieComposéeDeBudgetDeJetons

Compose plusieurs stratégies dans un pipeline séquentiel piloté par un budget de jeton. Chaque stratégie enfant s'exécute dans l'ordre et s'arrête dès que le budget est rempli. Un mécanisme de repli intégré exclut les groupes les plus anciens si les stratégies seules ne peuvent pas atteindre l'objectif.

• Les stratégies s’exécutent dans l’ordre ; placez d’abord les stratégies les plus douces.
• early_stop=True (valeur par défaut) s’arrête dès que le budget du jeton est satisfait.

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),
    ],
)

Ce pipeline :

1. Réduit les anciens résultats de l’outil (doux).
2. Résume les anciens segments de conversation (modérée).
3. Garde uniquement les 20 derniers groupes (agressifs).
4. Revient à l'exclusion la plus ancienne si le budget est encore dépassé (mécanisme de sauvegarde d'urgence).

Encapsulez une stratégie de compactage dans un CompactionProvider et enregistrez-la en tant que AIContextProvider. Passez soit une stratégie unique, soit un PipelineCompactionStrategy au constructeur.

Inscription auprès de l’API builder

Inscrivez le fournisseur sur ChatClientBuilder en utilisant UseAIContextProviders. Le fournisseur s’exécute à l’intérieur de la boucle d’appel d’outils, compactant les messages avant chaque appel 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));

Si une seule stratégie est nécessaire, transmettez-la directement à CompactionProvider sans l’encapsuler dans un PipelineCompactionStrategy:

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

Inscription à travers ChatClientAgentOptions

Le fournisseur peut également être spécifié directement sur ChatClientAgentOptions.AIContextProviders:

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

Compactage ad hoc

CompactionProvider.CompactAsync applique une stratégie à une liste de messages arbitraires sans session d’agent active :

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

CompactionProvider est un fournisseur de contexte qui applique des stratégies de compactage avant et après l’exécution de chaque agent. Ajoutez-le en parallèle avec un fournisseur d'historique dans la liste de l'agent context_providers.

• before_strategy — s’exécute avant l’appel du modèle, compactant les messages déjà chargés dans le contexte.
• after_strategy — s’exécute après l’appel du modèle, en compactant les messages stockés par le fournisseur d’historique afin que le tour suivant démarre plus petit.
• history_source_id — le source_id fournisseur d’historique dont les messages after_strategy stockés "in_memory"doivent être compactés (par défaut).

Inscription auprès d’un agent

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))

Si une seule stratégie est nécessaire, transmettez-la directement sous la forme de before_strategy.

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

Compression de l'historique persistant après chaque exécution

Permet after_strategy de compacter les messages stockés par le fournisseur d’historique afin que les futurs tours commencent par un contexte réduit :

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

Compactage ad hoc

apply_compaction applique une stratégie à une liste de messages arbitraire en dehors d’une session d’agent active :

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,
)