使用 Gemini 2.0、Gradio 和 WebRTC 构建实时视频聊天 👀👂
2024 年 12 月,Google 发布了 Gemini 2.0——对其旗舰 AI 模型进行了全面改革。其中最酷的新功能之一是能够通过 多模态实时 API 与 Gemini 进行自然、人性化的视频聊天对话。在本教程中,我们将构建一个网络应用程序,让您可以使用 Python 与 Gemini 进行实时视频聊天。
我们的应用程序将实现:
- 使用您的网络摄像头与 Gemini 进行实时视频聊天
- 实时音频流,实现自然对话
- 可选的图片上传功能
- 简洁、用户友好的界面
先决条件
- 基本的 Python 知识
- 一个拥有 Gemini API 密钥的 Google Cloud 账户。请访问 此处
- 以下 Python 包:
pip install gradio-webrtc==0.0.28 google-generativeai==0.3.0
我们的应用程序将使用 Gradio 构建,这是一个完全用 Python 构建 AI 驱动的网络应用程序的框架。Gradio 将为我们处理所有 UI 元素,因此我们只需专注于应用程序的核心逻辑。`gradio-webrtc` 包是 Gradio 的一个扩展,它通过 WebRTC(一种实时通信协议)实现低延迟音频/视频流。
`google-generativeai` 包是 Google 用于与 Gemini 交互的官方包。
实现音视频处理程序
我们应用程序的核心是 `GeminiHandler` 类,它将建立到 Gemini 服务器的音频/视频流。让我们将实现分为几个部分。
类构造函数和复制方法
类构造函数将创建我们需要处理视频会话的所有变量。即,用于存储音频和视频输出的队列,以及用于存储与 Gemini 服务器连接的会话变量。
复制方法确保每个用户都有自己独特的流处理程序。
import asyncio
from gradio_webrtc import AsyncAudioVideoStreamHandler
class GeminiHandler(AsyncAudioVideoStreamHandler):
def __init__(
self, expected_layout="mono", output_sample_rate=24000, output_frame_size=480
) -> None:
super().__init__(
expected_layout,
output_sample_rate,
output_frame_size,
input_sample_rate=16000,
)
self.audio_queue = asyncio.Queue()
self.video_queue = asyncio.Queue()
self.quit = asyncio.Event()
self.session = None
self.last_frame_time = 0
def copy(self) -> "GeminiHandler":
"""Copy gets called whenever a new user connects to the server.
This ensures that each user has an independent handler.
"""
return GeminiHandler(
expected_layout=self.expected_layout,
output_sample_rate=self.output_sample_rate,
output_frame_size=self.output_frame_size,
)
音频处理
音频处理由 `emit` 和 `receive` 方法处理。当从用户接收到新的音频帧时调用 `receive` 方法,而 `emit` 返回 Gemini 的下一个音频帧。
在 `emit` 方法中,我们将通过在后台调用 `connect` 方法(这就是 `asyncio.create_task` 的含义)连接到 Gemini API。Gemini python 库使用上下文管理器来打开和关闭连接。我们使用一个 `asyncio.Event` 来保持此上下文管理器打开,直到用户点击停止按钮或关闭页面。此时,调用 shutdown 方法并设置 `asyncio.Event`。
async def connect(self, api_key: str):
"""Connect to the Gemini API"""
if self.session is None:
client = genai.Client(api_key=api_key, http_options={"api_version": "v1alpha"})
config = {"response_modalities": ["AUDIO"]}
async with client.aio.live.connect(
model="gemini-2.0-flash-exp", config=config
) as session:
self.session = session
asyncio.create_task(self.receive_audio())
# Wait for the quit event to keep the connection open
await self.quit.wait()
async def generator(self):
while not self.quit.is_set():
turn = self.session.receive()
async for response in turn:
if data := response.data:
yield data
async def receive_audio(self):
async for audio_response in async_aggregate_bytes_to_16bit(
self.generator()
):
self.output_queue.put_nowait(audio_response)
async def receive(self, frame: tuple[int, np.ndarray]) -> None:
_, array = frame
array = array.squeeze()
audio_message = encode_audio(array)
if self.session:
await self.session.send(audio_message)
async def emit(self) -> AudioEmitType:
if not self.args_set.is_set():
await self.wait_for_args()
if self.session is None:
asyncio.create_task(self.connect(self.latest_args[1]))
array = await self.output_queue.get()
return (self.output_sample_rate, array)
def shutdown(self) -> None:
self.quit.set()
self.connection = None
self.args_set.clear()
self.quit.clear()
视频处理
视频处理将由 `video_receive` 和 `video_emit` 方法处理。对于我们的应用程序,我们将简单地将网络摄像头流显示回用户,但每 1 秒,我们将最新的网络摄像头帧以及可选的上传图像发送到 Gemini 服务器。
async def video_receive(self, frame: np.ndarray):
"""Send video frames to the server"""
if self.session:
# send image every 1 second
# otherwise we flood the API
if time.time() - self.last_frame_time > 1:
self.last_frame_time = time.time()
await self.session.send(encode_image(frame))
if self.latest_args[2] is not None:
await self.session.send(encode_image(self.latest_args[2]))
self.video_queue.put_nowait(frame)
async def video_emit(self) -> VideoEmitType:
"""Return video frames to the client"""
return await self.video_queue.get()
用户界面
最后,让我们创建具有适当样式和组件的 Gradio 界面。
在 HTML 标题下方,我们放置两行——一行用于输入 Gemini API 密钥,另一行用于启动视频聊天。页面首次打开时,只有第一行可见。一旦输入 API 密钥,第二行将可见,而第一行将不可见。
`webrtc.stream` 方法设置了我们的视频聊天。作为此事件的输入,我们将传入 API 密钥和可选的 `image_input` 组件。我们将 `time_limit=90` 设置为将视频聊天限制为 90 秒。Gemini API 的免费层级只允许两个并发连接,因此我们将 `concurrency_limit=2` 设置为确保一次只有两个用户连接。
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 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 api_key_row:
api_key = gr.Textbox(label="API Key", type="password", placeholder="Enter your API Key", value=os.getenv("GOOGLE_API_KEY"))
with gr.Row(visible=False) as row:
with gr.Column():
webrtc = WebRTC(
label="Video Chat",
modality="audio-video",
mode="send-receive",
elem_id="video-source",
# See for changes needed to deploy behind a firewall
# https://freddyaboulton.github.io/gradio-webrtc/deployment/
rtc_configuration=None,
icon="https://www.gstatic.com/lamda/images/gemini_favicon_f069958c85030456e93de685481c559f160ea06b.png",
pulse_color="rgb(35, 157, 225)",
icon_button_color="rgb(35, 157, 225)",
)
with gr.Column():
image_input = gr.Image(label="Image", type="numpy", sources=["upload", "clipboard"])
webrtc.stream(
GeminiHandler(),
inputs=[webrtc, api_key, image_input],
outputs=[webrtc],
time_limit=90,
concurrency_limit=2,
)
api_key.submit(
lambda: (gr.update(visible=False), gr.update(visible=True)),
None,
[api_key_row, row],
)
if __name__ == "__main__":
demo.launch()
结论
此实现为 Gemini AI 创建了一个功能齐全的语音聊天界面,支持音频和图像输入。WebRTC 的使用实现了实时、低延迟通信,而异步设计确保了流的有效处理。
我们的应用程序托管在 Hugging Face 这里。要了解有关 Python 的 WebRTC 流媒体的更多信息,请访问 gradio-webrtc 文档。Gradio 是一个用于在 Python 中构建自定义 UI 的出色工具,它适用于任何类型的 AI 应用程序。请在此处查看 文档。