如何在Python应用中实现Twilio通话实时流至Google Voice API转写响应

嘿，我之前刚好折腾过类似的实时通话转写功能，结合Twilio和Google Speech-to-Text完全能实现你的需求！别纠结Record功能了，Twilio的Media Streams才是实时流处理的正确打开方式，给你一步步拆解方案和代码示例：

核心思路 #

Twilio的Media Streams可以把通话中的音频数据通过WebSocket实时推送到你的Python服务，你只需要：

1. 配置Twilio通话，让它把音频流发送到你的WebSocket端点
2. 在Python服务中接收音频帧，转换为Google Speech API支持的格式
3. 调用Google Speech的实时转写接口，拿到转写文本
4. 根据转写内容生成响应，再通过Twilio推送给通话者

步骤1：让Twilio开启音频流 #

当Twilio发起通话时，你需要返回一段TwiML指令，告诉Twilio要开启实时流，并指定你的WebSocket服务地址。比如：

<Response>
  <Start>
    <!-- 替换成你的公网WebSocket地址，本地测试可以用ngrok转发 -->
    <Stream url="wss://your-public-domain.com/twilio-stream" />
  </Start>
  <Say>请说话，我会实时转写并回应你</Say>
  <Pause length="60" />
</Response>

这段TwiML可以通过你的Python后端接口返回给Twilio，比如用Flask写一个简单的路由：

from flask import Flask, Response

app = Flask(__name__)

@app.route("/twiml", methods=["POST"])
def twiml_response():
    twiml = """
    <Response>
      <Start>
        <Stream url="wss://your-public-domain.com/twilio-stream" />
      </Start>
      <Say>请说话，我会实时转写并回应你</Say>
      <Pause length="60" />
    </Response>
    """
    return Response(twiml, mimetype="application/xml")

if __name__ == "__main__":
    app.run(port=5000)

步骤2：搭建Python WebSocket服务接收音频流 #

用websockets库创建WebSocket服务，接收Twilio发来的音频数据。Twilio发送的是PCMU编码（8kHz单声道），得转换成Google Speech支持的LINEAR16格式。这里直接上代码：

import asyncio
import websockets
import json
import base64
from google.cloud import speech_v1p1beta1 as speech
from google.cloud.speech_v1p1beta1 import types
from pydub import AudioSegment
from twilio.rest import Client

# 初始化Google Speech客户端（记得配置GOOGLE_APPLICATION_CREDENTIALS环境变量）
speech_client = speech.SpeechClient()

# 初始化Twilio客户端
TWILIO_ACCOUNT_SID = "你的Twilio账户SID"
TWILIO_AUTH_TOKEN = "你的TwilioAuthToken"
twilio_client = Client(TWILIO_ACCOUNT_SID, TWILIO_AUTH_TOKEN)

# Google Speech实时识别配置
recognition_config = types.RecognitionConfig(
    encoding=speech.RecognitionConfig.AudioEncoding.LINEAR16,
    sample_rate_hertz=8000,
    language_code="zh-CN",  # 根据你的需求切换语言
    enable_automatic_punctuation=True
)
streaming_config = types.StreamingRecognitionConfig(
    config=recognition_config,
    interim_results=True  # 返回中间结果，提升实时感
)

# PCMU转LINEAR16的工具函数
def pcmu_to_linear16(pcmu_data):
    audio = AudioSegment(
        data=pcmu_data,
        sample_width=1,
        frame_rate=8000,
        channels=1
    )
    return audio.set_sample_width(2).raw_data

async def handle_twilio_stream(websocket, path):
    call_sid = None
    # 生成Google Speech的流式请求
    async def audio_request_generator():
        nonlocal call_sid
        while True:
            message = await websocket.recv()
            data = json.loads(message)
            
            # 记录通话SID，后续用来发送响应
            if data["event"] == "start":
                call_sid = data["start"]["callSid"]
                print(f"通话开始：{call_sid}")
            
            # 处理音频数据
            elif data["event"] == "media":
                pcmu_payload = base64.b64decode(data["media"]["payload"])
                linear16_payload = pcmu_to_linear16(pcmu_payload)
                yield types.StreamingRecognizeRequest(audio_content=linear16_payload)
            
            # 通话结束
            elif data["event"] == "stop":
                print(f"通话结束：{call_sid}")
                break

    # 处理Google Speech的识别结果
    responses = speech_client.streaming_recognize(streaming_config, audio_request_generator())
    async for response in responses:
        for result in response.results:
            transcript = result.alternatives[0].transcript
            if result.is_final:
                print(f"最终转写内容：{transcript}")
                # 这里写你的业务逻辑，生成响应文本
                response_text = f"我听到你说：{transcript}"
                # 通过Twilio API发送响应
                send_twilio_response(call_sid, response_text)
            else:
                print(f"实时中间转写：{transcript}")

def send_twilio_response(call_sid, response_text):
    """通过Twilio API更新通话TwiML，播放响应"""
    twiml = f'<Response><Say>{response_text}</Say></Response>'
    twilio_client.calls(call_sid).update(twiml=twiml)

# 启动WebSocket服务
start_server = websockets.serve(handle_twilio_stream, "0.0.0.0", 8765)

asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()

步骤3：本地测试技巧 #

本地开发时，WebSocket服务需要公网可访问，推荐用ngrok做端口转发：

ngrok http 8765 --protocol=http

然后把生成的wss地址（比如wss://xxxx-xx-xx-xx-xx.ngrok.io）替换到TwiML的Stream URL里。

几个要注意的坑 #

• 音频格式转换：Twilio默认发PCMU，必须转LINEAR16，否则Google识别会失败，我用了pydub，记得提前装ffmpeg
• Google认证：要配置GOOGLE_APPLICATION_CREDENTIALS环境变量，指向你的服务密钥JSON文件
• 实时性平衡：interim_results=True会返回中间转写结果，虽然实时但可能不准，你可以根据业务需求决定是否只处理最终结果
• 错误处理：记得加WebSocket断开、API调用失败的异常捕获，避免服务崩溃

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