asr — speech recognition

Wire up a private cloud or local recognition engine. Engines are currently mutually exclusive: only one ASR is active at a time.

---

Declaration

{
  "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 }
  ]
}

After enabling it, go to Settings → Speech Recognition and select it as the active engine.

tags determines host behaviour:

• Contains cloud → cloud API (no local model card is attached; we recommend also setting "requires_hf": false to skip the transformers warm-up)
• Contains local / offline → local offline (together with model_artifacts, the host automatically attaches the download card and hardware rating)

---

Minimal implementation

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 mono, 16 kHz, range [-1, 1]
        return Transcript(text="识别文字", language=language)

Hard signature constraints (see Provider signatures):

• with_alignment must appear in the signature, defaulting to False. If you have no word-level aligner, ignore it and leave Transcript.words as None. Do not pass this parameter through to a lower-level client that takes fewer parameters.
• on_token is a streaming incremental callback; ignore it if you don't support it.

---

Capabilities and streaming

@dataclass
class ASRCapabilities:
    has_native_vad: bool = False             # The engine segments sentences / detects endpoints itself, no forced external splitting
    supports_streaming: bool = False         # Whether real-time streaming recognition over a long-lived connection is supported
    supports_word_timestamps: bool = False   # Whether word-level timestamps are supported
    supports_language_detection: bool = True # Whether automatic language detection is supported

Streaming interface (optional; all have default implementations):

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]]   # (text, is_final)
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            # default False
@property
def supports_concurrent_streams(self) -> bool     # default True

• When both tracks (upper track peer / lower track self) stream at the same time, the two sessions must be able to coexist (independent handles, weights may be shared). If the underlying client is single-handle, return supports_concurrent_streams = False and the host will automatically degrade the other track.
• Not supporting streaming is fine too: the host splits sentences by VAD and then calls transcribe.

---

Local offline ASR (important)

For a local recognition plugin the host automatically takes over model download, hardware rating and storage management — provided you declare model_artifacts + tags: ["local"] in mod.json:

• Automatically attaches HardwareRatingBanner (compute rating) and LocalModelStorageCard (mirror / proxy / progress / clear cache)
• Put only inference hyper-parameters (device / port / sampling parameters…) in config_schema; never hand-write your own "model download path / domestic mirror switch / download button"

{
  "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"]
    }
  ]
}

The full field table is in Model artifacts. To get the files:

info = self.host.inspect_model_artifact(self.manifest.id, "custom-asr-weights")
root = info["target_dir"]      # Always present (usable even before download)
real = info["model_dir"]       # The directory where the weights were actually found; "" when not ready

If your plugin starts its own local inference service (llama-server / a custom HTTP service), you must read the Local-service engine contract (lazy startup / port fallback / shutdown kills child processes).

---

Reference implementations

Directory	Type
extensions/community_cloud_mimo_asr/	Cloud API
extensions/community_cloud_dashscope_asr/	Cloud API (with a custom settings view)
extensions/workshop_common_qwen3_asr/	Local service type (llama.cpp, complete contract example)

---

Related

• Provider signatures & data types
• Local-service engine contract
• Model artifacts
• Capabilities & read-only contract