💡 LLM 成本优化 / Prompt 压缩

🗜️ Prompt 压缩

减少输入 token 数量本身——不改变模型，不改变缓存策略，只是让每个 token 更有价值。30-50% 的压缩率在大多数场景下可行。

压缩方法对比

方法压缩率质量损失实现难度推荐场景
工具输出截断20-70%低极低所有 Agent 产品
对话历史摘要50-80%中低长对话场景
系统提示词精简20-40%低低所有场景
LLMLingua30-50%中中文档密集型
RepoMap 结构化60-90%低中代码 Agent
选择性上下文40-60%中中RAG 场景

工具输出截断

✅ 最简单有效的压缩：Agent 调用工具后，返回结果通常远比需要的长。一个 ls 命令可能返回 1000 行，但 Agent 只需要知道关键信息。截断是最容易实现、ROI 最高的压缩方法。 
class ToolOutputTruncator:
    """工具输出截断器"""
    
    # 默认截断策略
    DEFAULT_MAX_CHARS = 2000
    DEFAULT_MAX_LINES = 50
    
    def __init__(self, max_chars: int = None, max_lines: int = None):
        self.max_chars = max_chars or self.DEFAULT_MAX_CHARS
        self.max_lines = max_lines or self.DEFAULT_MAX_LINES
    
    def truncate(self, output: str, tool_name: str = None) -> str:
        """截断工具输出"""
        # 按工具类型定制截断策略
        strategy = self._get_strategy(tool_name)
        
        if strategy == 'smart':
            return self._smart_truncate(output)
        elif strategy == 'tail':
            return self._tail_truncate(output)
        else:
            return self._simple_truncate(output)
    
    def _get_strategy(self, tool_name: str) -> str:
        """根据工具类型选择截断策略"""
        strategies = {
            'shell': 'tail',       # 命令行输出：保留最后 N 行
            'file_read': 'smart',  # 文件读取：保留头尾
            'web_fetch': 'smart',  # 网页抓取：保留关键段落
            'search': 'smart',     # 搜索结果：保留摘要
        }
        return strategies.get(tool_name, 'simple')
    
    def _simple_truncate(self, text: str) -> str:
        """简单截断"""
        if len(text) <= self.max_chars:
            return text
        
        lines = text.split('\n')
        if len(lines) <= self.max_lines:
            return text[:self.max_chars] + "\n...[截断]"
        
        kept = lines[:self.max_lines]
        return '\n'.join(kept)[:self.max_chars] + f"\n...[截断，共{len(lines)}行]"
    
    def _tail_truncate(self, text: str) -> str:
        """保留末尾（适合命令行输出）"""
        lines = text.split('\n')
        if len(lines) <= self.max_lines:
            return text[:self.max_chars]
        
        kept = lines[-self.max_lines:]
        result = '\n'.join(kept)[:self.max_chars]
        return f"...[省略前{len(lines)-self.max_lines}行]\n{result}"
    
    def _smart_truncate(self, text: str) -> str:
        """智能截断：保留头尾 + 关键段落"""
        if len(text) <= self.max_chars:
            return text
        
        budget = self.max_chars
        head_size = budget // 3
        tail_size = budget // 3
        
        head = text[:head_size]
        tail = text[-tail_size:]
        
        # 尝试提取包含关键词的段落
        keywords = ['error', 'warning', 'failed', 'success', '错误', '警告']
        key_lines = []
        for line in text.split('\n'):
            if any(kw in line.lower() for kw in keywords):
                key_lines.append(line)
        
        key_text = '\n'.join(key_lines[:5])[:budget // 3] if key_lines else ""
        
        parts = [head, "...[截断]", key_text, "...[截断]", tail]
        return '\n'.join(p for p in parts if p)

# 使用
truncator = ToolOutputTruncator(max_chars=2000, max_lines=50)

# Agent 工具调用后
raw_output = execute_tool("shell", "ls -la /var/log/")
compressed = truncator.truncate(raw_output, tool_name="shell")
# 从可能的 10000+ 字符 → 2000 字符，80% 压缩

LLMLingua 压缩

# LLMLingua: 微软开发的 Prompt 压缩工具
# 核心思想：用小模型判断每个 token 的重要性，删除不重要的 token
# pip install llmlingua

from llmlingua import PromptCompressor

class LLMLinguaCompressor:
    """LLMLingua Prompt 压缩器"""
    
    def __init__(self, model_name: str = "microsoft/llmlingua-2-xlm-roberta-large-meetingbank"):
        self.compressor = PromptCompressor(model_name=model_name)
    
    def compress(self, prompt: str, target_token: int = None, 
                 rate: float = 0.5) -> dict:
        """压缩 prompt
        
        Args:
            prompt: 原始 prompt
            target_token: 目标 token 数（与 rate 二选一）
            rate: 压缩率（0.5 = 压缩到 50%）
        """
        result = self.compressor.compress_prompt(
            prompt,
            rate=rate,
            target_token=target_token,
        )
        
        return {
            'original': prompt,
            'compressed': result['compressed_prompt'],
            'original_tokens': result['origin_tokens'],
            'compressed_tokens': result['compressed_tokens'],
            'rate': result['compressed_tokens'] / result['origin_tokens'],
            'savings_pct': (1 - result['compressed_tokens'] / result['origin_tokens']) * 100
        }

# 使用
compressor = LLMLinguaCompressor()

original = """Please analyze the following customer feedback and categorize it 
into one of these categories: Product Quality, Customer Service, Shipping, 
Pricing, or Other. Also determine the sentiment (positive, negative, or neutral) 
and extract the key issue mentioned.

Customer feedback: I received my order yesterday and the product quality is 
excellent, but the shipping took way too long - almost 3 weeks! I would have 
given 5 stars if it arrived sooner. The packaging was also a bit damaged."""

result = compressor.compress(original, rate=0.6)
print(f"原始: {result['original_tokens']} tokens")
print(f"压缩后: {result['compressed_tokens']} tokens")
print(f"节省: {result['savings_pct']:.1f}%")
print(f"压缩结果: {result['compressed']}")

OpenHands Condenser 策略

# OpenHands (原 OpenDevin) 的 Condenser 策略
# 核心思想：对 Agent 的长历史进行智能压缩，保留关键决策信息

class AgentHistoryCondenser:
    """Agent 历史压缩器（灵感来自 OpenHands Condenser）"""
    
    def __init__(self, max_history_events: int = 20, 
                 summary_model: str = "gpt-4o-mini"):
        self.max_events = max_history_events
        self.summary_model = summary_model
    
    def condense(self, events: list[dict]) -> list[dict]:
        """压缩 Agent 事件历史"""
        if len(events) <= self.max_events:
            return events
        
        # 保留最近的 N 个事件
        recent = events[-self.max_events:]
        
        # 对早期事件生成摘要
        early_events = events[:-self.max_events]
        summary = self._summarize_events(early_events)
        
        # 返回：摘要 + 最近事件
        return [
            {
                'role': 'system',
                'content': f'[之前操作的摘要]\n{summary}\n[/摘要结束]'
            },
            *recent
        ]
    
    def _summarize_events(self, events: list[dict]) -> str:
        """摘要早期事件"""
        # 提取关键信息
        key_info = []
        for event in events:
            if event.get('type') == 'tool_call':
                key_info.append(
                    f"- 执行了 {event['tool']}({self._summarize_args(event.get('args', {}))})"
                )
            elif event.get('type') == 'observation':
                result = str(event.get('result', ''))[:100]
                key_info.append(f"- 结果: {result}")
            elif event.get('type') == 'thought':
                key_info.append(f"- 思考: {str(event.get('content', ''))[:100]}")
        
        # 如果摘要本身太长，用小模型进一步压缩
        summary_text = '\n'.join(key_info)
        if len(summary_text) > 1000:
            # 用小模型压缩
            return self._llm_compress(summary_text)
        return summary_text
    
    def _summarize_args(self, args: dict) -> str:
        """精简参数显示"""
        if not args:
            return ""
        parts = []
        for k, v in args.items():
            val = str(v)[:50]
            parts.append(f"{k}={val}")
        return ', '.join(parts)

RepoMap 结构化摘要

# RepoMap: 代码仓库的结构化摘要方法
# 只发送文件结构和关键代码片段，而不是完整代码
# 灵感来自 Aider 的 repo-map 功能

class RepoMapGenerator:
    """代码仓库结构化摘要生成器"""
    
    def __init__(self, max_tokens: int = 4000):
        self.max_tokens = max_tokens
    
    def generate(self, repo_path: str, focus_files: list[str] = None) -> str:
        """生成仓库的结构化摘要"""
        # 1. 文件树
        tree = self._get_file_tree(repo_path)
        
        # 2. 每个文件的结构概要（类、函数签名）
        structures = self._extract_structures(repo_path, focus_files)
        
        # 3. 聚焦文件的完整内容
        focused = self._get_focused_content(repo_path, focus_files)
        
        # 4. 组合，控制在 token 预算内
        parts = [
            f"# 文件结构\n{tree}",
            f"# 代码结构\n{structures}",
        ]
        
        if focused:
            parts.append(f"# 聚焦文件\n{focused}")
        
        result = '\n\n'.join(parts)
        
        # 如果超预算，逐步删减
        while self._estimate_tokens(result) > self.max_tokens and len(parts) > 1:
            parts.pop(-2)  # 先删结构概要
            result = '\n\n'.join(parts)
        
        return result
    
    def _get_file_tree(self, path: str, max_depth: int = 3) -> str:
        """获取文件树"""
        import os
        tree_lines = []
        for root, dirs, files in os.walk(path):
            depth = root.replace(path, '').count(os.sep)
            if depth > max_depth:
                continue
            indent = '  ' * depth
            tree_lines.append(f"{indent}{os.path.basename(root)}/")
            sub_indent = '  ' * (depth + 1)
            for file in files[:10]:  # 每个目录最多10个文件
                tree_lines.append(f"{sub_indent}{file}")
        return '\n'.join(tree_lines[:50])
    
    def _extract_structures(self, path: str, 
                             focus_files: list[str] = None) -> str:
        """提取代码结构（类、函数签名）"""
        # 简化实现：使用正则提取
        structures = []
        for file in (focus_files or []):
            content = open(f"{path}/{file}").read()
            # 提取 class 和 function 定义
            for line in content.split('\n'):
                if any(line.strip().startswith(kw) for kw in 
                       ['class ', 'def ', 'async def ', 'function ']):
                    structures.append(f"{file}: {line.strip()}")
        return '\n'.join(structures)
    
    def _get_focused_content(self, path: str, 
                              focus_files: list[str]) -> str:
        """获取聚焦文件的完整内容"""
        if not focus_files:
            return ""
        parts = []
        for file in focus_files[:3]:  # 最多3个文件
            try:
                content = open(f"{path}/{file}").read()
                if len(content) > 1000:
                    content = content[:800] + "\n...[截断]..." + content[-200:]
                parts.append(f"## {file}\n```{content}```")
            except FileNotFoundError:
                pass
        return '\n\n'.join(parts)
    
    def _estimate_tokens(self, text: str) -> int:
        """粗略估算 token 数"""
        return len(text) // 4  # 英文约4字符/token，中文约2字符/token

系统提示词精简

# 系统提示词优化：减少冗余，保持效果

# ❌ 冗长的系统提示词（~800 tokens）
BAD_SYSTEM_PROMPT = """You are an AI assistant that helps users with their questions. 
You should always be helpful, accurate, and provide detailed responses. 
When you don't know the answer, you should honestly say so instead of making things up.
Please format your responses in a clear and organized manner.
If the user asks about something dangerous or illegal, politely decline.
Always maintain a professional tone.
Try to provide examples when explaining complex concepts.
Break down long answers into sections with headers.
Use bullet points for lists.
Summarize key takeaways at the end.
If relevant, suggest follow-up questions the user might want to ask."""

# ✅ 精简的系统提示词（~200 tokens，压缩 75%）
GOOD_SYSTEM_PROMPT = """Rules: 1)Helpful+accurate 2)Admit unknowns 3)Decline harmful requests 4)Use headers/bullets for clarity 5)Brief summary at end"""

# 测试：精简版是否影响输出质量？
# 经验法则：指令越简洁，模型反而越容易遵守
# 因为长指令中有些会互相矛盾，模型需要"平衡"这些指令

压缩效果实测

场景原始 tokens压缩后压缩率质量影响
工具输出截断5000150070%↓极小（Agent通常只看关键信息）
对话摘要（10轮→摘要）8000200075%↓小（保留关键信息）
LLMLingua（文档）6000300050%↓中（部分细节丢失）
RepoMap（代码库）50000400092%↓小（保留结构+聚焦文件）
系统提示词精简80020075%↓小（模型对简洁指令更遵守）
✅ 推荐的压缩优先级：
  1. 工具输出截断 — 最简单，每个 Agent 产品都应该做
  2. 系统提示词精简 — 一次性工作，长期收益
  3. 对话历史摘要 — 长对话场景必须做
  4. RepoMap — 代码 Agent 专用，效果显著
  5. LLMLingua — 需要额外模型，ROI 取决于场景

← 上一篇：Prompt Caching | 下一篇：模型路由 → | 返回首页

相关知识

🤗 Smolagents → 异步任务与队列 → ✅ 上线安全清单 - AI 安全合规 → Goose → 🌐 Browser Use →