wehub/skillhub-102-context-optimization
1
0
Fork 0
Code Issues Pull Requests Actions Deployments Models Agent Notes Packages Projects Releases Wiki Activity

chore: import zh skill context-optimization

Browse Source
This commit is contained in:
wehub-skill-sync
2026-07-13 21:36:25 +08:00
commit 2ab0cadabf
4 changed files with 1068 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.wehub.md
+9
View File
@@ -0,0 +1,9 @@
# WeHub 来源说明
- Skill 名称：`context-optimization`
- 中文类目：Token 成本与 KV-Cache 优化
- 上游仓库：`muratcankoylan__Agent-Skills-for-Context-Engineering`
- 上游路径：`skills/context-optimization/SKILL.md`
- 上游链接：https://github.com/muratcankoylan/Agent-Skills-for-Context-Engineering/blob/HEAD/skills/context-optimization/SKILL.md
- 本仓库为 WeHub 中文 Skill 汉化包，基于 skill 市场筛选 Top200 清单整理
- 原作者、版权和许可证信息以上游仓库为准
SKILL.md
+219
View File
@@ -0,0 +1,219 @@
---
name: context-optimization
description: 该技能应用于提升上下文效率：上下文预算管理、观察遮蔽、前缀或 KV 缓存策略、分区、令牌成本降低、检索范围限定，以及在不降低回答质量的前提下扩展有效上下文容量。
---
# 上下文优化技术
上下文优化通过战略性压缩、遮蔽、缓存和分区，扩展有限上下文窗口的有效容量。有效的优化能在不依赖更大模型或更长窗口的前提下提升有用容量——但前提是必须基于度量纪律进行应用。以下技术按影响程度和风险排序。
## 何时激活
在以下情形中激活此技能：
- 上下文预算或令牌成本限制了任务复杂度
- 观察遮蔽可将冗长的工具输出替换为可检索的引用
- 前缀或 KV 缓存命中率需要提升
- 检索范围限定可减少不相关的已加载上下文
- 上下文分区可跨代理扩展有效容量
- 需要预算触发器来执行遮蔽、压缩或分区
对于由其他技能负责的相邻工作，请勿激活此技能：
- 解释注意力机制或上下文窗口为何如此表现：`context-fundamentals`
- 诊断活跃的中间丢失、投毒、干扰、混淆或冲突：`context-degradation`
- 为长对话设计结构化的交接摘要：`context-compression`
- 将大型输出、计划或日志存储为文件：`filesystem-context`
## 核心概念
按以下优先级应用四种主要策略：
1. **KV 缓存优化**——重新排序并稳定提示结构，使推理引擎能重用缓存的 Key/Value 张量。当运行时支持前缀缓存时，这是成本最低的优化：质量风险低，能立即节省成本和延迟。当存在稳定前缀时优先应用。
2. **观察遮蔽**——一旦冗长工具输出的目的已完成，将其替换为紧凑引用。工具输出可能主导代理轨迹（claim-context-optimization-tool-output-dominance），因此遮蔽通常能带来最大的容量收益。若下游需要，原始内容仍可检索。
3. **压缩**——当上下文利用率超过 70% 时，对累积的上下文进行总结，然后用总结重新初始化。这能提炼窗口内容，同时保留任务关键状态。压缩是有损的——应在遮蔽已移除低价值主体后再进行。
4. **上下文分区**——当单个窗口无法容纳完整问题时，将工作拆分为多个具有隔离上下文的子代理。每个子代理在其专注于子任务的干净上下文中运行。当预计上下文超过窗口限制的 60% 时保留此策略，因为协调开销是实际存在的。
基本原则：上下文质量比数量更重要。每一次优化都应在保留信号的同时减少噪声。优化前先度量，再度量优化的效果。
## 详细主题
### 压缩策略
当上下文利用率超过 70% 时触发压缩：总结当前上下文，然后用总结重新初始化。这能以高保真的方式提炼窗口内容，使后续执行能以最小的性能降级继续。优先压缩工具输出（它们消耗 80% 以上的令牌），然后是旧的对话轮次，最后是检索到的文档。切勿压缩系统提示——它锚定模型行为，移除它会导致不可预测的性能下降。
按消息类型保留不同元素：
- **工具输出**：提取关键发现、指标、错误代码和结论。剔除冗长的原始输出、堆栈跟踪（除非正在调试）和样板式页眉/页脚。
- **对话轮次**：保留决策、承诺、用户偏好和上下文转变。移除填充内容、客套话以及已导向已记录结论的探索性来回对话。
- **检索到的文档**：保留与当前任务相关的声明、事实和数据点。移除仅用于一次性推理目的的支持性证据和详细阐述。
目标是将令牌减少 50-70%，质量下降不超过 5%。如果压缩幅度超过 70%，请检查摘要是否存在关键信息丢失——过度压缩是最常见的失败模式。
### 观察遮蔽
根据时效性和持续相关性有选择地遮蔽观察，而非统一处理。应用以下规则：
- **从不遮蔽**：对当前任务至关重要的观察、最近一轮的观察、活跃推理链中使用的观察，以及调试进行中的错误输出。
- **3 轮后遮蔽**：关键点已被提取到对话流中的冗长输出。替换为紧凑引用：`[Obs:{ref_id} 已省略。要点：{summary}。完整内容可检索。]`
- **始终立即遮蔽**：重复/重复输出、样板式页眉和页脚、已在对话早期总结过的输出。
遮蔽应使被遮蔽的观察减少 60-80%，质量影响低于 2%。关键是要保持可检索性——将完整内容存储在外部，并在上下文中保留引用 ID，以便代理在需要时能请求原始内容。
### KV 缓存优化
通过构建提示结构使稳定内容占据前缀、动态内容出现在末尾，最大化前缀缓存命中率。KV 缓存存储推理过程中计算的 Key 和 Value 张量；当连续请求共享相同前缀时，会重用缓存的张量，从而节省成本和延迟。
在每个提示中应用此顺序：
1. 系统提示（最稳定——在会话内从不改变）
2. 工具定义（跨请求稳定）
3. 频繁重用的模板和少量示例
4. 对话历史（不断增长，但与前几轮共享前缀）
5. 当前查询和动态内容（最不稳定——始终放在最后）
设计提示以获得缓存稳定性：从系统提示中移除时间戳、会话计数器和请求 ID。将动态元数据移入单独的用户消息或工具结果中，使其不会破坏前缀。前缀中哪怕一个空白字符的变化都会使该变化之后的所有缓存块失效。
对于稳定工作负载，目标缓存命中率达到 70% 以上。在规模应用下，这相当于缓存令牌的成本降低 50% 以上、延迟降低 40% 以上。
### 上下文分区
当单个上下文无法容纳完整问题且会触发激进压缩时，将工作分区到多个子代理。每个子代理在其专注于子任务的干净、聚焦的上下文中运行，然后将结构化结果返回给协调代理。
当预计任务上下文超过窗口限制的 60% 时，规划分区。将任务分解为独立的子任务，分配给各个子代理，并汇总结果。验证所有分区是否已完成，合并兼容的结果，如果聚合输出仍超出预算则应用摘要。
这种方法实现了关注点分离——详细的搜索上下文隔离在子代理内部，而协调器专注于综合。然而，协调有实际的令牌成本：协调器提示、结果聚合和错误处理都会消耗令牌。仅在节省的成本超过此开销时才进行分区。
### 预算管理
在会话开始前，为上下文类别分配明确的令牌预算：系统提示、工具定义、检索到的文档、消息历史、工具输出以及预留缓冲区（占总量的 5-10%）。持续监控各分类的使用情况与预算的对比，并在任一分类超出其分配或总利用率超过 70% 时触发优化。
使用基于触发器的优化而非定期优化。监控以下信号：
- 令牌利用率超过 80%——触发压缩
- 注意力退化指标（重复、遗漏指令）——触发遮蔽 + 压缩
- 质量评分低于基准——在优化前审计上下文组成
## 实践指导
### 优化决策框架
根据主导上下文的因素选择优化技术：
| 上下文组成 | 首要操作 | 次要操作 |
|---|---|---|
| 工具输出占主导（>50%） | 观察遮蔽 | 压缩剩余轮次 |
| 检索到的文档占主导 | 摘要 | 如果文档相互独立则分区 |
| 消息历史占主导 | 选择性保留的压缩 | 为新子任务分区 |
| 多个组件共同贡献 | 先做 KV 缓存优化，然后分层使用遮蔽 + 压缩 |
| 接近限制且正在调试 | 仅遮蔽已解决的工具输出——保留错误详情 |
### 性能目标
跟踪以下指标以验证优化效果：
- **压缩**：令牌减少 50-70%，质量下降 <5%，压缩步骤本身带来的延迟开销 <10%
- **遮蔽**：被遮蔽的观察减少 60-80%，质量影响 <2%，延迟开销接近为零
- **缓存优化**：稳定工作负载命中率 70% 以上，成本降低 50% 以上，延迟降低 40% 以上
- **分区**：扣除协调器开销后的净令牌节省；通常需要 3 个以上子任务才能达到盈亏平衡
根据度量结果迭代策略。如果某项优化技术未能可衡量地改善目标指标，则将其移除——优化机制本身也会消耗令牌并增加延迟。
## 示例
**示例 1：压缩触发**
```python
if context_tokens / context_limit > 0.8:
context = compact_context(context)
```
**示例 2：观察遮蔽**
```python
if len(observation) > max_length:
ref_id = store_observation(observation)
return f"[Obs:{ref_id} 已省略。要点：{extract_key(observation)}]"
```
**示例 3：缓存友好排序**
```python
# 稳定内容在前
context = [system_prompt, tool_definitions] # 可缓存
context += [reused_templates] # 可重用
context += [unique_content] # 唯一内容
```
**示例 4：预算触发的优化策略**
```yaml
budgets:
tool_outputs: 35%
message_history: 30%
retrieved_documents: 20%
reserved_buffer: 15%
triggers:
tool_outputs_over_budget: 遮蔽已解决的观察
total_context_over_70_percent: 压缩消息历史
repeated_irrelevant_retrievals: 收紧检索范围
```
## 指南
1. 优化前先度量——了解当前状态
2. 先遮蔽再压缩——先移除低价值主体，再总结剩余内容
3. 通过一致的提示设计实现缓存稳定性
4. 在上下文变得有问题之前进行分区
5. 持续监控优化效果
6. 在令牌节省与质量保持之间取得平衡
7. 在生产规模下测试优化
8. 为边缘情况实现优雅降级
## 陷阱
1. **空白字符会破坏 KV 缓存**：提示前缀中哪怕一个空白字符或换行符的变化，都会使该点之后的所有 KV 缓存块失效。将系统提示固定为不可变字符串——不要向其中插入时间戳、版本号或会话 ID。在部署之间按字节比对提示模板。
2. **系统提示中的时间戳会破坏缓存命中率**：在系统提示中包含 `Current date: {today}` 或类似动态内容，会迫使每个新的一天（如果使用时间更细粒度，则是每个请求）都发生完全的缓存未命中。将动态元数据移入用户消息或附加在稳定前缀之后的单独工具结果中。
3. **压力下的压缩会丢失关键状态**：当执行压缩的模型本身处于上下文压力之下（利用率 >85%）时，其摘要质量会下降——它会遗漏任务目标、丢弃用户约束、将细微的状态扁平化。在 70-80% 时触发压缩，而非 90% 以上。如果必须在晚期进行压缩，请使用单独的模型调用，并提供一个仅包含待总结材料的干净上下文。
4. **遮蔽错误输出会破坏调试循环**：过度激进的遮蔽会隐藏代理在后续轮次诊断和修复问题所需的错误消息、堆栈跟踪和失败详情。在活跃调试期间（最近 3 轮内有错误），暂停对所有错误相关观察的遮蔽，直到问题解决。
5. **分区开销可能超过节省**：每个子代理需要自己的系统提示、工具定义和协调消息。对于少于 3 个独立子任务的任务，协调开销通常超过上下文节省。在决定分区前，估算总令牌数（协调器 + 所有子代理）。
6. **部署变更后的缓存未命中成本激增**：在部署之间重新排序工具、修改系统提示或更改少量示例，会使整个前缀缓存失效，导致成本暂时激增 2-5 倍，直到新缓存预热完成。逐步推出提示变更，并在部署窗口期间监控缓存命中率。
7. **压缩会造成对过期摘要的错误自信**：上下文一旦被压缩，摘要看起来具有权威性，但可能反映的是过时的状态。如果自压缩以来任务已发生变化（新的用户需求、修正的假设），摘要会默默携带过时信息。压缩后，在继续之前重新验证摘要与当前任务目标的一致性。
## 集成
此技能负责令牌效率策略和预算策略。相邻技能负责诊断、存储和架构：
- `context-fundamentals`：解释为什么上下文质量和注意力位置很重要的心智模型。
- `context-degradation`：在输出质量已下降时进行诊断。
- `context-compression`：有损摘要和交接策略。
- `filesystem-context`：基于文件的后台卸载，用于完整输出和日志。
- `multi-agent-patterns`：跨隔离代理上下文进行工作分区。
- `latent-briefing`：在兼容运行时中跨编排器-工作者边界的选择性 KV 保留。
- `evaluation`：衡量优化是否改善了质量、成本或延迟。
- `memory-systems`：为上下文提供即时检索的持久化检索层。
## 参考资料
内部参考：
- [优化技术参考](./references/optimization_techniques.md)——何时阅读：在实现特定优化技术且需要技能主体未提供的详细代码模式、阈值表或集成示例时。
本集合中的相关技能：
- context-fundamentals——何时阅读：不熟悉上下文窗口机制、令牌计数或注意力分布基础知识时。
- context-degradation——何时阅读：诊断代理性能为何下降，需要在选择优化方案前识别正在发生的退化模式时。
- evaluation——何时阅读：设置指标和基准来衡量某项优化技术是否确实改善了结果时。
外部资源：
- 上下文窗口限制研究——何时阅读：评估模型特定的上下文行为（例如中间丢失效应、注意力衰减曲线）时。
- KV 缓存优化技术——何时阅读：在推理基础设施层面（vLLM、TGI 或云提供商 API）实现前缀缓存时。
- 生产工程指南——何时阅读：在生产管道中部署上下文优化，需要可运维模式（监控、告警、回滚）时。
---
## 技能元数据
**创建日期**：2025-12-20
**最后更新**：2026-05-15
**作者**：Agent Skills for Context Engineering Contributors
**版本**：2.1.0
references/optimization_techniques.md
+278
View File
@@ -0,0 +1,278 @@
---
name: context-optimization-reference
description: 上下文优化技术参考文档
metadata:
type: reference
---
# 上下文优化参考
本文档提供了关于上下文优化技术与策略的详细技术参考。
## 压缩策略
### 基于摘要的压缩
基于摘要的压缩用简洁的摘要替代冗长的内容，同时保留关键信息。该方法通过识别可压缩的部分、生成捕捉要点的摘要，以及用摘要替换完整内容来实现。
压缩的有效性取决于保留哪些信息。关键决策、用户偏好以及当前任务状态绝不能压缩。中间结果和支持性证据可以更积极地摘要化。样板内容、重复信息以及探索性推理通常可以直接移除。
### Token 预算分配
有效的上下文预算管理需要了解不同上下文组件如何消耗 token，并进行策略性分配：
| 组件 | 典型范围 | 说明 |
|-----------|---------------|-------|
| 系统提示词 | 500–2000 tokens | 整个会话期间保持稳定 |
| 工具定义 | 每个工具 100–500 | 随工具数量增长 |
| 检索到的文档 | 不固定 | 通常是最大消耗者 |
| 消息历史 | 不固定 | 随对话增长 |
| 工具输出 | 不固定 | 可能占据绝大部分上下文 |
### 压缩阈值
在适当的阈值触发压缩，以维持性能：
- 达到有效上下文限制 70% 时发出警告阈值
- 达到有效上下文限制 80% 时触发压缩
- 达到有效上下文限制 90% 时进行激进压缩
具体阈值取决于模型行为与任务特征。部分模型会呈现渐进式退化，而另一些模型则表现出明显的性能悬崖。
## 观测屏蔽模式
### 选择性屏蔽
并非所有观测结果都应同等屏蔽。应考虑屏蔽那些已完成使命、不再需要用于活跃推理的观测结果。保留当前任务的核心观测结果。保留最近一轮的观测结果。保留可能被再次引用的观测结果。
### 屏蔽实现
```python
def selective_mask(observations: List[Dict], current_task: Dict) -> List[Dict]:
"""
根据相关性选择性屏蔽观测结果。
返回带有 mask 字段的观测结果，指示已屏蔽的内容。
"""
masked = []
for obs in observations:
relevance = calculate_relevance(obs, current_task)
if relevance < 0.3 and obs["age"] > 3:
# 低相关性且陈旧——屏蔽
masked.append({
**obs,
"masked": True,
"reference": store_for_reference(obs["content"]),
"summary": summarize_content(obs["content"])
})
else:
masked.append({
**obs,
"masked": False
})
return masked
```
## KV 缓存优化
### 前缀稳定性
KV 缓存的命中率取决于前缀的稳定性。稳定的前缀支持跨请求复用缓存。动态前缀会使缓存失效并强制重新计算。
应保持稳定的元素包括：系统提示词、工具定义以及常用模板。可能变化的元素包括：时间戳、会话标识符以及查询相关的内容。
### 缓存友好的设计
设计提示词以最大化缓存命中率：
1. 将稳定内容放在开头
2. 跨请求使用一致的格式
3. 尽可能避免在提示词中使用动态内容
4. 对动态内容使用占位符
```python
# 不利于缓存：提示词中包含动态时间戳
system_prompt = f"""
Current time: {datetime.now().isoformat()}
You are a helpful assistant.
"""
# 有利于缓存：提示词稳定，动态时间作为变量
system_prompt = """
You are a helpful assistant.
Current time is provided separately when relevant.
"""
```
## 上下文分区策略
### 子代理隔离
将工作分配到多个子代理中，以防止任何一个上下文增长过大。每个子代理使用干净的上下文，专注于其子任务。
### 分区规划
```python
def plan_partitioning(task: Dict, context_limit: int) -> Dict:
"""
根据上下文限制规划如何对任务进行分区。
返回分区策略与子任务定义。
"""
estimated_context = estimate_task_context(task)
if estimated_context <= context_limit:
return {
"strategy": "single_agent",
"subtasks": [task]
}
# 规划多代理方案
subtasks = decompose_task(task)
return {
"strategy": "multi_agent",
"subtasks": subtasks,
"coordination": "hierarchical"
}
```
## 优化决策框架
### 何时进行优化
在以下情况考虑上下文优化：上下文利用率超过 70%，回复质量随对话延长而下降，长上下文导致成本上升，或对话长度增加导致延迟升高。
### 应用何种优化
根据上下文构成选择优化策略：
如果工具输出占据上下文主导地位，应用观测屏蔽。如果检索到的文档占据主导地位，应用摘要化或分区。如果消息历史占据主导地位，应用带摘要的压缩。如果多个组件共同贡献，则组合多种策略。
### 优化效果评估
应用优化后，评估效果：
- 衡量 token 减少量
- 衡量质量保留程度（输出质量不应下降）
- 衡量延迟改善情况
- 衡量成本降低情况
根据评估结果迭代优化策略。
## 常见陷阱
### 过度激进的压缩
压缩过于激进可能会移除关键信息。务必保留任务目标、用户偏好以及最近的对话上下文。逐步增加压缩强度进行测试，以找到最佳平衡点。
### 屏蔽关键观测结果
屏蔽仍在使用的观测结果可能导致错误。跟踪观测结果的使用情况，仅屏蔽不再被引用的内容。考虑保留已屏蔽内容的引用，以便在需要时检索。
### 忽略注意力分布
中间丢失现象意味着信息放置位置至关重要。将关键信息放在注意力有利的位置（上下文的开头和结尾）。使用显式标记突出重要内容。
### 过早优化
并非所有上下文都需要优化。添加优化机制本身有开销。仅在上下文限制确实制约了代理性能时进行优化。
## 监控与告警
### 关键指标
跟踪以下指标以了解优化需求：
- 上下文 token 数量随时间变化
- 重复模式的缓存命中率
- 按上下文大小划分的回复质量指标
- 按上下文长度划分的每次对话成本
- 按上下文大小划分的延迟
### 告警阈值
针对以下情况设置告警：
- 上下文利用率超过 80%
- 缓存命中率低于 50%
- 质量分数下降超过 10%
- 成本超过基线上升
## 集成模式
### 与代理框架集成
将优化集成到代理工作流中：
```python
class OptimizingAgent:
def __init__(self, context_limit: int = 80000):
self.context_limit = context_limit
self.optimizer = ContextOptimizer()
def process(self, user_input: str, context: Dict) -> Dict:
# 检查是否需要优化
if self.optimizer.should_compact(context):
context = self.optimizer.compact(context)
# 使用优化后的上下文进行处理
response = self._call_model(user_input, context)
# 跟踪指标
self.optimizer.record_metrics(context, response)
return response
```
### 与记忆系统集成
将优化与记忆系统对接：
```python
class MemoryAwareOptimizer:
def __init__(self, memory_system, context_limit: int):
self.memory = memory_system
self.limit = context_limit
def optimize_context(self, current_context: Dict, task: str) -> Dict:
# 检查信息是否已在记忆中
relevant_memories = self.memory.retrieve(task)
# 将不需要留在上下文中的信息移至记忆
for mem in relevant_memories:
if mem["importance"] < threshold:
current_context = remove_from_context(current_context, mem)
# 保留引用，以便需要时从记忆检索
return current_context
```
## 性能基准
### 压缩性能
压缩应在保留质量的同时减少 token 数量。目标：
- 激进压缩实现 50–70% 的 token 减少
- 压缩带来的质量退化低于 5%
- 压缩开销导致的延迟增加低于 10%
### 屏蔽性能
观测屏蔽应显著减少 token 数量：
- 被屏蔽的观测结果减少 60–80%
- 屏蔽带来的质量影响低于 2%
- 延迟开销接近为零
### 缓存性能
KV 缓存优化应改善成本与延迟：
- 稳定工作负载下缓存命中率达 70% 以上
- 缓存命中实现 50% 以上的成本降低
- 缓存命中实现 40% 以上的延迟降低
scripts/compaction.py
+562
View File
@@ -0,0 +1,562 @@
"""
Context Optimization Utilities — compaction, masking, budgeting, and cache optimization.
Public API
----------
Functions:
estimate_token_count(text) -> int
estimate_message_tokens(messages) -> int
categorize_messages(messages) -> dict
summarize_content(content, category, max_length) -> str
design_stable_prompt(template, dynamic_values) -> str
calculate_cache_metrics(requests, cache) -> dict
Classes:
ObservationStore — Store and mask verbose tool outputs with retrievable references.
ContextBudget — Token budget allocation and optimization trigger detection.
PRODUCTION NOTES:
- Token estimation uses simplified heuristics (~4 chars/token for English).
Production systems should use model-specific tokenizers:
- OpenAI: tiktoken library
- Anthropic: anthropic tokenizer
- Local models: HuggingFace tokenizers
- Summarization functions use simple heuristics for demonstration.
Production systems should use:
- LLM-based summarization for high-quality compression
- Domain-specific summarization models
- Schema-based summarization for structured outputs
- Cache metrics are illustrative. Production systems should integrate
with actual inference infrastructure metrics.
"""
from typing import List, Dict, Optional, Tuple
import hashlib
import re
import time
__all__ = [
"estimate_token_count",
"estimate_message_tokens",
"categorize_messages",
"summarize_content",
"summarize_tool_output",
"summarize_conversation",
"summarize_document",
"summarize_general",
"ObservationStore",
"ContextBudget",
"design_stable_prompt",
"calculate_cache_metrics",
"generate_cache_recommendations",
]
# ---------------------------------------------------------------------------
# Token estimation
# ---------------------------------------------------------------------------
def estimate_token_count(text: str) -> int:
"""
Estimate token count for text.
Use when: a quick token budget check is needed and a model-specific
tokenizer is unavailable or too slow for the hot path.
Uses approximation: ~4 characters per token for English.
WARNING: This is a rough estimate. Actual tokenization varies by:
- Model (GPT-5.2, Claude 4.5, Gemini 3 have different tokenizers)
- Content type (code typically has higher token density)
- Language (non-English may have 2-3x higher token/char ratio)
Production usage::
import tiktoken
enc = tiktoken.encoding_for_model("gpt-4")
token_count = len(enc.encode(text))
"""
return len(text) // 4
def estimate_message_tokens(messages: List[Dict[str, str]]) -> int:
"""
Estimate token count for a message list.
Use when: checking whether the current conversation is approaching
the context budget threshold before deciding to compact or mask.
"""
total = 0
for msg in messages:
content = msg.get("content", "")
total += estimate_token_count(content)
# Add overhead for role/formatting
total += 10
return total
# ---------------------------------------------------------------------------
# Compaction functions
# ---------------------------------------------------------------------------
def categorize_messages(messages: List[Dict]) -> Dict[str, List[Dict]]:
"""
Categorize messages for selective compaction.
Use when: preparing to compact context and needing to apply different
summarization strategies per category (tool outputs first, then old
conversation turns, then retrieved documents — never the system prompt).
Returns a dict mapping category name to list of messages.
"""
categories: Dict[str, List[Dict]] = {
"system_prompt": [],
"tool_definition": [],
"tool_output": [],
"conversation": [],
"retrieved_document": [],
"other": [],
}
for msg in messages:
role = msg.get("role", "user")
content = msg.get("content", "")
if role == "system":
categories["system_prompt"].append({**msg, "category": "system_prompt"})
elif "tool_use" in msg.get("type", ""):
categories["tool_output"].append({**msg, "category": "tool_output"})
elif role == "user":
categories["conversation"].append({**msg, "category": "conversation"})
elif "retrieved" in msg.get("tags", []):
categories["retrieved_document"].append({**msg, "category": "retrieved_document"})
else:
categories["other"].append({**msg, "category": "other"})
return categories
def summarize_content(content: str, category: str, max_length: int = 500) -> str:
"""
Summarize content for compaction, dispatching by category.
Use when: compacting context and needing category-aware summarization
(tool outputs get metric extraction, conversations get decision
extraction, documents get lead-paragraph extraction).
"""
if category == "tool_output":
return summarize_tool_output(content, max_length)
elif category == "conversation":
return summarize_conversation(content, max_length)
elif category == "retrieved_document":
return summarize_document(content, max_length)
else:
return summarize_general(content, max_length)
def summarize_tool_output(content: str, max_length: int = 500) -> str:
"""
Summarize tool output by extracting metrics and key findings.
Use when: a tool output has served its immediate purpose and needs
to be compacted while preserving actionable data points.
"""
# Look for metrics (numbers with context)
metrics = re.findall(r'(\w+):\s*([\d.,]+)', content)
# Look for key findings (lines with important keywords)
keywords = ["result", "found", "total", "success", "error", "value"]
findings = []
for line in content.split('\n'):
if any(kw in line.lower() for kw in keywords):
findings.append(line.strip())
summary_parts = []
if metrics:
summary_parts.append(f"Metrics: {', '.join([f'{k}={v}' for k, v in metrics])}")
if findings:
summary_parts.append("Key findings: " + "; ".join(findings[:3]))
result = " | ".join(summary_parts) if summary_parts else "[Tool output summarized]"
return result[:max_length]
def summarize_conversation(content: str, max_length: int = 500) -> str:
"""
Summarize conversational content by extracting decisions and questions.
Use when: older conversation turns need compaction and the key
decisions/commitments must survive while filler is removed.
"""
decisions = re.findall(r'(?i)(?:decided|decision|chose|chosen)[:\s]+([^.]+)', content)
questions = re.findall(r'(?:\?|question)[:\s]+([^.]+)', content)
summary_parts = []
if decisions:
decision_texts = [d.strip() for d in decisions[:5]]
summary_parts.append(f"Decisions: {'; '.join(decision_texts)}")
if questions:
question_texts = [q.strip() for q in questions[:3]]
summary_parts.append(f"Open questions: {'; '.join(question_texts)}")
if not summary_parts:
# Fallback: extract the first few substantive sentences
sentences = [s.strip() for s in content.split('.') if len(s.strip()) > 20]
if sentences:
summary_parts.append('. '.join(sentences[:3]) + '.')
result = " | ".join(summary_parts) if summary_parts else "[Conversation summarized]"
return result[:max_length]
def summarize_document(content: str, max_length: int = 500) -> str:
"""
Summarize document content using lead-paragraph extraction.
Use when: a retrieved document has been consumed for reasoning and
only a brief reference needs to remain in context.
"""
paragraphs = content.split('\n\n')
if paragraphs:
first_para = paragraphs[0].strip()
sentences = first_para.split('. ')
if len(sentences) > 2:
first_para = '. '.join(sentences[:2]) + '.'
return first_para[:max_length]
return "[Document summarized]"
def summarize_general(content: str, max_length: int = 500) -> str:
"""
General-purpose summarization via truncation.
Use when: content does not fit a specific category and a simple
truncation with ellipsis is acceptable.
"""
return content[:max_length] + "..." if len(content) > max_length else content
# ---------------------------------------------------------------------------
# Observation masking
# ---------------------------------------------------------------------------
class ObservationStore:
"""
Store and mask verbose tool outputs with retrievable references.
Use when: tool outputs dominate context (>50% of tokens) and older
observations have already served their reasoning purpose. Stores the
full content externally and replaces it with a compact reference
containing a key-point summary.
Example::
store = ObservationStore(max_size=500)
masked, ref_id = store.mask(long_tool_output, max_length=200)
# masked: "[Obs:a1b2c3d4 elided. Key: ... Full content retrievable.]"
# Later retrieval:
original = store.retrieve(ref_id)
"""
def __init__(self, max_size: int = 1000) -> None:
self.observations: Dict[str, Dict] = {}
self.order: List[str] = []
self.max_size = max_size
def store(self, content: str, metadata: Optional[Dict] = None) -> str:
"""Store observation and return reference ID."""
ref_id = self._generate_ref_id(content)
self.observations[ref_id] = {
"content": content,
"metadata": metadata or {},
"stored_at": time.time(),
"last_accessed": time.time(),
}
self.order.append(ref_id)
# Evict oldest if over limit
if len(self.order) > self.max_size:
oldest = self.order.pop(0)
del self.observations[oldest]
return ref_id
def retrieve(self, ref_id: str) -> Optional[str]:
"""Retrieve observation by reference ID."""
if ref_id in self.observations:
self.observations[ref_id]["last_accessed"] = time.time()
return self.observations[ref_id]["content"]
return None
def mask(self, content: str, max_length: int = 200) -> Tuple[str, Optional[str]]:
"""
Mask observation if longer than max_length.
Use when: deciding per-observation whether to keep inline or
replace with a compact reference. Returns (masked_content, ref_id)
where ref_id is None if the content was short enough to keep.
"""
if len(content) <= max_length:
return content, None
ref_id = self.store(content)
key_point = self._extract_key_point(content)
masked = f"[Obs:{ref_id} elided. Key: {key_point}. Full content retrievable.]"
return masked, ref_id
def _generate_ref_id(self, content: str) -> str:
"""Generate unique reference ID."""
hash_input = f"{content[:100]}{time.time()}"
return hashlib.md5(hash_input.encode()).hexdigest()[:8]
def _extract_key_point(self, content: str) -> str:
"""Extract key point from observation."""
lines = [line for line in content.split('\n') if len(line) > 20]
if lines:
return lines[0][:50] + "..."
sentences = content.split('. ')
if sentences:
return sentences[0][:50] + "..."
return content[:50] + "..."
# ---------------------------------------------------------------------------
# Context budget management
# ---------------------------------------------------------------------------
class ContextBudget:
"""
Token budget allocation and optimization trigger detection.
Use when: building an agent loop that needs to monitor context usage
across categories and trigger compaction/masking at the right thresholds
rather than waiting until the window overflows.
Example::
budget = ContextBudget(total_limit=128_000)
budget.allocate("system_prompt", 1500)
budget.allocate("tool_definitions", 3000)
# ... after each agent turn:
should_act, reasons = budget.should_optimize(current_usage)
if should_act:
# apply masking or compaction based on reasons
pass
"""
def __init__(self, total_limit: int) -> None:
self.total_limit = total_limit
self.allocated: Dict[str, int] = {
"system_prompt": 0,
"tool_definitions": 0,
"retrieved_docs": 0,
"message_history": 0,
"tool_outputs": 0,
"other": 0,
}
self.reserved = 5000 # Reserved buffer
self.reservation_limit = total_limit - self.reserved
def allocate(self, category: str, amount: int) -> bool:
"""
Allocate budget to category. Returns True on success, False if
the allocation would exceed the reservation limit.
"""
if category not in self.allocated:
category = "other"
current = sum(self.allocated.values())
proposed = current + amount
if proposed > self.reservation_limit:
return False
self.allocated[category] += amount
return True
def remaining(self) -> int:
"""Get remaining unallocated budget."""
current = sum(self.allocated.values())
return self.reservation_limit - current
def get_usage(self) -> Dict[str, object]:
"""
Get current usage breakdown.
Use when: logging or displaying context budget state for
monitoring dashboards or debug output.
"""
total = sum(self.allocated.values())
return {
"total_used": total,
"total_limit": self.total_limit,
"remaining": self.remaining(),
"by_category": dict(self.allocated),
"utilization_ratio": total / self.total_limit,
}
def should_optimize(
self, current_usage: int, metrics: Optional[Dict[str, float]] = None
) -> Tuple[bool, List[Tuple[str, object]]]:
"""
Determine if optimization should trigger.
Use when: called at the end of each agent loop iteration to
decide whether to apply compaction, masking, or both before
the next model call.
Returns (should_optimize, list_of_reasons).
"""
reasons: List[Tuple[str, object]] = []
# Check utilization
utilization = current_usage / self.total_limit
if utilization > 0.8:
reasons.append(("high_utilization", utilization))
# Check degradation metrics if provided
if metrics:
if metrics.get("attention_degradation", 0) > 0.3:
reasons.append(("attention_degradation", True))
if metrics.get("quality_score", 1.0) < 0.8:
reasons.append(("quality_degradation", True))
return len(reasons) > 0, reasons
# ---------------------------------------------------------------------------
# Cache optimization
# ---------------------------------------------------------------------------
def design_stable_prompt(template: str, dynamic_values: Optional[Dict] = None) -> str:
"""
Stabilize a prompt template for maximum KV-cache hit rate.
Use when: constructing system prompts or few-shot prefixes that will
be reused across many requests. Replaces dynamic content (timestamps,
session IDs, counters) with stable placeholders so the prefix hash
remains constant.
"""
result = template
# Replace timestamps
date_pattern = r'\d{4}-\d{2}-\d{2}'
result = re.sub(date_pattern, '[DATE_STABLE]', result)
# Replace session IDs
session_pattern = r'Session \d+'
result = re.sub(session_pattern, 'Session [STABLE]', result)
# Replace counters
counter_pattern = r'\d+/\d+'
result = re.sub(counter_pattern, '[COUNTER_STABLE]', result)
return result
def calculate_cache_metrics(
requests: List[Dict], cache: Dict[str, Dict]
) -> Dict[str, object]:
"""
Calculate KV-cache hit metrics for a request sequence.
Use when: evaluating whether prompt restructuring improved cache
utilization. Feed in the request log and current cache state to
get hit/miss rates and actionable recommendations.
"""
hits = 0
misses = 0
for req in requests:
prefix = req.get("prefix_hash", "")
token_count = req.get("token_count", 0)
if prefix in cache:
hits += token_count * cache[prefix].get("hit_ratio", 0)
else:
misses += token_count
total = hits + misses
return {
"hit_rate": hits / total if total > 0 else 0,
"cache_hits": hits,
"cache_misses": misses,
"recommendations": generate_cache_recommendations(hits, misses),
}
def generate_cache_recommendations(hits: int, misses: int) -> List[str]:
"""
Generate recommendations for cache optimization based on hit/miss ratio.
Use when: cache metrics indicate sub-optimal hit rates and concrete
next steps are needed.
"""
recommendations: List[str] = []
hit_rate = hits / (hits + misses) if (hits + misses) > 0 else 0
if hit_rate < 0.5:
recommendations.append("Consider stabilizing system prompts")
recommendations.append("Reduce variation in request prefixes")
if hit_rate < 0.8:
recommendations.append("Group similar requests together")
recommendations.append("Use consistent formatting across requests")
return recommendations
# ---------------------------------------------------------------------------
# Demo / smoke test
# ---------------------------------------------------------------------------
if __name__ == "__main__":
print("=== Context Optimization Utilities — Demo ===\n")
# 1. Token estimation
sample_text = "The quick brown fox jumps over the lazy dog. " * 20
tokens = estimate_token_count(sample_text)
print(f"1. Token estimate for {len(sample_text)}-char text: ~{tokens} tokens\n")
# 2. Observation masking
store = ObservationStore(max_size=100)
long_output = (
"Result: 42 items found\n"
"Total processing time: 3.2s\n"
"Details:\n" + "\n".join([f" Item {i}: value={i*10}" for i in range(20)])
)
masked, ref_id = store.mask(long_output, max_length=100)
print(f"2. Masked observation:\n {masked}")
print(f" Ref ID: {ref_id}")
retrieved = store.retrieve(ref_id)
print(f" Retrievable: {retrieved is not None}\n")
# 3. Context budget
budget = ContextBudget(total_limit=128_000)
budget.allocate("system_prompt", 1500)
budget.allocate("tool_definitions", 3000)
budget.allocate("message_history", 95_000)
usage = budget.get_usage()
print(f"3. Budget utilization: {usage['utilization_ratio']:.1%}")
should_opt, reasons = budget.should_optimize(
current_usage=int(128_000 * 0.85)
)
print(f" Should optimize: {should_opt}, reasons: {reasons}\n")
# 4. Cache-stable prompt
raw_prompt = "Session 42 started on 2025-12-20. Progress: 3/10 tasks."
stable = design_stable_prompt(raw_prompt)
print(f"4. Original prompt: {raw_prompt}")
print(f" Stabilized: {stable}\n")
# 5. Summarization
tool_out = "count: 150\nstatus: success\nFound 3 errors in module A."
summary = summarize_content(tool_out, "tool_output", max_length=200)
print(f"5. Tool output summary: {summary}\n")
print("=== Demo complete ===")