Headroom 通用压缩（Universal Compression）详解：ML 内容检测、结构掩码与可逆压缩实现

Headroom 通用压缩Universal Compression详解ML 内容检测、结构掩码与可逆压缩实现【免费下载链接】headroomCompress tool outputs, logs, files, and RAG chunks before they reach the LLM. 20% fewer tokens for coding agents, 60-95% fewer tokens for JSON, same answers. Library, proxy, MCP server.项目地址: https://gitcode.com/GitHub_Trending/head/headroom本篇技术指南基于 Headroom 的 Universal Compression 模块文档wiki/compression.md系统讲解该模块的四大核心机制——Magika ML 内容检测、结构掩码Structure Mask保护、Kompress 智能压缩与 CCR 可逆检索并结合 headroom/compression/ 目录下的源码实现梳理完整调用链、全部配置参数、JSON 与代码处理器的保留规则以及降级回退策略。读完后你将掌握如何在代码中一行调用compress()、如何用UniversalCompressorConfig调优压缩目标、如何注册自定义结构处理器以及压缩结果对象中每个字段的准确含义。一、模块定位四大技术组合Headroom 的 Universal Compression 模块提供智能、自动的压缩能力。它并非单一算法而是把四种技术组合成一条流水线见 模块入口ML-based Detection使用 Magika 深度学习模型自动识别内容类型JSON、代码、日志、文本无需配置文件扩展名或脆弱正则Structure Preservation通过结构掩码structure masks保留键、函数签名、模板等导航性内容让 LLM 在值被压缩后仍能看到完整 schemaIntelligent Compression对非结构性内容使用可选的 ML 压缩器 Kompress 进行有损压缩Reversible via CCR把原文存入 CCRCompress-Cache-Retrieve存储当 LLM 需要完整上下文时可凭ccr_key检索原文实现可逆压缩。模块 docstringuniversal.py明确了这条流水线检测内容类型Magika→ 提取结构handler→ 压缩非结构内容Kompress→ 可选存入 CCR。二、快速开始一行式调用最简单的用法是compress()便捷函数内部创建一个默认配置的UniversalCompressorfrom headroom.compression import compress result compress(content) print(result.compressed) print(fSaved {result.savings_percentage:.0f}% tokens)带配置调用from headroom.compression import UniversalCompressor, UniversalCompressorConfig config UniversalCompressorConfig( compression_ratio_target0.5, # 保留 50% 内容 use_entropy_preservationTrue, # 保留 UUID、哈希等高熵内容 ) compressor UniversalCompressor(configconfig) result compressor.compress(content)注意compress()便捷函数会把**kwargs透传给UniversalCompressor.compress()因此可以直接传入content_type覆盖检测器结果见 universal.py。三、压缩主流程源码级调用链UniversalCompressor.compress()universal.py的执行顺序与文档中的检测流程图一一对应┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ Content │───│ Detect │───│ Extract │───│ Compress │ │ Input │ │ Type │ │ Structure │ │ Content │ └─────────────┘ └─────────────┘ └─────────────┘ └─────────────┘ │ │ │ ▼ ▼ ▼ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ Magika │ │ Handler │ │ Kompress │ │ (ML) │ │ (JSON, │ │ (ML, opt- │ │ │ │ Code...) │ │ in [ml]) │ └─────────────┘ └─────────────┘ └─────────────┘源码层面的关键步骤短内容短路内容为空或长度小于min_content_length默认 100 字符时直接原样返回metadata标记{skipped: content too short}preservation_ratio为 1.0universal.py类型检测未显式传入content_type时调用self._detector.detect(content)显式传入则构造confidence1.0的DetectionResult跳过 ML 检测结构掩码提取按检测到的ContentType从 handler 表取处理器JSON →JSONStructureHandlerCODE →CodeStructureHandler其余 →NoOpHandler把内容做字符级分词tokens list(content)后调用handler.get_mask(content, tokens)得到StructureMask熵保留叠加当use_entropy_preservationTrue时调用compute_entropy_mask_for_content()计算高熵词掩码并与结构掩码做union()——任一掩码标记保留即保留universal.py按掩码分段压缩_compress_with_mask()把掩码切成连续 span结构性 span 原样保留非结构性 span 仅在长度超过 50 字符时才交给压缩函数过短的片段压缩收益太低直接保留universal.pyCCR 存储ccr_enabledTrue时懒加载headroom.cache.compression_store.CompressionStore存入原文与压缩文返回ccr_key。一个值得注意的实现细节token 估算采用约 4 字符 1 token的启发式len(text) // 4见 universal.py因此tokens_before/after是估算值而非精确 tokenizer 计数。压缩函数与降级策略构造函数中compress_fn参数允许注入自定义压缩函数默认路径的选择逻辑_get_default_compress_fnuniversal.pyuse_kompressTrue时优先返回 Kompress 包装器懒加载headroom.transforms.kompress_compressor.KompressCompressor。Kompress 未安装ImportError或压缩抛异常时都会自动降级到简单压缩_simple_compress简单压缩_simple_compress按compression_ratio_target计算目标长度保留前 2/3 与后 1/3中间插入分隔符 ...[compressed]... 。源码注释特别说明该分隔符不能含控制字符RFC 8259 §7——因为这个回退路径可能作用于 JSON 字符串值内部的片段裸换行会产生非法 JSONuniversal.py。这种ML 缺失自动回退的设计意味着即使只装了基础包无 magika、无 Kompress、无 tree-sittercompress()依然可用——检测走启发式回退压缩走截断回退。四、ML 内容检测Magika 与回退检测器检测逻辑在 headroom/compression/detector.py。ContentType 枚举检测结果是 7 个高层类别detector.pyJSON、CODE、LOG、DIFF、MARKDOWN、TEXT、UNKNOWN。MagikaDetectorMagikaDetector封装 Google Magika 深度模型单例懒加载首次使用时才Magika()避免不必要的模型加载开销detect()对content.encode(utf-8)调用identify_bytes()取原始 label 与置信度经_map_label()映射到ContentType置信度低于min_confidence默认 0.5时降级为UNKNOWN。label 映射是唯一的标签映射点源码注释强调 no hardcoding elsewheredetector.py代码类约 38 种语言标签python、javascript、typescript、go、rust、java、c、cpp、ruby、shell、sql 等→ContentType.CODE并保留language字段如python供代码处理器选 parser结构化数据json/jsonl/yaml/toml/xml/html/csv/tsv/ini/properties一律 →ContentType.JSON由 JSON 处理器做类 JSON 的结构化处理日志log、syslog→LOG文档markdown/rst/asciidoc/org→MARKDOWNdiff→DIFFtxt/text/ascii/utf8/empty及无法识别的标签 →TEXT默认按文本处理。FallbackDetectoruse_magikaFalse或未安装 magika 时get_detector(prefer_magikaTrue)自动探测使用零依赖的FallbackDetector启发式规则detector.py判定顺序规则结果置信度1以{或[开头且json.loads成功JSON1.02含def、class、function、import、const、fn、package等代码指示词CODE0.73含ERROR、WARN、INFO、DEBUG、FATAL等日志指示词LOG0.64其余TEXT0.5五、结构掩码StructureMask 与熵保留掩码系统的实现在 headroom/compression/masks.py其设计思想模块 docstring是把结构检测handler 负责与内容压缩Kompress 负责解耦——掩码本身就是与 token 序列对齐的布尔数组True 保留结构性/导航性False 可压缩。StructureMask 核心 API构造时强校验len(tokens) len(mask)长度不一致直接ValueErrorpreservation_ratio被标记保留的 token 占比会出现在CompressionResult.preservation_ratio中union(other)/intersection(other)掩码合并。union用于叠加多种保留策略任一保留即保留intersection可用于更激进的压缩两者都保留才保留mask_to_spans()把布尔数组转成连续区间列表MaskSpan(start, end, is_structural)供按段差异化处理——_compress_with_mask()正是基于此实现结构段原样、非结构段压缩。熵保留一个精心设计的信号EntropyScore.compute()用归一化 Shannon 熵字符频率统计后除以该字母表的最大熵评分超过entropy_threshold默认 0.85即建议保留。保留高熵内容UUID、哈希、随机串的理由信息密集不可重建、常为语义重要的标识符、且 token 级压缩器容易把它们打碎。源码中有一个关键的工程修正masks.py主流程使用字符级 tokenlist(content)单字符 token 永远达不到最小长度门槛会让基于 token 的熵掩码在纯文本上静默失效。因此UniversalCompressor实际调用的是compute_entropy_mask_for_content()按空白切出整词、对长词评分、再把命中词的字符区间映射回字符级掩码。另一个细节是长度下限SECRET_ENTROPY_MIN_LENGTH 20masks.py归一化熵无法区分 40 字符的 API key 和 8 字符的高多样性英文单词如 detailed因此用长度下限补足信号——该值对齐了 trufflehog、detect-secrets 等密钥扫描器的熵检测下限避免普通散文被过度保留而破坏压缩率。各内容类型的保留/压缩规则总表内容类型保留压缩JSON键、括号、布尔、null、短值、UUID长字符串值、空白代码import、函数签名、类定义、类型标注函数体、注释日志时间戳、日志级别、错误消息重复模式、冗长细节文本高熵 tokenID、哈希低信息量内容六、JSON 处理器JSONStructureHandler实现见 headroom/compression/handlers/json_handler.py完整构造参数含默认值from headroom.compression.handlers.json_handler import JSONStructureHandler handler JSONStructureHandler( preserve_short_valuesTrue, # 保留短字符串值 short_value_threshold20, # 短的长度阈值按去引号后的值长度计 preserve_high_entropyTrue, # 保留 UUID、哈希 entropy_threshold0.85, # 熵阈值 max_array_items_full3, # 数组前 N 项完整保留 max_number_digits10, # 不超过 N 位的数字保留常为 ID )判定规则对照源码_should_preserve_token永远保留所有键导航性——LLM 由此看到 schema、结构语法{ } [ ] : ,、布尔值、null见 JSONTokenType数字位数 ≤max_number_digits10时保留因为它们常是 ID 或关键数值字符串值分三层判定json_handler.py所在数组的项索引 ≥max_array_items_full时不再保留——大数组中更激进地压缩数组项计数只在数组自身的逗号上累加嵌套对象内的逗号不会误计源码专门用container_stack/array_item_stack处理该边界去引号后的值长度 ≤short_value_threshold20且preserve_short_valuesTrue时保留注释说明去引号是为了让20 字符阈值真正作用于值本身而非含引号的 tokenpreserve_high_entropyTrue且值中无空格最廉价的标识符信号避免把英文散文误判成 UUID且熵达标时保留。空白永远不保留。can_handle()通过json.loads校验内容是否为合法 JSON处理器还附带extract_json_schema()工具函数可抽取仅键 类型的 schemajson_handler.py。效果示例继承自原文档# Before {id: usr_abc123, name: Alice Johnson, bio: A long description that goes on and on...} # After结构保留长值被压缩 {id: usr_abc123, name: Alice Johnson, bio: A long...[compressed]...}七、代码处理器CodeStructureHandler实现见 headroom/compression/handlers/code_handler.pyfrom headroom.compression.handlers.code_handler import CodeStructureHandler handler CodeStructureHandler( preserve_commentsFalse, # 是否把注释视为结构性内容 use_tree_sitterTrue, # 优先使用 tree-sitter AST 解析 default_languagepython, # 检测失败时的默认语言 )保留import 语句、函数/方法签名、类定义、类型标注、装饰器压缩函数体实现细节、注释除非preserve_commentsTrue。# Before def process_data(items: List[str]) - Dict[str, int]: Process items and count occurrences. result {} for item in items: item item.strip().lower() if item in result: result[item] 1 else: result[item] 1 return result # After签名保留函数体被压缩 def process_data(items: List[str]) - Dict[str, int]: Process items and count occurrences. result {} for item in items: ...[compressed]...语言支持与工程细节文档给出的支持矩阵语言Parser支持级别Python / JavaScript / TypeScript / Go / Rust / Java / C / Ctree-sitter完整 AST源码中每个语言都定义了结构性 AST 节点类型白名单如 Python 的import_statement、function_definition、class_definition、decorated_definitionJavaScript 的function_declaration、class_declaration、arrow_function等见 code_handler.pytree-sitter 不可用时回退到正则模式匹配。从源码结构看还有两处稳健性设计线程局部 parsertree-sitter 的Parser对象是 pyo3 unsendable 的跨线程访问会 panic代理场景下 handler 运行在 executor 线程池上因此每个 (线程, 语言) 组合持有独立 parsercode_handler.py双 API 兼容层_ts_parse/_ts_kind/_ts_start_byte等 shim 同时兼容经典 pybind API.type、parse(bytes)与 tree-sitter-language-pack ≥1.0 的 Rust 绑定 APIkind()、parse(str)避免新版本下 AST 路径抛TypeError后静默退化到正则code_handler.py。_check_tree_sitter()还会在首次使用时实际执行一次最小解析把 ABI 不匹配问题提前暴露而不是等请求期才发现。八、完整配置参数速查UniversalCompressorConfig定义在 universal.py默认值与文档表格一致参数默认值说明use_magikaTrue使用 Magika ML 内容检测需要 magika 包否则自动回退到启发式检测器use_kompressTrue使用 Kompress 做内容压缩旧参数use_llmlingua已退役传入会抛TypeErrorcompression_ratio_target0.3目标保留比例0.3 保留 30%即 70% 缩减min_content_length100短于该字符数的内容直接跳过不压缩use_entropy_preservationTrue叠加高熵 tokenUUID、哈希、密钥类标识符保留掩码entropy_threshold0.85归一化熵保留阈值0.0–1.0越高越严格ccr_enabledTrue把原文存入 CCR 供后续检索配置示例继承自原文档use_kompress需安装headroom-ai[ml]才生效否则自动降级from headroom.compression import UniversalCompressorConfig config UniversalCompressorConfig( # 检测 use_magikaTrue, # ML 内容检测需要 magika # 压缩 compression_ratio_target0.3, # 保留 30%70% 缩减 min_content_length100, # 短于此长度跳过 # 结构保护 use_entropy_preservationTrue, # 保留高熵 token entropy_threshold0.85, # CCR ccr_enabledTrue, # 存入原文供检索 )九、CompressionResult 结果对象compress()/compressor.compress()返回CompressionResultuniversal.py完整字段from headroom.compression import compress result compress(content) # 压缩内容 print(result.compressed) # 压缩后内容 print(result.original) # 原始内容引用 print(result.compression_ratio) # 压缩后长度 / 原始长度如 0.35 print(result.tokens_before) # 压缩前估算 token 数 print(result.tokens_after) # 压缩后估算 token 数 print(result.tokens_saved) # tokens_before - tokens_after属性≥0 print(result.savings_percentage) # 节省百分比如 65.0 # 检测信息 print(result.content_type) # ContentType.JSON / CODE / TEXT ... print(result.detection_confidence) # 0.0-1.0 # 结构信息 print(result.handler_used) # json / code / noop print(result.preservation_ratio) # 被标记为结构性而保留的比例 # CCR print(result.ccr_key) # 检索用 keyccr_enabled 且存储成功时其中metadata还携带检测原始 label、代码语言metadata[detection][language]以及 handler 元数据如 JSON 的 token 数、key 数。十、批量压缩与自定义 Handler批量压缩compress_batch()对多内容压缩更高效若检测器支持detect_batchMagikaDetector 提供则一次性批量检测再把每项带已检测类型逐条压缩universal.pyfrom headroom.compression import UniversalCompressor compressor UniversalCompressor() contents [ {users: [...]}, def hello(): pass, Plain text content, ] results compressor.compress_batch(contents) for result in results: print(f{result.content_type}: {result.savings_percentage:.0f}% saved)注册自定义 HandlerHandler 遵循StructureHandler协议get_mask() 可选can_handle()见 handlers/base.py。继承BaseStructureHandler只需实现_extract_mask(content, tokens, **kwargs)并返回HandlerResult(mask, handler_name, confidence)即可通过register_handler()挂到任意ContentType上universal.pyfrom headroom.compression import UniversalCompressor from headroom.compression.detector import ContentType from headroom.compression.handlers.base import BaseStructureHandler, HandlerResult from headroom.compression.masks import StructureMask class LogStructureHandler(BaseStructureHandler): 自定义日志处理器。 def __init__(self): super().__init__(namelog) def can_handle(self, content: str) - bool: return [INFO] in content or [ERROR] in content def _extract_mask(self, content, tokens, **kwargs): # 把时间戳、日志级别标记为结构性 mask [False] * len(content) # ... (自定义标记逻辑) return HandlerResult( maskStructureMask(tokenstokens, maskmask), handler_nameself.name, confidence0.9, ) # 注册自定义 handler compressor UniversalCompressor() compressor.register_handler(ContentType.TEXT, LogStructureHandler())BaseStructureHandler.get_mask()的公共职责已内置空内容直接返回空掩码、未传 tokens 时用字符级分词兜底未注册任何 handler 的内容类型落到NoOpHandler——全部标记为可压缩。十一、CCR 集成可逆压缩开启ccr_enabledTrue后压缩成功的原文会被写入 CCR 存储_store_in_ccr懒加载headroom.cache.compression_store.CompressionStore并附带压缩前后的估算 token 数universal.py。CCR 不可用时仅记 debug 日志、ccr_key为None不影响压缩本身from headroom.compression import UniversalCompressor, UniversalCompressorConfig config UniversalCompressorConfig(ccr_enabledTrue) compressor UniversalCompressor(configconfig) result compressor.compress(large_content) if result.ccr_key: print(fOriginal stored with key: {result.ccr_key}) # LLM 需要完整上下文时可通过 CCR 检索原文完整的 CCRCompress-Cache-Retrieve架构文档见 wiki/ccr.md。十二、性能参考原文档给出的性能参考数据基于官方 wiki 文档具体数值会随硬件与内容变化内容类型压缩率速度准确性JSON大数组70–90%~1ms键完整保留代码Python50–70%~10ms签名完整保留纯文本60–80%~5ms高熵内容保留文档标注的额外开销约为每次压缩 1–10ms取决于内容大小与类型。十三、安装与依赖# 基础压缩无 ML 依赖时自动降级为启发式检测 简单压缩 pip install headroom-ai # 带 ML 内容检测推荐 pip install headroom-ai[magika] # 带 Kompress ML 压缩 pip install headroom-ai[ml] # 带 AST 代码处理 pip install headroom-ai[code] # 全部 pip install headroom-ai[all]与 pyproject.toml 对照需要注意两点[llmlingua]extra 已在 0.9.x 中移除源码注释明确no live code path used it使用[ml]代替——这正是UniversalCompressorConfig中use_llmlingua参数被use_kompress取代的原因[ml]实际包含torchmacOS Intel x86_64 上因 PyTorch 无对应 wheel 被排除、transformers5.5.0,6.0与huggingface-hub1.5.0,2.0后者防止同机其他包把版本拖低导致 Kompress 静默不可用[code]extra 的版本锁定tree-sitter-language-pack0.10.0,1.0tree-sitter0.25.2,0.27——因为 tree-sitter-language-pack ≥1.0 移除了内置 tree-sitter 并改用了不兼容的节点 APIissue #1216代码压缩器内部虽有 API 兼容 shim官方推荐仍按此锁定安装。另外magika0.6.0本身已包含在[proxy]extra 中。十四、完整流水线示例from headroom.compression import UniversalCompressor, UniversalCompressorConfig # 配置激进压缩 config UniversalCompressorConfig( compression_ratio_target0.25, # 保留 25% use_magikaTrue, use_kompressTrue, ccr_enabledTrue, ) compressor UniversalCompressor(configconfig) # 压缩一个 JSON API 响应 json_content { users: [ {id: usr_123, name: Alice, bio: Software engineer...}, {id: usr_456, name: Bob, bio: Product manager...} ], total: 2, page: 1 } result compressor.compress(json_content) print(fType: {result.content_type}) # ContentType.JSON print(fHandler: {result.handler_used}) # json print(fSaved: {result.savings_percentage:.0f}%) # ~60% print(fStructure: {result.preservation_ratio:.0%} preserved) # ~40% print(fCCR Key: {result.ccr_key}) # 用于检索十五、小结与延伸阅读Universal Compression 的价值在于把什么该留结构掩码 熵信号与怎么压缩Kompress/截断回退彻底解耦并以 CCR 兜底保证可逆性即使全部 ML 依赖缺失检测、处理、压缩三层都有确定性回退路径API 行为保持稳定。相关文档CCR 指南——可逆压缩架构Compress-Cache-RetrieveTransforms 参考——其他压缩变换含 Kompress 等Text Compression——面向搜索/日志的可选文本压缩工具源码入口headroom/compression/universal.py、headroom/compression/detector.py、headroom/compression/masks.py、headroom/compression/handlers/json_handler.py、headroom/compression/handlers/code_handler.py【免费下载链接】headroomCompress tool outputs, logs, files, and RAG chunks before they reach the LLM. 20% fewer tokens for coding agents, 60-95% fewer tokens for JSON, same answers. Library, proxy, MCP server.项目地址: https://gitcode.com/GitHub_Trending/head/headroom创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考