Gemini Live Telephony App 实战：用 Twilio + FastAPI + Gemini Live 构建云原生实时语音 AI 通话系统

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 生产部署的完整链路，读完你将掌握：如何用 TwiML<Connect><Stream>建立电话侧 WebSocket 媒体流、如何用流式 DSP 重采样桥接 8kHz µ-law 与 16/24kHz PCM 三种音频格式、如何用session_handle实现断线会话恢复，以及如何通过 Cloud Run 的一组关键参数把有状态、低延迟的通话服务稳定跑在无状态、无服务器的平台上。

一、项目定位与三大核心挑战

该示例应用提供了一个"实时、双向、语音到 AI"应用的完整架构蓝图与实现：Twilio 负责电话接入，FastAPI 负责实时处理与媒体流编排，Google Gemini Live API 负责会话式 AI。项目的设计文档 design_doc.md 明确指出，这套集成面临三个核心挑战：

1. 系统级集成：把 Twilio 的电话信令与媒体流，通过双向 WebSocket 与自定义后端对接；
2. 音频转码管线设计：在 Twilio（8kHz µ-law）、Gemini Live 输入（16kHz PCM）与输出（24kHz PCM）三种截然不同的音频格式之间，构建高保真、低延迟的实时转码流水线；
3. 平台化部署：把有状态、低延迟的实时通话服务，部署到无状态、无服务器的 Google Cloud Run 上，并规避平台固有的冷启动与状态管理问题。

其中最关键、也最不直观的设计决策是音频重采样环节的选型：必须使用流式（streaming）数字信号处理库，而非朴素的逐块（chunk-by-chunk）处理，否则会产生可闻的音频伪影。设计文档给出的最优解是python-samplerate（即 libsamplerate 的 Python 封装），其有状态的 "Full API" 正是为高质量、实时的分块音频处理而设计。

二、端到端架构：从 HTTP Webhook 到双向 WebSocket 流

整个系统的数据流可以概括为"一次 HTTP 握手，一条持久 WebSocket 双向流"：

1. 发起（HTTP）：用户拨打为服务预配的 Twilio 电话号码。
2. Webhook 触发：Twilio 收到来电后，向应用预定义的 HTTP 端点/twiml发送一个同步的 HTTP POST 请求。
3. TwiML 响应（HTTP）：部署在 Cloud Run 上的 FastAPI 服务收到请求，动态生成并返回一份 TwiML（Twilio Markup Language）文档。
4. WebSocket 连接（WSS）：TwiML 响应中包含<Connect><Stream>动词，Twilio 媒体服务器据此向应用的 WebSocket 端点/ws/twilio发起持久、安全的 WSS 连接。
5. 双向流式传输（WSS）：连接建立后，FastAPI 基于asyncio并发管理两条音频流——
   • 入站流（用户 → AI）：接收 Twilio 的 8kHz µ-law 音频，实时转码为 16kHz PCM，转发给 Gemini Live API；
   • 出站流（AI → 用户）：接收 Gemini Live API 的 24kHz PCM 音频，实时转码为 8kHz µ-law，回传给 Twilio。
6. 结束：用户挂断或连接关闭后，应用清理资源。

这条链路的关键切换点在于：TwiML 中的<Stream>动词让 Twilio 从"HTTP 请求-响应"模型平滑过渡到"持久 WebSocket 媒体流"模型，这正是电话 AI 应用的标准接驳方式。

三、FastAPI 后端与 WebSocket 编排

后端核心是 main.py，FastAPI 应用包含两个端点：

3.1/twiml：动态生成 TwiML

main.py 中，服务从环境变量SERVICE_URL读取部署地址（去掉协议前缀），拼出 WebSocket 地址后注入 TwiML：

@app.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 = f"""<Response><Connect><Stream url="wss://{service_url}/ws/twilio" /></Connect></Response>""" return Response(content=twiml, media_type="application/xml")

3.2/ws/twilio：三任务并发编排

main.py 中，WebSocket 端点接受连接后立即创建两个asyncio.Queue（in_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( vertexai=True, project=os.getenv("GOOGLE_CLOUD_PROJECT"), location=os.getenv("GOOGLE_CLOUD_LOCATION"), ) # client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])

默认走 Vertex AI 认证（需要GOOGLE_CLOUD_PROJECT与GOOGLE_CLOUD_LOCATION）；注释掉的那行给出了使用普通 Gemini API Key 的备选路径，两种接入方式可以根据你的部署环境切换。

四、音频转码管线：流式重采样与 µ-law 编解码

这是本项目的技术核心，全部实现集中在 utils/audio_transcoding.py。管线需要桥接三种音频格式：

4.1 入站：Twilio（8kHz µ-law）→ Gemini（16kHz PCM）

handle_twilio_to_gemini 的核心逻辑：

1. Base64 解码：Twilio 媒体消息中的 payload 是 base64 编码的 µ-law 音频；
2. µ-law → PCM：用audioop.ulaw2lin(chunk_ulaw, 2)转成 16-bit 线性 PCM；
3. PCM → Float32：np.frombuffer读取为int16数组，再除以32768.0归一化为float32；
4. 8kHz → 16kHz 升采样：调用resampler.process(arr, ratio=2.0, end_of_input=False)；
5. Float32 → Int16：乘以32767还原为int16字节流，放入in_q队列交给 Gemini 会话。

这里end_of_input=False是关键：它告诉重采样器"后续还有数据"，让内部滤波器状态（状态历史/延迟补偿）跨块延续，从而避免逐块独立处理导致的边界伪影。

4.2 出站：Gemini（24kHz PCM）→ Twilio（8kHz µ-law）

handle_gemini_to_twilio 是入站的逆过程：

1. 从out_q接收 Gemini 返回的 24kHz PCM 字节（asyncio.wait_for带 1 秒超时）；
2. 字节转float32数组并归一化；
3. 用ratio=(8000/24000)降采样到 8kHz；
4. 还原为int16PCM 后，用audioop.lin2ulaw(arr_8k.tobytes(), 2)编码为 µ-law；
5. base64 编码后，通过 WebSocket 以 Twilio 媒体消息格式回发（携带event: media、streamSid与payload）。

4.3 为什么必须用python-samplerate

两个方向的重采样器都在 main.py 中创建：

resampler_in = samplerate.Resampler("sinc_fastest", channels=1) resampler_out = samplerate.Resampler("sinc_fastest", channels=1)

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_instruction=types.Content( parts=[types.Part(text=BASE_SYSTEM_INSTRUCTION)] ), response_modalities=["AUDIO"], session_resumption=types.SessionResumptionConfig(handle=session_handle), speech_config=types.SpeechConfig( voice_config=types.VoiceConfig( prebuilt_voice_config=types.PrebuiltVoiceConfig( voice_name="Achird", ) ), language_code="en-US", ), realtime_input_config=types.RealtimeInputConfig( automatic_activity_detection=types.AutomaticActivityDetection( disabled=False, start_of_speech_sensitivity=types.StartSensitivity.START_SENSITIVITY_LOW, end_of_speech_sensitivity=types.EndSensitivity.END_SENSITIVITY_LOW, prefix_padding_ms=20, silence_duration_ms=150, ) ), )

各配置项要点：

• response_modalities=["AUDIO"]：只接受音频响应，电话场景不需要文本/工具输出；
• speech_config：选用预置语音Achird，语言en-US，决定 AI 的音色；
• realtime_input_config/ 自动活动检测（VAD）：start_of_speech_sensitivity与end_of_speech_sensitivity都设为LOW（低灵敏度，容忍更长的停顿，避免说话间隙被打断），prefix_padding_ms=20表示语音起始前保留 20ms 音频，silence_duration_ms=150表示静音 150ms 判定为一段语音结束；
• session_resumption：携带会话恢复句柄，实现断线续聊。

5.2 三个并发协程：发送、心跳与接收

会话建立后并发运行三个协程：

sender_loop（发送）：从in_q取 16kHz PCM 音频块，以audio/pcm;rate=16000的 Blob 通过session.send_realtime_input发给 Gemini；wait_for超时 0.01 秒实现高频率轮询，保持低延迟。

heartbeat_loop（心跳保活）：每 5 秒发送一段 10ms 的 16kHz 静音（320 字节 16-bit PCM：b"\x00" * 320），防止长静默期间流被服务端判定为无活动而断开。

主接收循环：迭代session.receive()，处理三类消息——

• session_resumption_update：捕获new_handle并保存为session_handle，用于后续重连时恢复完整对话上下文；
• server_content.model_turn：提取part.inline_data（即 24kHz PCM 音频），放入out_q交由出站转码任务回传 Twilio；
• turn_complete：单轮回复结束，保持会话存活。

5.3 自动重连与会话恢复

外层是一个while call_state.get("active")循环：任何一次连接异常（网络抖动、服务端断流）都会捕获后asyncio.sleep(2)重连；重连时把已保存的session_handle重新塞进LiveConnectConfig，Gemini 侧据此恢复完整对话上下文，用户几乎感知不到中断。这就是"基于句柄的会话恢复"——相比文件式历史，状态保存在 API 内部，后端无需持久化对话内容。

5.4 人设与系统指令

utils/prompt.py 集中存放BASE_SYSTEM_INSTRUCTION：示例中 AI 扮演 Northwestern Medicine 的护理团队成员 Sam，给患者 Vishnu 拨打术后随访电话。指令明确约束了人设（亲切、专业、有同理心）、对话风格（自然轮流对话）、目标（鼓励患者管理健康、安排后续预约）以及边界（不提供诊断、不施压、不虚构预约），并给定了开场白。实际接入时，替换这里的提示词即可快速定制任意行业的"外呼关怀助手"人设。

六、状态管理：内存态 + 会话句柄的混合模型

针对"无状态平台跑有状态通话"的矛盾，本项目采用混合状态模型：

• 内存态（in-memory）：call_state字典维护单实例内的实时通话状态（active标志、stream_sid），配合 Cloud Run 的session affinity保证同一通话的请求始终落在同一实例；
• 会话恢复句柄：利用 Gemini 的session_handle属性——从会话更新消息中捕获 token，重连时通过SessionResumptionConfig恢复 API 内部的完整上下文，避免在本地维护历史文件。

设计文档进一步建议：生产环境应引入 Google Memorystore（Redis）将对话状态外部化，从而支持水平扩展与更高的韧性。这是从示例走向生产的关键演进路径。

七、本地快速开始：从零打通一通测试电话

本地联调流程（README 中的 Quickstart 部分）按以下步骤执行，环境要求Python 3.12。

7.1 项目与依赖

python3.12 -m venv venv source venv/bin/activate pip install -r requirements.txt

requirements.txt 中核心依赖及用途：

7.2 云环境与认证

1. 在 Google Cloud Console 创建项目并启用Vertex AI API；
2. 本地终端认证：

gcloud auth login gcloud auth application-default login # 为应用和代码提供凭据

3. 在项目目录下参照.env.example（README 中的说明）创建.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 做隧道：

1. 从 ngrok 官网下载 Linux 版本并解压，移动到 PATH 目录（如/usr/local/bin）；
2. 注册并配置 authtoken：

ngrok config add-authtoken $YOUR_AUTHTOKEN ngrok --version # 验证安装

3. 新开一个终端，把本地 8000 端口暴露到公网：

ngrok http 8000

ngrok 会输出一个公网Forwarding URL，形如https://<random-string>.ngrok-free.dev。

7.4 配置 SERVICE_URL 并启动应用

关键点：应用用SERVICE_URL告诉 Twilio 去哪里建立 WebSocket 连接，因此必须把它设置为 ngrok 转发地址（写入.env）：

SERVICE_URL=https://<random-string>.ngrok-free.dev

更新.env后再启动 FastAPI 应用：

uvicorn main:app --host 0.0.0.0 --port 8000

7.5 配置 Twilio 并拨打电话

1. 注册 Twilio 试用账号，获得一个试用电话号码与免费额度；
2. 在 Twilio Console 的电话号码配置中，把"A CALL COMES IN"Webhook 指向https://<random-string>.ngrok-free.dev/twiml，方法设为 HTTP POST；
3. 等待约 2 分钟让配置生效，然后拨打你的 Twilio 号码。

Twilio 会把 Webhook 发给公网 ngrok 地址，再转发到本地应用——此时你可以在本地终端实时观察完整链路的日志，进行端到端调试。

八、部署到 Google Cloud Run

8.1 自动化部署脚本

deploy.sh 自动化了"构建镜像 → 推送 GCR → 部署 Cloud Run → 回填 SERVICE_URL"的完整流程：

1. 修改脚本：把PROJECT_ID=[YOUR_PROJECT_ID]替换为你的 GCP 项目 ID（脚本使用gcr.io仓库，并以时间戳生成唯一镜像 tag，强制拉取新镜像）；
2. 配置 Docker 与 gcloud（可选）：

gcloud auth configure-docker

3. 执行部署：

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 关键参数（低延迟的有状态通话）

部署脚本中的核心参数及其设计意图如下：

其中--concurrency=1是对这类有状态通话服务最关键的设置：每个实例专职服务一路通话，从根本上规避多路并发对内存态call_state与重采样器实例的共享问题。

8.3 部署后的 Twilio 配置

部署脚本成功后会输出Service URL（形如https://<your-service-url>.a.run.app），随后：

1. 复制该 URL；
2. 在 Twilio Console 电话号码配置中，把"A CALL COMES IN"Webhook 设为https://<your-service-url>.a.run.app/twiml，方法设为HTTP POST；
3. 保存配置，你的 AI Care Assistant 即可接听真实来电。

8.4 IAM 权限与容器化注意事项

• IAM：部署前需确保账号具备 Cloud Run 与 Cloud Build 服务账号所需权限（如Cloud Run Invoker等角色）；
• 容器镜像：Dockerfile 基于python:3.12-slim，系统层必须安装libsamplerate0（samplerate库的运行时依赖），并安装了build-essential、cmake、git用于编译安装 Python 包；启动命令用gunicorn --bind :$PORT --workers 1 --worker-class uvicorn.workers.UvicornWorker --threads 8 main:app，绑定 Cloud Run 注入的$PORT环境变量。

九、生产化演进路径

从示例到生产，设计文档与代码共同指向以下几个方向：

1. 状态外部化：用 Google Memorystore（Redis）替代单实例内存态，支撑水平扩展与故障转移；
2. 会话句柄持久化：把session_handle存入 Redis，实现跨实例/跨重启的会话恢复；
3. 横向扩容与配额：concurrency=1意味着实例数 = 并发通话数，生产上需配合 Cloud Run 的实例上限与扩容策略规划容量；
4. 多路通话的流控：对 Twilio 媒体消息、Gemini 返回块做背压与超时管理（代码中已通过asyncio.Queue与wait_for做了基础处理）；
5. 人设与话术沉淀：把 utils/prompt.py 中的系统指令按业务场景参数化。

十、总结

gemini-live-telephony-app用约两百行核心代码演示了一条完整的"电话 ↔ AI"实时语音链路：TwiML 握手建立 WSS 媒体流、python-samplerate有状态重采样桥接三种音频格式、session_handle实现断线续聊、一套精心调校的 Cloud Run 参数（min-instances=1、session-affinity、concurrency=1、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