本章目錄如下：

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+YOLO的實時目標檢測》
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="+8615558675309",from_="+8615017250604", 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。