开篇语音聊天到底难在哪“对着手机说一句对方秒回”听起来简单背后却是一条超长链路麦克风采集 → 前端编码 → 网络传输 → 云端 ASR → LLM 推理 → TTS → 音频回传 → 播放器渲染。任何一环掉链子用户就会吐槽“卡顿、延迟、机器人味儿”。用 com.google.genai 做语音聊天核心挑战可以浓缩成三点音频流是“胖数据”64 kbit/s 的 Opus 单路 20 ms 帧每秒就要 50 个包网络抖一下就直接“炸麦”。全双工场景下ASR、LLM、TTS 三个模型要并行跑还要共享上下文资源调度比传统 HTTP 请求复杂一个量级。Google 的 API 默认走全球负载均衡第一次 TLS 握手 OAuth 刷新就可能 300 ms不加优化直接输在起跑线。下面把我踩过的坑、调优脚本、线上配置一条线捋清让你少熬两周夜。接入方式选哪家gRPC vs REST vs WebSocketcom.google.genai 对外暴露三套端口协议优点缺点适用场景gRPC (HTTP/2)自带流式、多路复用、官方 Python/Java SDK 原生支持端口 443 需允许 HTTP/2部分老旧代理会降级低延迟双向流生产首选REST (JSON)调试简单curl 一把梭无服务器推送只能轮询延迟500 ms后台离线批处理WebSocket浏览器直接开麦需要自己做帧同步、重连、指数退避H5 网页 Demo结论终端到服务器走 gRPC 流式纯后台任务例如把 1 万小时录音批量转文字用 RESTWebSocket 留给快速原型上线前务必迁到 gRPC。环境配置 3 步走开通服务Cloud Console → Vertex AI → “Generative AI” → 勾选 “Speech/LLM/TTS” API记下 Project ID。建服务账号IAM Admin → Service Accounts → 新建genai-voice-chat→ 角色Vertex AI User→ 下载 JSON。装 SDK# Python 3.10 虚拟环境 pip install google-cloud-aiplatform1.38.0 google-genai0.3.0把刚才的 JSON 路径写进环境变量后面代码会自动卷export GOOGLE_APPLICATION_CREDENTIALS/secure/genai-voice-chat.json音频流处理Python 示例带注释下面这段代码演示“麦克风 → 实时 ASR → LLM → TTS → 扬声器”全双工回路单线程异步方便你插到 asyncio 框架里。关键逻辑用pyaudio以 20 ms 帧喂给 gRPCASR 返回is_final后触发 LLMLLM 每输出一个 sentence 就调用 TTSTTS 返回的音频流直接塞进pyaudio输出缓冲区全程 ring-buffer 缓存网络抖动时自动补包。import asyncio, pyaudio, logging, time from google.api_core import retry from google.genai import speech, llm, tts FORMAT pyaudio.paInt16 CHANNELS 1 RATE 16000 CHUNK 320 # 20 ms class VoiceChat: def __init__(self): self.speech_client speech.SpeechClient() self.llm_client llm.LLMClient() self.tts_client tts.TextToSpeechClient() self.audio pyaudio.PyAudio() self.in_stream self.audio.open(formatFORMAT, channelsCHANNELS, rateRATE, inputTrue, frames_per_bufferCHUNK) self.out_stream self.audio.open(formatFORMAT, channelsCHANNELS, rateRATE, outputTrue, frames_per_bufferCHUNK) async def listen(self): Producer把麦克风帧推给 ASR 流 config speech.RecognitionConfig( encodingspeech.RecognitionConfig.AudioEncoding.LINEAR16, sample_rate_hertzRATE, language_codeen-US, enable_automatic_punctuationTrue, ) streaming_config speech.StreamingRecognitionConfig( configconfig, interim_resultsTrue ) # 双向流 requests self.audio_request_generator() responses self.speech_client.streaming_recognize( requests, timeout300 ) async for response in responses: for result in response.results: if result.is_final: transcript result.alternatives[0].transcript logging.info(ASR: %s, transcript) # 直接调度 LLM不阻塞 asyncio.create_task(self.think_and_speak(transcript)) async def audio_request_generator(self): 异步生成器yield 音频帧 while True: data await asyncio.to_thread(self.in_stream.read, CHUNK) yield speech.StreamingRecognizeRequest(audio_contentdata) async def think_and_speak(self, transcript): ConsumerLLM TTS prompt fUser: {transcript}\nAssistant: # 流式 LLM返回 sentence 级切片 llm_stream self.llm_client.predict_stream( modelgemini-pro, promptprompt, max_tokens150 ) assistant_text async for piece in llm_stream: assistant_text piece # 简单断句遇到句号就发 TTS if piece.endswith((., !, ?)): await self.speak(assistant_text) assistant_text retry.Retry(predicateretry.if_transient_error) async def speak(self, text): TTS 并播放 tts_resp await asyncio.to_thread( self.tts_client.synthesize_speech, inputtts.SynthesisInput(texttext), voicetts.VoiceSelectionParams( language_codeen-US, nameen-US-Wavenet-D ), audio_configtts.AudioConfig( audio_encodingtts.AudioEncoding.LINEAR16 ), ) # 直接写扬声器 await asyncio.to_thread(self.out_stream.write, tts_resp.audio_content) if __name__ __main__: logging.basicConfig(levellogging.INFO) vc VoiceChat() asyncio.run(vc.listen())时间复杂度ASR 流式识别为 O(n) 帧级增量n 为音频采样点数LLM 生成 O(m) tokenm150 时平均延迟 180 msTTS 实时因子 0 RTF 0.3整体链路延迟 P95 可压到 600 ms见下节指标。Java 双工示例Android 端Android 官方 Sample 已经封装好AudioRecord gRPC这里只贴关键片段// proto 双向流 StreamObserverStreamingRecognizeRequest requestObserver speechStub.streamingRecognize(new StreamObserver() { Override public void onNext(StreamingRecognizeResponse resp) { if (resp.getResultsCount() 0 resp.getResults(0).getIsFinal()) { String txt resp.getResults(0) .getAlternatives(0) .getTranscript(); // 切换到 UI 线程 runOnUiThread(() - sendToLLM(txt)); } } ... }); // 麦克风循环 while (recording) { short[] buf new short[320]; audioRecord.read(buf, 0, 320); requestObserver.onNext( StreamingRecognizeRequest.newBuilder() .setAudioContent(ByteString.copyFrom(short2bytes(buf))) .build()); }要点用audioRecord.getTimestamp打 WallClock方便后端做漂移校准gRPC 通道加keepAliveWithoutCallstrue()防止 NAT 超时断流。性能指标与优化清单延迟拆解实测 Pixel 6 Wi-Fi 6美国西海岸 endpoint麦克风采集 前端编码20 ms网络 RTT40 msASR 首帧响应120 msLLM 首 token80 msTTS 首包100 ms播放器缓冲60 ms合计 420 msP95 600 ms达到“准实时”门槛。并发模型单核 Gemini Pro 可支撑 120 QPSQuery Per Second若每通对话平均 7 轮则 1 vCPU ≈ 17 路并发。生产建议K8s HPA 按 CPU 70% 扩容把 ASR/TTS 与 LLM 拆成独立 Pod避免互相挤占。省流技巧启用voice_activity_detection静音段直接丢包省 30% 流量TTS 选MP3_64K比LINEAR16小 4 倍解码 CPU 增加 5%移动端更划算gRPC 打开gzip压缩文本 payload 可再降 60%。生产环境注意事项认证管理把服务账号 JSON 挂进 K8s Secret不要打包进镜像每 12 小时调用auth.refresh()防止 401 风暴。错误重试gRPC 状态码映射UNAVAILABLE/DEADLINE_EXCEEDED→ 指数退避最大 3 次RESOURCE_EXHAUSTED→ 立刻限流等待配额窗口INVALID_ARGUMENT→ 直接抛给客户端避免死循环。监控用 OpenTelemetry 把asr_latency/llm_latency/tts_latency打成 HistogramP991 s 就 paging音频层再挂packet_loss、jitter一眼定位是网络还是模型。隐私合规欧盟用户先过 GDPR录音落盘前调用speech_client.delete_recognizer()清除临时日志提供“一键遗忘”接口LLM 上下文 24 h 后强制淘汰。扩展思考题多语言混说场景下如何动态切换 ASRlanguage_code而不重启流如果用户网络掉到 3G音频码率自适应降到 24 kbit/sTTS 音质如何同步降级在浏览器里直接用 WebRTC Insertable Streams能否把 gRPC 音频帧封装成 RTP绕过 WebSocket把这三个问题想透你的语音聊天就真正从“能跑”进化到“能抗”。—— 先记录到这里祝各位上线不炸服延迟稳稳压在 500 ms 以内。