拓冰网站定制 
Project Livewire 服务端实战解析用 Gemini Multimodal Live API 构建实时多模态 WebSocket 代理与工具调用【免费下载链接】generative-aiSample code and notebooks for Generative AI on Google Cloud, with Gemini Enterprise Agent Platform项目地址: https://gitcode.com/GitHub_Trending/ge/generative-aiProject Livewire 是展示 Gemini Multimodal Live API 能力的实时多模态对话应用其服务端server 组件是一个基于 Pythonwebsockets与google-genai库构建的 WebSocket 应用承担着客户端与 Gemini API 之间的代理、会话管理和工具调用枢纽角色。本文以 server/README.md 为主线结合仓库内源码与部署配置完整讲解该服务端从环境准备、配置加载、本地运行到 Cloud Run 部署、工具扩展与故障排查的全过程读完后你可以独立搭建一个语音 文本 图像 工具调用的实时 AI 对话后端。一、服务端定位它到底做了什么该服务端是 Project Livewire 的核心后端逻辑围绕五大职责展开WebSocket 通信与前端 UI 建立并维护双向、实时的 WebSocket 长连接Gemini API 交互连接并管理 Gemini Multimodal Live API 会话处理用户输入文本、音频、视频并转发 AI 生成的响应工具处理与函数调用实现工具处理机制将 Gemini 发起的函数调用路由到对应的工具天气、日历等这些工具以独立的 Google Cloud Functions 实现通过 HTTP 请求安全调用会话管理为每个连接的客户端维护会话状态保证多轮交互的上下文连续配置管理从环境变量与 Google Cloud Secret Manager 中加载 API Key、Cloud Function URL 与服务器设置。从部署定位看该服务端面向 Google Cloud Run 设计以利用其弹性伸缩与托管能力同时支持本地运行以用于开发、测试与实验。二、架构与核心组件源码解读2.1 目录结构总览gemini/multimodal-live-api/project-livewire/server/ ├── core/ │ ├── gemini_client.py # Gemini API 客户端初始化与会话创建 │ ├── session.py # 客户端会话状态与活动会话管理 │ ├── tool_handler.py # 工具执行与 Cloud Function 路由 │ └── websocket_handler.py # WebSocket 连接处理与消息流转 ├── config/ │ ├── config.py # 配置加载与管理Secret Manager 环境变量 │ └── system-instructions.txt # Gemini 模型的系统指令 ├── Dockerfile # 服务端容器镜像定义 ├── cloudbuild.yaml # Cloud Build 部署到 Cloud Run 的配置 ├── requirements.txt # Python 依赖清单 ├── README.md # 本文档 └── server.py # 主入口启动 WebSocket 服务器2.2 入口文件 server.pyasyncio 事件循环与 WebSocket 服务server.py 是整个服务端的入口。它通过websockets.serve将handle_client挂载到0.0.0.0:8081并通过await asyncio.Future()让事件循环永久运行port 8081 async with websockets.serve( handle_client, 0.0.0.0, port, ping_interval30, ping_timeout10, ): logger.info(fRunning websocket server on 0.0.0.0:{port}...) await asyncio.Future() # run forever值得注意的两个细节心跳参数ping_interval30每 30 秒发送一次 WebSocket ping、ping_timeout1010 秒未收到 pong 判定连接失效这是 Cloud Run 空闲实例回收环境下保持长连接活跃的关键配置日志治理服务器启动时通过logging.getLogger(...).setLevel(logging.ERROR)将google、urllib3.connectionpool、websockets.client、httpx等第三方库的日志级别压到 ERROR保留应用自身的 DEBUG 信息避免海量 SDK 日志淹没调试输出。日志级别本身由环境变量LOG_LEVEL控制默认INFOlogging.basicConfig(levelgetattr(logging, os.getenv(LOG_LEVEL, INFO).upper()), ...)。2.3 core/gemini_client.py双通道客户端初始化core/gemini_client.py 的create_gemini_session()根据use_vertex标志选择两种认证与接入方式Vertex AI 模式GOOGLE_GENAI_USE_VERTEXAItrue使用 ADCApplication Default Credentials初始化需要GOOGLE_CLOUD_PROJECT区域默认us-central1client genai.Client(vertexaiTrue, locationlocation, projectproject_id)Dev API 模式默认使用 API Key 初始化http_options{api_version: v1alpha}client genai.Client(vertexaiFalse, http_options{api_version: v1alpha}, api_keyapi_config.api_key)最终通过client.aio.live.connect(modelMODEL, configCONFIG)建立 Gemini Live 会话aio为异步客户端。模型与语音在 config/config.py 中按接入模式分别配置接入模式模型语音Vertex AIgemini-2.0-flash-expMODEL_GOOGLE_GENAI_USE_VERTEXAIAoedeVOICE_GOOGLE_GENAI_USE_VERTEXAIDev APImodels/gemini-2.0-flash-expMODEL_DEV_APIPuckVOICE_DEV_API2.4 core/session.py会话状态管理core/session.py 用dataclass SessionState记录单个客户端的完整状态dataclass class SessionState: is_receiving_response: bool False interrupted: bool False current_tool_execution: Optional[asyncio.Task] None current_audio_stream: Optional[Any] None genai_session: Optional[Any] None received_model_response: bool False其中current_tool_execution保存正在执行的工具任务句柄用于中断时取消active_sessions是模块级字典Dict[str, SessionState]配合create_session/get_session/remove_session完成会话生命周期管理。会话 ID 由str(id(websocket))生成。2.5 core/websocket_handler.py双向消息流与工具队列core/websocket_handler.py 是整个服务端最核心的编排模块其关键机制如下。建立连接handle_client首先create_gemini_session()创建 Gemini 会话向客户端发送{ready: true}握手消息然后进入handle_messages。双任务并发模型handle_messages使用asyncio.TaskGroup同时运行两个协程——handle_client_messages客户端 → Gemini与handle_gemini_responsesGemini → 客户端。当任一任务因Quota exceeded或connection closed异常退出时服务端会向客户端发送{type: error, data: {...}}结构化错误消息并在finally中取消未完成的任务。客户端入站消息协议handle_client_messages消息类型行为audio以mime_type: audio/pcm发送给 Gemini并置end_of_turnTrueimage以mime_type: image/jpeg发送给 Geminitext作为纯文本发送置end_of_turnTrueend记录结束信号不做转发Gemini 出站响应处理handle_gemini_responsesprocess_server_content若响应含tool_call放入asyncio.Queue由后台process_tool_queue任务异步执行主循环继续处理其他响应工具执行不阻塞音频流server_content.model_turn.parts中的inline_data音频经base64.b64encode编码后以{type: audio, data: base64}下发part.text以{type: text, data: ...}下发收到server_content.interrupted时向客户端发送{type: interrupted}通知支持用户随时打断模型回答turn_complete时发送{type: turn_complete}并重置会话中的响应状态位。工具调用协议process_tool_queue对每个function_call先向客户端发送{type: function_call, data: {name: ..., args: ...}}用于 UI 反馈然后执行工具再发送{type: function_response, data: ...}最后通过types.LiveClientToolResponse(function_responses[...])将结果回传给 Gemini 会话。异常兜底handle_client对浏览器刷新导致的异常连接code 1006、空闲超时asyncio.TimeoutError与普通异常分别发送不同error_typeconnection_closed/timeout/general的错误消息finally中通过cleanup_session取消进行中的工具任务、关闭 Gemini 会话并从活动会话表移除记录。2.6 core/tool_handler.py动态工具路由core/tool_handler.py 的execute_tool(tool_name, params)是工具的通用执行器从CLOUD_FUNCTIONS字典取工具名对应的 Cloud Function URL用urlencode将参数拼为查询串通过aiohttp发起 GET 请求并解析 JSON 响应。其返回结构为Dict[str, Any]失败时返回{error: ...}以便 Gemini 理解错误。该执行器完全由配置驱动——新增工具时无需修改 tool_handler.py详见第六节。三、环境准备本地开发与 Cloud Run 部署的前置条件3.1 本地开发环境Python 3.11与pip依赖见 requirements.txt核心包括websockets14.1、google-genai1.3.0、aiohttp3.12.14、python-dotenv1.0.1、google-cloud-secret-manager2.19.0等。3.2 Google Cloud 项目准备Cloud Run 部署在 Google Cloud Console 创建项目并执行gcloud config set project YOUR_GOOGLE_CLOUD_PROJECT启用以下 APISecret Manager API、Cloud Run API、Cloud Build API、Cloud Functions API以及若使用 Vertex APIVertex AI API安装并初始化 gcloud CLIgcloud init与gcloud auth login。3.3 API Key 与密钥Gemini API Key从 Google AI Studio 获取生产环境推荐存入 Secret Managersecret 名GOOGLE_API_KEY本地开发可用环境变量OpenWeather API Key可选用于天气工具注册 OpenWeather 获取存入 Secret ManagerOPENWEATHER_API_KEY或环境变量。3.4 服务账号Service AccountCloud Run 部署创建服务账号并授予Secret Manager Secret Accessorroles/secretmanager.secretAccessor角色部署时将服务配置为使用该账号本地开发可选下载服务账号 JSON 密钥文件设置GOOGLE_APPLICATION_CREDENTIALS环境变量指向该文件即可通过 ADC 访问 Secret Manager。3.5 Cloud Functions工具实现工具函数作为独立的 Google Cloud Functions 部署详细步骤见 cloud-functions/README.md天气工具位于 weather-tools/get-weather-tool/main.py。部署后记录各函数的触发 URL用于服务端配置。四、安装与配置4.1 安装步骤git clone https://github.com/heiko-hotz/project-livewire.git cd project-livewire/server python3 -m venv venv source venv/bin/activate # Linux/macOSWindows 使用 venv\Scripts\activate pip install -r requirements.txt4.2 环境变量与 .env 文件在server/目录创建.env文件示例内容如下GOOGLE_CLOUD_PROJECTyour-gcp-project-id LOG_LEVELINFO # Cloud Function URLs替换为实际部署后的函数 URL WEATHER_FUNCTION_URLhttps://REGION-GOOGLE_CLOUD_PROJECT.cloudfunctions.net/get-weather-tool CALENDAR_FUNCTION_URLhttps://REGION-GOOGLE_CLOUD_PROJECT.cloudfunctions.net/get-calendar-tool # 可选本地开发时作为 Secret Manager 的兜底 API Key GOOGLE_API_KEYyour_gemini_api_key OPENWEATHER_API_KEYyour_openweather_api_key # 可选本地访问 Secret Manager 时显式指定服务账号密钥文件 # GOOGLE_APPLICATION_CREDENTIALSpath/to/service-account-key.json重要务必把.env加入.gitignore与.dockerignore防止敏感信息被提交。结合源码config/config.py 还支持以下扩展变量GOOGLE_GENAI_USE_VERTEXAItrue切换到 Vertex AI 接入模式GOOGLE_CLOUD_LOCATIONVertex AI 区域默认us-central1MODEL_DEV_API/MODEL_GOOGLE_GENAI_USE_VERTEXAI覆盖模型名VOICE_DEV_API/VOICE_GOOGLE_GENAI_USE_VERTEXAI覆盖语音Dev 默认PuckVertex 默认AoedeFORECAST_FUNCTION_URL、PAST_APPOINTMENTS_FUNCTION_URL天气预报与历史日程工具 URL。CLOUD_FUNCTIONS字典在模块加载时即对 URL 做校验缺失或非https://开头会输出警告日志CLOUD_FUNCTIONS { get_weather: os.getenv(WEATHER_FUNCTION_URL), get_weather_forecast: os.getenv(FORECAST_FUNCTION_URL), get_next_appointment: os.getenv(CALENDAR_FUNCTION_URL), get_past_appointments: os.getenv(PAST_APPOINTMENTS_FUNCTION_URL), }4.3 配置加载优先级服务端按以下顺序解析配置Google Cloud Secret Manager在 Cloud Run 环境或本地配置 ADC中可访问时优先使用读取GOOGLE_API_KEY与OPENWEATHER_API_KEY。其实现位于 config/config.py 的get_secret通过projects/{project_id}/secrets/{secret_id}/versions/latest拉取最新版本环境变量含 .envSecret Manager 不可达或未配置对应项时回退到环境变量。代码中先load_dotenv()加载.env再在get_secret失败时logger.warning(...)并回退到os.getenv(...)LOG_LEVEL始终从环境变量加载默认INFO。五、运行服务端本地开发与 Cloud Run 部署5.1 本地运行配置好.env后直接启动python server.py服务端将在0.0.0.0:8081监听 WebSocket 连接。5.2 部署到 Google Cloud Run使用仓库内 cloudbuild.yaml 一键构建、推送并部署gcloud builds submit --config server/cloudbuild.yaml该流水线依次执行构建镜像gcr.io/$GOOGLE_CLOUD_PROJECT/livewire-backend→ 推送到 Container Registry → 以服务账号livewire-backend${GOOGLE_CLOUD_PROJECT}.iam.gserviceaccount.com部署到us-central1的livewire-backend服务并通过--set-env-vars GOOGLE_CLOUD_PROJECT...,LOG_LEVELINFO注入运行环境变量。镜像定义见 Dockerfile基于python:3.11-slim暴露 8081 端口启动命令python server.py。获取服务 URLgcloud run services describe livewire-backend --platform managed --region us-central1 --format value(status.url) gcloud run services list # 查看所有服务以找到前端服务名 gcloud run services describe livewire-frontend --platform managed --region us-central1 --format value(status.url)5.3 用 wscat 测试部署的服务端安装 WebSocket 命令行测试工具npm install -g wscat连接并发送测试消息export CLOUD_RUN_URL$(gcloud run services describe livewire-backend --platform managed --region us-central1 --format value(status.url)) # macOS wscat -c $(echo $CLOUD_RUN_URL | sed s|https:|wss:|) # Linux wscat -c $(echo $CLOUD_RUN_URL | sed s/https:/wss:/) # 或手动指定wscat -c wss://YOUR_CLOUD_RUN_URL连接后发送{type: text, data: Hello, are you there?}服务端应返回来自 Gemini 模型的响应语音响应对应{type: audio, data: base64 音频}消息。六、扩展与修改工具四步流程向 Gemini 暴露新工具如新的天气或日历能力通常只需修改四处Cloud Function 实现在 cloud-functions 目录下实现/修改工具逻辑并部署记录其函数 URL服务端配置config/config.py在CLOUD_FUNCTIONS字典中增加条目将代码中的工具名映射到函数 URL在CONFIG[tools]的function_declarations中新增声明包含nameGemini 调用的函数名需与CLOUD_FUNCTIONS的键一致description工具作用描述帮助 Gemini 判断何时调用parameters函数接受的参数 schema含required列表这是 Gemini 正确生成调用参数的关键当前CONFIG还通过generation_config: {response_modalities: [AUDIO], speech_config: VOICE}将响应模态固定为音频并指定语音环境变量.env新增与配置中变量名一致的 URL 变量如YOUR_NEW_TOOL_URL系统指令system-instructions.txt告知模型新工具的存在、用途与使用规则。仓库当前示例指令为询问天气时必须使用 get_weather 工具并在运行时作为system_instruction随会话配置下发。重点tool_handler.py通常无需修改——它是完全配置驱动的动态执行器仅依据CLOUD_FUNCTIONS映射执行 HTTP 调用。七、故障排查指南7.1 连接问题服务端日志检查启动与运行期日志中的 WebSocket 连接错误或异常网络连通性确认 8081 端口或自定义端口未被防火墙或网络策略拦截客户端配置核对客户端应用中的 WebSocket URL 是否指向正确的服务端地址客户端代码位于 client 目录。7.2 API Key 错误Secret Manager 配置确认运行服务尤其 Cloud Run的服务账号具有Secret Manager Secret Accessor角色并在 Console 中确认GOOGLE_API_KEY、OPENWEATHER_API_KEY已创建且有值环境变量本地开发时确认.env变量名与值正确、且被服务端正常加载服务端日志查找密钥获取或认证失败的相关错误信息。7.3 工具执行错误函数 URL核对.env与config/config.py中的 URL 是否正确、函数是否已部署且可访问Cloud Function 日志工具实现内部的报错需在函数日志中排查权限确认 Cloud Function 使用的服务账号有权限访问其依赖的外部 API如 Calendar API、天气 API参数传递通过服务端与函数两侧日志追踪参数数据流确认参数格式与完整性。7.4 配额超限Quota Exceeded在 Cloud Console 监控 Gemini API、Cloud Functions 等服务的用量与配额关注服务端日志或客户端 UI 中的配额错误提示服务端已内置quota_exceeded类型的结构化错误消息与文本提示可配合重试机制或优雅的错误处理来缓解。八、总结Project Livewire 服务端是一个清晰展示WebSocket 长连接 Gemini Multimodal Live API 配置驱动工具路由三层架构的参考实现websocket_handler用双任务并发模型维持客户端与 Gemini 的双向流tool_handler让工具扩展从改代码退化为改配置而config.py的Secret Manager 优先、环境变量兜底策略则为生产环境密钥安全提供了范本。无论是想快速搭建一个可打断的实时语音助手还是研究 Live API 的函数调用与异步编排模式本文对应的 server 目录 都是一个可直接运行、易于拆解的起点。【免费下载链接】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），仅供参考 Project Livewire 服务端实战解析：用 Gemini Multimodal Live API 构建实时多模态 WebSocket 代理与工具调用
Project Livewire 服务端实战解析用 Gemini Multimodal Live API 构建实时多模态 WebSocket 代理与工具调用【免费下载链接】generative-aiSample code and notebooks for Generative AI on Google Cloud, with Gemini Enterprise Agent Platform项目地址: https://gitcode.com/GitHub_Trending/ge/generative-aiProject Livewire 是展示 Gemini Multimodal Live API 能力的实时多模态对话应用其服务端server 组件是一个基于 Pythonwebsockets与google-genai库构建的 WebSocket 应用承担着客户端与 Gemini API 之间的代理、会话管理和工具调用枢纽角色。本文以 server/README.md 为主线结合仓库内源码与部署配置完整讲解该服务端从环境准备、配置加载、本地运行到 Cloud Run 部署、工具扩展与故障排查的全过程读完后你可以独立搭建一个语音 文本 图像 工具调用的实时 AI 对话后端。一、服务端定位它到底做了什么该服务端是 Project Livewire 的核心后端逻辑围绕五大职责展开WebSocket 通信与前端 UI 建立并维护双向、实时的 WebSocket 长连接Gemini API 交互连接并管理 Gemini Multimodal Live API 会话处理用户输入文本、音频、视频并转发 AI 生成的响应工具处理与函数调用实现工具处理机制将 Gemini 发起的函数调用路由到对应的工具天气、日历等这些工具以独立的 Google Cloud Functions 实现通过 HTTP 请求安全调用会话管理为每个连接的客户端维护会话状态保证多轮交互的上下文连续配置管理从环境变量与 Google Cloud Secret Manager 中加载 API Key、Cloud Function URL 与服务器设置。从部署定位看该服务端面向 Google Cloud Run 设计以利用其弹性伸缩与托管能力同时支持本地运行以用于开发、测试与实验。二、架构与核心组件源码解读2.1 目录结构总览gemini/multimodal-live-api/project-livewire/server/ ├── core/ │ ├── gemini_client.py # Gemini API 客户端初始化与会话创建 │ ├── session.py # 客户端会话状态与活动会话管理 │ ├── tool_handler.py # 工具执行与 Cloud Function 路由 │ └── websocket_handler.py # WebSocket 连接处理与消息流转 ├── config/ │ ├── config.py # 配置加载与管理Secret Manager 环境变量 │ └── system-instructions.txt # Gemini 模型的系统指令 ├── Dockerfile # 服务端容器镜像定义 ├── cloudbuild.yaml # Cloud Build 部署到 Cloud Run 的配置 ├── requirements.txt # Python 依赖清单 ├── README.md # 本文档 └── server.py # 主入口启动 WebSocket 服务器2.2 入口文件 server.pyasyncio 事件循环与 WebSocket 服务server.py 是整个服务端的入口。它通过websockets.serve将handle_client挂载到0.0.0.0:8081并通过await asyncio.Future()让事件循环永久运行port 8081 async with websockets.serve( handle_client, 0.0.0.0, port, ping_interval30, ping_timeout10, ): logger.info(fRunning websocket server on 0.0.0.0:{port}...) await asyncio.Future() # run forever值得注意的两个细节心跳参数ping_interval30每 30 秒发送一次 WebSocket ping、ping_timeout1010 秒未收到 pong 判定连接失效这是 Cloud Run 空闲实例回收环境下保持长连接活跃的关键配置日志治理服务器启动时通过logging.getLogger(...).setLevel(logging.ERROR)将google、urllib3.connectionpool、websockets.client、httpx等第三方库的日志级别压到 ERROR保留应用自身的 DEBUG 信息避免海量 SDK 日志淹没调试输出。日志级别本身由环境变量LOG_LEVEL控制默认INFOlogging.basicConfig(levelgetattr(logging, os.getenv(LOG_LEVEL, INFO).upper()), ...)。2.3 core/gemini_client.py双通道客户端初始化core/gemini_client.py 的create_gemini_session()根据use_vertex标志选择两种认证与接入方式Vertex AI 模式GOOGLE_GENAI_USE_VERTEXAItrue使用 ADCApplication Default Credentials初始化需要GOOGLE_CLOUD_PROJECT区域默认us-central1client genai.Client(vertexaiTrue, locationlocation, projectproject_id)Dev API 模式默认使用 API Key 初始化http_options{api_version: v1alpha}client genai.Client(vertexaiFalse, http_options{api_version: v1alpha}, api_keyapi_config.api_key)最终通过client.aio.live.connect(modelMODEL, configCONFIG)建立 Gemini Live 会话aio为异步客户端。模型与语音在 config/config.py 中按接入模式分别配置接入模式模型语音Vertex AIgemini-2.0-flash-expMODEL_GOOGLE_GENAI_USE_VERTEXAIAoedeVOICE_GOOGLE_GENAI_USE_VERTEXAIDev APImodels/gemini-2.0-flash-expMODEL_DEV_APIPuckVOICE_DEV_API2.4 core/session.py会话状态管理core/session.py 用dataclass SessionState记录单个客户端的完整状态dataclass class SessionState: is_receiving_response: bool False interrupted: bool False current_tool_execution: Optional[asyncio.Task] None current_audio_stream: Optional[Any] None genai_session: Optional[Any] None received_model_response: bool False其中current_tool_execution保存正在执行的工具任务句柄用于中断时取消active_sessions是模块级字典Dict[str, SessionState]配合create_session/get_session/remove_session完成会话生命周期管理。会话 ID 由str(id(websocket))生成。2.5 core/websocket_handler.py双向消息流与工具队列core/websocket_handler.py 是整个服务端最核心的编排模块其关键机制如下。建立连接handle_client首先create_gemini_session()创建 Gemini 会话向客户端发送{ready: true}握手消息然后进入handle_messages。双任务并发模型handle_messages使用asyncio.TaskGroup同时运行两个协程——handle_client_messages客户端 → Gemini与handle_gemini_responsesGemini → 客户端。当任一任务因Quota exceeded或connection closed异常退出时服务端会向客户端发送{type: error, data: {...}}结构化错误消息并在finally中取消未完成的任务。客户端入站消息协议handle_client_messages消息类型行为audio以mime_type: audio/pcm发送给 Gemini并置end_of_turnTrueimage以mime_type: image/jpeg发送给 Geminitext作为纯文本发送置end_of_turnTrueend记录结束信号不做转发Gemini 出站响应处理handle_gemini_responsesprocess_server_content若响应含tool_call放入asyncio.Queue由后台process_tool_queue任务异步执行主循环继续处理其他响应工具执行不阻塞音频流server_content.model_turn.parts中的inline_data音频经base64.b64encode编码后以{type: audio, data: base64}下发part.text以{type: text, data: ...}下发收到server_content.interrupted时向客户端发送{type: interrupted}通知支持用户随时打断模型回答turn_complete时发送{type: turn_complete}并重置会话中的响应状态位。工具调用协议process_tool_queue对每个function_call先向客户端发送{type: function_call, data: {name: ..., args: ...}}用于 UI 反馈然后执行工具再发送{type: function_response, data: ...}最后通过types.LiveClientToolResponse(function_responses[...])将结果回传给 Gemini 会话。异常兜底handle_client对浏览器刷新导致的异常连接code 1006、空闲超时asyncio.TimeoutError与普通异常分别发送不同error_typeconnection_closed/timeout/general的错误消息finally中通过cleanup_session取消进行中的工具任务、关闭 Gemini 会话并从活动会话表移除记录。2.6 core/tool_handler.py动态工具路由core/tool_handler.py 的execute_tool(tool_name, params)是工具的通用执行器从CLOUD_FUNCTIONS字典取工具名对应的 Cloud Function URL用urlencode将参数拼为查询串通过aiohttp发起 GET 请求并解析 JSON 响应。其返回结构为Dict[str, Any]失败时返回{error: ...}以便 Gemini 理解错误。该执行器完全由配置驱动——新增工具时无需修改 tool_handler.py详见第六节。三、环境准备本地开发与 Cloud Run 部署的前置条件3.1 本地开发环境Python 3.11与pip依赖见 requirements.txt核心包括websockets14.1、google-genai1.3.0、aiohttp3.12.14、python-dotenv1.0.1、google-cloud-secret-manager2.19.0等。3.2 Google Cloud 项目准备Cloud Run 部署在 Google Cloud Console 创建项目并执行gcloud config set project YOUR_GOOGLE_CLOUD_PROJECT启用以下 APISecret Manager API、Cloud Run API、Cloud Build API、Cloud Functions API以及若使用 Vertex APIVertex AI API安装并初始化 gcloud CLIgcloud init与gcloud auth login。3.3 API Key 与密钥Gemini API Key从 Google AI Studio 获取生产环境推荐存入 Secret Managersecret 名GOOGLE_API_KEY本地开发可用环境变量OpenWeather API Key可选用于天气工具注册 OpenWeather 获取存入 Secret ManagerOPENWEATHER_API_KEY或环境变量。3.4 服务账号Service AccountCloud Run 部署创建服务账号并授予Secret Manager Secret Accessorroles/secretmanager.secretAccessor角色部署时将服务配置为使用该账号本地开发可选下载服务账号 JSON 密钥文件设置GOOGLE_APPLICATION_CREDENTIALS环境变量指向该文件即可通过 ADC 访问 Secret Manager。3.5 Cloud Functions工具实现工具函数作为独立的 Google Cloud Functions 部署详细步骤见 cloud-functions/README.md天气工具位于 weather-tools/get-weather-tool/main.py。部署后记录各函数的触发 URL用于服务端配置。四、安装与配置4.1 安装步骤git clone https://github.com/heiko-hotz/project-livewire.git cd project-livewire/server python3 -m venv venv source venv/bin/activate # Linux/macOSWindows 使用 venv\Scripts\activate pip install -r requirements.txt4.2 环境变量与 .env 文件在server/目录创建.env文件示例内容如下GOOGLE_CLOUD_PROJECTyour-gcp-project-id LOG_LEVELINFO # Cloud Function URLs替换为实际部署后的函数 URL WEATHER_FUNCTION_URLhttps://REGION-GOOGLE_CLOUD_PROJECT.cloudfunctions.net/get-weather-tool CALENDAR_FUNCTION_URLhttps://REGION-GOOGLE_CLOUD_PROJECT.cloudfunctions.net/get-calendar-tool # 可选本地开发时作为 Secret Manager 的兜底 API Key GOOGLE_API_KEYyour_gemini_api_key OPENWEATHER_API_KEYyour_openweather_api_key # 可选本地访问 Secret Manager 时显式指定服务账号密钥文件 # GOOGLE_APPLICATION_CREDENTIALSpath/to/service-account-key.json重要务必把.env加入.gitignore与.dockerignore防止敏感信息被提交。结合源码config/config.py 还支持以下扩展变量GOOGLE_GENAI_USE_VERTEXAItrue切换到 Vertex AI 接入模式GOOGLE_CLOUD_LOCATIONVertex AI 区域默认us-central1MODEL_DEV_API/MODEL_GOOGLE_GENAI_USE_VERTEXAI覆盖模型名VOICE_DEV_API/VOICE_GOOGLE_GENAI_USE_VERTEXAI覆盖语音Dev 默认PuckVertex 默认AoedeFORECAST_FUNCTION_URL、PAST_APPOINTMENTS_FUNCTION_URL天气预报与历史日程工具 URL。CLOUD_FUNCTIONS字典在模块加载时即对 URL 做校验缺失或非https://开头会输出警告日志CLOUD_FUNCTIONS { get_weather: os.getenv(WEATHER_FUNCTION_URL), get_weather_forecast: os.getenv(FORECAST_FUNCTION_URL), get_next_appointment: os.getenv(CALENDAR_FUNCTION_URL), get_past_appointments: os.getenv(PAST_APPOINTMENTS_FUNCTION_URL), }4.3 配置加载优先级服务端按以下顺序解析配置Google Cloud Secret Manager在 Cloud Run 环境或本地配置 ADC中可访问时优先使用读取GOOGLE_API_KEY与OPENWEATHER_API_KEY。其实现位于 config/config.py 的get_secret通过projects/{project_id}/secrets/{secret_id}/versions/latest拉取最新版本环境变量含 .envSecret Manager 不可达或未配置对应项时回退到环境变量。代码中先load_dotenv()加载.env再在get_secret失败时logger.warning(...)并回退到os.getenv(...)LOG_LEVEL始终从环境变量加载默认INFO。五、运行服务端本地开发与 Cloud Run 部署5.1 本地运行配置好.env后直接启动python server.py服务端将在0.0.0.0:8081监听 WebSocket 连接。5.2 部署到 Google Cloud Run使用仓库内 cloudbuild.yaml 一键构建、推送并部署gcloud builds submit --config server/cloudbuild.yaml该流水线依次执行构建镜像gcr.io/$GOOGLE_CLOUD_PROJECT/livewire-backend→ 推送到 Container Registry → 以服务账号livewire-backend${GOOGLE_CLOUD_PROJECT}.iam.gserviceaccount.com部署到us-central1的livewire-backend服务并通过--set-env-vars GOOGLE_CLOUD_PROJECT...,LOG_LEVELINFO注入运行环境变量。镜像定义见 Dockerfile基于python:3.11-slim暴露 8081 端口启动命令python server.py。获取服务 URLgcloud run services describe livewire-backend --platform managed --region us-central1 --format value(status.url) gcloud run services list # 查看所有服务以找到前端服务名 gcloud run services describe livewire-frontend --platform managed --region us-central1 --format value(status.url)5.3 用 wscat 测试部署的服务端安装 WebSocket 命令行测试工具npm install -g wscat连接并发送测试消息export CLOUD_RUN_URL$(gcloud run services describe livewire-backend --platform managed --region us-central1 --format value(status.url)) # macOS wscat -c $(echo $CLOUD_RUN_URL | sed s|https:|wss:|) # Linux wscat -c $(echo $CLOUD_RUN_URL | sed s/https:/wss:/) # 或手动指定wscat -c wss://YOUR_CLOUD_RUN_URL连接后发送{type: text, data: Hello, are you there?}服务端应返回来自 Gemini 模型的响应语音响应对应{type: audio, data: base64 音频}消息。六、扩展与修改工具四步流程向 Gemini 暴露新工具如新的天气或日历能力通常只需修改四处Cloud Function 实现在 cloud-functions 目录下实现/修改工具逻辑并部署记录其函数 URL服务端配置config/config.py在CLOUD_FUNCTIONS字典中增加条目将代码中的工具名映射到函数 URL在CONFIG[tools]的function_declarations中新增声明包含nameGemini 调用的函数名需与CLOUD_FUNCTIONS的键一致description工具作用描述帮助 Gemini 判断何时调用parameters函数接受的参数 schema含required列表这是 Gemini 正确生成调用参数的关键当前CONFIG还通过generation_config: {response_modalities: [AUDIO], speech_config: VOICE}将响应模态固定为音频并指定语音环境变量.env新增与配置中变量名一致的 URL 变量如YOUR_NEW_TOOL_URL系统指令system-instructions.txt告知模型新工具的存在、用途与使用规则。仓库当前示例指令为询问天气时必须使用 get_weather 工具并在运行时作为system_instruction随会话配置下发。重点tool_handler.py通常无需修改——它是完全配置驱动的动态执行器仅依据CLOUD_FUNCTIONS映射执行 HTTP 调用。七、故障排查指南7.1 连接问题服务端日志检查启动与运行期日志中的 WebSocket 连接错误或异常网络连通性确认 8081 端口或自定义端口未被防火墙或网络策略拦截客户端配置核对客户端应用中的 WebSocket URL 是否指向正确的服务端地址客户端代码位于 client 目录。7.2 API Key 错误Secret Manager 配置确认运行服务尤其 Cloud Run的服务账号具有Secret Manager Secret Accessor角色并在 Console 中确认GOOGLE_API_KEY、OPENWEATHER_API_KEY已创建且有值环境变量本地开发时确认.env变量名与值正确、且被服务端正常加载服务端日志查找密钥获取或认证失败的相关错误信息。7.3 工具执行错误函数 URL核对.env与config/config.py中的 URL 是否正确、函数是否已部署且可访问Cloud Function 日志工具实现内部的报错需在函数日志中排查权限确认 Cloud Function 使用的服务账号有权限访问其依赖的外部 API如 Calendar API、天气 API参数传递通过服务端与函数两侧日志追踪参数数据流确认参数格式与完整性。7.4 配额超限Quota Exceeded在 Cloud Console 监控 Gemini API、Cloud Functions 等服务的用量与配额关注服务端日志或客户端 UI 中的配额错误提示服务端已内置quota_exceeded类型的结构化错误消息与文本提示可配合重试机制或优雅的错误处理来缓解。八、总结Project Livewire 服务端是一个清晰展示WebSocket 长连接 Gemini Multimodal Live API 配置驱动工具路由三层架构的参考实现websocket_handler用双任务并发模型维持客户端与 Gemini 的双向流tool_handler让工具扩展从改代码退化为改配置而config.py的Secret Manager 优先、环境变量兜底策略则为生产环境密钥安全提供了范本。无论是想快速搭建一个可打断的实时语音助手还是研究 Live API 的函数调用与异步编排模式本文对应的 server 目录 都是一个可直接运行、易于拆解的起点。【免费下载链接】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），仅供参考