2. Token 节省技术详解
2.1 过滤策略分类学
RTK 采用针对性压缩策略,而非通用算法。针对 30+ 种命令类型,RTK 实现了 12 种不同的过滤策略,每种策略都针对特定输出模式进行了优化。
策略矩阵总览
| 策略编号 | 策略名称 | 适用模块 | 技术手段 | 压缩率 |
|---|---|---|---|---|
| 1 | 统计信息提取 | git status, git log, pnpm list | 计数/聚合,丢弃细节 | 90-99% |
| 2 | 仅错误模式 | runner (err 模式) | 仅 stderr,丢弃 stdout | 60-80% |
| 3 | 模式分组 | lint, tsc, grep | 按规则/文件分组,计数/摘要 | 80-90% |
| 4 | 去重 | log_cmd | 唯一行 + 计数 | 70-85% |
| 5 | 仅结构 | json_cmd | 键 + 类型,剥离值 | 80-95% |
| 6 | 代码过滤 | read, smart | 语言感知剥离(注释/函数体) | 0-90% |
| 7 | 失败聚焦 | vitest, playwright, runner | 仅失败项,隐藏通过项 | 94-99% |
| 8 | 树压缩 | ls | 树状层级,按目录聚合 | 50-70% |
| 9 | 进度过滤 | wget, pnpm install | 剥离进度条,仅最终结果 | 85-95% |
| 10 | JSON/文本双模式 | ruff, pip | JSON(结构化)+ 文本(回退) | 80%+ |
| 11 | 状态机解析 | pytest | 状态追踪(IDLE→TEST→结果) | 90%+ |
| 12 | NDJSON 流 | go test | 逐行 JSON 解析,聚合结果 | 90%+ |
2.2 策略详解与实现
策略 1:统计信息提取(Stats Extraction)
原理:将冗长的原始输出转换为紧凑的统计摘要。
示例:
# 原始:git status (5000 字符,~200 tokens)
On branch main
Your branch is up to date with 'origin/main'.
Changes not staged for commit:
(use "git add <file>..." to update what will be committed)
(use "git restore <file>..." to discard changes in working directory)
modified: src/main.rs
modified: src/utils.rs
modified: src/config.rs
...
Untracked files:
(use "git add <file>..." to include in what will be committed)
target/
Cargo.lock
# RTK 输出:(~20 字符,~5 tokens, 96% 节省)
3 modified, 2 untracked ✓
实现逻辑(git.rs 简化版):
pub fn format_git_status(output: &str, verbose: u8) -> String {
let mut modified = 0;
let mut untracked = 0;
for line in output.lines() {
if line.starts_with("M ") {
modified += 1;
} else if line.starts_with("?? ") {
untracked += 1;
}
}
format!("{} modified, {} untracked ✓", modified, untracked)
}
适用场景:
- Git 状态/日志/差异
- 包管理器列表(pnpm list, pip list)
- 容器列表(docker ps, kubectl pods)
策略 2:仅错误模式(Error Only)
原理:混合输出中仅保留 stderr(错误),丢弃 stdout(正常输出)。
示例:
# 原始:build 命令 (10000 字符)
[stdout] Compiling module1...
[stdout] Compiling module2...
[stdout] Build complete.
[stderr] ERROR: Failed to resolve dependency 'xyz'
# RTK 输出:(仅 stderr, ~100 字符)
ERROR: Failed to resolve dependency 'xyz'
实现逻辑(runner.rs 简化版):
pub fn run_err(args: &[String], verbose: u8) -> Result<()> {
let output = execute_command(args)?;
// 仅输出 stderr
let stderr = String::from_utf8_lossy(&output.stderr);
if !stderr.is_empty() {
println!("{}", stderr);
}
// stdout 被丢弃
Ok(())
}
适用场景:
- 构建错误排查
- 运行时错误诊断
- 仅需关注失败的场景
策略 3:模式分组(Grouping by Pattern)
原理:将分散的条目按规则/文件/错误代码分组,计数汇总。
示例:
# 原始:lint 输出 (100 行错误,分散)
src/app.ts:5:3 no-unused-vars 'x' is defined but never used
src/app.ts:12:7 no-unused-vars 'y' is defined but never used
src/utils.ts:3:1 semi Missing semicolon
src/utils.ts:8:1 semi Missing semicolon
...
# RTK 输出:(按规则 + 文件分组,~50 字符)
no-unused-vars: 23 issues (app.ts: 15, utils.ts: 8)
semi: 45 issues (app.ts: 30, utils.ts: 15)
实现逻辑(lint_cmd.rs 简化版):
use std::collections::HashMap;
pub fn format_lint_output(output: &str) -> String {
let mut by_rule: HashMap<String, HashMap<String, u32>> = HashMap::new();
for line in output.lines() {
// 解析:file:line:col rule message
if let Some((file, rule)) = parse_lint_line(line) {
by_rule.entry(rule.clone())
.or_insert_with(HashMap::new)
.entry(file.clone())
.and_modify(|c| *c += 1)
.or_insert(1);
}
}
// 格式化为分组摘要
let mut result = String::new();
for (rule, files) in by_rule {
let total: u32 = files.values().sum();
result.push_str(&format!("{}: {} issues\n", rule, total));
}
result
}
适用场景:
- Lint 输出(eslint, biome, ruff, golangci-lint)
- TypeScript 编译器错误(tsc)
- 搜索结果(grep, rg 按文件分组)
策略 4:去重(Deduplication)
原理:识别重复模式,保留唯一行并附加计数。
示例:
# 原始:日志输出 (重复模式)
[ERROR] Connection timeout
[ERROR] Connection timeout
[ERROR] Connection timeout
[INFO] Retrying...
[ERROR] Connection timeout
# RTK 输出:(去重 + 计数)
[ERROR] Connection timeout (×4)
[INFO] Retrying...
实现逻辑(log_cmd.rs 简化版):
use std::collections::HashMap;
pub fn deduplicate_logs(output: &str) -> String {
let mut counts: HashMap<String, u32> = HashMap::new();
for line in output.lines() {
counts.entry(line.to_string())
.and_modify(|c| *c += 1)
.or_insert(1);
}
let mut result = String::new();
for (line, count) in counts {
if count > 1 {
result.push_str(&format!("{} (×{})\n", line, count));
} else {
result.push_str(&format!("{}\n", line));
}
}
result
}
适用场景:
- 重复日志消息
- 循环错误
- 批处理输出
策略 5:仅结构(Structure Only)
原理:保留 JSON 键名和类型,剥离具体值。
示例:
// 原始:大型 JSON (10000 字符)
{
"users": [
{"id": 1, "name": "Alice", "email": "alice@example.com", "bio": "..."},
{"id": 2, "name": "Bob", "email": "bob@example.com", "bio": "..."}
],
"metadata": {"total": 1000, "page": 1, "data": "..."}
}
// RTK 输出:(~200 字符)
{
"users": [{"id": number, "name": string, "email": string, "bio": string}, ...],
"metadata": {"total": number, "page": number, "data": string}
}
实现逻辑(json_cmd.rs 简化版):
use serde_json::Value;
pub fn strip_json_values(value: &Value) -> Value {
match value {
Value::Object(map) => {
Value::Object(map.iter().map(|(k, v)| {
(k.clone(), strip_json_values(v))
}).collect())
}
Value::Array(arr) => {
Value::Array(arr.iter().map(|v| strip_json_values(v)).collect())
}
Value::String(_) => Value::String("<string>".to_string()),
Value::Number(_) => Value::Number(serde_json::Number::from(0)),
Value::Bool(_) => Value::Bool(true),
Value::Null => Value::Null,
}
}
适用场景:
- API 响应结构探索
- 配置文件 schema 查看
- 大数据集结构概览
策略 6:代码过滤(Code Filtering)
原理:基于语言语法树,剥离注释和/或函数体。
过滤级别(filter.rs):
pub enum FilterLevel {
None, // 保留全部(0% 压缩)
Minimal, // 仅剥离注释(20-40% 压缩)
Aggressive, // 剥离注释 + 函数体(60-90% 压缩)
}
示例:
// 原始代码
fn calculate_total(items: &[Item]) -> i32 {
// Sum all items in the list
// This is a helper function
let sum = items.iter().map(|i| i.value).sum();
println!("Total: {}", sum);
sum
}
// FilterLevel::Minimal (仅去注释)
fn calculate_total(items: &[Item]) -> i32 {
let sum = items.iter().map(|i| i.value).sum();
println!("Total: {}", sum);
sum
}
// FilterLevel::Aggressive (去注释 + 函数体)
fn calculate_total(items: &[Item]) -> i32 { ... }
支持语言:Rust, Python, JavaScript, TypeScript, Go, C, C++, Java
检测机制:基于文件扩展名,带回退启发式。
适用场景:
- 代码库探索(rtk read -l aggressive)
- 签名级代码审查
- 大型文件快速浏览
策略 7:失败聚焦(Failure Focus)
原理:仅显示失败项,隐藏所有通过项。
示例:
# 原始:cargo test (200+ 行,含 15 个测试)
running 15 tests
test utils::test_parse ... ok
test utils::test_format ... ok
test auth::test_login ... FAILED
thread 'auth::test_login' panicked at 'assertion failed', src/auth.rs:42
test db::test_query ... ok
...
# RTK 输出:(~20 行,仅失败)
FAILED: 2/15 tests
test auth::test_login: assertion failed at src/auth.rs:42
test db::test_overflow: panic at utils.rs:18
实现逻辑(vitest_cmd.rs 简化版):
pub fn format_vitest_output(output: &str) -> String {
let mut failures = Vec::new();
let mut passed = 0;
let mut total = 0;
for line in output.lines() {
if line.contains("FAIL") || line.contains("✗") {
failures.push(extract_failure_info(line));
} else if line.contains("PASS") || line.contains("✓") {
passed += 1;
}
total += 1;
}
if failures.is_empty() {
format!("✓ {} tests passed", passed)
} else {
let mut result = format!("FAILED: {}/{} tests\n", failures.len(), total);
for failure in failures {
result.push_str(&format!(" {}\n", failure));
}
result
}
}
适用场景:
- 测试运行(vitest, playwright, pytest, go test, cargo test)
- CI/CD 失败排查
- 仅需关注失败的场景
策略 8:树压缩(Tree Compression)
原理:将扁平列表转换为树状层级,按目录聚合计数。
示例:
# 原始:ls -la (45 行,~800 tokens)
drwxr-xr-x 15 user staff 480 ...
-rw-r--r-- 1 user staff 1234 ...
-rw-r--r-- 1 user staff 5678 ...
...
# RTK 输出:(12 行,~150 tokens, 树状)
my-project/
├── src/ (8 files)
│ ├── main.rs
│ ├── utils.rs
│ └── config.rs
├── tests/ (3 files)
├── Cargo.toml
└── README.md
实现逻辑(ls.rs 简化版):
use std::collections::HashMap;
pub fn format_tree(output: &str) -> String {
let mut dirs: HashMap<String, Vec<String>> = HashMap::new();
for line in output.lines() {
let path = extract_path(line);
let parent = get_parent_dir(&path);
dirs.entry(parent).or_insert_with(Vec::new).push(path);
}
// 构建树状输出
let mut result = String::from("project/\n");
for (dir, files) in dirs {
result.push_str(&format!("├── {}/ ({} files)\n", dir, files.len()));
}
result
}
适用场景:
- 目录浏览(rtk ls)
- 文件搜索(rtk find)
- 项目结构概览
策略 9:进度过滤(Progress Filtering)
原理:剥离 ANSI 转义序列和进度条,仅保留最终结果。
示例:
# 原始:pnpm install (带 ANSI 进度条)
Progress: resolved 1, reused 0, downloaded 0, added 0
Progress: resolved 5, reused 0, downloaded 2, added 2
Progress: resolved 10, reused 0, downloaded 5, added 5
...
Progress: resolved 100, reused 0, downloaded 98, added 98
✓ 100 packages installed.
# RTK 输出:(剥离 ANSI,仅最终结果)
✓ 100 packages installed.
实现逻辑(utils.rs 中的 ANSI 剥离):
use regex::Regex;
pub fn strip_ansi(text: &str) -> String {
let re = Regex::new(r"\x1b\[[0-9;]*[a-zA-Z]").unwrap();
re.replace_all(text, "").to_string()
}
pub fn filter_progress(output: &str) -> String {
let clean = strip_ansi(output);
let lines: Vec<&str> = clean.lines()
.filter(|l| !l.starts_with("Progress:"))
.collect();
lines.join("\n")
}
适用场景:
- 包安装(pnpm install, pip install)
- 下载进度(wget, curl)
- 构建进度(cargo build, npm run build)
策略 10:JSON/文本双模式(JSON/Text Dual Mode)
原理:优先使用 JSON API 获取结构化数据,回退到文本解析。
示例(ruff check):
# JSON 模式(ruff check --output-format=json)
{
"violations": [
{"rule": "F401", "file": "app.py", "line": 5, "message": "'x' imported but unused"},
{"rule": "F401", "file": "app.py", "line": 8, "message": "'y' imported but unused"}
]
}
# RTK 输出:(分组摘要)
F401 (unused import): 2 issues in app.py
实现逻辑(ruff_cmd.rs 简化版):
pub fn run_ruff_check(args: &[String], verbose: u8) -> Result<()> {
// 尝试 JSON 输出
let mut cmd = Command::new("ruff");
cmd.args(args).arg("--output-format=json");
let output = cmd.output()?;
let stdout = String::from_utf8_lossy(&output.stdout);
// 尝试解析 JSON
if let Ok(json) = serde_json::from_str::<RuffOutput>(&stdout) {
// 按规则分组
let grouped = group_by_rule(&json.violations);
println!("{}", format_grouped(&grouped));
} else {
// 回退到文本解析
println!("{}", format_text_output(&stdout));
}
Ok(())
}
适用场景:
- Ruff check(JSON)vs ruff format(文本)
- Pip list/show(JSON API)
- Golangci-lint(JSON API)
策略 11:状态机解析(State Machine Parsing)
原理:追踪测试生命周期状态,提取测试结果。
状态转换:
IDLE → TEST_START → (PASSED | FAILED) → SUMMARY
示例(pytest):
# 原始输出
test_auth.py::test_login PASSED
test_auth.py::test_logout PASSED
test_db.py::test_query FAILED
AssertionError: Expected 1 row, got 0
test_db.py::test_insert PASSED
# RTK 输出:(状态机解析)
PASSED: 3/4 tests
FAILED: test_db.py::test_query
AssertionError: Expected 1 row, got 0
实现逻辑(pytest_cmd.rs 简化版):
enum State { Idle, TestStart, Passed, Failed }
pub fn parse_pytest_output(output: &str) -> String {
let mut state = State::Idle;
let mut failures = Vec::new();
let mut passed = 0;
let mut total = 0;
for line in output.lines() {
match state {
State::Idle => {
if line.starts_with("test_") {
state = State::TestStart;
total += 1;
}
}
State::TestStart => {
if line.contains("PASSED") {
passed += 1;
state = State::Idle;
} else if line.contains("FAILED") {
state = State::Failed;
}
}
State::Failed => {
failures.push(line.to_string());
if line.is_empty() {
state = State::Idle;
}
}
}
}
format_summary(passed, total, failures)
}
适用场景:
- Pytest 输出解析
- 其他测试框架(Jest, Mocha 等)
- 结构化测试生命周期追踪
策略 12:NDJSON 流(NDJSON Streaming)
原理:逐行解析 NDJSON(Newline Delimited JSON),聚合交错的事件。
示例(go test):
// 原始 NDJSON 输出(交错的包事件)
{"Action":"run","Package":"pkg1","Test":"TestAuth"}
{"Action":"fail","Package":"pkg1","Test":"TestAuth"}
{"Action":"run","Package":"pkg2","Test":"TestDB"}
{"Action":"pass","Package":"pkg2","Test":"TestDB"}
// RTK 输出:(聚合结果)
2 packages tested: pkg1 (1 failed), pkg2 (passed)
实现逻辑(go_cmd.rs 简化版):
use serde::Deserialize;
#[derive(Deserialize)]
struct GoTestEvent {
Action: String, // "run", "pass", "fail"
Package: String,
Test: Option<String>,
}
pub fn parse_go_test_ndjson(output: &str) -> String {
let mut failures: HashMap<String, Vec<String>> = HashMap::new();
for line in output.lines() {
if let Ok(event) = serde_json::from_str::<GoTestEvent>(line) {
if event.Action == "fail" {
failures.entry(event.Package)
.or_insert_with(Vec::new)
.push(event.Test.unwrap_or_default());
}
}
}
format_summary(&failures)
}
适用场景:
- Go 测试(NDJSON 流)
- 其他流式 JSON 输出工具
- 交错包/测试事件聚合
2.3 策略选择决策树
RTK 根据以下决策树选择过滤策略:
输出格式已知?
├─ 工具提供 JSON 标志?
│ ├─ 需要结构化数据? → 使用 JSON API
│ │ 示例:ruff check, pip list, golangci-lint
│ │
│ └─ 简单输出? → 使用文本模式
│ 示例:ruff format, go build 错误
│
├─ 流式事件(NDJSON)?
│ └─ 逐行 JSON 解析
│ 示例:go test(交错的包事件)
│
└─ 仅纯文本?
├─ 需要状态追踪? → 状态机
│ 示例:pytest(测试生命周期追踪)
│
└─ 简单过滤? → 文本过滤器
示例:go vet, go build
2.4 量化节省数据
30 分钟 Claude Code 会话基准测试
| 操作 | 频率 | 标准输出 | RTK 输出 | 节省率 |
|---|---|---|---|---|
ls / tree | 10x | 2,000 | 400 | -80% |
cat / read | 20x | 40,000 | 12,000 | -70% |
grep / rg | 8x | 16,000 | 3,200 | -80% |
git status | 10x | 3,000 | 600 | -80% |
git diff | 5x | 10,000 | 2,500 | -75% |
git log | 5x | 2,500 | 500 | -80% |
git add/commit/push | 8x | 1,600 | 120 | -92% |
cargo test / npm test | 5x | 25,000 | 2,500 | -90% |
ruff check | 3x | 3,000 | 600 | -80% |
pytest | 4x | 8,000 | 800 | -90% |
go test | 3x | 6,000 | 600 | -90% |
docker ps | 3x | 900 | 180 | -80% |
总计:~118,000 → ~23,900 tokens(-80% 整体节省)
各命令类别节省分布
┌─────────────────────────────────────────────────────────┐
│ Token 节省率分布(按命令类别) │
├─────────────────────────────────────────────────────────┤
│ │
│ Git 操作 ████████████████████ 85-99% │
│ 测试运行 ██████████████████ 90-99% │
│ Lint/编译 ████████████████ 80-90% │
│ 文件操作 ████████████ 50-90% │
│ 包管理 ██████████████ 70-90% │
│ 容器/基础设施 ████████████ 60-80% │
│ 网络请求 ██████████████ 70-95% │
│ │
│ 平均节省率:78.5% │
└─────────────────────────────────────────────────────────┘
2.5 策略对比分析
通用压缩 vs 针对性优化
| 维度 | 通用压缩(如 gzip) | RTK 针对性优化 |
|---|---|---|
| 原理 | 字节级统计压缩 | 语义级结构压缩 |
| 可理解性 | 需解压,LLM 不可读 | 直接可读,保留语义 |
| 压缩率 | 文本 60-70% | 针对性 60-99% |
| LLM 友好 | ❌ 需先解压 | ✅ 直接输入 |
| 上下文感知 | ❌ 无 | ✅ 命令类型感知 |
| 错误容忍 | ❌ 一位错误全损 | ✅ 部分丢失可接受 |
12 种策略对比矩阵
| 策略 | 压缩率 | 信息损失 | 适用面 | 实现复杂度 |
|---|---|---|---|---|
| 统计提取 | 90-99% | 中 | 窄(Git/列表) | 低 |
| 仅错误 | 60-80% | 高 | 中(构建/运行) | 低 |
| 模式分组 | 80-90% | 低 | 宽(lint/编译) | 中 |
| 去重 | 70-85% | 低 | 中(日志) | 低 |
| 仅结构 | 80-95% | 高 | 窄(JSON) | 中 |
| 代码过滤 | 0-90% | 可调 | 中(代码文件) | 高 |
| 失败聚焦 | 94-99% | 中 | 窄(测试) | 中 |
| 树压缩 | 50-70% | 低 | 窄(文件系统) | 中 |
| 进度过滤 | 85-95% | 无 | 中(安装/下载) | 低 |
| JSON/文本 | 80%+ | 无 | 宽(工具输出) | 高 |
| 状态机 | 90%+ | 低 | 窄(测试) | 高 |
| NDJSON | 90%+ | 低 | 窄(Go 测试) | 高 |
2.6 结论
RTK 的 token 节省技术采用针对性压缩策略,而非通用算法。通过 12 种精心设计的过滤策略,RTK 能够针对 30+ 种常见开发命令实现 60-99% 的压缩率。
关键技术特点:
- 语义级压缩:保留可读性,LLM 可直接理解
- 策略多样性:12 种策略覆盖所有开发场景
- 信息损失可控:通过 Tee 机制保留完整输出
- 性能优异:<15ms 开销,90%+ 压缩率
- 高度可配置:3 级 verbose + ultra-compact 模式
下一章将详细分析信息精度损失和权限处理风险。