语音识别与合成中的延迟问题及解决策略

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

def speech_recognize_continuous_async_from_microphone():
    """异步从麦克风输入进行连续语音识别"""
    speech_config = speechsdk.SpeechConfig(subscription=os.getenv("SUBSCRIPTION_KEY"), 
    region="centralIndia")
    speech_recognizer = speechsdk.SpeechRecognizer(speech_config=speech_config)
    done = False
    def recognized_cb(evt: speechsdk.SpeechRecognitionEventArgs):
        print('RECOGNIZED: {}'.format(evt.result.text))

    def stop_cb(evt: speechsdk.SessionEventArgs):
        """回调函数，用于停止连续识别"""
        print('CLOSING on {}'.format(evt))
        nonlocal done
        done = True

    # 连接语音识别器触发的事件的回调函数
    speech_recognizer.recognized.connect(recognized_cb)
    speech_recognizer.session_stopped.connect(stop_cb)


    # 在识别开始时，可以在此线程上执行其他任务...
    result_future = speech_recognizer.start_continuous_recognition_async()
    result_future.get()  # 等待 voidfuture，这样我们就知道引擎初始化已完成。
    print('Continuous Recognition is now running, say something.')
    while not done:
        print('type "stop" then enter when done')
        stop = input()
        if (stop.lower() == "stop"):
            print('Stopping async recognition.')
            speech_recognizer.stop_continuous_recognition_async()
            break
    print("recognition stopped, main thread can exit now.")

speech_recognize_continuous_async_from_microphone()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

speech_config = speechsdk.SpeechConfig(subscription="YourSubscriptionKey", 
region="YourServiceRegion") 
speech_config.speech_recognition_language = "zh-CN" # 设置默认语言
## 或者 
speech_config.set_property(property_id=speechsdk.PropertyId
.SpeechServiceConnection_LanguageIdMode, value = "AtStart")
auto_detect_source_language_config = speechsdk.languageconfig
.AutoDetectSourceLanguageConfig(languages=["zh-CN", "en-US"])
speech_recognizer = speechsdk.SpeechRecognizer(speech_config=speech_config, 
audio_config=audio_config,
auto_detect_source_language_config=auto_detect_source_language_config)

1
2
3

speech_recognizer.start_continuous_recognition_async(); # 执行其他任务

speech_recognizer.stop_continuous_recognition_async()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

from pydub import AudioSegment  
import concurrent.futures  

def transcribe_chunk(chunk):  
    # 每个片段的转录逻辑  
    pass  
audio = AudioSegment.from_file("large_audio_file.wav")  
chunk_length_ms = 10000  # 10秒  
chunks = [audio[i:i + chunk_length_ms] 
for i in range(0, len(audio), chunk_length_ms)]  
  
with concurrent.futures.ThreadPoolExecutor() as executor:  
    futures = [executor.submit(transcribe_chunk, chunk) for chunk in chunks]  
    results = [f.result() for f in concurrent.futures.as_completed(futures)]  

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

def increase_audio_speed(filename, 
output_filename = "modified_audio_file.wav", speed_change_factor = 1.7):  
    # 加载音频文件  
    audio = AudioSegment.from_file(filename)  # 更改为您的文件格式  
    # 改变速度：加速（例如，1.5倍）  
    speed_change_factor = speed_change_factor  # 增加，使其变快，减少以放慢  
    new_audio = audio._spawn(audio.raw_data, 
    overrides={'frame_rate': int(audio.frame_rate * speed_change_factor)})  
    # 设置新音频的帧率  
    new_audio = new_audio.set_frame_rate(audio.frame_rate)  
    # 导出修改后的音频  
    new_audio.export(output_filename, format="wav")  # 更改为您所需的格式  

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

from pydub import AudioSegment
input_audio = 'raw_tts.wav'
output_audio = 'compressed_audio.mp3'
try:
    # 加载音频文件
    audio = AudioSegment.from_file(input_audio)  
    # 导出音频文件，使用较低的比特率进行压缩
    audio.export(output_audio, format="mp3", bitrate="64k")
    print(f"Compressed audio saved as {output_audio}")
except Exception as e:
    print(f"An error occurred: {e}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

# tts 句子结束标记
tts_sentence_end = [ ".", "!", "?", ";", "。", "！", "？", "；", "\
" ]
completion = gpt_client.chat.completions.create(
        model="gpt-4o",
    messages=[,
        {"role": "user", "content": <prompt>}
    ],
    stream=True
)
collected_messages = []
last_tts_request = None

for chunk in completion:
    if len(chunk.choices) > 0:
        chunk_text = chunk.choices[0].delta.content
        if chunk_text:
            collected_messages.append(chunk_text)
            if chunk_text in tts_sentence_end:
                text = "".join(collected_messages)
                .strip() # 将收到的消息一起连接起来构成一个句子
                last_tts_request = speech_synthesizer
                .start_speaking_text_async(text).get()
                collected_messages.clear()