Python中如何捕获Google Speech API流式识别的GRPC期限过短异常

这个错误是Google Speech-to-Text API的硬性限制导致的——流式识别请求的GRPC deadline必须满足 至少3倍当前音频时长 + 5秒，当你设置的deadline低于这个阈值时就会触发这个InvalidArgument异常。下面是具体的解决思路和代码调整方案：

核心解决方案：动态计算并调整Deadline #

不要设置固定的deadline值，而是根据当前音频流的实时时长动态计算符合要求的deadline，甚至可以额外增加一点冗余时间来应对网络波动或API处理延迟。

修改后的代码示例 #

结合你的代码结构，调整后可以这样处理（包含异常捕获和重试逻辑）：

from google.api_core.exceptions import InvalidArgument

with self.MicrophoneStream(self.RATE, self.CHUNK) as stream:
    # 维护变量跟踪已发送的音频总时长
    total_audio_duration = 0.0
    while True:
        try:
            # 计算符合要求的最小deadline，额外加5秒冗余避免边界问题
            required_deadline = 3 * total_audio_duration + 10
            
            # 生成音频请求流，同时累加时长
            audio_generator = stream.generator()
            requests = []
            for content in audio_generator:
                requests.append(speech.StreamingRecognizeRequest(audio_content=content))
                # 累加当前音频块的时长
                total_audio_duration += self.CHUNK / self.RATE
            
            # 发起识别请求时设置动态计算的deadline
            operation = self.client.streaming_recognize(
                self.config, 
                requests, 
                timeout=required_deadline
            )
            
            # 处理识别结果的逻辑
            for result in operation:
                # 你的结果处理代码，比如打印转录文本
                print(result.alternatives[0].transcript)
                
        except InvalidArgument as e:
            # 捕获到deadline不足的异常，打印信息后重试
            print(f"触发Deadline异常: {e}，将重新计算Deadline后重试")
            continue

关键细节说明 #

• 时长跟踪逻辑：如果你的MicrophoneStream没有内置时长统计，就像示例里那样，每次获取音频块时累加CHUNK / RATE来计算总时长，这个值就是当前已发送音频的秒数。
• 冗余时间的必要性：直接用3*total_audio_duration+5刚好踩阈值，容易因为网络延迟或API处理慢再次触发异常，额外加几秒冗余会让请求更稳定。
• 重试逻辑：捕获异常后直接重试即可，因为此时重新计算的deadline已经满足API的最低要求。

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