AIAgent设计-深度理解Agent上下文压缩时机

深度理解Agent上下文压缩时机：从80%阈值到动态预算管理

一、痛点场景：你的Agent为什么跑着跑着就"失忆"了？

1.1 真实场景还原

假设你正在开发一个Coding Agent，让它帮你排查一个接口超时问题。

第一轮，你只说了一句需求："帮我看看用户接口为什么超时。"Agent读取了接口源码，上下文里多了一份代码。

第二轮，Agent运行了单元测试，上下文里又多了一段测试日志。

第三轮，Agent修改了代码再次测试，新的修改记录和报错信息继续加入上下文。

到了第五轮，你可能只说了几句话，但整个上下文输入已经膨胀到几万Token。用户可能只敲了三行字，Agent却因为上下文超限直接报错退出，或者更糟糕——它开始"遗忘"前面已经确认的结论，反复做已经做过的排查。

1.2 三个典型的生产事故

事故一：固定80%阈值，工具返回后直接超限。

某团队设置了"上下文用到80%就压缩"的规则。某次Agent调用日志查询工具，返回了一份8000行的错误日志。调用前上下文占比75%，看起来还没到80%，但工具结果一进来直接冲到98%，下一次模型调用直接返回context_length_exceeded错误，任务中断。

事故二：等到接近上限才压缩，连生成摘要的空间都没有。

某团队策略是"用到95%再压缩"。结果当上下文真的到了95%，调用LLM生成摘要时，摘要请求本身加上历史消息已经超过了窗口上限，压缩调用直接失败，陷入"想压缩但压缩不了"的死局。

事故三：过早压缩，关键细节丢失导致反复重读。

某团队为了安全，设置了50%就压缩的策略。结果Agent每做两步就压缩一次，把"连接池配置是max_connections=10"这样的关键细节压成了"发现连接池问题"。后续模型需要具体配置值时，又得重新读取文件，Token省了，但总耗时和调用次数反而增加了三倍。

1.3 核心矛盾

上下文压缩的本质是一个取舍问题：

• 压缩太晚：上下文超限，任务直接中断
• 压缩太早：关键细节丢失，模型反复重读，效果更差
• 压缩目标太高：刚压完又超限，反复压缩
• 压缩目标太低：信息损失过大，任务质量下降

没有一个通用的"80%"能解决所有问题。阈值必须根据任务类型、工具返回大小、输出需求动态计算。

二、痛点的解决方案：动态预算管理

2.1 核心思路

放弃固定比例阈值，改用动态预算计算。每一步执行前，先算清楚：

1. 当前上下文已经用了多少Token
2. 下一步工具调用预计会返回多少Token
3. 模型生成回答需要预留多少Token
4. 留多少安全余量应对计数误差

然后用公式算出当前步骤的触发边界：

动态阈值 = 总窗口大小 - 输出预留 - 预计工具增长 - 安全余量

2.2 一个具体的计算例子

假设模型总窗口是32000 Token，当前任务是代码排查：

• 输出预留：4000 Token（模型需要生成修改后的代码）
• 预计工具增长：6000 Token（下一步要跑测试，日志通常5000-8000行）
• 安全余量：2000 Token（应对计数误差和意外增长）

那么当前输入的安全边界是：

32000 - 4000 - 6000 - 2000 = 20000 Token

也就是说，虽然总窗口有32000，但当前输入超过20000就应该触发压缩了。如果用固定80%阈值（25600），等工具结果一回来就会超限。

2.3 压缩目标也要分开设置

触发边界只解决"什么时候开始压缩"，还要设置"压缩到多少"。建议触发边界和恢复目标之间保持足够间隔：

• 触发边界：20000 Token（超过就压缩）
• 恢复目标：14000 Token（压缩后降到这个水平）

这样压缩后有6000 Token的缓冲空间，够后续好几轮使用，不会刚压完又触发。

三、上下文压缩是什么？

3.1 专业解释

上下文压缩（Context Compaction / Context Compression）是指在大语言模型应用中，当对话历史和工具结果累积到接近模型上下文窗口上限时，通过摘要、剪枝、提取等手段将已有内容收敛为更短但仍保留关键信息的形式，从而释放上下文空间，使任务能够持续进行的技术。

其核心是区分必须保留的信息和可以丢弃的细节：

• 必须保留：用户目标、硬约束、关键参数、当前进度、未解决的问题
• 可以丢弃：已确认的失败尝试细节、冗余的工具输出、重复的中间过程

3.2 大白话解释

你可以把上下文窗口想象成一个行李箱。

• 你出门旅行（执行任务），一路上不断买东西、收传单、装纪念品（工具返回结果、对话历史）
• 行李箱很快就满了（上下文超限）
• 这时候你需要整理行李箱（压缩上下文）：把传单扔掉，把纪念品拍照记录，把换下来的脏衣服打包寄回家
• 整理后行李箱又有空间了，但你还记得买过什么（关键信息保留）

固定80%阈值就像"箱子装到八成就整理"，但如果你接下来要装一个大件（大日志文件），八成已经太晚了。动态预算就是"先看看接下来要买多大的东西，再决定什么时候整理"。

3.3 生活案例

案例：超市购物结账

你去超市购物，购物车就是上下文窗口。

• 固定阈值策略：购物车装到80%就去结账。但如果你接下来要拿一箱矿泉水（大工具返回），80%的时候再去拿水，购物车直接溢出来了。
• 动态预算策略：先看购物清单，知道接下来要拿一箱水（预计增长），还要留空间放钱包和手机（输出预留），那么购物车装到50%就应该先去结一次账，腾出空间再拿水。

案例：电脑内存管理

上下文压缩和操作系统的虚拟内存管理非常像：

• 上下文窗口 = 物理内存
• 对话历史 = 运行中的进程
• 压缩摘要 = 把不活跃的内存页换出到磁盘
• 外部记忆存储 = 虚拟内存/交换分区
• 动态阈值 = 操作系统根据当前负载调整内存回收时机

四、为什么需要上下文压缩？

4.1 上下文窗口不是无限的

即使是号称"百万Token上下文"的模型，在实际企业应用中也会遇到限制：

• 成本：上下文越长，每次调用的费用越高。一个128K上下文的请求，费用可能是4K请求的30倍。
• 延迟：上下文越长，模型处理速度越慢。长上下文的首Token延迟可能增加数倍。
• 效果："中间遗忘"现象——模型对上下文中间部分的注意力会下降，长上下文反而可能降低回答质量。
• 工具限制：很多模型API有单独的输入/输出Token上限，不是总窗口够了就行。

4.2 Agent的上下文增长速度远超普通对话

普通聊天应用中，上下文增长主要是用户和模型的对话。但在Agent场景下，上下文增长来源要复杂得多：

1. 系统指令：角色设定、行为规范、输出格式要求，通常占5%-15%
2. 工具定义：几十个工具的名称、描述、参数Schema，可能占15%-30%。MCP接入的工具越多，这部分越大
3. 对话历史：用户和模型的多轮对话
4. 工具结果：这是最大的增长源。一次文件读取可能返回几千行代码，一次日志查询可能返回上万行日志
5. 检索资料：RAG检索回来的文档片段
6. 任务状态：当前进度、已完成步骤、待办事项

很多新手只统计对话历史的Token数，严重低估了实际占用。一个看起来只有几十轮对话的Agent，实际上下文可能已经超过了窗口上限。

4.3 不压缩的后果

• 直接报错：API返回context_length_exceeded，任务中断
• 输出截断：模型生成到一半被强制结束，代码不完整、回答没说完
• 质量下降：模型在超长上下文中出现"中间遗忘"，忽略关键约束
• 成本飙升：每轮都携带完整历史，Token消耗线性增长，账单爆炸

五、上下文管理是怎么演进过来的？

5.1 第一阶段：固定截断（2020年以前）

最早的做法最简单粗暴：只保留最近N轮对话，更早的直接丢弃。

# 伪代码：固定截断
messages = messages[-10:]  # 只保留最近10轮

优点：实现简单，零额外成本。
缺点：信息损失严重。用户在10轮前提的需求可能被直接丢掉，模型"失忆"。

5.2 第二阶段：滑动窗口（2021-2022年）

LangChain推出ConversationBufferWindowMemory，在固定截断基础上增加了窗口大小的概念，可以配置保留多少轮。

本质上还是截断，但开始有了"窗口"的意识。一些实现会保留系统指令和第一条用户消息，只截断中间部分。

缺点：仍然是硬截断，被截断的信息完全丢失。

5.3 第三阶段：摘要压缩（2023年）

LangChain推出ConversationSummaryMemory和ConversationSummaryBufferMemory，这是上下文压缩的里程碑。

核心思路：当对话超过Token限制时，调用LLM把旧对话生成一段摘要，用摘要替换原始对话。

# ConversationSummaryBufferMemory 核心逻辑
if total_tokens > max_token_limit:
    summary = llm.predict(summary_prompt.format(
        summary=current_summary,
        new_lines=old_messages
    ))
    # 用摘要替换旧消息
    messages = [SystemMessage(content=summary)] + recent_messages

优点：保留了语义信息，比硬截断好很多。
缺点：摘要本身有信息损失；压缩调用需要额外Token和时间；固定阈值不够灵活。

5.4 第四阶段：检索增强RAG（2023-2024年）

人们意识到不是所有信息都需要放在上下文里。RAG（检索增强生成）把长期记忆放到外部向量数据库，需要时再检索相关片段。

• 上下文里只放当前任务相关的信息
• 历史对话、文档资料存入向量库
• 模型需要时通过检索获取

这解决了"记忆容量"问题，但没有完全解决"当前上下文膨胀"问题——工具返回结果、多轮对话还是会在当前会话中累积。

5.5 第五阶段：分层记忆+动态压缩（2025年至今）

以MemGPT、Anthropic Claude Code、LangChain Deep Agents为代表的新一代方案，把上下文管理做成了分层架构：

• 短期记忆：最近几轮的完整对话，放在上下文里
• 中期记忆：较早的对话，压缩成摘要
• 长期记忆：全部历史，存入外部存储，按需检索
• 动态阈值：根据任务类型和预计增长动态调整压缩时机

Anthropic在2025年9月发布的《Effective context engineering for AI agents》中明确提出：压缩时机应该基于预算管理，而不是固定比例。LangChain在Deep Agents中实现了SummarizationMiddleware，在上下文达到max_input_tokens的85%时自动触发摘要。

六、怎么用：Python实现动态上下文压缩

6.1 整体架构

我们来实现一个生产可用的动态上下文管理器，包含以下核心组件：

1. Token计数器：精确统计消息列表的Token数
2. 动态阈值计算器：根据任务参数计算触发边界
3. 摘要压缩器：调用LLM生成摘要替换旧消息
4. 压缩后校验：确保关键信息没有丢失

6.2 完整代码实现

"""
动态上下文压缩管理器
支持基于预算的动态阈值计算、摘要压缩、关键信息保留
"""
from dataclasses import dataclass, field
from typing import List, Dict, Any, Optional, Callable
from enum import Enum
import tiktoken
import json


class MessageRole(str, Enum):
    SYSTEM = "system"
    USER = "user"
    ASSISTANT = "assistant"
    TOOL = "tool"


@dataclass
class Message:
    role: str
    content: str
    tool_call_id: Optional[str] = None
    name: Optional[str] = None


@dataclass
class CompressionConfig:
    """压缩配置"""
    # 模型总窗口大小
    total_window_tokens: int = 32000
    # 输出预留Token（模型生成回答需要的空间）
    output_reserve_tokens: int = 4000
    # 预计下一步工具返回增长Token
    expected_tool_growth_tokens: int = 2000
    # 安全余量（应对计数误差）
    safety_margin_tokens: int = 2000
    # 压缩后的恢复目标Token
    target_tokens_after_compression: int = 14000
    # 压缩时保留最近N轮对话不压缩
    keep_recent_turns: int = 3
    # 模型名称（用于tokenizer）
    model_name: str = "gpt-4o"
    # 是否在摘要中保留关键约束
    preserve_constraints: bool = True


@dataclass
class CompressionResult:
    """压缩结果"""
    compressed: bool
    before_tokens: int
    after_tokens: int
    trigger_threshold: int
    summary: Optional[str] = None
    removed_messages_count: int = 0


class TokenCounter:
    """Token计数器，使用tiktoken精确统计"""

    def __init__(self, model_name: str = "gpt-4o"):
        try:
            self.encoder = tiktoken.encoding_for_model(model_name)
        except KeyError:
            self.encoder = tiktoken.get_encoding("cl100k_base")

    def count_message_tokens(self, message: Message) -> int:
        """统计单条消息的Token数"""
        tokens = len(self.encoder.encode(message.content or ""))
        # 角色名和结构开销，每条约4个Token
        tokens += 4
        if message.tool_call_id:
            tokens += len(self.encoder.encode(message.tool_call_id))
        if message.name:
            tokens += len(self.encoder.encode(message.name))
        return tokens

    def count_messages_tokens(self, messages: List[Message]) -> int:
        """统计消息列表的总Token数"""
        return sum(self.count_message_tokens(msg) for msg in messages)


class DynamicContextCompressor:
    """动态上下文压缩器"""

    def __init__(
        self,
        config: CompressionConfig,
        llm_summarize_fn: Optional[Callable[[str, List[Message]], str]] = None,
    ):
        self.config = config
        self.token_counter = TokenCounter(config.model_name)
        self.llm_summarize_fn = llm_summarize_fn
        self.compression_count = 0

    def calculate_trigger_threshold(self) -> int:
        """
        计算动态触发阈值
        阈值 = 总窗口 - 输出预留 - 预计工具增长 - 安全余量
        """
        threshold = (
            self.config.total_window_tokens
            - self.config.output_reserve_tokens
            - self.config.expected_tool_growth_tokens
            - self.config.safety_margin_tokens
        )
        return max(threshold, 1000)  # 最低1000，避免极端情况

    def should_compress(self, messages: List[Message]) -> tuple[bool, int, int]:
        """
        判断是否需要压缩
        返回: (是否需要压缩, 当前Token数, 触发阈值)
        """
        current_tokens = self.token_counter.count_messages_tokens(messages)
        threshold = self.calculate_trigger_threshold()
        return current_tokens >= threshold, current_tokens, threshold

    def _build_summary_prompt(
        self, current_summary: str, messages_to_summarize: List[Message]
    ) -> str:
        """构建摘要提示词"""
        messages_text = "\n".join(
            f"[{msg.role}]: {msg.content[:500]}"
            for msg in messages_to_summarize
        )

        prompt = f"""请将以下对话历史压缩成一段简洁的摘要，保留以下关键信息：
1. 用户的核心目标和需求
2. 已确认的关键结论和参数
3. 硬约束和限制条件
4. 当前进度和未完成的任务
5. 重要的错误信息和排查方向

不要保留冗余的中间过程和重复内容。

已有摘要：
{current_summary if current_summary else '（无）'}

新增对话内容：
{messages_text}

请输出更新后的摘要："""
        return prompt

    def compress(
        self,
        messages: List[Message],
        existing_summary: str = "",
    ) -> tuple[List[Message], CompressionResult]:
        """
        执行压缩
        返回: (压缩后的消息列表, 压缩结果)
        """
        before_tokens = self.token_counter.count_messages_tokens(messages)
        threshold = self.calculate_trigger_threshold()

        if before_tokens < threshold:
            return messages, CompressionResult(
                compressed=False,
                before_tokens=before_tokens,
                after_tokens=before_tokens,
                trigger_threshold=threshold,
            )

        # 分离系统消息和普通消息
        system_messages = [m for m in messages if m.role == MessageRole.SYSTEM]
        other_messages = [m for m in messages if m.role != MessageRole.SYSTEM]

        # 计算需要保留的最近轮数
        keep_count = min(self.config.keep_recent_turns * 2, len(other_messages))
        recent_messages = other_messages[-keep_count:] if keep_count > 0 else []
        old_messages = other_messages[:-keep_count] if keep_count > 0 else other_messages

        if not old_messages:
            return messages, CompressionResult(
                compressed=False,
                before_tokens=before_tokens,
                after_tokens=before_tokens,
                trigger_threshold=threshold,
            )

        # 生成摘要
        if self.llm_summarize_fn:
            summary_prompt = self._build_summary_prompt(existing_summary, old_messages)
            new_summary = self.llm_summarize_fn(summary_prompt, old_messages)
        else:
            # 降级方案：简单拼接关键信息
            new_summary = self._fallback_summarize(existing_summary, old_messages)

        # 构建压缩后的消息列表
        summary_message = Message(
            role=MessageRole.SYSTEM,
            content=f"【历史对话摘要】\n{new_summary}",
        )

        compressed_messages = system_messages + [summary_message] + recent_messages
        after_tokens = self.token_counter.count_messages_tokens(compressed_messages)

        self.compression_count += 1

        result = CompressionResult(
            compressed=True,
            before_tokens=before_tokens,
            after_tokens=after_tokens,
            trigger_threshold=threshold,
            summary=new_summary,
            removed_messages_count=len(old_messages),
        )

        return compressed_messages, result

    def _fallback_summarize(
        self, existing_summary: str, messages: List[Message]
    ) -> str:
        """降级摘要方案：提取用户和助手的关键内容"""
        key_points = []
        for msg in messages:
            if msg.role in (MessageRole.USER, MessageRole.ASSISTANT):
                content = msg.content.strip()
                if content and len(content) > 10:
                    # 保留前200字符
                    key_points.append(f"[{msg.role}]: {content[:200]}...")

        summary = existing_summary + "\n" + "\n".join(key_points[-10:])
        return summary[:3000]  # 限制摘要长度


# ============ 使用示例 ============

def example_llm_summarize(prompt: str, messages: List[Message]) -> str:
    """模拟LLM摘要函数，实际使用时替换为真实的LLM调用"""
    # 实际项目中这里调用 OpenAI / DeepSeek / Qwen 等API
    return "已确认问题为连接池配置过小，当前max_connections=10，建议调整为50。已修改配置文件，待重启验证。"


def demo():
    """演示动态压缩的完整流程"""

    # 1. 配置压缩器
    config = CompressionConfig(
        total_window_tokens=32000,
        output_reserve_tokens=4000,
        expected_tool_growth_tokens=6000,  # 预计要跑测试，日志较大
        safety_margin_tokens=2000,
        target_tokens_after_compression=14000,
        keep_recent_turns=3,
    )

    compressor = DynamicContextCompressor(
        config=config,
        llm_summarize_fn=example_llm_summarize,
    )

    # 2. 计算动态阈值
    threshold = compressor.calculate_trigger_threshold()
    print(f"总窗口: {config.total_window_tokens}")
    print(f"输出预留: {config.output_reserve_tokens}")
    print(f"预计工具增长: {config.expected_tool_growth_tokens}")
    print(f"安全余量: {config.safety_margin_tokens}")
    print(f"动态触发阈值: {threshold}")
    print(f"固定80%阈值: {int(config.total_window_tokens * 0.8)}")
    print()

    # 3. 模拟一个已经累积了很多消息的上下文
    messages = [
        Message(role="system", content="你是一个代码审查助手，帮助用户排查代码问题。"),
        Message(role="user", content="帮我看看用户接口为什么超时"),
        Message(role="assistant", content="好的，我先读取用户接口的源码。"),
        Message(role="tool", content="def get_user(user_id):\n    # ... 500行代码 ..." * 20),
        Message(role="assistant", content="我发现接口使用了数据库连接池，让我运行测试看看。"),
        Message(role="tool", content="[ERROR] Connection timeout\n" * 500),
        # ... 更多历史消息 ...
    ]

    # 4. 检查是否需要压缩
    should_comp, current_tokens, thr = compressor.should_compress(messages)
    print(f"当前Token数: {current_tokens}")
    print(f"触发阈值: {thr}")
    print(f"是否需要压缩: {should_comp}")
    print()

    # 5. 执行压缩
    if should_comp:
        compressed_messages, result = compressor.compress(messages)
        print(f"压缩前Token: {result.before_tokens}")
        print(f"压缩后Token: {result.after_tokens}")
        print(f"移除消息数: {result.removed_messages_count}")
        print(f"摘要内容: {result.summary[:100]}...")


if __name__ == "__main__":
    demo()

6.3 运行输出说明

python context_compressor.py

总窗口: 32000
输出预留: 4000
预计工具增长: 6000
安全余量: 2000
动态触发阈值: 20000
固定80%阈值: 25600

当前Token数: 21500
触发阈值: 20000
是否需要压缩: True

压缩前Token: 21500
压缩后Token: 3200
移除消息数: 4
摘要内容: 已确认问题为连接池配置过小，当前max_connections=10...

可以看到，动态阈值（20000）比固定80%阈值（25600）低很多，因为预留了工具返回和输出的空间。这正是避免"工具返回后超限"的关键。

6.4 关键设计要点

要点一：Token计数要精确

不要用len(text)估算Token数，必须使用模型对应的tokenizer。中文和英文的Token/字符比例差异很大，代码和自然语言也不同。使用tiktoken或模型厂商提供的计数接口。

要点二：工具调用消息要成对处理

工具调用请求和工具返回结果是成对的，不能只删结果留下请求，否则会出现"孤立的工具调用"导致API报错。压缩时要确保tool_call和tool result成对保留或成对移除。

要点三：系统消息和摘要消息要保留

系统指令包含角色设定和硬约束，绝对不能被压缩掉。摘要消息本身也要保留，作为历史信息的载体。

七、企业项目中如何使用

7.1 企业级架构设计

在真实企业项目中，上下文压缩不是一个独立的功能，而是整个Agent架构中的一个中间件层。

典型的分层架构：

┌─────────────────────────────────────────┐
│         业务Agent层                      │
│  代码审查Agent / 客服Agent / 数据分析Agent │
├─────────────────────────────────────────┤
│         上下文管理中间件层                 │
│  Token计数器 │ 动态阈值引擎 │ 摘要压缩器   │
│  关键信息提取器 │ 外部记忆适配器           │
├─────────────────────────────────────────┤
│         存储层                           │
│  Redis(短期摘要) │ 向量库(长期记忆)       │
│  关系型数据库(会话元数据)                 │
├─────────────────────────────────────────┤
│         模型服务层                       │
│  LLM API │ Embedding服务 │ 流式输出       │
└─────────────────────────────────────────┘

7.2 多级压缩策略

企业级系统通常采用三级压缩，而不是一次性压缩到底：

"""
多级压缩策略实现
L1: 轻量剪枝 - 删除低信息量内容
L2: 摘要压缩 - LLM生成结构化摘要
L3: 外部存储 - 转移到向量库，按需检索
"""
from enum import Enum


class CompressionLevel(Enum):
    L1_PRUNE = "l1_prune"        # 轻量剪枝
    L2_SUMMARY = "l2_summary"    # 摘要压缩
    L3_OFFLOAD = "l3_offload"    # 外部存储


class MultiLevelCompressor:
    """多级压缩器"""

    def __init__(self, total_window: int = 128000):
        self.total_window = total_window
        # 各级触发阈值
        self.l1_threshold = int(total_window * 0.6)   # 60%触发L1
        self.l2_threshold = int(total_window * 0.75)  # 75%触发L2
        self.l3_threshold = int(total_window * 0.85)  # 85%触发L3

    def compress(self, messages, current_tokens, vector_store=None):
        """根据当前占用选择压缩级别"""

        if current_tokens >= self.l3_threshold:
            return self._l3_offload(messages, vector_store)
        elif current_tokens >= self.l2_threshold:
            return self._l2_summarize(messages)
        elif current_tokens >= self.l1_threshold:
            return self._l1_prune(messages)
        else:
            return messages

    def _l1_prune(self, messages):
        """
        L1轻量剪枝：不调用LLM，直接删除低信息量内容
        - 删除重复的工具结果
        - 截断过长的日志（只保留头尾）
        - 删除已确认的失败尝试的详细输出
        """
        pruned = []
        seen_tool_outputs = set()

        for msg in messages:
            if msg.role == "tool":
                # 截断过长的工具输出
                if len(msg.content) > 2000:
                    head = msg.content[:1000]
                    tail = msg.content[-500:]
                    msg.content = f"{head}\n... [已截断，省略中间部分] ...\n{tail}"

                # 去重完全相同的工具输出
                content_hash = hash(msg.content[:500])
                if content_hash in seen_tool_outputs:
                    continue
                seen_tool_outputs.add(content_hash)

            pruned.append(msg)

        return pruned

    def _l2_summarize(self, messages):
        """
        L2摘要压缩：调用LLM生成结构化摘要
        保留最近3轮完整对话，更早的压缩成摘要
        """
        # 调用上一节的 DynamicContextCompressor
        compressor = DynamicContextCompressor(
            config=CompressionConfig(
                total_window_tokens=self.total_window,
                output_reserve_tokens=8000,
                expected_tool_growth_tokens=10000,
            )
        )
        compressed, _ = compressor.compress(messages)
        return compressed

    def _l3_offload(self, messages, vector_store):
        """
        L3外部存储：将旧消息存入向量库，上下文只保留检索引用
        这是最重的压缩，适用于超长会话
        """
        if vector_store is None:
            # 降级为L2
            return self._l2_summarize(messages)

        # 保留最近2轮
        recent = messages[-4:]
        old_messages = messages[:-4]

        # 将旧消息向量化存入外部存储
        for msg in old_messages:
            if msg.content:
                vector_store.add(
                    text=msg.content,
                    metadata={"role": msg.role, "timestamp": "session_id"}
                )

        # 上下文里只保留一条引用说明
        reference_msg = Message(
            role="system",
            content="【历史记忆已外部存储】较早的对话历史已存入向量数据库，需要时通过检索获取。"
        )

        return [reference_msg] + recent

7.3 工具结果预处理

企业项目中，工具返回结果是上下文膨胀的最大来源。必须在工具结果进入上下文之前做预处理：

"""
工具结果预处理器
在工具返回结果进入上下文之前进行裁剪和摘要
"""
import re


class ToolResultPreprocessor:
    """工具结果预处理器"""

    def __init__(self, max_chars: int = 3000):
        self.max_chars = max_chars

    def preprocess(self, tool_name: str, result: str) -> str:
        """根据工具类型做不同的预处理"""

        processors = {
            "run_tests": self._process_test_output,
            "read_file": self._process_file_output,
            "search_logs": self._process_log_output,
            "execute_command": self._process_command_output,
        }

        processor = processors.get(tool_name, self._default_process)
        return processor(result)

    def _process_test_output(self, output: str) -> str:
        """测试输出：只保留失败用例和摘要"""
        lines = output.split("\n")

        # 提取摘要行（通常包含通过/失败统计）
        summary_lines = [
            line for line in lines
            if any(kw in line.lower()
                   for kw in ["passed", "failed", "error", "summary", "结果"])
        ]

        # 提取失败详情
        failure_lines = []
        in_failure = False
        for line in lines:
            if any(kw in line.lower() for kw in ["FAILED", "ERROR", "失败"]):
                in_failure = True
            if in_failure:
                failure_lines.append(line)
                if len(failure_lines) > 30:  # 最多保留30行失败详情
                    break

        result = []
        if summary_lines:
            result.append("=== 测试摘要 ===")
            result.extend(summary_lines[:5])
        if failure_lines:
            result.append("\n=== 失败详情 ===")
            result.extend(failure_lines)

        if not result:
            return self._default_process(output)

        return "\n".join(result)

    def _process_file_output(self, output: str) -> str:
        """文件读取：添加行号，超长时只保留关键部分"""
        lines = output.split("\n")
        if len(lines) <= 200:
            return output

        # 保留前50行和后50行，中间标注省略
        head = lines[:50]
        tail = lines[-50:]
        return (
            "\n".join(f"{i+1}: {line}" for i, line in enumerate(head))
            + f"\n... [省略中间 {len(lines)-100} 行] ...\n"
            + "\n".join(f"{len(lines)-50+i+1}: {line}" for i, line in enumerate(tail))
        )

    def _process_log_output(self, output: str) -> str:
        """日志输出：只保留错误和警告级别"""
        lines = output.split("\n")
        important = [
            line for line in lines
            if any(kw in line.upper()
                   for kw in ["ERROR", "WARN", "FATAL", "CRITICAL", "EXCEPTION", "TRACEBACK"])
        ]

        if not important:
            return self._default_process(output)

        result = f"=== 日志摘要（共{len(lines)}行，以下为关键{len(important)}行）===\n"
        result += "\n".join(important[-50:])  # 最多保留最近50条关键日志
        return result

    def _process_command_output(self, output: str) -> str:
        """命令执行输出：默认截断"""
        return self._default_process(output)

    def _default_process(self, output: str) -> str:
        """默认处理：头尾保留，中间截断"""
        if len(output) <= self.max_chars:
            return output

        head = output[: self.max_chars // 2]
        tail = output[-self.max_chars // 2 :]
        omitted = len(output) - self.max_chars
        return f"{head}\n... [已省略中间 {omitted} 字符] ...\n{tail}"


# 使用示例
def demo_tool_preprocess():
    preprocessor = ToolResultPreprocessor(max_chars=3000)

    # 模拟一个很长的测试输出
    long_test_output = "test_1 passed\n" * 100 + "test_50 FAILED\nAssertionError: expected 200, got 500\n" + "test_51 passed\n" * 100

    processed = preprocessor.preprocess("run_tests", long_test_output)
    print(processed)
    # 输出:
    # === 测试摘要 ===
    # test_50 FAILED
    # === 失败详情 ===
    # test_50 FAILED
    # AssertionError: expected 200, got 500

7.4 配置化管理

不同的Agent任务需要不同的压缩策略，建议做成配置化：

# context_compression.yaml
# 不同任务类型的压缩配置

default:
  total_window_tokens: 128000
  output_reserve_tokens: 8000
  expected_tool_growth_tokens: 5000
  safety_margin_tokens: 4000
  keep_recent_turns: 3
  compression_levels:
    l1_threshold: 0.6
    l2_threshold: 0.75
    l3_threshold: 0.85

code_review_agent:
  # 代码审查：工具返回大（源码、日志），输出也大（修改后的代码）
  total_window_tokens: 128000
  output_reserve_tokens: 16000
  expected_tool_growth_tokens: 20000
  safety_margin_tokens: 4000
  keep_recent_turns: 5
  tool_preprocess:
    read_file:
      max_lines: 300
    run_tests:
      keep_failed_only: true

customer_service_agent:
  # 客服：对话轮次多，但单轮内容短
  total_window_tokens: 32000
  output_reserve_tokens: 2000
  expected_tool_growth_tokens: 1000
  safety_margin_tokens: 2000
  keep_recent_turns: 10
  enable_summary: true

data_analysis_agent:
  # 数据分析：文件大，查询结果可能很大
  total_window_tokens: 128000
  output_reserve_tokens: 8000
  expected_tool_growth_tokens: 30000
  safety_margin_tokens: 6000
  keep_recent_turns: 3
  tool_preprocess:
    query_database:
      max_rows: 100
      summarize_large_result: true

"""配置加载器"""
import yaml


def load_compression_config(task_type: str, config_path: str = "context_compression.yaml") -> CompressionConfig:
    """根据任务类型加载压缩配置"""
    with open(config_path, "r", encoding="utf-8") as f:
        all_configs = yaml.safe_load(f)

    # 合并默认配置和任务特定配置
    default = all_configs.get("default", {})
    task_config = all_configs.get(task_type, {})
    merged = {**default, **task_config}

    return CompressionConfig(
        total_window_tokens=merged["total_window_tokens"],
        output_reserve_tokens=merged["output_reserve_tokens"],
        expected_tool_growth_tokens=merged["expected_tool_growth_tokens"],
        safety_margin_tokens=merged["safety_margin_tokens"],
        keep_recent_turns=merged["keep_recent_turns"],
    )

7.5 压缩效果监控

上线后必须监控压缩效果，及时发现问题：

"""
压缩效果监控指标
"""
from dataclasses import dataclass
from typing import List
from collections import defaultdict


@dataclass
class CompressionMetrics:
    """压缩监控指标"""
    total_sessions: int = 0
    total_compressions: int = 0
    compression_failures: int = 0
    avg_compression_ratio: float = 0.0
    repeated_read_count: int = 0  # 压缩后重新读取同一文件的次数
    task_success_rate: float = 0.0
    avg_tokens_per_call: float = 0.0


class CompressionMonitor:
    """压缩监控器"""

    def __init__(self):
        self.metrics = defaultdict(CompressionMetrics)

    def record_compression(
        self,
        task_type: str,
        before_tokens: int,
        after_tokens: int,
        success: bool,
    ):
        """记录一次压缩"""
        m = self.metrics[task_type]
        m.total_compressions += 1
        if not success:
            m.compression_failures += 1
        else:
            ratio = (before_tokens - after_tokens) / before_tokens
            m.avg_compression_ratio = (
                m.avg_compression_ratio * (m.total_compressions - 1) + ratio
            ) / m.total_compressions

    def record_repeated_read(self, task_type: str, file_path: str):
        """记录压缩后的重复读取（可能是压缩丢了关键信息）"""
        self.metrics[task_type].repeated_read_count += 1

    def get_report(self) -> str:
        """生成监控报告"""
        lines = ["=== 上下文压缩监控报告 ==="]
        for task_type, m in self.metrics.items():
            lines.append(f"\n[{task_type}]")
            lines.append(f"  总压缩次数: {m.total_compressions}")
            lines.append(f"  压缩失败率: {m.compression_failures / max(m.total_compressions, 1):.2%}")
            lines.append(f"  平均压缩率: {m.avg_compression_ratio:.2%}")
            lines.append(f"  重复读取次数: {m.repeated_read_count}")

            # 告警判断
            if m.avg_compression_ratio > 0.9:
                lines.append("  [告警] 压缩率过高，可能丢失关键信息")
            if m.repeated_read_count > m.total_compressions * 0.3:
                lines.append("  [告警] 重复读取过多，压缩时机可能过早")

        return "\n".join(lines)

八、四种上下文管理策略对比

对比维度	滑动窗口	摘要压缩	检索增强RAG	分层记忆+动态压缩
核心思想	只保留最近N轮，更早的直接丢弃	用LLM生成摘要替换旧消息	历史存外部向量库，按需检索	短期完整+中期摘要+长期检索，动态阈值触发
信息保留度	低，被截断的完全丢失	中，保留语义但丢细节	高，全量存储但检索可能漏	高，分层存储，关键信息不丢
实现复杂度	极低，几行代码	中等，需要LLM调用和提示词设计	高，需要向量库和检索逻辑	很高，需要多层存储和调度
Token节省效果	中，固定窗口大小	高，摘要通常只有原文10%-30%	极高，上下文只放当前相关内容	极高，按需加载，动态调整
额外成本	零	每次压缩需要LLM调用费用	向量库存储和检索费用	综合成本最高但效果最好
适用场景	短对话、简单任务	中等长度对话、通用场景	长文档问答、知识库场景	企业级Agent、复杂多轮任务
典型框架	ConversationBufferWindowMemory	ConversationSummaryMemory	LangChain Retriever	MemGPT / Deep Agents
信息丢失风险	高，硬截断	中，摘要可能遗漏	低，全量存储但检索不全	低，多层备份
响应延迟	无额外延迟	压缩时有额外延迟	检索有额外延迟	压缩和检索都可能有延迟
阈值灵活性	固定窗口大小	通常固定Token阈值	不涉及阈值	动态计算，按任务调整

8.1 选型建议

新手入门/简单场景：滑动窗口 + 摘要压缩。实现简单，能解决80%的问题。

文档问答/知识库场景：RAG为主，摘要压缩为辅。长文档不需要全部放上下文。

企业级Agent/复杂任务：分层记忆+动态压缩。虽然实现复杂，但这是唯一能在长任务中保持稳定性的方案。

成本敏感场景：优先做工具结果预处理，很多时候把工具返回裁剪好就不需要压缩了。

九、常用场景教学

9.1 场景一：代码审查Agent

特点：工具返回大（源码、测试日志），输出也大（修改后的代码），多轮排查。

配置建议：

code_review_config = CompressionConfig(
    total_window_tokens=128000,
    output_reserve_tokens=16000,      # 生成修改后的代码需要大量空间
    expected_tool_growth_tokens=20000, # 读文件+跑测试可能返回很多
    safety_margin_tokens=4000,
    keep_recent_turns=5,              # 代码排查需要保留最近的上下文
)

关键技巧：

• 读取文件时只返回相关函数，不要返回整个文件
• 测试日志只保留失败用例，通过的用例只保留计数
• 压缩时必须保留"已修改的文件列表"和"待验证项"

9.2 场景二：智能客服

特点：对话轮次多，单轮内容短，用户偏好和订单信息需要长期保留。

配置建议：

customer_service_config = CompressionConfig(
    total_window_tokens=32000,
    output_reserve_tokens=2000,
    expected_tool_growth_tokens=1000,
    safety_margin_tokens=2000,
    keep_recent_turns=10,  # 客服需要保留较多对话上下文
)

关键技巧：

• 用户明确说的偏好（“我要退款”、“不要打电话”）必须提取到摘要中
• 订单号、用户ID等关键信息要单独存储，不能只在对话里
• 可以用"用户画像"机制，把用户信息从对话中提取出来独立维护

9.3 场景三：数据分析Agent

特点：数据文件大，SQL查询结果可能有上万行，图表生成需要空间。

配置建议：

data_analysis_config = CompressionConfig(
    total_window_tokens=128000,
    output_reserve_tokens=8000,
    expected_tool_growth_tokens=30000,  # 查询结果可能非常大
    safety_margin_tokens=6000,
    keep_recent_turns=3,
)

关键技巧：

• 查询结果超过100行时只返回摘要和前20行
• 数据文件的元信息（列名、行数、统计量）要保留，原始数据不进上下文
• 压缩时保留"已验证的结论"和"待分析的问题"

9.4 场景四：文档问答

特点：文档本身很长，但单次问答只需要相关片段。

配置建议：

• 不使用上下文压缩，使用RAG
• 上下文只放系统指令+检索到的相关片段+当前对话
• 文档全部存在向量库，每次问答检索Top-K片段

关键技巧：

• 文档分块时要有重叠，避免答案被切断
• 检索结果要做重排序，不是相似度高就一定相关
• 多轮问答要把历史问题也加入检索查询

9.5 场景五：多工具调用Agent

特点：工具定义本身就占很多Token，工具返回结果大小差异大。

配置建议：

multi_tool_config = CompressionConfig(
    total_window_tokens=128000,
    output_reserve_tokens=8000,
    expected_tool_growth_tokens=15000,
    safety_margin_tokens=4000,
    keep_recent_turns=3,
)

关键技巧：

• 工具定义要精简，描述不要写小作文
• 可以用"工具分组"机制，当前阶段只加载相关工具的定义
• 工具返回结果必须预处理，大文件先存外部再返回路径

9.6 场景六：多Agent协作

特点：Agent之间传递消息，每个Agent有自己的上下文，消息传递会重复占用。

配置建议：

• Agent之间传递的不是完整对话，而是结构化的"任务交接单"
• 交接单包含：任务目标、已完成工作、关键结论、下一步建议
• 每个Agent独立管理自己的上下文，不继承完整历史

@dataclass
class AgentHandover:
    """Agent间交接单"""
    task_goal: str
    completed_work: List[str]
    key_findings: List[str]
    next_steps: List[str]
    constraints: List[str]
    relevant_files: List[str]

十、面试官高频面试题

10.1 基础概念题

Q1：什么是上下文压缩？为什么Agent需要上下文压缩？

参考答案：
上下文压缩是指当对话历史和工具结果累积到接近模型上下文窗口上限时，通过摘要、剪枝、提取等手段将已有内容收敛为更短但保留关键信息的形式，从而释放空间使任务持续进行。

Agent需要压缩的原因：第一，模型上下文窗口有上限，超限会报错或输出截断；第二，Agent的上下文增长远快于普通对话，工具返回结果可能一次就增加几万Token；第三，长上下文会增加成本、降低响应速度、还可能出现"中间遗忘"导致质量下降。

Q2：Token和字数有什么区别？为什么不能用字数估算Token？

参考答案：
Token是模型处理文本的基本单位，不等于字数。中文通常1个字约1-2个Token，英文1个单词约1-1.5个Token，代码和特殊符号的Token密度更高。不同模型的分词器不同，同样的文本在不同模型下Token数可能不同。因此必须使用模型对应的tokenizer精确统计，不能用len(text)估算。

Q3：内部状态和外部状态在上下文压缩中怎么理解？

参考答案：
可以类比享元模式的概念。上下文里的"系统指令、角色设定、硬约束"相当于内部状态，是不变的、必须保留的；"对话历史、工具结果、中间过程"相当于外部状态，是可变的、可以压缩的。压缩时要确保内部状态完整保留，只对外部状态做处理。

10.2 触发策略题

Q4：固定80%阈值压缩有什么问题？

参考答案：
三个问题：第一，没有预留模型输出空间，等输入到80%时，加上输出可能已经超限；第二，没有考虑工具返回的增长，下一步工具可能返回大文件，80%时再压缩已经来不及；第三，没有区分任务类型，简单对话和代码排查用同一个阈值不合理。正确的做法是动态计算阈值：总窗口减去输出预留、预计工具增长、安全余量。

Q5：怎么确定压缩阈值？给出计算公式。

参考答案：
动态阈值公式：触发阈值 = 总窗口大小 - 输出预留 - 预计工具增长 - 安全余量。

其中输出预留根据任务类型确定（生成代码需要多，简单问答需要少），预计工具增长根据下一步要调用的工具历史返回大小估算，安全余量通常取总窗口的5%-10%应对计数误差。压缩后的恢复目标要比触发阈值低20%-30%，避免刚压完又触发。

Q6：压缩到多少合适？压缩目标怎么定？

参考答案：
压缩目标（恢复目标）要和触发阈值保持足够间隔，通常比触发阈值低20%-30%。比如触发阈值20000，恢复目标设14000，这样压缩后有6000的缓冲空间。目标太低会丢失过多信息，目标太高会频繁触发压缩。具体值要根据任务的平均每轮Token增长量来定，增长快的任务间隔要大一些。

10.3 实现方案题

Q7：滑动窗口、摘要压缩、RAG三种方案的区别和适用场景？

参考答案：
滑动窗口是硬截断，只保留最近N轮，实现简单但信息丢失严重，适合短对话简单任务。摘要压缩用LLM把旧对话生成摘要，保留语义但丢细节，适合中等长度通用场景，是最常用的方案。RAG把历史存外部向量库按需检索，信息保留最完整但实现复杂，适合长文档和知识库场景。企业级复杂Agent通常三种结合使用：短期用滑动窗口保留最近对话，中期用摘要压缩，长期用RAG存储。

Q8：压缩时怎么保证关键信息不丢失？

参考答案：
四个措施：第一，系统消息和硬约束永远不压缩，单独保留；第二，摘要提示词中明确要求保留用户目标、关键参数、已确认结论、待办事项；第三，压缩后做校验，检查摘要中是否包含关键约束词；第四，重要信息（如用户ID、订单号、已修改文件）提取到独立的结构化字段中，不依赖摘要保留。

Q9：工具返回结果太大怎么办？

参考答案：
不能只靠压缩，要在工具结果进入上下文之前做预处理。具体手段：第一，截断，长日志只保留错误行和摘要；第二，分页，大文件分块读取，每次只读相关部分；第三，外部存储，大结果存文件或Redis，上下文只存路径；第四，工具侧限制，让工具本身只返回必要信息，比如数据库查询限制行数、日志查询只返回错误级别。

10.4 工程实践题

Q10：压缩后模型反复重读同一个文件，说明什么？怎么解决？

参考答案：
说明压缩时机过早或摘要质量太差，把模型后续还需要的细节丢掉了。解决方法：第一，提高触发阈值，晚一点压缩；第二，优化摘要提示词，要求保留具体参数值和配置信息；第三，压缩时保留最近更多轮对话；第四，监控重复读取率，如果超过30%就说明压缩策略有问题。

Q11：多轮对话中怎么保持连贯性？

参考答案：
三个要点：第一，摘要中必须包含"当前进度"和"下一步计划"，让模型知道任务做到哪了；第二，保留最近3-5轮的完整对话，这是连贯性的关键；第三，工具调用和结果要成对保留，不能只删结果留下调用，否则模型会困惑。

Q12：压缩本身也需要调用LLM，如果压缩时上下文已经满了怎么办？

参考答案：
这就是为什么不能等到接近上限才压缩。触发阈值要留足压缩调用本身需要的空间。如果已经遇到这种情况，降级方案是：不调用LLM，直接做硬截断（保留系统消息+最近几轮），先把空间腾出来，后续再做高质量压缩。更稳妥的做法是设置两级阈值：第一级触发正常摘要压缩，第二级（更高）触发紧急硬截断。

10.5 对比辨析题

Q13：上下文压缩和KV Cache压缩是一回事吗？

参考答案：
不是一回事。上下文压缩是应用层的策略，在调用模型前对输入内容做裁剪和摘要，减少发送给模型的Token数。KV Cache压缩是推理引擎层的优化，在模型内部对注意力机制的Key-Value缓存做压缩，减少显存占用。两者目标都是节省资源，但发生在不同层面，可以同时使用。

Q14：上下文压缩和RAG有什么关系？是替代关系吗？

参考答案：
不是替代，是互补。RAG解决"知识从哪里来"的问题，把外部知识按需检索进上下文；上下文压缩解决"上下文满了怎么办"的问题，把已有的内容精简。实际系统中通常同时使用：RAG控制进入上下文的内容量，压缩控制已累积内容的膨胀。RAG做得好可以减少压缩的频率，但不能完全替代压缩，因为对话历史和工具结果还是会累积。

Q15：扩大上下文窗口（比如从32K到128K）还需要压缩吗？

参考答案：
仍然需要。第一，成本随上下文线性增长，128K的调用费用是32K的4倍；第二，长上下文有"中间遗忘"问题，模型对中间部分的注意力下降；第三，Agent的上下文增长速度很快，再大的窗口也会满，只是时间问题；第四，很多API有单独的输出限制，不是总窗口够了就行。扩大窗口可以降低压缩频率，但不能消除压缩的必要性。

---

本文为原创文章，如需转载，请联系作者获得授权，并注明出处。

转载自 CSDN-专业IT技术社区

原文链接：https://blog.csdn.net/zxc18344522713/article/details/168652795