基于 Qwen3-TTS 和 Whisper ASR 的双向语音对话系统搭建

2.3 保存和播放生成的语音

import soundfile as sf
import numpy as np

def save_audio(audio_data, filename="output.wav", sample_rate=24000):
    if len(audio_data.shape) > 1:
        audio_data = audio_data[0]
    if audio_data.max() > 1.0 or audio_data.min() < -1.0:
        audio_data = audio_data / np.max(np.abs(audio_data))
    sf.write(filename, audio_data, sample_rate)

save_audio(audio_data, "first_speech.wav")

3. Whisper ASR 模型部署与测试

3.1 加载 Whisper 模型

from transformers import WhisperProcessor, WhisperForConditionalGeneration
import librosa

model_size = "base"
processor = WhisperProcessor.from_pretrained(f"openai/whisper-{model_size}")
asr_model = WhisperForConditionalGeneration.from_pretrained(f"openai/whisper-{model_size}")
if torch.cuda.is_available():
    asr_model = asr_model.cuda()

3.2 实现语音识别功能

def speech_to_text(audio_path, language="zh"):
    audio, sr = librosa.load(audio_path, sr=16000)
    input_features = processor(audio, sampling_rate=sr, return_tensors="pt").input_features
    if torch.cuda.is_available():
        input_features = input_features.cuda()
    predicted_ids = asr_model.generate(input_features)
    transcription = processor.batch_decode(predicted_ids, skip_special_tokens=True)[0]
    return transcription

4. 构建双向语音对话系统

4.1 系统架构设计

流程：用户说话 → Whisper 识别文字 → 处理文字（LLM） → Qwen3-TTS 转语音 → 播放。

4.2 基础对话系统实现

import time
from pydub import AudioSegment
from pydub.playback import play
import tempfile

class SimpleVoiceChat:
    def __init__(self, tts_model_name="Qwen/Qwen3-TTS-12Hz-1.7B-VoiceDesign", asr_model_size="base"):
        self.tts_tokenizer = AutoTokenizer.from_pretrained(tts_model_name, trust_remote_code=True)
        self.tts_model = AutoModelForTextToSpeech.from_pretrained(tts_model_name, torch_dtype=torch.float16, device_map="auto", trust_remote_code=True)
        self.asr_processor = WhisperProcessor.from_pretrained(f"openai/whisper-{asr_model_size}")
        self.asr_model = WhisperForConditionalGeneration.from_pretrained(f"openai/whisper-{asr_model_size}")
        if torch.cuda.is_available():
            self.asr_model = self.asr_model.cuda()

    def listen(self, audio_path):
        try:
            audio, sr = librosa.load(audio_path, sr=16000)
            input_features = self.asr_processor(audio, sampling_rate=sr, return_tensors="pt").input_features
            if torch.cuda.is_available():
                input_features = input_features.cuda()
            predicted_ids = self.asr_model.generate(input_features)
            text = self.asr_processor.batch_decode(predicted_ids, skip_special_tokens=True)[0]
            return text
        except Exception as e:
            print(f"语音识别失败：{e}")
            return ""

    def think(self, user_text):
        response = f"你说的是：{user_text}"
        return response

    def speak(self, text, language="中文", voice="友好的助手"):
        try:
            input_text = f"{language}|{voice}|{text}"
            inputs = self.tts_tokenizer(input_text, return_tensors="pt")
            if torch.cuda.is_available():
                inputs = {k: v.cuda() for k, v in inputs.items()}
            with torch.no_grad():
                speech = self.tts_model.generate(**inputs)
                speech = speech.cpu().numpy()
            with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmp:
                temp_path = tmp.name
                if len(speech.shape) > 1:
                    speech = speech[0]
                sf.write(temp_path, speech, 24000)
                audio = AudioSegment.from_wav(temp_path)
                play(audio)
            return True
        except Exception as e:
            print(f"语音合成失败：{e}")
            return False

4.3 接入大语言模型增强对话能力

import openai

class SmartVoiceChat(SimpleVoiceChat):
    def __init__(self, openai_api_key, **kwargs):
        super().__init__(**kwargs)
        openai.api_key = openai_api_key
        self.conversation_history = []

    def think(self, user_text):
        self.conversation_history.append({"role": "user", "content": user_text})
        try:
            response = openai.ChatCompletion.create(
                model="gpt-3.5-turbo",
                messages=[{"role": "system", "content": "你是一个友好的语音助手。"}] + self.conversation_history[-6:],
                max_tokens=150
            )
            assistant_reply = response.choices[0].message.content
            self.conversation_history.append({"role": "assistant", "content": assistant_reply})
            return assistant_reply
        except Exception as e:
            return "抱歉，我现在无法处理你的请求。"

5. 优化技巧与常见问题

5.1 性能优化建议

1. 模型量化：使用 8 位量化减少内存占用。

from transformers import BitsAndBytesConfig
quantization_config = BitsAndBytesConfig(load_in_8bit=True)

2. 使用更小的模型：根据需求选择 tiny 或 base 版本。
3. 批处理：批量处理音频以提高效率。

5.2 常见问题与解决

• 内存不足：使用更小模型、启用量化或使用 CPU 模式。
• 识别不准：确保音频质量，尝试更大模型，指定正确语言参数。
• 合成不自然：检查输入文本格式，调整音色描述。
• 延迟太高：使用流式生成模式，优化代码，使用 GPU 加速。

6. 总结

通过本教程，我们完成了环境准备、Qwen3-TTS 与 Whisper ASR 部署、系统集成及功能增强。该方案可作为智能语音助手的基础平台，支持后续扩展多轮对话、情感分析等功能。