本章目錄如下：

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.9 使用FastRTC+Gemini創建實時沉浸式音頻+視頻的藝術評論家

作為對本章知識的總結，本節將實現一個綜合演示：讓Gemini扮演藝術評論家，對用戶通過FastRTC上傳的藝術作品進行點評。本節內容包括準備工作及音頻圖像編碼器、實現Gemini音視頻處理程序使用gr.Blocks替換Stream.ui。

11.9.1 準備工作及音頻圖像編碼器

本節先介紹項目及安裝等準備工作，然后實現音頻和圖像編碼器。

1. 項目說明及準備工作

FastRTC是一個支持通過WebRTC構建低延遲實時應用的庫，Gemini是DeepMind發布的支持多模態和實時任務的大模型。本演示將完成以下工作：

• 將網絡攝像頭和麥克風數據流式傳輸至Gemini的實時會話。
• 定期發送視頻幀（及可選上傳圖像）至模型。
• 實時流式返回模型的音頻響應。
• 創建精美的全屏Gradio WebRTC用戶界面。

在開始之前，需已安裝Python>=3.10并獲取GEMINI_API_KEY，并安裝以下依賴：

pip install "fastrtc[vad, tts]" gradio google-genai python-dotenv websockets pillow

2. 音頻和圖像編碼器

本節實現編碼器功能，它將音頻轉換為base64編碼數據，將圖像轉換為base64編碼的JPEG格式，轉換代碼如下所示：

import base64
import numpy as np
from io import BytesIO
from PIL import Image
def encode_audio(data: np.ndarray) -> dict:"""Encode audio data (int16 mono) for Gemini."""return {"mime_type": "audio/pcm","data": base64.b64encode(data.tobytes()).decode("UTF-8"),}
def encode_image(data: np.ndarray) -> dict:with BytesIO() as output_bytes:pil_image = Image.fromarray(data)pil_image.save(output_bytes, "JPEG")bytes_data = output_bytes.getvalue()base64_str = str(base64.b64encode(bytes_data), "utf-8")return {"mime_type": "image/jpeg", "data": base64_str}

這段代碼包含兩個函數，分別用于音頻和圖像數據的編碼處理，解讀如下：

• encode_audio：接收int16格式的單聲道音頻numpy數組，將音頻數據轉換為字節流并進行base64編碼，最后返回包含MIME類型和編碼數據的字典。輸出格式適用于Gemini系統。
• encode_image：接收圖像數據的numpy數組，使用Pillow庫將數組轉為JPEG格式圖像。將圖像數據轉換為字節流并進行base64編碼，最后返回包含JPEG類型和編碼數據的字典。

兩個函數都實現了將原始二進制數據轉換為base64編碼字符串的功能，并附帶相應的MIME類型信息，這種編碼方式常用于網絡傳輸或API交互場景。音頻處理保持原始PCM格式，而圖像處理則轉換為JPEG格式進行壓縮。

11.9.2 使用Gemini+fastrtc.Stream進行音視頻處理

本節先詳細講述實時進行音視頻發送與接受的GeminiHandler對象的處理邏輯，然后設置fastrtc.Stream并啟動其UI進行展示。

1. GeminiHandler：實時音視頻發送與接受

異步音視頻流處理核心類GeminiHandler的實現代碼如下所示：

import asyncio
import os
import time
import numpy as np
import websockets
from dotenv import load_dotenv
from google import genai
from fastrtc import AsyncAudioVideoStreamHandler, wait_for_item, WebRTCError
load_dotenv()
class GeminiHandler(AsyncAudioVideoStreamHandler):def __init__(self) -> None:super().__init__("mono",output_sample_rate=24000,input_sample_rate=16000,)self.audio_queue = asyncio.Queue()self.video_queue = asyncio.Queue()self.session = Noneself.last_frame_time = 0.0self.quit = asyncio.Event()async def start_up(self):await self.wait_for_args()api_key = self.latest_args[3]hf_token = self.latest_args[4]if hf_token is None or hf_token == "":raise WebRTCError("HF Token is required")os.environ["HF_TOKEN"] = hf_tokenclient = genai.Client(api_key=api_key, http_options={"api_version": "v1alpha"})config = {"response_modalities": ["AUDIO"], "system_instruction": "You are an art critic that will critique the artwork passed in as an image to the user. Critique the artwork in a funny and lighthearted way. Be concise and to the point. Be friendly and engaging. Be helpful and informative. Be funny and lighthearted."}async with client.aio.live.connect(model="gemini-2.0-flash-exp", # Replaceable version: gemini-2.0-flash. Latest version: gemini-live-2.5-flash-previewconfig=config,) as session:self.session = sessionwhile not self.quit.is_set():turn = self.session.receive()try:async for response in turn:if data := response.data:audio = np.frombuffer(data, dtype=np.int16).reshape(1, -1)self.audio_queue.put_nowait(audio)except websockets.exceptions.ConnectionClosedOK:print("connection closed")break# Video: receive and (optionally) send frames to Geminiasync def video_receive(self, frame: np.ndarray):self.video_queue.put_nowait(frame)if self.session and (time.time() - self.last_frame_time > 1.0):self.last_frame_time = time.time()await self.session.send(input=encode_image(frame))# If there is an uploaded image passed alongside the WebRTC component,# it will be available in latest_args[2]if self.latest_args[2] is not None:await self.session.send(input=encode_image(self.latest_args[2]))async def video_emit(self) -> np.ndarray:frame = await wait_for_item(self.video_queue, 0.01)if frame is not None:return frame# Fallback while waiting for first framereturn np.zeros((100, 100, 3), dtype=np.uint8)# Audio: forward microphone audio to Geminiasync def receive(self, frame: tuple[int, np.ndarray]) -> None:_, array = framearray = array.squeeze()  # (num_samples,)audio_message = encode_audio(array)if self.session:await self.session.send(input=audio_message)# Audio: emit Gemini’s audio back to the clientasync def emit(self):array = await wait_for_item(self.audio_queue, 0.01)if array is not None:return (self.output_sample_rate, array)return arrayasync def shutdown(self) -> None:if self.session:self.quit.set()await self.session.close()self.quit.clear()

該類繼承自AsyncAudioVideoStreamHandler，實現了一個基于Gemini Live API的、通過WebRTC進行實時音視頻處理的雙向流傳輸系統，它使用Gemini模型進行實時藝術評論（圖像分析）和對話處理，核心功能解析如下：

1. __init__(self)方法：初始化類參數。①音頻隊列（audio_queue）和視頻隊列（video_queue）：使用asyncio實現非阻塞IO操作，用于異步數據交換。②quit事件：傳輸關閉控制機制。③采樣率配置：支持16kHz輸入/24kHz輸出的音頻流轉換。
2. start_up(self)：初始化Gemini會話配置并連接，處理響應消息。 config：系統指令設置響應模態為AUDIO，并讓模型扮演藝術評論角色。client.aio.live.connect(...) ：配置模型參數（gemini-2.0-flash-exp），實時持續接收Gemini響應并存入客戶端的音頻隊列audio_queue。
3. 音頻處理流程。receive()：接受麥克風輸入，編碼后發送至Gemini。emit()：從音頻隊列audio_queue獲取Gemini響應音頻，并返回給客戶端處理。兩個函數通過隊列緩沖處理音頻數據，最終實現音頻流的雙向傳輸。
4. 視頻處理流程。video_receive()：接收攝像頭視頻流圖像幀并存入video_queue，按1秒間隔發送至Gemini，且每秒最多發送一幀視頻（避免對API造成過載）。同時可額外處理預設圖像（latest_args[2] ），因此可選擇同時發送上傳的圖像（gr.Image）和網絡攝像頭視頻幀。video_emit()：從視頻隊列video_queue獲取視頻幀輸出（含空幀處理），將輸出發送到客戶端。注意：Gemini Live API只返回評論視頻幀和圖像幀的音頻流，視頻隊列video_queue中的視頻直接來自客戶端的攝像頭。

系統在初始化時加載環境變量，然后建立Gemini長連接并循環處理音視頻數據流：每秒發送一幀視頻進行分析，并實時雙向傳輸音頻，最后再關閉時清理會話資源。GeminiHandler類的模型版本可替換，支持自定義系統指令，其模塊化的隊列設計非常便于擴展 。

2. 設置流媒體fastrtc.Stream對象并啟動UI

我們將在WebRTC組件旁添加一個可選的gr.Image輸入組件，在向Gemini發送幀時，處理程序可通過self.latest_args[2]訪問該圖像。fastrtc.Stream代碼如下所示：

import gradio as gr
from fastrtc import Stream, WebRTC, get_hf_turn_credentials
stream = Stream(handler=GeminiHandler(),modality="audio-video",mode="send-receive",server_rtc_configuration=get_hf_turn_credentials(ttl=600*10000),rtc_configuration=get_hf_turn_credentials(),additional_inputs=[gr.Markdown("## 🎨 Art Critic\n\n""Provide an image of your artwork or hold it up to the webcam, and Gemini will critique it for you.""To get a Gemini API key, please visit the [Gemini API Key](https://aistudio.google.com/apikey) page.""To get an HF Token, please visit the [HF Token](https://huggingface.co/settings/tokens) page."),gr.Image(label="Artwork", value="mona_lisa.jpg", type="numpy", sources=["upload", "clipboard"]),gr.Textbox(label="Gemini API Key", type="password"),gr.Textbox(label="HF Token", type="password"),],ui_args={"icon": "https://www.gstatic.com/lamda/images/gemini_favicon_f069958c85030456e93de685481c559f160ea06b.png","pulse_color": "rgb(255, 255, 255)","icon_button_color": "rgb(255, 255, 255)","title": "Gemini Audio Video Chat",},time_limit=90,concurrency_limit=5,
)
if __name__ == "__main__":stream.ui.launch()

在FastRTC的Stream中，設置各項配置，核心功能有：

1. FastRTC流配置。首先，配置自定義的Gemini處理程序，設置為支持音視頻模態及雙向通信模式。然后，設置服務器端TURN和客戶端TURN，使用Hugging Face的TURN服務器實現NAT穿透，10小時(600*10000毫秒)的憑證有效期確保長時會話穩定性。
2. Gradio界面設計。首先，定義輸入組件，包括Markdown說明文檔、圖像上傳組件、和安全憑證輸入。然后，實現UI定制，包括Gemini官方圖標URL、按鈕顏色及標題。
3. 設置系統約束，包括單次會話最長90秒和最大并發連接數5。

該界面結合WebRTC+P2P通信減少中間環節延遲，實現低延遲架構。通過密碼框保護API密鑰，符合敏感信息處理規范。運行效果如圖11-10所示：

11.9.3 Gradio.Blocks替換Stream.ui實現自定義界面

在11.10.2節中，使用fastrtc.Stream對象的默認ui啟動，如果希望實現自定義界面且保留Stream設置，該怎么辦呢？

1. 實現代碼及解讀

使用gr.Blocks替換Stream.ui實現代碼如下所示：

stream = Stream(handler=GeminiHandler(),modality="audio-video",mode="send-receive",rtc_configuration=get_cloudflare_turn_credentials_async,time_limit=180 if get_space() else None,additional_inputs=[gr.Image(label="Image", type="numpy", sources=["upload", "clipboard"])],ui_args={"icon": "https://www.gstatic.com/lamda/images/gemini_favicon_f069958c85030456e93de685481c559f160ea06b.png","pulse_color": "rgb(255, 255, 255)","icon_button_color": "rgb(255, 255, 255)","title": "Gemini Audio Video Chat",},
)css = """
#video-source {max-width: 600px !important; max-height: 600 !important;}
"""
with gr.Blocks(css=css) as demo:gr.HTML("""<div style='display: flex; align-items: center; justify-content: center; gap: 20px'><div style="background-color: var(--block-background-fill); border-radius: 8px"><img src="https://www.gstatic.com/lamda/images/gemini_favicon_f069958c85030456e93de685481c559f160ea06b.png" style="width: 100px; height: 100px;"></div><div><h1>Gen AI SDK Voice Chat</h1><p>Speak with Gemini using real-time audio + video streaming</p><p>Powered by <a href="https://gradio.app/">Gradio</a> and <a href=https://freddyaboulton.github.io/gradio-webrtc/">WebRTC</a>??</p><p>Get an API Key <a href="https://support.google.com/googleapi/answer/6158862?hl=en">here</a></p></div></div>""")with gr.Row() as row:with gr.Column():webrtc = WebRTC(label="Video Chat", modality="audio-video",mode="send-receive", elem_id="video-source",rtc_configuration=get_cloudflare_turn_credentials_async,icon="https://www.gstatic.com/lamda/images/gemini_favicon_f069958c85030456e93de685481c559f160ea06b.png",pulse_color="rgb(255, 255, 255)",icon_button_color="rgb(255, 255, 255)",)with gr.Column():image_input = gr.Image(label="Image", type="numpy", sources=["upload", "clipboard"])webrtc.stream(GeminiHandler(), inputs=[webrtc, image_input],outputs=[webrtc], time_limit=180 if get_space() else None,concurrency_limit=2 if get_space() else None)
stream.ui = demo
if __name__ == "__main__":if (mode := os.getenv("MODE")) == "UI":stream.ui.launch(server_port=7860)elif mode == "PHONE":raise ValueError("Phone mode not supported for this demo")else:stream.ui.launch(server_port=7860)

本段代碼先定義fastrtc.Stream配置，然后構建Gradio界面，最后定義組件交互邏輯，詳細解讀如下：

• fastrtc.Stream流媒體配置。核心組件，處理音視頻流傳輸和AI集成，使用GeminiHandler作為數據處理核心。支持雙工通信模式(send-receive)，使用Cloudflare TURN服務器處理NAT穿透。動態設置會話時長限制180秒，通過get_space()函數判斷是否啟用限制。
• Gradio界面構建。采用Blocks模式創建響應式布局。首先，通過CSS約束視頻源尺寸(600x600像素)。然后，左側顯示Gemini品牌圖標和產品標題，右側提供API獲取指引。最后在雙列布局中，左列是實時音視頻組件(WebRTC)，右列是圖像輸入組件(支持上傳/粘貼)。嚴格使用Gemini官方視覺元素（圖標、配色），通過HTML/CSS實現品牌展示區。
• 組件交互邏輯。WebRTC組件綁定GeminiHandler處理器，同時接收音視頻流和圖像輸入。設置時長限制和并發限制數，通過get_space()函數實現動態資源配置。啟動時根據環境變量MODE選擇UI模式或Phone模式（后者暫不支持）

2. 運行效果及參考資源

最終運行效果如圖11-11所示：

如需進一步學習，請參考資源：

• Gemini音視頻聊天示例——Gemini藝術評論家參考代碼：gradio/Gemini-Art-Critic🖇?鏈接11-61。Gradio Blocks界面版本：fastrtc/gemini-audio-video🖇?鏈接11-62。
• FastRTC的Audio + Video音視頻用戶指南：🖇?鏈接11-63。
• Gradio參考資料：Create a Real-Time Immersive Audio + Video Demo with FastRTC🖇?鏈接11-64。