realtime-audio-architecture

macOS におけるリアルタイムオーディオのアーキテクチャ

Kokoro TTS パイプラインの構築で得られた、macOS Apple Silicon でジッターのない音声再生を実現するための、実戦で検証されたパターンとアンチパターンです。

自己進化型スキル：このスキルは使用を通じて改善されます。手順に誤りがある場合、パラメーターにずれが生じた場合、または回避策が必要になった場合は、先送りにせず、直ちにこのファイルを修正してください。実際に再現できる問題がある場合にのみ更新してください。

判断の枠組み

macOS で Python を使って音声再生を構築するときは、次の優先順位に基づいて選択してください。

1. Write-based sd.OutputStream     ← DEFAULT CHOICE
2. Callback-based sd.OutputStream  ← Only if you need sample-level control
3. afplay subprocess               ← Only for one-shot playback of existing files
4. macOS say                       ← NEVER for production TTS

パターン（推奨）

パターン 1：書き込みベースの sounddevice.OutputStream

Python での音声再生における標準的な選択肢です。 stream.write() は、デバイスバッファーに空きができるまで PortAudio の C コード内でブロックします。音声スレッドでは Python コードが実行されないため、GIL の影響を受けません。

import sounddevice as sd
import numpy as np

def open_audio_stream() -> sd.OutputStream:
    # Refresh PortAudio to discover hot-plugged devices (Bluetooth, HDMI)
    sd._terminate()
    sd._initialize()
    stream = sd.OutputStream(
        samplerate=24000,
        channels=1,
        dtype="float32",
        blocksize=2048,    # ~85ms blocks at 24kHz
        latency="high",    # large internal buffer (not live, so latency is fine)
    )
    stream.start()
    return stream

# Open per request — close after each to follow device changes
stream = open_audio_stream()

# Play audio — blocks in C code, no GIL contention
audio = np.array([...], dtype=np.float32).reshape(-1, 1)
WRITE_BLOCK = 4096  # ~170ms — responsive to stop, smooth playback
for i in range(0, len(audio), WRITE_BLOCK):
    if interrupted:
        break
    stream.write(audio[i:i + WRITE_BLOCK])

stream.close()  # close after request so next open uses current default device

この方法が有効な理由：

• stream.write() は PortAudio の C 層を呼び出すため、音声スレッドでは Python が実行されません
• バッファリング、タイミング、デバイスとのやり取りはすべて PortAudio が内部で処理します
• CPU 負荷の高い処理（MLX 推論、numpy オペレーション）が GIL を保持しても、音声のタイミングに影響しません
• 約 170ms 単位で書き込むことで、すばやく割り込みを確認できます
• デバイスの変更に対応するため、ストリームは起動時ではなくリクエストごとに開きます

停止方法： stream.abort() を呼び出すと、再生を直ちに停止し、write() のブロックを解除します。次の再生時にはストリームを開き直します。

参考： write-based-stream.md

パターン 2：パイプライン合成（N を再生中に N+1 を合成）

チャンク単位の TTS では、合成と再生を並行させます。

from concurrent.futures import ThreadPoolExecutor

with ThreadPoolExecutor(max_workers=1) as pool:
    ahead = pool.submit(synthesize, chunks[0])
    for i in range(len(chunks)):
        audio = ahead.result()
        if i + 1 < len(chunks):
            ahead = pool.submit(synthesize, chunks[i + 1])
        stream.write(audio)  # plays while next chunk synthesizes

理由： チャンクごとの合成には 500～2000ms かかります。パイプライン化しない場合、合成が終わるまでチャンク間に無音が生じます。パイプライン化すれば、通常は再生時間のほうが合成時間より長いため、チャンク N の再生が終わるまでにチャンク N+1 の準備が整います。

パターン 3：ネイティブ形式として Float32 PCM を使用

CoreAudio のネイティブなサンプル形式は 32 ビット浮動小数点です。最初から最後までこの形式を使用します。

# Synthesis output → float32 directly
audio = model.synthesize(text)
if audio.dtype != np.float32:
    audio = audio.astype(np.float32)
    if np.max(np.abs(audio)) > 2.0:  # int16 range
        audio = audio / 32768.0

理由： WAV のエンコード／デコードにかかるオーバーヘッドを回避できます。一時ファイルは不要です。再生時の形式変換もありません。CoreAudio が、最適な形式でデータを受け取れます。

パターン 4：境界でのフェード（2ms）

チャンク境界にごく短いフェードイン／フェードアウトを適用し、クリックノイズを防ぎます。

FADE_SAMPLES = 48  # 2ms at 24kHz

def apply_boundary_fades(audio: np.ndarray) -> np.ndarray:
    if len(audio) < FADE_SAMPLES * 2:
        return audio
    audio = audio.copy()
    audio[:FADE_SAMPLES] *= np.linspace(0, 1, FADE_SAMPLES, dtype=np.float32)
    audio[-FADE_SAMPLES:] *= np.linspace(1, 0, FADE_SAMPLES, dtype=np.float32)
    return audio

理由： 隣接するチャンクでは DC オフセットや位相が異なる場合があります。2ms のフェードは聞こえませんが、不連続によるクリックノイズを防ぎます。チャンク間のクロスフェードよりシンプルで、信頼性も高くなります。

パターン 5：オーディオプロセスに対する launchd QoS

<key>Nice</key>
<integer>-10</integer>
<key>ProcessType</key>
<string>Adaptive</string>

理由：

• Nice: -10 を設定すると、CPU スケジューリングの優先度が高くなります（範囲：-20 が最高、20 が最低）
• ProcessType: Adaptive を設定すると、プロセスが実際に動作している間、macOS が優先度を引き上げられます
• launchd はユーザーエージェントに負の nice 値を設定できます（root として実行されます）

パターン 6：オーディオサーバーの一元化

1 つのサーバー、1 つの発話キューを用意し、すべてのクライアント（BTT、Telegram ボット、CLI）で共有します。

BTT shortcut  →  POST /v1/audio/speak  →  [server queue]  →  synthesize  →  play
Telegram bot  →  POST /v1/audio/speak  →  [server queue]  →  synthesize  →  play

理由： 音声の競合を防げます。ロックプロトコルも 1 つで済みます。調整するプロセスも 1 つです。クライアントはシンプルな HTTP POST 呼び出し元になります。

パターン 7：オーディオデバイスのホットスイッチ

PortAudio は Pa_Initialize() の実行時にデバイス一覧をキャッシュします。後から接続した Bluetooth デバイス（AirPods）は認識されません。2 層の方法で対応します。

def _refresh_audio_devices():
    """Re-init PortAudio to discover hot-plugged devices (~1ms)."""
    sd._terminate()
    sd._initialize()

def open_audio_stream():
    """Open stream with fresh device discovery."""
    _refresh_audio_devices()  # ← discovers AirPods, new HDMI, etc.
    stream = sd.OutputStream(samplerate=24000, channels=1, dtype="float32",
                             blocksize=2048, latency="high")
    stream.start()
    return stream

def maybe_reopen_stream(stream):
    """Between-chunk check for device switching (cached devices only).

    CRITICAL: Do NOT call _refresh_audio_devices() here — it invalidates
    the active stream pointer (PaErrorCode -9988).
    """
    current_default = sd.query_devices(kind='output')['index']
    if stream.device != current_default:
        stream.close()
        return open_audio_stream()
    return stream

2 つの層：

層	タイミング	対応するケース	方法
リクエスト間	ストリームを開くとき	Bluetooth のホットプラグ、HDMI 接続	_refresh_audio_devices() + 新しいストリーム
チャンク間	再生中	認識済みデバイス間の切り替え	キャッシュ済み一覧に対する sd.query_devices()

重要： ストリームがアクティブな間は、絶対に sd._terminate() を呼び出さないでください。PortAudio のすべてのストリームポインターが無効になります。

参考： device-routing.md

アンチパターン（非推奨）

アンチパターン 1：Python キューを使ったコールバックベースの sd.OutputStream

# DON'T — GIL contention causes jitter
def callback(outdata, frames, time_info, status):
    data = audio_queue.get_nowait()  # needs GIL!
    outdata[:, 0] = data

stream = sd.OutputStream(callback=callback, ...)

失敗する理由： コールバックは PortAudio のリアルタイムオーディオスレッドで実行されますが、queue.get_nowait() の実行には Python の GIL を取得する必要があります。MLX による合成（または CPU を集中的に使用する Python の処理）が GIL を保持している間は、たとえ 10ms でもコールバックが遅延し、バッファアンダーランから音声の乱れが発生します。

コールバック自体は C レベルで実行されますが、その中の Python コードには GIL が必要です。 これが根本的な落とし穴です。sounddevice のドキュメントには「コールバックはリアルタイムスレッドで実行される」と記載されています。これは C のラッパーについては正しいものの、その中の Python コードは依然として GIL を取り合います。

アンチパターン 2：チャンクごとにサブプロセスを起動する（afplay）

# DON'T — process spawn + device acquisition per chunk = jitter
for chunk in chunks:
    wav_path = write_temp_wav(chunk)
    subprocess.run(["afplay", wav_path])  # new process each time!
    os.unlink(wav_path)

失敗する理由：

1. プロセス起動のオーバーヘッド： チャンクごとに fork() + exec() を実行します。
2. オーディオデバイスの再取得： afplay は実行ごとにオーディオデバイスを開き、フォーマットをネゴシエートし、再生を開始してから解放します。チャンク間に無音とクリック音が生じます。
3. ファイル I/O のオーバーヘッド： WAV をディスクに書き込み、再び読み込みます。メモリ上に numpy 配列がある場合は不要です。
4. パイプライン化できない： 現在のチャンクの再生中に次のチャンクを合成できません（プロセスがブロックされます）。

afplay が適切な場合： 既存のファイルを一度だけ再生するとき（例：通知音）です。ストリーミングやチャンク単位のオーディオには適しません。

アンチパターン 3：オーディオに launchd のバックグラウンド QoS を使用する

<key>Nice</key>
<integer>5</integer>
<key>ProcessType</key>
<string>Background</string>

失敗する理由： ProcessType: Background は、プロセスがタイムリーな CPU アクセスを必要としないことを macOS に伝えます。その結果、macOS は次のような処理を行う可能性があります。

• CPU スケジューリングの優先度を下げる
• I/O 帯域幅を制限する
• システム負荷が高いときに実行を延期する

オーディオ再生では、他のプロセスが動作しているときに限って発生する、再現が難しい突発的なジッターの原因になります。

アンチパターン 4：TTS のフォールバックとして macOS の say を使用する

# DON'T — quality cliff, unexpected behavior
if ! kokoro_synthesize "$text"; then
    say "$text"  # "fallback"
fi

失敗する理由：

• 音質に大きな差があるため（ロボットのような音声とニューラル音声）、ユーザーが混乱します。
• say はタイミング、音量、動作が異なります。
• 「動くが品質が悪い」という状態になり、明確に失敗する場合よりもデバッグが難しくなります。
• TTS エンジンが複数あると、ロックのプロトコル、プロセス管理、エッジケースも増えます。

代わりに： 通知を出して明確に失敗させます。ユーザーに TTS サーバーが停止していることと、復旧方法を知らせます。

アンチパターン 5：起動時に静的ストリームを開く

# DON'T — stream binds to whatever device was default at process start
stream = sd.OutputStream(samplerate=24000, channels=1, dtype="float32")
stream.start()
# ... reuse forever, never close/reopen

失敗する理由：

1. デバイスへの固定： ストリームは、開いた時点でデフォルトデバイスにバインドされます。その後システムのデフォルトデバイスを切り替えても反映されず、オーディオは以前のデバイスから再生され続けます。
2. launchd の起動タイミング： MacBook のスピーカーがデフォルトになっているログイン時にサーバーが起動することがあります。外部モニターや Bluetooth デバイスは、まだ接続されていない可能性があります。
3. PortAudio のデバイスキャッシュ： Pa_Initialize() はデバイスを一度だけスキャンします。その後に接続した Bluetooth デバイスは認識されず、それらを使ってストリームを開こうとすると、失敗が通知されないか、再生ワーカーがクラッシュします。

代わりに： リクエストごとに必要に応じてストリームを開き、使用後に閉じます。デバイス一覧を更新するため、ストリームを開く前に sd._terminate() と sd._initialize() を呼び出します。

クイック診断

音声にジッターや途切れがある場合：

1. プロセスの優先度を確認します： ps -o pid,nice,pri,command -p $(pgrep -f tts_server)
   • Nice は 0 以下である必要があります（5 以上ではありません）。
2. 再生方法を確認します： grep -c afplay ~/.local/state/launchd-logs/kokoro-tts-server/stdout.log
   • 0 である必要があります（afplay が起動されていない状態）。
3. GIL 競合を確認します： ログで audio callback status: output underflow を探します。
   • 存在する場合 → コールバック方式から write ベースのストリームに切り替えます。
4. launchd の QoS を確認します： plutil -p ~/Library/LaunchAgents/com.terryli.kokoro-tts-server.plist | grep -E 'Nice|ProcessType'
   • Nice: -10、ProcessType: Adaptive である必要があります。

音声が間違ったデバイスから出力される場合：

1. ログでストリームのデバイスを確認します： grep "Audio stream opened" ~/.local/state/launchd-logs/kokoro-tts-server/stdout.log | tail -3
   • 期待するデバイス名が表示される必要があります。
2. PortAudio のエラーを確認します： grep "PaErrorCode\|PortAudio error" ~/.local/state/launchd-logs/kokoro-tts-server/stdout.log | tail -5
   • PaErrorCode -9988 = ストリームポインターが無効化されています（ストリームがアクティブな状態でデバイスが更新されました）。
3. システムのデフォルトを確認します： ~/.local/share/kokoro/.venv/bin/python3 -c "import sounddevice as sd; print(sd.query_devices(kind='output'))"

参考資料

• write ベースのストリーム実装
• launchd QoS のリファレンス
• パイプライン合成パターン
• デバイスのルーティングとホットスイッチ

関連項目

• devops-tools:macbook-desktop-mode — USB デバイスの_耐障害性_（スリープ／復帰後の回復、uhubctl によるポートの切り替え、バッテリーの長寿命化、pmset によるデスクトップ設定）を扱う補完的なスキルです。このスキルはアプリケーション／再生レイヤーを扱い、もう一方のスキルはシステム／USB レイヤーを扱います。

実行後の振り返り

このスキルの実行後、終了前に確認します：

1. コマンドは成功しましたか？ — 成功しなかった場合は、失敗の原因となった手順またはエラーテーブルを修正します。
2. パラメーターや出力に変更はありましたか？ — 基盤となるツールのインターフェースに差異が生じていた場合は、使用例とパラメーターテーブルを一致するように更新します。
3. 回避策が必要でしたか？ — 即興の対応（別のフラグや追加手順）が必要だった場合は、次回の実行で同じ回避策が不要になるよう、この SKILL.md を更新します。

問題が実際に発生し、再現可能な場合にのみ更新してください。憶測に基づいて更新しないでください。