realtime-audio-architecture

macOS 上的实时音频架构

在 macOS Apple Silicon 上实现无抖动音频播放的实战模式与反模式，这些经验来自构建 Kokoro TTS 流水线。

自我演进技能：本技能会在使用过程中不断改进。如果指令有误、参数发生漂移，或需要采用变通方案，请立即修复此文件，不要拖延。仅针对真实且可复现的问题进行更新。

决策框架

在 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

原因： 相邻分块的直流偏移或相位可能不同。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：集中式音频服务

一个服务、一个语音队列，供所有客户端（BTT、Telegram 机器人、CLI）共享：

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

原因： 防止音频冲突。统一锁协议。只需调优一个进程。客户端只是轻量级的 HTTP POST 调用方。

模式 7：音频设备热切换

PortAudio 会在 Pa_Initialize() 时缓存设备列表。之后连接的蓝牙设备（AirPods）不会显示。采用双层策略：

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

两个层级：

层级	时机	适用情况	机制
请求之间	打开流时	蓝牙设备热插拔、连接 HDMI	_refresh_audio_devices() + 新建流
分块之间	播放过程中	在已知设备之间切换	在缓存列表上调用 sd.query_devices()

重要： 流处于活动状态时，切勿调用 sd._terminate() —— 这会使所有 PortAudio 流指针失效。

参考： device-routing.md

反模式（不推荐）

反模式 1：基于回调的 sd.OutputStream 与 Python 队列

# 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 会告诉 macOS，该进程不需要及时访问 CPU。macOS 将会：

• 降低 CPU 调度优先级
• 限制 I/O 带宽
• 系统负载较高时可能会推迟执行

对于音频播放，这会导致偶发且难以复现的抖动——只有其他进程处于活动状态时才会发生。

反模式 4：将 macOS say 用作 TTS 后备方案

# 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 Speakers 可能是默认设备。外接显示器或蓝牙设备尚未连接。
3. PortAudio 设备缓存：Pa_Initialize() 只扫描一次设备。之后连接的蓝牙设备不可见，尝试打开对应流会静默失败，或导致播放工作线程崩溃。

**建议做法：**按需为每个请求打开流，完成后关闭。打开前调用 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. 参数或输出是否发生变化？ — 如果底层工具的接口有所变化，请相应更新 Usage 示例和 Parameters 表。
3. 是否需要变通处理？ — 如果你不得不临时采用其他做法（不同的标志、额外步骤），请更新此 SKILL.md，确保下次执行时无需再采用相同的变通方法。

仅在问题真实且可复现时才进行更新，不要更新推测性问题。