如何将Google Cloud Speech-to-Text流式文本发送至Gemini API？

核心逻辑 #

要实现流式处理，关键是正确处理Speech-to-Text的临时/最终分块，并节流式地将最新文本同步到Gemini进行流式翻译：

• 维护两个缓冲区：存储已确认的最终文本，以及临时更新的未确认文本，合并后作为当前待翻译内容
• 对Gemini调用做节流控制，避免因临时文本频繁更新导致的重复请求

具体实现步骤 #

1. 依赖安装

安装Google Cloud Speech-to-Text和Gemini的Python SDK：

pip install google-cloud-speech google-generativeai

2. 代码实现

以下是完整的流式处理示例（包含Speech-to-Text流式接收和Gemini流式翻译）：

import asyncio
from google.cloud import speech
from google import generativeai as genai
import os

# 配置认证信息
os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = "path/to/speech-service-account.json"
genai.configure(api_key="your-gemini-api-key")

# 全局缓冲区与节流配置
final_transcript = ""
temp_transcript = ""
last_call_time = 0
THROTTLE_INTERVAL = 0.2  # 200ms间隔，控制Gemini调用频率

# 初始化Gemini模型
gemini_model = genai.GenerativeModel('gemini-1.5-flash')

async def handle_speech_stream():
    global final_transcript, temp_transcript, last_call_time

    # 配置Speech-to-Text参数
    speech_client = speech.SpeechClient()
    recognition_config = speech.RecognitionConfig(
        encoding=speech.RecognitionConfig.AudioEncoding.LINEAR16,
        sample_rate_hertz=16000,
        language_code="zh-CN",  # 替换为你的源语言
        enable_automatic_punctuation=True
    )
    streaming_config = speech.StreamingRecognitionConfig(
        config=recognition_config,
        interim_results=True  # 开启临时结果返回
    )

    # 替换为真实音频输入流（比如麦克风读取）
    def audio_input_generator():
        # 示例：模拟音频数据，实际需替换为从麦克风/文件读取的二进制音频
        while True:
            yield b""

    # 发起流式语音识别请求
    requests = (speech.StreamingRecognizeRequest(audio_content=chunk) 
               for chunk in audio_input_generator())
    responses = speech_client.streaming_recognize(streaming_config, requests)

    # 处理识别结果
    for response in responses:
        for result in response.results:
            current_text = result.alternatives[0].transcript
            is_final = result.is_final

            if is_final:
                # 最终文本块：追加到已确认缓冲区，清空临时内容
                final_transcript += f"{current_text} "
                temp_transcript = ""
            else:
                # 临时文本：覆盖当前临时缓冲区
                temp_transcript = current_text

            # 节流控制：达到间隔才调用Gemini
            current_time = asyncio.get_event_loop().time()
            if current_time - last_call_time > THROTTLE_INTERVAL:
                last_call_time = current_time
                # 异步启动翻译任务，避免阻塞语音识别流程
                asyncio.create_task(run_stream_translate(final_transcript + temp_transcript))

async def run_stream_translate(input_text):
    if not input_text.strip():
        return

    # 构造翻译提示（替换为目标语言要求）
    prompt = f"将以下文本翻译成英文：{input_text}"

    # 流式调用Gemini生成翻译
    translation_stream = gemini_model.generate_content(prompt, stream=True)
    # 实时更新翻译结果（前端可替换为DOM更新逻辑）
    print("\r当前翻译：", end="")
    full_translation = ""
    for chunk in translation_stream:
        if chunk.text:
            full_translation += chunk.text
            print(f"\r当前翻译：{full_translation}", end="")
    print()

if __name__ == "__main__":
    asyncio.run(handle_speech_stream())

关键注意事项 #

• 节流控制：必须设置调用间隔，否则临时文本的高频更新会导致Gemini请求过载，同时用户体验也会因翻译频繁闪烁变差
• 缓冲区管理：严格区分临时和最终文本，确保发送给Gemini的是当前最完整的有效内容
• 错误处理：实际生产环境需添加异常捕获（如API连接失败、音频流中断），保证流程稳定性
• 成本与配额：流式调用会增加API请求次数，需关注Google Cloud的Speech-to-Text和Gemini的配额限制与计费规则
• 语言适配：根据实际需求修改Speech-to-Text的language_code和Gemini提示中的目标语言

内容的提问来源于stack exchange，提问作者MohamedLEGH