摘要
上下文压缩是处理长文本的核心技术，目的是在有限的上下文窗口内塞入更多信息。这篇为零基础读者讲解：什么是上下文压缩、为什么需要压缩、三大压缩方法（摘要、抽取、混合）、具体实现代码、以及生产环境中的最佳实践。看完全篇，你就能写出高效的上下文压缩系统了。
先理解一个问题：上下文为什么需要”压缩”？
想象你搬家
你要把一整个房间的东西装进一个行李箱：
房间物品：
- 衣服：50件
- 书籍：30本
- 电子产品：10个
- 生活用品：20个
总共：110件物品
行李箱容量：30件物品
怎么办？
→ 压缩！把同类合并、丢弃不重要的、只带必需品
AI的”搬家”问题
你的文档：
- 100页PDF
- 包含：正文、表格、图片注释、引用、脚注
- 总计：50000 tokens
AI的行李箱（上下文窗口）：
- GPT-4：8192 tokens
- Claude 2：200000 tokens
- 各种限制...
差距：50000 vs 8192 ❌ 装不下！
解决方案：压缩！
一、上下文压缩的三种核心方法
方法一览
| 方法 | 原理 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|---|
| 摘要压缩 | 理解内容后重新组织 | 信息密度高 | 可能丢失细节 | 需要理解全文的场景 |
| 抽取压缩 | 直接提取关键句子/段落 | 保留原始表达 | 可能缺乏连贯性 | 检索类场景 |
| 混合压缩 | 先抽取后摘要 | 平衡密度和质量 | 实现复杂 | 通用场景 |
三种方法的可视化对比
原文（100句话）：
[句1][句2][句3][句4][句5]...[句100]
┌─────────────────────────────────────────────────────────────┐
│ 方法1：摘要压缩 │
│ 输出：1段精炼的摘要 │
│ │
│ [总结1-100的核心观点，重新组织，表达精炼] │
└─────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────┐
│ 方法2：抽取压缩 │
│ 输出：10句关键句子 │
│ │
│ [句3][句7][句12][句25][句38][句52][句67][句81][句89][句95] │
└─────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────┐
│ 方法3：混合压缩 │
│ 输出：主题分段摘要 │
│ │
│ 主题A：[抽取关键句] + [一句话摘要] │
│ 主题B：[抽取关键句] + [一句话摘要] │
│ 主题C：[抽取关键句] + [一句话摘要] │
└─────────────────────────────────────────────────────────────┘
二、方法1：摘要压缩
核心思想
摘要压缩就像让AI读完全文后写一篇”观后感”：
输入：一篇5000字的文章
处理：AI理解文章核心内容
输出：300字的摘要
保留：核心观点、关键数据、重要结论
删除：例子、细节、重复表达、冗余描述
实现代码
from typing import List, Optional
from dataclasses import dataclass
@dataclass
class CompressionResult:
"""压缩结果"""
original_tokens: int
compressed_tokens: int
compression_ratio: float
content: str
metadata: dict
class SummaryCompressor:
"""
摘要压缩器
使用LLM生成压缩摘要
"""
def __init__(
self,
llm_client,
target_ratio: float = 0.2, # 压缩到20%
style: str = "concise" # concise | detailed | bullet
):
self.llm = llm_client
self.target_ratio = target_ratio
self.style = style
def compress(
self,
text: str,
preserve_topics: List[str] = None,
focus_question: str = None
) -> CompressionResult:
"""
压缩文本
Args:
text: 要压缩的文本
preserve_topics: 必须保留的主题列表
focus_question: 如果有，优先保留与此问题相关的内容
"""
original_tokens = self._estimate_tokens(text)
# 构建压缩提示
prompt = self._build_compression_prompt(
text,
preserve_topics,
focus_question
)
# 生成压缩文本
compressed = self.llm.generate(prompt)
compressed_tokens = self._estimate_tokens(compressed)
return CompressionResult(
original_tokens=original_tokens,
compressed_tokens=compressed_tokens,
compression_ratio=compressed_tokens / original_tokens,
content=compressed,
metadata={
'style': self.style,
'preserve_topics': preserve_topics,
'focus_question': focus_question
}
)
def _build_compression_prompt(
self,
text: str,
preserve_topics: List[str],
focus_question: str
) -> str:
"""构建压缩提示"""
topic_constraint = ""
if preserve_topics:
topic_constraint = f"\n【必须保留的主题】：{', '.join(preserve_topics)}"
question_constraint = ""
if focus_question:
question_constraint = f"\n【重点关注问题】：{focus_question}\n请优先保留与此问题相关的内容。"
style_instruction = {
"concise": "使用简洁的语言，直接陈述核心内容，不要铺垫。",
"detailed": "保留必要的细节和例子，但避免重复。",
"bullet": "使用要点列表，每点一句话，清晰列出关键信息。"
}[self.style]
return f"""请压缩以下文本，保留核心信息，删除冗余内容。
{topic_constraint}
{question_constraint}
压缩要求：
1. {style_instruction}
2. 保持原文的事实准确性
3. 保留关键数据和结论
4. 删除重复表达和不必要的例子
5. 最终长度约为原文的{self.target_ratio*100}%
原文：
{text}
压缩后："""
@staticmethod
def _estimate_tokens(text: str) -> int:
"""估算token数"""
chinese = sum(1 for c in text if '\u4e00' <= c <= '\u9fff')
english = len(text.split()) - chinese
return int(chinese * 0.5 + english * 0.25)
# 使用示例
def demo_summary_compression():
"""演示摘要压缩"""
compressor = SummaryCompressor(
llm_client=llm,
target_ratio=0.15, # 压缩到15%
style="concise"
)
# 模拟长文本
long_text = """
今天我们开会讨论了公司的发展战略。会议于上午9点开始，持续到下午5点。
会议第一部分由张总主持，主要讨论了去年的业绩情况。去年我们的营收达到了1000万，
同比增长30%。其中核心产品A贡献了60%的营收，B产品贡献了25%，其他产品贡献15%。
会议第二部分讨论了今年的目标。经过热烈讨论，我们确定了以下目标：
1. 营收目标：1500万，同比增长50%
2. 用户目标：新增100万注册用户
3. 产品目标：推出3款新产品
4. 技术目标：完成核心系统重构
会议第三部分讨论了具体执行方案。技术部需要在Q1完成系统重构，
产品部需要在Q2推出第一款新产品，市场部需要加大推广力度。
会议还讨论了一些其他事项，包括办公室装修、团队建设活动等。
下次会议定在两周后继续讨论执行细节。
""" * 5 # 放大文本
result = compressor.compress(
long_text,
preserve_topics=["目标", "数据", "决策"]
)
print(f"原始长度：{result.original_tokens} tokens")
print(f"压缩后：{result.compressed_tokens} tokens")
print(f"压缩比：{result.compression_ratio:.1%}")
print(f"\n压缩结果：\n{result.content}")进阶：多级摘要
class HierarchicalSummarizer:
"""
多级摘要器
逐级压缩：全文 → 分段摘要 → 最终摘要
"""
def __init__(self, llm_client):
self.llm = llm_client
def summarize(
self,
text: str,
num_levels: int = 3,
target_tokens: int = 500
) -> dict:
"""
多级摘要
Args:
text: 原始文本
num_levels: 摘要级别数
target_tokens: 最终目标token数
"""
# 按段落分割
paragraphs = [p.strip() for p in text.split('\n\n') if p.strip()]
# 第一级：每个段落生成一句话摘要
level1_summaries = self._level1_summaries(paragraphs)
# 第二级：将段落摘要分组，每组生成一个摘要
if num_levels >= 2:
level2_summaries = self._level2_summaries(level1_summaries)
# 第三级：生成最终摘要
final_summary = self._final_summary(
level2_summaries if num_levels >= 2 else level1_summaries,
target_tokens
)
return {
'level1': level1_summaries,
'level2': level2_summaries if num_levels >= 2 else None,
'final': final_summary,
'full_summary': '\n\n'.join([
level2_summaries[0] if num_levels >= 2
else level1_summaries[0]
])
}
def _level1_summaries(self, paragraphs: List[str]) -> List[str]:
"""第一级：段落摘要"""
summaries = []
for para in paragraphs:
prompt = f"""用一句话概括以下段落的中心思想：
{para}
一句话概括："""
summary = self.llm.generate(prompt).strip()
summaries.append(summary)
return summaries
def _level2_summaries(self, level1: List[str]) -> List[str]:
"""第二级：分组摘要"""
# 每3-5个段落为一组
group_size = 4
groups = [
level1[i:i+group_size]
for i in range(0, len(level1), group_size)
]
summaries = []
for i, group in enumerate(groups):
combined = '\n'.join([f"- {s}" for s in group])
prompt = f"""基于以下要点，用2-3句话总结核心内容：
{combined}
总结："""
summary = self.llm.generate(prompt).strip()
summaries.append(summary)
return summaries
def _final_summary(
self,
content: List[str],
target_tokens: int
) -> str:
"""最终摘要"""
combined = '\n\n'.join(content)
prompt = f"""将以下内容整合成一个简洁的摘要（目标{target_tokens}tokens）：
{combined}
整合摘要："""
return self.llm.generate(prompt).strip()三、方法2：抽取压缩
核心思想
抽取压缩就像”划重点”——不做改写，直接提取原文中最重要的部分：
输入：100句话
处理：判断每句话的重要性
输出：最重要的10句话（按原文顺序）
保留：原文表达
删除：次要内容、过渡句、重复表达
实现代码
class ExtractionCompressor:
"""
抽取压缩器
直接从原文中提取关键内容
"""
def __init__(
self,
llm_client,
max_tokens: int = 4000,
min_sentences: int = 3
):
self.llm = llm_client
self.max_tokens = max_tokens
self.min_sentences = min_sentences
def compress(
self,
text: str,
query: str = None,
mode: str = "automatic" # automatic | query-focused | topic-based
) -> CompressionResult:
"""
抽取压缩
Args:
text: 要压缩的文本
query: 可选的查询，用于相关性排序
mode: 压缩模式
"""
original_tokens = self._estimate_tokens(text)
# 分割句子
sentences = self._split_sentences(text)
if mode == "automatic":
# 自动模式：评估每句话的重要性
important = self._score_and_select(sentences, query)
elif mode == "query-focused":
# 查询聚焦模式：优先选择与查询相关的内容
important = self._query_focused_select(sentences, query)
elif mode == "topic-based":
# 主题模式：按主题分组，每个主题选一句
important = self._topic_based_select(sentences)
# 按原文顺序排列
important_ordered = self._order_by_original(
important,
sentences
)
# 构建压缩文本
compressed = '\n\n'.join(important_ordered)
compressed_tokens = self._estimate_tokens(compressed)
# 如果还是太长，继续精简
while compressed_tokens > self.max_tokens and len(important_ordered) > self.min_sentences:
important_ordered = important_ordered[:-2] # 删除最不重要的2句
compressed = '\n\n'.join(important_ordered)
compressed_tokens = self._estimate_tokens(compressed)
return CompressionResult(
original_tokens=original_tokens,
compressed_tokens=compressed_tokens,
compression_ratio=compressed_tokens / original_tokens,
content=compressed,
metadata={
'mode': mode,
'query': query,
'selected_count': len(important_ordered),
'original_count': len(sentences)
}
)
def _split_sentences(self, text: str) -> List[dict]:
"""分割句子"""
# 简单分割（实际应该用更好的句子分割器）
import re
raw_sentences = re.split(r'[。！？\n]+', text)
sentences = []
for i, sent in enumerate(raw_sentences):
sent = sent.strip()
if len(sent) > 10: # 过滤太短的
sentences.append({
'text': sent,
'index': i,
'tokens': self._estimate_tokens(sent)
})
return sentences
def _score_and_select(
self,
sentences: List[dict],
query: str
) -> List[str]:
"""评分并选择"""
# 评估每句话的重要性
scored = []
for sent in sentences:
prompt = f"""评估以下句子在文档中的重要程度（0-10分）：
句子：{sent['text']}
重要程度评分标准：
- 10分：包含关键结论、重要数据、核心观点
- 7-9分：包含重要细节或论据
- 4-6分：普通内容
- 1-3分：过渡句、例子、不重要的细节
只输出数字，不要其他内容。"""
score = self.llm.generate(prompt).strip()
try:
score = float(score)
except:
score = 5.0 # 默认分数
scored.append({
'text': sent['text'],
'score': score,
'tokens': sent['tokens']
})
# 按分数排序，选择高分句子
scored.sort(key=lambda x: x['score'], reverse=True)
selected = []
total_tokens = 0
for item in scored:
if total_tokens + item['tokens'] > self.max_tokens:
break
selected.append(item['text'])
total_tokens += item['tokens']
return selected
def _query_focused_select(
self,
sentences: List[dict],
query: str
) -> List[str]:
"""查询聚焦选择"""
if not query:
return self._score_and_select(sentences, None)
scored = []
for sent in sentences:
# 计算与查询的相关性
relevance_prompt = f"""评估以下句子与查询的相关性（0-10分）：
查询：{query}
句子：{sent['text']}
相关性评分标准：
- 10分：直接回答或解释查询
- 7-9分：包含与查询相关的重要内容
- 4-6分：部分相关
- 1-3分：不太相关
只输出数字，不要其他内容。"""
score = self.llm.generate(relevance_prompt).strip()
try:
score = float(score)
except:
score = 5.0
scored.append({
'text': sent['text'],
'score': score,
'tokens': sent['tokens']
})
# 选择最相关的
scored.sort(key=lambda x: x['score'], reverse=True)
selected = []
total_tokens = 0
for item in scored:
if total_tokens + item['tokens'] > self.max_tokens:
break
selected.append(item['text'])
total_tokens += item['tokens']
return selected
def _topic_based_select(self, sentences: List[dict]) -> List[str]:
"""基于主题的选择"""
# 识别主题
topic_prompt = f"""将以下句子按主题分组，每组一句话总结：
{chr(10).join([f"{i+1}. {s['text']}" for i, s in enumerate(sentences)])}
分组要求：
1. 识别出2-5个主要主题
2. 每个主题选择1-2句最代表性的句子
3. 输出格式：主题1：[选中的句子]
选中的句子："""
result = self.llm.generate(topic_prompt).strip()
# 解析结果，提取句子
# （简化处理，实际应该更robust）
selected = []
for sent in sentences:
if sent['text'] in result:
selected.append(sent['text'])
return selected if selected else [s['text'] for s in sentences[:5]]
def _order_by_original(
self,
selected: List[str],
sentences: List[dict]
) -> List[str]:
"""按原文顺序排列"""
# 创建索引映射
text_to_idx = {s['text']: s['index'] for s in sentences}
# 按原文顺序排序
selected_with_idx = [
(text, text_to_idx.get(text, 0))
for text in selected
]
selected_with_idx.sort(key=lambda x: x[1])
return [text for text, _ in selected_with_idx]
@staticmethod
def _estimate_tokens(text: str) -> int:
chinese = sum(1 for c in text if '\u4e00' <= c <= '\u9fff')
english = len(text.split()) - chinese
return int(chinese * 0.5 + english * 0.25)四、方法3：混合压缩
核心思想
混合压缩结合了摘要和抽取的优点：
第一步：抽取 - 选出与主题相关的段落
第二步：摘要 - 对每个段落进行精炼
效果：保留关键细节 + 表达精炼
实现代码
class HybridCompressor:
"""
混合压缩器
先抽取关键内容，再进行摘要精炼
"""
def __init__(
self,
llm_client,
max_tokens: int = 4000,
extract_ratio: float = 0.3,
compress_ratio: float = 0.5
):
self.llm = llm_client
self.max_tokens = max_tokens
self.extract_ratio = extract_ratio # 抽取比例
self.compress_ratio = compress_ratio # 摘要压缩比例
def compress(
self,
text: str,
query: str = None,
preserve_structure: bool = True
) -> CompressionResult:
"""
混合压缩
Args:
text: 原始文本
query: 可选的查询
preserve_structure: 是否保留原文结构
"""
original_tokens = self._estimate_tokens(text)
# 第一步：抽取关键内容
extracted = self._extract_key_content(text, query)
# 第二步：对抽取的内容进行摘要
compressed = self._summarize_content(extracted, preserve_structure)
# 调整到目标长度
while self._estimate_tokens(compressed) > self.max_tokens:
compressed = self._further_compress(compressed)
compressed_tokens = self._estimate_tokens(compressed)
return CompressionResult(
original_tokens=original_tokens,
compressed_tokens=compressed_tokens,
compression_ratio=compressed_tokens / original_tokens,
content=compressed,
metadata={
'method': 'hybrid',
'query': query,
'preserve_structure': preserve_structure
}
)
def _extract_key_content(
self,
text: str,
query: str
) -> str:
"""抽取关键内容"""
# 按段落分割
paragraphs = [p.strip() for p in text.split('\n\n') if p.strip()]
if not paragraphs:
return text
# 评估每段与查询的相关性
scored_paras = []
for para in paragraphs:
if query:
prompt = f"""评估以下段落与查询的相关性（0-10分）：
查询：{query}
段落：{para}
只输出数字："""
else:
prompt = f"""评估以下段落的重要程度（0-10分）：
段落：{para}
只输出数字："""
score = self.llm.generate(prompt).strip()
try:
score = float(score)
except:
score = 5.0
scored_paras.append({
'text': para,
'score': score,
'tokens': self._estimate_tokens(para)
})
# 选择最相关的内容
scored_paras.sort(key=lambda x: x['score'], reverse=True)
target_tokens = original_tokens * self.extract_ratio if 'original_tokens' in dir() else self.max_tokens * 0.5
selected = []
total_tokens = 0
for para in scored_paras:
if total_tokens + para['tokens'] > target_tokens:
continue
selected.append(para)
total_tokens += para['tokens']
# 按原文顺序排列
selected.sort(key=lambda x: paragraphs.index(x['text']))
return '\n\n'.join([p['text'] for p in selected])
def _summarize_content(
self,
content: str,
preserve_structure: bool
) -> str:
"""摘要精炼"""
if preserve_structure:
# 保留结构：每个段落单独摘要
paragraphs = [p.strip() for p in content.split('\n\n') if p.strip()]
summarized = []
for para in paragraphs:
prompt = f"""精炼以下段落，保留核心信息：
{para}
要求：
- 删除重复和冗余
- 保留关键数据
- 一句话能说清楚的，不要说两句
精炼后："""
summary = self.llm.generate(prompt).strip()
summarized.append(summary)
return '\n\n'.join(summarized)
else:
# 不保留结构：整体摘要
prompt = f"""精炼以下内容，保留核心信息：
{content}
要求：
- 删除重复和冗余
- 保留关键数据
- 表达精炼
精炼后："""
return self.llm.generate(prompt).strip()
def _further_compress(self, content: str) -> str:
"""进一步压缩"""
prompt = f"""进一步压缩以下内容，保留最核心的信息：
{content}
目标长度：{int(self._estimate_tokens(content) * 0.7)} tokens
压缩后："""
return self.llm.generate(prompt).strip()
@staticmethod
def _estimate_tokens(text: str) -> int:
chinese = sum(1 for c in text if '\u4e00' <= c <= '\u9fff')
english = len(text.split()) - chinese
return int(chinese * 0.5 + english * 0.25)五、生产级压缩系统
自适应压缩管道
class AdaptiveCompressionPipeline:
"""
自适应压缩管道
根据文本长度和内容自动选择压缩策略
"""
def __init__(
self,
llm_client,
max_context_tokens: int = 8000,
quality_threshold: float = 0.7
):
self.llm = llm_client
self.max_tokens = max_context_tokens
self.quality_threshold = quality_threshold
# 各类型压缩器
self.summary_compressor = SummaryCompressor(llm_client, 0.2)
self.extraction_compressor = ExtractionCompressor(llm_client, max_tokens)
self.hybrid_compressor = HybridCompressor(llm_client, max_tokens)
def compress(
self,
text: str,
query: str = None,
force_method: str = None
) -> CompressionResult:
"""
自适应压缩
Args:
text: 要压缩的文本
query: 可选的查询
force_method: 强制使用的方法（summary/extraction/hybrid）
"""
original_tokens = self._estimate_tokens(text)
# 判断是否需要压缩
if original_tokens <= self.max_tokens:
return CompressionResult(
original_tokens=original_tokens,
compressed_tokens=original_tokens,
compression_ratio=1.0,
content=text,
metadata={'method': 'none', 'reason': '不需要压缩'}
)
# 选择压缩方法
method = force_method or self._select_method(
text,
query
)
# 执行压缩
if method == "summary":
result = self.summary_compressor.compress(
text,
focus_question=query
)
elif method == "extraction":
result = self.extraction_compressor.compress(
text,
query=query,
mode="query-focused" if query else "automatic"
)
else: # hybrid
result = self.hybrid_compressor.compress(
text,
query=query
)
# 质量检查
result = self._quality_check(result, query)
return result
def _select_method(self, text: str, query: str) -> str:
"""选择压缩方法"""
# 长文本 + 有查询 → 混合
if query and self._estimate_tokens(text) > 20000:
return "hybrid"
# 有查询 + 需要精确匹配 → 抽取
if query:
return "extraction"
# 长文本 → 摘要
if self._estimate_tokens(text) > 50000:
return "summary"
# 默认 → 混合
return "hybrid"
def _quality_check(
self,
result: CompressionResult,
query: str
) -> CompressionResult:
"""质量检查"""
if not query:
return result
# 检查查询相关内容是否保留
check_prompt = f"""评估压缩后的内容是否保留了回答问题的关键信息：
问题：{query}
压缩后内容：
{result.content}
评估标准：
- 1分：完全丢失关键信息
- 5分：部分保留
- 10分：完整保留
只输出数字："""
quality_score = self.llm.generate(check_prompt).strip()
try:
quality_score = float(quality_score) / 10.0
except:
quality_score = 0.5
# 如果质量太低，尝试换方法
if quality_score < self.quality_threshold:
# 尝试不同的压缩方法
alt_method = "extraction" if result.metadata.get('method') == 'summary' else "summary"
result = self.compress(
result.content[:1000] + "...", # 用原文重新压缩
query=query,
force_method=alt_method
)
result.metadata['retry'] = True
result.metadata['quality_score'] = quality_score
return result
@staticmethod
def _estimate_tokens(text: str) -> int:
chinese = sum(1 for c in text if '\u4e00' <= c <= '\u9fff')
english = len(text.split()) - chinese
return int(chinese * 0.5 + english * 0.25)
# 使用示例
def demo_adaptive_compression():
"""演示自适应压缩"""
pipeline = AdaptiveCompressionPipeline(
llm_client=llm,
max_context_tokens=8000
)
# 模拟不同类型的文档
documents = {
"长论文": "..." * 1000,
"对话记录": "用户：..." * 100,
"技术文档": "..." * 500,
}
for name, doc in documents.items():
result = pipeline.compress(
doc,
query="主要结论是什么？"
)
print(f"\n{name}:")
print(f" 原始：{result.original_tokens} tokens")
print(f" 压缩后：{result.compressed_tokens} tokens")
print(f" 压缩比：{result.compression_ratio:.1%}")
print(f" 方法：{result.metadata.get('method')}")
print(f" 质量：{result.metadata.get('quality_score', 'N/A'):.2f}" if isinstance(result.metadata.get('quality_score'), float) else f" 方法：{result.metadata.get('method')}")六、压缩质量评估
评估维度
class CompressionEvaluator:
"""压缩质量评估器"""
def __init__(self, llm_client):
self.llm = llm_client
def evaluate(
self,
original: str,
compressed: str,
query: str = None
) -> dict:
"""
评估压缩质量
Returns:
dict: 包含各维度评分的字典
"""
return {
'compression_ratio': self._estimate_tokens(compressed) / max(1, self._estimate_tokens(original)),
'faithfulness': self._evaluate_faithfulness(original, compressed),
'coverage': self._evaluate_coverage(original, compressed, query),
'fluency': self._evaluate_fluency(compressed),
'information_density': self._evaluate_density(original, compressed)
}
def _evaluate_faithfulness(
self,
original: str,
compressed: str
) -> float:
"""忠实度评估 - 压缩后是否忠实于原文"""
prompt = f"""评估压缩是否忠实于原文（0-10分）：
原文核心观点：
{self._get_main_points(original)}
压缩后内容：
{compressed}
评分标准：
- 10分：完全忠实，没有歪曲原意
- 5分：基本忠实，有轻微偏差
- 1分：严重歪曲原意
只输出数字："""
score = self.llm.generate(prompt).strip()
try:
return float(score) / 10.0
except:
return 0.5
def _evaluate_coverage(
self,
original: str,
compressed: str,
query: str
) -> float:
"""覆盖度评估 - 关键信息是否保留"""
if not query:
return 0.8 # 没有查询，给默认分数
prompt = f"""评估压缩后内容对回答问题的覆盖程度（0-10分）：
问题：{query}
原文：
{original}
压缩后：
{compressed}
评分标准：
- 10分：完整覆盖，能完全回答问题
- 5分：部分覆盖
- 1分：几乎无法回答问题
只输出数字："""
score = self.llm.generate(prompt).strip()
try:
return float(score) / 10.0
except:
return 0.5
def _evaluate_fluency(self, compressed: str) -> float:
"""流畅度评估"""
prompt = f"""评估压缩后内容的语言流畅度（0-10分）：
{compressed}
评分标准：
- 10分：非常流畅，像人写的
- 5分：基本可读
- 1分：语句不通顺
只输出数字："""
score = self.llm.generate(prompt).strip()
try:
return float(score) / 10.0
except:
return 0.5
def _evaluate_density(
self,
original: str,
compressed: str
) -> float:
"""信息密度评估"""
orig_tokens = self._estimate_tokens(original)
comp_tokens = self._estimate_tokens(compressed)
ratio = comp_tokens / max(1, orig_tokens)
# 密度分数：压缩越狠，密度应该越高
if ratio > 0.8:
return 0.3 # 压缩太少
elif ratio > 0.5:
return 0.7
elif ratio > 0.2:
return 1.0 # 理想范围
else:
return 0.8 # 压缩太狠
@staticmethod
def _estimate_tokens(text: str) -> int:
chinese = sum(1 for c in text if '\u4e00' <= c <= '\u9fff')
english = len(text.split()) - chinese
return int(chinese * 0.5 + english * 0.25)
def _get_main_points(self, text: str, num_points: int = 5) -> str:
"""提取原文主要观点"""
prompt = f"""列出以下文本的{num_points}个核心观点：
{text}
格式：每行一个观点"""
return self.llm.generate(prompt).strip()七、实战案例：构建智能文档助手
class SmartDocumentAssistant:
"""
智能文档助手
使用上下文压缩处理长文档问答
"""
def __init__(
self,
llm_client,
vector_store,
max_context_tokens: int = 8000
):
self.llm = llm_client
self.vector_store = vector_store
self.max_tokens = max_context_tokens
self.compressor = AdaptiveCompressionPipeline(
llm_client,
max_context_tokens
)
def query(
self,
document_id: str,
question: str
) -> str:
"""
文档问答
流程：
1. 检索相关片段
2. 压缩到上下文限制
3. 生成回答
"""
# 1. 检索相关片段
chunks = self.vector_store.search(
query=question,
document_id=document_id,
top_k=20
)
# 2. 合并并压缩
combined = '\n\n'.join([c['content'] for c in chunks])
# 3. 压缩
compressed = self.compressor.compress(
combined,
query=question
)
# 4. 生成回答
prompt = f"""基于以下文档内容回答问题。如果文档中没有答案，说明不知道。
[文档]
{compressed.content}
[/文档]
问题：{question}
回答要求：
1. 如果文档有答案，给出完整回答
2. 引用相关原文
3. 如果文档没有答案，说明"根据提供的文档，无法回答这个问题"
回答："""
answer = self.llm.generate(prompt)
return {
'answer': answer,
'compression_info': {
'original_tokens': compressed.original_tokens,
'compressed_tokens': compressed.compressed_tokens,
'ratio': f"{compressed.compression_ratio:.1%}",
'method': compressed.metadata.get('method')
}
}
# 使用示例
def demo_document_assistant():
"""演示智能文档助手"""
assistant = SmartDocumentAssistant(
llm_client=llm,
vector_store=vector_store,
max_context_tokens=8000
)
# 问答
result = assistant.query(
document_id="doc_001",
question="这篇论文的主要贡献是什么？"
)
print(f"回答：{result['answer']}")
print(f"\n压缩信息：{result['compression_info']}")八、总结
┌─────────────────────────────────────────────────────────────────────┐
│ 上下文压缩技术速查表 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ 🎯 三大方法 │
│ ├─ 摘要压缩：理解后重新组织，信息密度高 │
│ ├─ 抽取压缩：直接提取关键句，保留原文表达 │
│ └─ 混合压缩：先抽后摘，平衡质量 │
│ │
│ 📊 选择指南 │
│ ├─ 理解类任务 → 摘要压缩 │
│ ├─ 检索类任务 → 抽取压缩 │
│ └─ 通用场景 → 混合压缩 │
│ │
│ 💡 质量评估 │
│ ├─ 忠实度：是否忠于原意 │
│ ├─ 覆盖度：关键信息是否保留 │
│ ├─ 流畅度：语言是否通顺 │
│ └─ 密度：信息是否精炼 │
│ │
│ ⚠️ 注意事项 │
│ ├─ 压缩比不宜过低（<10%可能丢失关键信息） │
│ ├─ 重要数据不要压缩 │
│ └─ 结构化内容优先保留格式 │
│ │
└─────────────────────────────────────────────────────────────────────┘
相关主题
- 上下文窗口深度解析 - 理解上下文限制的根本原因
- 滑动窗口技术 - 另一种处理长文本的方法
- RAG上下文优化指南 - 压缩在RAG中的应用
参考文献
- Mohtaj, S., et al. (2024). A Survey on Text Simplification. ACM Computing Surveys.
- Liu, Y., et al. (2024). Query-Focused Text Summarization. arXiv.
- Zhou, Y., et al. (2024). Context Compression in Large Language Models. ACL 2024.