Provider 签名与数据类型

宿主与所有调用方只允许传基类参数。底层客户端参数更少时，在实现内部适配；禁止把多余 kwargs 透传下去。

MT

python

def translate(
    self, text: str,
    source_lang: str = "en", target_lang: str = "zh",
    glossary_context: str | None = None,
    target: str | None = None, source: str | None = None,
) -> Translation: ...

宿主门面 host.translate(text, source, target, glossary_context=None) 与 host.translate_batch(...) 会把 glossary_context 传给当前活跃 MT。

可选增量接口（默认回退到整句）：

python

def translate_partial(
    self, text, source_lang="en", target_lang="zh",
    committed_prefix="", glossary_context=None,
    target=None, source=None,
) -> Translation: ...

ASR

python

def transcribe(self, audio, sample_rate=16000, language="auto", stream_id="default",
               with_alignment=False, on_token=None) -> Transcript: ...

with_alignment 默认 False，签名里必须有。没有词级对齐器的引擎收下这个参数并忽略，此时 Transcript.words 为 None（内置 Qwen3 已去掉 ForcedAligner，不会因为该参数去跑对齐）。
调用方不要把 with_alignment= 透传给参数更少的底层 transcribe。

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

python

def stream_open(self, stream_id: str, language: str = "auto") -> Any
def stream_feed(self, stream_id: str, pcm_chunk) -> 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

能力声明：

python

@dataclass
class ASRCapabilities:
    has_native_vad: bool = False
    supports_streaming: bool = False
    supports_word_timestamps: bool = False
    supports_language_detection: bool = True

TTS

python

class BaseTTSProvider(BasePlugin):
    supports_voice_clone: bool = False   # 支持注册克隆音色
    supports_speed: bool = False         # 原生支持语速

    @property
    def sample_rate(self) -> int: ...            # 默认 24000
    @property
    def is_streaming_capable(self) -> bool: ...  # 默认 False

    def synthesize(self, text, voice_id="", voice=None, language="zh"
                   ) -> tuple[np.ndarray, int]: ...
    def stream_tts(self, text, voice_id="", voice=None, language="zh"
                   ) -> Iterator[np.ndarray]: ...
    def list_voices(self) -> list[dict]: ...
    def register_voice(self, voice_name, audio_samples, reference_text="",
                       sample_rate=24000) -> bool: ...
    def remove_voice(self, voice_name: str) -> bool: ...

项	约定
`synthesize`	返回 `(float32 单声道 PCM, 采样率)`；`voice` 优先于 `voice_id`（宿主可能只传其一）
`stream_tts`	可选。实现了就按块 `yield`（宿主按预缓冲逐块播，起播更早）；未实现时宿主退化为整段 `synthesize`
`list_voices()`	→ `[{"id", "name", "lang", "kind"}]`，`kind` ∈ `speaker`（模型自带） / `registered`（用户克隆，只有它能被删除）。只返回真实可用项；服务未启动时返回缓存/空表
`register_voice(voice_name, audio_samples, reference_text, sample_rate)`	`audio_samples` = float32 单声道 numpy（1~30 秒）。宿主负责读入音频（用户选文件，或设置页内软件内录制）。插件只负责转码/上传/落盘，返回 `True/False`
`remove_voice(name)`	删除已注册克隆音色，返回 `True/False`

语速：

声明 supports_speed = True → 宿主把 speed=（1.0 = 原速，范围 0.5~2.0）传给你，你自己在模型内实现。
未声明 → 宿主在合成后统一做保音高时间伸缩，你什么都不用做。
speed != 1.0 且未声明 supports_speed 时，宿主不会调用 stream_tts。

音色配置键：宿主把用户选中的音色写进插件配置的 voice（分语种时还有 voice_en / voice_other，缺省回落 voice）。实现里读 self._config.get("voice")。

克隆音色持久化（强制）：服务端只存内存的话必须自己落盘 + 服务就绪后自动补注册，详见 tts § 克隆音色的持久化。

interpretation

python

def start_session(self, session_id="peer", source_lang="zh", target_lang="en",
                  voice=None, tts_enabled=False) -> bool
def feed_audio(self, session_id: str, pcm_chunk) -> None
def stop_session(self, session_id: str = "peer") -> None
def is_active(self, session_id: str = "peer") -> bool
# 向上吐结果（基类实现）：emit_transcript / emit_translation / emit_tts_audio / emit_status / emit_error

详见 interpretation。

数据类型

python

@dataclass
class WordTime:
    word: str
    start: float
    end: float

@dataclass
class Transcript:
    text: str
    language: str | None = None
    words: list[WordTime] | None = None

@dataclass
class Translation:
    text: str
    source_prefix: str | None = None
    complete: bool = True     # False 表示增量片段

@dataclass
class PipelineContext:        # 中间件钩子的 context
    session_id: str = ""
    sample_rate: int = 16000
    source: str = "peer"      # 'peer' | 'self'
    source_lang: str = "auto"
    target_lang: str = "zh"
    extra: dict = {}

生命周期钩子（所有 Provider 通用）

钩子	什么时候被调	你该做什么
`initialize(config)`	建实例时（外部插件跑在后台线程）	只做准备：解析配置、查文件；不启动服务
`ensure_loaded()`	宿主「真正要用」时（开音轨、启流水线、点测试连接）	拉起本地服务并等就绪（可阻塞）
`update_config(delta)`	用户改配置	热同步；只有服务在跑才重启服务
`check_ready()`	就绪门禁	判断「能不能用」；服务没起 ≠ 未就绪
`shutdown()`	退出 / 释放显存 / 禁用 / 改配置	必须 kill 子进程
`wait_async_init()`	宿主在调 `ensure_loaded` 前后	基类提供，等异步 `initialize` 收敛

细节见本地服务型引擎强制契约。

Provider 签名与数据类型 ​

MT ​

ASR ​

TTS ​

interpretation ​

数据类型 ​

生命周期钩子（所有 Provider 通用） ​

相关 ​

Provider 签名与数据类型

MT

ASR

TTS

interpretation

数据类型

生命周期钩子（所有 Provider 通用）

相关