用 Quart Gemini Live API 构建非阻塞实时对话应用并部署到 Cloud Run【免费下载链接】generative-aiSample code and notebooks for Generative AI on Google Cloud, with Gemini Enterprise Agent Platform项目地址: https://gitcode.com/GitHub_Trending/ge/generative-ai导读本文围绕 gemini-quart-cloudrun 示例应用讲解如何用异步 Python Web 框架 Quart、Gemini Live API 与 Cloud Run 构建一个非阻塞non-blocking的实时流式对话应用。文章先剖析 Quart 相对 Flask 在并发模型上的差异、以及相比裸 WebSocket 代理方案的优势再带你从 Cloud Shell 本地运行到 Cloud Run 部署的全流程最后深入app.py与index.html的源码实现理解双工作协程upstream/downstream worker如何实现真正的全双工对话以及上线前需要补齐的生产化改进点。示例应用概览这个应用演示的是Quart Gemini Live API Cloud Run三者的组合Quart 提供基于 ASGI 的异步 Web 服务与原生 WebSocket 支持Gemini Live API 提供低延迟的流式对话能力Cloud Run 负责无服务器容器化部署。应用本身是一个极简的聊天页面用户在输入框中发送文本消息页面通过 WebSocket 与服务器建立全双工连接服务器再把消息转发给 Gemini 并以流式方式把回复实时推送回页面——用户可以在 Gemini 尚未回复完上一句时连续发送下一条消息会话不会像传统同步模型那样被阻塞。应用的目录结构如下gemini/sample-apps/gemini-quart-cloudrun/ ├── README.md # 本应用说明文档 └── app/ ├── app.py # Quart 应用 Gemini Live API 服务端逻辑 ├── Dockerfile # 容器镜像构建定义 ├── deploy.sh # Cloud Build Cloud Run 部署脚本 ├── run.sh # Cloud Shell 本地运行脚本 ├── requirements.txt # 依赖声明 └── public/ └── index.html # 前端聊天页面原生 WebSocket 客户端设计理念为什么是 Quart Gemini Live APIQuart 的核心优势Quart 是构建在 ASGI 标准之上的异步 Python Web 框架专为高性能并发应用而设计。对生成式 AI 应用而言它的几个特性尤其关键异步架构基于asyncio能高效处理并发 I/O 密集操作——与 AI 模型交互、实时数据流处理正是典型的 I/O 密集场景异步模型可以避免性能退化。原生 WebSocket 支持框架内置对 WebSocket 的完整支持可建立持续的、双向的通信通道满足交互式 AI 应用对实时数据交换的需求。Flask 风格 APIAPI 设计与广为人知的 Flask 保持一致开发者迁移成本低、上手快。面向多模态流式数据优化擅长处理并转发大体积数据流这对于生成式 AI 模型可能产生的多模态输出至关重要。用 Quart Gemini Live API 构建 Gen AI 应用的关键收益响应性与自然对话Quart 原生支持非阻塞的全双工 WebSocket 通信。它在等待 Gemini 响应期间不会挂起整个请求处理回复更快、对话更流畅尤其是当应用支持音频、图片等多模态输入、对网络延迟敏感时用户可快速连续发送文本或语音消息Quart 能以更小延迟处理并支持打断interruption。并发与可扩展性可同时处理大量用户和消息多个请求与 Gemini 的回复能够并发进行。单线程事件循环设计让服务器资源利用更充分从而降低运营成本、提升可扩展性。Flask阻塞vs Quart非阻塞Flask 的工作方式是同步阻塞的一次只能处理一个请求等待 Gemini 响应期间线程被挂起客户端也必须等上一条回复返回后才能发送下一条消息交互显得缓慢。Quart 则是非阻塞并发的多个请求可以同时被处理客户端可以连续发送消息不会因等待 Gemini 而阻塞交互更顺畅。两种模型在架构层面的关键差异可对比如下特性Flask同步Quart异步请求处理一次一个、阻塞式并发、非阻塞服务器接口WSGIASGI并发模型多进程/多线程WSGI单线程 事件循环asyncio视图函数普通def函数async def函数I/O 操作阻塞非阻塞使用awaitI/O 密集任务吞吐较低较高编写复杂度起步更简单异步/await 有学习曲线说明原 README 中引用的seq_flask.png与seq_quart.png两张时序图托管在外部对象存储上当前仓库内仅保留了 seq_flask.png展示 Client → WSGIServer → FlaskApp 的同步请求/响应链路仓库内未包含对应的seq_quart.png与demo_anim.png图片资源因此本文不另行插入示意图。原文档内嵌的 Mermaid 时序图注释代码也保留在 README.md 中可在支持 Mermaid 的渲染器中查看两种模型的完整消息时序。Raw WebSocket 代理 vs Quart在仓库的 websocket-demo-app 示例中采用的是裸 WebSocket API实现一个代理函数把客户端与 Gemini Live API 直接桥接起来。这是一种实现可扩展非阻塞 Gen AI 应用的替代方案通常在你需要最大程度控制力、有非常具体的性能要求、或要实现高度自定义协议时选用。相比之下Quart 提供了更高层的抽象让基于 WebSocket 的实时应用更易开发、管理和扩展它简化了常见任务、与 HTTP 集成良好并且能受益于整个 Python 生态。尤其重要的是它能够与 Google Gen AI Python SDK 无缝配合让你更轻松地借助高层 API 在服务端处理多模态内容与函数调用。在 Cloud Shell 上运行示例应用下载源码在 Cloud Shell 中执行git clone https://github.com/GoogleCloudPlatform/generative-ai.git \ gemini/sample-apps/gemini-quart-cloudrun cd gemini/sample-apps/gemini-quart-cloudrun本地运行设置项目 ID将YOUR_PROJECT_ID替换为你的真实项目gcloud config set project YOUR_PROJECT_ID安装依赖pip install -r app/requirements.txt依赖文件 requirements.txt 声明了两个包google-genai0.8.0 quart0.20.0其中google-genai提供 Gemini Live API 的客户端与AsyncSessionquart提供异步 Web 框架与 WebSocket 支持。启动应用cd app chmod x run.sh ./run.sh脚本 run.sh 的内容如下#!/bin/bash #set -e # To run the app with Vertex AI, use this script as is. # To run the app with Gemini API key, comment out these. PROJECT_ID$(gcloud config get-value project) export PROJECT_ID LOCATIONus-central1 export LOCATION # To run the app with Gemini API key, uncomment this and specify your key # (See: https://aistudio.google.com/apikey) #export GEMINI_API_KEYYOUR GEMINI API KEY # Quart debug mode (True or False) QUART_DEBUG_MODETrue export QUART_DEBUG_MODE python3 app.py默认通过Vertex AI方式运行PROJECT_ID从gcloud config自动读取区域默认为us-central1并开启 Quart 调试模式QUART_DEBUG_MODETrue。可选使用 Gemini API Key 运行如果想改用 Gemini API Key 而非 Vertex AI编辑run.sh取消export GEMINI_API_KEYYOUR GEMINI API KEY一行的注释并填入你的 Gemini API Key。应用启动后点击 Cloud Shell 右上角的Web 预览web preview按钮打开预览页面也可以在浏览器中直接访问。本地运行时的环境变量如何生效从 app.py 可以看到客户端初始化逻辑会依据环境变量自动选择接入方式PROJECT_ID: str os.environ.get(PROJECT_ID, ) LOCATION: str os.environ.get(LOCATION, us-central1) GEMINI_API_KEY: str os.environ.get(GEMINI_API_KEY, ) QUART_DEBUG_MODE: bool os.environ.get(QUART_DEBUG_MODE) True # Gemini API Client: Use either one of the following APIs gemini_client: Client ( Client(vertexaiTrue, projectPROJECT_ID, locationLOCATION) if not GEMINI_API_KEY else Client(api_keyGEMINI_API_KEY, http_options{api_version: v1alpha}) )关键点只要没有设置GEMINI_API_KEY就走 Vertex AIvertexaiTrue此时必须提供PROJECT_ID和LOCATION一旦设置了GEMINI_API_KEY则改用 Gemini Developer API并显式指定api_versionv1alphaLive API 需要该 alpha 版本接口。因此本地用脚本运行与容器内用环境变量注入行为完全一致。构建并部署到 Cloud Run设置项目 IDgcloud config set project YOUR_PROJECT_ID执行部署脚本cd app chmod x deploy.sh ./deploy.shdeploy.sh 完成镜像构建与部署两个动作#!/bin/bash #set -e PROJECT_ID$(gcloud config get-value project) export PROJECT_ID LOCATIONus-central1 export LOCATION # To run the app with Gemini API key, uncomment this and specify your key. #export GEMINI_API_KEYYOUR GEMINI API KEY # Quart debug mode (True or False) QUART_DEBUG_MODEFalse export QUART_DEBUG_MODE # build an image gcr_image_pathgcr.io/$PROJECT_ID/gemini-quart-cloudrun gcloud builds submit --tag $gcr_image_path # deploy gcloud run deploy gemini-quart-cloudrun \ --image $gcr_image_path \ --platform managed \ --allow-unauthenticated \ --project$PROJECT_ID --region$LOCATION \ --set-env-varsPROJECT_ID$PROJECT_ID \ --set-env-varsLOCATION$LOCATION \ --set-env-varsGEMINI_API_KEY$GEMINI_API_KEY \ --set-env-varsQUART_DEBUG_MODE$QUART_DEBUG_MODE注意两点差异部署时QUART_DEBUG_MODEFalse关闭调试模式且--allow-unauthenticated允许匿名访问仅适合演示。部署成功后脚本会返回一个 Cloud Run 服务 URL浏览器打开即可体验。容器镜像定义Dockerfile 采用 Python 3.13 slim 基础镜像并通过HypercornQuart 官方推荐的 ASGI 服务器对外服务FROM python:3.13-slim WORKDIR /app COPY requirements.txt requirements.txt RUN pip install --no-cache-dir -r requirements.txt COPY . . EXPOSE 8080 CMD [hypercorn, app:app, --bind, 0.0.0.0:8080]由于 app.py 的__main__分支使用app.run(host0.0.0.0, port8080)直接以开发服务器启动用于本地运行而容器内则由 Hypercorn 加载app:app对象并绑定0.0.0.0:8080与 Cloud Run 要求的容器监听端口一致。遇到RESOURCE_EXHAUSTED报错怎么办使用 Vertex AI 运行应用时偶尔会在 Cloud Run 日志页看到RESOURCE_EXHAUSTED错误。这通常意味着你超出了Gemini API 允许的并发会话数配额。此时有两种处理方式等待几分钟后重试应用在run.sh或deploy.sh中指定 Gemini API Key改用 Gemini Developer API 绕过该配额限制。应用工作原理app.pyQuart 服务端如何工作app.py 定义了完整的服务端逻辑核心包含三层结构。1. 静态页面路由/app.route(/) async def index() - Response: return await send_from_directory(public, index.html)首页通过send_from_directory从public目录返回 index.html。2. WebSocket 端点/live这是应用交互功能的核心。它使用 Quart 的app.websocket装饰器建立与客户端的 WebSocket 连接并利用async with上下文管理器连接到 Gemini Live APIasync with gemini_client.aio.live.connect( modelGEMINI_MODEL, configgemini_config ) as gemini_session: upstream_task asyncio.create_task(upstream_worker(gemini_session, websocket)) downstream_task asyncio.create_task(downstream_worker(gemini_session, websocket))其中模型固定为gemini-2.0-flash-live-preview-04-09连接配置LiveConnectConfig(response_modalities[TEXT])声明只接收文本模态的响应。3. 双工作协程upstream_worker与downstream_worker在/live处理器内部创建两个并发运行的异步任务upstream_worker上行持续从客户端的 WebSocket 读取消息并通过gemini_session.send(inputmessage, end_of_turnTrue)发送给 Gemini。每条客户端消息被视为一次对话轮次async def upstream_worker(gemini_session: AsyncSession, client_websocket: Websocket) - None: while True: message: str await client_websocket.receive() await gemini_session.send(inputmessage, end_of_turnTrue)downstream_worker下行通过gemini_session.receive()持续接收 Gemini 的流式响应把文本与轮次结束状态打包成 JSON 后经 WebSocket 回传给客户端async def downstream_worker(gemini_session: AsyncSession, client_websocket: Websocket) - None: while True: async for response in gemini_session.receive(): if not response: continue packet: Dict[str, Any] { text: response.text if response.text else , turn_complete: response.server_content.turn_complete, } await client_websocket.send(json.dumps(packet))并发与异常管理两个 worker 通过asyncio.create_task并发调度实现客户端与 Gemini 之间的双向实时通信。asyncio.wait(..., return_whenasyncio.FIRST_EXCEPTION)用于监控两个任务的异常一旦任一任务抛错就取消其余任务并向上重抛异常done, pending await asyncio.wait( [downstream_task, upstream_task], return_whenasyncio.FIRST_EXCEPTION ) for task in pending: task.cancel() for task in done: exc task.exception() if exc: raise exc会话管理gemini_session在async with块内建立确保 WebSocket 连接终止时会话被正确关闭避免资源泄漏。finally块中还会取消遗留任务并调用gemini_session.close()收尾同时捕获asyncio.CancelledError客户端断开场景优雅退出。index.html前端如何工作前端页面结构很简单标题Gemini Live API Test、消息展示区messagesdiv以及一个消息发送表单。核心逻辑在 JavaScript 部分WebSocket 连接以wss:// 当前主机 /live建立连接与页面同源const ws_url wss:// window.location.host /live; let ws new WebSocket(ws_url);事件处理器onopen连接建立成功后启用Send按钮显示Connection opened消息并为表单绑定提交处理器onmessage解析服务器推送的 JSON 数据包。若turn_complete为true重置currentMessageId本轮结束否则为新一轮对话创建新的消息元素并把packet.text逐段追加到当前消息元素上实现逐字流式显示同时自动滚动到底部ws.onmessage function (event) { const packet JSON.parse(event.data); if (packet.turn_complete packet.turn_complete true) { currentMessageId null; return; } if (currentMessageId null) { currentMessageId Math.random().toString(36).substring(7); const message document.createElement(p); message.id currentMessageId; messagesDiv.appendChild(message); } const message document.getElementById(currentMessageId); message.textContent packet.text; messagesDiv.scrollTop messagesDiv.scrollHeight; };onclose连接关闭时禁用发送按钮、显示Connection closed并启动一个 5 秒定时器自动重连ws.onclose function () { document.getElementById(sendButton).disabled true; setTimeout(function () { ws new WebSocket(ws_url); addWebSocketHandlers(ws); }, 5000); };提交处理表单提交时把用户输入显示为 消息的本地回显随后通过ws.send(message)发给服务器再清空输入框。生产化部署的改进方向README 明确指出这是一个极简演示应用若要投入生产可从以下四个方面增强音频与图像支持应用目前仅处理文本可扩展支持音频和图片等多模态输入处理方式可参考仓库中的 intro_multimodal_live_api_genai_sdk.ipynb。Gemini Live API 速率限制当前未处理速率限制。生产环境中需要针对每 key 并发会话数和每分钟 token 数实现限流机制以应对多个客户端的流量。安全性deploy.sh中的--allow-unauthenticated使应用可被公开访问生产环境应实现认证与授权来控制访问。会话管理当前 WebSocket 处理器内的会话管理虽可用但涉及多用户或持久会话场景时需要探索更健壮的会话管理方案。参考资料Gemini Multimodal Live APIVertex AI 模型参考文档仓库内的 websocket-demo-app 示例裸 WebSocket 代理实现Google Gen AI Python SDKGetting Started with the Multimodal Live API using Gen AI SDKQuart 官方文档【免费下载链接】generative-aiSample code and notebooks for Generative AI on Google Cloud, with Gemini Enterprise Agent Platform项目地址: https://gitcode.com/GitHub_Trending/ge/generative-ai创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
用 Quart + Gemini Live API 构建非阻塞实时对话应用并部署到 Cloud Run
用 Quart Gemini Live API 构建非阻塞实时对话应用并部署到 Cloud Run【免费下载链接】generative-aiSample code and notebooks for Generative AI on Google Cloud, with Gemini Enterprise Agent Platform项目地址: https://gitcode.com/GitHub_Trending/ge/generative-ai导读本文围绕 gemini-quart-cloudrun 示例应用讲解如何用异步 Python Web 框架 Quart、Gemini Live API 与 Cloud Run 构建一个非阻塞non-blocking的实时流式对话应用。文章先剖析 Quart 相对 Flask 在并发模型上的差异、以及相比裸 WebSocket 代理方案的优势再带你从 Cloud Shell 本地运行到 Cloud Run 部署的全流程最后深入app.py与index.html的源码实现理解双工作协程upstream/downstream worker如何实现真正的全双工对话以及上线前需要补齐的生产化改进点。示例应用概览这个应用演示的是Quart Gemini Live API Cloud Run三者的组合Quart 提供基于 ASGI 的异步 Web 服务与原生 WebSocket 支持Gemini Live API 提供低延迟的流式对话能力Cloud Run 负责无服务器容器化部署。应用本身是一个极简的聊天页面用户在输入框中发送文本消息页面通过 WebSocket 与服务器建立全双工连接服务器再把消息转发给 Gemini 并以流式方式把回复实时推送回页面——用户可以在 Gemini 尚未回复完上一句时连续发送下一条消息会话不会像传统同步模型那样被阻塞。应用的目录结构如下gemini/sample-apps/gemini-quart-cloudrun/ ├── README.md # 本应用说明文档 └── app/ ├── app.py # Quart 应用 Gemini Live API 服务端逻辑 ├── Dockerfile # 容器镜像构建定义 ├── deploy.sh # Cloud Build Cloud Run 部署脚本 ├── run.sh # Cloud Shell 本地运行脚本 ├── requirements.txt # 依赖声明 └── public/ └── index.html # 前端聊天页面原生 WebSocket 客户端设计理念为什么是 Quart Gemini Live APIQuart 的核心优势Quart 是构建在 ASGI 标准之上的异步 Python Web 框架专为高性能并发应用而设计。对生成式 AI 应用而言它的几个特性尤其关键异步架构基于asyncio能高效处理并发 I/O 密集操作——与 AI 模型交互、实时数据流处理正是典型的 I/O 密集场景异步模型可以避免性能退化。原生 WebSocket 支持框架内置对 WebSocket 的完整支持可建立持续的、双向的通信通道满足交互式 AI 应用对实时数据交换的需求。Flask 风格 APIAPI 设计与广为人知的 Flask 保持一致开发者迁移成本低、上手快。面向多模态流式数据优化擅长处理并转发大体积数据流这对于生成式 AI 模型可能产生的多模态输出至关重要。用 Quart Gemini Live API 构建 Gen AI 应用的关键收益响应性与自然对话Quart 原生支持非阻塞的全双工 WebSocket 通信。它在等待 Gemini 响应期间不会挂起整个请求处理回复更快、对话更流畅尤其是当应用支持音频、图片等多模态输入、对网络延迟敏感时用户可快速连续发送文本或语音消息Quart 能以更小延迟处理并支持打断interruption。并发与可扩展性可同时处理大量用户和消息多个请求与 Gemini 的回复能够并发进行。单线程事件循环设计让服务器资源利用更充分从而降低运营成本、提升可扩展性。Flask阻塞vs Quart非阻塞Flask 的工作方式是同步阻塞的一次只能处理一个请求等待 Gemini 响应期间线程被挂起客户端也必须等上一条回复返回后才能发送下一条消息交互显得缓慢。Quart 则是非阻塞并发的多个请求可以同时被处理客户端可以连续发送消息不会因等待 Gemini 而阻塞交互更顺畅。两种模型在架构层面的关键差异可对比如下特性Flask同步Quart异步请求处理一次一个、阻塞式并发、非阻塞服务器接口WSGIASGI并发模型多进程/多线程WSGI单线程 事件循环asyncio视图函数普通def函数async def函数I/O 操作阻塞非阻塞使用awaitI/O 密集任务吞吐较低较高编写复杂度起步更简单异步/await 有学习曲线说明原 README 中引用的seq_flask.png与seq_quart.png两张时序图托管在外部对象存储上当前仓库内仅保留了 seq_flask.png展示 Client → WSGIServer → FlaskApp 的同步请求/响应链路仓库内未包含对应的seq_quart.png与demo_anim.png图片资源因此本文不另行插入示意图。原文档内嵌的 Mermaid 时序图注释代码也保留在 README.md 中可在支持 Mermaid 的渲染器中查看两种模型的完整消息时序。Raw WebSocket 代理 vs Quart在仓库的 websocket-demo-app 示例中采用的是裸 WebSocket API实现一个代理函数把客户端与 Gemini Live API 直接桥接起来。这是一种实现可扩展非阻塞 Gen AI 应用的替代方案通常在你需要最大程度控制力、有非常具体的性能要求、或要实现高度自定义协议时选用。相比之下Quart 提供了更高层的抽象让基于 WebSocket 的实时应用更易开发、管理和扩展它简化了常见任务、与 HTTP 集成良好并且能受益于整个 Python 生态。尤其重要的是它能够与 Google Gen AI Python SDK 无缝配合让你更轻松地借助高层 API 在服务端处理多模态内容与函数调用。在 Cloud Shell 上运行示例应用下载源码在 Cloud Shell 中执行git clone https://github.com/GoogleCloudPlatform/generative-ai.git \ gemini/sample-apps/gemini-quart-cloudrun cd gemini/sample-apps/gemini-quart-cloudrun本地运行设置项目 ID将YOUR_PROJECT_ID替换为你的真实项目gcloud config set project YOUR_PROJECT_ID安装依赖pip install -r app/requirements.txt依赖文件 requirements.txt 声明了两个包google-genai0.8.0 quart0.20.0其中google-genai提供 Gemini Live API 的客户端与AsyncSessionquart提供异步 Web 框架与 WebSocket 支持。启动应用cd app chmod x run.sh ./run.sh脚本 run.sh 的内容如下#!/bin/bash #set -e # To run the app with Vertex AI, use this script as is. # To run the app with Gemini API key, comment out these. PROJECT_ID$(gcloud config get-value project) export PROJECT_ID LOCATIONus-central1 export LOCATION # To run the app with Gemini API key, uncomment this and specify your key # (See: https://aistudio.google.com/apikey) #export GEMINI_API_KEYYOUR GEMINI API KEY # Quart debug mode (True or False) QUART_DEBUG_MODETrue export QUART_DEBUG_MODE python3 app.py默认通过Vertex AI方式运行PROJECT_ID从gcloud config自动读取区域默认为us-central1并开启 Quart 调试模式QUART_DEBUG_MODETrue。可选使用 Gemini API Key 运行如果想改用 Gemini API Key 而非 Vertex AI编辑run.sh取消export GEMINI_API_KEYYOUR GEMINI API KEY一行的注释并填入你的 Gemini API Key。应用启动后点击 Cloud Shell 右上角的Web 预览web preview按钮打开预览页面也可以在浏览器中直接访问。本地运行时的环境变量如何生效从 app.py 可以看到客户端初始化逻辑会依据环境变量自动选择接入方式PROJECT_ID: str os.environ.get(PROJECT_ID, ) LOCATION: str os.environ.get(LOCATION, us-central1) GEMINI_API_KEY: str os.environ.get(GEMINI_API_KEY, ) QUART_DEBUG_MODE: bool os.environ.get(QUART_DEBUG_MODE) True # Gemini API Client: Use either one of the following APIs gemini_client: Client ( Client(vertexaiTrue, projectPROJECT_ID, locationLOCATION) if not GEMINI_API_KEY else Client(api_keyGEMINI_API_KEY, http_options{api_version: v1alpha}) )关键点只要没有设置GEMINI_API_KEY就走 Vertex AIvertexaiTrue此时必须提供PROJECT_ID和LOCATION一旦设置了GEMINI_API_KEY则改用 Gemini Developer API并显式指定api_versionv1alphaLive API 需要该 alpha 版本接口。因此本地用脚本运行与容器内用环境变量注入行为完全一致。构建并部署到 Cloud Run设置项目 IDgcloud config set project YOUR_PROJECT_ID执行部署脚本cd app chmod x deploy.sh ./deploy.shdeploy.sh 完成镜像构建与部署两个动作#!/bin/bash #set -e PROJECT_ID$(gcloud config get-value project) export PROJECT_ID LOCATIONus-central1 export LOCATION # To run the app with Gemini API key, uncomment this and specify your key. #export GEMINI_API_KEYYOUR GEMINI API KEY # Quart debug mode (True or False) QUART_DEBUG_MODEFalse export QUART_DEBUG_MODE # build an image gcr_image_pathgcr.io/$PROJECT_ID/gemini-quart-cloudrun gcloud builds submit --tag $gcr_image_path # deploy gcloud run deploy gemini-quart-cloudrun \ --image $gcr_image_path \ --platform managed \ --allow-unauthenticated \ --project$PROJECT_ID --region$LOCATION \ --set-env-varsPROJECT_ID$PROJECT_ID \ --set-env-varsLOCATION$LOCATION \ --set-env-varsGEMINI_API_KEY$GEMINI_API_KEY \ --set-env-varsQUART_DEBUG_MODE$QUART_DEBUG_MODE注意两点差异部署时QUART_DEBUG_MODEFalse关闭调试模式且--allow-unauthenticated允许匿名访问仅适合演示。部署成功后脚本会返回一个 Cloud Run 服务 URL浏览器打开即可体验。容器镜像定义Dockerfile 采用 Python 3.13 slim 基础镜像并通过HypercornQuart 官方推荐的 ASGI 服务器对外服务FROM python:3.13-slim WORKDIR /app COPY requirements.txt requirements.txt RUN pip install --no-cache-dir -r requirements.txt COPY . . EXPOSE 8080 CMD [hypercorn, app:app, --bind, 0.0.0.0:8080]由于 app.py 的__main__分支使用app.run(host0.0.0.0, port8080)直接以开发服务器启动用于本地运行而容器内则由 Hypercorn 加载app:app对象并绑定0.0.0.0:8080与 Cloud Run 要求的容器监听端口一致。遇到RESOURCE_EXHAUSTED报错怎么办使用 Vertex AI 运行应用时偶尔会在 Cloud Run 日志页看到RESOURCE_EXHAUSTED错误。这通常意味着你超出了Gemini API 允许的并发会话数配额。此时有两种处理方式等待几分钟后重试应用在run.sh或deploy.sh中指定 Gemini API Key改用 Gemini Developer API 绕过该配额限制。应用工作原理app.pyQuart 服务端如何工作app.py 定义了完整的服务端逻辑核心包含三层结构。1. 静态页面路由/app.route(/) async def index() - Response: return await send_from_directory(public, index.html)首页通过send_from_directory从public目录返回 index.html。2. WebSocket 端点/live这是应用交互功能的核心。它使用 Quart 的app.websocket装饰器建立与客户端的 WebSocket 连接并利用async with上下文管理器连接到 Gemini Live APIasync with gemini_client.aio.live.connect( modelGEMINI_MODEL, configgemini_config ) as gemini_session: upstream_task asyncio.create_task(upstream_worker(gemini_session, websocket)) downstream_task asyncio.create_task(downstream_worker(gemini_session, websocket))其中模型固定为gemini-2.0-flash-live-preview-04-09连接配置LiveConnectConfig(response_modalities[TEXT])声明只接收文本模态的响应。3. 双工作协程upstream_worker与downstream_worker在/live处理器内部创建两个并发运行的异步任务upstream_worker上行持续从客户端的 WebSocket 读取消息并通过gemini_session.send(inputmessage, end_of_turnTrue)发送给 Gemini。每条客户端消息被视为一次对话轮次async def upstream_worker(gemini_session: AsyncSession, client_websocket: Websocket) - None: while True: message: str await client_websocket.receive() await gemini_session.send(inputmessage, end_of_turnTrue)downstream_worker下行通过gemini_session.receive()持续接收 Gemini 的流式响应把文本与轮次结束状态打包成 JSON 后经 WebSocket 回传给客户端async def downstream_worker(gemini_session: AsyncSession, client_websocket: Websocket) - None: while True: async for response in gemini_session.receive(): if not response: continue packet: Dict[str, Any] { text: response.text if response.text else , turn_complete: response.server_content.turn_complete, } await client_websocket.send(json.dumps(packet))并发与异常管理两个 worker 通过asyncio.create_task并发调度实现客户端与 Gemini 之间的双向实时通信。asyncio.wait(..., return_whenasyncio.FIRST_EXCEPTION)用于监控两个任务的异常一旦任一任务抛错就取消其余任务并向上重抛异常done, pending await asyncio.wait( [downstream_task, upstream_task], return_whenasyncio.FIRST_EXCEPTION ) for task in pending: task.cancel() for task in done: exc task.exception() if exc: raise exc会话管理gemini_session在async with块内建立确保 WebSocket 连接终止时会话被正确关闭避免资源泄漏。finally块中还会取消遗留任务并调用gemini_session.close()收尾同时捕获asyncio.CancelledError客户端断开场景优雅退出。index.html前端如何工作前端页面结构很简单标题Gemini Live API Test、消息展示区messagesdiv以及一个消息发送表单。核心逻辑在 JavaScript 部分WebSocket 连接以wss:// 当前主机 /live建立连接与页面同源const ws_url wss:// window.location.host /live; let ws new WebSocket(ws_url);事件处理器onopen连接建立成功后启用Send按钮显示Connection opened消息并为表单绑定提交处理器onmessage解析服务器推送的 JSON 数据包。若turn_complete为true重置currentMessageId本轮结束否则为新一轮对话创建新的消息元素并把packet.text逐段追加到当前消息元素上实现逐字流式显示同时自动滚动到底部ws.onmessage function (event) { const packet JSON.parse(event.data); if (packet.turn_complete packet.turn_complete true) { currentMessageId null; return; } if (currentMessageId null) { currentMessageId Math.random().toString(36).substring(7); const message document.createElement(p); message.id currentMessageId; messagesDiv.appendChild(message); } const message document.getElementById(currentMessageId); message.textContent packet.text; messagesDiv.scrollTop messagesDiv.scrollHeight; };onclose连接关闭时禁用发送按钮、显示Connection closed并启动一个 5 秒定时器自动重连ws.onclose function () { document.getElementById(sendButton).disabled true; setTimeout(function () { ws new WebSocket(ws_url); addWebSocketHandlers(ws); }, 5000); };提交处理表单提交时把用户输入显示为 消息的本地回显随后通过ws.send(message)发给服务器再清空输入框。生产化部署的改进方向README 明确指出这是一个极简演示应用若要投入生产可从以下四个方面增强音频与图像支持应用目前仅处理文本可扩展支持音频和图片等多模态输入处理方式可参考仓库中的 intro_multimodal_live_api_genai_sdk.ipynb。Gemini Live API 速率限制当前未处理速率限制。生产环境中需要针对每 key 并发会话数和每分钟 token 数实现限流机制以应对多个客户端的流量。安全性deploy.sh中的--allow-unauthenticated使应用可被公开访问生产环境应实现认证与授权来控制访问。会话管理当前 WebSocket 处理器内的会话管理虽可用但涉及多用户或持久会话场景时需要探索更健壮的会话管理方案。参考资料Gemini Multimodal Live APIVertex AI 模型参考文档仓库内的 websocket-demo-app 示例裸 WebSocket 代理实现Google Gen AI Python SDKGetting Started with the Multimodal Live API using Gen AI SDKQuart 官方文档【免费下载链接】generative-aiSample code and notebooks for Generative AI on Google Cloud, with Gemini Enterprise Agent Platform项目地址: https://gitcode.com/GitHub_Trending/ge/generative-ai创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
