使用Google Cloud Speech-to-Text实时转录Asterisk录制的GSM音频失败求助

核心问题 #

1. 编码配置不匹配：代码中指定了WEBM_OPUS编码，但实际处理的是GSM格式音频，Google Speech无法正确解析，最终触发识别超时。
2. 流读取逻辑缺陷：当前GSMFileStream生成器读完文件现有内容后直接终止，未等待Asterisk写入新音频数据，导致Google Speech服务长时间收不到新数据触发超时。
3. Asterisk文件写入缓冲：Asterisk默认会缓冲录制数据，不会实时刷新到磁盘，导致读取进程无法获取最新录制内容。

具体修复步骤 #

1. 纠正音频编码配置

将RecognitionConfig中的编码改为GSM对应的枚举值：

config = speech.RecognitionConfig(
    encoding=speech.RecognitionConfig.AudioEncoding.GSM,  # 替换原WEBM_OPUS
    sample_rate_hertz=RATE,
    language_code=language_code,
)

2. 修改流读取逻辑，持续等待新数据

更新GSMFileStream的generator方法，当读取不到数据时短暂等待后继续尝试，直到流被关闭：

def generator(self):
    while not self.closed:
        self.file.seek(self.offset)
        chunk = self.file.read(self.chunk_size)
        if chunk:
            self.offset += len(chunk)
            yield chunk
        else:
            # 无新数据时等待100ms，避免CPU空转
            time.sleep(0.1)

3. 确保Asterisk实时刷新录制文件

修改record_file函数，调用Asterisk AGI的record_file时添加f选项强制刷新缓冲区：

def record_file(filename, agi):
    # 示例：使用agi.record_file时添加flush选项
    agi.record_file(
        f"/app/softphone_app/wavs/{filename}.gsm",
        format="gsm",
        options="f",  # 强制实时刷新到磁盘
        timeout=30000  # 根据需求设置超时
    )

4. 优化多进程启动逻辑，替代固定sleep

不要依赖time.sleep(5)，改为监控文件大小，当文件有足够数据时再启动转写进程：

import os

def record_and_tts_task(agi):
    p1 = None
    p2 = None
    audio_file = "/app/softphone_app/wavs/client.gsm"

    try:
        p1 = multiprocessing.Process(target=record_file, args=('client', agi))
        p1.start()
        logger.info("RECORDING FILE STARTED!")

        # 等待文件创建并达到最小数据量（比如1024字节）
        while not os.path.exists(audio_file) or os.path.getsize(audio_file) < 1024:
            time.sleep(0.5)

        p2 = multiprocessing.Process(target=stt_stream)
        p2.start()
        logger.info("TTS Stream STARTED!")

    except Exception as e:
        logger.error(f"Error while starting processes: {e}")

    finally:
        if p1 and p1.is_alive():
            p1.join()
            logger.info("RECORDING FILE ENDED!")
        if p2 and p2.is_alive():
            p2.join()
            logger.info("TTS Stream ENDED!")

5. 调整转写循环，支持持续实时识别

修改listen_print_loop，不要在第一个final结果时返回，而是持续处理所有结果：

def listen_print_loop(responses: object) -> None:
    for response in responses:
        if not response.results:
            continue

        result = response.results[0]
        if not result.alternatives:
            continue

        alternative = result.alternatives[0]
        if result.is_final:
            logger.info(f"Final Transcript: {alternative.transcript}")
        else:
            logger.info(f"Interim Transcript: {alternative.transcript}")

额外注意事项 #

• 确保Google Speech-to-Text的配额足够支持实时流识别。
• 可考虑使用共享内存或管道替代文件读取，减少磁盘IO延迟，提升实时性。
• 测试时可先手动生成一个正在写入的GSM文件（比如用ffmpeg持续写入），验证转写逻辑正常后再对接Asterisk。

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