minicpm-o5-sdk 0.0.3

MiniCPMO5 training data protocol toolkit

MiniCPMO5 SDK

MiniCPMO5 SDK 是面向 MiniCPM-O 系列双工（duplex）训练数据协议的 Python 工具包：把"一通对话"的多轨观测序列化成可训练的 token 流，并提供电子书式的可视化教程。

你写的对话 / 观测
      ↓        构造五轨 TrainingData（schema 强校验）
O5DuplexTrainingData
      ↓        arrange()  把 start_trigger 解到 unit 网格
O5DuplexArrangement
      ↓        tokenize(tokenizer)  五轨 slot 模板展开
O5TokenizedDuplexData
      ↓        可视化（5 视图）
Guidebook / Dataset Viewer

当前版本 0.0.3：只覆盖数据 + token 协议 + 质检 + 可视化。 训练支持在 v0.1，推理支持在 v0.2。详见 CHANGELOG.md。

安装

pip install minicpm-o5-sdk
# 或从源码 editable 安装：
pip install -e .

依赖：python ≥ 3.10、pydantic ≥ 2.11、torch ≥ 2.0、transformers ≥ 5.0、tokenizers ≥ 0.20.3、jsonschema、watchfiles。

快速上手

五轨概览（每个 unit 内严格按此顺序展开 token）

Track	类型	说明
user_video	timed	用户视频帧（v0.0.1 不实现 content tokenize）
user_audio	timed	用户语音 + 可选字级 alignment
input_event	text	外部观测事件：tool_response / tool_created（框架注入）/ standalone event
ai_spoken	timed	AI 语音输出，字级 alignment 必填且 schema 校验拼接一致
ai_non_spoken	text	AI 思考 / 工具调用（text + think + tool_call）

直接加载内置 Case

SDK 随 wheel 分发一组 Guidebook 原子案例，可直接导入为 O5DuplexTrainingData：

from minicpm_o5_sdk import O5CaseID, O5TokenizerID, load_case

training_data = load_case(O5CaseID.CASE_1_02)
result = training_data.tokenize(tokenizer_id=O5TokenizerID.O5)

脚本场景也可以用字符串 ID：

from minicpm_o5_sdk import load_case

training_data = load_case("case_1_02")

可用 ID 通过 list_case_ids() 查看；当前发布包内置 13 个 active case。

源码仓库中另有 examples/load_builtin_case.ipynb 作为交互式教程；notebook 不随 wheel 打包，pip 用户可直接使用上面的代码片段或启动 Guidebook。

构造一条最小 TrainingData（用户问 → AI 答）

import torch
from minicpm_o5_sdk import (
    O5AISpokenSegment, O5AISpokenTrack, O5AINonSpokenTrack,
    O5Alignment, O5DuplexTrainingData, O5DuplexTrainingTracks,
    O5GlobalTime, O5InputEventTrack, O5MediaSegmentTime,
    O5LazyAudio, O5SystemContent, O5SystemTextSegment,
    O5StartTrigger, O5UnitPolicy, O5UserAudioSegment,
    O5UserAudioTrack, O5WordInterval,
)

def alignment_for(text: str, dur: float) -> O5Alignment:
    return O5Alignment(word_intervals=[O5WordInterval(text=text, start_sec=0.0, end_sec=dur)])

training_data = O5DuplexTrainingData(
    unit_policy=O5UnitPolicy(
        unit_sec=1.0,
        non_spoken_budgets_while_listening=[100],
        non_spoken_budgets_while_speaking=[100],
    ),
    system=O5SystemContent(segments=[O5SystemTextSegment(text="你是一个友好的中文 AI 助手。")]),
    tracks=O5DuplexTrainingTracks(
        user_video=None,
        user_audio=O5UserAudioTrack(segments=[
            O5UserAudioSegment(
                audio=O5LazyAudio(
                    duration_sec=1.2,
                    get_tensor_fn=lambda: torch.zeros(int(1.2 * 16000), dtype=torch.float32),
                ),
                start_trigger=O5StartTrigger(refs=[O5GlobalTime()]),
                transcript="你好",
            ),
        ]),
        input_event=O5InputEventTrack(segments=[]),
        ai_spoken=O5AISpokenTrack(segments=[
            O5AISpokenSegment(
                audio=O5LazyAudio(
                    duration_sec=0.8,
                    get_tensor_fn=lambda: torch.zeros(int(0.8 * 16000), dtype=torch.float32),
                ),
                start_trigger=O5StartTrigger(refs=[
                    O5MediaSegmentTime(
                        track="user_audio",
                        segment_index=0,
                        at="end",
                        offset_sec=0.3,
                    ),
                ]),
                text="你好",
                alignment=alignment_for("你好", 0.8),
            ),
        ]),
        ai_non_spoken=O5AINonSpokenTrack(segments=[]),
    ),
)

Tokenize

from minicpm_o5_sdk import O5TokenizerID

result = training_data.tokenize(tokenizer_id=O5TokenizerID.O5)

assert result.tokenized_data.tokenizer_target == "o5"
for t in result.tokenized_data.token_provenance[:30]:
    print(t.token_text, t.track, t.unit_index, t.trainable)

Tokenizer target：O5 和 O45_FC 是当前 active line；O45 仅作为 deprecated shim 保留。两者基础词表和 token id 不可互换，调用方必须显式选 target：

from minicpm_o5_sdk import O5TokenizerID

result = training_data.tokenize(tokenizer_id=O5TokenizerID.O45_FC)

推理代码指导示例

如果你要在推理代码里拼 system/tool prefill、解析 tool_call、写回 tool_response， 可参考源码仓库中的 examples/inference_runtime_guide.ipynb；该 notebook 不随 wheel 打包。

Token 查表与资料卡

如果你要确认当前 tokenizer target 下某个 token 的 id、语义 key、track、loss 或 hf_added_special，可参考源码仓库中的 examples/token_lookup_guide.ipynb；该 notebook 不随 wheel 打包。

它演示：

tok.token_to_id("<|speak|>")
tok.id_to_token(248103)
tok.tokens_to_ids(["<ai_spoken_slot>", "<|speak|>"])
tok.ids_to_tokens([248161, 248103])
tok.token_info("<tool_call>")
tok.list_special_tokens(track="ai_non_spoken")

启动 Guidebook（电子书式教程，随 wheel 分发）

python -m minicpm_o5_sdk.visualizers.duplex.guidebook \
    --host 0.0.0.0 --port 10035 \
    --tokenizer-id o5

浏览器打开 http://localhost:10035。涵盖：

• Ch1 为什么需要双工
• Ch2 Token 协议
• Ch3 TrainingData 五轨
• Ch5 原子案例（当前 13 个 active case，含真实 TTS+CTC alignment）

启动 Dataset Viewer（任意数据集浏览）

python -m minicpm_o5_sdk.visualizers.duplex.dataset_viewer \
    /path/to/your/cases \
    --port 10036 --tokenizer-id o5 \
    --case-cache-size 100

每个 case 提供 5 视图：raw / aligned / tokenized / provenance / issues。--tokenizer-id 默认是 o5；如需按 O45_FC token id 查看，改为 --tokenizer-id o45_fc。

协议要点

• unit_sec 协议自由度：协议层允许任意值，O5 训练规范主线 1s，少数 case 用 0.5s。
• start_trigger 时序声明：每个 segment 用 O5StartTrigger(refs=[...]) 表达启动条件；常用 ref 是 O5GlobalTime、O5MediaSegmentTime 和 O5TextSegmentTime，多 ref 是 AND 语义。
• alignment 必填：ai_spoken.alignment 是 schema 强 invariant，且 "".join(wi.text for wi in word_intervals) == text 严格相等（中文天然字级，英文把空格放进 wi.text）。
• ai_spoken start_unit_index 是 perceived ceil+1：模型解码 ai_spoken 比物理时间晚一个 unit 才被感知到，与 input_event 在感知延迟上对称。timeline_start_sec/end_sec 保留物理真相。
• token 模板每 unit 五轨 slot 固定顺序：user_video → user_audio → input_event → ai_spoken → ai_non_spoken。AI 两个输出 slot 必出现（空时分别走 <|listen|> / <|no_action|>）。

开发（贡献者）

cd minicpm_o5_sdk
PYTHONPATH=src python -m pytest tests/protocols/duplex -q
# 120 passed

PYTHONPATH=src python -m mypy src/minicpm_o5_sdk/protocols/duplex/

python -m build --wheel --outdir dist

发版前检查：docs/before_release_checklist.md。 0.0.3 发布说明：docs/releases/0.0.3.md。 飞书 latest 使用入口本地源：docs/feishu/latest_usage_entry.md。 0.0.1 历史里程碑：docs/0.0.1-milestone.md。 变更日志：CHANGELOG.md。 v0.0.1 之后的剩余 follow-up：docs/future-work.md。

当前不支持

• 多模态 event content（音频/图像）序列化：v0.0.1 入口抛 NotImplementedError。
• Track-level tensor 合成（如 O5UserAudioTrack.build_tensor()）：后续 v0.1 训练 collator 对接时设计。
• 训练 / 推理脚本：分别在 v0.1 / v0.2 引入。
• abort（pending lifecycle）：v0.0.1 已保留 <|non_spoken_abort|> token 和 abort_trigger schema，但暂不在 arranger/tokenization 中生产 abort token。

类型提示

wheel 内包含 py.typed，IDE / pyright / mypy 可以识别 SDK 自带的 inline type hints。常用 public API 推荐从顶层导入，例如 from minicpm_o5_sdk import O5DuplexTrainingData, load_case；内部实现路径后续重构时会通过顶层 facade 保持用户侧 import 稳定。

版权与许可

Author / maintainer: Weiyue Sun [e-mail].

Copyright (c) 2026 ModelBest Inc.

本 SDK 使用 MIT License 发布。详见 LICENSE。