Discord机器人Whisper语音转录优化：解决不一致性与提升准确性

import numpy as np
from typing import Any

class WhisperProcessor:
    def __init__(self, model: Any): # model would be your loaded Whisper model
        self.model = model

    def binary_transcribe(self, audio_data: bytes) -> str:
        # 将二进制数据从int16转换为float32并归一化
        audio_np = np.frombuffer(audio_data, dtype=np.int16)
        audio_np = audio_np.astype('float32') / 32767.0

        # 直接传递给Whisper模型
        result = self.model.transcribe(audio_np)
        print("Transcription completed...")
        return result["text"]

import whisperx
import numpy as np
import io
import soundfile as sf # 用于将PCM数据转换为WAV，如果whisperx直接支持numpy float32则不需要

# 假设这是从Discord机器人获取的原始PCM二进制数据
# 实际应用中，audio_data_bytes 会通过interactions库捕获
# 这里我们模拟一个PCM数据
def simulate_pcm_data(duration_seconds=5, sample_rate=16000, frequency=440):
    t = np.linspace(0, duration_seconds, int(sample_rate * duration_seconds), endpoint=False)
    # 模拟一个简单的正弦波作为语音内容，并添加一些静音和噪声
    pure_tone = 0.5 * np.sin(2 * np.pi * frequency * t)
    noise = np.random.normal(0, 0.05, pure_tone.shape)
    # 加入一些静音
    silence_start = int(sample_rate * 0.5)
    silence_end = int(sample_rate * 1.5)
    pure_tone[silence_start:silence_end] = 0

    audio_float = (pure_tone + noise).astype(np.float32)
    # 转换为int16模拟原始PCM
    audio_int16 = (audio_float * 32767).astype(np.int16)
    return audio_int16.tobytes()

# 从Discord机器人获取的原始PCM二进制数据
discord_audio_data_bytes = simulate_pcm_data(duration_seconds=10)

# 1. 将原始PCM二进制数据转换为WhisperX兼容的numpy数组 (float32, 16kHz)
# 假设原始PCM是16kHz采样率，单声道
audio_np_int16 = np.frombuffer(discord_audio_data_bytes, dtype=np.int16)
audio_np_float32 = audio_np_int16.astype(np.float32) / 32767.0 # 归一化到-1.0到1.0

# 2. 加载WhisperX模型
# 可以选择不同的模型大小，例如 "large-v2", "medium", "small"
# device 可以是 "cuda" (如果可用) 或 "cpu"
# compute_type 可以是 "float16" (更快，需要GPU) 或 "int8" (更省内存，可能略降精度)
device = "cuda" if whisperx.utils.is_accelerate_available() else "cpu"
batch_size = 16 # 批处理大小，根据GPU内存调整
compute_type = "float16" if device == "cuda" else "int8"

print(f"Loading whisperX model on {device} with compute_type={compute_type}...")
model = whisperx.load_model("large-v2", device, compute_type=compute_type)

# 3. 使用WhisperX进行转录 (内置VAD)
print("Starting transcription with whisperX...")
result = model.transcribe(audio_np_float32, batch_size=batch_size)

# 打印转录结果
print("\nTranscription Result:")
for segment in result["segments"]:
    print(f"[{segment['start']:.2f}s - {segment['end']:.2f}s] {segment['text']}")

# 可选：加载对齐模型以获取更精确的单词级时间戳
# if result["language"] != "zh": # 假设我们主要处理中文，但如果语言检测到其他语言，可以加载对应对齐模型
#     model_a, metadata = whisperx.load_align_model(language_code=result["language"], device=device)
#     result = whisperx.align(result["segments"], model_a, metadata, audio_np_float32, device, return_char_alignments=False)
#     print("\nAligned Transcription Result:")
#     for segment in result["segments"]:
#         print(f"[{segment['start']:.2f}s - {segment['end']:.2f}s] {segment['text']}")

# 终止模型资源（如果需要）
# del model
# if 'model_a' in locals():
#     del model_a