想象这样一个场景：你的Agent正在帮用户分析一份财务报告。对话已经进行了30轮：
此时，上下文窗口的使用率已经达到了85%。还有15%的空间。但用户刚刚提出了一个新的子任务，预计还需要10轮才能完成。
你面临一个选择：
显然，第三个选择是最优的。但"智能压缩"怎么做？压缩什么？保留什么？用什么来压缩？这些问题就是本节的核心内容。
压缩的核心思想很简单：用更少的token表达相同的信息量。
原始消息（500 tokens）：
"根据查询结果，2026年第二季度的营收为1.2亿元，同比增长15%。其中产品A贡献了60%的营收，产品B贡献了30%，其他产品线贡献了10%。毛利率为45%，比第一季度提升了3个百分点。运营费用为3000万元，其中研发费用占40%，销售费用占35%，管理费用占25%..."
压缩后（80 tokens）：
"[Q2财报摘要] 营收1.2亿(+15%)，毛利率45%(+3pp)，运营费用3000万"
信息密度从"1倍"提升到了"6倍"。这就是压缩的价值。
不同场景需要不同程度的压缩：
| 压缩级别 | 压缩方法 | 保留内容特点 | 压缩比 |
|---|---|---|---|
| 第1级：轻量压缩（提取关键句） | 从长文本中提取核心句子 | 保留大部分细节 | 2-3倍 |
| 第2级：中度压缩（结构化摘要） | 将对话转换为结构化的信息列表 | 保留关键数据和结论 | 5-10倍 |
| 第3级：深度压缩（语义摘要） | 使用LLM生成摘要 | 只保留"对后续推理必要"的信息 | 10-50倍 |
原则：先尝试轻量压缩，不行再中度，最后才用深度压缩。因为压缩级别越高，信息丢失的风险越大，成本也越高。
轻量压缩的目标是：从冗长的消息中提取"信息量最大"的句子。
import re
from typing import List, Tuple
class LightweightCompressor:
"""轻量压缩器：提取关键句"""
def __init__(self, max_sentences: int = 3):
self.max_sentences = max_sentences
def compress(self, text: str) -> str:
"""将长文本压缩为关键句"""
if len(text) < 200:
return text # 短文本不需要压缩
sentences = self._split_sentences(text)
if len(sentences) <= self.max_sentences:
return text
# 对每个句子评分
scored = []
for sentence in sentences:
score = self._score_sentence(sentence, text)
scored.append((sentence, score))
# 按分数排序，保留最高分的句子
scored.sort(key=lambda x: x[1], reverse=True)
key_sentences = [s for s, _ in scored[:self.max_sentences]]
# 按原文顺序排列
key_sentences.sort(key=lambda s: text.index(s))
result = " ".join(key_sentences)
return result + f" ...[已压缩：原{len(text)}字符，保留{len(result)}字符]"
def _split_sentences(self, text: str) -> List[str]:
"""将文本分割为句子"""
# 按句号、分号、换行分割
sentences = re.split(r'[。；;\n]+', text)
return [s.strip() for s in sentences if len(s.strip()) > 10]
def _score_sentence(self, sentence: str, full_text: str) -> float:
"""评估句子的重要性"""
score = 0.0
# 包含数字：可能包含关键数据
if re.search(r'\d', sentence):
score += 2.0
# 包含百分比
if '%' in sentence or '％' in sentence:
score += 2.0
# 包含关键词（结论性词汇）
conclusion_keywords = ["因此", "所以", "总之", "结论", "结果", "最终",
"关键", "重要", "注意", "但是", "然而", "相比",
"增长", "下降", "提升", "减少", "超过", "低于"]
for keyword in conclusion_keywords:
if keyword in sentence:
score += 1.5
break
# 句子长度适中（太短信息量少，太长难理解）
length = len(sentence)
if 20 <= length <= 100:
score += 1.0
elif length > 200:
score -= 0.5
# 在文本中出现的位置（首段和尾段通常更重要）
position = full_text.index(sentence) / len(full_text)
if position < 0.2 or position > 0.8:
score += 1.0
return score
轻量压缩的适用场景：工具返回的长结果、需要保留但不需要完整细节的信息。
中度压缩将对话历史转换为结构化的信息列表。它比提取关键句更进一步——不只是选择保留什么句子，而是重新组织信息。
class StructuralCompressor:
"""中度压缩器：生成结构化摘要"""
def compress_conversation(self, messages: List[Dict[str, Any]]) -> str:
"""将对话历史压缩为结构化摘要"""
summary_parts = []
# 提取用户的目标和需求
user_goals = self._extract_user_goals(messages)
if user_goals:
summary_parts.append(f"用户目标：{'；'.join(user_goals)}")
# 提取已确认的事实
confirmed_facts = self._extract_confirmed_facts(messages)
if confirmed_facts:
summary_parts.append(f"已确认信息：{'；'.join(confirmed_facts)}")
# 提取工具执行结果的关键数据
tool_results = self._extract_tool_results(messages)
if tool_results:
summary_parts.append(f"工具执行结果：{'；'.join(tool_results)}")
# 提取当前任务状态
task_status = self._extract_task_status(messages)
if task_status:
summary_parts.append(f"任务状态：{task_status}")
# 提取未解决的问题
open_questions = self._extract_open_questions(messages)
if open_questions:
summary_parts.append(f"待解决问题：{'；'.join(open_questions)}")
if not summary_parts:
return ""
return "[对话摘要]\n" + "\n".join(summary_parts)
def _extract_user_goals(self, messages: List[Dict[str, Any]]) -> List[str]:
"""提取用户的目标"""
goals = []
for msg in messages:
if msg.get("role") == "user":
content = msg.get("content", "")
# 检测目标性表达
goal_patterns = [
r"帮我(.{2,50})",
r"我想(.{2,50})",
r"需要(.{2,50})",
r"查一下(.{2,50})",
]
for pattern in goal_patterns:
match = re.search(pattern, content)
if match:
goals.append(match.group(0))
break
return goals[-3:] if len(goals) > 3 else goals # 最多保留最近3个目标
def _extract_confirmed_facts(self, messages: List[Dict[str, Any]]) -> List[str]:
"""提取已确认的事实"""
facts = []
for msg in messages:
if msg.get("role") == "tool":
content = msg.get("content", "")
# 提取数字和关键信息
if "成功" in content or "已" in content:
facts.append(content[:80])
return facts[-5:] if len(facts) > 5 else facts
def _extract_tool_results(self, messages: List[Dict[str, Any]]) -> List[str]:
"""提取工具执行结果的关键数据"""
results = []
for msg in messages:
if msg.get("role") == "tool":
content = msg.get("content", "")
# 提取包含数字的行
lines = content.split("\n")
for line in lines:
if re.search(r'\d', line) and len(line) < 100:
results.append(line.strip())
break
return results[-5:] if len(results) > 5 else results
def _extract_task_status(self, messages: List[Dict[str, Any]]) -> str:
"""提取当前任务状态"""
# 找最后一条助手消息
for msg in reversed(messages):
if msg.get("role") == "assistant":
content = msg.get("content", "")
if len(content) > 20:
return content[:100] + "..."
return ""
def _extract_open_questions(self, messages: List[Dict[str, Any]]) -> List[str]:
"""提取未解决的问题"""
questions = []
for msg in messages:
if msg.get("role") == "assistant":
content = msg.get("content", "")
if "?" in content or "？" in content:
# 提取问题
q_pattern = r'[^。；\n]*[?？]'
matches = re.findall(q_pattern, content)
questions.extend(matches)
return questions[-3:] if len(questions) > 3 else questions
中度压缩的适用场景：对话历史超过窗口的60%时，将较早的对话压缩为结构化摘要。
深度压缩使用LLM来生成摘要。这是质量最高但成本也最高的压缩方式。
class LLMCompressor:
"""使用LLM生成高质量摘要"""
def __init__(self, client, model: str = "gpt-4o-mini"):
self.client = client
self.model = model
def compress(self, messages: List[Dict[str, Any]],
max_summary_tokens: int = 500) -> str:
"""使用LLM将对话历史压缩为摘要"""
if not messages:
return ""
# 构建压缩提示词
prompt = (
"你是一个对话摘要助手。请将以下对话历史压缩为一段简短的摘要。\n"
"摘要要求：\n"
"1. 保留用户的核心目标和需求\n"
"2. 保留已确认的关键事实和数据\n"
"3. 保留工具执行的重要结果\n"
"4. 保留未解决的问题和待确认项\n"
"5. 忽略寒暄、重复确认、无信息量的内容\n"
f"6. 摘要控制在{max_summary_tokens}字以内\n\n"
"对话历史：\n"
)
for msg in messages:
role = msg.get("role", "unknown")
content = msg.get("content", "")
# 截断长消息
if len(content) > 500:
content = content[:500] + "..."
prompt += f"[{role}] {content}\n"
prompt += "\n摘要："
try:
response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": prompt}],
max_tokens=max_summary_tokens,
temperature=0.0,
)
summary = response.choices[0].message.content
return f"[历史摘要] {summary}"
except Exception as e:
print(f"LLM摘要生成失败：{e}")
# 降级到结构化压缩
fallback = StructuralCompressor()
return fallback.compress_conversation(messages)
深度压缩的适用场景：上下文窗口即将爆满（>85%使用率），或者对话历史中有大量冗余内容时。
压缩不是随时进行的。它消耗资源（尤其是LLM压缩），不应该每轮都做。一个合理的触发策略：
class CompressionScheduler:
"""压缩调度器：决定何时压缩、压缩什么"""
def __init__(self, working_memory,
llm_compressor: Optional[LLMCompressor] = None):
self.working_memory = working_memory
self.llm_compressor = llm_compressor
self.lightweight = LightweightCompressor()
self.structural = StructuralCompressor()
# 阈值配置
self.light_threshold = 0.6 # 60%使用率触发轻量压缩
self.medium_threshold = 0.75 # 75%使用率触发中度压缩
self.deep_threshold = 0.88 # 88%使用率触发深度压缩
def check_and_compress(self):
"""检查窗口使用率，必要时执行压缩"""
stats = self.working_memory.get_stats()
utilization = stats["total_estimated_tokens"] / stats["window_capacity"]
if utilization < self.light_threshold:
return # 空间充足，不需要压缩
if utilization < self.medium_threshold:
# 轻量压缩：压缩最早的长消息
self._compress_oldest_messages(level="light")
return
if utilization < self.deep_threshold:
# 中度压缩：结构化摘要最早的30%消息
self._compress_oldest_messages(level="medium")
return
# 深度压缩：使用LLM摘要最早的50%消息
self._compress_oldest_messages(level="deep")
def _compress_oldest_messages(self, level: str,
proportion: float = 0.3):
"""压缩最早的一部分消息"""
messages = self.working_memory.messages
if not messages:
return
n_to_compress = max(1, int(len(messages) * proportion))
old_messages = messages[:n_to_compress]
# 选择压缩器
if level == "light":
# 对每条长消息做轻量压缩
for i, msg in enumerate(old_messages):
if len(msg.content) > 200:
msg.content = self.lightweight.compress(msg.content)
msg.token_estimate = len(msg.content) // 2
elif level == "medium":
# 结构化摘要
summary = self.structural.compress_conversation(
[m.to_api_format() for m in old_messages]
)
self._replace_with_summary(old_messages, summary)
elif level == "deep":
# LLM摘要
if self.llm_compressor:
summary = self.llm_compressor.compress(
[m.to_api_format() for m in old_messages]
)
self._replace_with_summary(old_messages, summary)
else:
# 降级
summary = self.structural.compress_conversation(
[m.to_api_format() for m in old_messages]
)
self._replace_with_summary(old_messages, summary)
def _replace_with_summary(self, old_messages, summary: str):
"""用摘要替换一组消息"""
if not summary:
return
from datetime import datetime
# 创建摘要消息
summary_msg = ManagedMessage(
role="system",
content=summary,
importance=MessageImportance.HIGH,
token_estimate=len(summary) // 2,
timestamp=datetime.now().timestamp(),
)
# 替换
indices_to_replace = [self.working_memory.messages.index(m)
for m in old_messages]
if not indices_to_replace:
return
first_idx = min(indices_to_replace)
last_idx = max(indices_to_replace)
self.working_memory.messages = (
self.working_memory.messages[:first_idx] +
[summary_msg] +
self.working_memory.messages[last_idx + 1:]
)
在5.1节中提到过：工具调用和工具结果是成对的。压缩时如果只保留其中一个，LLM会看到"悬空"的信息。
处理策略：压缩时将工具调用和结果合并为一条记录：
def _merge_tool_pairs(self, messages: List[ManagedMessage]) -> List[ManagedMessage]:
"""将工具调用和结果合并"""
merged = []
i = 0
while i < len(messages):
msg = messages[i]
# 如果当前消息是工具调用，找到对应的结果
if msg.role == "assistant" and "tool_calls" in msg.content:
# 查找后续的tool消息
tool_results = []
j = i + 1
while j < len(messages) and messages[j].role == "tool":
tool_results.append(messages[j])
j += 1
if tool_results:
# 合并为一条记录
merged_content = (
f"[工具调用] {msg.content[:100]}\n"
f"[工具结果] {tool_results[0].content[:200]}"
)
merged_msg = ManagedMessage(
role="system",
content=merged_content,
importance=MessageImportance.HIGH,
token_estimate=len(merged_content) // 2,
timestamp=msg.timestamp,
)
merged.append(merged_msg)
i = j
continue
merged.append(msg)
i += 1
return merged
压缩后的摘要是否保留了关键信息？这是一个需要验证的问题。
def validate_compression(original_messages: List[Dict[str, Any]],
summary: str) -> Dict[str, Any]:
"""验证压缩质量"""
# 提取原始消息中的关键数字
original_numbers = set()
for msg in original_messages:
content = msg.get("content", "")
numbers = re.findall(r'\d+(?:\.\d+)?%?', content)
original_numbers.update(numbers)
# 检查摘要中是否保留了这些数字
preserved_numbers = set()
for num in original_numbers:
if num in summary:
preserved_numbers.add(num)
# 计算保留率
retention_rate = len(preserved_numbers) / len(original_numbers) if original_numbers else 1.0
# 检查关键结论性词汇
key_terms = ["完成", "成功", "失败", "错误", "确认", "取消", "增长", "下降"]
preserved_terms = [t for t in key_terms if t in summary]
return {
"numeric_retention_rate": f"{retention_rate * 100:.1f}%",
"preserved_numbers": sorted(preserved_numbers)[:10],
"lost_numbers": sorted(original_numbers - preserved_numbers)[:10],
"preserved_key_terms": preserved_terms,
"summary_length": len(summary),
"original_length": sum(len(m.get("content", "")) for m in original_messages),
"compression_ratio": f"{sum(len(m.get('content', '')) for m in original_messages) / max(len(summary), 1):.1f}x",
}
验证的核心指标：
如果验证发现关键信息丢失严重，需要调整压缩策略（降低压缩级别、修改摘要提示词、增加摘要长度）。
【上下文状态】
窗口使用率：82%
消息总数：45条
总token估算：16400
窗口容量：20000
【压缩调度器触发】
使用率82% > 75%阈值 → 执行中度压缩
【压缩过程】
1. 选择最早的30%消息（14条）
2. 将工具调用和结果合并（14条→7条合并记录）
3. 对合并后的记录生成结构化摘要
4. 摘要生成：约300 tokens
5. 替换原14条消息（原约6000 tokens）
【压缩结果】
窗口使用率：82% → 54%
消息总数：45 → 32条
总token估算：16400 → 10700
节省空间：5700 tokens（35%）
【验证】
数字保留率：92%
压缩比：20倍
丢失关键信息：无（只有冗余的确认消息被丢弃）
记忆压缩与摘要策略是上下文窗口管理的最后一道防线。本节讨论了：
核心原则：压缩是有损的——总会丢失一些信息。好的压缩策略的目标不是"什么都不丢"，而是"只丢不重要的，保留关键的"。压缩的级别选择取决于空间压力和任务对信息完整性的要求。
在下一节中，我们将通过一个完整的实战案例，将短期记忆、长期记忆、会话状态和压缩策略整合到一个可运行的Agent中。
评论专区
评论加载中...登录后即可发表评论