Whisper+T5-translate实现python实时语音翻译

1.首先下载模型，加载模型

1. import torch
2. import numpy as np
3. import webrtcvad
4. import pyaudio
5. import queue
6. import threading
7. from datetime import datetime
8. from faster_whisper import WhisperModel
9. from transformers import AutoTokenizer, AutoModelForSeq2SeqLM,pipeline
10. from transformers import T5ForConditionalGeneration, T5Tokenizer
11. device = "cuda:0" if torch.cuda.is_available() else "cpu"
12. torch_dtype = torch.float16
13. save_directory = "./faster-distil-whiper-large-v3-local"  # 替换为你希望保存的本地路径
14. # en_zh_directory = "./opus-mt-en-zh-local"  # 替换为你希望保存的本地路径
15. en_zh_directory = "./t5-translate-en-ru-zh-base-200-sent-local"  # 替换为你希望保存的本地路径
16. whisperModel = WhisperModel(save_directory, device="cuda", compute_type="float32")
19. model = T5ForConditionalGeneration.from_pretrained(en_zh_directory)
20. model.eval()
21. model.to(device)
22. tokenizer = T5Tokenizer.from_pretrained(en_zh_directory)
23. vad = webrtcvad.Vad(3)  # 设置 VAD 灵敏度（0-3，3 最敏感）
24. prefix = 'translate to zh: '

复制代码

2.配置麦克风

1. # 初始化 PyAudio
2. p = pyaudio.PyAudio()
3. # 设置音频流参数
4. FORMAT = pyaudio.paInt16  # 16-bit 音频格式
5. CHANNELS = 1              # 单声道
6. RATE = 16000              # 采样率（Whisper 需要 16kHz）
7. FRAME_DURATION = 20       # 每帧的时长（ms）
8. CHUNK = int(RATE * FRAME_DURATION / 1000)  # 每帧的帧数
9. MIN_SILENCE_DURATION = 0.2  # 最小静音时长（秒）

复制代码

3.队列构建，构建录音基本参数

1. # 共享队列，用于录音和推理线程之间的数据交换
2. audio_queue = queue.Queue()
4. silence_frames = 0
5. silence_frames_lock = threading.Lock()

复制代码

4.构建录音函数

1. # 录音线程
2. def record_audio():
3.     global silence_frames
4.     stream = p.open(
5.         format=FORMAT,
6.         channels=CHANNELS,
7.         rate=RATE,
8.         input=True,
9.         frames_per_buffer=CHUNK,
10.     )
11.     print("开始录音...按 Ctrl+C 停止")
12.     try:
13.         while True:
14.             # 从麦克风读取音频数据
15.             data = stream.read(CHUNK)
16.             audio_data = np.frombuffer(data, dtype=np.int16).astype(np.float32) / 32768.0
18.             # 使用 VAD 检测语音活动
19.             if vad.is_speech(data, RATE):
20.                 audio_queue.put(audio_data)
21.                 with silence_frames_lock:
22.                     silence_frames = 0  # 重置静音计数器
23.             else:
24.                 with silence_frames_lock:
25.                     silence_frames += 1  # 重置静音计数器
26.     except KeyboardInterrupt:
27.         print("录音停止")
28.     finally:
29.         stream.stop_stream()
30.         stream.close()
31.         p.terminate()

复制代码

5.构建翻译函数

1. def process_audio():
2.     global silence_frames
3.     audio_buffer = np.array([], dtype=np.float32)
4.     silence_frames = 0
6.     while True:
7.         try:
8.             # 从队列中获取音频数据
9.             audio_data = audio_queue.get(timeout=1)  # 超时 1 秒
10.             audio_buffer = np.concatenate((audio_buffer, audio_data))
12.         except Exception as e:
13.             pass
14.         current_silence_frames = 0
15.         # 检查静音计数器
16.         with silence_frames_lock:
17.             current_silence_frames = silence_frames
19.         # 如果检测到静音时间超过阈值，处理累积的音频
20.         if (current_silence_frames > MIN_SILENCE_DURATION * (RATE / CHUNK)) or len(audio_buffer) > 320 * 200:
21.             if(len(audio_buffer) > 0):#, language="en"
22.                 segments, _ = whisperModel.transcribe(audio_buffer,vad_filter=True,vad_parameters=dict(min_silence_duration_ms=200), language="en", condition_on_previous_text=True)
23.                 for segment in segments:
24.                     if(segment.text == ""):
25.                         continue
26.                     elif(segment.text == "Thank you."):
27.                         print("[%s] %s (%s)" % (str(datetime.now()), "感谢", segment.text))
28.                     else:            
29.                         src_text = prefix + segment.text
30.                         
31.                         input_ids = tokenizer(src_text, return_tensors="pt")
32.                         generated_tokens = model.generate(**input_ids.to(device))
33.                         result = tokenizer.batch_decode(generated_tokens, skip_special_tokens=True)
34.                         print("[%s] %s (%s)" % (str(datetime.now()), result[0], segment.text))
35.                     
36.                     
37.                     # result = pipeline(segment.text)
38.                     # print("[%s] %s (%s)" % (str(datetime.now()), result[0]['translation_text'], segment.text))
39.             
40.                 audio_buffer = np.array([], dtype=np.float16)
41.                 silence_frames = 0

复制代码

6.启动线程，启动步伐

1. # 启动录音线程和推理线程
2. record_thread = threading.Thread(target=record_audio)
3. process_thread = threading.Thread(target=process_audio)
5. record_thread.start()
6. process_thread.start()

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。