Logo
热心市民王先生

2. Token 节省技术详解

2.1 过滤策略分类学

RTK 采用针对性压缩策略,而非通用算法。针对 30+ 种命令类型,RTK 实现了 12 种不同的过滤策略,每种策略都针对特定输出模式进行了优化。

策略矩阵总览

策略编号策略名称适用模块技术手段压缩率
1统计信息提取git status, git log, pnpm list计数/聚合,丢弃细节90-99%
2仅错误模式runner (err 模式)仅 stderr,丢弃 stdout60-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%
10JSON/文本双模式ruff, pipJSON(结构化)+ 文本(回退)80%+
11状态机解析pytest状态追踪(IDLE→TEST→结果)90%+
12NDJSON 流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 / tree10x2,000400-80%
cat / read20x40,00012,000-70%
grep / rg8x16,0003,200-80%
git status10x3,000600-80%
git diff5x10,0002,500-75%
git log5x2,500500-80%
git add/commit/push8x1,600120-92%
cargo test / npm test5x25,0002,500-90%
ruff check3x3,000600-80%
pytest4x8,000800-90%
go test3x6,000600-90%
docker ps3x900180-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%+窄(测试)
NDJSON90%+窄(Go 测试)

2.6 结论

RTK 的 token 节省技术采用针对性压缩策略,而非通用算法。通过 12 种精心设计的过滤策略,RTK 能够针对 30+ 种常见开发命令实现 60-99% 的压缩率。

关键技术特点:

  • 语义级压缩:保留可读性,LLM 可直接理解
  • 策略多样性:12 种策略覆盖所有开发场景
  • 信息损失可控:通过 Tee 机制保留完整输出
  • 性能优异:<15ms 开销,90%+ 压缩率
  • 高度可配置:3 级 verbose + ultra-compact 模式

下一章将详细分析信息精度损失和权限处理风险。