Wie nutzt man die Anecho-Sprecherisolation als Pipecat-Audiofilter?

Sie nutzen die Anecho-Sprecherisolation in Pipecat, indem Sie einen kleinen BaseAudioFilter schreiben, der das eingehende Audio des Transports in 240-Sample-Hops bei 16 kHz neu zerlegt und das Anecho-SDK auf jeden Hop anwendet. Wir liefern kein Pipecat-Paket aus. Pipecats Audiofilter-Slot enthält bereits Krisp VIVA, ai-coustics, RNNoise, Koala und noisereduce (docs.pipecat.ai, abgerufen am 2026-10-15); unserer ist der Code unten, der Ihnen gehört und den Sie anpassen können.

Wo er sitzt

Ein Pipecat-Transport akzeptiert audio_in_filter, der die rohen eingehenden Bytes vor VAD und STT sieht. Das ist der richtige Ort für ein Modell, das auf einem kontinuierlichen Strom laufen muss. Der SDK-Vertrag: Processor.process() nimmt einen float32-Block und gibt gleich viele Samples 15 ms später zurück; die ersten 240 Samples sind Stille (die algorithmische Verzögerung). Der UsageReporter sendet Dauern, nie Audio.

Der Filter

Pipecat reicht dem Filter int16-Bytechunks beliebiger Größe in der Rate des Transports. Das Modell will 240-Sample-Hops bei 16 kHz, und der Filter muss so viele Bytes zurückgeben, wie er erhält; deshalb puffert er auf beiden Seiten.

import os
import numpy as np
from pipecat.audio.filters.base_audio_filter import BaseAudioFilter
from pipecat.frames.frames import FilterControlFrame
from anecho import Model, Processor, UsageReporter

RATE, HOP = 16_000, 240          # model rate; 240 samples = 15 ms


class AnechoFilter(BaseAudioFilter):
    """Speaker isolation on the transport's inbound audio. Our code, not a package."""

    def __init__(self, model_path: str):
        self._model = Model.from_file(model_path)
        self._reporter = UsageReporter(os.environ["ANECHO_API_KEY"])
        self._proc = None
        self._pending = np.zeros(0, dtype=np.float32)    # input not yet a full hop
        self._out = np.zeros(HOP, dtype=np.float32)      # one-hop prefill keeps output ahead of input

    async def start(self, sample_rate: int):
        if sample_rate != RATE:                          # resample at the transport, not here
            raise ValueError(f"AnechoFilter needs 16 kHz input, got {sample_rate}")
        self._proc = Processor(self._model, self._reporter.ensure_license())
        self._reporter.attach(self._proc)
        self._reporter.start()

    async def stop(self):
        self._proc = None

    async def process_frame(self, frame: FilterControlFrame):
        pass                                             # no runtime controls

    async def filter(self, audio: bytes) -> bytes:
        x = np.frombuffer(audio, dtype=np.int16).astype(np.float32) / 32768.0
        self._pending = np.concatenate([self._pending, x])
        while len(self._pending) >= HOP:
            block, self._pending = self._pending[:HOP], self._pending[HOP:]
            self._out = np.concatenate([self._out, self._proc.process(block)])
        y, self._out = self._out[: len(x)], self._out[len(x):]   # same byte count out as in
        return (np.clip(y, -1.0, 1.0) * 32767).astype(np.int16).tobytes()

transport = SmallWebRTCTransport(
    webrtc_connection=conn,
    params=TransportParams(
        audio_in_enabled=True,
        audio_in_filter=AnechoFilter("anecho.ai_focus_model_16khz_v8_2.anecho"),
    ),
)

Das ist eine Skizze gegen das BaseAudioFilter-Interface (start, stop, process_frame, filter). Wir haben keine laufende Pipecat-Pipeline gebenchmarkt; prüfen Sie das Interface gegen Ihre Pipecat-Version und testen Sie mit eigenem Audio. Da die Ausgabe auf die Eingabelänge gepuffert wird, fügt der Filter die 15 ms des Modells plus bis zu einen Hop Pufferung hinzu.

Hinweise

• Der Filter verlangt 16-kHz-Eingang und wirft sonst einen Fehler; setzen Sie audio_in_sample_rate des Transports oder resamplen Sie davor. Hochrechnen von 8-kHz-Telefonie stellt keine Bandbreite wieder her, und in unserem Lauf verschlechterte das Isolationsmodell die Telefonie-Bedingungen leicht; siehe 8 kHz ist, wo Sprach-KI bricht.
• Erzeugen Sie einen Filter pro Anruf, da der Prozessor Zustand hält.
• Pipecat hat auch auf Lärm trainierte Turn-Detection-Modelle; Isolation und Turn-Detection beheben verschiedene Fehlermodi (warum Sprachagenten bei lauten Anrufen scheitern).

Was wir gemessen haben

Posten	Wert	Quelle und Datum
Algorithmische Verzögerung	15 ms	Null Test, 2026-08-14
Echtzeitfaktor, ein CPU-Thread	0.067	Null Test, 2026-08-14 (Apple M1 Max)
Schlechteste Blockzeit, p99	3.96 ms	Null Test, 2026-08-14
Zwei Stimmen bei 0 dB, WER roh vs unsere	84.74% vs 74.21%	Null Test, 2026-08-14
Gepoolt über 19 Bedingungen, WER roh vs unsere	14.32% vs 16.59%	Null Test, 2026-08-14

Die gepoolte Zeile ist ein Verlust, deshalb ist dies ein Filter, den man pro Deployment nach dem Messen aktiviert: wie man Sprecherisolation bewertet.

FAQ

Gibt es ein Anecho-Paket für Pipecat?

Nein. Es gibt kein veröffentlichtes Paket. Der Filter oben ist Code, den Sie gegen das Pipecat-Interface BaseAudioFilter schreiben.

Welchen Pipecat-Erweiterungspunkt nutze ich?

Den audio_in_filter des Transports, der das eingehende Audio vor VAD und STT sieht.

Warum puffert der Filter?

Das Modell arbeitet in 240-Sample-Hops bei 16 kHz, während Pipecat Chunks beliebiger Größe in der Transportrate liefert, und der Filter muss so viele Bytes zurückgeben, wie er erhält.