在实际 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 models 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\activate2.2 安装必要依赖库我们将使用以下几个库tiktokenOpenAI 开源的令牌编码器用于精确计算文本的令牌数即使不使用 OpenAI 的 API其编码方式也是业内的参考标准之一。difflibPython 标准库用于比较序列如字符串、列表之间的差异是生成差异报告的核心。colorama可选用于在终端中为差异输出添加颜色提升可读性。通过 pip 安装pip install tiktoken colorama2.3 项目结构规划创建一个清晰的项目目录有助于组织代码。compactdiff_demo/ ├── requirements.txt # 依赖列表 ├── compactdiff.py # 主工具脚本 ├── session_loader.py # 会话数据加载模块 ├── compaction_simulator.py # 压缩策略模拟器 ├── diff_visualizer.py # 差异可视化模块 └── sample_session.json # 示例会话数据requirements.txt文件内容如下tiktoken0.5.0 colorama0.4.63. 构建核心模块从会话加载到差异生成现在我们开始实现“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(messagescompacted_messages, model_context_windowself.session.model_context_window, encodingself.session.encoding) current_tokens new_session.total_tokens() return AgentSession(messagescompacted_messages, model_context_windowself.session.model_context_window, encodingself.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(messagescompacted_messages, model_context_windowself.session.model_context_window, encodingself.session.encoding) current_tokens new_session.total_tokens() if to_summarize: # 调用外部总结函数生成一条总结性消息 summary_text summarizer_func(to_summarize) summary_msg Message(rolesystem, contentf[Summary of earlier conversation]: {summary_text}) summary_msg.calculate_tokens(self.session.encoding) # 将总结消息插入回会话开头或合适的位置 compacted_messages.insert(0, summary_msg) # 注意插入后令牌数可能仍超限需要递归或进一步处理。此处为简化示例。 return AgentSession(messagescompacted_messages, model_context_windowself.session.model_context_window, encodingself.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(keylambda 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(messagescompacted_messages, model_context_windowself.session.model_context_window, encodingself.session.encoding) current_tokens new_session.total_tokens() # 移除后列表已重新排序下一条待移除的仍然是当前列表中优先级最低的 # 最后将消息按原始顺序或时间顺序重新排列 # 这里我们假设原始会话顺序就是时间顺序我们需要根据消息内容或ID恢复顺序。 # 由于示例简化我们跳过复杂的重排序仅作演示。 # 实际应用中消息应带有唯一ID或时间戳。 return AgentSession(messagescompacted_messages, model_context_windowself.session.model_context_window, encodingself.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(autoresetTrue) # 初始化 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, fromfileoriginal_session, tofilecompacted_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(fOriginal Session: {orig_msg_count} messages, {orig_token_count} tokens (est.)) report_lines.append(fCompacted Session: {comp_msg_count} messages, {comp_token_count} tokens (est.)) report_lines.append(fTokens Removed: {orig_token_count - comp_token_count}) report_lines.append(fMessages 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, encodingutf-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_text4. 整合与实战运行一个完整的 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\npython\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\npython\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比如n10000也能正确计算只是计算时间会很长。如果你担心性能可以考虑使用矩阵快速幂算法将时间复杂度降到O(log n)。 }, { role: user, content: 请解释一下矩阵快速幂算法的原理并给出代码。 }, { role: assistant, content: 矩阵快速幂算法基于这样一个事实斐波那契数列可以通过矩阵幂运算来表示。\n令 F [[1, 1], [1, 0]]那么 F^n [[F_{n1}, F_n], [F_n, F_{n-1}]]。\n因此计算F^n就可以得到F_n。而计算矩阵的n次幂可以使用快速幂算法在O(log n)时间内完成。\n\n以下是Python实现\n\npython\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\npython\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, headersheaders)\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, indexFalse, encodingutf-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, encodingutf-8) as f: data json.load(f) messages [Message(rolemsg[role], contentmsg[content]) for msg in data[messages]] session AgentSession(messagesmessages, model_context_windowdata.get(model_context_window, 4096), encodingdata.get(encoding, cl100k_base)) # 预计算所有消息的令牌数 _ session.total_tokens() return session def main(): parser argparse.ArgumentParser(description模拟会话压缩并显示差异。) parser.add_argument(--session-file, defaultsample_session.json, help会话数据JSON文件路径 (默认: sample_session.json)) parser.add_argument(--target-tokens, typeint, default3000, help压缩后目标令牌数 (默认: 3000)) parser.add_argument(--strategy, choices[truncate, priority], defaulttruncate, 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观察按角色优先级toolassistantusersystem压缩的结果。你可能会发现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 测试对比不同压缩策略下智能体在关键业务指标如任务完成率、用户满意度上的表现用数据驱动策略选择。6.4 扩展工具能力当前的演示工具主要关注“发生了什么”。你可以将其扩展为更强大的“上下文治理”工具预测性压缩不仅仅在超限时被动压缩可以实时预测按照当前对话速度何时会超限并提前进行温和的、渐进式的压缩避免突然的大段信息丢失。交互式审查提供一个界面允许开发者在压缩发生前或发生后手动审查被标记为“待移除”的消息并可以手动调整或恢复。根本原因分析将频繁导致压缩的会话模式例如用户持续发送极长消息识别出来反馈给产品或设计侧从源头优化交互设计。理解并掌控上下文压缩是构建可靠、可预测的长对话 AI 应用不可或缺的一环。通过实现和集成类似 Compactdiff 的工具你将能从黑盒中取出关键信息将调试过程从猜测变为精确的分析从而显著提升智能体系统的稳定性和用户体验。 LLM上下文压缩原理与Compactdiff工具：可视化会话压缩差异
在实际 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 models 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\activate2.2 安装必要依赖库我们将使用以下几个库tiktokenOpenAI 开源的令牌编码器用于精确计算文本的令牌数即使不使用 OpenAI 的 API其编码方式也是业内的参考标准之一。difflibPython 标准库用于比较序列如字符串、列表之间的差异是生成差异报告的核心。colorama可选用于在终端中为差异输出添加颜色提升可读性。通过 pip 安装pip install tiktoken colorama2.3 项目结构规划创建一个清晰的项目目录有助于组织代码。compactdiff_demo/ ├── requirements.txt # 依赖列表 ├── compactdiff.py # 主工具脚本 ├── session_loader.py # 会话数据加载模块 ├── compaction_simulator.py # 压缩策略模拟器 ├── diff_visualizer.py # 差异可视化模块 └── sample_session.json # 示例会话数据requirements.txt文件内容如下tiktoken0.5.0 colorama0.4.63. 构建核心模块从会话加载到差异生成现在我们开始实现“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(messagescompacted_messages, model_context_windowself.session.model_context_window, encodingself.session.encoding) current_tokens new_session.total_tokens() return AgentSession(messagescompacted_messages, model_context_windowself.session.model_context_window, encodingself.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(messagescompacted_messages, model_context_windowself.session.model_context_window, encodingself.session.encoding) current_tokens new_session.total_tokens() if to_summarize: # 调用外部总结函数生成一条总结性消息 summary_text summarizer_func(to_summarize) summary_msg Message(rolesystem, contentf[Summary of earlier conversation]: {summary_text}) summary_msg.calculate_tokens(self.session.encoding) # 将总结消息插入回会话开头或合适的位置 compacted_messages.insert(0, summary_msg) # 注意插入后令牌数可能仍超限需要递归或进一步处理。此处为简化示例。 return AgentSession(messagescompacted_messages, model_context_windowself.session.model_context_window, encodingself.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(keylambda 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(messagescompacted_messages, model_context_windowself.session.model_context_window, encodingself.session.encoding) current_tokens new_session.total_tokens() # 移除后列表已重新排序下一条待移除的仍然是当前列表中优先级最低的 # 最后将消息按原始顺序或时间顺序重新排列 # 这里我们假设原始会话顺序就是时间顺序我们需要根据消息内容或ID恢复顺序。 # 由于示例简化我们跳过复杂的重排序仅作演示。 # 实际应用中消息应带有唯一ID或时间戳。 return AgentSession(messagescompacted_messages, model_context_windowself.session.model_context_window, encodingself.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(autoresetTrue) # 初始化 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, fromfileoriginal_session, tofilecompacted_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(fOriginal Session: {orig_msg_count} messages, {orig_token_count} tokens (est.)) report_lines.append(fCompacted Session: {comp_msg_count} messages, {comp_token_count} tokens (est.)) report_lines.append(fTokens Removed: {orig_token_count - comp_token_count}) report_lines.append(fMessages 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, encodingutf-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_text4. 整合与实战运行一个完整的 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\npython\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\npython\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比如n10000也能正确计算只是计算时间会很长。如果你担心性能可以考虑使用矩阵快速幂算法将时间复杂度降到O(log n)。 }, { role: user, content: 请解释一下矩阵快速幂算法的原理并给出代码。 }, { role: assistant, content: 矩阵快速幂算法基于这样一个事实斐波那契数列可以通过矩阵幂运算来表示。\n令 F [[1, 1], [1, 0]]那么 F^n [[F_{n1}, F_n], [F_n, F_{n-1}]]。\n因此计算F^n就可以得到F_n。而计算矩阵的n次幂可以使用快速幂算法在O(log n)时间内完成。\n\n以下是Python实现\n\npython\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\npython\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, headersheaders)\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, indexFalse, encodingutf-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, encodingutf-8) as f: data json.load(f) messages [Message(rolemsg[role], contentmsg[content]) for msg in data[messages]] session AgentSession(messagesmessages, model_context_windowdata.get(model_context_window, 4096), encodingdata.get(encoding, cl100k_base)) # 预计算所有消息的令牌数 _ session.total_tokens() return session def main(): parser argparse.ArgumentParser(description模拟会话压缩并显示差异。) parser.add_argument(--session-file, defaultsample_session.json, help会话数据JSON文件路径 (默认: sample_session.json)) parser.add_argument(--target-tokens, typeint, default3000, help压缩后目标令牌数 (默认: 3000)) parser.add_argument(--strategy, choices[truncate, priority], defaulttruncate, 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观察按角色优先级toolassistantusersystem压缩的结果。你可能会发现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 测试对比不同压缩策略下智能体在关键业务指标如任务完成率、用户满意度上的表现用数据驱动策略选择。6.4 扩展工具能力当前的演示工具主要关注“发生了什么”。你可以将其扩展为更强大的“上下文治理”工具预测性压缩不仅仅在超限时被动压缩可以实时预测按照当前对话速度何时会超限并提前进行温和的、渐进式的压缩避免突然的大段信息丢失。交互式审查提供一个界面允许开发者在压缩发生前或发生后手动审查被标记为“待移除”的消息并可以手动调整或恢复。根本原因分析将频繁导致压缩的会话模式例如用户持续发送极长消息识别出来反馈给产品或设计侧从源头优化交互设计。理解并掌控上下文压缩是构建可靠、可预测的长对话 AI 应用不可或缺的一环。通过实现和集成类似 Compactdiff 的工具你将能从黑盒中取出关键信息将调试过程从猜测变为精确的分析从而显著提升智能体系统的稳定性和用户体验。