如何将Google Speech V1实时麦克风转写代码迁移至V2？

报错原因分析 #

你遇到的Exception iterating requests!错误，核心原因是Speech V2的请求流格式与V1完全不同：

• V2要求第一个请求必须携带完整的streaming_config和识别器资源信息
• 后续请求仅需传递音频数据块
• V2必须指定项目下的识别器（Recognizer）资源，而V1无需此步骤

正确实现步骤及代码 #

1. 安装依赖 #

pip install google-cloud-speech-v2 pyaudio

2. 核心实现代码 #

import pyaudio
from google.cloud import speech_v2
from google.cloud.speech_v2.types import cloud_speech

# 替换为你的GCP项目ID
PROJECT_ID = "your-gcp-project-id"
# 识别器名称（如果已提前在控制台创建，直接填写完整名称；否则代码会自动创建）
RECOGNIZER_ID = "auto-language-detect-recognizer"

def init_speech_client():
    return speech_v2.SpeechClient()

def create_or_get_recognizer(client):
    # 尝试获取已存在的识别器，不存在则创建
    recognizer_name = f"projects/{PROJECT_ID}/locations/global/recognizers/{RECOGNIZER_ID}"
    try:
        return client.get_recognizer(name=recognizer_name)
    except:
        # 创建支持自动语言检测的识别器
        recognizer = client.create_recognizer(
            parent=f"projects/{PROJECT_ID}/locations/global",
            recognizer_id=RECOGNIZER_ID,
            recognizer=cloud_speech.Recognizer(
                # 可选：指定可能的语言列表，提升自动检测准确率
                language_codes=["zh-CN", "en-US", "ja-JP"],
                model="latest_long",
                auto_decoding_config=cloud_speech.AutoDetectDecodingConfig(),
            ),
        )
        return recognizer

def init_microphone_stream():
    RATE = 16000
    CHUNK = int(RATE / 10)  # 100ms 音频块，平衡实时性与稳定性
    
    p = pyaudio.PyAudio()
    stream = p.open(
        format=pyaudio.paInt16,
        channels=1,
        rate=RATE,
        input=True,
        frames_per_buffer=CHUNK,
    )
    return p, stream, RATE, CHUNK

def generate_requests(recognizer_name, stream, chunk_size):
    # 第一个请求：发送识别配置
    yield cloud_speech.StreamingRecognizeRequest(
        recognizer=recognizer_name,
        streaming_config=cloud_speech.StreamingRecognitionConfig(
            config=cloud_speech.RecognitionConfig(
                auto_decoding_config=cloud_speech.AutoDetectDecodingConfig(),
                model="latest_long",
            ),
            interim_results=True,  # 启用实时中间结果
        ),
    )
    # 后续请求：持续发送麦克风音频数据
    while True:
        audio_data = stream.read(chunk_size)
        yield cloud_speech.StreamingRecognizeRequest(audio=audio_data)

def handle_responses(responses):
    for response in responses:
        if not response.results:
            continue
        result = response.results[0]
        if not result.alternatives:
            continue
        
        transcript = result.alternatives[0].transcript
        # 区分最终结果与实时中间结果
        if result.is_final:
            print(f"\n最终转写: {transcript}")
        else:
            print(f"实时转写: {transcript}", end="\r")

def main():
    client = init_speech_client()
    recognizer = create_or_get_recognizer(client)
    p, stream, rate, chunk = init_microphone_stream()
    
    try:
        requests_gen = generate_requests(recognizer.name, stream, chunk)
        responses = client.streaming_recognize(requests=requests_gen)
        handle_responses(responses)
    except Exception as e:
        print(f"\n运行错误: {str(e)}")
    finally:
        stream.stop_stream()
        stream.close()
        p.terminate()

if __name__ == "__main__":
    main()

关键注意事项 #

• GCP认证配置：确保本地环境已通过GOOGLE_APPLICATION_CREDENTIALS环境变量配置服务账号密钥，或使用gcloud auth登录
• 识别器管理：识别器是GCP上的持久化资源，可提前在控制台创建，避免代码重复创建
• 音频参数匹配：麦克风的采样率、通道数、编码格式需与识别配置兼容（示例中使用16kHz单声道PCM编码，是Speech API推荐的通用格式）
• 自动语言检测：通过AutoDetectDecodingConfig和language_codes列表优化检测效果，指定的语言越多，检测精度可能略有下降，建议仅添加业务常用语言

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