基于Solara：如何用ipywebrtc向Azure Speech流式传输音频？

问题背景 #

我正在用Solara构建Python Web应用，通过ipywebrtc捕获客户端浏览器音频，目前能录制到临时文件再传给Azure Speech，但需要实现流式传输。尝试直接将ipywebrtc的音频对象传给Azure的AudioInputStream时遇到类型转换错误，修改为recorder.codecs后又出现SPXERR_INVALID_ARG错误。

错误原因分析 #

• Azure Speech SDK的AudioInputStream仅接受符合其规范的流对象（如PushAudioInputStream或PullAudioInputStream），而ipywebrtc的recorder.audio是自定义媒体对象，无法直接转换为Azure所需的流类型。
• recorder.codecs是编码配置信息，并非实际音频数据，传入后必然触发无效参数错误。

解决方案 #

核心思路：在浏览器端捕获音频流，将原始PCM格式音频数据通过Solara前后端通信机制实时传到Python后端，再写入Azure的PushAudioInputStream实现流式转录。

步骤1：安装依赖 #

pip install solara ipywebrtc azure-cognitiveservices-speech numpy

步骤2：完整实现代码 #

import solara
from ipywebrtc import CameraStream, AudioRecorder
import azure.cognitiveservices.speech as speechsdk
from azure.cognitiveservices.speech.audio import PushAudioInputStream, AudioStreamFormat
import threading
import numpy as np

# Azure Speech配置
speech_config = speechsdk.SpeechConfig(subscription="你的订阅密钥", region="你的区域")
speech_config.speech_recognition_language = "pt-BR"  # 根据需求设置语言

# 匹配ipywebrtc的音频格式：16kHz、16位、单声道PCM
audio_format = AudioStreamFormat(samples_per_second=16000, bits_per_sample=16, channels=1)
push_stream = PushAudioInputStream(audio_format)
audio_config = speechsdk.audio.AudioConfig(stream=push_stream)

# 初始化会话转录器
transcriber = speechsdk.transcription.ConversationTranscriber(
    speech_config=speech_config, audio_config=audio_config
)

# 响应式变量存储转录结果
transcription_result = solara.reactive("")

def handle_transcription(evt):
    """处理Azure返回的转录结果"""
    if evt.result.reason == speechsdk.ResultReason.RecognizedSpeech:
        transcription_result.set(f"{transcription_result.value} {evt.result.text}")
    elif evt.result.reason == speechsdk.ResultReason.NoMatch:
        transcription_result.set(f"{transcription_result.value} [未识别音频]")

# 绑定转录器事件
transcriber.transcribed.connect(handle_transcription)
transcriber.session_started.connect(lambda evt: transcription_result.set("转录开始..."))
transcriber.session_stopped.connect(lambda evt: transcription_result.set(f"{transcription_result.value} \n转录结束"))

def audio_data_callback(data):
    """接收浏览器端音频数据，转换后写入Azure流"""
    if data is not None:
        # 将ipywebrtc的float32格式转换为16位PCM字节流
        pcm_data = (data * 32767).astype(np.int16).tobytes()
        push_stream.write(pcm_data)

def start_transcription():
    """启动实时转录"""
    transcriber.start_continuous_recognition()

def stop_transcription():
    """停止转录并关闭流"""
    transcriber.stop_continuous_recognition()
    push_stream.close()

@solara.component
def Page():
    # 初始化音频捕获组件
    camera = solara.use_memo(lambda: CameraStream(constraints={'audio': True, 'video': False}))
    recorder = solara.use_memo(lambda: AudioRecorder(
        stream=camera, 
        filename=None, 
        codecs='audio/wav', 
        audio_bitrate=16000
    ))
    
    # 监听音频数据回调
    solara.use_effect(lambda: recorder.audio.on_data(audio_data_callback), [recorder])
    
    # UI布局
    solara.Markdown("# 实时音频转录到Azure Speech")
    with solara.Row():
        solara.Button("开始转录", on_click=start_transcription)
        solara.Button("停止转录", on_click=stop_transcription)
    solara.Markdown(f"### 转录结果:\n{transcription_result.value}")

关键要点说明 #

• 格式匹配：必须保证ipywebrtc捕获的音频格式（16kHz、16位、单声道）与Azure Speech的AudioStreamFormat完全一致，否则会出现解码失败。
• 数据转换：ipywebrtc输出的是float32类型的音频样本，需转换为16位整数的PCM字节流才能被Azure Speech处理。
• 流式传输：通过recorder.audio.on_data回调实时获取浏览器端的音频数据，写入Azure的PushAudioInputStream实现无文件中转的流式传输。
• 线程安全：Azure Speech的转录操作在后台线程运行，Solara的响应式变量确保UI更新线程安全。

注意事项 #

• 确保Azure Speech的订阅密钥和区域配置正确，账户具备语音转录权限。
• 浏览器需授予麦克风访问权限。
• 生产环境建议添加错误处理（如网络中断、格式不匹配等异常场景）。

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