¿Cómo usar el aislamiento de hablante de Anecho como filtro de audio de Pipecat?

Se usa el aislamiento de hablante de Anecho en Pipecat escribiendo un pequeño BaseAudioFilter que reagrupa el audio entrante del transporte en saltos de 240 muestras a 16 kHz y ejecuta el SDK de Anecho en cada salto. No distribuimos un paquete de Pipecat. El hueco de filtros de audio de Pipecat ya incluye Krisp VIVA, ai-coustics, RNNoise, Koala y noisereduce (docs.pipecat.ai, consultado el 2026-10-15); el nuestro es el código de abajo, que es tuyo y puedes adaptar.

Dónde se sitúa

Un transporte de Pipecat acepta audio_in_filter, que ve los bytes entrantes sin procesar antes del VAD y del STT. Es el lugar adecuado para un modelo que debe correr sobre un flujo continuo. El contrato del SDK: Processor.process() toma un bloque float32 y devuelve el mismo número de muestras 15 ms después; las primeras 240 muestras son silencio (el retardo algorítmico). El UsageReporter envía duraciones, nunca audio.

El filtro

Pipecat entrega al filtro fragmentos de bytes int16 de tamaño arbitrario a la frecuencia del transporte. El modelo quiere saltos de 240 muestras a 16 kHz y el filtro debe devolver tantos bytes como recibe, así que almacena en búfer a ambos lados.

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"),
    ),
)

Es un boceto contra la interfaz BaseAudioFilter (start, stop, process_frame, filter). No hemos medido un pipeline de Pipecat en vivo, así que comprueba la interfaz con tu versión de Pipecat y prueba con tu propio audio. Como la salida se almacena hasta la longitud de la entrada, el filtro añade los 15 ms del modelo más hasta un salto de búfer.

Notas

• El filtro exige entrada a 16 kHz y falla en caso contrario, así que ajusta audio_in_sample_rate del transporte o remuestrea antes. Subir telefonía de 8 kHz no recupera ancho de banda, y en nuestra ejecución el modelo de aislamiento empeoró ligeramente las condiciones de telefonía; ver 8 kHz es donde la IA de voz se rompe.
• Crea un filtro por llamada, ya que el procesador guarda estado.
• Pipecat también tiene modelos de detección de turnos entrenados para ruido; el aislamiento y la detección de turnos arreglan modos de fallo distintos (por qué fallan los agentes de voz en llamadas ruidosas).

Lo que medimos

Elemento	Valor	Fuente y fecha
Retardo algorítmico	15 ms	Null Test, 2026-08-14
Factor de tiempo real, un hilo de CPU	0.067	Null Test, 2026-08-14 (Apple M1 Max)
Peor tiempo de bloque, p99	3.96 ms	Null Test, 2026-08-14
Dos voces a 0 dB, WER sin procesar vs nuestro	84.74% vs 74.21%	Null Test, 2026-08-14
Agregado de 19 condiciones, WER sin procesar vs nuestro	14.32% vs 16.59%	Null Test, 2026-08-14

La fila agregada es una pérdida, por eso este es un filtro que se activa por despliegue, tras medir: cómo evaluar el aislamiento de hablante.

FAQ

¿Hay un paquete de Anecho para Pipecat?

No. No hay paquete publicado. El filtro de arriba es código que escribes contra la interfaz BaseAudioFilter de Pipecat.

¿Qué punto de extensión de Pipecat se usa?

El audio_in_filter del transporte, que ve el audio entrante antes del VAD y del STT.

¿Por qué almacena en búfer el filtro?

El modelo trabaja en saltos de 240 muestras a 16 kHz mientras Pipecat entrega fragmentos de tamaño arbitrario a la frecuencia del transporte, y el filtro debe devolver el mismo número de bytes que recibe.