共用方式為


壓縮

隨著對話增多,聊天歷史的代碼數量可能超過模型的上下文窗口,或導致成本上升。 壓縮策略在保留重要上下文的同時,縮小對話歷史的大小,使代理人能在長時間互動中持續運作。

這很重要

壓縮框架目前仍處於實驗階段。 若要使用它,您必須新增 #pragma warning disable MAAI001

這很重要

壓縮框架目前仍在 Python 中進行實驗。 從agent_framework._compaction中匯入策略。

為什麼壓縮很重要

每次呼叫大型語言模型時都會包含完整的對話歷史。 無壓縮:

  • 代幣限制 — 對話最終超過模型的上下文視窗,導致錯誤。
  • 成本 — 較大的提示詞會消耗更多代幣,導致 API 成本增加。
  • 延遲 — 輸入標記越多,回應時間越慢。

壓縮透過有選擇地移除、摺疊或摘要較舊的對話部分來解決這些問題。

核心概念

適用性:僅限於記憶體內歷史代理

壓縮僅適用於在記憶體中管理自己對話歷史的代理。 依賴服務管理上下文或對話狀態的代理程式不會從壓縮中受益,因為服務本身已經處理上下文管理。 服務管理代理的範例包括:

  • Foundry 代理 — 情境由 Azure AI Foundry 服務在伺服器端管理。
  • 啟用儲存(預設)的回應 API — 對話狀態由 OpenAI 服務儲存和管理。
  • Copilot Studio 客服 人員 — 對話上下文由 Copilot Studio 服務維護。

對於這些代理類型,配置壓縮策略不會產生影響。 壓縮只有在代理維護自身記憶體訊息清單並在每次呼叫中將完整歷史傳給模型時才有意義。

壓縮是基於 MessageIndex 一種結構化的扁平訊息清單視圖,將訊息分組成稱為 MessageGroup 實例的原子單元。 每個群組追蹤其訊息數量、位元組數及估計的令牌數。

訊息群組

A MessageGroup 代表邏輯相關的訊息,必須同時保留或移除。 例如,包含工具呼叫及其對應工具結果訊息的助理訊息會形成原子群組——若將其中一組移除,則會導致 LLM API 錯誤。

每個群組有一個 MessageGroupKind

種類 說明
System 一個或多個系統訊息。 在壓實過程中始終會被保存。
User 一個使用者訊息,開啟新回合。
AssistantText 此純助理文字回覆(無工具呼叫)。
ToolCall 一個包含工具呼叫及相應工具結果訊息的助理訊息,視為一個原子單元。
Summary 由摘要壓縮產生的濃縮訊息。

觸發器

A CompactionTrigger 是根據現有 MessageIndex 指標評估是否應該進行壓實的代表:

public delegate bool CompactionTrigger(MessageIndex index);

CompactionTriggers 類別提供常見的工廠方法:

Trigger 當事件觸發時
CompactionTriggers.Always 每一次(無條件地)。
CompactionTriggers.Never 絕不(禁止壓縮)。
CompactionTriggers.TokensExceed(maxTokens) 所包含的代幣數量超過門檻。
CompactionTriggers.MessagesExceed(maxMessages) 包含的訊息數量超過門檻。
CompactionTriggers.TurnsExceed(maxTurns) 包含的使用者回合數超過上限。
CompactionTriggers.GroupsExceed(maxGroups) 納入的組別數超過門檻。
CompactionTriggers.HasToolCalls() 至少存在一個非排除的工具呼叫群組。

結合觸發條件與邏輯運算:CompactionTriggers.All(...)(邏輯 AND)或 CompactionTriggers.Any(...)(邏輯 OR)

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

觸發器與標的

每個策略都有兩個謂詞:

  • 觸發器 — 控制 何時 開始壓實。 如果觸發器返回 false,則策略將被完全跳過。
  • 目標 — 控制 壓實停止的時間 。 策略逐步排除群組,每完成一步就重新評估目標,當目標回到true時就停止。

當未指定目標時,預設為觸發器的反向——當觸發條件不再觸發時,壓縮即停止。

壓縮是在一個平面物件清單 Message 上運作。 訊息會以輕量級群組元資料加以標註,策略會直接修改這些註解,將群組標記為排除,然後再將訊息清單投影至模型中。

訊息群組

訊息被分組成原子單位。 每個群組被分配一個 GroupKind

種類 說明
system 系統訊息。 在壓實過程中始終會被保存。
user 一則單一使用者訊息。
assistant_text 一個簡單的助理文字回應(沒有函數調用)。
tool_call 一個帶有函式呼叫的助理訊息,以及對應的工具結果訊息,視為一個原子單元。

壓縮策略

A CompactionStrategy 是一個協定——任何 async 可調用的協議,接受 a list[Message] 並在原地進行變異,當它改變任何東西時返回 True

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

Tokenizer

代幣感知策略接受一個 TokenizerProtocol 實作。 內建 CharacterEstimatorTokenizer 使用每個標記 4 字元的啟發式:

from agent_framework._compaction import CharacterEstimatorTokenizer

tokenizer = CharacterEstimatorTokenizer()

當你需要特定模型的精確標記數時,可以傳入自訂分詞器。

壓縮策略

所有策略都繼承自抽象 CompactionStrategy 基底類別。 每種策略都保留系統訊息,並尊重 MinimumPreserved 保護最新非系統群組不被移除的底線。

壓縮策略從agent_framework._compaction匯入。

截斷壓縮策略

截斷策略

最直接的方法:移除最早的非系統訊息群組,直到達成目標條件為止。

  • 尊重原子群邊界(工具呼叫與結果訊息同時移除)。
  • 最適合硬性代幣預算的後備方案。
  • MinimumPreserved 預設為 32
// Drop oldest groups when tokens exceed 32K, keeping at least 10 recent groups
TruncationCompactionStrategy truncation = new(
    trigger: CompactionTriggers.TokensExceed(0x8000),
    minimumPreserved: 10);
  • 當提供 tokenizer 時,度量為標記數,否則則為包含訊息數量。
  • 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(),
)

滑動視窗壓縮策略

滑動視窗策略

移除較舊的對話內容,只保留最近的對話視窗,尊重邏輯對話單元而非任意的訊息數量。 系統訊息會被完整保留。

  • 最適合用來設定對話長度的可預測範圍。

移除最早的使用者 回合 及其相關回應群組,採用邏輯回合邊界而非個別群組運作。

  • 輪流以使用者訊息開始,包含所有後續助理及工具呼叫群組,直到下一則使用者訊息。
  • MinimumPreserved 預設為 1 (至少保留最近的非系統群組)。
// Keep only the last 4 user turns
SlidingWindowCompactionStrategy slidingWindow = new(
    trigger: CompactionTriggers.TurnsExceed(4));

只保留最新的 keep_last_groups 非系統群組,排除所有較舊的群組。

  • preserve_system 預設為 True
from agent_framework._compaction import SlidingWindowStrategy

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

工具結果壓縮策略

將較舊的工具呼叫群組壓縮成緊湊的摘要訊息,保留易讀的追蹤資料,且不會產生完整訊息的多餘開銷。

  • 不會觸碰使用者訊息或一般助理回應。
  • 作為第一遍策略,用於從冗長的工具結果中回收空間。
  • 以簡短摘要 [Tool calls: get_weather, search_docs]取代多訊息工具呼叫群組(助理呼叫 + 工具結果)。
  • MinimumPreserved 預設為 2,確保當前回合的工具互動保持可見。
// Collapse old tool results when tokens exceed 512
ToolResultCompactionStrategy toolCompaction = new(
    trigger: CompactionTriggers.TokensExceed(0x200));
  • 壓縮成如 [Tool results: get_weather: sunny, 18°C]的簡要摘要訊息。
  • 最新的 keep_last_tool_call_groups 工具呼叫群組則保持不變。
from agent_framework._compaction import ToolResultCompactionStrategy

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

摘要壓縮策略

摘要策略

使用大型語言模型(LLM)來將對話的舊部分進行摘要,並以單一摘要訊息替換。

  • 預設提示會保留關鍵事實、決策、使用者偏好及工具呼叫結果。
  • 需要單獨的LLM客戶端來進行摘要工作,建議使用較小且較快速的模型。
  • 最適合保留對話上下文,同時大幅減少標記數量。
  • 你可以提供自訂的摘要提示。
  • 保護系統訊息及最新的 MinimumPreserved 非系統群組(預設: 4)。
  • 將舊訊息發送到一個獨立的 IChatClient,附上摘要提示,然後將摘要作為 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);

你可以提供自訂的摘要提示:

SummarizationCompactionStrategy summarization = new(
    chatClient: summarizerChatClient,
    trigger: CompactionTriggers.TokensExceed(0x500),
    summarizationPrompt: "Summarize the key decisions and user preferences only.");
  • 當非系統訊息的數量超過 target_count + threshold 時觸發。
  • 保留最新的 target_count 訊息,並總結所有較舊的訊息。
  • 需要客户端。SupportsChatGetResponse
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,
)

提供自訂摘要提示:

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

管線壓縮策略

將多種策略組合成連續的流程。 每種策略或策略方法都基於前一種的結果運行,從而實現從溫和到積極的分層壓縮。

  • 管線本身的觸發器是 CompactionTriggers.Always。每個子策略會獨立評估自己的觸發器。
  • 策略是依序執行的,所以先把最溫和的策略放在第一位。
PipelineCompactionStrategy pipeline = new(
    new ToolResultCompactionStrategy(CompactionTriggers.TokensExceed(0x200)),
    new SummarizationCompactionStrategy(summarizerChatClient, CompactionTriggers.TokensExceed(0x500)),
    new SlidingWindowCompactionStrategy(CompactionTriggers.TurnsExceed(4)),
    new TruncationCompactionStrategy(CompactionTriggers.TokensExceed(0x8000)));

此管線:

  1. 舊工具結果會塌陷(溫和)。
  2. 總結較舊的對話片段(適中)。
  3. 只保留最後4次使用者互動(非常積極)。
  4. 如果還超支,會把最老的團體剔除(緊急後備方案)。

SelectiveToolCallCompactionStrategy

完全排除較舊的工具呼叫群組,僅保留最後一個 keep_last_tool_call_groups

  • 不會接觸使用者或一般助理訊息。
  • 最佳狀態是工具對話主導代幣使用,且不需要完整工具歷史。
from agent_framework._compaction import SelectiveToolCallCompactionStrategy

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

組合策略代幣預算

將多種策略組合成由代幣預算驅動的序列流程。 每個子策略依序進行,預算滿足後提前結束。 策略若無法達成目標,則內建的後備機制將排除最早的群組。

  • 策略依序執行;先採取最溫和的策略。
  • early_stop=True (預設)在代幣預算被滿足後立即停止。
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),
    ],
)

此管線:

  1. 舊工具結果會塌陷(溫和)。
  2. 總結較舊的對話部分(中等)。
  3. 只保留最後20組(積極)。
  4. 如果預算仍然超支,則會回到最舊優先排除(緊急後備方案)。

使用代理進行壓縮

將壓縮策略包裹在 a CompactionProvider 中並註冊為 AIContextProvider。 將單一策略或PipelineCompactionStrategy傳遞給建構函式。

註冊建構者 API

請使用UseAIContextProvidersChatClientBuilder上註冊提供者。 提供者運行於工具呼叫迴圈內,並在每次 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));

小提示

使用較小且較便宜的模型(例如 gpt-4o-mini)來做摘要聊天客戶端,以降低成本同時維持摘要品質。

若只需一種策略,直接將其傳遞至 CompactionProvider ,無需用 PipelineCompactionStrategy 包裹。

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

通過 ChatClientAgentOptions 註冊

提供者也可以直接在ChatClientAgentOptions.AIContextProviders上指定。

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

備註

當透過ChatClientAgentOptions註冊時,CompactionProvider不會在工具呼叫迴圈中被接入。 代理層級的上下文提供者會在聊天歷史儲存之前執行,因此任何由 CompactionProvider 所產生的合成摘要訊息當使用 ChatHistoryProvider 時,可能成為持久保存的歷史的一部分。 若要僅壓縮飛行中請求的上下文並保留原始儲存的歷史記錄,請在 ChatClientBuilder 之上透過 UseAIContextProviders(...) 註冊提供者。

臨時壓縮

CompactionProvider.CompactAsync 對無主動代理會話的任意訊息清單套用策略:

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

CompactionProvider 是一個情境提供者,會在每次代理執行前後套用壓縮策略。 把它和歷史提供者一起加入代理的 context_providers 列表。

  • before_strategy — 在模型呼叫前執行,將已載入上下文的訊息壓縮。
  • after_strategy — 在模型呼叫之後執行,壓縮歷史提供者儲存的訊息,使下一回合開始時較小。
  • history_source_id — 儲存訊息的歷史提供者 after_strategy 應壓縮(預設為 "in_memory")。

與代理人註冊

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

小提示

使用較小且較便宜的模型(例如 gpt-4o-mini)作為摘要客戶端,以降低成本同時維持摘要品質。

如果只需要一個策略,直接傳遞為:before_strategy

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

每次運行後壓縮持久化的歷史記錄

after_strategy 來壓縮歷史提供者儲存的訊息,使未來回合以簡化上下文開始:

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

臨時性壓縮

apply_compaction 對主動代理會話外的任意訊息清單套用策略:

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

策略選擇

策略 侵略性 保留上下文 需要大型語言模型(LLM) 最適合用於
ToolResultCompactionStrategy 高 — 僅使工具結果崩潰 No 從冗長工具輸出中回收空間
SummarizationCompactionStrategy 中等 媒介 — 以摘要取代歷史 是的 在長時間的對話中,情境尤為重要。
SlidingWindowCompactionStrategy 低 — 跳過整回合 No 硬性回合數限制
TruncationCompactionStrategy 低 — 刪除最古老的群組 No 緊急代幣預算後備方案
PipelineCompactionStrategy Configurable 這取決於小孩的策略 視情況而定 具有多重回退的層疊壓縮
策略 侵略性 保留上下文 需要大型語言模型(LLM) 最適合用於
ToolResultCompactionStrategy 高 — 摺疊工具會產生摘要訊息 No 從冗長工具輸出中回收空間
SelectiveToolCallCompactionStrategy 低–中 Medium — 完全排除舊有的工具呼叫群組 No 當不再需要結果時,移除工具歷史
SummarizationStrategy 中等 媒介 — 以摘要取代歷史 是的 在長時間的對話中,情境是非常重要的。
SlidingWindowStrategy 低 — 刪除最古老的群組 No 硬性群組計數限制
TruncationStrategy 低 — 刪除最古老的群組 No 緊急訊息或代幣預算後備機制
TokenBudgetComposedStrategy Configurable 這取決於小孩的策略 視情況而定 層疊壓縮,目標為代幣預算,並有多種備援

下一步