Gemini Live Telephony App 实战用 Twilio FastAPI Gemini Live 构建云原生实时语音 AI 通话系统【免费下载链接】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-live-telephony-app 示例应用为核心完整拆解如何将 Twilio 电话网络、FastAPI 后端与 Google Gemini Live API 三者打通构建一套端到端的实时双向语音 AI 应用。全文覆盖从架构设计、音频转码管线、会话管理到本地联调与 Cloud Run 生产部署的完整链路读完你将掌握如何用 TwiMLConnectStream建立电话侧 WebSocket 媒体流、如何用流式 DSP 重采样桥接 8kHz µ-law 与 16/24kHz PCM 三种音频格式、如何用session_handle实现断线会话恢复以及如何通过 Cloud Run 的一组关键参数把有状态、低延迟的通话服务稳定跑在无状态、无服务器的平台上。一、项目定位与三大核心挑战该示例应用提供了一个实时、双向、语音到 AI应用的完整架构蓝图与实现Twilio 负责电话接入FastAPI 负责实时处理与媒体流编排Google Gemini Live API 负责会话式 AI。项目的设计文档 design_doc.md 明确指出这套集成面临三个核心挑战系统级集成把 Twilio 的电话信令与媒体流通过双向 WebSocket 与自定义后端对接音频转码管线设计在 Twilio8kHz µ-law、Gemini Live 输入16kHz PCM与输出24kHz PCM三种截然不同的音频格式之间构建高保真、低延迟的实时转码流水线平台化部署把有状态、低延迟的实时通话服务部署到无状态、无服务器的 Google Cloud Run 上并规避平台固有的冷启动与状态管理问题。其中最关键、也最不直观的设计决策是音频重采样环节的选型必须使用流式streaming数字信号处理库而非朴素的逐块chunk-by-chunk处理否则会产生可闻的音频伪影。设计文档给出的最优解是python-samplerate即 libsamplerate 的 Python 封装其有状态的 Full API 正是为高质量、实时的分块音频处理而设计。二、端到端架构从 HTTP Webhook 到双向 WebSocket 流整个系统的数据流可以概括为一次 HTTP 握手一条持久 WebSocket 双向流发起HTTP用户拨打为服务预配的 Twilio 电话号码。Webhook 触发Twilio 收到来电后向应用预定义的 HTTP 端点/twiml发送一个同步的 HTTP POST 请求。TwiML 响应HTTP部署在 Cloud Run 上的 FastAPI 服务收到请求动态生成并返回一份 TwiMLTwilio Markup Language文档。WebSocket 连接WSSTwiML 响应中包含ConnectStream动词Twilio 媒体服务器据此向应用的 WebSocket 端点/ws/twilio发起持久、安全的 WSS 连接。双向流式传输WSS连接建立后FastAPI 基于asyncio并发管理两条音频流——入站流用户 → AI接收 Twilio 的 8kHz µ-law 音频实时转码为 16kHz PCM转发给 Gemini Live API出站流AI → 用户接收 Gemini Live API 的 24kHz PCM 音频实时转码为 8kHz µ-law回传给 Twilio。结束用户挂断或连接关闭后应用清理资源。这条链路的关键切换点在于TwiML 中的Stream动词让 Twilio 从HTTP 请求-响应模型平滑过渡到持久 WebSocket 媒体流模型这正是电话 AI 应用的标准接驳方式。三、FastAPI 后端与 WebSocket 编排后端核心是 main.pyFastAPI 应用包含两个端点端点方法职责/twimlPOST接收 Twilio 初始 Webhook动态生成包含ConnectStream的 TwiML返回application/xml/ws/twilioWebSocket双向音频流主端点编排 Twilio 与 Gemini Live 之间的音频流动3.1/twiml动态生成 TwiMLmain.py 中服务从环境变量SERVICE_URL读取部署地址去掉协议前缀拼出 WebSocket 地址后注入 TwiMLapp.post(/twiml) async def get_twiml(): Generates TwiML response to initiate a WebSocket stream with Twilio. service_url ( os.getenv(SERVICE_URL).replace(https://, ).replace(http://, ) ) twiml fResponseConnectStream urlwss://{service_url}/ws/twilio //Connect/Response return Response(contenttwiml, media_typeapplication/xml)3.2/ws/twilio三任务并发编排main.py 中WebSocket 端点接受连接后立即创建两个asyncio.Queuein_q、out_q作为音频块的传递通道并预创建两个流式重采样器实例随后用asyncio.create_task启动三个并发任务tasks [ asyncio.create_task( handle_twilio_to_gemini(websocket, in_q, resampler_in, call_state) ), asyncio.create_task( handle_gemini_to_twilio(websocket, out_q, resampler_out, call_state) ), asyncio.create_task( run_gemini_session(client, MODEL_ID, in_q, out_q, call_state) ), ]handle_twilio_to_gemini处理 Twilio → Gemini 的入站音频handle_gemini_to_twilio处理 Gemini → Twilio 的出站音频run_gemini_session在 utils/live_api.py 中管理 Gemini Live 高层会话。call_state字典用于在单实例内维护通话的实时状态活动标志、stream SID。任务结束后统一取消并复位状态完成资源清理。3.3 Gemini 客户端初始化main.py 使用google-genaiSDK 初始化客户端默认模型为gemini-live-2.5-flash-native-audio可通过环境变量GOOGLE_GENAI_MODEL覆盖MODEL_ID os.getenv(GOOGLE_GENAI_MODEL, gemini-live-2.5-flash-native-audio) client genai.Client( vertexaiTrue, projectos.getenv(GOOGLE_CLOUD_PROJECT), locationos.getenv(GOOGLE_CLOUD_LOCATION), ) # client genai.Client(api_keyos.environ[GEMINI_API_KEY])默认走 Vertex AI 认证需要GOOGLE_CLOUD_PROJECT与GOOGLE_CLOUD_LOCATION注释掉的那行给出了使用普通 Gemini API Key 的备选路径两种接入方式可以根据你的部署环境切换。四、音频转码管线流式重采样与 µ-law 编解码这是本项目的技术核心全部实现集中在 utils/audio_transcoding.py。管线需要桥接三种音频格式阶段采样率编码格式Twilio 上行8kHzµ-lawG.711Gemini Live 输入16kHz16-bit PCMGemini Live 输出24kHz16-bit PCM4.1 入站Twilio8kHz µ-law→ Gemini16kHz PCMhandle_twilio_to_gemini 的核心逻辑Base64 解码Twilio 媒体消息中的 payload 是 base64 编码的 µ-law 音频µ-law → PCM用audioop.ulaw2lin(chunk_ulaw, 2)转成 16-bit 线性 PCMPCM → Float32np.frombuffer读取为int16数组再除以32768.0归一化为float328kHz → 16kHz 升采样调用resampler.process(arr, ratio2.0, end_of_inputFalse)Float32 → Int16乘以32767还原为int16字节流放入in_q队列交给 Gemini 会话。这里end_of_inputFalse是关键它告诉重采样器后续还有数据让内部滤波器状态状态历史/延迟补偿跨块延续从而避免逐块独立处理导致的边界伪影。4.2 出站Gemini24kHz PCM→ Twilio8kHz µ-lawhandle_gemini_to_twilio 是入站的逆过程从out_q接收 Gemini 返回的 24kHz PCM 字节asyncio.wait_for带 1 秒超时字节转float32数组并归一化用ratio(8000/24000)降采样到 8kHz还原为int16PCM 后用audioop.lin2ulaw(arr_8k.tobytes(), 2)编码为 µ-lawbase64 编码后通过 WebSocket 以 Twilio 媒体消息格式回发携带event: media、streamSid与payload。4.3 为什么必须用python-samplerate两个方向的重采样器都在 main.py 中创建resampler_in samplerate.Resampler(sinc_fastest, channels1) resampler_out samplerate.Resampler(sinc_fastest, channels1)samplerate.Resampler对应 libsamplerate 的Full API——它是有状态的每次process()都会携带上一次调用的残余状态适合把任意长度的音频块喂给重采样器而不产生接缝噪声。sinc_fastest在保真度与计算开销之间取得平衡对实时通话场景是合理选择。设计文档特别强调naive 的逐块独立重采样chunk-by-chunk无状态会引入可闻的音频伪影这正是本项目选型该库的根本原因。五、Gemini Live 会话持久连接、VAD 与会话恢复会话编排在 utils/live_api.py 的run_gemini_session中完成它负责与 Gemini Live API 维持整通电话的持久连接。5.1 LiveConnectConfig一次配齐指令、语音与 VAD每次连接都构造一份LiveConnectConfig设计文档中提到的live_api_config.py配置集中化思路在当前仓库中以内联方式体现在run_gemini_session内config types.LiveConnectConfig( system_instructiontypes.Content( parts[types.Part(textBASE_SYSTEM_INSTRUCTION)] ), response_modalities[AUDIO], session_resumptiontypes.SessionResumptionConfig(handlesession_handle), speech_configtypes.SpeechConfig( voice_configtypes.VoiceConfig( prebuilt_voice_configtypes.PrebuiltVoiceConfig( voice_nameAchird, ) ), language_codeen-US, ), realtime_input_configtypes.RealtimeInputConfig( automatic_activity_detectiontypes.AutomaticActivityDetection( disabledFalse, start_of_speech_sensitivitytypes.StartSensitivity.START_SENSITIVITY_LOW, end_of_speech_sensitivitytypes.EndSensitivity.END_SENSITIVITY_LOW, prefix_padding_ms20, silence_duration_ms150, ) ), )各配置项要点response_modalities[AUDIO]只接受音频响应电话场景不需要文本/工具输出speech_config选用预置语音Achird语言en-US决定 AI 的音色realtime_input_config/ 自动活动检测VADstart_of_speech_sensitivity与end_of_speech_sensitivity都设为LOW低灵敏度容忍更长的停顿避免说话间隙被打断prefix_padding_ms20表示语音起始前保留 20ms 音频silence_duration_ms150表示静音 150ms 判定为一段语音结束session_resumption携带会话恢复句柄实现断线续聊。5.2 三个并发协程发送、心跳与接收会话建立后并发运行三个协程sender_loop发送从in_q取 16kHz PCM 音频块以audio/pcm;rate16000的 Blob 通过session.send_realtime_input发给 Geminiwait_for超时 0.01 秒实现高频率轮询保持低延迟。heartbeat_loop心跳保活每 5 秒发送一段 10ms 的 16kHz 静音320 字节 16-bit PCMb\x00 * 320防止长静默期间流被服务端判定为无活动而断开。主接收循环迭代session.receive()处理三类消息——session_resumption_update捕获new_handle并保存为session_handle用于后续重连时恢复完整对话上下文server_content.model_turn提取part.inline_data即 24kHz PCM 音频放入out_q交由出站转码任务回传 Twilioturn_complete单轮回复结束保持会话存活。5.3 自动重连与会话恢复外层是一个while call_state.get(active)循环任何一次连接异常网络抖动、服务端断流都会捕获后asyncio.sleep(2)重连重连时把已保存的session_handle重新塞进LiveConnectConfigGemini 侧据此恢复完整对话上下文用户几乎感知不到中断。这就是基于句柄的会话恢复——相比文件式历史状态保存在 API 内部后端无需持久化对话内容。5.4 人设与系统指令utils/prompt.py 集中存放BASE_SYSTEM_INSTRUCTION示例中 AI 扮演 Northwestern Medicine 的护理团队成员 Sam给患者 Vishnu 拨打术后随访电话。指令明确约束了人设亲切、专业、有同理心、对话风格自然轮流对话、目标鼓励患者管理健康、安排后续预约以及边界不提供诊断、不施压、不虚构预约并给定了开场白。实际接入时替换这里的提示词即可快速定制任意行业的外呼关怀助手人设。六、状态管理内存态 会话句柄的混合模型针对无状态平台跑有状态通话的矛盾本项目采用混合状态模型内存态in-memorycall_state字典维护单实例内的实时通话状态active标志、stream_sid配合 Cloud Run 的session affinity保证同一通话的请求始终落在同一实例会话恢复句柄利用 Gemini 的session_handle属性——从会话更新消息中捕获 token重连时通过SessionResumptionConfig恢复 API 内部的完整上下文避免在本地维护历史文件。设计文档进一步建议生产环境应引入 Google MemorystoreRedis将对话状态外部化从而支持水平扩展与更高的韧性。这是从示例走向生产的关键演进路径。七、本地快速开始从零打通一通测试电话本地联调流程README 中的 Quickstart 部分按以下步骤执行环境要求Python 3.12。7.1 项目与依赖python3.12 -m venv venv source venv/bin/activate pip install -r requirements.txtrequirements.txt 中核心依赖及用途依赖版本用途fastapi0.111.0Web 服务框架uvicorn[standard]0.30.1ASGI 开发服务器gunicorn23.0.0生产 WSGI 服务器托管 Uvicorn workergoogle-genai1.28.0Gemini Live SDKtwilio9.8.6Twilio 辅助库numpy2.3.4音频数值运算samplerate0.2.2高质量流式音频重采样7.2 云环境与认证在 Google Cloud Console 创建项目并启用Vertex AI API本地终端认证gcloud auth login gcloud auth application-default login # 为应用和代码提供凭据在项目目录下参照.env.exampleREADME 中的说明创建.env文件。从 main.py 可以看到实际需要的变量GOOGLE_CLOUD_PROJECT、GOOGLE_CLOUD_LOCATION、GOOGLE_GENAI_MODEL可选默认gemini-live-2.5-flash-native-audio、SERVICE_URL。7.3 用 ngrok 暴露本地服务电话媒体流需要公网可达的 WebSocket 端点本地开发用 ngrok 做隧道从 ngrok 官网下载 Linux 版本并解压移动到 PATH 目录如/usr/local/bin注册并配置 authtokenngrok config add-authtoken $YOUR_AUTHTOKEN ngrok --version # 验证安装新开一个终端把本地 8000 端口暴露到公网ngrok http 8000ngrok 会输出一个公网Forwarding URL形如https://random-string.ngrok-free.dev。7.4 配置 SERVICE_URL 并启动应用关键点应用用SERVICE_URL告诉 Twilio 去哪里建立 WebSocket 连接因此必须把它设置为 ngrok 转发地址写入.envSERVICE_URLhttps://random-string.ngrok-free.dev更新.env后再启动 FastAPI 应用uvicorn main:app --host 0.0.0.0 --port 80007.5 配置 Twilio 并拨打电话注册 Twilio 试用账号获得一个试用电话号码与免费额度在 Twilio Console 的电话号码配置中把A CALL COMES INWebhook 指向https://random-string.ngrok-free.dev/twiml方法设为 HTTP POST等待约 2 分钟让配置生效然后拨打你的 Twilio 号码。Twilio 会把 Webhook 发给公网 ngrok 地址再转发到本地应用——此时你可以在本地终端实时观察完整链路的日志进行端到端调试。八、部署到 Google Cloud Run8.1 自动化部署脚本deploy.sh 自动化了构建镜像 → 推送 GCR → 部署 Cloud Run → 回填 SERVICE_URL的完整流程修改脚本把PROJECT_ID[YOUR_PROJECT_ID]替换为你的 GCP 项目 ID脚本使用gcr.io仓库并以时间戳生成唯一镜像 tag强制拉取新镜像配置 Docker 与 gcloud可选gcloud auth configure-docker执行部署bash deploy.sh脚本用gcloud builds submit --tag ${IMAGE_NAME} --no-cache构建推送镜像随后执行gcloud run deploy服务名gemini-live-health区域us-central1--allow-unauthenticated部署完成后再用gcloud run services update把服务自己的公网 URL 写回SERVICE_URL环境变量——这样/twiml生成的 WebSocket 地址就能自动指向正确的部署地址。8.2 Cloud Run 关键参数低延迟的有状态通话部署脚本中的核心参数及其设计意图如下参数值设计意图--min-instances11常驻一个实例避免冷启动——电话随时可能打进绝不能等实例拉起--timeout36003600s1 小时超时适配长时间存活的 WebSocket 通话连接--memory2Gi2Gi为 CPU 密集的音频重采样预留足够内存--cpu22双核保障重采样与并发流的算力--session-affinity开启同一客户端Twilio的请求固定路由到同一实例维持 WebSocket 与内存 DSP 状态--concurrency11每个实例同一时刻只处理一路通话避免单实例内多路有状态通话互相干扰--no-cpu-throttling开启禁用 CPU 节流让实例始终可访问完整分配的 CPU保证实时音频处理的低延迟其中--concurrency1是对这类有状态通话服务最关键的设置每个实例专职服务一路通话从根本上规避多路并发对内存态call_state与重采样器实例的共享问题。8.3 部署后的 Twilio 配置部署脚本成功后会输出Service URL形如https://your-service-url.a.run.app随后复制该 URL在 Twilio Console 电话号码配置中把A CALL COMES INWebhook 设为https://your-service-url.a.run.app/twiml方法设为HTTP POST保存配置你的 AI Care Assistant 即可接听真实来电。8.4 IAM 权限与容器化注意事项IAM部署前需确保账号具备 Cloud Run 与 Cloud Build 服务账号所需权限如Cloud Run Invoker等角色容器镜像Dockerfile 基于python:3.12-slim系统层必须安装libsamplerate0samplerate库的运行时依赖并安装了build-essential、cmake、git用于编译安装 Python 包启动命令用gunicorn --bind :$PORT --workers 1 --worker-class uvicorn.workers.UvicornWorker --threads 8 main:app绑定 Cloud Run 注入的$PORT环境变量。九、生产化演进路径从示例到生产设计文档与代码共同指向以下几个方向状态外部化用 Google MemorystoreRedis替代单实例内存态支撑水平扩展与故障转移会话句柄持久化把session_handle存入 Redis实现跨实例/跨重启的会话恢复横向扩容与配额concurrency1意味着实例数 并发通话数生产上需配合 Cloud Run 的实例上限与扩容策略规划容量多路通话的流控对 Twilio 媒体消息、Gemini 返回块做背压与超时管理代码中已通过asyncio.Queue与wait_for做了基础处理人设与话术沉淀把 utils/prompt.py 中的系统指令按业务场景参数化。十、总结gemini-live-telephony-app用约两百行核心代码演示了一条完整的电话 ↔ AI实时语音链路TwiML 握手建立 WSS 媒体流、python-samplerate有状态重采样桥接三种音频格式、session_handle实现断线续聊、一套精心调校的 Cloud Run 参数min-instances1、session-affinity、concurrency1、no-cpu-throttling让有状态通话跑稳在无状态平台上。无论你要构建的是医疗随访助手、客服外呼、订餐机器人还是任何电话场景的生成式 AI 应用本文的架构蓝图、转码管线与部署配置都可作为可直接落地的起点。深入细节可继续阅读 design_doc.md 与上述各源码文件。【免费下载链接】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），仅供参考