Gradio全解11——Streaming：流式传输的视频应用（2）——Twilio：网络服务提供商

本章目录如下：

1. 《Gradio全解11——Streaming：流式传输的视频应用（1）——FastRTC：Python实时通信库》
2. 《Gradio全解11——Streaming：流式传输的视频应用（2）——Twilio：网络服务提供商》
3. 《Gradio全解11——Streaming：流式传输的视频应用（3）——YOLO系列模型技术架构与实战》
4. 《Gradio全解11——Streaming：流式传输的视频应用（4）——基于Gradio WebRTC的实时目标检测》
5. 《Gradio全解11——Streaming：流式传输的视频应用（5）——RT-DETR：实时端到端检测模型》
6. 《Gradio全解10——Streaming：流式传输的视频应用（6）——基于RT-DETR模型构建视频流目标检测系统》
7. 《Gradio全解11——Streaming：流式传输的视频应用（7）——多模态Gemini模型及其思考模式》
8. 《Gradio全解11——Streaming：流式传输的视频应用（8）——Gemini Live API：实时音视频连接》
9. 《Gradio全解11——Streaming：流式传输的视频应用（9）——使用FastRTC+Gemini创建沉浸式音频+视频的艺术评论家》

11.2 Twilio：网络服务提供商

本节先讲解Twillo与TURN服务器概念，然后进行Twilio网络穿透服务实战，包括通过Twilio建立连接和使用Twilio拨打外呼电话。

11.2.1 Twillo穿透服务与TURN服务器

如果我们想在任意云提供商部署Gradio应用程序，则需要使用Twilio的API来获取他们的TURN服务器。那什么是TURN服务器，Twilio如何计费呢？

1. 什么是STUN、TURN和ICE？

STUN、TURN和ICE是IETF（The Internet Engineering Task Force：国际互联网工程任务组）制定的标准协议组，用于在建立点对点通信会话时穿透NAT（Network Address Translation，网络地址转换）。STUN、TURN和ICE的功能如下所述：

• 当主机位于NAT防火墙后方时，可通过NAT会话穿透工具STUN（ Session Traversal Utilities for NAT）发现主机的公网IP地址；若该主机需接收对端连接，会将主机公网IP地址作为可连接地址提供。若NAT防火墙仍阻止主机直连，双方则连接至NAT中继穿透服务器TURN（Traversal Using Relay around NAT），通过该服务器中转媒体流。
• WebRTC及其他VoIP技术栈通过支持交互式连接建立协议ICE（Interactive Connectivity Establishment ）来提升IP通信的可靠性，ICE是协调STUN与TURN以实现主机间连接的综合性标准。

关于STUN、TURN和ICE具体如何工作，请参考：How do STUN, TURN and ICE work?🖇️链接11-16。

2. Twilio介绍及网络穿透服务

Twilio是一家提供云通信平台即服务（CPaaS）的美国公司，其产品允许开发者通过API集成短信、语音通话、视频、电子邮件、发送WhatsApp消息和身份验证等功能到应用程序中。Twilio提供的服务包括以下几类：

• Twilio基础组件： Programmable SMS、Programmable Voice、2FA with Verify、Twilio API、Webhooks等。
• 开发者工具：Code Exchange、OpenAPI、Functions and Assets、Marketplace、Studio、TwiML Bins、Twilio CLI等。
• 开发资源：API status、Changelog、Error and Warning Dictionary、Glossary等。
• Twilio产品类：Twilio Content Template Builder、Conversational Intelligence、Elastic SIP Trunking、Flex、Event Streams、Network Traversal Service等。

除了网络穿透服务，其它内容已超出本书范围，感兴趣读者请参阅官网文档：🖇️链接11-17。Twilio的网络穿透服务（Network Traversal Service）是一项全球分布式STUN/TURN服务，它兼容ICE的客户端（如支持WebRTC标准的浏览器），可帮助厂商部署更可靠的P2P通信应用。在WebRTC和VoIP应用中可使用该服务实现NAT/防火墙穿透和中继，确保用户每次都能成功建立连接。更多信息请参考：Network Traversal Service🖇️链接11-18。

3. Twilio计费原理

Twilio的计费基于TURN服务器中转的数据量，TURN Client负责在TURN服务器上分配中继地址（也称为TURN会话）。中转数据总量按TURN Client发送和接收的字节数总和计算，系统随后会向创建该会话的TURN Client关联的Twilio Account SID收取费用。计费以兆字节为单位中转数据总量为依据，请注意不同Twilio区域的资费标准有所不同，登录Twilio Console（🖇️链接11-19）可查看发送记录、账单等。

11.2.2 通过Twilio建立连接

Twilio网络穿透服务实战包括两部分：使用Twilio建立连接并发送消息，以及使用Twilio拨打外呼电话，先讲述第一部分。本节内容包括安装Twilio并测试，然后使用Stream建立Twilio连接。

1. 安装Twilio并测试两类Client

从Python包管理工具PiPy安装Twilio，命令如下：

pip3 install twilio

测试安装，尝试给自己发送一条短信。更新以下代码示例中的account_sid、auth_token和from_，其中from_使用用户Twilio账户中的电话号码，to应为个人手机号码。代码如下所示：

from twilio.rest import Client
# Your Account SID and Auth Token from console.twilio.com
account_sid = "ACXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX"
auth_token  = "your_auth_token"
client = Client(account_sid, auth_token)
message = client.messages.create(to="[phone]",from_="[phone]", body="Hello from Python!")
print(message.sid)

运行代码并稍等片刻后，将在手机上收到短信。

也可以使用api_key和api_secret代替auth_token进行认证，并将发送短信改为拨打电话，代码如下所示：

from twilio.rest import Client
api_key = "XXXXXXXXXXXXXXXXX"
api_secret = "YYYYYYYYYYYYYYYYYY"
account_sid = "ACXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX"
client = Client(api_key, api_secret, account_sid)
call = client.calls.create(to="9991231234", from_="9991231234",url="http://twimlets.com/holdmusic?Bucket=com.twilio.music.ambient")
print(call.sid)

关于url的设置请参考11.3.3节。关于twilio-python库更多信息请参考：🖇️链接11-20。

2. 使用Stream建立Twilio连接

如需在WebRTC应用中使用Twilio网络穿透服务，只需获取令牌并传入Stream的构造函数。请按以下步骤操作：

• 首先，从Web服务器发起请求以获取网络穿透服务令牌，然后将其传递至WebRTC应用。获取网络穿透服务令牌需使用Twilio账户的SID和认证token，为保障Twilio账户凭证安全，请务必通过服务器而非客户端浏览器发起该请求。

import os
from twilio.rest import Client
# Find your Account SID and Auth Token at twilio.com/console
# and set the environment variables. See http://twil.io/secure
account_sid = os.environ["TWILIO_ACCOUNT_SID"]
auth_token = os.environ["TWILIO_AUTH_TOKEN"]
client = Client(account_sid, auth_token)
token = client.tokens.create()
print(token)

打印结果与如下所示内容类似：

{"username": "dc2d2894d5a9023620c467b0e71cfa6a35457e6679785ed6ae9856fe5bdfa269","ice_servers": [{"urls": "stun:global.stun.twilio.com:3478"},{"username": "dc2d2894d5a9023620c467b0e71cfa6a35457e6679785ed6ae9856fe5bdfa269","credential": "tE2DajzSJwnsSbc123","urls": "turn:global.turn.twilio.com:3478?transport=udp"},{"username": "dc2d2894d5a9023620c467b0e71cfa6a35457e6679785ed6ae9856fe5bdfa269","credential": "tE2DajzSJwnsSbc123","urls": "turn:global.turn.twilio.com:3478?transport=tcp"},{"username": "dc2d2894d5a9023620c467b0e71cfa6a35457e6679785ed6ae9856fe5bdfa269","credential": "tE2DajzSJwnsSbc123","urls": "turn:global.turn.twilio.com:443?transport=tcp"}],"date_updated": "Fri, 01 May 2020 01:42:57 +0000","account_sid": "ACXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX","ttl": "86400","date_created": "Fri, 01 May 2020 01:42:57 +0000","password": "tE2DajzSJwnsSbc123"
}

• 接下来，构造RTC配置。在建立呼叫时，将包含iceServers属性的RTCConfiguration对象传入Stream构造函数的属性rtc_configuration中，如下所示：

rtc_configuration = {"iceServers": token.ice_servers,"iceTransportPolicy": "relay"}
stream = Stream(handler=detection, modality="video",mode="send-receive", ..., rtc_configuration=rtc_configuration)

从此节点开始，已使用网络穿透服务建立连接，用户可以像往常一样交换SDP（Session Description Protocol，会话描述协议）的offer/answer。

11.2.3 电话集成：使用Twilio拨打外呼电话

电话集成用例主要为了演示FastRTC的Stream如何结合Twilio，实现发起呼叫、接听语音助手和流式传输音频功能。内容包括电话集成注意事项，设置集成流程并配置Twilio电话号码，以及使用Twilio外呼电话代码。

1. handler注意事项

使用电话集成时，Stream的处理函数handler应注意事项包括：

• 不要求任何额外输入：要使Stream的handler能通过电话正常工作，必须确保handler除了音频数据外，不要求任何额外输入。若调用await self.wait_for_args()，Stream将永久等待额外输入数据。StreamHandler具有phone_mode属性，当运行在电话模式时该属性值为True，我们可利用此属性判断是否需要等待额外输入数据。
• ReplyOnPause与电话集成：传递给ReplyOnPause的生成器函数必须为除audio外的所有参数设置默认值。若生成AdditionalOutputs，这些输出将在下次调用生成器时，作为输入参数传递给生成器。请参阅11.2.1节提到的示例Talk To Claude，了解作为handler的ReplyOnPause如何兼容电话使用；同时注意观察每次调用时，如何将输入的聊天历史记录生成为AdditionalOutput。

2. 设置专属电话号码指向FastAPI URL

在电话集成示例中，我们可以将Twilio等SIP提供商与Stream集成，并为应用程序设置专属电话号码。设置流程如下：

1. 创建Twilio账户：注册Twilio账户并购买支持语音功能的电话号码。对于试用账户，仅允许注册时使用的电话号码连接至Stream。
2. 挂载Stream：通过方法stream.mount(app)将FastAPI应用挂载至Stream，并启动服务器，得到应用URL。示例代码如下：

from fastrtc import Stream, ReplyOnPause
from fastapi import FastAPI
def echo(audio):yield audio
app = FastAPI()
stream = Stream(ReplyOnPause(echo), modality="audio", mode="send-receive")
stream.mount(app)
# run with `uvicorn main:app`

3. 配置Twilio Webhook：将您购买的Twilio电话号码指向第二步生成的网络钩子URL。

第三步配置Twilio电话号码详细步骤如下：

1. 进入Twilio账户控制台左侧导航栏，然后依次单击Develop→Manage→TwiML apps→Create new TwiML App。如图11-1所示：
   

   图11-1

2. 为TwiML应用命名后（如FastRTCPhone），将Voice Configuration中实现语音功能的Request URL设置为FastAPI应用地址，并在末尾添加/telephone/incoming，例如：https://your-app-url.com/telephone/incoming。最后保存即可，界面如图11-2所示：
   

   图11-2

技术拓展：在本地开发时，可通过Ngrok（🖇️链接11-21）将本地服务器暴露至公网。操作步骤：先启动ngrok服务：

ngrok http <port>

然后将Twilio Voice URL设置为：https://your-ngrok-subdomain.ngrok.io/telephone/incoming-call。

3. 示例：使用Twilio外呼电话代码

本节演示如何通过twilio-python模块实现电话呼叫，包括发起呼叫、接听语音助手和流式传输音频功能，代码如下所示：

from twilio.rest import Client
import gradio as gr
from fastrtc import Stream, ReplyOnPause
from fastapi import FastAPI
def echo(audio):yield audio
app = FastAPI()
stream = Stream(ReplyOnPause(echo), modality="audio", mode="send-receive")@app.post("/call")
async def start_call(req: Request):body = await req.json()from_no = body.get("from")to_no = body.get("to")account_sid = os.getenv("TWILIO_ACCOUNT_SID")auth_token = os.getenv("TWILIO_AUTH_TOKEN")client = Client(account_sid, auth_token)# Use the public URL of your application# here we're using ngrok to expose an app running locallycall = client.calls.create(to=to_no, from_=from_no,url="https://[your_ngrok_subdomain].ngrok.app/incoming-call")return {"sid": f"{call.sid}"}@app.api_route("/incoming-call", methods=["GET", "POST"])
async def handle_incoming_call(request: Request):from twilio.twiml.voice_response import VoiceResponse, Connectresponse = VoiceResponse()response.say("Connecting to AI assistant"connect = Connect()path = request.url.path.removesuffix("/telephone/incoming")connect.stream(url=f"wss://{request.url.hostname}{path}/telephone/handler")response.append(connect)response.say("The call has been disconnected.")return HTMLResponse(content=str(response), media_type="application/xml")@app.websocket("/media-stream")
async def handle_media_stream(websocket: WebSocket):# stream is a FastRTC stream defined elsewhereawait stream.telephone_handler(websocket)
app = gr.mount_gradio_app(app, stream.ui, path="/")
stream.ui.launch()

代码中函数结合Twilio，依次实现呼叫、接听语音助手和流式传输音频功能，最后将FastAPI挂载到Gradio界面并启动。其中流式传输音频是基于WebSocket实现的，而不是WebRTC。各个函数解析如下：

• start_call：启动呼叫。从请求Request中获取呼叫和被呼叫号码，然后创建客户端并开始呼叫。
• handle_incoming_call：处理来电，例如通过Twilio（取决于电话设置）。它生成TwiML指令，将来电连接至WebSocket处理程序（如/telephone/handler）以进行音频流传输。参数request是用于处理来电webhook的FastAPI Request对象，返回包含TwiML指令（XML格式）的HTMLResponse。
• stream.telephone_handler：电话处理程序，它接受传入的WebSocket连接对象，是通过Twilio电话进行流式传输音频的WebSocket连接处理端点。

更多信息请参考：Telephone Integration🖇️链接11-22。