オーディオシステム技術ドキュメント

オーディオシステムの概要

ASR-LLM-TTS バックエンドのオーディオシステムは、音声認識の入力、音声合成の出力、BGM の再生を担当します。多層アーキテクチャ設計を採用し、複数デバイスの管理、同時再生の制御、インテリジェントなデバイス切り替えに対応しています。

️ アーキテクチャ設計

オーディオシステムのアーキテクチャ図

┌─────────────────────────────────────────────────────────────┐
│                    音频系统架构                              │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  ┌─────────────────┐    ┌─────────────────┐                │
│  │   ASR Service   │    │  TTS Player     │                │
│  │  (语音识别)     │    │  (语音合成)     │                │
│  └─────────┬───────┘    └─────────┬───────┘                │
│            │                      │                        │
│            ▼                      ▼                        │
│  ┌─────────────────┐    ┌─────────────────┐                │
│  │  Microphone     │    │ AudioNewPlayer  │                │
│  │  (麦克风输入)   │    │ (音频播放器)    │                │
│  └─────────────────┘    └─────────┬───────┘                │
│                                   │                        │
│                         ┌─────────▼───────┐                │
│                         │ Audio Device    │                │
│                         │ Manager         │                │
│                         │ (设备管理器)    │                │
│                         └─────────┬───────┘                │
│                                   │                        │
│            ┌──────────────────────┼──────────────────────┐ │
│            │                      │                      │ │
│            ▼                      ▼                      ▼ │
│  ┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐ │
│  │  Music Player   │    │  TTS Output     │    │  System Audio   │ │
│  │  (背景音乐)     │    │  (语音输出)     │    │  (系统音频)     │ │
│  └─────────────────┘    └─────────────────┘    └─────────────────┘ │
│            │                      │                      │ │
│            ▼                      ▼                      ▼ │
│  ┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐ │
│  │   pygame mixer  │    │   sounddevice   │    │   PulseAudio    │ │
│  │   (音乐播放)    │    │   (实时播放)    │    │   (系统音频)    │ │
│  └─────────────────┘    └─────────────────┘    └─────────────────┘ │
└─────────────────────────────────────────────────────────────────┘

主要コンポーネント

AudioNewPlayer リアルタイム音声プレーヤー

主な特長

• 非同期ストリーミング再生：非同期ジェネレーターからの音声データ受信に対応
• インテリジェントなバッファリング戦略：プリバッファリングの仕組みによりスムーズな再生を確保
• ステートマシンによる管理：再生状態をきめ細かく追跡
• 音量制御：リアルタイムの音量調整
• デバイスへの自動適応：最適なオーディオデバイスを自動選択

プレーヤーのステートマシン

class PlayerState(Enum):
    IDLE = "idle"                    # 空闲状态
    BUFFERING = "buffering"          # 缓冲中
    PLAYING = "playing"              # 播放中
    PAUSED = "paused"                # 暂停
    STOPPING = "stopping"            # 停止中
    STOPPED = "stopped"              # 已停止
    ERROR = "error"                  # 错误状态

非同期再生の実装

class AudioNewPlayer:
    async def play_stream(self, audio_generator, sample_rate: int, channels: int = 1):
        """流式播放音频（完全异步）"""
        try:
            # 1. 初始化播放队列
            self._audio_queue = queue.Queue(maxsize=self.buffer_size)
            self._set_state(PlayerState.BUFFERING)

            # 2. 启动音频输入任务
            input_task = asyncio.create_task(
                self._feed_audio_queue(audio_generator)
            )

            # 3. 等待预缓冲完成
            await self._wait_for_prebuffer_or_error()

            # 4. 创建音频输出流
            self._stream = sd.OutputStream(
                samplerate=sample_rate,
                channels=channels,
                dtype=np.float32,
                callback=self._audio_callback,
                blocksize=self.blocksize,
                device=self.device,
                finished_callback=self._stream_finished_callback
            )

            # 5. 开始播放
            self._stream.start()
            self._set_state(PlayerState.PLAYING)

            # 6. 等待播放完成
            await self._wait_for_playback_finished_or_stopped()

        except Exception as e:
            logger.error(f"播放过程中发生错误: {e}", exc_info=True)
            self._set_state(PlayerState.ERROR)
            raise AudioPlayerError(f"播放失败: {e}")
        finally:
            await self._cleanup_playback()

バッファリング戦略

プリバッファリングの仕組み：

async def _wait_for_prebuffer_or_error(self):
    """等待预缓冲完成或出错"""
    while True:
        if self._error_event.is_set():
            error_msg = self._error_message or "未知播放错误"
            raise AudioPlayerError(f"预缓冲阶段出错: {error_msg}")

        queue_size = self._audio_queue.qsize()
        if queue_size >= self.min_prebuffer:
            logger.info(f"预缓冲完成，队列大小: {queue_size}")
            break

        await asyncio.sleep(0.01)  # 短暂等待

オーディオコールバック関数：

def _audio_callback(self, outdata, frames, time, status):
    """sounddevice音频回调函数"""
    if status:
        logger.warning(f"音频回调状态警告: {status}")

    try:
        # 从队列中获取音频数据
        if not self._audio_queue.empty():
            audio_chunk = self._audio_queue.get_nowait()

            # 应用音量控制
            if self._volume != 1.0:
                audio_chunk = audio_chunk * self._volume

            # 填充输出缓冲区
            if len(audio_chunk) == frames:
                outdata[:] = audio_chunk.reshape(-1, 1)
            else:
                # 处理长度不匹配的情况
                outdata[:] = self._resize_audio_chunk(audio_chunk, frames)
        else:
            # 队列为空，输出静音
            outdata.fill(0)

    except queue.Empty:
        outdata.fill(0)
    except Exception as e:
        logger.error(f"音频回调函数错误: {e}")
        outdata.fill(0)

️ AudioDeviceManager デバイスマネージャー

デバイスタイプの定義

class AudioDeviceType(Enum):
    TTS_PLAYER = "tts_player"        # TTS播放器
    MUSIC_PLAYER = "music_player"    # 背景音乐播放器
    SYSTEM_AUDIO = "system_audio"    # 系统音频

デバイスアクセス権の管理

class AudioDeviceManager:
    def __init__(self):
        self._device_locks: Dict[AudioDeviceType, asyncio.Lock] = {
            device_type: asyncio.Lock() for device_type in AudioDeviceType
        }
        self._active_devices: Dict[AudioDeviceType, bool] = {}
        self._volume_callbacks: Dict[AudioDeviceType, Callable] = {}
        self._original_volumes: Dict[AudioDeviceType, float] = {}

    def request_device_access_sync(
        self, 
        device_type: AudioDeviceType, 
        device_info: dict,
        force_stop_others: bool = False
    ) -> bool:
        """同步方式请求设备访问权限"""
        try:
            if force_stop_others:
                # 停止其他设备的播放
                for other_type in AudioDeviceType:
                    if other_type != device_type and self._active_devices.get(other_type):
                        self._stop_device_sync(other_type)

            # 标记设备为活跃状态
            self._active_devices[device_type] = True
            logger.info(f"设备 {device_type.value} 获得访问权限")
            return True

        except Exception as e:
            logger.error(f"请求设备访问权限失败: {e}")
            return False

    def release_device_access_sync(self, device_type: AudioDeviceType):
        """同步方式释放设备访问权限"""
        try:
            self._active_devices[device_type] = False

            # 恢复其他设备的原始音量
            for other_type in AudioDeviceType:
                if other_type != device_type:
                    original_volume = self._original_volumes.get(other_type)
                    if original_volume is not None:
                        callback = self._volume_callbacks.get(other_type)
                        if callback:
                            callback(original_volume)

            logger.info(f"设备 {device_type.value} 释放访问权限")

        except Exception as e:
            logger.error(f"释放设备访问权限失败: {e}")

音量の協調制御の仕組み

def register_volume_callback(
    self, 
    device_type: AudioDeviceType, 
    callback: Callable[[float], None]
):
    """注册音量控制回调"""
    self._volume_callbacks[device_type] = callback
    logger.info(f"注册 {device_type.value} 的音量控制回调")

def set_original_volume(self, device_type: AudioDeviceType, volume: float):
    """设置设备的原始音量"""
    self._original_volumes[device_type] = volume
    logger.info(f"设置 {device_type.value} 原始音量: {volume}")

def update_music_player_volume(self, volume: float):
    """更新音乐播放器音量并记录"""
    self.set_original_volume(AudioDeviceType.MUSIC_PLAYER, volume)
    logger.info(f"音乐播放器音量已更新: {volume}")

MusicPlayer BGM システム

pygame mixer の初期化

def _init_pygame_mixer(self):
    """初始化pygame mixer，动态尝试多个设备"""
    potential_devices = [
        'UACDemoV1.0',      # RDKX5 USB喇叭描述名
        'UACDemoV10',       # 替代描述名
        'hw:0,0',           # ALSA硬件设备
        'plughw:0,0',       # ALSA插件设备
        None                # 系统默认设备
    ]

    os.environ['SDL_AUDIODRIVER'] = 'pulse'

    for device in potential_devices:
        try:
            logger.info(f"尝试初始化设备: {device or '默认'}")

            pygame.mixer.pre_init(
                frequency=44100,
                size=-16,
                channels=2,
                buffer=4096
            )

            if device:
                pygame.mixer.init(devicename=device)
            else:
                pygame.mixer.init()

            logger.info(f"✅ pygame mixer 初始化成功: {device or '默认'}")
            return

        except pygame.error as e:
            logger.warning(f"设备 {device} 初始化失败: {e}")
            pygame.mixer.quit()

    raise RuntimeError("无法初始化pygame mixer")

インテリジェントな楽曲マッチング

def _find_music_by_name(self, music_name: str, music_list: List[str]) -> Optional[str]:
    """智能音乐名称匹配"""
    # 1. 完全匹配
    if music_name in music_list:
        return music_name

    # 2. 去扩展名匹配
    name_without_ext = os.path.splitext(music_name)[0]
    for music_file in music_list:
        if os.path.splitext(music_file)[0] == name_without_ext:
            return music_file

    # 3. 前缀匹配
    for music_file in music_list:
        if os.path.splitext(music_file)[0].startswith(name_without_ext):
            return music_file

    # 4. 包含匹配
    for music_file in music_list:
        if name_without_ext in os.path.splitext(music_file)[0]:
            return music_file

    return None

再生監視の仕組み

def _monitor_music_playback(self):
    """监控音乐播放状态，播放结束时释放设备权限"""
    while not self._stop_monitor.wait(1.0):  # 每秒检查一次
        try:
            if not pygame.mixer.music.get_busy():
                # 音乐播放结束
                logger.info("检测到音乐播放结束")
                self.current_music = None

                # 释放设备访问权限
                self._release_device_access_sync()

                # 恢复原始音量
                pygame.mixer.music.set_volume(self._volume)
                break

        except Exception as e:
            logger.error(f"监控播放状态错误: {e}")
            break

ASR 音声認識システム

マルチプロセスアーキテクチャ

# 在server.py中启动ASR服务
if input_mode == "asr":
    asr_config_obj = config_loader.get_asr_config()
    if asr_config_obj:
        asr_result_queue = multiprocessing.Queue(maxsize=20)

        asr_service_process = multiprocessing.Process(
            target=run_asr_service,
            args=(asr_config_obj, None, asr_result_queue),
            name="ASRServiceProcess",
            daemon=True
        )
        asr_service_process.start()

        # 启动监听任务
        asr_listener_task = asyncio.create_task(
            listen_and_forward_asr_results()
        )

VAD による発話区間検出

def run_asr_service(asr_config: ASRConfig, audio_queue, result_queue):
    """ASR服务主函数"""
    try:
        # 1. 初始化VAD模型
        vad_model = init_vad_model(asr_config.vad_model_path)

        # 2. 初始化ASR模型  
        asr_recognizer = init_asr_recognizer(asr_config)

        # 3. 初始化音频输入
        microphone = init_microphone(asr_config)

        logger.info("ASR服务启动完成")

        # 4. 主处理循环
        audio_buffer = []
        vad_state = VadState.SILENCE

        for audio_chunk in microphone.stream():
            # VAD检测
            is_speech = vad_model.predict(audio_chunk)

            if vad_state == VadState.SILENCE and is_speech:
                # 检测到语音开始
                vad_state = VadState.SPEECH
                audio_buffer = [audio_chunk]
                logger.debug("检测到语音开始")

            elif vad_state == VadState.SPEECH:
                audio_buffer.append(audio_chunk)

                if not is_speech:
                    # 检测到语音结束
                    vad_state = VadState.SILENCE

                    # 进行ASR识别
                    if len(audio_buffer) > MIN_AUDIO_LENGTH:
                        audio_data = np.concatenate(audio_buffer)
                        recognition_result = asr_recognizer.recognize(audio_data)

                        if recognition_result.strip():
                            result_queue.put(recognition_result)
                            logger.info(f"ASR识别结果: {recognition_result}")

                    audio_buffer = []

    except Exception as e:
        logger.error(f"ASR服务运行错误: {e}", exc_info=True)
        result_queue.put(None)  # 发送错误信号

オーディオ設定ツール

対話型設定ツール

# utils/audio_config/setup_audio_devices.py
class AudioDeviceSetup:
    def __init__(self):
        self.config_manager = AudioDeviceConfigManager()

    def run_interactive_setup(self):
        """交互式音频设备配置"""
        print("🎵 音频设备配置工具")
        print("=" * 50)

        while True:
            self._display_menu()
            choice = input("请选择操作: ").strip()

            if choice == "1":
                self._scan_and_display_devices()
            elif choice == "2":
                self._configure_output_device()
            elif choice == "3":
                self._configure_input_device()
            elif choice == "4":
                self._test_audio_output()
            elif choice == "5":
                self._show_current_config()
            elif choice == "0":
                break
            else:
                print("❌ 无效选择，请重试")

    def _scan_and_display_devices(self):
        """扫描并显示音频设备"""
        print("\n🔍 正在扫描音频设备...")

        input_devices = self.config_manager.get_input_devices()
        output_devices = self.config_manager.get_output_devices()

        print(f"\n📥 发现 {len(input_devices)} 个输入设备:")
        for i, device in enumerate(input_devices):
            status = "✅ 活跃" if device.get('active') else "⭕ 非活跃"
            print(f"  {i+1}. {device['name']} {status}")

        print(f"\n📤 发现 {len(output_devices)} 个输出设备:")
        for i, device in enumerate(output_devices):
            status = "✅ 活跃" if device.get('active') else "⭕ 非活跃"
            print(f"  {i+1}. {device['name']} {status}")

デバイス状態のチェック

# utils/audio_config/check_audio_devices.py
def check_audio_status():
    """检查音频设备状态"""
    print("🔊 音频设备状态检查")
    print("=" * 40)

    try:
        # 1. 检查PulseAudio状态
        print("\n1️⃣ PulseAudio 服务状态:")
        pa_status = subprocess.run(['pulseaudio', '--check'], 
                                 capture_output=True, text=True)
        if pa_status.returncode == 0:
            print("   ✅ PulseAudio 服务正常运行")
        else:
            print("   ❌ PulseAudio 服务未运行")

        # 2. 检查默认设备
        print("\n2️⃣ 默认音频设备:")

        default_sink = get_default_sink()
        if default_sink:
            print(f"   🔊 输出设备: {default_sink}")
        else:
            print("   ❌ 未设置默认输出设备")

        default_source = get_default_source()
        if default_source:
            print(f"   🎤 输入设备: {default_source}")
        else:
            print("   ❌ 未设置默认输入设备")

        # 3. 检查项目配置
        print("\n3️⃣ 项目配置检查:")
        check_project_config()

    except Exception as e:
        print(f"❌ 检查过程中发生错误: {e}")

トラブルシューティングガイド

よくある質問

1. 音声を再生しても音が出ない

症状：TTS や音楽を再生しても音が出力されない 対処手順：

# 1. 检查音频设备状态
python utils/audio_config/check_audio_devices.py

# 2. 测试系统音频
pactl info

# 3. 检查音量设置
pactl list sinks

# 4. 测试音频播放器
python test/audio_tests/test_audio_player_new.py

2. ASR による認識が機能しない

症状：音声認識が反応しない、または誤認識率が高い 対処手順：

# 1. 检查麦克风设备
arecord -l

# 2. 测试录音功能
python test/audio_tests/test_recording_simple.py

# 3. 检查VAD模型
ls -la Backend/model/vad/

# 4. 查看ASR日志
tail -f backend.log | grep ASR

3. デバイスの競合

症状：複数の音声コンポーネントが同時に動作しない 解決方法：

# 在代码中使用AudioDeviceManager协调访问
access_granted = audio_device_manager.request_device_access_sync(
    AudioDeviceType.MUSIC_PLAYER,
    device_info={"music_name": music_name},
    force_stop_others=True  # 强制停止其他播放器
)

パフォーマンス最適化の推奨事項

1. バッファサイズのチューニング

# config.yaml中的音频配置
audio_player:
  blocksize: 1024      # 减小延迟，增加CPU使用
  buffer_size: 50      # 增加缓冲，提高稳定性
  min_prebuffer: 2     # 预缓冲大小

2. デバイス選択の最適化

# 优先选择低延迟设备
preferred_devices = [
    "hw:0,0",           # 硬件设备，延迟最低
    "plughw:0,0",       # 插件设备，兼容性好
    None                # 系统默认，稳定性好
]

3. メモリ使用量の最適化

# 音频数据块大小控制
OPTIMAL_CHUNK_SIZE = 1024 * 4  # 4KB
MAX_BUFFER_SIZE = 1024 * 1024  # 1MB

def optimize_audio_buffer(audio_data):
    """优化音频缓冲区使用"""
    if len(audio_data) > MAX_BUFFER_SIZE:
        # 分块处理大音频数据
        return split_audio_chunks(audio_data, OPTIMAL_CHUNK_SIZE)
    return audio_data

オーディオシステムは ASR-LLM-TTS の中核となる基盤であり、綿密に設計されたアーキテクチャと最適化戦略により、音声インタラクションの滑らかさと安定性を確保しています。