Gradio全解11——Streaming：流式传输的视频应用（1）——FastRTC：Python实时通信库

前言

本系列文章主要介绍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所示：

需要说明的是，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: str
    conf_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 like
            yield 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 variable
    token="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 audio
    yield 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",
    ).text
    response = claude_client.messages.create(
        model="claude-3-5-haiku-20241022",
        max_tokens=512,
        messages=[{"role": "user", "content": prompt}],
    )
    response_text = " ".join(block.text
        for block in response.content
        if 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)