跳转到内容
OmniRoute source

RTK Compression (中文 (简体))

RTK 支持 3 个强度级别,用于在压缩激进程度与安全性之间进行权衡。该级别通过引擎配置中的 config.intensity 设置。

级别 截断阈值 token 节省量 风险 最适合
minimal 每个分段 24 行 ~20-40% 极低 具有关键上下文的生产环境
standard(默认) 每个分段 24 行 ~50-70% 低 日常编码会话
aggressive 每个分段 16 行 ~70-90% 中等 长会话、最大程度节省 token

截断阈值会影响 lineFilter.ts:

// 来自 open-sse/services/compression/engines/rtk/index.ts:329-330
config.intensity === "aggressive" ? 16 : 24,
config.intensity === "aggressive" ? 16 : 24,

每个分段的开头和结尾都会保留;触发截断时,中间内容会被丢弃。

内容 minimal standard aggressive
错误 / 堆栈跟踪 ✅ 保留 ✅ 保留 ✅ 保留
测试失败 ✅ 保留 ✅ 保留 ✅ 保留
构建错误 ✅ 保留 ✅ 保留 ✅ 保留
测试通过(详细输出) ✅ 保留 🟡 折叠 🟡 折叠
常规输出(信息日志) 🟡 折叠 🟡 折叠 ❌ 丢弃
进度条 🟡 折叠 ❌ 丢弃 ❌ 丢弃
横幅 / ASCII 艺术字 🟡 折叠 ❌ 丢弃 ❌ 丢弃
丢失上下文是否会造成灾难性后果?
│
┌───────────┼───────────┐
│ │ │
是 否 不确定
│ │ │
▼ │ │
minimal │ │
│ │ │
│ ▼ ▼
│ 吞吐量有多 先尝试 `standard`
│ 重要? (适用于 80% 的
│ │ 情况)
│ ┌────┴────┐
│ │ │
│ 低 高
│ │ │
│ ▼ ▼
│ standard aggressive
│ │ │
└──────┴─────────┘

按组合配置(在组合配置中):

{
"combo": "my-coding-combo",
"routing": {/* ... */},
"compression": {
"engine": "rtk",
"intensity": "aggressive"
}
}

通过编程方式:

rtkEngine(@omniroute/open-sse/services/compression/engines/rtk)是一个 CompressionEngine,且没有 updateConfig 方法。请改为通过注册表辅助函数 更新引擎配置:

import { updateEngineConfig } from "@omniroute/open-sse/services/compression/engines/registry";
updateEngineConfig("rtk", { intensity: "aggressive" });

使用验证门禁(见下文)确认过滤器在所选强度下是安全的:

import { runRtkFilterTests } from "omniroute/compression/engines/rtk/verify";
const result = runRtkFilterTests({ intensity: "aggressive" });
if (!result.passed) {
console.error("过滤器在 aggressive 强度下未通过");
}

engines/rtk/filters/ 目录包含 49+ 个内置过滤器 JSON 文件。你可以添加自己的过滤器,以压缩默认过滤器未覆盖的自定义工具输出。

{
"id": "string", // 必填。过滤器标识符(kebab-case,例如 "python-traceback")
"label": "string", // 必填。人类可读的过滤器名称
"description": "string", // 可选(默认值:"String")。过滤器功能的简短描述
"category": "git|test|build|shell|docker|package|infra|cloud|generic",
"priority": number, // 可选(0-100,默认值:50)。执行顺序(数值越高越先执行)
"match": {
"commands": ["string"], // 要匹配的命令名称(例如 "python"、"pytest")
"patterns": ["string"], // 用于匹配输出的正则表达式模式
"outputTypes": ["string"] // 检测到的输出类别(例如 "test-failure")
},
"rules": {
"stripAnsi": boolean, // 可选(默认值:false)。移除 ANSI 颜色代码
"replace": [ // 查找和替换规则(默认值:[])
{ "pattern": "regex", "replacement": "..." }
],
"matchOutput": [ // 模式匹配时短路处理(默认值:[])
{
"pattern": "regex",
"message": "short summary",
"unless": "regex" // 如果此模式匹配,则跳过
}
],
"includePatterns": ["string"], // 要保留的行(正则表达式模式,默认值:[])
"dropPatterns": ["string"], // 要删除的行(正则表达式模式,默认值:[])
"collapsePatterns": ["string"], // 要折叠为单次出现的行(默认值:[])
"deduplicate": boolean, // 可选(默认值:false)。删除重复行
"truncateLineAt": number, // 可选(默认值:0)。将行截断到最大字符数
"maxLines": number, // 可选(默认值:0)。总行数硬性上限
"headLines": number, // 可选(默认值:20)。保留匹配输出的前 N 行
"tailLines": number, // 可选(默认值:20)。保留匹配输出的后 N 行
"onEmpty": "string", // 可选(默认值:"String")。所有行都被过滤时的后备消息
"filterStderr": boolean // 可选(默认值:false)。同时过滤 stderr 输出
},
"preserve": {
"errorPatterns": ["string"], // 必须始终保留的模式(默认值:[])
"summaryPatterns": ["string"] // 最终摘要行的模式(默认值:[])
},
"tests": [ // 用于验证的内联测试(默认值:[])
{
"name": "string", // 必填。测试名称
"input": "sample output", // 必填。示例输入文本
"expected": "expected output", // 必填。预期的压缩输出
"command": "optional command" // 可选。命令上下文
}
]
}
{
"id": "python-traceback",
"label": "Python Traceback Filter",
"description": "Compresses Python tracebacks to essential file/line locations and error type",
"category": "test",
"priority": 60,
"match": {
"commands": ["python", "python3", "pytest", "uv", "poetry"],
"patterns": ["Traceback \\(most recent call last\\)", "Error", "Exception"],
"outputTypes": ["error-traceback"]
},
"rules": {
"stripAnsi": true,
"includePatterns": [
"Traceback \\(most recent call last\\)",
"^\\s*File \".+\", line \\d+",
"^\\s*[A-Z][a-zA-Z]+Error:",
"^\\s*[A-Z][a-zA-Z]+Exception"
],
"dropPatterns": ["site-packages/", "^\\s+[a-z_]+\\([^)]*\\)$"],
"headLines": 5,
"tailLines": 3,
"maxLines": 25,
"filterStderr": true
},
"preserve": {
"errorPatterns": ["Error:", "Exception:", "Traceback"],
"summaryPatterns": ["^[A-Z][a-zA-Z]+(?:Error|Exception):"]
},
"tests": [
{
"name": "preserves-error-type-and-location",
"input": "Traceback (most recent call last):\n File \"app.py\", line 42, in main\n do_thing()\n File \"lib/utils.py\", line 17, in helper\n return 1 / 0\nZeroDivisionError: division by zero",
"expected": "Traceback (most recent call last):\n File \"app.py\", line 42, in main\n File \"lib/utils.py\", line 17, in helper\nZeroDivisionError: division by zero",
"command": "python app.py"
}
]
}

将文件放置在可识别的位置:

~/.omniroute/rtk/filters/my-filter.json # 用户级
<project>/.rtk/filters/my-filter.json # 项目级

过滤器会在启动时通过 open-sse/services/compression/engines/rtk/filterLoader.ts 中的 loadRtkFilters() 自动加载。加载器会从以下位置发现过滤器:

  • 内置目录:open-sse/services/compression/engines/rtk/filters/
  • 用户目录:~/.omniroute/rtk/filters/
  • 项目目录:<project>/.rtk/filters/

要以编程方式加载过滤器:

import { loadRtkFilters } from "@omniroute/open-sse/services/compression/engines/rtk/filterLoader";
// 选项:customFiltersEnabled(加载用户/项目过滤器,默认启用)、
// trustProjectFilters、refresh。
const filters = loadRtkFilters({ customFiltersEnabled: true });

过滤器在加载时会根据 Zod 模式进行验证。结构错误的过滤器将加载失败,并记录错误:

RTK_FILTER_LOADER: filter "my-filter" failed validation:
- rules.replace.0.pattern: Invalid regex
- match.commands: must not be empty

要验证所有已安装的过滤器,请调用从 open-sse/services/compression/engines/rtk/verify.ts 导出的 runRtkFilterTests()。

  1. 始终包含 tests[] — 它们可以证明过滤器有效,并防止回归
  2. 使用 matchOutput 进行短路处理 — 如果单行就能说明情况,则替换整个代码块
  3. 优先使用 keep 而不是 strip — 明确的“始终保留”规则比“始终删除”更安全
  4. 在所有 3 个强度级别进行测试 — minimal 应不执行任何操作,aggressive 仍应保留错误
  5. 使用 unless 字段 — 使用“如果存在 X,则不触发”来保护短路逻辑

当 RTK 大幅压缩输出时,你可以恢复原始文本,用于调试、审计或重放。

原始输出(10K 个 token)
│
▼
RTK 压缩(rawOutput.enabled=true)
│
├─▶ 压缩后的输出(2K 个 token) ──▶ 发送至 LLM
│
└─▶ 原始输出(10K 个 token) ──▶ 存储在数据库中
(通过 request_id 关联)

按请求启用(在组合配置中):

{
"compression": {
"engine": "rtk",
"intensity": "aggressive",
"rawOutput": {
"enabled": true,
"maxBytes": 1048576 // 上限为 1MB
}
}
}

默认值:rawOutput.enabled: false(可节省存储空间)。

每个请求 上限 1MB 上限 10MB
平均压缩输出 ~5KB ~5KB
存储的原始输出 ~50-500KB ~500KB-5MB
每天 1000 个请求时 50-500MB/天 500MB-5GB/天

建议:仅在调试会话或抽样审计期间启用原始输出,不要始终开启。

import { readRtkRawOutput } from "omniroute/compression/engines/rtk/rawOutput";
const raw = readRtkRawOutput(pointerId); // pointerId 来自压缩统计信息
if (raw) {
console.log("Original output:", raw);
}

压缩后,pointerId 会在 CompressionStats.rtkRawOutputPointers[] 中返回。 有关函数签名,请参阅 open-sse/services/compression/engines/rtk/rawOutput.ts:102。

RTK 过滤器验证(open-sse/services/compression/engines/rtk/verify.ts)会根据各过滤器的 tests[] 对其进行验证,并确保在全部 3 个强度级别下行为均正确。

调用 runRtkFilterTests() 以运行验证:

import { runRtkFilterTests } from "open-sse/services/compression/engines/rtk/verify";
const result = runRtkFilterTests();
console.log(`Passed: ${result.outcomes.filter((o) => o.passed).length}`);
console.log(`Failed: ${result.outcomes.filter((o) => !o.passed).length}`);
if (!result.passed) {
console.error("Filters failed verification");
result.outcomes
.filter((o) => !o.passed)
.forEach((o) => {
console.error(
` - ${o.filterId} / ${o.testName}: expected "${o.expected}", got "${o.actual}"`
);
});
}

验证内容:

  1. 每个过滤器都能加载并通过模式验证
  2. 每个 tests[] 条目都能生成预期输出
  3. minimal 强度不执行任何操作(保留原始内容,仅应用结构过滤器)
  4. aggressive 强度会保留错误、测试失败信息和堆栈跟踪
  5. 压缩后的输出绝不会大于原始输入
  • 源代码:open-sse/services/compression/engines/rtk/(63 个文件,约 70KB)

  • 合并过滤器更改之前 — 始终确保测试通过

  • 升级 RTK 引擎之后 — 模式可能已发生变化

  • 在监控中定期执行 — 防止测试固件发生偏移

  • 添加新的工具/命令系列时 — 证明新过滤器能够正常工作



OmniRoute 源码 (a58000c7685f)

HagiCode

HagiCode 是一套智能体编码工作台:结构化工作流、多 Agent 并行执行与 Hero Dungeon 视图,把想法变成真正交付的软件。

让想法更快变成好用的软件,让智能编码更聪明、更高效,也更有趣。

HagiCode 浅色主题主界面截图
  • Smart结构化工作流将意图转化为从想法到交付的可执行路径。
  • Efficient多 Agent 工作流让调研、实现与审阅并行推进。
  • FunHero Dungeon 让长时间编码协作更直观、更有参与感。
访问 HagiCode