通过

压缩

随着聊天的增长，聊天历史记录的令牌计数可能会超过模型上下文窗口或增加成本。 压缩策略可减少聊天历史记录的大小，同时保留重要上下文，因此代理可以在长时间运行的交互中继续运行。

重要

压缩框架目前是实验性的。 若要使用它，需要添加 #pragma warning disable MAAI001。

重要

压缩框架目前在 Python 中是实验性的。 从 agent_framework._compaction中导入策略。

为什么压缩很重要

每次调用 LLM 都包括完整的对话历史记录。 不压缩：

  • 令牌限制 - 对话最终超过模型的上下文窗口，导致错误。
  • 成本 - 较大的提示消耗更多令牌，增加 API 成本。
  • 延迟 - 更多的输入令牌意味着响应时间变慢。

压缩通过选择性地删除、折叠或汇总对话的较旧部分来解决这些问题。

核心概念

适用性：仅限内存型历史记录代理程序

压缩仅适用于在内存中管理其自己的聊天历史记录的代理。 依赖于服务托管上下文或会话状态的代理不会从精简中受益，因为服务已经处理了上下文管理。 服务托管代理的示例包括：

  • Foundry 代理 - 上下文由 Microsoft Foundry 服务并行管理。
  • 启用了存储的响应 API （默认值） - 聊天状态由 OpenAI 服务存储和管理。
  • Copilot Studio 代理 - 聊天上下文由 Copilot Studio 服务维护。

对于这些代理类型，配置压缩策略不起作用。 仅当代理维护自己的内存中消息列表，并在每次调用时将完整历史记录传递给模型时，压缩才相关。

压缩对 MessageIndex 平面消息列表的结构化视图进行作，该视图将消息分组为称为 MessageGroup 实例的原子单元。 每个组跟踪其消息计数、字节计数和估计令牌计数。

消息组

一个 MessageGroup 表示必须一起保留或删除的逻辑相关消息。 例如，包含工具调用的助手消息及其相应的工具结果消息形成一个原子组，删除其中一个而不删除另一个就会导致 LLM API 错误。

每个组都有一个 MessageGroupKind：

种类 说明
System 一个或多个系统消息。 在压缩期间始终保留。
User 启动新轮次的单个用户消息。
AssistantText 纯助理文本响应（无工具调用）。
ToolCall 带有工具调用和相应工具结果消息的助理消息，被视为原子单元。
Summary 汇总压缩生成的简化消息。

触发器

一个 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 具有函数调用的助手消息加上相应的工具结果消息，被视为原子单元。

压缩策略

CompactionStrategy 是一种协议 —— 任何可调用的对象接受一个 async，并在原地修改它，如果改变了任何内容，则返回 list[Message]。

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

分词器

令牌感知策略接受 TokenizerProtocol 实现。 内置 CharacterEstimatorTokenizer 使用每令牌 4 个字符的启发式：

from agent_framework._compaction import CharacterEstimatorTokenizer

tokenizer = CharacterEstimatorTokenizer()

当您需要特定模型编码的准确令牌计数时，请传递自定义分词器。

压缩策略

所有策略都继承自抽象 CompactionStrategy 基类。 每个策略都会保留系统消息，并尊重一个 MinimumPreserved 可保护最新非系统组免受删除的底层。

压缩策略是从 agent_framework._compaction中导入的。

TruncationCompactionStrategy

截断策略

最直接的方法：在满足目标条件之前删除最早的非系统消息组。

  • 尊重原子组边界（工具调用和结果消息一起删除）。
  • 最适合硬令牌预算的安全保护。
  • 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(),
)

SlidingWindowCompactionStrategy

SlidingWindowStrategy（滑动窗口策略）

删除较旧的聊天内容以仅保留最近的交换窗口，尊重逻辑对话单元而不是任意消息计数。 系统消息将在整个过程中保留。

  • 最适合在可预测的范围内限制对话长度。

删除最早的用户 轮次 及其关联的响应组，在逻辑轮次边界而不是单个组上运行。

  • 轮次从用户消息开始，并包括所有后续助理和工具呼叫组，直到下一个用户消息。
  • 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.",
)

PipelineCompactionStrategy

将多个策略组合到一个顺序管道中。 每个策略都基于上一个策略的结果进行操作，实现从柔和到强势的分层压缩。

  • 管道的触发器为 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. 如果预算仍然超出限制（紧急后备措施），将移除最早的组。

选择性工具调用压缩策略

完全排除较旧的工具调用组，只保留最后 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)

TokenBudgetComposedStrategy

将多个策略组合到由令牌预算驱动的顺序流程中。 每个子策略按顺序执行，一旦预算得到满足，就会提前终止。 如果仅策略无法达到目标，则内置的回退机制将排除最旧的一组。

  • 策略按顺序执行;首先放置最温和的策略。
  • 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. 如果仍然超过预算（紧急后备），则回退到最早的排除。

将压缩与代理程序配合使用

将压缩策略包装在一个 CompactionProvider 中，并将其注册为一个 AIContextProvider。 将单个策略或 PipelineCompactionStrategy 传递给构造函数。

注册在生成器 API 上

在 ChatClientBuilder上使用 UseAIContextProviders注册提供程序。 提供者在工具调用循环内运行，在每次调用 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 — 其存储消息source_id应进行压缩的历史记录提供程序（默认为after_strategy）。

向代理注册

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 低 高 - 仅折叠工具结果 否 从冗长的工具输出中回收空间
SummarizationCompactionStrategy 中等 中 - 将历史记录替换为摘要 是的 上下文很重要的长对话
SlidingWindowCompactionStrategy 高 低 - 下降整个轮次 否 硬轮次计数限制
TruncationCompactionStrategy 高 低 - 删除最早的组 否 紧急令牌预算后盾
PipelineCompactionStrategy 可配置 取决于子策略 视情况而定 具有多种备选机制的分层压缩
策略 攻击性 保留上下文 需要 LLM 最适用于
ToolResultCompactionStrategy 低 高级 - 将工具结果折叠为摘要信息 否 从冗长的工具输出中回收空间
SelectiveToolCallCompactionStrategy 低-中 中等—完全排除旧的工具调用组 否 不再需要结果时删除工具历史记录
SummarizationStrategy 中等 中 - 将历史记录替换为摘要 是的 上下文很重要的长对话
SlidingWindowStrategy 高 低 - 删除最早的组 否 固定组数量限制
TruncationStrategy 高 低 - 删除最早的组 否 紧急消息与令牌预算后援措施
TokenBudgetComposedStrategy 可配置 取决于子策略 视情况而定 分层压实与令牌分配目标及多个回退机制

后续步骤

Middleware

  • Last updated on