asr — 语音识别

接私有云端或本地识别引擎。当前引擎互斥：同一时间只有一个 ASR 生效。

声明

json

{
  "id": "myteam.asr.deeplx",
  "name": "My ASR",
  "version": "1.0.0",
  "target_app_version": "0.2",
  "type": "asr",
  "entrypoint": "my_asr.py:MyASRProvider",
  "tags": ["cloud"],
  "requires_hf": false,
  "config_schema": [
    { "key": "api_key", "label": "API Key", "type": "password", "is_secret": true }
  ]
}

启用后到 设置 → 语音识别 选为当前引擎。

tags 决定宿主行为：

含 cloud → 云端 API（不挂本地模型卡；建议同时 "requires_hf": false 跳过 transformers 预热）
含 local / offline → 本地离线（配合 model_artifacts 自动挂上下载卡与硬件评级）

最小实现

python

import numpy as np
from transbox.plugins.base import BaseASRProvider, ASRCapabilities
from transbox.asr.engine import Transcript


class CustomASRProvider(BaseASRProvider):
    @property
    def capabilities(self) -> ASRCapabilities:
        return ASRCapabilities(has_native_vad=False, supports_streaming=False)

    def transcribe(self, audio: np.ndarray, sample_rate: int = 16000,
                   language: str = "auto", stream_id: str = "default",
                   with_alignment: bool = False, on_token=None) -> Transcript:
        # audio: float32 单声道，16 kHz，范围 [-1, 1]
        return Transcript(text="识别文字", language=language)

签名硬约束（见 Provider 签名）：

with_alignment 必须出现在签名里，默认 False。没有词级对齐器就忽略它，Transcript.words 留 None。不要把这个参数透传给参数更少的底层客户端。
on_token 是流式增量回调，不支持就忽略。

能力与流式

python

@dataclass
class ASRCapabilities:
    has_native_vad: bool = False             # 引擎自带断句/端点检测，无需外部强制切句
    supports_streaming: bool = False         # 是否支持实时长连接流式识别
    supports_word_timestamps: bool = False   # 是否支持词级时间戳
    supports_language_detection: bool = True # 是否支持自动语言检测

流式接口（可选，全部有默认实现）：

python

def stream_open(self, stream_id: str, language: str = "auto") -> Any
def stream_feed(self, stream_id: str, pcm_chunk: np.ndarray) -> Any
def stream_reopen(self, stream_id: str, language: str = "auto") -> Any
def stream_events(self, stream_id: str) -> Iterator[Tuple[str, bool]]   # (文本, 是否定稿)
def stream_close(self, stream_id: str) -> str
def has_active_stream(self, stream_id: str) -> bool
def close_all_streams(self) -> None

@property
def is_streaming_capable(self) -> bool            # 默认 False
@property
def supports_concurrent_streams(self) -> bool     # 默认 True

双轨（上轨 peer / 下轨 self）同时走流式时，两条会话必须能并存（各自独立句柄、权重可共享）。底层是单活句柄就返回 supports_concurrent_streams = False，宿主会自动降级另一轨。
不支持流式也没关系：宿主会按 VAD 切句后调 transcribe。

本地离线 ASR（重要）

本地识别插件宿主会自动接管模型下载、硬件评级与存储管理——前提是你在 mod.json 声明 model_artifacts + tags: ["local"]：

自动挂 HardwareRatingBanner（算力评级）与 LocalModelStorageCard（镜像/代理/进度/清缓存）
config_schema 里只放推理超参（device / port / 采样参数…），严禁再手写「模型下载路径 / 国内镜像开关 / 下载按钮」

json

{
  "tags": ["asr", "local", "offline", "cpu-friendly"],
  "requires_cuda": false,
  "model_artifacts": [
    {
      "id": "custom-asr-weights",
      "name": "Custom-ASR-0.6B",
      "category": "asr",
      "repo": "org/Custom-ASR-0.6B-GGUF",
      "est_bytes": 1018000000,
      "sources": ["huggingface", "modelscope"],
      "ready_globs": ["Custom-ASR-Q8_0.gguf", "mmproj*.gguf"],
      "allow_patterns": ["Custom-ASR-Q8_0.gguf", "mmproj*.gguf"]
    }
  ]
}

字段全表见模型工件。取文件用：

python

info = self.host.inspect_model_artifact(self.manifest.id, "custom-asr-weights")
root = info["target_dir"]      # 恒有值（未下载也能用）
real = info["model_dir"]       # 实际命中权重的目录；未就绪时是 ""

若你的插件自己起本地推理服务（llama-server / 自研 HTTP），必须读本地服务型引擎强制契约（惰性启动 / 端口顺延 / shutdown 杀子进程）。

参考实现

目录	类型
`extensions/community_cloud_mimo_asr/`	云端 API
`extensions/community_cloud_dashscope_asr/`	云端 API（含自定义设置页）
`extensions/community_qwen3_asr/`	本地服务型（llama.cpp，完整契约范例）

asr — 语音识别 ​

声明 ​

最小实现 ​

能力与流式 ​

本地离线 ASR（重要） ​

参考实现 ​

相关 ​

asr — 语音识别

声明

最小实现

能力与流式

本地离线 ASR（重要）

参考实现

相关