Published2026-06-04
TopicAgent · 源码
Series每日 GitHub 解析 · 一
Size4272 chars · 14 code blocks

每日github项目解析：（一）20260604

每日github项目解析：（一）20260604

headroom

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是闭源的，这是怎么做到的呢？其实是开了个代理服务器

  flowchart LR
      CC["Claude Code<br/>(闭源, 不动它)"]
      HR["Headroom 本地代理<br/>127.0.0.1:8787<br/>压缩 + 缓存对齐"]
      API["api.anthropic.com<br/>(真正的大模型)"]
      CC -->|"① 请求 (被 ANTHROPIC_BASE_URL 重定向)"| HR
      HR -->|"② 压缩后转发"| API
      API -->|"③ 回复"| HR
      HR -->|"④ 透传回去"| CC

Claude Code 本身就支持一个官方环境变量 ANTHROPIC_BASE_URL,让你自定义 API 地址(原本是给企业代理/网关用的)。

Headroom 就是钻这个官方留的口子——把这个地址改成本地的127.0.0.1:8787。

于是 CC 以为自己在跟 Anthropic 说话,其实先经过了 Headroom。

Headroom 把请求里的长内容压缩掉,再原样转发给真正的 Anthropic。

它的结果强调的是压缩结构化、重复性高的内容(JSON、搜索结果、日志)压得越狠(90%+)，而信息密度高的散文/源码,压得越少(40%多)，不算吹，还算是良心。

本地运行，用rust，针对性压缩，数据很华丽，但是具体如何还是要自己做。

  flowchart TD
      IN["一段待压内容"] --> R{ContentRouter<br/>判断类型}
      R -->|"JSON / 数组 / 表格型"| SC["SmartCrusher<br/>纯 Rust 算法"]
      R -->|"源代码"| CC["CodeCompressor<br/>tree-sitter AST"]
      R -->|"散文 / 自然语言"| KP["Kompress-base<br/>本地小模型 ModernBERT"]
      SC --> OUT["压缩结果"]
      CC --> OUT
      KP --> OUT

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 是"打分/分类"的,本地小模型,毫秒级,不联网,免费。

Scrapling

https://github.com/D4Vinci/Scrapling

Scrapling 是一个 Python 的「反检测 + 自适应」爬虫框架。

flowchart LR
    U["你 / AI 助手"] --> CFG["注入解析配置<br/>adaptive·storage·huge_tree"]
    CFG --> PICK{"选哪一档抓取器?"}
    PICK -->|"纯 HTTP·最快"| E1["static 引擎<br/>curl_cffi"]
    PICK -->|"要跑 JS"| E2["DynamicFetcher<br/>Playwright 浏览器"]
    PICK -->|"要过 WAF"| E3["StealthyFetcher<br/>隐身浏览器"]
    E1 --> RF["ResponseFactory<br/>统一适配"]
    E2 --> RF
    E3 --> RF
    RF --> R["Response<br/>= HTTP响应 + Selector"]
    R --> SEL["css / xpath / re<br/>find_similar / find_by_text"]
    SEL --> DATA["结构化数据"]
    R -.->|"yield Request"| SP["Spider 框架<br/>并发·去重·续爬"]
    SP -.-> DATA

按照洋葱分层来说，应该是这样的：

flowchart TB
    L1["门面层 Facade<br/>scrapling/__init__.py·fetchers/*<br/>懒加载，只暴露 Selector + 4 个 Fetcher"]
    L2["抓取引擎层 Fetch Engines<br/>engines/static.py·engines/_browsers/*<br/>真正联网、渲染、反检测"]
    L3["适配/工具带层 Toolbelt<br/>convertor·proxy_rotation·fingerprints·ad_domains<br/>把异构响应归一 + 跨引擎复用能力"]
    L4["解析层 Parsing<br/>parser.py·core/translator·custom_types·storage<br/>选取、CSS↔XPath、AutoMatch 重定位"]
    L5["爬虫编排层 Crawling<br/>spiders/*<br/>async 调度·去重·限流·断点续爬"]
    L6["周边集成层 Peripheral<br/>cli.py·core/shell.py·core/ai.py<br/>命令行·交互式 Shell·MCP for AI"]
    L1 --> L2 --> L3 --> L4
    L4 --> L5
    L1 --> L6
    L6 -.-> L2

我对爬虫一直好奇的有两个问题：

首先，怎么把一万个请求跑得又快又不被封？

最朴素也最划算的优化：别每个请求都重新建连接，会话复用。

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 一个

这里有两个层次：

  1. CapacityLimiter（容量限流器） 是 async 原语——满了就挂起协程（而不是阻塞线程），全局一道、每个域名再来一道，既能跑满又能对单个站点礼貌。
  2. 背压（backpressure）：主循环不是一次性把上万个任务全 start_soon 出去，而是在飞任务满了就稍等，按需取队列、按需 spawn。

还有指纹浏览器和中间过程的保存就不说了，感觉上面的重要一点。

其次，脚本元素失效了怎么救回来？

AutoMatch 自适应——这是 Scrapling 区别于一切传统解析库的核心。分「记忆」和「回忆」两步

第 1 步「记忆」：把元素拍成指纹存进 SQLite

开启 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 一致，就能取回同一个元素的指纹。

第 2 步「回忆」：relocate 用相似度打分认人

旧选择器失效后，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 的 token

extraction_type 决定喂给模型的格式，css_selector 让 AI 能精准截取页面某一块，main_content_only 则只保留正文。

整个「记忆 → 回忆」闭环画出来是这样：

sequenceDiagram
    participant U as 你的代码
    participant S as Selector
    participant DB as SQLite 指纹库
    Note over U,DB: 第一次抓取(网站旧版)
    U->>S: css('#price', identifier='price', auto_save=True, adaptive=True)
    S->>S: xpath 命中元素
    S->>DB: save(指纹) 存档 tag/属性/文本/路径/父/兄弟
    Note over U,DB: 几周后(网站已改版, #price 失效)
    U->>S: css('#price', identifier='price', adaptive=True)
    S->>S: xpath 命中为空!
    S->>DB: retrieve('price') 取回旧指纹
    DB-->>S: element_data
    S->>S: relocate: 遍历全页, 每个元素打相似度分
    S-->>U: 返回最高分且≥40% 的元素 = 找回成功