Prompt 压缩技术：让上下文窗口利用率提升 50%

Prompt 压缩
├── 无损压缩
│   ├── 符号化压缩（缩写、代码化）
│   ├── 结构化压缩（JSON→紧凑格式）
│   └── 去冗余压缩（删除重复信息）
├── 有损压缩
│   ├── 语义压缩（LLM 总结）
│   ├── 选择性保留（截断低重要性内容）
│   └── 信息蒸馏（提取关键信息）
└── 混合压缩
    └── 分层压缩策略

class SymbolicCompressor:
    """符号化压缩——用短符号替代长文本"""
    
    SYMBOL_MAP = {
        # 常见指令缩写
        "请分析以下内容并给出": "分析:",
        "请根据以上信息回答": "回答:",
        "以下是相关的背景信息": "背景:",
        "请注意以下重要事项": "注意:",
        
        # 角色缩写
        "你是一个专业的": "角色:",
        "你的核心职责是": "职责:",
        
        # 格式缩写
        "请用Markdown表格格式输出": "→MD表格",
        "请用JSON格式输出": "→JSON",
        "请用列表格式输出": "→列表",
        
        # 常见短语
        "需要注意的是": "⚠",
        "重要提醒": "‼",
        "例如": "如",
        "也就是说": "即",
    }
    
    def compress(self, prompt: str) -> str:
        for full, symbol in self.SYMBOL_MAP.items():
            prompt = prompt.replace(full, symbol)
        return prompt
    
    def decompress_guide(self) -> str:
        """生成符号说明（添加到System Prompt）"""
        guide = "符号说明: "
        for full, symbol in self.SYMBOL_MAP.items():
            guide += f"{symbol}={full[:4]}.. "
        return guide

class StructuralCompressor:
    """结构化压缩——压缩冗余的格式"""
    
    def compress_table(self, markdown_table: str) -> str:
        """压缩 Markdown 表格"""
        lines = markdown_table.strip().split('\n')
        if len(lines) < 3:
            return markdown_table
        
        # 提取表头和数据
        headers = [h.strip() for h in lines[0].split('|')[1:-1]]
        data_rows = []
        for line in lines[2:]:  # 跳过分隔行
            cells = [c.strip() for c in line.split('|')[1:-1]]
            data_rows.append(cells)
        
        # 紧凑格式：用 | 分隔，不用对齐
        compact = '|'.join(headers) + '\n'
        for row in data_rows:
            compact += '|'.join(row) + '\n'
        
        return compact
    
    def compress_json(self, json_str: str) -> str:
        """压缩 JSON"""
        import json
        data = json.loads(json_str)
        return json.dumps(data, ensure_ascii=False, separators=(',', ':'))
    
    def compress_list(self, markdown_list: str) -> str:
        """压缩列表"""
        lines = markdown_list.strip().split('\n')
        items = [l.lstrip('- *').strip() for l in lines if l.strip()]
        return '; '.join(items)

class RedundancyRemover:
    """去冗余压缩"""
    
    def compress(self, prompt: str) -> str:
        # 1. 移除重复段落
        prompt = self._remove_duplicate_paragraphs(prompt)
        
        # 2. 移除重复句子
        prompt = self._remove_duplicate_sentences(prompt)
        
        # 3. 移除空白行
        prompt = self._remove_blank_lines(prompt)
        
        # 4. 合并连续空格
        import re
        prompt = re.sub(r' {2,}', ' ', prompt)
        
        return prompt
    
    def _remove_duplicate_paragraphs(self, text: str) -> str:
        paragraphs = text.split('\n\n')
        seen = set()
        unique = []
        for p in paragraphs:
            normalized = p.strip().lower()
            if normalized and normalized not in seen:
                seen.add(normalized)
                unique.append(p)
        return '\n\n'.join(unique)
    
    def _remove_duplicate_sentences(self, text: str) -> str:
        import re
        sentences = re.split(r'(?<=[。.!?！？])\s+', text)
        seen = set()
        unique = []
        for s in sentences:
            if s.strip() and s.strip() not in seen:
                seen.add(s.strip())
                unique.append(s)
        return ' '.join(unique)

class SemanticCompressor:
    """使用 LLM 进行语义压缩"""
    
    COMPRESSION_PROMPT = """请压缩以下文本，要求：
1. 保留所有关键信息和数据
2. 保留逻辑结构和因果关系
3. 移除冗余描述和过渡语句
4. 用更简洁的表达替代冗长表达
5. 保持事实准确性

原始文本（{original_tokens} tokens）：
{text}

输出压缩后的文本，目标：{target_tokens} tokens以内。"""
    
    def compress(self, text: str, target_ratio: float = 0.5,
                 llm_client=None) -> str:
        original_tokens = self._estimate_tokens(text)
        target_tokens = int(original_tokens * target_ratio)
        
        prompt = self.COMPRESSION_PROMPT.format(
            original_tokens=original_tokens,
            text=text,
            target_tokens=target_tokens
        )
        
        compressed = llm_client.generate(prompt)
        return compressed
    
    def _estimate_tokens(self, text: str) -> int:
        # 粗略估算
        chinese_chars = sum(1 for c in text if '\u4e00' <= c <= '\u9fff')
        other_chars = len(text) - chinese_chars
        return chinese_chars * 2 + other_chars // 4

class SelectiveCompressor:
    """选择性保留——基于重要性的压缩"""
    
    def compress(self, text: str, target_ratio: float = 0.5) -> str:
        # 1. 分割为句子
        sentences = self._split_sentences(text)
        
        # 2. 计算每个句子的重要性分数
        scored = self._score_sentences(sentences)
        
        # 3. 保留高重要性句子
        target_count = int(len(sentences) * target_ratio)
        top_sentences = sorted(scored, key=lambda x: -x[1])[:target_count]
        
        # 4. 按原顺序排列
        top_sentences.sort(key=lambda x: x[2])  # 按原始位置排序
        
        return ' '.join(s[0] for s in top_sentences)
    
    def _score_sentences(self, sentences: list) -> list:
        """使用 TextRank 思想计算句子重要性"""
        # 基于句子间相似度构建图
        n = len(sentences)
        scores = [1.0] * n
        
        for iteration in range(10):  # 迭代计算
            new_scores = []
            for i, sent in enumerate(sentences):
                score = 0.15  # 基础分
                for j, other in enumerate(sentences):
                    if i != j:
                        sim = self._sentence_similarity(sent, other)
                        score += 0.85 * sim * scores[j] / max(
                            sum(self._sentence_similarity(other, s2) 
                                for k, s2 in enumerate(sentences) if k != j),
                            1e-8
                        )
                new_scores.append(score)
            scores = new_scores
        
        return [(sentences[i], scores[i], i) for i in range(n)]
    
    def _sentence_similarity(self, s1: str, s2: str) -> float:
        """计算两个句子的相似度"""
        words1 = set(s1.split())
        words2 = set(s2.split())
        intersection = words1 & words2
        union = words1 | words2
        return len(intersection) / max(len(union), 1)

class InformationDistiller:
    """信息蒸馏——提取关键信息，丢弃细节"""
    
    DISTILL_PROMPT = """从以下文本中提取关键信息，使用紧凑格式输出。

输出格式：
- 主题：[1-2句话]
- 关键事实：[每条1行，最多5条]
- 数据：[数值和单位]
- 结论：[1句话]

文本：
{text}"""
    
    def distill(self, text: str, llm_client) -> str:
        prompt = self.DISTILL_PROMPT.format(text=text)
        return llm_client.generate(prompt)

class LayeredCompressor:
    """分层压缩策略——不同内容用不同压缩方法"""
    
    def __init__(self, llm_client):
        self.llm = llm_client
        self.symbolic = SymbolicCompressor()
        self.structural = StructuralCompressor()
        self.redundancy = RedundancyRemover()
        self.semantic = SemanticCompressor()
        self.selective = SelectiveCompressor()
        self.distiller = InformationDistiller(llm_client)
    
    def compress(self, prompt: str, target_ratio: float = 0.5) -> str:
        """分层压缩"""
        original_tokens = self._estimate_tokens(prompt)
        target_tokens = int(original_tokens * target_ratio)
        
        # Layer 1: 无损压缩（总是执行）
        prompt = self.symbolic.compress(prompt)
        prompt = self.structural.compress_json(prompt)
        prompt = self.redundancy.compress(prompt)
        
        current_tokens = self._estimate_tokens(prompt)
        if current_tokens <= target_tokens:
            return prompt  # 无损压缩已达标
        
        # Layer 2: 内容分类
        sections = self._classify_sections(prompt)
        
        # Layer 3: 按类别压缩
        compressed_sections = []
        for section_type, content in sections:
            if section_type == 'rules':
                # 规则类：仅做符号压缩
                compressed_sections.append(self.symbolic.compress(content))
            elif section_type == 'knowledge':
                # 知识类：语义压缩
                compressed_sections.append(
                    self.semantic.compress(content, 0.4, self.llm)
                )
            elif section_type == 'examples':
                # 示例类：选择性保留
                compressed_sections.append(
                    self.selective.compress(content, 0.6)
                )
            elif section_type == 'context':
                # 上下文类：信息蒸馏
                compressed_sections.append(
                    self.distiller.distill(content, self.llm)
                )
            else:
                compressed_sections.append(content)
        
        result = '\n\n'.join(compressed_sections)
        
        # Layer 4: 如果仍超标，全局压缩
        if self._estimate_tokens(result) > target_tokens:
            result = self.semantic.compress(result, 
                target_tokens / self._estimate_tokens(result), 
                self.llm)
        
        return result
    
    def _classify_sections(self, prompt: str) -> list:
        """将 Prompt 分为不同类型的段落"""
        sections = []
        current_section = ""
        current_type = "other"
        
        for line in prompt.split('\n'):
            if line.startswith('规则') or line.startswith('约束'):
                if current_section:
                    sections.append((current_type, current_section))
                current_section = line + '\n'
                current_type = 'rules'
            elif line.startswith('知识') or line.startswith('背景'):
                if current_section:
                    sections.append((current_type, current_section))
                current_section = line + '\n'
                current_type = 'knowledge'
            elif line.startswith('示例') or line.startswith('例子'):
                if current_section:
                    sections.append((current_type, current_section))
                current_section = line + '\n'
                current_type = 'examples'
            elif line.startswith('上下文') or line.startswith('历史'):
                if current_section:
                    sections.append((current_type, current_section))
                current_section = line + '\n'
                current_type = 'context'
            else:
                current_section += line + '\n'
        
        if current_section:
            sections.append((current_type, current_section))
        
        return sections

class ConversationHistoryCompressor:
    """对话历史压缩——长对话的上下文管理"""
    
    def __init__(self, llm_client, max_history_tokens: int = 4000):
        self.llm = llm_client
        self.max_tokens = max_history_tokens
    
    def compress_history(self, messages: list) -> list:
        """压缩对话历史"""
        total_tokens = sum(self._estimate_tokens(m['content']) for m in messages)
        
        if total_tokens <= self.max_tokens:
            return messages  # 不需要压缩
        
        # 策略：保留最近 N 轮 + 压缩早期对话
        recent_count = min(6, len(messages))  # 保留最近3轮
        recent = messages[-recent_count:]
        old = messages[:-recent_count]
        
        # 压缩早期对话
        summary = self._summarize_conversation(old)
        
        # 构建压缩后的历史
        compressed = [
            {"role": "system", "content": f"对话摘要：{summary}"},
            *recent
        ]
        
        return compressed
    
    def _summarize_conversation(self, messages: list) -> str:
        """总结早期对话"""
        conversation_text = '\n'.join(
            f"{m['role']}: {m['content'][:200]}" for m in messages
        )
        
        prompt = f"""请总结以下对话的关键信息，保留：
1. 用户的核心需求和偏好
2. 已达成的结论和决定
3. 未解决的问题
4. 重要的事实和数据

对话内容：
{conversation_text}

总结（不超过200字）："""
        
        return self.llm.generate(prompt)

class CompressionEvaluator:
    """压缩效果评估器"""
    
    def evaluate(self, original: str, compressed: str,
                 test_cases: list, llm_client) -> dict:
        results = {
            'compression_ratio': len(compressed) / len(original),
            'token_reduction': 1 - self._tokens(compressed) / self._tokens(original),
        }
        
        # 信息保留率评估
        info_retention = self._evaluate_info_retention(original, compressed, llm_client)
        results['info_retention'] = info_retention
        
        # 任务效果评估
        original_scores = []
        compressed_scores = []
        
        for case in test_cases:
            # 使用原始 Prompt
            original_response = llm_client.generate(original + case['input'])
            original_scores.append(self._score(original_response, case['expected']))
            
            # 使用压缩 Prompt
            compressed_response = llm_client.generate(compressed + case['input'])
            compressed_scores.append(self._score(compressed_response, case['expected']))
        
        results['original_accuracy'] = sum(original_scores) / len(original_scores)
        results['compressed_accuracy'] = sum(compressed_scores) / len(compressed_scores)
        results['accuracy_drop'] = results['original_accuracy'] - results['compressed_accuracy']
        
        # 成本节省
        results['cost_saving'] = results['token_reduction']
        
        return results