https://github.com/chopratejas/headroom
官方的说法是它是个给 AI agent 省 token 的"压缩中间层"。
AI agent(比如 Claude Code)干活时,要把一大堆东西塞给大模型读——工具输出、日志、报错、检索结果、文件内容、聊天历史。
这些东西又臭又长,烧token、烧钱、还容易把上下文撑爆。
Headroom 在这些内容到达大模型之前先压缩一遍,号称答案不变、token 砍掉 60–95%。
怎么做到的呢？
它是把本来要发给 api.anthropic.com 的那个 HTTP 请求体,在半路被拿下来压一遍。
你给 agent 的 prompt
工具执行结果(比如一次代码搜索返回 100 条结果)
日志、报错堆栈
RAG 检索出来的文档片段
读进来的源码文件
之前的对话历史
问题是CC是闭源的，这是怎么做到的呢？其实是开了个代理服务器
Claude Code 本身就支持一个官方环境变量 ANTHROPIC_BASE_URL,让你自定义 API 地址(原本是给企业代理/网关用的)。
Headroom 就是钻这个官方留的口子——把这个地址改成本地的127.0.0.1:8787。
于是 CC 以为自己在跟 Anthropic 说话,其实先经过了 Headroom。
Headroom 把请求里的长内容压缩掉,再原样转发给真正的 Anthropic。
它的结果强调的是压缩结构化、重复性高的内容(JSON、搜索结果、日志)压得越狠(90%+)，而信息密度高的散文/源码,压得越少(40%多)，不算吹，还算是良心。
本地运行，用rust，针对性压缩，数据很华丽，但是具体如何还是要自己做。
SmartCrusher是出场率最高的(代码搜索 92%、日志 92% 的省幅基本靠它)。它根本不"理解"内容,纯粹做结构统计:
一个 JSON 数组里 100 条记录,很多字段是重复的/同模式的，它做去重、保留有代表性的"变化点"行、丢掉冗余行
被丢掉的行不是删除,而是换成一个标记 <<ccr:哈希 100_rows_offloaded>>,原文存在本地,大模型要看再用工具捞回来(这就是它说的"可逆 CCR")。
CodeCompressor用 tree-sitter 把代码解析成 AST(抽象语法树),然后
保留 import、函数签名、类型标注、错误处理(这些是骨架,删了就读不懂);
只压缩函数体内部;
保证输出仍是合法能解析的代码(syntax_valid=True)。
Kompress-base是唯一带 ML 的，它用的是 ModernBERT(一个几百 MB 的 encoder/编码器模型)，通过 ONNX 或 PyTorch 在你本机 CPU 上跑推理,判断每个 token 该不该保留。
ModernBERT 是"打分/分类"的,本地小模型,毫秒级,不联网,免费。
https://github.com/D4Vinci/Scrapling
Scrapling 是一个 Python 的「反检测 + 自适应」爬虫框架。
按照洋葱分层来说，应该是这样的：
我对爬虫一直好奇的有两个问题：
首先，怎么把一万个请求跑得又快又不被封？
最朴素也最划算的优化：别每个请求都重新建连接，会话复用。
SessionManager（会话管理器）在整个爬取开始时，一次性把会话curl_cffi 的 AsyncSession）建好并保活，所有请求复用同一个连接池；用不到的会话还能「懒启动」——第一次真正用到某个 sid（session id）时才连。
# scrapling/spiders/session.py:83
async def start(self):
if self._started:
return
for sid, session in self._sessions.items():
if sid not in self._lazy_sessions and not session._is_alive:
await session.__aenter__() # 进场即建长连接
self._started = True
# 懒会话：用到才连
if sid in self._lazy_sessions and not session._is_alive:
async with self._lazy_lock: # 双重检查，防并发重复启动
if not session._is_alive:
await session.__aenter__()Scheduler（调度器）做两件事：排序和去重。先爬重要的，重复的根本不发。
排序：底层是 asyncio.PriorityQueue，元素是 (-priority, counter, request) 三元组。负优先级让数值大的先出；counter（自增计数）既做先进先出的「平局打破器」，又避免直接比较 Request 对象。
去重：入队那一刻就用 SHA1 指纹判重，重复 URL 直接丢弃，连网络请求都不会发起。
# scrapling/spiders/scheduler.py:30
async def enqueue(self, request):
fingerprint = request.update_fingerprint(...) # canonicalize_url + 排序后 sha1
if not request.dont_filter and fingerprint in self._seen:
log.debug("Dropped duplicate request: %s", request)
return False # 重复，丢弃
self._seen.add(fingerprint)
item = (-request.priority, next(self._counter), request)
await self._queue.put(item)光排好队还不够，得控制「同一时刻有多少请求在飞」。
Scrapling 用两道闸门把带宽跑满又不爆内存：
# scrapling/spiders/engine.py:63 / 125 / 394
self._global_limiter = CapacityLimiter(spider.concurrent_requests) # 全局总闸
def _rate_limiter(self, domain): # 每域名分闸
if self.spider.concurrent_requests_per_domain:
self._domain_limiters.setdefault(domain, CapacityLimiter(self.spider.concurrent_requests_per_domain))
return self._domain_limiters[domain]
return self._global_limiter
# 主循环里的第二道闸门：背压
if self._active_tasks >= self.spider.concurrent_requests:
await anyio.sleep(0.01); continue # 在飞任务满了，先别再 spawn
request = await self.scheduler.dequeue()
self._active_tasks += 1
tg.start_soon(self._task_wrapper, request) # 按需取一个、spawn 一个这里有两个层次：
CapacityLimiter（容量限流器） 是 async 原语——满了就挂起协程（而不是阻塞线程），全局一道、每个域名再来一道，既能跑满又能对单个站点礼貌。
背压（backpressure）：主循环不是一次性把上万个任务全 start_soon 出去，而是在飞任务满了就稍等，按需取队列、按需 spawn。
还有指纹浏览器和中间过程的保存就不说了，感觉上面的重要一点。
其次，脚本元素失效了怎么救回来？
AutoMatch 自适应——这是 Scrapling 区别于一切传统解析库的核心。分「记忆」和「回忆」两步
开启 adaptive=True 且 auto_save=True 时，css() 命中元素后会顺手把它存档：
# scrapling/parser.py:568（css 的核心分支）
if elements := self._root.xpath(selector, **kwargs):
if self.__adaptive_enabled and auto_save:
self.save(elements[0], identifier or selector) # 命中→存指纹
return self.__handle_elements(elements)
elif self.__adaptive_enabled:
if adaptive: # 没命中且开了 adaptive
element_data = self.retrieve(identifier or selector) # 取出旧指纹
if element_data:
elements = self.relocate(element_data, percentage) # 重新定位!
return self.__handle_elements(elements)这段代码就是整个主题的总开关与分发点：命中走快路径并存档；没命中且开了 adaptive，就触发回退——取出指纹、重新定位。
「指纹」由 element_to_dict 生成，记录的线索相当全：
# scrapling/core/utils/_utils.py:83
result = {
"tag": str(element.tag),
"attributes": cls.__clean_attributes(element),
"text": element.text.strip() if element.text else None,
"path": cls._get_element_path(element), # 从根到它的标签路径
}
if parent is not None:
result.update({"parent_name": parent.tag,
"parent_attribs": dict(parent.attrib),
"parent_text": parent.text.strip() if parent.text else None})
siblings = [c.tag for c in parent.iterchildren() if c != element]存储用 SQLiteStorageSystem，按 (网站主域名, identifier) 唯一索引，WAL 模式 + RLock 做到线程安全：
# scrapling/core/storage.py:97
CREATE TABLE IF NOT EXISTS storage (
id INTEGER PRIMARY KEY, url TEXT, identifier TEXT, element_data TEXT,
UNIQUE (url, identifier) -- 同站同标识只存一份
)注意 identifier（标识符）是存/取的主键：哪怕你日后换了一个完全不同的选择器字符串，只要 identifier 一致，就能取回同一个元素的指纹。
旧选择器失效后，relocate 遍历全页每个元素，逐个和指纹打分，取最高分且过阈值者：
# scrapling/parser.py:519
for node in _find_all_elements(self._root): # 遍历全页所有元素
score = self.__calculate_similarity_score(element, node)
score_table.setdefault(score, []).append(node) # 不提前停，允许并列
highest = max(score_table.keys())
if highest >= percentage: # 默认阈值 40%
return score_table[highest]打分函数是算法的心脏——它不靠单一线索，而是把标签、文本、class/id/href/src、标签路径、父节点、兄弟节点全部用 SequenceMatcher 求相似比，加权平均成一个百分数：
# scrapling/parser.py:807
score += 1 if original["tag"] == data["tag"] else 0 # 标签
if original["text"]:
score += SequenceMatcher(None, original["text"], data.get("text") or "").ratio() # 文本
for attrib in ("class", "id", "href", "src"): # 关键属性各算一项
if original["attributes"].get(attrib):
score += SequenceMatcher(None, original["attributes"][attrib],
data["attributes"].get(attrib) or "").ratio()
score += SequenceMatcher(None, original["path"], data["path"]).ratio() # 结构路径
return round((score / checks) * 100, 2)最近的更新也把抓取/抽取封装成 MCP 工具，其中 get 工具的参数直接复用了选取能力：
# AI 端可调用的工具(节选)
get(url, extraction_type='markdown', # 返回 markdown / html / text
css_selector='article', # 复用 CSS 选取做精准截取
main_content_only=True) # 只取正文, 省 AI 的 tokenextraction_type 决定喂给模型的格式，css_selector 让 AI 能精准截取页面某一块，main_content_only 则只保留正文。
整个「记忆 → 回忆」闭环画出来是这样：