Context Optimizer - 上下文优化专家
核心理念
上下文优化 是长对话性能的关键：
┌─────────────────────────────────────────────────────────┐ │ 识别问题 → 压缩历史 → 掩码加载 → 缓存重用 → 性能提升 │ └─────────────────────────────────────────────────────────┘
核心问题：
- ❌ Lost-in-Middle：关键信息被中间内容淹没
- ❌ Context Poisoning：冲突信息干扰判断
- ❌ Distraction：无关信息浪费 token
- ❌ Context Clash：多信息源冲突
何时使用本技能
在以下场景时激活：
- 长对话导致性能下降
- Context window 接近限制
- AI 遗忘之前的信息
- 提到"上下文"、"token 限制"、"效率"
上下文问题识别
问题 1：Lost-in-Middle
表现：
- AI 遗忘对话中间的关键信息
- 首尾信息记住，中间信息遗忘
- 需要重复提供相同信息
检测：
pythondef detect_lost_in_middle(conversation: list) -> bool: """检测是否出现 lost-in-middle 问题""" # 1. 检查对话长度 if len(conversation) < 10: return False # 2. 检查是否有重复提问 questions = [msg for msg in conversation if '?' in msg] unique_questions = set(questions) if len(questions) > len(unique_questions) * 1.5: return True # 存在重复提问 # 3. 检查中间内容是否被引用 middle_start = len(conversation) // 3 middle_end = len(conversation) * 2 // 3 middle_content = conversation[middle_start:middle_end] # 检查后续对话是否引用中间内容 later_refs = sum( 1 for msg in conversation[middle_end:] if any(keyword in msg for keyword in extract_keywords(middle_content)) ) if later_refs < len(middle_content) * 0.1: return True # 中间内容被遗忘 return False
问题 2：Context Poisoning
表现：
- AI 产生矛盾的回答
- 错误信息影响判断
- 不同来源信息冲突
检测：
pythondef detect_context_poisoning(conversation: list) -> list: """检测上下文污染""" conflicts = [] # 1. 提取所有事实陈述 facts = extract_facts(conversation) # 2. 检测矛盾 for fact1, fact2 in combinations(facts, 2): if are_contradictory(fact1, fact2): conflicts.append({ 'type': 'contradiction', 'fact1': fact1, 'fact2': fact2, 'severity': 'high' }) # 3. 检测信息源冲突 sources = group_by_source(facts) for source, source_facts in sources.items(): if has_internal_conflicts(source_facts): conflicts.append({ 'type': 'source_conflict', 'source': source, 'severity': 'medium' }) return conflicts
优化策略
策略 1：压缩策略
历史压缩
pythonclass ContextCompressor: """上下文压缩器""" def compress_history( self, conversation: list, max_tokens: int, retention_priority: list[str] = None ) -> list: """ 压缩对话历史 Args: conversation: 对话历史 max_tokens: 最大 token 数 retention_priority: 保留优先级 ["current_task", "decisions", "errors"] 返回值： 压缩后的对话 """ priority = retention_priority or ["current_task", "decisions", "errors"] # 1. 分类消息 categorized = self._categorize_messages(conversation) # 2. 按优先级保留 retained = [] current_tokens = 0 for category in priority: messages = categorized.get(category, []) for msg in messages: tokens = self._count_tokens(msg) if current_tokens + tokens > max_tokens: # 尝试压缩 compressed = self._compress_message(msg) if current_tokens + self._count_tokens(compressed) <= max_tokens: retained.append(compressed) current_tokens += self._count_tokens(compressed) else: retained.append(msg) current_tokens += tokens return retained def _categorize_messages(self, conversation: list) -> dict: """分类消息""" categories = { 'current_task': [], 'decisions': [], 'errors': [], 'context': [] } for msg in conversation: if self._is_task_related(msg): categories['current_task'].append(msg) elif self._is_decision(msg): categories['decisions'].append(msg) elif self._is_error(msg): categories['errors'].append(msg) else: categories['context'].append(msg) return categories def _compress_message(self, message: str) -> str: """压缩单条消息""" # 提取关键信息 key_points = extract_key_points(message) # 生成摘要 summary = summarize(key_points) return f"[摘要] {summary}" def _count_tokens(self, text: str) -> int: """估算 token 数量""" return len(text.split()) * 1.3 # 粗略估计
增量摘要
pythonclass IncrementalSummarizer: """增量摘要器""" def __init__(self, summary_interval: int = 10): self.summary_interval = summary_interval self.summaries = [] def add_messages(self, messages: list) -> str: """添加消息并生成摘要""" # 每隔 N 条消息生成一次摘要 if len(messages) % self.summary_interval == 0: summary = self._generate_summary(messages[-self.summary_interval:]) self.summaries.append(summary) # 返回完整的摘要历史 return "\n\n".join(self.summaries) def _generate_summary(self, messages: list) -> str: """生成消息摘要""" # 提取关键信息 key_info = { 'tasks': self._extract_tasks(messages), 'decisions': self._extract_decisions(messages), 'errors': self._extract_errors(messages), 'outcomes': self._extract_outcomes(messages) } # 格式化摘要 summary_parts = [] if key_info['tasks']: summary_parts.append(f"任务: {', '.join(key_info['tasks'])}") if key_info['decisions']: summary_parts.append(f"决策: {', '.join(key_info['decisions'])}") if key_info['errors']: summary_parts.append(f"错误: {', '.join(key_info['errors'])}") if key_info['outcomes']: summary_parts.append(f"结果: {', '.join(key_info['outcomes'])}") return " | ".join(summary_parts)
策略 2：掩码策略
按需加载
pythonclass LazyContextLoader: """懒加载上下文""" def __init__(self): self.loaded_references = {} self.reference_metadata = {} def load_reference( self, ref_name: str, force: bool = False ) -> str | None: """ 按需加载参考文档 Args: ref_name: 参考文档名称 force: 是否强制重新加载 """ # 已加载且不强制 if ref_name in self.loaded_references and not force: return self.loaded_references[ref_name] # 检查元数据 metadata = self.reference_metadata.get(ref_name) if not metadata: return None # 按需决策 if self._should_load(metadata): content = self._load_from_disk(ref_name) self.loaded_references[ref_name] = content return content return None def _should_load(self, metadata: dict) -> bool: """判断是否应该加载""" # 判断逻辑： # 1. 是否被明确请求 # 2. 相关性分数 # 3. 当前 token 使用率 relevance = metadata.get('relevance', 0) token_usage = metadata.get('token_usage', 0) return relevance > 0.7 or token_usage < 0.8
策略 3：缓存策略
智能缓存
pythonclass SmartCache: """智能缓存系统""" def __init__(self, max_size: int = 100): self.cache = {} self.max_size = max_size self.access_count = {} def get(self, key: str) -> any: """获取缓存""" if key in self.cache: # 更新访问计数 self.access_count[key] = self.access_count.get(key, 0) + 1 return self.cache[key] return None def set(self, key: str, value: any, priority: int = 1): """设置缓存""" # 缓存已满，清理低优先级项 if len(self.cache) >= self.max_size: self._evict_low_priority() self.cache[key] = value self.access_count[key] = 0 def _evict_low_priority(self): """淘汰低优先级缓存""" # 按 (访问次数 * 优先级) 排序 items = list(self.cache.items()) items.sort(key=lambda x: self.access_count.get(x[0], 0) * x[1].get('priority', 1)) # 移除最低分项 if items: key_to_remove = items[0][0] del self.cache[key_to_remove] del self.access_count[key_to_remove] # 使用示例 cache = SmartCache() # 缓存解析结果 code_structure = parse_code('main.py') cache.set('code:main.py', code_structure, priority=2) # 获取缓存 cached = cache.get('code:main.py') if cached: use_cached_structure(cached)
优化检查清单
问题诊断
- 对话长度是否合理
- 是否有重复提问
- 中间信息是否被遗忘
- 是否存在信息冲突
压缩策略
- 历史对话已摘要
- 关键决策已保留
- 错误信息已保留
- 无关信息已过滤
掩码策略
- 参考文档按需加载
- 详细信息延迟加载
- 避免一次性加载所有内容
缓存策略
- 解析结果已缓存
- 频繁访问内容已缓存
- 缓存有淘汰机制
最佳实践
1. 分阶段处理
python# ❌ 一次性处理所有信息 def process_large_file(filename): content = read_file(filename) # 可能很大 result = analyze(content) return result # ✅ 分阶段处理 def process_large_file(filename): # 第一阶段：获取结构 structure = get_file_structure(filename) # 第二阶段：按需加载 for section in structure.sections: content = load_section(filename, section) result = analyze_section(content) return aggregate_results(results)
2. 渐进式信息披露
python# ❌ 一次性提供所有信息 def provide_context(): return """ 这是项目的完整文档，包括架构、API、配置等... （可能 10000+ tokens） """ # ✅ 渐进式披露 def provide_context(): return """ 项目概述：这是一个 Web 应用 需要详细信息时，可查阅： - [架构设计](docs/architecture.md) - [API 文档](docs/api.md) - [配置指南](docs/config.md) （约 100 tokens） """
相关参考
约束
公共硬约束
本块由 docs/templates/skill-common-constraints.md 统一维护；每个 SKILL.md 的 ## 约束 必须逐字同步本块，不得在副本中改写公共规则。
- 任务需要落盘时，使用唯一的
./.bensz-api/task-{yyyymmdd-hhmm}-{简短描述}/根目录；共享材料放入shared/，Skill 专属材料放入该 Skill 的input/、output/、log/。 - 正式交付物、源代码和正式计划按项目约定保存，不写入任务工作区；未经授权不覆盖、删除、迁移或远程写入。
- 项目维护变更检查 BAC 可用性并记录需求、AI 产出、工具结果、文件改动和验证摘要；BAC 只做过程审计，不替代署名、责任或合规判断。
- 不记录 API Key、访问令牌、密码、Cookie、环境/凭据文件、私有 Prompt、身份信息、本地用户名、主机名或不必要的大体积原始数据。
- 文件路径必须规范化并限制在授权项目范围内；外部 URL、子进程和网络访问遵循最小权限，防止路径遍历、SSRF 和命令注入。
- Skill 版本唯一记录在自身
config.yaml:skill_info.version；公开 API、协议、目录或配置变更同步文档与CHANGELOG.md。 bensz-collect-bugs是一个 Agent Skill；仅将 Bensz Agent Skill 或 Bensz 基础设施本身的设计缺陷交给它。先脱敏写入~/.bensz-skills/bugs/，当前任务不中断，只有用户明确要求才公开上报，禁止直接修改用户已安装的 Skill 源码。
