Komprimierung

Die Komprimierung arbeitet auf einer MessageIndex strukturierten Ansicht der flachen Nachrichtenliste, die Nachrichten in atomare Einheiten gruppiert, die als MessageGroup-Instanzen bezeichnet werden. Jede Gruppe verfolgt die Nachrichtenanzahl, byteanzahl und die geschätzte Tokenanzahl nach.

Nachrichtengruppen

Ein MessageGroup stellt logisch zusammengehörige Nachrichten dar, die zusammen beibehalten oder entfernt werden müssen. Eine Assistentennachricht, die Toolaufrufe enthält, und die entsprechenden Toolergebnismeldungen bilden beispielsweise eine atomische Gruppe – das Entfernen eines Ohne den anderen würde ZU LLM-API-Fehlern führen.

Jede Gruppe hat ein MessageGroupKind:

Auslöser

Ein CompactionTrigger ist ein Delegat, der auswertet, ob die Komprimierung auf Basis der aktuellen MessageIndex Metriken fortgesetzt werden soll.

public delegate bool CompactionTrigger(MessageIndex index);

Die CompactionTriggers Klasse stellt allgemeine Fabrikmethoden bereit:

Kombinieren von Triggern mit CompactionTriggers.All(...) (logisches UND) oder CompactionTriggers.Any(...) (logisches ODER):

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

Trigger im Vergleich zum Ziel

Jede Strategie hat zwei Prädikate:

• Trigger – Steuerelemente , wenn die Komprimierung beginnt. Wenn der Trigger false zurückgibt, wird die gesamte Strategie übersprungen.
• Ziel – Steuert , wenn die Komprimierung beendet wird. Strategien schließen Gruppen schrittweise aus und bewerten das Ziel nach jedem Schritt neu, wobei der Prozess endet, sobald das Ziel zurückkehrt true.

Wenn kein Ziel angegeben wird, wird standardmäßig das Gegenteil des Triggers verwendet: Die Verdichtung wird beendet, sobald die Triggerbedingung nicht mehr erfüllt ist.

Komprimierung wird auf einer flachen Liste von Message Objekten ausgeführt. Nachrichten werden mit einfachen Gruppenmetadaten versehen, und Strategien mutieren diese Anmerkungen, um Gruppen als ausgeschlossen zu markieren, bevor die Nachrichtenliste auf das Modell projiziert wird.

Nachrichtengruppen

Nachrichten werden in atome Einheiten gruppiert. Jedem Gruppe wird ein GroupKind zugewiesen.

Komprimierungsstrategien

Ein CompactionStrategy ist ein Protokoll – jedes async aufrufbares Element, das es list[Message] annehmen und direkt verändern kann, zurückkehrt True, sobald etwas geändert wurde:

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

Tokenisierer

Tokenfähige Strategien akzeptieren eine TokenizerProtocol Implementierung. Das integrierte System CharacterEstimatorTokenizer verwendet eine Heuristik mit 4 Zeichen pro Token:

from agent_framework._compaction import CharacterEstimatorTokenizer

tokenizer = CharacterEstimatorTokenizer()

Übergeben Sie einen benutzerdefinierten Tokenizer, wenn Sie genaue Tokenanzahlen für die Codierung eines bestimmten Modells benötigen.

Alle Strategien erben von der abstrakten CompactionStrategy Basisklasse. Jede Strategie behält Systemmeldungen bei und respektiert eine MinimumPreserved Schwelle, die die neuesten Nicht-System-Gruppen vor der Entfernung schützt.

Komprimierungsstrategien werden aus agent_framework._compactionimportiert.

Trunkierungs-Kompressionsstrategie

Abkürzungsstrategie

• Der Standardwert von MinimumPreserved ist 32.

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

• Wenn ein tokenizer Wert angegeben wird, ist die Metrik die Tokenanzahl. Andernfalls wird die Nachrichtenanzahl eingeschlossen.
• Der Standardwert von preserve_system ist 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

Gleitfensterstrategie

Entfernt die ältesten Benutzer turns und ihre zugehörigen Antwortgruppen, wobei logische Abgrenzungen der Turns verwendet werden und nicht auf einzelne Gruppen abgezielt wird.

• Ein Vorgang beginnt mit einer Benutzernachricht und umfasst alle nachfolgenden Assistenten- und Werkzeugaufrufgruppen bis zur nächsten Benutzernachricht.
• MinimumPreserved wird standardmäßig auf 1 gesetzt (behält mindestens die neueste nicht-systemdefinierte Gruppe bei).

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

Behält nur die neuesten keep_last_groups Nicht-Systemgruppen bei, mit Ausnahme aller älteren Gruppen.

• Der Standardwert von preserve_system ist True.

from agent_framework._compaction import SlidingWindowStrategy

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

• Ersetzt Tool-Anrufgruppen mit mehreren Nachrichten (Assistentenanruf + Ergebnisse der Werkzeuge) durch eine kurze Zusammenfassung, wie [Tool calls: get_weather, search_docs].
• MinimumPreserved wird standardmäßig auf 2 zurückgesetzt, wodurch sichergestellt wird, dass die Toolinteraktionen des aktuellen Zuges sichtbar bleiben.

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

• Reduziert sich in kompakte Zusammenfassungsmeldungen wie [Tool results: get_weather: sunny, 18°C]z. B. .
• Die neuesten keep_last_tool_call_groups Toolanrufgruppen bleiben unberührt.

from agent_framework._compaction import ToolResultCompactionStrategy

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

SummarizationCompactionStrategy

Strategie zur Zusammenfassung

• Schützt Systemmeldungen und die neuesten MinimumPreserved Nicht-Systemgruppen (Standard: 4).
• Sendet die älteren Nachrichten an separates IChatClient mit einer Aufforderung zur Zusammenfassung und fügt dann die Zusammenfassung als MessageGroupKind.Summary Gruppe ein.

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

Sie können eine benutzerdefinierte Zusammenfassungsaufforderung bereitstellen:

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

• Trigger, wenn die Anzahl der nicht systemfremden target_count + thresholdNachrichten überschreitet.
• Behält die neuesten target_count Nachrichten bei; fasst alles zusammen, was älter ist.
• Erfordert einen 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,
)

Geben Sie eine benutzerdefinierte Zusammenfassungsaufforderung an:

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

Pipeline-Verdichtungsstrategie

Kombiniert mehrere Strategieansätze zu einer sequenziellen Pipeline. Jede Strategie basiert auf dem Ergebnis der vorherigen, wodurch eine abgestufte Komprimierung von sanft bis aggressiv ermöglicht wird.

• Der eigene Trigger der Pipeline lautet CompactionTriggers.Always – jede untergeordnete Strategie wertet ihren eigenen Trigger unabhängig aus.
• Strategien werden in der Reihenfolge ausgeführt, also setzen Sie die sanftsten Strategien zuerst.

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

Diese Pipeline:

1. Klappt alte Werkzeugergebnisse auf sanfte Weise zusammen.
2. Fasst ältere Gesprächszeiträume (mäßig) zusammen.
3. Behält nur die letzten 4 Benutzerwechsel bei (aggressiv).
4. Entfernt älteste Gruppen, wenn das Budget noch überschritten ist (Notfall-Backstop).

Selektive Werkzeuganruf-Komprimierungsstrategie

Ältere Werkzeugaufrufgruppen werden vollständig ausgeschlossen, wobei nur die letzte keep_last_tool_call_groups beibehalten wird.

• Berührt weder Benutzer- noch einfache Assistenten-Nachrichten.
• Am besten, wenn Der Toolchatter die Tokenverwendung dominiert und der vollständige Toolverlauf nicht benötigt wird.

from agent_framework._compaction import SelectiveToolCallCompactionStrategy

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

TokenBudgetVerbundStrategie

Kombiniert mehrere Strategien in einer von einem Token-Budget gesteuerten sequenziellen Pipeline. Jede untergeordnete Strategie wird in der Reihenfolge ausgeführt und frühzeitig beendet, sobald das Budget erfüllt ist. Ein integrierter Fallback schließt die ältesten Gruppen aus, wenn die Strategien allein das Ziel nicht erreichen können.

• Strategien werden in der Reihenfolge ausgeführt; platzieren Sie die sanftsten Strategien zuerst.
• early_stop=True (Standardeinstellung) wird beendet, sobald das Tokenbudget erfüllt ist.

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

Diese Pipeline:

1. Klappt alte Werkzeugergebnisse zusammen (sanft).
2. Fasst ältere Unterhaltungen (mäßig) zusammen.
3. Speichert nur die letzten 20 Gruppen (aggressive Vorgehensweise).
4. Fällt auf den ältesten Ausschluss zurück, wenn der Haushalt noch über dem Budget liegt (Notfall-Backstop).

Umschließen Sie eine Komprimierungsstrategie in einer CompactionProvider und registrieren Sie sie als eine AIContextProvider. Übergeben Sie entweder eine einzelne Strategie oder eine PipelineCompactionStrategy an den Konstruktor.

Registrieren mit der Generator-API

Registrieren Sie den Anbieter auf ChatClientBuilder mit UseAIContextProviders. Der Anbieter wird innerhalb der Werkzeugaufrufschleife ausgeführt und komprimiert Nachrichten vor jedem LLM-Aufruf.

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

Wenn nur eine Strategie benötigt wird, übergeben Sie sie direkt an CompactionProvider, ohne dass Sie sie in eine PipelineCompactionStrategy umschließen müssen.

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

Registrieren über ChatClientAgentOptions

Der Anbieter kann auch direkt bei ChatClientAgentOptions.AIContextProviders spezifiziert werden.

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

Ad-hoc-Komprimierung

CompactionProvider.CompactAsync wendet eine Strategie auf eine beliebige Nachrichtenliste ohne aktive Agentsitzung an:

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

CompactionProvider ist ein Kontextanbieter, der Komprimierungsstrategien vor und nach jeder Agentausführung anwendet. Fügen Sie ihn zusammen mit einem Historienanbieter in die Liste des context_providers-Agents hinzu.

• before_strategy — wird vor dem Modellaufruf ausgeführt und komprimiert Nachrichten, die bereits in den Kontext geladen wurden.
• after_strategy — wird nach dem Modellaufruf ausgeführt und komprimiert die vom Verlaufsanbieter gespeicherten Nachrichten, so dass der nächste Durchlauf kleiner startet.
• history_source_id — des source_id Verlaufsanbieters, dessen gespeicherte Nachrichten after_strategy komprimiert werden sollen (Standardeinstellung "in_memory").

Registrieren bei einem 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))

Wenn nur eine Strategie benötigt wird, übergeben Sie sie direkt als before_strategy.

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

Komprimieren des gespeicherten Verlaufs nach jeder Ausführung

Verwenden Sie after_strategy, um die vom Verlaufsanbieter gespeicherten Nachrichten zu komprimieren, damit zukünftige Schritte mit einem reduzierten Kontext beginnen.

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

Ad-hoc-Komprimierung

apply_compaction wendet eine Strategie auf eine beliebige Nachrichtenliste außerhalb einer aktiven Agentsitzung an:

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