使用 Gemini 2.0、Gradio 和 WebRTC 构建实时视频聊天 👀👂

社区文章 发布于 2025 年 1 月 13 日

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()

image/png

用户界面

最后,让我们创建具有适当样式和组件的 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 应用程序。请在此处查看 文档

社区

我尝试在 Hugging Face 上使用它,它显示等待了很长时间,并且不像您的视频演示那样工作。

您从哪里连接?能再试一次吗?也许网站流量很大。

·

嘿,我运行了应用程序,干得漂亮,喜欢你的作品
我想知道是否有可能不使用 Gemini,而是从我托管在 Runpod 上的 Ollama llava 通过 ollama API 连接,而且我在网页应用程序的设置中看到了屏幕共享功能,Gemini 或 ollama 可以看到我的屏幕并提问吗?
顺便说一句,你用 gradio 做的,尽量简化,真的很棒。

此评论已被隐藏(标记为无关话题)

注册登录 评论