如何用PyAV与aiortc实时合并多WebRTC音频流并生成转录文本？

你当前的代码只是循环逐个接收不同track的音频帧并依次编码写入，本质是交替播放各流的声音，并非真正的音频混音合并。要实现多流合并为单音频输出，同时支持转录，可按以下思路优化：

一、核心实现逻辑 #

要完成多音频流合并，关键要做这几件事：

• 统一所有音频流的参数（采样率、声道数、采样格式）
• 异步并行接收各track的帧，按时间戳对齐缓存
• 将同一时间窗口的音频帧采样数据叠加混音（需做音量衰减避免失真）
• 编码混音后的帧输出，同时将原始PCM数据传给转录服务

二、具体优化步骤与代码示例 #

1. 统一音频参数 #

WebRTC默认音频参数通常是48kHz单声道16位PCM，所有track必须统一此格式，不一致的要先转码。

2. 并行接收与帧对齐 #

用异步任务单独接收每个track的帧，存入缓存后按时间戳对齐，保证混音的是同一时间段的音频。

3. 混音处理 #

将对齐后的帧采样数据叠加，每个帧的采样值先除以track数量做衰减，防止音量过载削波。

4. 编码输出与转录集成 #

优化后的完整代码示例：

import asyncio
import av
import numpy as np
from aiortc import rtcrtpreceiver

class AudioMixerRecorder:
    _task = None
    _container = None
    _audio_stream = None
    _tracks: list[rtcrtpreceiver.RemoteStreamTrack] = []
    _frame_buffers = {}  # 存储每个track的帧缓存

    def __init__(self, file_path: str, sample_rate=48000, channels=1, sample_fmt='s16'):
        self._container = av.open(file=file_path, format=None, mode="w")
        self._audio_stream = self._container.add_stream("aac", rate=sample_rate)
        self._audio_stream.channels = channels
        self._sample_rate = sample_rate
        self._channels = channels
        self._sample_fmt = sample_fmt

    def add_track(self, track: rtcrtpreceiver.RemoteStreamTrack):
        self._tracks.append(track)
        self._frame_buffers[track] = []

    async def _receive_frames(self, track):
        """单独接收每个track的帧，统一格式后存入缓存"""
        while True:
            frame = await track.recv()
            # 统一音频格式，不一致则转码
            if (frame.format.name != self._sample_fmt 
                or frame.sample_rate != self._sample_rate 
                or frame.channels != self._channels):
                frame = frame.reformat(
                    sample_rate=self._sample_rate,
                    format=self._sample_fmt,
                    channels=self._channels
                )
            self._frame_buffers[track].append(frame)

    def _mix_frames(self):
        """从缓存中取出可对齐的帧进行混音"""
        # 找到所有缓存中最早的帧时间戳
        min_pts = None
        for buf in self._frame_buffers.values():
            if buf and (min_pts is None or buf[0].pts < min_pts):
                min_pts = buf[0].pts

        if min_pts is None:
            return None

        # 收集所有track中待混音的帧
        frames_to_mix = []
        for buf in self._frame_buffers.values():
            if buf:
                frames_to_mix.append(buf.pop(0))

        if not frames_to_mix:
            return None

        # 初始化混音帧，复用第一个帧的参数
        mixed_frame = av.AudioFrame(
            format=self._sample_fmt,
            sample_rate=self._sample_rate,
            channels=self._channels
        )
        mixed_frame.samples = frames_to_mix[0].samples
        mixed_frame.pts = min_pts

        # 提取各帧的采样数据并混音
        samples_list = [frame.to_ndarray() for frame in frames_to_mix]
        # 衰减后叠加，避免削波
        mixed_samples = sum(samples for samples in samples_list) / len(samples_list)
        # 转换为对应格式的整数
        mixed_samples = mixed_samples.astype(np.int16)

        mixed_frame.from_ndarray(mixed_samples)
        return mixed_frame

    async def _start(self):
        # 启动所有track的异步接收任务
        receive_tasks = [asyncio.create_task(self._receive_frames(track)) for track in self._tracks]

        try:
            while True:
                mixed_frame = self._mix_frames()
                if mixed_frame:
                    # 编码并写入容器
                    for packet in self._audio_stream.encode(mixed_frame):
                        self._container.mux(packet)
                    # 这里可以调用转录逻辑，比如把帧转成PCM传给Whisper
                    # await self._transcribe_audio(mixed_frame)
                await asyncio.sleep(0.01)  # 降低CPU占用
        finally:
            # 清理所有接收任务
            for task in receive_tasks:
                task.cancel()
            # 编码剩余帧并关闭容器
            for packet in self._audio_stream.encode(None):
                self._container.mux(packet)
            self._container.close()

    async def start(self):
        self._task = asyncio.create_task(self._start())

    async def stop(self):
        if self._task:
            self._task.cancel()
            await self._task

    async def _transcribe_audio(self, frame):
        """示例转录逻辑，可替换为实际转录服务"""
        # 将帧转为PCM字节流
        pcm_data = frame.to_ndarray().tobytes()
        # 这里可以调用Whisper等模型进行转录
        # 比如用openai-whisper的本地模型：
        # result = whisper_model.transcribe(pcm_data, sample_rate=self._sample_rate)
        # print(result["text"])

三、音频转录集成建议 #

• 选择转录工具：可以用OpenAI Whisper（支持本地部署或API），或者轻量级模型如Vosk，适合实时转录场景
• 异步处理：把混音后的PCM数据放入异步队列，单独开一个任务处理转录，避免阻塞混音流程
• 格式适配：转录模型通常需要原始PCM数据，直接用frame.to_ndarray().tobytes()即可获取符合要求的字节流

四、关键注意事项 #

• 音量控制：混音时必须做衰减，否则多轨叠加会导致采样值超出范围，出现刺耳的削波失真
• 时间对齐：实际场景中WebRTC帧的时间戳可能有偏差，可进一步优化缓存逻辑，按帧的持续时间计算对齐窗口
• 资源优化：多track接收和混音会占用CPU，可限制缓存的最大帧数，避免内存泄漏
• 异常处理：要处理track断开的情况，及时从缓存中移除失效的track

内容的提问来源于stack exchange，提问作者Paul Salmon