middleware — pipeline middleware

Attach interceptors at any stage of capture → VAD → ASR → MT → TTS. Multiple instances allowed, ordered by priority.

---

Declaration

{
  "id": "community.mod.text_filter",
  "name": "流水线文字过滤与润色中间件",
  "version": "1.0.0",
  "target_app_version": "0.2",
  "type": "middleware",
  "entrypoint": "filter_middleware.py:TextFilterMiddleware",
  "enabled": false
}

Unlike engine types, a middleware plugin is attached to the pipeline as soon as it is enabled (there is no "select as the active engine" step).

---

Base class and hooks

from transbox.plugins.base import BasePipelineMiddleware, Transcript, Translation

class TextFilterMiddleware(BasePipelineMiddleware):
    @property
    def priority(self) -> int:
        return 50  # Lower runs first; default 100 (can also be overridden by a `priority` option in config_schema)

    # 1. Audio before VAD / ASR (float32 PCM)
    def process_audio(self, audio, context):
        return audio

    # 2. ASR result (before it is handed to MT)
    def process_transcript(self, transcript: Transcript, context) -> Transcript:
        return Transcript(
            text=transcript.text.replace("那个...", ""),
            language=transcript.language,
        )

    # 3. MT translation result (before it goes to the UI and TTS)
    def process_translation(self, translation: Translation, context) -> Translation:
        return Translation(
            text=translation.text.replace("坏词", "***"),
            complete=translation.complete,
        )

    # 4. TTS synthesized audio (before sound-card playback) — returns (pcm, sample_rate)
    def process_tts_audio(self, pcm_audio, sample_rate, context):
        return pcm_audio, sample_rate

Execution order: process_audio → process_transcript → process_translation → process_tts_audio.

Every hook has a default pass-through implementation; override only the ones you need.

---

The context PipelineContext

@dataclass
class PipelineContext:
    session_id: str = ""
    sample_rate: int = 16000
    source: str = "peer"          # 'peer' (the other party / system sound) | 'self' (my microphone)
    source_lang: str = "auto"
    target_lang: str = "zh"
    extra: dict = {}

priority can also be overridden through plugin config: the base class reads self._config.get("priority", 100) by default.

---

Notes

• Hooks sit on the hot path for audio/inference, so they must be fast. Do not put heavy work (network, models) here; if you really need it, make it async and accept that "the current sentence is unaffected".
• Return values must have the correct type: process_tts_audio returns a tuple (pcm, sample_rate), not a single array.
• In dual-track scenarios context.source distinguishes peer / self; branch on it when the two need different handling.

---

Related

• Provider signatures & data types — Transcript / Translation / PipelineContext
• Runtime architecture & lifecycle — dual-track routing
• Example library