Compactação

A compactação opera em uma MessageIndex exibição estruturada da lista de mensagens simples que agrupa mensagens em unidades atômicas chamadas MessageGroup instâncias. Cada grupo controla sua contagem de mensagens, contagem de bytes e contagem de tokens estimada.

Grupos de mensagens

Um MessageGroup representa mensagens logicamente relacionadas que devem ser mantidas ou removidas juntas. Por exemplo, uma mensagem de assistente que contém chamadas de ferramenta e suas mensagens de resultado de ferramenta correspondentes formam um grupo atômico, e remover uma sem a outra causaria erros de API LLM.

Cada grupo tem um MessageGroupKind:

Gatilhos

Um CompactionTrigger é um delegado que avalia se a compactação deve continuar com base nas métricas atuais MessageIndex :

public delegate bool CompactionTrigger(MessageIndex index);

A CompactionTriggers classe fornece métodos comuns de fábrica:

Combinar gatilhos com CompactionTriggers.All(...) (AND lógico) ou CompactionTriggers.Any(...) (OR lógico):

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

Gatilho versus alvo

Cada estratégia tem dois predicados:

• Gatilho – Controla quando a compactação começa. Se o gatilho retornar false, a estratégia será totalmente ignorada.
• Destino – Controla quando a compactação é interrompida. As estratégias excluem incrementalmente grupos e reavaliam o alvo após cada etapa, parando assim que o alvo retornar true.

Quando nenhum destino é especificado, ele usa como padrão o inverso do gatilho – a compactação é interrompida assim que a condição de gatilho não é mais disparada.

A compactação opera em uma lista simples de Message objetos. As mensagens são anotadas com metadados de grupo leves e as estratégias modificam essas anotações em vigor para marcar grupos como excluídos antes que a lista de mensagens seja projetada para o modelo.

Grupos de mensagens

As mensagens são agrupadas em unidades atômicas. Cada grupo recebe um GroupKind:

Estratégias de compactação

Um CompactionStrategy é um protocolo - qualquer async chamável que aceite um list[Message] e o altere no local, retornando True quando tenha mudado qualquer coisa:

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

Gerador de token

Estratégias com reconhecimento de tokens aceitam uma TokenizerProtocol implementação. O interno CharacterEstimatorTokenizer usa uma heurística de 4 caracteres por token:

from agent_framework._compaction import CharacterEstimatorTokenizer

tokenizer = CharacterEstimatorTokenizer()

Passe um tokenizador personalizado quando for necessário obter contagens precisas de tokens para a codificação de um modelo específico.

Todas as estratégias herdam da classe base abstrata CompactionStrategy . Cada estratégia preserva as mensagens do sistema e mantém uma base MinimumPreserved que protege os grupos não-sistêmicos mais recentes contra remoção.

As estratégias de compressão são importadas de agent_framework._compaction.

TruncationCompactionStrategy

TruncationStrategy (EstratégiaDeTruncamento)

• MinimumPreserved usa como padrão 32.

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

• Quando uma tokenizer é fornecida, a métrica é a contagem de tokens; caso contrário, ela está incluída na contagem de mensagens.
• preserve_system usa como padrão 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(),
)

Estratégia de Compactação Janela Deslizante

Estratégia de Janela Deslizante

Remove os turnos de usuário mais antigos e seus grupos de resposta associados, operando em limites lógicos de turnos em vez de em grupos individuais.

• Um turno começa com uma mensagem do usuário e inclui todos os grupos de assistente e chamadas de ferramenta subsequentes até a próxima mensagem do usuário.
• MinimumPreserved é definido como 1 (preserva, pelo menos, o grupo não-sistema mais recente).

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

Mantém apenas os grupos não-sistema mais recentes keep_last_groups , excluindo tudo o que é mais antigo.

• preserve_system usa como padrão True.

from agent_framework._compaction import SlidingWindowStrategy

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

• Substitui grupos de chamadas de ferramentas que envolvem várias mensagens (incluindo chamada de assistente e resultados da ferramenta) por um resumo conciso, como [Tool calls: get_weather, search_docs].
• O padrão do MinimumPreserved é 2, garantindo que as interações de ferramenta do turno atual permaneçam visíveis.

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

• Colapse em mensagens de resumo compactas, como [Tool results: get_weather: sunny, 18°C].
• Os grupos de chamadas de ferramentas mais recentes keep_last_tool_call_groups permanecem inalterados.

from agent_framework._compaction import ToolResultCompactionStrategy

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

SummarizationCompactionStrategy

Estratégia de Resumir

• Protege as mensagens do sistema e os grupos não-sistema mais recentes MinimumPreserved (padrão: 4).
• Envia as mensagens mais antigas para um local separado IChatClient com um prompt de resumo e, em seguida, insere o resumo como MessageGroupKind.Summary grupo.

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

Você pode fornecer um prompt de resumo personalizado:

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

• Dispara quando a contagem de mensagens não sistêmicas incluídas excede target_count + threshold.
• Retém as mensagens mais recentes target_count; resume tudo que é mais antigo.
• Requer um SupportsChatGetResponse cliente.

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

Forneça um prompt de resumo personalizado:

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

PipelineCompactionStrategy

Compõe várias estratégias em um pipeline sequencial. Cada estratégia opera com base no resultado da anterior, permitindo uma compactação em camadas que vai de leve a agressiva.

• O próprio gatilho do pipeline é CompactionTriggers.Always, cada estratégia filha avalia seu próprio gatilho de forma independente.
• As estratégias são executadas em ordem, portanto, coloque as estratégias mais suaves em primeiro lugar.

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

Este pipeline (fluxo de processo):

1. Recolhe resultados antigos da ferramenta (com delicadeza).
2. Resume intervalos de conversa mais antigos (moderado).
3. Mantém apenas as últimas 4 voltas de usuário (agressivas).
4. Descarta os grupos mais antigos se ainda estiver acima do orçamento (backstop de emergência).

SelectiveToolCallCompactionStrategy

Exclui totalmente grupos de chamadas de ferramentas mais antigos, mantendo apenas o último keep_last_tool_call_groups.

• Não toca mensagens de usuário ou mensagens de assistente simples.
• Melhor quando as conversas de ferramentas dominam o uso de token e o histórico completo de ferramentas não é necessário.

from agent_framework._compaction import SelectiveToolCallCompactionStrategy

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

TokenBudgetComposedStrategy

Compõe várias estratégias em um pipeline sequencial impulsionado por um orçamento de tokens. Cada estratégia filial é executada em ordem, parando mais cedo quando o orçamento for atendido. Um recurso interno de reserva exclui os grupos mais antigos se as estratégias por si só não puderem atingir o destino.

• As estratégias devem ser executadas em ordem; coloque primeiro as estratégias menos agressivas.
• early_stop=True (o padrão) para assim que o limite de tokens é atingido.

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

Este processo:

1. Recolhe resultados antigos da ferramenta de forma suave.
2. Resume intervalos de conversa mais antigos (moderado).
3. Mantém apenas os últimos 20 grupos (agressivos).
4. Volta à exclusão mais antiga se ainda estiver acima do orçamento (backstop de emergência).

Encapsular uma estratégia de compactação em uma CompactionProvider e registrá-la como um AIContextProvider. Passe uma única estratégia ou um PipelineCompactionStrategy ao construtor.

Registrando-se com a API do construtor

Registre o provedor no ChatClientBuilder usando UseAIContextProviders. O provedor é executado dentro do loop de chamada da ferramenta, compactando mensagens antes de cada chamada ao 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 apenas uma estratégia for necessária, passe-a diretamente para CompactionProvider sem envolvê-la em um PipelineCompactionStrategy.

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

Registrando-se por meio ChatClientAgentOptions

O provedor também pode ser especificado diretamente em ChatClientAgentOptions.AIContextProviders:

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

Compactação ad hoc

CompactionProvider.CompactAsync aplica uma estratégia a uma lista de mensagens arbitrárias sem uma sessão de agente ativo:

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

CompactionProvider é um provedor de contexto que aplica estratégias de compactação antes e depois da execução de cada agente. Adicione-o juntamente com um provedor de histórico na lista do agente context_providers.

• before_strategy — é executado antes da chamada de modelo, compactando mensagens já carregadas no contexto.
• after_strategy — é executado após a chamada do modelo, compactando as mensagens armazenadas pelo provedor de histórico para que a próxima interação comece de forma mais compacta.
• history_source_id — o source_id provedor de histórico cujas mensagens after_strategy armazenadas devem ser compactadas (o padrão é "in_memory").

Registrando-se com um 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 apenas uma estratégia for necessária, passe-a diretamente como before_strategy:

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

Compactação do histórico persistente após cada execução

Use after_strategy para compactar as mensagens armazenadas pelo provedor de histórico para que as voltas futuras comecem com um contexto reduzido:

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

Compactação ad hoc

apply_compaction aplica uma estratégia a uma lista de mensagens arbitrária fora de uma sessão de agente ativo:

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