Trace Converter - 统一的Trace格式转换工具

将各种trace格式转换为Optimizer标准的Get+Write格式。

✨ 特性

自动发现 Converter

系统会自动扫描并发现所有可用的 converter：

内置 converter: converters/ 目录下的标准格式
自动发现: 通过路径推断自动加载额外的 converter
用户扩展: 支持通过参数添加自定义 converter

# 使用内置 converter
python3 trace_converter.py -i input.jsonl -o output.jsonl -f qwen_bailian

# 使用自定义 converter
python3 trace_converter.py -i input.jsonl -o output.jsonl -f custom \
    --converter-module /path/to/custom_converter.py

安装

cd kv_cache_manager/optimizer/tools/trace_converter

# 方式1: 直接使用 (推荐)
pip install -r requirements.txt

# 方式2: 安装为Python包 (可选)
pip install -e .

快速开始

自动输出文件名: 如果不指定-o参数,会自动在输入文件同目录生成<input_name>_optimizer.jsonl。

Publisher Log转换

自动识别所有instance (从日志的source字段提取)

# 最简单用法 (自动生成输出文件)
python trace_converter.py \
    -i /path/to/publisher.log \
    -f publisher_log \
    --mode optimizer
# → 输出: /path/to/publisher_optimizer.jsonl

# 指定输出文件
python trace_converter.py \
    -i /path/to/publisher.log \
    -o /path/to/optimizer_trace.jsonl \
    -f publisher_log \
    --mode optimizer \
    --instance-block-sizes "instance1:16,instance2:32"

# 输出示例:
# 📌 Discovered instance: instance1 (block_size=16)
# 📌 Discovered instance: instance2 (block_size=32)
# ✅ Discovered 2 instance(s):
#    - instance1: block_size=16, traces=150
#    - instance2: block_size=32, traces=200

# 使用默认block_size=16
python trace_converter.py \
    -i /path/to/publisher.log \
    -o /path/to/optimizer_trace.jsonl \
    -f publisher_log \
    --mode optimizer

Qwen Bailian数据集转换

单instance (数据集本身没有instance概念,使用--instance-id指定)

# Optimizer模式 - 指定block_size
python trace_converter.py \
    -i /path/to/qwen_trace.jsonl \
    -o /path/to/optimizer_trace.jsonl \
    -f qwen_bailian \
    --mode optimizer \
    --instance-id my_instance \
    --instance-block-sizes "my_instance:32"

# 使用默认block_size=16
python trace_converter.py \
    -i /path/to/qwen_trace.jsonl \
    -o /path/to/optimizer_trace.jsonl \
    -f qwen_bailian \
    --instance-id my_instance

文本对话转换

单instance (文本trace本身没有instance概念,使用--instance-id指定)

# Optimizer模式 - 使用本地tokenizer
python trace_converter.py \
    -i /path/to/text_conversations.jsonl \
    -o /path/to/optimizer_trace.jsonl \
    -f text \
    --mode optimizer \
    --tokenizer-path /path/to/model/deepseek-v3 \
    --time-field time \
    --content-field prompt_messages \
    --instance-id my_text_instance \
    --instance-block-sizes "my_text_instance:32"

# 使用HuggingFace模型ID (自动下载)
python trace_converter.py \
    -i /path/to/text_conversations.jsonl \
    -o /path/to/optimizer_trace.jsonl \
    -f text \
    --mode optimizer \
    --tokenizer-path "Qwen/Qwen2.5-7B-Instruct" \
    --instance-id my_instance

# 优先使用本地缓存 (如果./model/deepseek-v3存在)
python trace_converter.py \
    -i /path/to/text_conversations.jsonl \
    -o /path/to/optimizer_trace.jsonl \
    -f text \
    --mode optimizer \
    --tokenizer-path "deepseek-ai/DeepSeek-V3" \
    --model-name deepseek-v3 \
    --instance-id my_instance

# 仅使用model-name (如果./model/qwen存在)
python trace_converter.py \
    -i /path/to/text_conversations.jsonl \
    -o /path/to/optimizer_trace.jsonl \
    -f text \
    --mode optimizer \
    --model-name qwen \
    --instance-id my_instance

# 使用默认block_size=16
python trace_converter.py \
    -i /path/to/text_conversations.jsonl \
    -o /path/to/optimizer_trace.jsonl \
    -f text \
    --tokenizer-path /path/to/model/deepseek-v3 \
    --instance-id my_text_instance

Tokenizer获取方式:

本地tokenizer (推荐,所有模型共享):

--tokenizer-path /path/to/local/tokenizer  # 或 HF模型ID
--model-name Qwen2.5-7B-Instruct          # 优先使用 ./model/{model_name}

自动从HF拉取 (需要模型映射):
- 默认映射: GLM-4.6 → zai-org/GLM-4.6
- 自定义映射: --model-mapping "Qwen2.5:Qwen/Qwen2.5-7B-Instruct,CustomModel:org/custom-model"

说明:

如果指定了--tokenizer-path或--model-name,所有模型将共享同一个tokenizer

输入格式

Publisher Log格式

JSON格式,每行一个事件:

{
  "type": "GetCacheLocation",
  "source": "instance_01",
  "trigger_time_us": 1704110400000000,
  "keys": [123, 456, 789],
  "query_type": "prefix_match"
}

Qwen Bailian格式

{
  "timestamp": 1704110400.0,
  "hash_ids": [123, 456, 789],  // 只包含input部分的hash
  "input_length": 48,
  "output_length": 32            // 仅用于统计,无对应KV Cache
}

注意: hash_ids只包含input部分,output_length仅供推理引擎模拟器使用。

文本格式

{
  "time": "2024-01-01 12:00:00.000000",
  "prompt_messages": "用户输入的文本"
}

或对话格式:

{
  "time": "2024-01-01 12:00:00.000000",
  "prompt_messages": {
    "messages": [
      {"role": "user", "content": "你好"},
      {"role": "assistant", "content": "你好!"}
    ]
  }
}

输出格式

Optimizer模式 (Get+Write)

GetLocationSchemaTrace (读操作):

{
  "instance_id": "instance",
  "trace_id": "trace_instance_1704110400000000000",
  "timestamp_ns": 1704110400000000000,
  "keys": [123, 456, 789],
  "query_type": "prefix_match",
  "block_mask": [],
  "sw_size": 0,
  "location_spec_names": []
}

WriteCacheSchemaTrace (写操作):

{
  "instance_id": "instance",
  "trace_id": "trace_instance_1704110400000000001",
  "timestamp_ns": 1704110400000000001,
  "keys": [123, 456, 789, 1011]
}

命令行参数

通用参数

参数	必需	说明	默认值
`-i, --input`	是	输入文件路径	-
`-o, --output`	否	输出文件路径	自动生成: `<input>_optimizer.jsonl`
`-f, --format`	是	输入格式: `publisher_log`, `qwen_bailian`, `text`	-
`--mode`	否	输出模式: `optimizer` (Get+Write)	`optimizer`
`--instance-id`	否	默认实例ID (仅当格式无instance信息时使用)	`instance`
`--instance-block-sizes`	否	每个instance的block_size (格式: `inst1:16,inst2:32`)	未指定使用默认16

Tokenizer参数 (text格式需要)

参数	必需	说明	默认值
`--tokenizer-path`	否*	Tokenizer路径 (本地路径或HuggingFace模型ID)	-
`--model-name`	否	模型名称 (如果指定且`./model/{model_name}`存在,优先使用)	-
`--model-mapping`	否	模型名称映射表 (格式见下方)	`GLM-4.6:zai-org/GLM-4.6`

环境变量:

HF_ENDPOINT: HuggingFace镜像地址 (默认: https://hf-mirror.com)
- 设置为空字符串可使用官方源: export HF_ENDPOINT=""
- 自定义镜像: export HF_ENDPOINT=https://your-mirror.com

格式特定参数

参数	适用格式	说明	默认值
`--time-field`	text	时间字段名	`time`
`--content-field`	text	内容字段名	`prompt_messages`

使用说明

Tokenizer参数 :

--tokenizer-path: tokenizer路径 (本地路径或HuggingFace模型ID)
- 本地路径: model/deepseek-v3, /path/to/model
- HF模型ID: deepseek-ai/DeepSeek-V3, Qwen/Qwen2.5-7B-Instruct
- 与--model-name二选一,或配合使用
--model-name: 模型名称 (如果./model/{model_name}存在,优先使用)
- 可单独使用 (如果本地已有tokenizer)
- 或作为--tokenizer-path的本地缓存优先级

加载逻辑 :

如果指定--model-name且./model/{model_name}存在 → 使用本地缓存
否则使用--tokenizer-path → 传给AutoTokenizer.from_pretrained()
AutoTokenizer自动判断是本地路径还是HF模型ID

Instance参数:

--instance-block-sizes: 格式 instance1:16,instance2:32,instance3:8
未指定的instance使用默认值16
publisher_log格式会自动识别所有instance
qwen_bailian和text格式使用--instance-id参数

与Optimizer集成

转换后的trace文件可直接用于Optimizer:

{
    "trace_file_path": "/path/to/optimizer_trace.jsonl",
    "output_result_path": "/path/to/output",
    "instance_groups": [
        {
            "group_name": "instance_group_01",
            "quota_capacity": 12000,
            "instances": [
                {
                    "instance_id": "instance",
                    "block_size": 16,
                    "eviction_policy_type": "lru"
                }
            ]
        }
    ]
}

bazel run //kv_cache_manager/optimizer:optimizer_main -- config.json

多Instance支持

自动识别 (Publisher Log)

Publisher Log格式会自动从日志的source字段识别所有instance,可以为每个instance指定不同的block_size:

# 为每个instance指定block_size
python trace_converter.py \
    -i publisher.log \
    -o output.jsonl \
    -f publisher_log \
    --instance-block-sizes "instance1:16,instance2:32,instance3:8"

# 输出:
# 📌 Discovered instance: instance1 (block_size=16)
# 📌 Discovered instance: instance2 (block_size=32)
# 📌 Discovered instance: instance3 (block_size=8)
# ✅ Discovered 3 instance(s):
#    - instance1: block_size=16, traces=150
#    - instance2: block_size=32, traces=200
#    - instance3: block_size=8, traces=100

# 使用默认block_size=16
python trace_converter.py \
    -i publisher.log \
    -o output.jsonl \
    -f publisher_log

生成的trace文件会包含所有instance的数据,每个trace的instance_id字段保留原始值。

单Instance (Qwen/Text)

Qwen Bailian和Text格式本身没有instance概念,使用--instance-id指定instance ID,使用--instance-block-sizes指定block_size:

# 指定block_size
python trace_converter.py \
    -i qwen.jsonl \
    -o output.jsonl \
    -f qwen_bailian \
    --instance-id my_instance \
    --instance-block-sizes "my_instance:32"

# 使用默认block_size=16
python trace_converter.py \
    -i qwen.jsonl \
    -o output.jsonl \
    -f qwen_bailian \
    --instance-id my_instance

前缀哈希算法

本工具使用前缀依赖的block ID生成算法,与C++ Optimizer内部的HashIntFunc完全一致 (Jenkins Hash变种):

# C++ 原始实现 (hash_util.h):
# hash ^= hasher(value) + 0x9e3779b97f4a7c15 + (hash << 12) + (hash >> 32);

def hash_int64_func(prev_hash: int, current_value: int) -> int:
    """Jenkins Hash 变种 - 与C++ HashIntFunc完全一致"""
    GOLDEN_RATIO = 0x9e3779b97f4a7c15
    value_hash = current_value  # std::hash<int64_t> 直接返回值
    
    hash_value = prev_hash & 0xFFFFFFFFFFFFFFFF
    left_shift = (hash_value << 12) & 0xFFFFFFFFFFFFFFFF
    right_shift = (hash_value >> 32) & 0xFFFFFFFFFFFFFFFF
    rhs = (value_hash + GOLDEN_RATIO + left_shift + right_shift) & 0xFFFFFFFFFFFFFFFF
    
    result = prev_hash ^ rhs
    # 转换为有符号int64
    result &= 0xFFFFFFFFFFFFFFFF
    if result >= 0x8000000000000000:
        result -= 0x10000000000000000
    return result

# 应用前缀哈希 (直接使用哈希值作为block key)
hash_value = 0
for token in tokens:
    hash_value = hash_int64_func(hash_value, token)
    block_keys.append(hash_value)  # 哈希值即为block ID

设计要点:

✅ 无状态设计: 完全消除ID映射,直接用哈希值作为block key
✅ 幂等性保证: 相同输入永远产生相同输出
✅ 多进程安全: 无共享状态,天然支持并行处理
✅ 跨语言一致: Python与C++生成完全相同的block key

合成时间戳策略

对于不能区分读写的trace (如Qwen Bailian, 文本对话),在Optimizer模式下:

原始时间戳: T

生成策略:
- GetLocationSchemaTrace.timestamp_ns = T
- WriteCacheSchemaTrace.timestamp_ns = T + 1 (纳秒)

理由: 保证Get在Write之前 (prefill先于decode)

时间戳冲突自动处理:

如果T或T+1已被使用,自动递增直到找到可用时间戳
确保所有trace的时间戳唯一且有序

扩展：添加自定义 Converter

步骤1: 创建 Converter 类

创建一个继承 BaseConverter 的类：

# my_converter.py
from converters.base import BaseConverter
import json

class MyConverter(BaseConverter):
    """自定义格式转换器"""
    
    def __init__(self, default_instance_id='instance',
                 instance_block_sizes=None, mode='optimizer', **kwargs):
        super().__init__(default_instance_id, instance_block_sizes, mode)
        self.converter_config = kwargs.get('converter_config', {})
    
    def convert_to_traces(self, input_file: str) -> list:
        """读取输入并返回标准 trace dict 列表"""
        traces = []
        with open(input_file, 'r', encoding='utf-8') as f_in:
            for line in f_in:
                if not line.strip():
                    continue
                raw = json.loads(line)
                traces.append({
                    'type': 'get',
                    'instance_id': raw.get('instance_id', self.default_instance_id),
                    'trace_id': raw['trace_id'],
                    'timestamp_ns': raw['timestamp_ns'],
                    'keys': raw['keys'],
                    'input_len': raw['input_len'],
                })
        return traces

命名约定：

类名以 Converter 结尾
系统自动推断 format 名称：MyConverter → my

步骤2: 使用自定义 Converter

# 方式A: 显式注册文件
python3 trace_converter.py -i input.jsonl -o output.jsonl -f my \
    --converter-module /path/to/my_converter.py

# 方式B: 扫描目录
python3 trace_converter.py -i input.jsonl -o output.jsonl -f my \
    --converter-dir /path/to/converters_dir

# 方式C: 放到 converters/ 目录（自动发现）
cp my_converter.py converters/
python3 trace_converter.py -i input.jsonl -o output.jsonl -f my

Converter 专属配置

框架支持把 JSON object 解析为 converter_config dict 传给 converter,用于内部或自定义 converter 扩展。公共 CLI 不需要知道这些参数的具体含义。

python3 trace_converter.py \
    -i input.jsonl \
    -o output.jsonl \
    -f my \
    --converter-module /path/to/my_converter.py \
    --converter-config-file /path/to/converter_config.json

也可以直接传 inline JSON:

python3 trace_converter.py \
    -i input.jsonl \
    -o output.jsonl \
    -f my \
    --converter-module /path/to/my_converter.py \
    --converter-config-json '{"field_name":"value"}'

Converter 输出接口

自定义 converter 至少实现以下接口之一:

def convert_to_traces(self, input_file: str) -> list:
    """返回 trace dict 列表,由 trace_converter.py 负责写 JSONL。"""

def convert_to_trace_jsonl(self, input_file: str, output_file: str) -> int:
    """直接写出 JSONL,每行一个 trace dict。"""

convert_to_trace_jsonl() 适合大文件流式转换。它的返回值 contract:

必须返回非负 int,表示写出的 trace 条数。
返回其他类型或负数会被视为 converter contract 错误。
trace_converter.py 不会对该接口写出的单文件内容做额外排序或顺序校验;如果需要特定顺序,由 converter 实现自行保证。
output_file 可能是 trace_converter.py 管理的临时路径,成功返回后再原子替换到最终输出路径;converter 只应写入传入的 output_file。

BaseConverter 提供的辅助方法

# 创建 Get trace
get_trace = self._create_get_trace(
    timestamp_ns=timestamp_ns,
    keys=block_keys,
    instance_id=instance_id,
    input_len=input_len
)

# 创建 paired Write trace；使用 get_trace["keys"]，保证 partial tail block 已按 input_len 截掉
self._create_write_trace(
    timestamp_ns=timestamp_ns,
    keys=get_trace["keys"],
    instance_id=instance_id
)

# 获取 instance 的 block_size
block_size = self.get_block_size(instance_id)

完整示例

查看现有 converter 的实现：

converters/publisher_log.py - 日志解析示例
converters/qwen_bailian.py - 简单数据集转换
converters/text_trace.py - 复杂的 tokenization 处理

Navigation Menu

Trace Converter - 统一的Trace格式转换工具

✨ 特性

自动发现 Converter

安装

快速开始

Publisher Log转换

Qwen Bailian数据集转换

文本对话转换

输入格式

Publisher Log格式

Qwen Bailian格式

文本格式

输出格式

Optimizer模式 (Get+Write)

命令行参数

通用参数

Tokenizer参数 (text格式需要)

格式特定参数

使用说明

与Optimizer集成

多Instance支持

自动识别 (Publisher Log)

单Instance (Qwen/Text)

前缀哈希算法

合成时间戳策略

扩展：添加自定义 Converter

步骤1: 创建 Converter 类

步骤2: 使用自定义 Converter

Converter 专属配置

Converter 输出接口

BaseConverter 提供的辅助方法

完整示例

相关文档

FilesExpand file tree

README.md

Latest commit

History

README.md

File metadata and controls

Trace Converter - 统一的Trace格式转换工具

✨ 特性

自动发现 Converter

安装

快速开始

Publisher Log转换

Qwen Bailian数据集转换

文本对话转换

输入格式

Publisher Log格式

Qwen Bailian格式

文本格式

输出格式

Optimizer模式 (Get+Write)

命令行参数

通用参数

Tokenizer参数 (text格式需要)

格式特定参数

使用说明

与Optimizer集成

多Instance支持

自动识别 (Publisher Log)

单Instance (Qwen/Text)

前缀哈希算法

合成时间戳策略

扩展：添加自定义 Converter

步骤1: 创建 Converter 类

步骤2: 使用自定义 Converter

Converter 专属配置

Converter 输出接口

BaseConverter 提供的辅助方法

完整示例

相关文档