向Twilio WebSocket发送mulaw 8kHz音频无法播放问题求助

我使用gTTS实现文本转语音功能，试图通过WebSocket向Twilio回传媒体音频，但Twilio无法播放该音频。已将音频转为单声道8kHz格式，尝试过audio/x-mulaw编码后仍然无效。以下是相关代码：

文本转语音代码 #

def text_to_mp3(self, f_path :str , f_key: str, text: str) -> str:
        tmp_fn = os.path.join(f_path, f_key + ".mp3"))
        tts = gTTS(text, lang="en")
        tts.save(tmp_fn) #save mp3
        new_fn = os.path.join(f_path, f_key + ".wav")
        audio = am.from_mp3(tmp_fn) #convert mp3 to wav 
        audio = audio.set_channels(1)  # Mono audio
        audio = audio.set_frame_rate(8000)  # 8000Hz sample rate
        audio.export(new_fn, format='wav')
        return new_fn

初始音频Payload处理代码 #

def get_encoded_payload(self, filename):
    with wave.open(filename, 'rb') as wav_in:
        params = wav_in.getparams()
        header_size = params.nframes * params.sampwidth + 44 # 44 is the size of the WAV header
        wav_in.readframes(header_size)
        audio_data = wav_in.readframes(params.nframes)
    base64_encoded_audio = base64.b64encode(audio_data).decode()
    return base64_encoded_audio

向Twilio发送媒体的代码 #

def play(self, base_data : str):
    media_data = {
        "event": "media",
        "streamSid": self.streamId,
        "media": {
            "payload": base_data
            }
        }
    self.ws.send(media_data)

发送标记消息的代码 #

def play_mark(self, unique_name: str):
    media_data = {
        "event": "mark",
        "streamSid": self.streamId,
        "mark": {
            "name": unique_name
            }
        }
    self.ws.send(media_data)

尝试过的μ-law编码修改版代码 #

def get_encoded_payload(self, filename):
    with wave.open(filename, 'rb') as wav_in:
        params = wav_in.getparams()
        header_size = params.nframes * params.sampwidth + 44 # 44 is the size of the WAV header
        wav_in.readframes(header_size)
        audio_data = wav_in.readframes(params.nframes)
        ulaw_audio_data= audioop.lin2ulaw(audio_data, params.sampwidth)
    base64_encoded_audio = base64.b64encode(ulaw_audio_data).decode()
    return base64_encoded_audio

问题根源与修复方案 #

1. WAV头处理逻辑错误
   标准PCM WAV的头长度是固定44字节，你之前的header_size计算错误（额外加上了音频数据长度），导致直接跳过了整个文件内容，后续读取的audio_data为空。实际上wave.open会自动处理文件头，直接调用readframes就能获取原始音频帧数据，无需手动跳过头。

2. 未按Twilio要求分块发送
   Twilio Media Stream要求每个media事件的payload对应20ms的音频数据（即160帧8kHz单声道音频，转成μ-law后是160字节），不能一次性发送完整音频文件。

3. 音频格式细节不匹配
   需要确保输出的WAV是16位线性PCM，这样lin2ulaw转换才会正确；同时发送WebSocket消息时要将字典转为JSON字符串，否则Twilio无法解析。

修复后的完整代码示例 #

修正文本转语音代码（确保16位PCM输出）

def text_to_mp3(self, f_path :str , f_key: str, text: str) -> str:
    tmp_fn = os.path.join(f_path, f_key + ".mp3")  # 修复语法错误：多余的右括号
    tts = gTTS(text, lang="en")
    tts.save(tmp_fn)
    new_fn = os.path.join(f_path, f_key + ".wav")
    audio = am.from_mp3(tmp_fn)
    audio = audio.set_channels(1)
    audio = audio.set_frame_rate(8000)
    # 明确设置16位采样宽度，保证是16位PCM格式
    audio = audio.set_sample_width(2)
    audio.export(new_fn, format='wav', codec='pcm_s16le')
    return new_fn

修正Payload处理（分块生成μ-law编码）

import audioop
import wave
import base64

def get_encoded_payload_chunks(self, filename):
    audio_chunks = []
    with wave.open(filename, 'rb') as wav_in:
        # 验证音频格式是否符合Twilio要求
        assert wav_in.getnchannels() == 1, "音频必须为单声道"
        assert wav_in.getframerate() == 8000, "采样率必须为8000Hz"
        assert wav_in.getsampwidth() == 2, "音频必须为16位PCM"
        
        # 20ms音频对应的帧数：8000Hz * 0.02s = 160帧
        frames_per_chunk = 160
        while True:
            # 直接读取帧数据，wave模块已自动跳过文件头
            raw_frames = wav_in.readframes(frames_per_chunk)
            if not raw_frames:
                break
            # 将16位线性PCM转为μ-law编码
            ulaw_data = audioop.lin2ulaw(raw_frames, 2)
            # base64编码后加入列表
            audio_chunks.append(base64.b64encode(ulaw_data).decode())
    return audio_chunks

修正媒体发送逻辑（分块发送+JSON序列化）

import json
import time

def play(self, audio_chunks):
    for chunk in audio_chunks:
        media_data = {
            "event": "media",
            "streamSid": self.streamId,
            "media": {
                "payload": chunk
            }
        }
        # 将字典转为JSON字符串发送，这是之前缺失的关键步骤
        self.ws.send(json.dumps(media_data))
        # 控制发送间隔，模拟实时音频流
        time.sleep(0.02)
    # 音频发送完成后发送标记
    self.play_mark("audio_playback_complete")

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