当前位置：首页 > news >正文

LLM上下文压缩原理与Compactdiff工具：可视化会话压缩差异

news 2026/8/11 2:25:48

在实际 AI 应用开发和调试过程中，尤其是在使用大语言模型（LLM）构建智能体（Agent）时，我们经常会遇到一个棘手的问题：会话（Session）随着交互轮次的增加，其包含的上下文（Context）会越来越长。当上下文长度超过模型的最大限制时，系统会触发一个称为“压缩（Compaction）”的机制，以丢弃部分历史信息，确保新的请求能够被模型处理。然而，这个过程通常是黑盒的，开发者无法直观地知道“压缩”到底丢弃了哪些对话内容，这给调试会话逻辑、理解智能体行为以及优化提示词（Prompt）带来了巨大困难。

“Compactdiff”这一工具概念，正是为了解决上述痛点而生。它旨在提供一个清晰的视图，让开发者能够对比压缩前后的会话内容，精确地看到哪些消息、指令或关键上下文在压缩过程中被移除。这对于诊断因上下文丢失导致的智能体行为异常、性能下降或“遗忘”先前指令等问题至关重要。本文将深入探讨会话压缩的原理，并基于“Compactdiff”的核心思想，构建一个可实操的、用于分析和可视化压缩过程的技术方案。无论你是正在开发基于 LLM 的聊天机器人、自动化工作流还是复杂的多步推理智能体，理解并监控上下文压缩都是提升系统稳定性和可预测性的关键一步。

1. 理解 LLM 上下文压缩：为什么需要以及如何发生

在深入实现之前，我们必须先厘清几个核心概念：上下文窗口、会话、令牌（Token）以及压缩策略。这是理解后续所有操作和排查工作的基础。

1.1 上下文窗口与令牌限制

大型语言模型（如 GPT 系列、Claude、LLaMA 等）在处理输入时，有一个硬性的技术限制，称为“上下文窗口”或“最大上下文长度”。这个限制通常以“令牌”为单位。一个令牌可以是一个单词、一个子词甚至一个标点符号。例如，一个模型可能拥有 4096、8192 或 128K 的上下文窗口。当我们将用户的问题、系统的指令、历史的对话记录以及模型的回复全部编码为令牌序列后，如果这个序列的总长度超过了模型的最大限制，请求就会失败，并返回类似error during compaction: api error: 400 this model's maximum context length的错误。

1.2 会话的生命周期与增长

在一个典型的智能体会话中，上下文通常以“消息”列表的形式组织，每条消息包含角色（如user,assistant,system）和内容。每次交互（一轮用户输入和模型输出）都会向这个列表追加新的消息。随着对话轮次增加，上下文列表不断增长，令牌数也随之攀升。即使单条消息不长，数十轮对话后也很容易触及模型的令牌上限。

1.3 压缩策略：黑盒下的数据丢弃

当上下文长度接近或超过限制时，应用层或中间件（而非模型本身）必须采取行动。常见的压缩策略包括：

截断（Truncation）：直接丢弃列表中最旧的消息（通常是对话的开头部分）。这是最简单粗暴的方式。
总结（Summarization）：调用模型本身，将早期的、冗长的对话内容总结成一段简短的摘要，然后用摘要替换原始的多条消息。这能保留更多语义，但消耗额外的计算资源。
选择性丢弃（Selective Dropping）：基于启发式规则（如消息的重要性、角色、长度）决定丢弃哪些消息。例如，可能优先保留system指令和最近的对话。
滑动窗口（Sliding Window）：始终保持上下文在一个固定长度内，新的消息加入时，最旧的消息被移出。

无论采用哪种策略，其核心动作都是“从原始会话消息列表中移除一部分内容”。而“Compactdiff”要做的，就是在压缩动作发生前后，对会话状态进行快照和差异比较。

1.4 压缩引发的典型问题

如果不清楚压缩丢弃了什么，你可能会遇到以下难以调试的现象：

智能体“忘记”了用户在对话早期设定的规则或偏好。
智能体突然无法引用之前讨论过的关键信息（如文件名、数字、决策依据）。
智能体的回复风格或能力发生突变，可能是因为关键的system提示词在压缩中被意外移除了。
在长文档问答中，智能体丢失了对文档前半部分内容的引用能力。

2. 环境准备与核心依赖

为了构建一个能够模拟和可视化压缩过程的工具，我们需要搭建一个简单的 Python 开发环境。这个环境将允许我们加载会话、模拟压缩算法、计算差异并生成报告。

2.1 Python 环境与包管理

建议使用 Python 3.8 或更高版本。使用venv或conda创建独立的虚拟环境以避免依赖冲突。

# 创建并激活虚拟环境 (Linux/macOS) python3 -m venv compactdiff_env source compactdiff_env/bin/activate # 创建并激活虚拟环境 (Windows) python -m venv compactdiff_env compactdiff_env\Scripts\activate

2.2 安装必要依赖库

我们将使用以下几个库：

tiktoken：OpenAI 开源的令牌编码器，用于精确计算文本的令牌数（即使不使用 OpenAI 的 API，其编码方式也是业内的参考标准之一）。
difflib：Python 标准库，用于比较序列（如字符串、列表）之间的差异，是生成差异报告的核心。
colorama（可选）：用于在终端中为差异输出添加颜色，提升可读性。

通过 pip 安装：

pip install tiktoken colorama

2.3 项目结构规划

创建一个清晰的项目目录，有助于组织代码。

compactdiff_demo/ ├── requirements.txt # 依赖列表 ├── compactdiff.py # 主工具脚本 ├── session_loader.py # 会话数据加载模块 ├── compaction_simulator.py # 压缩策略模拟器 ├── diff_visualizer.py # 差异可视化模块 └── sample_session.json # 示例会话数据

requirements.txt文件内容如下：

tiktoken>=0.5.0 colorama>=0.4.6

3. 构建核心模块：从会话加载到差异生成

现在，我们开始实现“Compactdiff”的核心功能。我们将分模块构建，最终将它们串联起来。

3.1 定义会话数据结构

首先，我们需要一个统一的数据结构来表示会话中的一条消息和一个完整的会话。

在session_loader.py中：

from dataclasses import dataclass from typing import List, Optional import json @dataclass class Message: """表示会话中的一条消息。""" role: str # 例如：'system', 'user', 'assistant', 'tool' content: str # 可以扩展其他字段，如 timestamp, id, tokens 等 tokens: Optional[int] = None # 缓存该条消息的令牌数 def calculate_tokens(self, encoding_name: str = "cl100k_base") -> int: """使用 tiktoken 计算消息内容的令牌数。""" import tiktoken try: enc = tiktoken.get_encoding(encoding_name) except KeyError: # 如果指定的编码不存在，回退到一种常见编码 enc = tiktoken.get_encoding("cl100k_base") # 通常角色名也会占用少量令牌，这里简化处理，只计算内容 # 更精确的计算需要将角色和内容按API实际格式拼接 self.tokens = len(enc.encode(self.content)) return self.tokens @dataclass class AgentSession: """表示一个智能体会话。""" messages: List[Message] model_context_window: int = 4096 # 假设模型上下文窗口为4096令牌 encoding: str = "cl100k_base" def total_tokens(self) -> int: """计算当前会话所有消息的总令牌数。""" total = 0 for msg in self.messages: if msg.tokens is None: msg.calculate_tokens(self.encoding) total += msg.tokens # 注意：实际API调用时，消息间的分隔符、特殊令牌等也会占用额外令牌。 # 此处为简化演示，仅作近似计算。生产环境需参考对应模型的令牌化规则。 return total def is_over_limit(self, reserve_tokens: int = 500) -> bool: """检查会话是否超过限制，预留一部分令牌给模型生成回复。""" return self.total_tokens() > (self.model_context_window - reserve_tokens)

3.2 实现压缩策略模拟器

接下来，在compaction_simulator.py中实现几种常见的压缩算法。

from session_loader import AgentSession, Message from typing import Callable, List import tiktoken class CompactionSimulator: """模拟各种上下文压缩策略。""" def __init__(self, session: AgentSession): self.session = session self.original_messages = session.messages.copy() # 保存原始副本用于比较 def compact_truncate_oldest(self, target_tokens: int) -> AgentSession: """策略1：从最旧的消息开始丢弃，直到总令牌数低于目标值。""" compacted_messages = self.session.messages.copy() current_tokens = self.session.total_tokens() while current_tokens > target_tokens and len(compacted_messages) > 0: # 移除列表第一条（最旧）消息 removed_msg = compacted_messages.pop(0) # 重新计算令牌数（简单起见，这里减去被移除消息的令牌） if removed_msg.tokens: current_tokens -= removed_msg.tokens else: # 如果未缓存，需要重新计算整个会话，成本较高 new_session = AgentSession(messages=compacted_messages, model_context_window=self.session.model_context_window, encoding=self.session.encoding) current_tokens = new_session.total_tokens() return AgentSession(messages=compacted_messages, model_context_window=self.session.model_context_window, encoding=self.session.encoding) def compact_summarize_oldest(self, target_tokens: int, summarizer_func: Callable[[List[Message]], str]) -> AgentSession: """ 策略2：总结最旧的N条消息。 注意：这是一个高级策略的框架，需要传入一个实际的总结函数（可能调用另一个LLM）。 """ compacted_messages = self.session.messages.copy() current_tokens = self.session.total_tokens() to_summarize = [] # 收集最旧的消息直到满足令牌要求 while current_tokens > target_tokens and len(compacted_messages) > 0: msg = compacted_messages.pop(0) to_summarize.append(msg) if msg.tokens: current_tokens -= msg.tokens else: new_session = AgentSession(messages=compacted_messages, model_context_window=self.session.model_context_window, encoding=self.session.encoding) current_tokens = new_session.total_tokens() if to_summarize: # 调用外部总结函数，生成一条总结性消息 summary_text = summarizer_func(to_summarize) summary_msg = Message(role='system', content=f"[Summary of earlier conversation]: {summary_text}") summary_msg.calculate_tokens(self.session.encoding) # 将总结消息插入回会话开头（或合适的位置） compacted_messages.insert(0, summary_msg) # 注意：插入后令牌数可能仍超限，需要递归或进一步处理。此处为简化示例。 return AgentSession(messages=compacted_messages, model_context_window=self.session.model_context_window, encoding=self.session.encoding) def compact_drop_by_role_priority(self, target_tokens: int, role_priority: List[str] = None) -> AgentSession: """ 策略3：基于角色优先级丢弃消息。 默认优先级从低到高：'tool' < 'assistant' < 'user' < 'system'。 优先丢弃优先级低的消息。 """ if role_priority is None: role_priority = ['tool', 'assistant', 'user', 'system'] # 越靠后优先级越高 compacted_messages = self.session.messages.copy() # 为每条消息附加优先级分数 for msg in compacted_messages: msg.priority_score = role_priority.index(msg.role) if msg.role in role_priority else -1 # 按优先级分数升序排序（分数低的先被考虑丢弃） compacted_messages.sort(key=lambda x: x.priority_score) current_tokens = self.session.total_tokens() index_to_remove = 0 while current_tokens > target_tokens and index_to_remove < len(compacted_messages): removed_msg = compacted_messages.pop(index_to_remove) # 总是移除当前列表的第一条（优先级最低的） if removed_msg.tokens: current_tokens -= removed_msg.tokens else: new_session = AgentSession(messages=compacted_messages, model_context_window=self.session.model_context_window, encoding=self.session.encoding) current_tokens = new_session.total_tokens() # 移除后，列表已重新排序，下一条待移除的仍然是当前列表中优先级最低的 # 最后，将消息按原始顺序（或时间顺序）重新排列 # 这里我们假设原始会话顺序就是时间顺序，我们需要根据消息内容或ID恢复顺序。 # 由于示例简化，我们跳过复杂的重排序，仅作演示。 # 实际应用中，消息应带有唯一ID或时间戳。 return AgentSession(messages=compacted_messages, model_context_window=self.session.model_context_window, encoding=self.session.encoding)

3.3 实现差异计算与可视化

这是“Compactdiff”的核心。在diff_visualizer.py中，我们将比较压缩前后的会话。

from session_loader import AgentSession, Message from difflib import Differ, unified_diff import sys from colorama import init, Fore, Back, Style init(autoreset=True) # 初始化 colorama class SessionDiffVisualizer: """计算并可视化会话压缩前后的差异。""" @staticmethod def messages_to_text_lines(messages: List[Message]) -> List[str]: """将会话消息列表转换为纯文本行列表，便于比较。""" lines = [] for i, msg in enumerate(messages): # 用清晰的格式标记每条消息 header = f"[{i:03d}] {msg.role.upper()}:" lines.append(header) # 将消息内容按行分割，并缩进 content_lines = msg.content.splitlines() for cl in content_lines: lines.append(f" {cl}") lines.append("") # 消息间空一行 return lines @staticmethod def compute_unified_diff(original: AgentSession, compacted: AgentSession) -> List[str]: """生成 unified diff 格式的差异。""" orig_lines = SessionDiffVisualizer.messages_to_text_lines(original.messages) comp_lines = SessionDiffVisualizer.messages_to_text_lines(compacted.messages) diff = list(unified_diff(orig_lines, comp_lines, fromfile='original_session', tofile='compacted_session', lineterm='')) return diff @staticmethod def compute_inline_diff(original: AgentSession, compacted: AgentSession): """生成行内差异（类似 git diff）。""" orig_lines = SessionDiffVisualizer.messages_to_text_lines(original.messages) comp_lines = SessionDiffVisualizer.messages_to_text_lines(compacted.messages) d = Differ() diff = list(d.compare(orig_lines, comp_lines)) return diff @staticmethod def print_colored_diff(diff_lines: List[str]): """在终端中打印带颜色的差异。""" for line in diff_lines: if line.startswith('---') or line.startswith('+++'): print(Fore.CYAN + line) elif line.startswith('@@'): print(Fore.YELLOW + line) elif line.startswith('+'): print(Fore.GREEN + line) elif line.startswith('-'): print(Fore.RED + line) else: print(line) @staticmethod def generate_diff_report(original: AgentSession, compacted: AgentSession, output_file: str = None): """生成一份完整的差异报告，包括统计信息。""" orig_token_count = original.total_tokens() comp_token_count = compacted.total_tokens() orig_msg_count = len(original.messages) comp_msg_count = len(compacted.messages) report_lines = [] report_lines.append("=" * 60) report_lines.append("COMPACTION DIFF REPORT") report_lines.append("=" * 60) report_lines.append(f"Original Session: {orig_msg_count} messages, {orig_token_count} tokens (est.)") report_lines.append(f"Compacted Session: {comp_msg_count} messages, {comp_token_count} tokens (est.)") report_lines.append(f"Tokens Removed: {orig_token_count - comp_token_count}") report_lines.append(f"Messages Removed: {orig_msg_count - comp_msg_count}") report_lines.append("-" * 60) report_lines.append("") # 找出被完全移除的消息（基于简单的内容匹配，实际应用可能需要消息ID） original_msg_set = set(f"{m.role}:{m.content[:100]}" for m in original.messages) # 前100字符作为指纹 compacted_msg_set = set(f"{m.role}:{m.content[:100]}" for m in compacted.messages) removed_msg_fingerprints = original_msg_set - compacted_msg_set if removed_msg_fingerprints: report_lines.append("MESSAGES COMPLETELY REMOVED:") for fp in removed_msg_fingerprints: # 从原始会话中找到对应的完整消息 for msg in original.messages: if f"{msg.role}:{msg.content[:100]}" == fp: report_lines.append(f" - [{msg.role}] {msg.content[:200]}...") break report_lines.append("") # 生成 unified diff report_lines.append("DETAILED UNIFIED DIFF:") report_lines.append("") diff = SessionDiffVisualizer.compute_unified_diff(original, compacted) report_lines.extend(diff) report_text = "\n".join(report_lines) if output_file: with open(output_file, 'w', encoding='utf-8') as f: f.write(report_text) print(f"[INFO] Diff report saved to: {output_file}") else: # 打印到控制台，并尝试着色 try: SessionDiffVisualizer.print_colored_diff(report_lines[:report_lines.index("DETAILED UNIFIED DIFF:")+2]) # 单独为详细的diff部分着色 detailed_diff_start = report_lines.index("DETAILED UNIFIED DIFF:") + 2 SessionDiffVisualizer.print_colored_diff(report_lines[detailed_diff_start:]) except (ValueError, ImportError): # 如果找不到索引或colorama不可用，打印全部 print(report_text) return report_text

4. 整合与实战：运行一个完整的 Compactdiff 流程

现在，我们将所有模块整合到一个主脚本compactdiff.py中，并提供一个示例会话来演示整个流程。

4.1 创建示例会话数据

在sample_session.json中，构造一个会触发压缩的长对话示例：

{ "model_context_window": 4096, "encoding": "cl100k_base", "messages": [ { "role": "system", "content": "你是一个乐于助人的AI助手。请用中文回答用户的问题。你的知识截止日期是2023年10月。" }, { "role": "user", "content": "请帮我写一个Python函数，计算斐波那契数列的第n项。" }, { "role": "assistant", "content": "当然，这是一个使用递归的Python函数，用于计算斐波那契数列的第n项。但请注意，递归方式在n较大时效率很低。\n\n```python\ndef fibonacci_recursive(n):\n if n <= 1:\n return n\n else:\n return fibonacci_recursive(n-1) + fibonacci_recursive(n-2)\n```" }, { "role": "user", "content": "递归的效率太低了。请提供一个使用迭代或动态规划的方法。" }, { "role": "assistant", "content": "你说得对。这是一个使用迭代（动态规划思想）的版本，时间复杂度为O(n)，空间复杂度为O(1)。\n\n```python\ndef fibonacci_iterative(n):\n if n <= 1:\n return n\n a, b = 0, 1\n for _ in range(2, n + 1):\n a, b = b, a + b\n return b\n```" }, { "role": "user", "content": "很好。现在假设这个数列的每一项都很大，可能会超过Python普通整数的范围，该如何处理？" }, { "role": "assistant", "content": "Python的整数是任意精度的，不会像其他语言那样发生溢出。所以上面的代码即使对于非常大的n（比如n=10000）也能正确计算，只是计算时间会很长。如果你担心性能，可以考虑使用矩阵快速幂算法，将时间复杂度降到O(log n)。" }, { "role": "user", "content": "请解释一下矩阵快速幂算法的原理，并给出代码。" }, { "role": "assistant", "content": "矩阵快速幂算法基于这样一个事实：斐波那契数列可以通过矩阵幂运算来表示。\n令 F = [[1, 1], [1, 0]]，那么 F^n = [[F_{n+1}, F_n], [F_n, F_{n-1}]]。\n因此，计算F^n就可以得到F_n。而计算矩阵的n次幂可以使用快速幂算法，在O(log n)时间内完成。\n\n以下是Python实现：\n\n```python\ndef matrix_multiply(a, b):\n return [\n [a[0][0]*b[0][0] + a[0][1]*b[1][0], a[0][0]*b[0][1] + a[0][1]*b[1][1]],\n [a[1][0]*b[0][0] + a[1][1]*b[1][0], a[1][0]*b[0][1] + a[1][1]*b[1][1]]\n ]\n\ndef matrix_power(matrix, n):\n result = [[1, 0], [0, 1]] # 单位矩阵\n while n > 0:\n if n % 2 == 1:\n result = matrix_multiply(result, matrix)\n matrix = matrix_multiply(matrix, matrix)\n n //= 2\n return result\n\ndef fibonacci_matrix(n):\n if n <= 1:\n return n\n F = [[1, 1], [1, 0]]\n powered = matrix_power(F, n - 1)\n return powered[0][0]\n```" }, { "role": "user", "content": "非常详细，谢谢。现在让我们换个话题。我想学习如何用Python进行Web爬虫，可以介绍一些基本的库和步骤吗？" }, { "role": "assistant", "content": "当然可以。Python进行Web爬虫常用的库有：\n1. **requests**: 用于发送HTTP请求，获取网页内容。\n2. **BeautifulSoup (bs4)**: 用于解析HTML/XML文档，提取数据。\n3. **Scrapy**: 一个强大的爬虫框架，适合大型、复杂的爬取任务。\n\n基本步骤：\n1. **分析目标网站**：查看网页结构，找到所需数据的位置。\n2. **发送请求**：使用requests.get(url)获取网页HTML。\n3. **解析内容**：使用BeautifulSoup(html_content, 'html.parser')创建解析对象，然后用find()或find_all()等方法定位元素。\n4. **提取数据**：从元素中获取文本、属性等。\n5. **存储数据**：保存到文件（如CSV、JSON）或数据库。\n6. **遵守规则**：检查robots.txt，设置合理的请求间隔，避免对服务器造成压力。\n\n一个简单的示例：爬取某个页面上的所有标题。" }, { "role": "user", "content": "请写一个爬取豆瓣电影Top250第一页电影名称和评分的示例代码。" }, { "role": "assistant", "content": "请注意，爬取网站数据前应确认其服务条款。以下是一个使用requests和BeautifulSoup的示例，仅用于学习目的：\n\n```python\nimport requests\nfrom bs4 import BeautifulSoup\nimport pandas as pd\n\nurl = 'https://movie.douban.com/top250'\nheaders = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'}\n\nresponse = requests.get(url, headers=headers)\nsoup = BeautifulSoup(response.text, 'html.parser')\n\nmovies = []\nfor item in soup.find_all('div', class_='item'):\n title_elem = item.find('span', class_='title')\n rating_elem = item.find('span', class_='rating_num')\n if title_elem and rating_elem:\n movies.append({\n 'title': title_elem.text,\n 'rating': rating_elem.text\n })\n\ndf = pd.DataFrame(movies)\nprint(df.head())\n# 可以保存为CSV: df.to_csv('douban_top250_page1.csv', index=False, encoding='utf-8-sig')\n```" } ] }

4.2 编写主程序逻辑

在compactdiff.py中：

#!/usr/bin/env python3 """ Compactdiff 演示工具：模拟并展示智能体会话压缩前后的差异。 """ import json import argparse from session_loader import AgentSession, Message from compaction_simulator import CompactionSimulator from diff_visualizer import SessionDiffVisualizer def load_session_from_file(filepath: str) -> AgentSession: """从JSON文件加载会话。""" with open(filepath, 'r', encoding='utf-8') as f: data = json.load(f) messages = [Message(role=msg['role'], content=msg['content']) for msg in data['messages']] session = AgentSession(messages=messages, model_context_window=data.get('model_context_window', 4096), encoding=data.get('encoding', 'cl100k_base')) # 预计算所有消息的令牌数 _ = session.total_tokens() return session def main(): parser = argparse.ArgumentParser(description='模拟会话压缩并显示差异。') parser.add_argument('--session-file', default='sample_session.json', help='会话数据JSON文件路径 (默认: sample_session.json)') parser.add_argument('--target-tokens', type=int, default=3000, help='压缩后目标令牌数 (默认: 3000)') parser.add_argument('--strategy', choices=['truncate', 'priority'], default='truncate', help='压缩策略: truncate (截断最旧) 或 priority (按角色优先级)') parser.add_argument('--output-report', help='将差异报告输出到指定文件') args = parser.parse_args() print(f"[INFO] 加载会话文件: {args.session_file}") original_session = load_session_from_file(args.session_file) print(f"[INFO] 原始会话: {len(original_session.messages)} 条消息, 约 {original_session.total_tokens()} 令牌。") print(f"[INFO] 模型上下文窗口: {original_session.model_context_window} 令牌。") print(f"[INFO] 压缩目标: <= {args.target_tokens} 令牌。") print(f"[INFO] 使用策略: {args.strategy}") simulator = CompactionSimulator(original_session) if args.strategy == 'truncate': compacted_session = simulator.compact_truncate_oldest(args.target_tokens) elif args.strategy == 'priority': # 使用默认角色优先级 compacted_session = simulator.compact_drop_by_role_priority(args.target_tokens) else: print(f"[ERROR] 未知策略: {args.strategy}") return print(f"[INFO] 压缩后会话: {len(compacted_session.messages)} 条消息, 约 {compacted_session.total_tokens()} 令牌。") print("\n" + "="*60) print("开始生成差异报告...") print("="*60) SessionDiffVisualizer.generate_diff_report(original_session, compacted_session, args.output_report) if __name__ == '__main__': main()

4.3 运行与验证

在项目根目录下运行命令，观察压缩差异：

python compactdiff.py --session-file sample_session.json --target-tokens 2500 --strategy truncate

预期你将看到类似以下的输出（在终端中会带有颜色）：

[INFO] 加载会话文件: sample_session.json [INFO] 原始会话: 12 条消息, 约 3892 令牌。 [INFO] 模型上下文窗口: 4096 令牌。 [INFO] 压缩目标: <= 2500 令牌。 [INFO] 使用策略: truncate [INFO] 压缩后会话: 8 条消息, 约 2487 令牌。 ============================================================ 开始生成差异报告... ============================================================ ============================================================ COMPACTION DIFF REPORT ============================================================ Original Session: 12 messages, 3892 tokens (est.) Compacted Session: 8 messages, 2487 tokens (est.) Tokens Removed: 1405 Messages Removed: 4 ------------------------------------------------------------ MESSAGES COMPLETELY REMOVED: - [system] 你是一个乐于助人的AI助手。请用中文回答用户的问题。你的知识截止日期是2023年10月。 - [user] 请帮我写一个Python函数，计算斐波那契数列的第n项。 - [assistant] 当然，这是一个使用递归的Python函数，用于计算斐波那契数列的第n项。但请注意，递归方式在n较大时效率很低。... - [user] 递归的效率太低了。请提供一个使用迭代或动态规划的方法。 DETAILED UNIFIED DIFF: --- original_session +++ compacted_session @@ -1,38 +1,26 @@ -[000] SYSTEM: - 你是一个乐于助人的AI助手。请用中文回答用户的问题。你的知识截止日期是2023年10月。 - -[001] USER: - 请帮我写一个Python函数，计算斐波那契数列的第n项。 - -[002] ASSISTANT: - 当然，这是一个使用递归的Python函数，用于计算斐波那契数列的第n项。但请注意，递归方式在n较大时效率很低。... - -[003] USER: - 递归的效率太低了。请提供一个使用迭代或动态规划的方法。 - -[004] ASSISTANT: +[000] ASSISTANT: 你说得对。这是一个使用迭代（动态规划思想）的版本，时间复杂度为O(n)，空间复杂度为O(1)。... ...

从报告中可以清晰地看到：

统计摘要：原始会话有12条消息约3892令牌，压缩后剩下8条消息约2487令牌，移除了4条消息和1405令牌。
被移除的消息列表：明确列出了被完全丢弃的4条消息，包括最初的system指令和前三轮关于斐波那契数列递归实现的对话。这是一个关键发现：压缩可能移除了至关重要的系统指令，导致智能体后续行为偏离预期。
详细的 Unified Diff：以标准 diff 格式展示了上下文的具体变化，-表示删除的行，+表示新增的行（在本例的截断策略中，通常没有新增行）。

尝试另一种策略：

python compactdiff.py --session-file sample_session.json --target-tokens 2500 --strategy priority

观察按角色优先级（tool<assistant<user<system）压缩的结果。你可能会发现system指令被保留了，但一些早期的user和assistant消息被移除。这演示了不同压缩策略对会话内容的不同影响。

5. 常见问题排查与调试指南

在实际集成或使用类似“Compactdiff”工具时，你可能会遇到以下问题。

5.1 令牌数计算不准确

现象：工具计算的令牌数与实际 API 调用消耗的令牌数有较大差异，导致压缩过早或过晚触发。原因与排查：

模型差异：不同模型（如gpt-3.5-turbo与gpt-4）的令牌化方式可能略有不同，即使使用相同的编码（如cl100k_base）。某些模型可能在消息格式中添加额外的特殊令牌。
格式封装：API 请求并非直接拼接消息内容。例如，OpenAI ChatCompletion API 会将消息列表序列化为特定的 JSON 格式，这些格式字符也会占用令牌。
函数调用/工具调用：如果消息中包含tool_calls或function_call等复杂结构，其令牌计算需要特殊处理。

解决方案：

使用官方库或近似计算：对于特定 API（如 OpenAI），使用其官方 SDK（如openai库）中的tokenizer工具进行精确计算是最可靠的。
增加安全余量：在计算出的令牌数基础上，增加一个安全余量（例如 10%-20%），避免因计算偏差导致请求失败。
实际测试校准：编写一个小脚本，发送不同长度的消息，从 API 响应头（如x-ratelimit-tokens-used）或账单中获取实际使用的令牌数，与本地计算值进行对比校准。

5.2 压缩后智能体行为异常

现象：压缩后，智能体“忘记”了关键信息，或开始以错误的身份（如不使用指定的语言）回答问题。排查步骤：

检查被移除的消息：使用 Compactdiff 报告，首先确认是否移除了system指令或包含核心规则的用户消息。
检查消息顺序：某些压缩策略可能打乱消息顺序。确保压缩后的消息列表保持了正确的时间或逻辑顺序。智能体通常依赖消息的顺序来理解对话流。
检查总结的保真度：如果使用总结策略，检查生成的摘要是否准确概括了原始对话的要点，是否存在信息扭曲或丢失。
模拟完整流程：在测试环境中，将压缩后的会话消息列表直接发送给模型 API，观察其回复，与压缩前的回复进行对比。

处理建议：

保护关键消息：在压缩逻辑中，为system消息和某些标记为关键（如important: true）的用户消息设置最高优先级，确保它们永远不会被丢弃。
使用语义保留策略：优先丢弃那些与当前对话主题相关性较低的消息（可通过嵌入向量计算相似度来判断），而不是简单地按时间或角色丢弃。
引入“记忆”模块：对于长对话，考虑引入外部记忆存储（如向量数据库），将重要的历史信息存储起来，并在需要时通过检索增强生成（RAG）的方式注入到上下文中，而不是全部塞进上下文窗口。

5.3 性能与效率问题

现象：压缩操作（尤其是总结策略）本身耗时过长，影响智能体响应速度。排查与优化：

分析瓶颈：使用性能分析工具（如 Python 的cProfile）确定时间是消耗在令牌计算、消息比较还是总结模型调用上。
缓存令牌数：如示例代码所示，为Message对象缓存计算好的令牌数，避免重复计算。
简化差异算法：对于超长会话，完整的difflib比较可能较慢。如果只关心哪些消息被完全移除，可以基于消息 ID 或内容哈希进行快速集合运算。
异步或离线压缩：如果压缩逻辑复杂，考虑将其移至后台异步任务，或定期进行离线压缩，而不是在每次请求前同步执行。

5.4 集成到现有框架

现象：不知道如何将 Compactdiff 的监控能力嵌入到现有的 LangChain、LlamaIndex 或自定义智能体框架中。集成思路：大多数框架在处理上下文窗口时，都会有一个“上下文管理”或“记忆管理”的环节。你需要找到框架中执行消息列表截断或总结的钩子（Hook）或回调函数（Callback）。

LangChain：可以自定义一个BaseMemory类，在其prune或相关方法中，在调用父类压缩逻辑前后，记录会话快照并进行差异计算。
LlamaIndex：在构建查询引擎时，可以关注chat_history的处理。在将历史记录送入模型前，可以复制一份用于后续比较。
自定义框架：在准备最终发送给模型 API 的消息列表（messages）之前，插入一个记录点。在调用压缩函数后，再插入一个记录点。比较这两个时间点的消息列表即可。

6. 生产环境最佳实践与扩展方向

将“Compactdiff”从演示工具升级为生产级监控组件，需要考虑以下几个方面：

6.1 增强监控与告警

记录压缩事件：每次发生压缩时，不仅记录差异报告，还应记录压缩前后的令牌数、被移除消息的元数据（角色、长度、时间戳）、使用的压缩策略以及会话 ID。这将构成宝贵的调试日志。
设置关键指标：定义并监控以下指标：
- compaction.rate：会话触发压缩的比例。
- compaction.tokens_dropped.avg：平均每次压缩丢弃的令牌数。
- compaction.critical_messages_lost：关键消息（如system）被丢弃的次数。
配置告警：当compaction.critical_messages_lost超过阈值，或单个会话丢弃的令牌比例异常高时，触发告警，提示可能需要优化提示词设计或调整压缩策略。

6.2 优化压缩策略

策略	优点	缺点	适用场景
截断最旧	实现简单，计算开销极低。	可能丢失对话早期的关键指令或设定。	对会话开头信息依赖性不高的简单对话。
按角色优先级	可以优先保留系统指令等关键消息。	规则较死板，可能误删重要的用户输入。	系统指令至关重要，且用户消息重要性随时间递减的场景。
总结	能最大程度保留历史语义信息。	实现复杂，需要调用模型，增加延迟和成本；总结可能失真。	对话历史连贯性极强的复杂任务，如多轮代码调试、长文档分析。
语义相似度筛选	丢弃与当前查询最不相关的历史消息，保留相关性高的。	需要计算嵌入向量，开销较大；依赖嵌入模型的质量。	基于检索的问答（RAG）场景，或话题频繁切换的长对话。
混合策略	结合多种策略优点，更灵活智能。	设计和调试更复杂。	大多数生产环境，可根据消息类型、长度、时间、语义等综合决策。

推荐做法：在生产环境中，通常从“按角色优先级”策略开始，并严格保护system消息。随着业务复杂化，逐步引入基于向量相似度的筛选，形成混合策略。同时，必须为压缩过程配备像 Compactdiff 这样的可视化监控工具，以便持续观察和调优策略效果。

6.3 将差异分析集成到开发工作流

自动化测试：在针对智能体的集成测试或端到端测试中，加入对压缩行为的断言。例如，可以断言“在任何压缩操作中，system消息不得被移除”。
调试面板：为你的智能体应用开发一个内部调试面板。当客服或测试人员报告智能体出现“遗忘”问题时，可以通过该面板输入会话 ID，直接查看该会话历史上的所有压缩记录和差异报告。
A/B 测试：对比不同压缩策略下，智能体在关键业务指标（如任务完成率、用户满意度）上的表现，用数据驱动策略选择。