开发指南

从理解框架架构到贡献你自己的模型——循序渐进的完整指南。

1. 架构概览

FunASR 围绕三个核心概念构建:注册表(Registry)用于组件发现,AutoModel 作为统一入口,config.yaml 作为模型的声明式定义。

用户代码 FunASR 框架 模型仓库 ───────── ──────────────── ───────── AutoModel(model="name") │ ├─→ download_model() ──────→ ModelScope / HuggingFace │ ↓ ↓ │ 读取 config.yaml 下载 model.pt │ ↓ ├─→ tables.model_classes["Name"] ← @tables.register 装饰器 │ ↓ ├─→ model_class(**config) ← __init__: 构建 encoder/decoder │ ↓ ├─→ load_pretrained_model() ← 从 model.pt 加载权重 │ ↓ └─→ model.eval() ← 准备就绪,可以推理 generate(input="audio.wav") │ ├─ 无 VAD → inference() ← 单条语音 │ ↓ │ model.inference(data_in, tokenizer, frontend, **kwargs) │ ↓ │ 返回 [{"key", "text", "timestamp", ...}] │ └─ 有 VAD → inference_with_vad() ← 长音频 ↓ 1. VAD:切分音频 → [[起始ms, 结束ms], ...] 2. 按长度排序(提高批处理效率) 3. ASR:逐段识别 4. 合并时间戳(加上 VAD 偏移量) 5. 标点恢复(可选) 6. 说话人分离(可选) ↓ 返回 [{"key", "text", "timestamp", "sentence_info"}]

注册表系统

FunASR 中的每个组件都通过名称注册。注册表是将配置字符串映射到 Python 类的查找表:

注册表用途示例
model_classesASR、VAD、标点、说话人模型"Paraformer", "FsmnVADStreaming"
encoder_classes编码器架构"SANMEncoder", "ConformerEncoder"
decoder_classes解码器架构"ParaformerSANMDecoder"
frontend_classes音频特征提取"WavFrontend", "WhisperFrontend"
tokenizer_classes文本分词"SentencepiecesTokenizer"
dataset_classes训练数据加载"AudioDataset"
from funasr.register import tables

# 注册一个新模型
@tables.register("model_classes", "MyModel")
class MyModel(nn.Module):
    ...

# 查看所有已注册的模型
tables.print("model")

2. 开发环境配置

克隆并以开发模式安装

git clone https://github.com/modelscope/FunASR.git
cd FunASR
pip install -e .              # 可编辑安装
pip install -e ".[train]"     # 包含训练依赖

验证安装

python -c "from funasr import AutoModel; print('OK')"
python -c "from funasr.register import tables; tables.print('model')"

运行已有测试

# 快速冒烟测试
python tests_models/test_fsmn_vad.py
python tests_models/test_paraformer.py

# 完整测试套件
cd tests_models && python run_all_tests.py

3. 推理流程详解

在添加新模型之前,必须理解推理的数据流。以下是调用 model.generate(input="audio.wav") 时的完整流程:

步骤 1:输入准备

prepare_data_iterator() 将各种输入类型(文件路径、URL、numpy 数组、bytes、列表)统一为 (key_list, data_list) 格式。

步骤 2:模型推理

每个模型的 inference() 方法接收以下参数:

def inference(self, data_in, data_lengths=None, key=None,
              tokenizer=None, frontend=None, **kwargs):
    # data_in: 音频样本列表(numpy 数组)
    # tokenizer: 将 token ID 解码为文本
    # frontend: 提取 fbank 特征
    # **kwargs: config.yaml 中的所有参数 + 用户运行时参数

    # 必须返回:(结果列表, 元数据字典)
    return [{"key": "id", "text": "hello", "timestamp": [...]}], {"batch_data_time": 5.5}

步骤 3:输出格式

results_list 必须是字典列表。必填/可选字段:

字段类型是否必填说明
keystr样本标识符
textstr是(ASR)识别文本
timestamplist说话人分离时需要[[起始ms, 结束ms], ...] 逐字时间戳
valuelist仅 VAD[[起始ms, 结束ms], ...] 有声段
spk_embeddingTensor仅说话人模型形状 [N, 192]
时间戳格式很重要!如果你的模型输出的时间戳是字典格式(如 Fun-ASR-Nano 的 {"start_time": 0.5, "end_time": 0.8}),FunASR 的 inference_with_vad 会自动转换。但标准的期望格式是 [[起始ms, 结束ms], ...](毫秒级的二元列表)。

4. 添加新模型

创建模型目录

funasr/models/my_model/
├── __init__.py      # 空文件
├── model.py         # 主模型类
├── encoder.py       # (可选)自定义编码器
└── decoder.py       # (可选)自定义解码器
规则:每个模型目录必须是自包含的。不要从其他模型目录导入。不要修改已有模型。

实现模型类

import torch.nn as nn
from funasr.register import tables

@tables.register("model_classes", "MyModel")
class MyModel(nn.Module):

    def __init__(self, **kwargs):
        super().__init__()    # ← 必须调用 super().__init__()
        # 从 kwargs 构建你的网络结构(kwargs 来自 config.yaml)
        # kwargs 包含:input_size, vocab_size, tokenizer, frontend 等

    def forward(self, speech, speech_lengths, text, text_lengths, **kwargs):
        """训练前向传播。返回 (loss, stats_dict, weight)。"""
        ...

    def inference(self, data_in, data_lengths=None, key=None,
                  tokenizer=None, frontend=None, **kwargs):
        """推理。返回 (results_list, meta_data)。"""
        ...

创建 config.yaml

# 该文件定义使用哪些组件
model: MyModel              # 与 @tables.register 中的名称对应
model_conf:
    hidden_size: 512

frontend: WavFrontend       # 复用已有的前端
frontend_conf:
    fs: 16000
    n_mels: 80
    frame_length: 25
    frame_shift: 10
    cmvn_file: null

tokenizer: SentencepiecesTokenizer
tokenizer_conf:
    bpemodel: null

创建 configuration.json(用于上传到 Hub)

{
  "framework": "pytorch",
  "task": "auto-speech-recognition",
  "model": {"type": "funasr"},
  "file_path_metas": {
    "init_param": "model.pt",
    "config": "config.yaml",
    "tokenizer_conf": {"bpemodel": "my_tokenizer.model"},
    "frontend_conf": {"cmvn_file": "am.mvn"}
  }
}

该文件告诉 AutoModel 如何解析相对路径。模型下载后,file_path_metas 中的每个路径都会加上模型目录前缀。

本地测试

from funasr import AutoModel

# 从本地目录加载
model = AutoModel(model="./my_model_dir")
res = model.generate(input="test.wav")
print(res)

5. 添加新的前端 / 分词器 / 数据集

所有组件都使用相同的注册表模式。以添加新前端为例:

from funasr.register import tables

@tables.register("frontend_classes", "MyFrontend")
class MyFrontend(nn.Module):
    def __init__(self, fs=16000, **kwargs):
        super().__init__()
        self.fs = fs

    def output_size(self):
        return 80  # 特征维度

    def forward(self, input, input_lengths):
        # input: 原始波形 (batch, samples)
        # 返回: 特征 (batch, frames, dim), 长度
        ...

然后在 config.yaml 中引用:

frontend: MyFrontend
frontend_conf:
    fs: 16000

分词器(tokenizer_classes)、数据集(dataset_classes)、编码器(encoder_classes)等都是同样的模式。

6. 独立仓库模式

你的模型不需要放在 FunASR 源码树内。通过 trust_remote_code=True,FunASR 可以从外部文件动态加载模型类:

# 用户代码 — 从独立仓库加载你的 model.py
model = AutoModel(
    model="your-org/your-model",      # HuggingFace/ModelScope 仓库
    trust_remote_code=True,
    remote_code="./model.py",          # 模型类定义文件路径
    hub="hf",
)

工作原理:

  1. FunASR 下载模型仓库(权重 + 配置 + model.py)
  2. remote_code="./model.py" 被动态导入
  3. 该文件中的 @tables.register 装饰器将模型类注册到注册表
  4. 后续正常执行 build_model() 流程

你的仓库结构:

your-model-repo/
├── model.py              # 带 @tables.register 的模型类
├── config.yaml           # 模型架构配置
├── configuration.json    # 路径解析配置
├── model.pt              # 训练好的权重
└── example/test.wav      # 示例音频

参考实现:Fun-ASR-NanoSenseVoice

7. 测试你的模型

编写测试脚本

# tests_models/test_my_model.py
import sys, time
from funasr import AutoModel

def main():
    model = AutoModel(model="path/to/model", device="cpu", disable_update=True)
    res = model.generate(input="test.wav")

    assert res and len(res) > 0, "结果为空"
    assert "text" in res[0], "缺少 text 字段"
    print("PASSED")
    return 0

if __name__ == "__main__":
    sys.exit(main())

测试 VAD + 说话人分离流水线

# 如果你的模型需要支持说话人分离:
model = AutoModel(
    model="path/to/model",
    vad_model="fsmn-vad",
    spk_model="cam++",
)
res = model.generate(input="meeting.wav", cache={})
assert "sentence_info" in res[0]
assert "spk" in res[0]["sentence_info"][0]

测试流式推理(如果支持)

cache = {}
for i in range(total_chunks):
    chunk = audio[i*stride:(i+1)*stride]
    res = model.generate(input=chunk, cache=cache,
                         is_final=(i == total_chunks-1), ...)
# 验证:相同音频在多次会话中应产生相同结果

8. 常见陷阱

❌ 忘记调用 super().__init__()

# 错误 — 会导致 "object has no attribute '_state_dict_pre_hooks'"
class MyEncoder(nn.Module):
    def __init__(self):
        pass

# 正确
class MyEncoder(nn.Module):
    def __init__(self):
        super().__init__()

❌ 在模型中检查 kwargs["batch_size"]

kwargs 中的 batch_sizeinference_with_vad 用于分段批处理的值(一个很大的毫秒数)。不要用它来判断实际的数据批量大小,应该使用 len(data_in)

❌ 未处理空输入或极短输入

VAD 可能产生空的片段。你的 inference() 应该能优雅地处理 data_in = [] 的情况。

❌ 时间戳格式不匹配

如果你的模型返回字典格式的时间戳({"start_time": 0.5, "end_time": 0.8}),流水线会自动转换。但如果你输出 [start_time, end_time, text](3 个元素),需要去掉文本——下游期望的是 [起始ms, 结束ms](2 个元素,毫秒单位)。

❌ 从其他模型目录导入

# 错误 — 产生紧耦合
from funasr.models.paraformer.model import Paraformer

# 正确 — 将需要的代码复制到自己的目录
# 或者在 config.yaml 中通过注册名引用

❌ 在推理过程中修改 self.kwargs

不要改动 AutoModel 传入的 kwargs。框架会在每次调用之间重置状态,但持久性的修改可能在不同会话间泄漏。

9. 贡献代码

代码风格

PR 检查清单

许可证

代码:MIT 许可。模型权重:FunASR 模型许可(允许商业使用,需注明出处)。