| name | context-optimizer |
| description | 上下文优化专家。专注于长对话中的上下文管理、token 效率和性能优化。解决 lost-in-middle、context poisoning 等问题，提升 AI 代理在复杂任务中的表现。 |
| metadata | {"short-description":"上下文管理与优化","keywords":["context-optimizer","上下文优化","token 效率","长对话","压缩策略","缓存机制","性能优化","上下文窗口"],"category":"性能优化","author":"Bensz Conan","platform":"Claude Code | OpenAI Codex | ChatGPT"} |
Context Optimizer - 上下文优化专家
BenszAPI 任务工作区
本 Skill 的新任务中间文件统一写入 ./.bensz-api/task-{yyyymmdd-hhmm}-{简短描述}/{skill名}/input|output|log/。同一任务复用一个任务根目录；多 Skill 协作才创建 shared/。正式交付物不写入该目录，历史隐藏目录只允许显式兼容读取、迁移或清理。
与 bensz-collect-bugs 的协作约定
- 因本 skill 设计缺陷导致的 bug，先用
bensz-collect-bugs 规范记录到 ~/.bensz-skills/bugs/，不要直接修改用户本地已安装的 skill 源码；若有 workaround，先记 bug，再继续完成任务。
- 只有用户明确要求“report bensz skills bugs”等公开上报时，才用本地
gh 上传新增 bug 到 huangwb8/bensz-bugs；不要 pull / clone 整个仓库。
核心理念
上下文优化 是长对话性能的关键：
┌─────────────────────────────────────────────────────────┐
│ 识别问题 → 压缩历史 → 掩码加载 → 缓存重用 → 性能提升 │
└─────────────────────────────────────────────────────────┘
核心问题：
- ❌ Lost-in-Middle：关键信息被中间内容淹没
- ❌ Context Poisoning：冲突信息干扰判断
- ❌ Distraction：无关信息浪费 token
- ❌ Context Clash：多信息源冲突
何时使用本技能
在以下场景时激活：
- 长对话导致性能下降
- Context window 接近限制
- AI 遗忘之前的信息
- 提到"上下文"、"token 限制"、"效率"
上下文问题识别
问题 1：Lost-in-Middle
表现：
- AI 遗忘对话中间的关键信息
- 首尾信息记住，中间信息遗忘
- 需要重复提供相同信息
检测：
def detect_lost_in_middle(conversation: list) -> bool:
"""检测是否出现 lost-in-middle 问题"""
if len(conversation) < 10:
return False
questions = [msg for msg in conversation if '?' in msg]
unique_questions = set(questions)
if len(questions) > len(unique_questions) * 1.5:
return True
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
问题 2：Context Poisoning
表现：
- AI 产生矛盾的回答
- 错误信息影响判断
- 不同来源信息冲突
检测：
def detect_context_poisoning(conversation: list) -> list:
"""检测上下文污染"""
conflicts = []
facts = extract_facts(conversation)
for fact1, fact2 in combinations(facts, 2):
if are_contradictory(fact1, fact2):
conflicts.append({
'type': 'contradiction',
'fact1': fact1,
'fact2': fact2,
'severity': 'high'
})
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：压缩策略
历史压缩
class 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"]
Returns:
压缩后的对话
"""
priority = retention_priority or ["current_task", "decisions", "errors"]
categorized = self._categorize_messages(conversation)
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: ) -> :
categories = {
: [],
: [],
: [],
: []
}
msg conversation:
._is_task_related(msg):
categories[].append(msg)
._is_decision(msg):
categories[].append(msg)
._is_error(msg):
categories[].append(msg)
:
categories[].append(msg)
categories
() -> :
key_points = extract_key_points(message)
summary = summarize(key_points)
() -> :
(text.split()) *
增量摘要
class IncrementalSummarizer:
"""增量摘要器"""
def __init__(self, summary_interval: int = 10):
self.summary_interval = summary_interval
self.summaries = []
def add_messages(self, messages: list) -> str:
"""添加消息并生成摘要"""
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 = []
key_info[]:
summary_parts.append()
key_info[]:
summary_parts.append()
key_info[]:
summary_parts.append()
key_info[]:
summary_parts.append()
.join(summary_parts)
策略 2：掩码策略
按需加载
class 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:
"""判断是否应该加载"""
relevance = metadata.get(, )
token_usage = metadata.get(, )
relevance > token_usage <
策略 3：缓存策略
智能缓存
class 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 = (.cache.items())
items.sort(key= x: .access_count.get(x[], ) * x[].get(, ))
items:
key_to_remove = items[][]
.cache[key_to_remove]
.access_count[key_to_remove]
cache = SmartCache()
code_structure = parse_code()
cache.(, code_structure, priority=)
cached = cache.get()
cached:
use_cached_structure(cached)
优化检查清单
问题诊断
压缩策略
掩码策略
缓存策略
最佳实践
1. 分阶段处理
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. 渐进式信息披露
def provide_context():
return """
这是项目的完整文档，包括架构、API、配置等...
（可能 10000+ tokens）
"""
def provide_context():
return """
项目概述：这是一个 Web 应用
需要详细信息时，可查阅：
- [架构设计](docs/architecture.md)
- [API 文档](docs/api.md)
- [配置指南](docs/config.md)
（约 100 tokens）
"""
相关参考
