如何在Twilio通话中同时实现双向流与多轮Gather转写？

核心思路 #

Twilio的<Stream>和<Gather>无法并行运行，因此放弃依赖<Gather>的内置转写功能，改用**全程双向媒体流（Media Stream）**同时处理音频收发和实时转写，完全覆盖原需求的两个功能模块。

具体实现步骤 #

1. 通话初始化直接启动双向流

   • 当Twilio触发来电Webhook时，直接返回包含<Stream>标签的TwiML，全程维持WebSocket连接，从通话开始就捕获音频，避免丢失：

     <Response>
       <Stream url="wss://your-public-server/ws" />
     </Response>
     

   • 纠正一个误解：Twilio的Media Stream本身就是双向的，Python可通过WebSocket服务实现双向音频传输，并非只能单向。

2. 基于WebSocket实现实时转写（替代多轮）

   • 在WebSocket服务中接收Twilio推送的PCMU格式音频帧（8kHz单声道），将帧缓存为连续音频片段。
   • 调用第三方ASR服务（如OpenAI Whisper、Google Speech-to-Text）对片段转写，通过语音停顿检测分割句子，实现逐句转写的效果，完全替代<Gather>的多轮转写需求。

3. 双向音频回传处理

   • 当需要向通话播放音频时，将音频文件转换为PCMU格式的帧，通过WebSocket直接发送给Twilio即可实现实时回传。

Python示例代码框架 #

import asyncio
import websockets
from twilio.twiml.voice_response import VoiceResponse, Stream
import speech_recognition as sr

# 生成通话初始化的TwiML
def get_incoming_call_twiml():
    resp = VoiceResponse()
    # 替换为你的WebSocket服务公网地址
    resp.stream(url="wss://your-domain.com/twilio-stream")
    return str(resp)

# WebSocket服务处理逻辑
async def twilio_stream_handler(websocket):
    audio_buffer = bytearray()
    recognizer = sr.Recognizer()

    async for msg in websocket:
        if isinstance(msg, bytes):
            # 接收Twilio发来的PCMU音频帧，存入缓冲区
            audio_buffer.extend(msg)
            # 示例：当缓冲区达到5秒音频量时触发转写
            if len(audio_buffer) > 8000 * 5:
                # 转换PCMU为WAV格式供ASR服务使用
                wav_data = convert_pcmu_to_wav(audio_buffer)
                # 调用Whisper进行转写
                with sr.AudioFile(wav_data) as source:
                    audio = recognizer.record(source)
                    transcript = recognizer.recognize_whisper(audio, model="base")
                    print(f"转写结果：{transcript}")
                # 清空缓冲区，准备下一轮转写
                audio_buffer.clear()
        else:
            # 处理Twilio的控制消息（如stream start/stop）
            pass

    # 示例：回传音频（需先将目标音频转为PCMU帧）
    # await websocket.send(pcmu_audio_frame)

# PCMU转WAV的辅助函数
def convert_pcmu_to_wav(pcmu_data):
    import wave
    from io import BytesIO
    wav_buffer = BytesIO()
    with wave.open(wav_buffer, 'wb') as wav_file:
        wav_file.setnchannels(1)
        wav_file.setsampwidth(2)
        wav_file.setframerate(8000)
        # 将PCMU转换为16位PCM
        pcm_data = bytes([(sample ^ 0xFF) << 1 for sample in pcmu_data])
        wav_file.writeframes(pcm_data)
    wav_buffer.seek(0)
    return wav_buffer

# 启动WebSocket服务
async def main():
    async with websockets.serve(twilio_stream_handler, "0.0.0.0", 8765):
        await asyncio.Future()  # 保持服务持续运行

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

关键注意事项 #

• 音频格式必须严格匹配Twilio要求：PCMU（μ-law）、8kHz采样率、单声道，否则无法正常收发。
• WebSocket服务必须具备公网可访问的HTTPS/WSS地址，Twilio不允许明文WS连接。
• ASR服务的实时处理逻辑可优化为语音端点检测（VAD），精准分割句子，提升转写体验。

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