Compactação

A compactação opera com base numa MessageIndex — uma visão estruturada da lista plana de mensagens que agrupa as mensagens em unidades atómicas chamadas MessageGroup instâncias. Cada grupo regista a sua contagem de mensagens, número de bytes e estimativa de tokens.

Grupos de mensagens

A MessageGroup representa mensagens logicamente relacionadas que devem ser mantidas ou removidas juntas. Por exemplo, uma mensagem de assistente contendo chamadas de ferramenta e as correspondentes mensagens de resultado de ferramenta formam um grupo atómico — remover uma sem a outra causaria erros na API do LLM.

Cada grupo tem um MessageGroupKind:

Acionadores

A CompactionTrigger é um delegado que avalia se a compactação deve prosseguir 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 vs. alvo

Cada estratégia tem dois predicados:

• Trigger — Controla quando a compactação começa. Se o gatilho devolver false, a estratégia é completamente ignorada.
• Alvo — Controla quando a compactação para. As estratégias excluem de forma incremental grupos e reavaliam o alvo após cada etapa, parando assim que o alvo retorna true.

Quando nenhum destino é especificado, por padrão passa para o oposto do gatilho — a compactação para assim que a condição do gatilho já não se aplica.

A compactação opera numa lista plana de Message objetos. As mensagens são anotadas com metadados leves de grupo, e as estratégias alteram essas anotações para marcar grupos como excluídos antes de a lista de mensagens ser projetada para o modelo.

Grupos de mensagens

As mensagens são agrupadas em unidades atómicas. Cada grupo é atribuído a um GroupKind:

Estratégias de compactação

A CompactionStrategy é um protocolo — qualquer async chamável que aceite a list[Message] e o mute no lugar, regressando True quando alterou algo:

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

Tokenizador

Estratégias sensíveis a tokens aceitam uma implementação TokenizerProtocol. O integrado CharacterEstimatorTokenizer utiliza uma heurística de 4 caracteres por token:

from agent_framework._compaction import CharacterEstimatorTokenizer

tokenizer = CharacterEstimatorTokenizer()

Passe um tokenizador personalizado quando precisar de 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 respeita um MinimumPreserved piso que protege da remoção os grupos mais recentes não-sistémicos.

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

Estratégia de Truncamento e Compactação

Estratégia de Truncação

• MinimumPreserved assume 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 um tokenizer é fornecido, a métrica é a contagem de tokens; caso contrário, a métrica é a contagem de mensagens.
• preserve_system assume 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 utilizador mais antigos e os seus grupos de resposta associados, operando sobre limites lógicos de turnos em vez de grupos individuais.

• Um turno começa com uma mensagem de utilizador e inclui todos os grupos subsequentes de assistentes e chamadas de ferramenta até à próxima mensagem do utilizador.
• MinimumPreserved por defeito, ( 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-se apenas os grupos mais recentes keep_last_groups que não são do sistema, excluindo tudo o que é mais antigo.

• preserve_system assume 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 com múltiplas mensagens (chamada assistente + resultados da ferramenta) por um resumo curto como [Tool calls: get_weather, search_docs].
• MinimumPreserved configura-se por padrão para 2, garantindo que as interações da ferramenta no turno atual permaneçam visíveis.

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

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

from agent_framework._compaction import ToolResultCompactionStrategy

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

Estratégia de Compactação por Resumo

Estratégia de Sumarização

• Protege as mensagens do sistema e os grupos mais recentes que não são de sistema MinimumPreserved (por defeito: 4).
• Envia as mensagens mais antigas para uma localização separada IChatClient com um prompt de resumo, depois insere o resumo como um grupo 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);

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.");

• Os disparadores ativam-se quando a contagem de mensagens não do sistema incluídas excede target_count + threshold.
• Mantém as mensagens mais target_count recentes; resume tudo o que é 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.",
)

Estratégia de Compactação de Pipeline

Compõe múltiplas estratégias num pipeline sequencial. Cada estratégia funciona com base no resultado da anterior, permitindo uma compactação em camadas de suave a agressiva.

• O próprio gatilho do pipeline é CompactionTriggers.Always — cada estratégia filho avalia o seu próprio gatilho de forma independente.
• As estratégias executam-se por ordem, por isso 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 fluxo de trabalho

1. Colapsa resultados de ferramentas antigas (suavemente).
2. Resume os períodos de conversa mais antigos (moderado).
3. Mantém apenas as últimas 4 interações do utilizador (agressivo).
4. Remove grupos antigos se ainda estiver acima do orçamento (salvaguarda de emergência).

Estratégia de Compactação de Chamadas de Ferramenta Seletiva

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

• Não toca nas mensagens do utilizador nem nas mensagens simples do assistente.
• O melhor é quando a conversa das ferramentas domina o uso dos tokens e o histórico completo da ferramenta 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 múltiplas estratégias num pipeline sequencial impulsionado por um orçamento de tokens. Cada estratégia de criança corre por ordem, parando mais cedo assim que o orçamento está satisfeito. Um plano B incorporado exclui os grupos mais antigos se as estratégias sozinhas não conseguirem atingir o alvo.

• As estratégias são executadas por ordem; Coloque as estratégias mais suaves em primeiro lugar.
• early_stop=True (o padrão) para assim que o orçamento dos tokens seja satisfeito.

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 fluxo de trabalho:

1. Colapsa resultados de ferramentas antigas (suavemente).
2. Resume os períodos de conversa mais antigos (moderado).
3. Mantém apenas os últimos 20 grupos (agressivo).
4. Retoma a exclusão pelo critério do mais antigo primeiro, caso ainda ultrapasse o orçamento (salvaguarda de emergência).

Envie uma estratégia de compactação em CompactionProvider e registe-a como um AIContextProvider. Passe uma única estratégia ou um PipelineCompactionStrategy ao construtor.

Registo na API do construtor

Registe o fornecedor no ChatClientBuilder usando UseAIContextProviders. O fornecedor opera dentro do ciclo de chamada de ferramentas, compactando mensagens antes de cada chamada de 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 for necessária apenas uma estratégia, passá-la diretamente para CompactionProvider sem a envolver num PipelineCompactionStrategy:

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

Registar-se através ChatClientAgentOptions

O prestador 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ária sem uma sessão ativa de agente:

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

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

• before_strategy — executa-se antes da chamada ao modelo, comprimindo mensagens já carregadas no contexto.
• after_strategy — executa após a chamada de modelo, compactando as mensagens armazenadas pelo fornecedor de histórico para que o próximo turno comece mais pequeno.
• history_source_id — o source_id do fornecedor de histórico cujas mensagens after_strategy armazenadas devem compactar (por defeito para "in_memory").

Registo 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 for necessária apenas uma estratégia, passá-la diretamente como before_strategy:

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

Após cada execução, compactar o histórico persistente

Use after_strategy para compactar as mensagens armazenadas pelo fornecedor de histórico, de modo a que os turnos futuros 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 ativa de agente:

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