Python中向Azure Speech Translation Service传入自定义音频字节的问题解决

我来帮你搞定这个问题！你遇到的AttributeError: 'PullAudioInputStreamCallback' object has no attribute '_impl'错误，核心原因是你直接把回调实例传给了AudioConfig，而正确的流程是先把回调包装成PullAudioInputStream对象，再交给音频配置。另外你的回调类实现和数据传递逻辑也有一些需要调整的地方，下面一步步给你修正：

错误原因分析 #

1. 你注释掉了PullAudioInputStream的创建代码，直接用PullAudioInputStreamCallback实例初始化AudioConfig，但SDK要求传入的是流对象而非回调对象。
2. CustomPullAudioInputStreamCallback的read方法实现错误：这个方法需要返回要提供给SDK的音频字节（不是长度），当没有更多数据时返回空字节。
3. 你手动调用pull_audio_input_stream_callback.read(audio_bytes)是不对的，应该让SDK在需要数据时自动调用回调的read方法。

正确实现代码 #

下面是调整后的完整代码，实现了直接传入音频字节进行翻译，无需保存文件：

import azure.cognitiveservices.speech as speechsdk
from azure.cognitiveservices.speech.audio import (
    AudioStreamFormat,
    PullAudioInputStream,
    PullAudioInputStreamCallback,
    AudioConfig
)

# 替换成你的密钥和区域
speech_key, service_region = "your-speech-key", "your-service-region"
channels = 1
bitsPerSample = 16
samplesPerSecond = 16000
audioFormat = AudioStreamFormat(samplesPerSecond, bitsPerSample, channels)

class CustomPullAudioInputStreamCallback(PullAudioInputStreamCallback):
    def __init__(self):
        super().__init__()
        self.audio_data = b""  # 存储要处理的音频字节
        self.offset = 0  # 当前读取的位置

    def set_audio_data(self, audio_bytes):
        # 设置要处理的音频字节
        self.audio_data = audio_bytes
        self.offset = 0

    def read(self, buffer_size):
        # SDK会调用这个方法获取音频数据，返回最多buffer_size字节的数据
        if self.offset >= len(self.audio_data):
            return b""  #  没有更多数据时返回空字节
        end = self.offset + buffer_size
        chunk = self.audio_data[self.offset:end]
        self.offset = end
        return chunk

    def close(self):
        super().close()

# 初始化翻译配置
translation_config = speechsdk.translation.SpeechTranslationConfig(
    subscription=speech_key,
    region=service_region
)
fromLanguage = 'en-US'
toLanguage = 'hi'
translation_config.speech_recognition_language = fromLanguage
translation_config.add_target_language(toLanguage)
translation_config.voice_name = "hi-IN-Kalpana-Apollo"

# 创建回调和流对象
pull_callback = CustomPullAudioInputStreamCallback()
pull_audio_stream = PullAudioInputStream(pull_callback, audioFormat)
audio_config = AudioConfig(stream=pull_audio_stream)

# 初始化翻译识别器
recognizer = speechsdk.translation.TranslationRecognizer(
    translation_config=translation_config,
    audio_config=audio_config
)

# 处理合成的音频结果（这里可以直接处理字节，不保存文件）
def synthesis_callback(evt):
    size = len(evt.result.audio)
    print(f'AUDIO SYNTHESIZED: {size} byte(s) {"(COMPLETED)" if size == 0 else ""}')
    if size > 0:
        # 这里可以直接使用evt.result.audio字节，无需保存到文件
        print("已获取翻译后的音频字节，长度：", size)
        # 示例：如果需要播放或进一步处理，直接用这个字节数据即可
    recognizer.stop_continuous_recognition_async()

# 处理识别和翻译结果
def recognized_complete(evt):
    result = evt.result  # 注意：之前的代码里漏了evt.，导致错误
    if result.reason == speechsdk.ResultReason.TranslatedSpeech:
        print(f"RECOGNIZED '{fromLanguage}': {result.text}")
        print(f"TRANSLATED into {toLanguage}: {result.translations['hi']}")
    elif result.reason == speechsdk.ResultReason.RecognizedSpeech:
        print(f"RECOGNIZED: {result.text} (text could not be translated)")
    elif result.reason == speechsdk.ResultReason.NoMatch:
        print(f"NOMATCH: Speech could not be recognized: {result.no_match_details}")
    elif result.reason == speechsdk.ResultReason.Canceled:
        print(f"CANCELED: Reason={result.cancellation_details.reason}")
        if result.cancellation_details.reason == speechsdk.CancellationReason.Error:
            print(f"CANCELED: ErrorDetails={result.cancellation_details.error_details}")

# 接收音频字节并启动翻译
def receiving_bytes(audio_bytes):
    # 设置要处理的音频字节
    pull_callback.set_audio_data(audio_bytes)
    # 绑定事件
    recognizer.synthesizing.connect(synthesis_callback)
    recognizer.recognized.connect(recognized_complete)
    # 启动连续识别
    recognizer.start_continuous_recognition_async()

# 示例调用：替换成你的音频字节数据
# audio_bytes = b"你的音频字节数据"
# receiving_bytes(audio_bytes)

关键调整说明 #

1. 回调类改进：新增set_audio_data方法来传入音频字节，read方法负责分块返回数据给SDK，符合SDK的调用逻辑。
2. 正确创建流对象：把回调包装成PullAudioInputStream后再传给AudioConfig，解决了_impl属性缺失的错误。
3. 事件处理修正：修复了recognized_complete里未使用evt.result的错误，避免变量未定义问题。
4. 无需保存文件：翻译后的音频字节直接通过evt.result.audio获取，你可以直接使用这些字节进行播放、存储或其他处理，不需要写入文件。

注意事项 #

• 确保你的音频字节格式和AudioStreamFormat定义的一致（16kHz采样率、16位单声道），否则会导致识别失败。
• 如果是长音频，这个方案可以正常处理，SDK会自动通过回调分块读取数据。

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