middleware — 管线中间件
在
捕获 → VAD → ASR → MT → TTS任意阶段挂拦截器。可多开,按priority排序。
声明
json
{
"id": "community.mod.text_filter",
"name": "流水线文字过滤与润色中间件",
"version": "1.0.0",
"target_app_version": "0.2",
"type": "middleware",
"entrypoint": "filter_middleware.py:TextFilterMiddleware",
"enabled": false
}引擎类之外,middleware 启用后即挂进流水线(不需要「选为当前引擎」)。
基类与钩子
python
from transbox.plugins.base import BasePipelineMiddleware, Transcript, Translation
class TextFilterMiddleware(BasePipelineMiddleware):
@property
def priority(self) -> int:
return 50 # 越小越先执行,默认 100(也可用 config_schema 的 priority 配置项覆盖)
# 1. VAD / ASR 之前的音频(float32 PCM)
def process_audio(self, audio, context):
return audio
# 2. ASR 识别结果(送 MT 之前)
def process_transcript(self, transcript: Transcript, context) -> Transcript:
return Transcript(
text=transcript.text.replace("那个...", ""),
language=transcript.language,
)
# 3. MT 翻译结果(送往 UI 与 TTS 之前)
def process_translation(self, translation: Translation, context) -> Translation:
return Translation(
text=translation.text.replace("坏词", "***"),
complete=translation.complete,
)
# 4. TTS 合成音频(声卡播放之前)—— 返回 (pcm, sample_rate)
def process_tts_audio(self, pcm_audio, sample_rate, context):
return pcm_audio, sample_rate执行顺序:process_audio → process_transcript → process_translation → process_tts_audio。
每个钩子都有默认透传实现,只覆盖你需要的那几个即可。
上下文 PipelineContext
python
@dataclass
class PipelineContext:
session_id: str = ""
sample_rate: int = 16000
source: str = "peer" # 'peer'(对方/系统声音) | 'self'(我方麦克风)
source_lang: str = "auto"
target_lang: str = "zh"
extra: dict = {}priority 也可通过插件配置覆盖:基类默认读 self._config.get("priority", 100)。
注意
- 钩子在音频/推理的热路径上,务必快。重活(网络、模型)不要放在这里, 真需要就异步化并接受"当前这一句不生效"。
- 返回值必须类型正确:
process_tts_audio返回的是 元组(pcm, sample_rate),不是单个数组。 - 双轨场景下
context.source会区分peer/self,需要区别对待时按它分支。
相关
- Provider 签名与数据类型 —
Transcript/Translation/PipelineContext - 运行架构与生命周期 — 双轨分流
- 示例库
