前言

本系列文章主要介紹WEB界面工具Gradio。Gradio是Hugging Face發布的簡易WebUI開發框架，它基于FastAPI和svelte，可以使用機器學習模型、python函數或API開發多功能界面，并可部署人工智能模型，是當前熱門的非常易于展示機器學習大語言模型LLM及擴散模型DM的WebUI框架。
本系列文章分為三部分：Gradio介紹、Gradio基礎功能實戰和Gradio高級功能實戰。第一部分Gradio介紹，方便讀者對Gradio整體把握，包括三章內容：第一章先介紹Gradio的概念，包括詳細技術架構、歷史、應用場景、與其他框架Gradio/NiceGui/StreamLit/Dash/PyWebIO的區別，然后詳細講述Gradio的安裝與運行，安裝包括Linux/Win/Mac三類系統安裝，運行包括普通方式和熱重載方式；第二章介紹Gradio的4種部署方式，包括本地部署launch()、huggingface托管、FastAPI掛載和Gradio-Lite瀏覽器集成；第三章介紹Gradio的三種客戶端（Client），包括python客戶端、javascript客戶端和curl客戶端。第二部分實戰Gradio基礎功能，進入本系列文章的核心，包括四章內容：第四章講解Gradio庫的模塊架構和環境變量，第五章講解Gradio高級抽象界面類Interface，第六章講解Gradio底層區塊類Blocks，第七章講解補充特性Additional Features。第三部分講解并實戰Gradio的高級功能，包括五章內容：第八章講解融合大模型的多模態聊天機器人組件Chatbot，第九章講解Gradio Tools工具庫的使用及構建方法，第十章講述講述數據科學與繪圖Data Science And Plots；第十一章講述流式傳輸Streaming的多模態應用；第十二章講述由Gradio App創建Discord Bot/Slack Bot/Website Widget；第十三章講述使用Gradio構建MCP的客戶端與服務器。
本系列文章講解細致，涵蓋Gradio及相關框架的大部分組件和功能，代碼均可運行并附有大量運行截圖，方便讀者理解并應用到開發中，Gradio一定會成為每個技術人員實現各種奇思妙想的最稱手工具。

本系列文章目錄如下：

1. 《Gradio全解1——Gradio簡介：大模型WebUI框架（上）》
2. 《Gradio全解1——Gradio簡介：大模型WebUI框架（下）》
3. 《Gradio全解2——Gradio的四種部署方式（上）》
4. 《Gradio全解2——Gradio的四種部署方式（下）》
5. 《Gradio全解3——Gradio三種客戶端：Python、JavaScript與Curl（一）——python》
6. 《Gradio全解3——Gradio三種客戶端：Python、JavaScript與Curl（二）——javascript》
7. 《Gradio全解3——Gradio三種客戶端：Python、JavaScript與Curl（三）——curl》
8. 《Gradio全解4——Gradio庫的模塊架構和環境變量》
9. 《Gradio全解5——Interface：高級抽象界面類（上）》
10. 《Gradio全解5——Interface：高級抽象界面類（下）》
11. 《Gradio全解6——Blocks：底層區塊類（上）》
12. 《Gradio全解6——Blocks：底層區塊類（下）》
13. 《Gradio全解7——Additional Features：補充特性（上）》
14. 《Gradio全解7——Additional Features：補充特性（下）》
15. 《Gradio全解8——ChatInterface&Chatbot：聊天界面類與聊天機器人》
16. 《Gradio全解9——Data Science And Plots：數據科學與繪圖》
17. 《Gradio全解10——Streaming：流式傳輸的音頻應用》
18. 《Gradio全解11——Streaming：流式傳輸的視頻應用》
19. 《Gradio全解12——Agents and Tools：代理與工具庫》
20. 《Gradio全解13——由Gradio App創建Discord Bot/Slack Bot/Website Widget》
21. 《Gradio全解14——使用Gradio構建MCP的客戶端與服務器》

本章目錄如下：

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章 Streaming：流式傳輸的視頻應用

本章講述流式傳輸的視頻應用，應用部分包括三部分：基于Gradio.WebRTC+YOLO的實時目標檢測，使用RT-DETR模型構建視頻流目標檢測系統，以及使用FastRTC+Gemini創建實時沉浸式音頻+視頻的藝術評論家。此外，用到的技術部分分為獨立的六節講解：Python實時通信庫FastRTC，網絡服務提供商Twilio，YOLO系列模型技術架構與實戰，實時端到端檢測模型RT-DETR、Gemini模型簡介及入門實戰，以及為更復雜的結合音視頻的Gemini Live API實時連接。

11.1 FastRTC：Python實時通信庫

在講解基于WebRTC的攝像頭實時目標檢測之前，先學習兩個知識點：FastRTC實時通信庫與Twilio網絡服務提供商。本節先學習FastRTC，Twilio將在11.3節講述。

在11.5節的攝像頭目標檢測中，我們將使用OpenCV進行圖像處理，并通過Gradio的WebRTC自定義組件FastRTC（GitHub倉庫🖇?鏈接11-1）傳輸數據。FastRTC被封裝為gradio_webrtc.WebRTC，它在底層使用WebRTC協議實現近乎零延遲。本節先講解WebRTC協議的概念與用法，然后講解FastRTC的特性和核心功能，最后進行FastRTC實戰。FastRTC實戰部分包括FastRTC安裝、內置功能與自定義路由，FastRTC核心特性Stream及其構造參數，Stream三種運行方式和四種典型示例，FastRTC的三種部署方案。

11.1.1 WebRTC協議與FastRTC介紹

本節先介紹WebRTC協議的概念與用法，從而引出FastRTC特性。

1. WebRTC協議的概念與用法

WebRTC（Web Real-Time Communication，網頁實時通信）是一項使網頁應用及網站能夠采集并持續傳輸音視頻媒體，同時支持瀏覽器間無需中介直接交換任意數據的技術。這套標準集合使得數據共享和點對點視頻會議成為可能，且用戶無需安裝插件或任何第三方軟件。其核心內容包括WebRTC的基礎架構原理、數據通道與媒體連接的配置方法、高級功能實現等。

WebRTC由多個相互關聯的API和協議共同構成，這些組件協同工作以實現實時通信功能。兩個對等方之間的連接由RTCPeerConnection接口表示，一旦使用RTCPeerConnection建立并打開了連接，就可以向連接添加下面兩類信息：

• 媒體流（MediaStream）：媒體流可以由任意數量的媒體信息軌道組成。軌道由基于MediaStreamTrack接口的對象表示，可以包含多種類型的媒體數據，包括音頻、視頻和文本（例如字幕甚至章節名稱）。大多數媒體流至少包含一個音頻軌道，可能也包含一個視頻軌道，并且可以用于發送和接收實時媒體流或存儲媒體流信息（例如電影）。
• 數據通道（RTCDataChannel）：使用RTCDataChannel接口，用戶還可以通過兩個對等方之間的連接來交換任意二進制數據。這可以用于反向信道信息、元數據交換、游戲狀態數據包、文件傳輸，甚至作為數據傳輸的主要通道。

借助WebRTC，我們可以為Gradio或其它應用添加基于開放標準運行的實時通信功能。它支持在對等設備之間發送視頻、語音和通用數據，使開發者能夠構建強大的語音和視頻通信解決方案，并且適用于所有現代瀏覽器以及所有主要平臺的原生客戶端。WebRTC采用的技術是開放網絡標準，以常規JavaScript API的形式在所有主流瀏覽器中提供。對于部分兼容性問題，可使用adapter.js庫作為兼容層，有效隔離應用與平臺間差異。關于WebRTC更多信息請參考：WebRTC API🖇?鏈接11-2。

2. FastRTC特性和核心功能

FastRTC是Python實時通信庫，它將任意Python函數轉換為通過WebRTC或WebSocket傳輸的實時音視頻流。其核心功能包括：

• 🗣? 自動語音檢測與話輪轉換：內置實現，開發者只需專注于用戶響應邏輯的開發。
• 💻 自動生成交互界面：使用.ui.launch()方法即可啟動支持WebRTC的Gradio內置界面。
• 🔌 自動WebRTC支持：通過.mount(app)方法將流媒體掛載到FastAPI應用，即可為開發者的自定義前端生成WebRTC端點！
• ?? WebSocket支持：使用.mount(app)方法將流媒體掛載到FastAPI應用，即可為開發者的自定義前端生成WebSocket端點！
• 📞 自動電話支持：通過Stream的fastphone()方法啟動應用，即可獲取免費臨時電話號碼！
• 🤖 完全可定制的后端：流媒體Stream可輕松掛載至FastAPI應用，方便開發者擴展以滿足生產需求。參考示例：Talk To Claude🖇?鏈接11-3，了解如何為自定義JS前端提供服務。

更多信息請參考FastRTC官方文檔🖇?鏈接11-4，其中Cookbook部分是使用FastRTC構建的應用程序集合，在這里可找到大量FastRTC示例程序。

11.1.2 FastRTC安裝、內置功能與自定義路由

FastRTC實戰部分從FastRTC安裝講起，然后介紹其內置功能：TTS和STT，最后介紹自定義路由與前端集成。

1. FastRTC安裝

安裝FastRTC命令：

pip install fastrtc

FastRTC常用內置功能包括停頓檢測功能（ReplyOnPause）、停頓詞檢測功能（ReplyOnStopWords ）、文本轉語音功能（Text-To-Speech）和語音轉文本功能（Speech-To-Text），要使用這四種功能需安裝擴展組件vad、stopword、tts和stt：

pip install "fastrtc[vad, stopword, tts, stt]"

這四種功能的詳細講解請參見官方Audio Streaming🖇?鏈接11-5，限于篇幅，下面只講述TTS與STT。

2. 內置功能：TTS

文本轉語音（TTS）：若已安裝fastrtc的tts擴展組件，就可以使用設備端文本轉語音模型。從fastrtc導入get_tts_model函數并通過名稱調用指定模型（當前僅支持kokoro模型），get_tts_model函數可以返回包含三個方法的模型對象：

• tts：同步文本轉語音。
• stream_tts_sync：同步文本轉語音并流式傳輸。
• stream_tts：異步文本轉語音并流式傳輸。

三種方法可根據需求調用，代碼如下所示：

from fastrtc import get_tts_model
model = get_tts_model(model="kokoro")
for audio in model.stream_tts_sync("Hello, world!"):yield audio
async for audio in model.stream_tts("Hello, world!"):yield audio
audio = model.tts("Hello, world!")

提示：通過向三種方法傳入KokoroTTSOptions實例可自定義音頻參數，可用音色列表參見hexgrad/Kokoro-82M🖇?鏈接11-6。調用代碼如下所示：

from fastrtc import KokoroTTSOptions, get_tts_model
model = get_tts_model(model="kokoro")
options = KokoroTTSOptions(voice="af_heart",speed=1.0, lang="en-us")
audio = model.tts("Hello, world!", options=options)

3. 內置功能：STT

語音轉文本（STT）：若已安裝fastrtc的stt或stopword擴展組件，就可以使用設備端語音轉文本模型。從fastrtc導入get_stt_model函數并通過名稱調用指定模型（當前僅支持moonshine/base和moonshine/tiny模型），get_stt_model函數返回包含以下方法的模型對象：

• stt：同步語音轉文本。

不過當前stt模型僅支持英語識別，調用代碼如下所示：

from fastrtc import get_stt_model
model = get_stt_model(model="moonshine/base")
audio = (16000, np.random.randint(-32768, 32768, size=(1, 16000)))
text = model.stt(audio)

可參閱示例LLM Voice Chat🖇?鏈接11-7，了解在ReplyOnPause的構造參數handler中使用stt方法的范例。

4. 自定義路由與前端集成

在將Stream掛載至FastAPI應用后，即可添加自定義路由以部署自有前端或實現附加功能。比如，將函數serve_frontend調用添加到根目錄，代碼如下所示：

from fastapi.responses import HTMLResponse
from fastapi import FastAPI
from fastrtc import Stream
stream = Stream(...)
app = FastAPI()
stream.mount(app)
# Serve a custom frontend
@app.get("/")
async def serve_frontend():return HTMLResponse(content=open("index.html").read())

這里出現了Stream類，它是FastRTC核心，下面重點講解。

11.1.3 FastRTC核心特性：Stream及其構造參數

FastRTC的核心是Stream對象，它可用于傳輸音頻、視頻或兩者皆可。由于Stream可用于大模型的多模態實現，所以本節單獨詳細講解Stream的各項構造參數和使用方法。

1. mode、modality與handler（StreamHandler）

以下是一個創建垂直翻轉視頻流的簡單示例，我們將用它來解釋Stream對象的核心概念。代碼如下所示：

from fastrtc import Stream
import gradio as gr
import numpy as np
def detection(image, slider):return np.flip(image, axis=0)stream = Stream(handler=detection, modality="video", mode="send-receive", additional_inputs=[gr.Slider(minimum=0, maximum=1, step=0.01, value=0.3)],additional_outputs=None, additional_outputs_handler=None)

Stream的前三項參數講解如下：

• mode：流模式。FastRTC支持三種流傳輸模式：①send-receive：雙向流傳輸（默認）。②send：僅客戶端到服務器。③receive：僅服務器到客戶端。
• modality：傳輸形態。FastRTC支持三種傳輸形態：①video：視頻流傳輸。②audio：音頻流傳輸。③audio-video：音視頻復合流傳輸。
• handler：處理程序。handler是Stream對象的主要參數，根據傳輸形態和流模式的不同，handler應是一個函數或是繼承自StreamHandler或AudioVideoStreamHandler的類，情況總結如表11-1所示：

表11-1

傳輸形態\流模式	send-receive	send	receive
video	接收視頻幀并返回新視頻幀的函數	接收視頻幀并返回新視頻幀的函數	接收視頻幀并返回新視頻幀的函數
audio	StreamHandler或AsyncStreamHandler子類	StreamHandler或AsyncStreamHandler子類	生成音頻幀的生成器
audio-video	AudioVideoStreamHandler或AsyncAudioVideoStreamHandler子類	暫不支持	暫不支持

需要說明的是，ReplyOnPause和ReplyOnStopWords是StreamHandler的具體實現，StreamHandler作為底層抽象接口，完全掌控輸入音頻流和輸出音頻流的生成過程，其成員函數有：receive()、emit()、copy()、start_up()、send_message()、reset()、shutdown()等。StreamHandler的完整復雜示例參考：Talk To Gemini🖇?鏈接11-8。

2. additional_inputs及輸入鉤子

通過additional_inputs參數，可為Stream添加額外輸入項。這些輸入將顯示在自動生成的Gradio界面中，并作為附加參數傳遞給處理程序handler。在自動生成的Gradio界面中，這些輸入項將與Gradio組件相對應的Python數據類型保持一致。例如使用gr.Slider作為附加輸入，因此傳遞的參數類型為float。有關組件完整列表及其對應類型請參閱4.3.3節。

請求輸入（Requesting Input）：在ReplyOnPause和ReplyOnStopWords模式下，所有附加輸入數據都會自動傳遞至生成器。而對于音頻流處理器StreamHandler，需要手動從客戶端請求輸入數據，具體實現方式如下：

• 對于AsyncStreamHandler：在emit或receive方法中調用await self.wait_for_args()。
• 對于StreamHandler：調用self.wait_for_args_sync()。

通過StreamHandler的latest_args屬性可訪問組件的最新值。該latest_args是一個存儲各參數值的列表，其中第0個索引為占位字符串__webrtc_value__。

輸入鉤子（Input Hook）：在Gradio界面之外，我們可以通過Stream對象的set_input方法自由更新輸入參數。常見做法是通過POST請求發送更新后的數據，代碼如下所示：

from pydantic import BaseModel, Field
from fastapi import FastAPI
class InputData(BaseModel):webrtc_id: strconf_threshold: float = Field(ge=0, le=1)
app = FastAPI()
stream.mount(app)
@app.post("/input_hook")
async def _(data: InputData):stream.set_input(data.webrtc_id, data.conf_threshold)

更新后的數據將在下一次調用時傳遞給函數。

3. additional_outputs及輸出鉤子

我們可以通過handler函數返回AdditionalOutputs實例來添加額外輸出項。以下示例除了返回數據幀，還可將畫面中檢測到的目標數量作為附加項返回：

from fastrtc import Stream, AdditionalOutputs
import gradio as gr
def detection(image, conf_threshold=0.3):processed_frame, n_objects = process_frame(image, conf_threshold)return processed_frame, AdditionalOutputs(n_objects)
stream = Stream(handler=detection, modality="video", mode="send-receive",additional_inputs=[gr.Slider(minimum=0, maximum=1, step=0.01, value=0.3)],additional_outputs=[gr.Number()], # (5)additional_outputs_handler=lambda component, n_objects: n_objects
)

我們向附加輸出項添加了gr.Number()組件，并提供了additional_outputs_handler函數。該函數僅用于Gradio界面，其功能是接收額外輸出中的組件當前狀態和AdditionalOutputs實例，并返回更新后的組件狀態。在本例中，接受gr.Number()組件和AdditionalOutputs中的檢測目標數量n_objects，并用n_objects更新gr.Number()組件。

需要注意，由于webRTC延遲極低，通常不建議在每幀畫面都返回附加輸出項。

輸出鉤子（Output Hook）：在Gradio界面之外，可以通過調用StreamingResponse對象的output_stream方法自由獲取輸出數據，常見做法是通過GET請求和webrtc_id獲取輸出數據流。代碼如下所示：

from fastapi.responses import StreamingResponse
@app.get("/updates")
async def stream_updates(webrtc_id: str):async def output_stream():async for output in stream.output_stream(webrtc_id):# Output is the AdditionalOutputs instance# Be sure to serialize it however you would likeyield f"data: {output.args[0]}\n\n"return StreamingResponse(output_stream(), media_type="text/event-stream")

4. track_constraints：控制流數據

您可通過指定track_constraints參數控制數據通過流傳輸至服務器的方式。例如，可通過以下方式控制從攝像頭捕獲的幀尺寸：

track_constraints = {"width": {"exact": 500},"height": {"exact": 500}, "frameRate": {"ideal": 30}}
webrtc = Stream(handler=..., track_constraints=track_constraints,modality="video", mode="send-receive")

完整約束條件文檔參見：MediaTrackConstraints - Constraints🖇?鏈接11-9。此外，Stream還有限制并發連接數參數concurrency_limit、時間控制參數time_limit、RTP加密的參數rtp_params，處理流數據的參數event_handler，底層Gradio WebRTC組件實例參數webrtc_component，Gradio Blocks UI示例_ui等，RTCPeerConnection配置參數rtc_configuration、server_rtc_configuration等。

更多Stream、ReplyOnPause、StreamHandler、Utils及TURN Credential參數請參閱：FastRTC - API Reference🖇?鏈接11-10。

11.1.4 Stream三種運行方式和三種典型示例

為了讓讀者更熟悉Stream用法，下面展示Stream三種運行方式：.ui.launch()、.fastphone()和.mount(app)，以及四種典型示例：重復音頻、LLM語音聊天、網絡攝像頭圖像翻轉和物體檢測。

1. Stream三種運行方式：ui.launch()、.fastphone()與.mount(app)

導入Stream類并傳入處理程序handler后，Stream有三種主要運行方式（以下為官方使用指南的簡化版本）：

• .ui.launch()：啟動內置UI交互界面，便于測試和共享基于Gradio構建的流媒體應用。開發者可以通過設置Stream對象的UI屬性來更改UI，另外關于如何使用fastrtc構建Gradio引用，請參考🖇?鏈接11-11。
• .fastphone()：獲取免費臨時電話號碼，用于呼叫接入Stream（需提供HuggingFace訪問令牌）。
• .mount(app)：將Stream掛載到FastAPI應用程序，適合與現有生產系統集成或構建自定義界面。

警告說明：WebSocket文檔僅適用于音頻流，電話集成文檔僅適用于雙向傳輸模式(send-receive)的音頻流。
Stream運行示例如下：

• Gradio：stream.ui.launch()，定義Stream對象后直接調用即可。
• Telephone Integration (Audio Only)：電話集成，FastRTC通過fastphone()方法提供內置電話支持：

# Launch with a temporary phone number
stream.fastphone(# Optional: If None, will use the default token in your machine or read from the HF_TOKEN environment variabletoken="your_hf_token", host="127.0.0.1", port=8000
)

調用該方法將顯示一個電話號碼及臨時驗證碼，用于連接至數據流，每個自然月通話時長限制為10分鐘。關于StreamHandler兼容電話使用場景的設置，請參考11.3.3節。

• FastAPI：通過.mount(app)方法提供WebRTC和WebSocket支持，代碼如下所示：

app = FastAPI()
stream.mount(app)
# Optional: Add routes
@app.get("/")
async def _():return HTMLResponse(content=open("index.html").read())
# uvicorn app:app --host 0.0.0.0 --port 8000

2. Stream簡單示例：重復音頻與圖像翻轉

本節展示FastRTC兩個簡單示例，包括：重復音頻、網絡攝像頭圖像翻轉。

重復語音（Echo Audio）示例代碼如下：

# main.py
from fastrtc import Stream, ReplyOnPause
import numpy as np
from fastapi import FastAPI
app = FastAPI()
def echo(audio: tuple[int, np.ndarray]):# The function will be passed the audio until the user pauses# Implement any iterator that yields audioyield audio
stream = Stream(handler=ReplyOnPause(echo),modality="audio",mode="send-receive")
stream.mount(app)
# run with `uvicorn main:app`

網絡攝像頭圖像翻轉（Webcam Stream）代碼如下所示：

from fastrtc import Stream
import numpy as np
def flip_vertically(image):return np.flip(image, axis=0)
stream = Stream(handler=flip_vertically,modality="video", mode="send-receive")

另外還有物體檢測（Object Detection），請參考11.5.1節。

3. Stream復雜示例：LLM語音聊天

更復雜的示例LLM語音聊天（LLM Voice Chat），代碼如下所示：

from fastrtc import (ReplyOnPause, AdditionalOutputs, Stream,audio_to_bytes, aggregate_bytes_to_16bit
)
import gradio as gr
from groq import Groq
import anthropic
from elevenlabs import ElevenLabs
groq_client = Groq()
claude_client = anthropic.Anthropic()
tts_client = ElevenLabs()
# See "Talk to Claude" in Cookbook for an example of how to keep 
# track of the chat history.
def response(audio: tuple[int, np.ndarray]):prompt = groq_client.audio.transcriptions.create(file=("audio-file.mp3", audio_to_bytes(audio)),model="whisper-large-v3-turbo",response_format="verbose_json",).textresponse = claude_client.messages.create(model="claude-3-5-haiku-20241022",max_tokens=512,messages=[{"role": "user", "content": prompt}],)response_text = " ".join(block.textfor block in response.contentif getattr(block, "type", None) == "text")iterator = tts_client.text_to_speech.stream(text=response_text,voice_id="JBFqnCBsd6RMkjVDRZzb",model_id="eleven_multilingual_v2",output_format="pcm_24000")for chunk in aggregate_bytes_to_16bit(iterator):audio_array = np.frombuffer(chunk, dtype=np.int16).reshape(1, -1)yield (24000, audio_array)
stream = Stream(modality="audio", mode="send-receive",handler=ReplyOnPause(response))

示例中函數ElevenLabs.text_to_speech.stream()將文本轉換為所選語音的音頻流，并以音頻流的形式返回音頻，使用詳情請訪問：Elevenlabs - ENDPOINTS - TTS - Stream speech🖇?鏈接11-12。

11.1.5 配置FastRTC連接TURN服務器的三種部署方案

在配置防火墻的網絡環境（如HuggingFace Spaces、RunPod）中部署時，WebRTC直連可能被阻斷，此時需通過TURN服務器中轉用戶間的音視頻流量，關于TURN服務器請參考11.3.1節。本節介紹配置FastRTC連接TURN服務器的三種方案，包括：Cloudflare Calls API方案、Twilio API方案和自檢TURN服務器。

技術說明：若構建獨立Gradio應用，Stream類的rtc_configuration參數可直接傳遞至WebRTC組件，見Twilio API方案。

1. Cloudflare Calls API方案

Cloudflare通過Cloudflare Calls提供托管式TURN服務，有兩種方法可以實現：（1）Hugging Face令牌方案。Cloudflare與Hugging Face合作，憑Hugging Face賬戶的令牌可免費使用10GB/月的WebRTC流量。設置代碼如下所示：

from fastrtc import Stream, get_cloudflare_turn_credentials_async, get_cloudflare_turn_credentials
# Make sure the HF_TOKEN environment variable is set
# Or pass in a callable with all arguments set
TOKEN = "hf_..."
async def get_credentials():return await get_cloudflare_turn_credentials_async(hf_token=TOKEN)stream = Stream(handler=..., modality="audio", mode="send-receive",rtc_configuration=get_credentials,server_rtc_configuration=get_cloudflare_turn_credentials(ttl=360_000))

server_rtc_configuration：服務器端RTCPeerConnection配置（如ICE服務器配置）。建議（非必須）在服務器端配置RTC參數，但最佳實踐是在客戶端設置服務器的短期憑證（例如調用get_cloudflare_turn_credentials*時默認TTL值為10分鐘），但也可以在服務器與客戶端之間共享相同憑證。

（2）使用Cloudflare API令牌。當用戶用完每月免費配額后，可創建免費Cloudflare賬戶繼續使用，步驟如下：注冊Cloudflare賬戶（🖇?鏈接11-13）并進入控制臺Calls模塊（🖇?鏈接11-14），選擇Create -> TURN App，命名應用（如fastrtc-demo）后單擊Create按鈕。然后記錄TURN的ID令牌（通常導出為TURN_KEY_ID）和API令牌（導出為TURN_KEY_API_TOKEN）。配置完成后，可通過如下方式連接WebRTC組件：

from fastrtc import Stream, get_cloudflare_turn_credentials_async
# Make sure the TURN_KEY_ID and TURN_KEY_API_TOKEN environment variables are set
stream = Stream(handler=..., modality="audio", mode="send-receive",rtc_configuration=get_cloudflare_turn_credentials_async)

2. Twilio API方案

實現配置FastRTC連接TURN服務器的另外一種簡便部署方式是使用類似Twilio的服務提供商。創建免費Twilio賬戶后，通過pip安裝twilio包，隨后可通過以下方式從WebRTC組件進行連接：

from fastrtc import Stream
from twilio.rest import Client
import os
account_sid = os.environ.get("TWILIO_ACCOUNT_SID")
auth_token = os.environ.get("TWILIO_AUTH_TOKEN")
client = Client(account_sid, auth_token)
token = client.tokens.create()
rtc_configuration = {"iceServers": token.ice_servers,"iceTransportPolicy": "relay",
}
Stream(handler=..., modality="audio", mode="send-receive",rtc_configuration=rtc_configuration)

還可使用get_twilio_turn_credentials輔助函數實現自動登錄，它的內部封裝了生成rtc_configuration的代碼，調用代碼如下所示：

from gradio_webrtc import get_twilio_turn_credentials
# Will automatically read the TWILIO_ACCOUNT_SID and TWILIO_AUTH_TOKEN
# env variables but you can also pass in the tokens as parameters
rtc_configuration = get_twilio_turn_credentials()

3. 自建AWS TURN服務器

官方開發了可自動部署TURN服務器至AWS（Amazon Web Services，亞馬遜網絡服務）的腳本，具體操作可參照以下步驟：
（1）環境準備，分為以下四步：
①若未安裝aws命令行工具，請執行pip install awscli。
②克隆以下代碼倉庫：turn-server-deploy🖇?鏈接11-15。
③登錄AWS賬戶，創建具有以下權限的IAM用戶：AWSCloudFormationFullAccess和AmazonEC2FullAccess。然后為該用戶創建密鑰對，記錄為"Access Key ID"和"Secret Access Key"，最后執行aws configure，輸入訪問秘鑰ID和機密訪問密鑰登錄aws命令行工具。可以使用以下命令來測試登錄aws sts get-caller-identity，如果返回當前用戶的身份信息，則表示已成功使用IAM用戶通過CLI登錄。
④使用aws命令行創建ec2密鑰對（將your-key-name替換為指定的名稱），創建命令如下所示：

aws ec2 create-key-pair --key-name your-key-name --query 'KeyMaterial' --output text > your-key-name.pem

（2）配置部署腳本：首先打開代碼倉庫turn-server-deploy中的parameters.json文件，填寫以下參數的正確值：
①KeyName：剛創建的密鑰文件名（如your-key-name，省略后綴.pem）。②TurnUserName：連接服務器所需用戶名。
③TurnPassword：連接服務器所需密碼。
④InstanceType：實例類型，可選值：t3.micro、t3.small、t3.medium、c4.large、c5.large。
（3）啟動堆棧，獲取EC2服務器的公網IP。首先根據倉庫turn-server-deploy中的部署文件創建堆棧，命令如下所示：

aws cloudformation create-stack \--stack-name turn-server \--template-body file://deployment.yml \--parameters file://parameters.json \--capabilities CAPABILITY_IAM

然后，通過以下命令等待堆棧啟動完成：

aws cloudformation wait stack-create-complete \--stack-name turn-server

最后，使用以下命令獲取EC2服務器的公網IP：

aws cloudformation describe-stacks \--stack-name turn-server \--query 'Stacks[0].Outputs' > server-info.json

server-info.json文件將包含服務器的公網IP和公共DNS信息：

[{"OutputKey": "PublicIP","OutputValue": "35.173.254.80","Description": "Public IP address of the TURN server"},{"OutputKey": "PublicDNS","OutputValue": "ec2-35-173-254-80.compute-1.amazonaws.com","Description": "Public DNS name of the TURN server"}
]

（4）通過rtc_configuration參數從Gradio WebRTC組件連接至EC2服務器，連接代碼如下所示：

from fastrtc import Stream
rtc_configuration = {"iceServers": [{"urls": "turn:35.173.254.80:80","username": "<my-username>", # TurnUserName"credential": "<my-password>" # TurnPassword},]
}
gr.WebRTC.stream = Stream(handler=..., modality="audio", mode="send-receive",rtc_configuration=rtc_configuration)