Anechoのスピーカーアイソレーションを Pipecat のオーディオフィルタとして使うには？
トランスポートの受信音声を16 kHz・240サンプルのホップに再分割し、各ホップにAnecho SDKを適用する小さなPipecatのBaseAudioFilterを書きます。公開プラグインではなく自分で書くコードです。フィルタの全体と数値を以下に示します。
PipecatでAnechoのスピーカーアイソレーションを使うには、トランスポートの受信音声を16 kHz・240サンプルのホップに再分割し、各ホップにAnecho SDKを適用する小さな BaseAudioFilter を書きます。 当社はPipecat用パッケージを提供していません。Pipecatのオーディオフィルタ枠にはすでにKrisp VIVA、ai-coustics、RNNoise、Koala、noisereduceがあります（docs.pipecat.ai、2026-10-15取得）。当社のものは以下のコードで、あなたが所有し、調整できます。
置く場所
Pipecatのトランスポートは audio_in_filter を受け付け、VADとSTTの前に生の受信バイトを見ます。連続ストリームで動かす必要があるモデルに適した場所です。SDKの契約：Processor.process() はfloat32のブロックを受け取り、同じ数のサンプルを15 ms遅れで返します。最初の240サンプルは無音（アルゴリズム遅延）です。UsageReporter が送るのは時間だけで、音声は送りません。
フィルタ
Pipecatはフィルタに、トランスポートのレートで任意サイズのint16バイト列を渡します。モデルは16 kHzの240サンプルホップを求め、フィルタは受け取ったのと同じバイト数を返す必要があるため、両側でバッファリングします。
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"),
),
)
これは BaseAudioFilter インターフェース（start、stop、process_frame、filter）に対するスケッチです。稼働中のPipecatパイプラインでのベンチマークはしていないため、お使いのPipecatバージョンでインターフェースを確認し、自社の音声で試してください。出力は入力の長さまでバッファされるため、モデルの15 msに加えて最大1ホップ分のバッファが加わります。
補足
- フィルタは16 kHz入力を必須とし、それ以外ではエラーにします。トランスポートの
audio_in_sample_rateを設定するか、手前でリサンプリングしてください。8 kHzの電話音声をアップサンプリングしても帯域は戻らず、当社の実行ではアイソレーションモデルは電話条件をわずかに悪化させました。8 kHzは音声AIが壊れる場所を参照してください。 - プロセッサが状態を持つため、フィルタは通話ごとに1つ作ります。
- Pipecatには雑音で学習したターン検出モデルもあります。アイソレーションとターン検出は別の故障モードを扱います（なぜ音声エージェントは騒がしい通話で失敗するのか）。
計測結果
| 項目 | 値 | 出典と日付 |
|---|---|---|
| アルゴリズム遅延 | 15 ms | Null Test、2026-08-14 |
| 実時間係数、CPU 1スレッド | 0.067 | Null Test、2026-08-14（Apple M1 Max） |
| 最悪ブロック時間、p99 | 3.96 ms | Null Test、2026-08-14 |
| 0 dBの2人の声、WER 未処理 対 当社 | 84.74% 対 74.21% | Null Test、2026-08-14 |
| 19条件のプール、WER 未処理 対 当社 | 14.32% 対 16.59% | Null Test、2026-08-14 |
プールの行は悪化なので、これは計測したうえで、デプロイごとに有効にするフィルタです：話者分離の評価方法。
FAQ
Pipecat用のAnechoパッケージはありますか？
いいえ。公開パッケージはありません。上のフィルタは、Pipecatの BaseAudioFilter インターフェースに対して自分で書くコードです。
Pipecatのどの拡張ポイントを使いますか？
トランスポートの audio_in_filter です。VADとSTTの前に受信音声を見ます。
なぜフィルタはバッファリングするのですか？
モデルは16 kHzの240サンプルホップで動作する一方、Pipecatはトランスポートのレートで任意サイズのチャンクを渡し、フィルタは受け取ったのと同じバイト数を返す必要があるためです。