基于Google Speech实现双人实时对话转写的最优方案咨询

是否需要分离麦克风和扬声器音频流？ #

是的，必须分离两路音频流。核心原因有两点：

• Google Speech API的**说话人分离（Speaker Diarization）**功能依赖清晰的音频来源区分，混合后的音频会大幅降低识别准确率，甚至无法正确标记说话人身份。
• 麦克风对应本地用户声音，扬声器对应远端用户声音，分离后既能独立处理转写，也能方便后续给结果打上明确的身份标签（如“本地用户”/“远端用户”）。

Google Meeting类工具的双人对话实时转写实现逻辑 #

这类工具的核心是分源捕获+说话人识别联动：

1. 音频捕获层：
   • 麦克风音频：直接通过系统默认输入设备捕获（即你当前代码的逻辑）。
   • 扬声器音频：需要捕获系统的“立体声混音”（Windows）、“Loopback”（macOS/Linux）设备，这类设备专门用于录制系统输出的声音。
2. 转写处理层：
   • 可以将两路音频分别发送至Google Speech API，或是合并为双声道音频（左声道麦克风、右声道扬声器）后发送，API会自动识别声道区分说话人。
   • 启用Google Speech的Speaker Diarization功能，配置enable_speaker_diarization=True并设置diarization_speaker_count=2（适配双人对话），API会返回携带说话人标签的转写结果。
3. 结果展示层：根据API返回的说话人标签，给转写文本标注对应的身份即可。

代码调整建议（基于你的PyAudio代码） #

1. 定位扬声器捕获设备 #

先遍历系统音频设备，找到对应Loopback设备的ID（以Windows为例，macOS/Linux需调整名称匹配逻辑）：

import pyaudio

audio = pyaudio.PyAudio()
loopback_device_id = None
# 遍历设备列表，查找立体声混音/loopback设备
for i in range(audio.get_device_count()):
    info = audio.get_device_info_by_index(i)
    if "立体声混音" in info["name"] or "Stereo Mix" in info["name"]:
        loopback_device_id = i
        print(f"找到Loopback设备: {info['name']}, ID: {loopback_device_id}")
        break

2. 同时开启两路音频捕获流 #

用异步任务分别处理麦克风和扬声器的音频捕获与发送：

import pyaudio
import asyncio
import socketio

FORMAT = pyaudio.paInt16
CHANNELS = 1
RATE = 16000  # Google Speech推荐采样率
CHUNK = 1024

sio = socketio.AsyncClient()

async def capture_mic():
    audio = pyaudio.PyAudio()
    stream = audio.open(
        format=FORMAT,
        channels=CHANNELS,
        rate=RATE,
        input=True,
        frames_per_buffer=CHUNK,
    )
    while True:
        data = stream.read(CHUNK)
        # 发送时标记音频来源
        await sio.emit("binaryAudioData", {"source": "mic", "data": data})
        await asyncio.sleep(0)

async def capture_speaker():
    audio = pyaudio.PyAudio()
    # 替换为你找到的Loopback设备ID
    loopback_device_id = 2
    stream = audio.open(
        format=FORMAT,
        channels=CHANNELS,
        rate=RATE,
        input=True,
        input_device_index=loopback_device_id,
        frames_per_buffer=CHUNK,
    )
    while True:
        data = stream.read(CHUNK)
        # 发送时标记音频来源
        await sio.emit("binaryAudioData", {"source": "speaker", "data": data})
        await asyncio.sleep(0)

async def main():
    await sio.connect("你的服务器地址")
    await asyncio.gather(capture_mic(), capture_speaker())
    await sio.wait()

if __name__ == "__main__":
    asyncio.run(main())

3. 后端启用说话人分离转写 #

在接收音频的后端，调用Google Speech API并启用说话人分离：

from google.cloud import speech_v1p1beta1 as speech

client = speech.SpeechClient()

def transcribe_with_diarization(audio_content):
    config = speech.RecognitionConfig(
        encoding=speech.RecognitionConfig.AudioEncoding.LINEAR16,
        sample_rate_hertz=16000,
        language_code="zh-CN",
        enable_speaker_diarization=True,
        diarization_speaker_count=2,
    )
    audio = speech.RecognitionAudio(content=audio_content)
    response = client.recognize(config=config, audio=audio)
    
    # 解析带说话人标签的结果
    result = response.results[-1]
    words_info = result.alternatives[0].words
    for word_info in words_info:
        print(f"说话人{word_info.speaker_tag}: {word_info.word}")

注意事项 #

• 不同系统的Loopback设备名称差异：macOS可使用Soundflower或系统自带Loopback设备，Linux可借助pulse的loopback模块配置。
• Google Speech的说话人分离功能需使用v1p1beta1版本的客户端库，且音频采样率、编码格式必须符合API要求。
• 若选择合并双声道音频发送，需在RecognitionConfig中设置audio_channel_count=2，API会自动按声道区分说话人。

内容的提问来源于stack exchange，提问作者p.magalhaes