Como usar o isolamento de locutor da Anecho como filtro de áudio do Pipecat?

Você usa o isolamento de locutor da Anecho no Pipecat escrevendo um pequeno BaseAudioFilter que reagrupa o áudio de entrada do transporte em saltos de 240 amostras a 16 kHz e roda o SDK da Anecho em cada salto. Não distribuímos um pacote do Pipecat. O slot de filtros de áudio do Pipecat já traz Krisp VIVA, ai-coustics, RNNoise, Koala e noisereduce (docs.pipecat.ai, consultado em 2026-10-15); o nosso é o código abaixo, que é seu e você pode adaptar.

Onde ele fica

Um transporte do Pipecat aceita audio_in_filter, que vê os bytes de entrada brutos antes do VAD e do STT. É o lugar certo para um modelo que precisa rodar sobre um fluxo contínuo. O contrato do SDK: Processor.process() recebe um bloco float32 e devolve o mesmo número de amostras 15 ms depois; as primeiras 240 amostras são silêncio (o atraso algorítmico). O UsageReporter envia durações, nunca áudio.

O filtro

O Pipecat entrega ao filtro pedaços de bytes int16 de tamanho arbitrário na taxa do transporte. O modelo quer saltos de 240 amostras a 16 kHz e o filtro deve devolver tantos bytes quanto recebe, então ele armazena em buffer dos dois 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"),
    ),
)

É um esboço contra a interface BaseAudioFilter (start, stop, process_frame, filter). Não medimos um pipeline Pipecat em produção, então confira a interface na sua versão do Pipecat e teste com o seu próprio áudio. Como a saída é bufferizada até o tamanho da entrada, o filtro adiciona os 15 ms do modelo mais até um salto de buffer.

Notas

• O filtro exige entrada a 16 kHz e falha caso contrário, então ajuste audio_in_sample_rate do transporte ou reamostre antes. Subir telefonia de 8 kHz não recupera largura de banda, e na nossa execução o modelo de isolamento piorou levemente as condições de telefonia; veja 8 kHz é onde a IA de voz quebra.
• Crie um filtro por chamada, pois o processador guarda estado.
• O Pipecat também tem modelos de detecção de turno treinados para ruído; isolamento e detecção de turno corrigem modos de falha diferentes (por que agentes de voz falham em chamadas ruidosas).

O que medimos

Item	Valor	Fonte e data
Atraso algorítmico	15 ms	Null Test, 2026-08-14
Fator de tempo real, uma thread de CPU	0.067	Null Test, 2026-08-14 (Apple M1 Max)
Pior tempo de bloco, p99	3.96 ms	Null Test, 2026-08-14
Duas vozes a 0 dB, WER bruto vs nosso	84.74% vs 74.21%	Null Test, 2026-08-14
Agregado de 19 condições, WER bruto vs nosso	14.32% vs 16.59%	Null Test, 2026-08-14

A linha agregada é uma perda, por isso este é um filtro que se ativa por implantação, depois de medir: como avaliar isolamento de locutor.

FAQ

Existe um pacote da Anecho para o Pipecat?

Não. Não há pacote publicado. O filtro acima é código que você escreve contra a interface BaseAudioFilter do Pipecat.

Qual ponto de extensão do Pipecat eu uso?

O audio_in_filter do transporte, que vê o áudio de entrada antes do VAD e do STT.

Por que o filtro usa buffer?

O modelo trabalha em saltos de 240 amostras a 16 kHz enquanto o Pipecat entrega pedaços de tamanho arbitrário na taxa do transporte, e o filtro deve devolver o mesmo número de bytes que recebe.