需求

之前我的文章里，提到基于FunAudioLLM和deepseek开发离线智能语音对话助手。之前的测试效果能够克隆音色的CosyVoice语音合成速度慢，响应速度快的Edge-TTS不能完全离线也不能克隆想要的音色。原文章地址如下：
https://blog.csdn.net/sinat_30065705/article/details/164022356?fromshare=blogdetail&sharetype=blogdetail&sharerId=164022356&sharerefer=PC&sharesource=sinat_30065705&sharefrom=from_link

综合上述需求，才有了现在的“基于CosyVoice定制AI语音对话助手提速优化”的版本。但我的GPU是GTX3070Ti，显存只有8G，要运行3个模型（deepseek-R1-7B、CosyVoice、SenseVoice）所以只能精打细算的优化。

优化技术路线

（1）检查cuda和显卡是否可用

加载完CosyVoice-TTS模型后，加入以下代码：

# ---- 检查 GPU 状态查看cuda和显卡是否可用，不可用先解决cuda和显卡可用性问题 ----
import torch
print(f"CUDA 可用: {torch.cuda.is_available()}")
print(f"当前设备: {torch.cuda.current_device()}")
print(f"设备名称: {torch.cuda.get_device_name(0)}")

print("\n=== tts_model 的属性 ===")
for attr in dir(tts_model):
    if not attr.startswith('_'):
        obj = getattr(tts_model, attr)
        print(f"  {attr}: {type(obj)}")
        # 如果是模块，打印其设备
        if isinstance(obj, torch.nn.Module):
            try:
                device = next(obj.parameters()).device
                print(f"    -> 设备: {device}")
            except:
                pass

（2）使用混合精度推理

加载CosyVoice-TTS模型后，加入以下代码：

# 启用混合精度推理提速
if torch.cuda.is_available():
    tts_model.fp16 = True # 
    print("✅ 已启用 FP16 模式")

（3）强制使用GPU运算

理论上启用混合精度推理会自动开启GPU计算，若没能自动开启，比如TTS模型加载前后显存占用没有明显的增加，则强制使用GPU运算，在上述代码中加入以下代码：

# ---- 若指定混合精度后没有自动在GPU上运行（显存占用在TTS模型加载前后无明显变化），则强制使用GPU ----
if torch.cuda.is_available():
    # 将 model 属性（即 CosyVoiceModel 实例）移到 GPU
    tts_model.model = tts_model.model.cuda()
    # 同时将 frontend 也移到 GPU（如果有 .cuda() 方法）
    if hasattr(tts_model.frontend, 'cuda'):
        tts_model.frontend = tts_model.frontend.cuda()
    print("✅ TTS 模型已迁移到 GPU")

（4）CosyVoice模型参数调整

修改程序中text_to_speech_and_play函数中的参数：
①将 PREFETCH_CHUNKS 从 8 改为 4或 5，该参数为缓冲的音频块数量，过小则说话断断续续，过大则等待时间久。
②将sd.OutputStream中的blocksize 改为 512，过小会吐字不全。

（5）加载模型前强制开启cudnn

torch.backends.cudnn.enabled：默认为 True，保持即可。
torch.backends.cudnn.benchmark：默认为 False。设置为 True 可以让 cuDNN 自动寻找最适合当前输入尺寸的卷积算法，通常能提升推理速度。但只适用于输入尺寸固定的情况。由于TTS 输入长度变化不大（文本长度不同，但音频块长度固定），可以考虑开启。

# 强制开启cudnn
if torch.cuda.is_available():
    torch.backends.cudnn.benchmark = True
    torch.backends.cudnn.enabled = True   # 默认就是 True，但显式写出

CosyVoice模型优化后的完整代码

以下代码中的参数兼顾了响应速度和语音流畅度。

import os
import time
import tempfile
import threading
import queue
import numpy as np
import soundfile as sf
import sounddevice as sd
import pyaudio
import ollama
import re
import torch
from funasr import AutoModel
from cosyvoice.cli.cosyvoice import CosyVoice

# ===== 开启 cuDNN 自动调优 =====
torch.backends.cudnn.enabled = True
torch.backends.cudnn.benchmark = True
torch.backends.cuda.matmul.allow_tf32 = True
torch.backends.cudnn.allow_tf32 = True

# ==================== 配置 ====================
os.environ["MODELSCOPE_CACHE"] = r"E:\Project\Fun_ASR"
import sys
sys.path.insert(0, r'E:\Project\Fun_ASR\CosyVoice-main\third_party\Matcha-TTS')

# 是否启用 DeepSeek
ENABLE_LLM = True

# 录音参数（实时流）
FORMAT = pyaudio.paInt16
CHANNELS = 1
RATE = 16000
CHUNK = int(16000 * 0.1)  # 0.1 秒一帧

# ==================== VAD 参数（精细调整） ====================
VAD_THRESHOLD = 0.95          # 值越低检测灵敏度越高
VAD_FRAME_DURATION = 0.8      # 0.5 秒帧长，稳定
SILENCE_DURATION = 0.5        # 0.6 秒静音结束，更快响应
MIN_SPEECH_DURATION = 0.8     # 最短语音时长 0.4 秒，过滤更短噪音

VAD_FRAME_SAMPLES = int(RATE * VAD_FRAME_DURATION)
SILENCE_FRAMES = int(SILENCE_DURATION / VAD_FRAME_DURATION)

# 优化流式参数：更快的首包响应
PREFETCH_CHUNKS = 6  # 降低预缓存，减少首包延迟
BLOCKSIZE = 1024  # 更小的块，降低延迟
LATENCY = 0.05 # 自定义延迟 50ms（或尝试 'high' ‘low’）

# TTS 参数
TTS_SR = 22050
TTS_SPEED = 1.0

# DeepSeek 模型名称
DEEPSEEK_MODEL = "deepseekr17Bq4"

# ASR 热词和置信度
HOTWORDS = "打开我的小宝贝:200 关闭再见:200 小傻瓜 龟儿子:90 中国人 地球人"
CONFIDENCE_THRESHOLD = 0.7

# ==================== 全局变量 ====================
result_queue = queue.Queue()
is_running = True
is_playing = False
_play_lock = threading.Lock()

# ==================== 设备选择 ====================
INPUT_DEVICE_INDEX = None
OUTPUT_DEVICE_INDEX = None

def select_audio_devices():
    p = pyaudio.PyAudio()
    print("\n===== 可用的录音设备（麦克风）=====")
    input_devices = []
    for i in range(p.get_device_count()):
        info = p.get_device_info_by_index(i)
        if info['maxInputChannels'] > 0:
            input_devices.append((i, info['name']))
            print(f"  [{i}] {info['name']} (输入通道: {info['maxInputChannels']})")
    if not input_devices:
        print("⚠️ 未找到任何录音设备，将使用默认设备。")
        input_choice = None
    else:
        while True:
            try:
                choice = input("\n请选择录音设备编号（直接回车使用默认）: ")
                if choice.strip() == "":
                    input_choice = None
                    break
                idx = int(choice)
                if any(idx == dev[0] for dev in input_devices):
                    input_choice = idx
                    break
                else:
                    print("⚠️ 编号无效，请重新输入。")
            except ValueError:
                print("⚠️ 请输入有效数字。")

    print("\n===== 可用的播放设备（扬声器/耳机）=====")
    output_devices = []
    for i in range(p.get_device_count()):
        info = p.get_device_info_by_index(i)
        if info['maxOutputChannels'] > 0:
            output_devices.append((i, info['name']))
            print(f"  [{i}] {info['name']} (输出通道: {info['maxOutputChannels']})")
    if not output_devices:
        print("⚠️ 未找到任何播放设备，将使用默认设备。")
        output_choice = None
    else:
        while True:
            try:
                choice = input("\n请选择播放设备编号（直接回车使用默认）: ")
                if choice.strip() == "":
                    output_choice = None
                    break
                idx = int(choice)
                if any(idx == dev[0] for dev in output_devices):
                    output_choice = idx
                    break
                else:
                    print("⚠️ 编号无效，请重新输入。")
            except ValueError:
                print("⚠️ 请输入有效数字。")

    p.terminate()
    return input_choice, output_choice

INPUT_DEVICE_INDEX, OUTPUT_DEVICE_INDEX = select_audio_devices()
print(f"\n✅ 录音设备 ID: {INPUT_DEVICE_INDEX if INPUT_DEVICE_INDEX is not None else '默认'}")
print(f"✅ 播放设备 ID: {OUTPUT_DEVICE_INDEX if OUTPUT_DEVICE_INDEX is not None else '默认'}")

# ==================== 加载模型 ====================
print("\n正在加载 SenseVoice (ASR) 模型...")
asr_model = AutoModel(
    model="iic/SenseVoiceSmall",
    trust_remote_code=True,
    device="cuda:0",
    disable_update=True,
)
print("ASR 模型加载完成！")

print("正在加载 VAD 模型...")
vad_model = AutoModel(
    model="fsmn-vad",
    device="cuda:0",
    trust_remote_code=True,
    disable_update=True,
)
print("VAD 模型加载完成！")

print("正在加载 CosyVoice (TTS) 模型...")
tts_model = CosyVoice('E:/Project/Fun_ASR/models/iic--CosyVoice-300M/snapshots/master', fp16=True)
print("TTS 模型加载完成！")

if ENABLE_LLM:
    print("DeepSeek 模式已启用（需Ollama服务运行）")
else:
    print("语音回环模式（仅ASR+TTS，不调用LLM）")

# ==================== VAD + ASR 工作线程（带回声消除 + 内存优化） ====================
def audio_and_asr_worker():
    global is_running, is_playing
    p = pyaudio.PyAudio()
    stream = p.open(
        format=FORMAT,
        channels=CHANNELS,
        rate=RATE,
        input=True,
        input_device_index=INPUT_DEVICE_INDEX,
        frames_per_buffer=CHUNK,
    )

    print(f"🎤 VAD 音频线程启动，帧长: {VAD_FRAME_DURATION}s, 静音阈值: {SILENCE_DURATION}s")
    print(f"📏 最小语音时长: {MIN_SPEECH_DURATION}s")

    speech_buffer = b''
    is_speaking = False
    silence_counter = 0
    frame_buffer = b''

    while is_running:
        try:
            # ----- 回声消除：播放期间丢弃音频 -----
            with _play_lock:
                if is_playing:
                    stream.read(CHUNK, exception_on_overflow=False)
                    time.sleep(0.005)
                    continue

            data = stream.read(CHUNK, exception_on_overflow=False)
            frame_buffer += data

            if len(frame_buffer) >= VAD_FRAME_SAMPLES * 2:
                vad_data = frame_buffer[:VAD_FRAME_SAMPLES * 2]
                frame_buffer = frame_buffer[VAD_FRAME_SAMPLES * 2:]

                # ---- VAD 检测（直接传内存数据，无临时文件） ----
                audio_int16 = np.frombuffer(vad_data, dtype=np.int16)
                audio_float = audio_int16.astype(np.float32) / 32768.0
                try:
                    # 直接传 numpy 数组给 VAD（避免临时文件）
                    vad_res = vad_model.generate(
                        input=[audio_float],
                        batch_size=1,
                        disable_progress=True,
                        threshold=VAD_THRESHOLD,
                    )
                    has_speech = False
                    if vad_res and len(vad_res) > 0:
                        for item in vad_res:
                            if 'value' in item and item['value']:
                                has_speech = True
                                break
                except Exception as e:
                    print(f"VAD 异常: {e}")
                    has_speech = False

                # ---- 状态机 ----
                if has_speech:
                    silence_counter = 0
                    if not is_speaking:
                        is_speaking = True
                        speech_buffer = b''
                    speech_buffer += vad_data
                else:
                    if is_speaking:
                        silence_counter += 1
                        speech_buffer += vad_data
                        if silence_counter >= SILENCE_FRAMES:
                            is_speaking = False
                            if speech_buffer:
                                speech_duration = len(speech_buffer) / (RATE * 2)
                                if speech_duration < MIN_SPEECH_DURATION:
                                    print(f"⚠️ 丢弃过短语音片段: {speech_duration:.2f}秒")
                                    speech_buffer = b''
                                    continue

                                # ---- 语音段识别（直接传内存数据） ----
                                audio_int16_full = np.frombuffer(speech_buffer, dtype=np.int16)
                                audio_float_full = audio_int16_full.astype(np.float32) / 32768.0

                                try:
                                    res = asr_model.generate(
                                        input=[audio_float_full],  # 直接传 numpy 数组
                                        batch_size=1,
                                        disable_progress=True,
                                        hotword=HOTWORDS,
                                        language="zh",
                                        use_itn=True,
                                        output_timestamp=True,
                                        beam_size=5,
                                        decoding_ctc_weight=0.3,
                                    )
                                    if res and len(res) > 0:
                                        timestamps = res[0].get("timestamps", [])
                                        if timestamps:
                                            filtered_tokens = []
                                            for token_info in timestamps:
                                                score = token_info.get("score", 0)
                                                token = token_info.get("token", "")
                                                if score >= CONFIDENCE_THRESHOLD:
                                                    filtered_tokens.append(token)
                                            filtered_text = ''.join(filtered_tokens).strip()
                                            if filtered_text:
                                                filtered_text = re.sub(r'<\|[^|]+\|>', '', filtered_text).strip()
                                                if filtered_text:
                                                    result_queue.put(filtered_text)
                                        else:
                                            text = res[0].get("text", "").strip()
                                            if text:
                                                text = re.sub(r'<\|[^|]+\|>', '', text).strip()
                                                if text:
                                                    result_queue.put(text)
                                except Exception as e:
                                    print(f"ASR 识别错误: {e}")
                                finally:
                                    speech_buffer = b''
        except Exception as e:
            print(f"音频线程异常: {e}")
            break

    stream.stop_stream()
    stream.close()
    p.terminate()
    print("音频线程已停止。")

# ==================== TTS + 流式播放（优化参数 + 显存管理） ====================
def text_to_speech_and_play(text):
    global is_playing
    if not text:
        return
    print(f"🔊 正在合成语音: {text}")

    with _play_lock:
        is_playing = True
    print("🔒 播放模式已启用（VAD 线程暂停）")

    stream_out = None
    chunk_queue = []
    total_samples = 0
    sample_rate = TTS_SR
    is_first_batch = True

    try:
        for result in tts_model.inference_zero_shot(
                tts_text=text,
                prompt_text='',
                prompt_wav='',
                zero_shot_spk_id='my_voice',
                speed=TTS_SPEED,
                stream=True
        ):
            if 'tts_speech' not in result:
                continue

            audio_chunk = result['tts_speech']

            if hasattr(audio_chunk, 'numpy'):
                audio_chunk = audio_chunk.numpy()
            elif hasattr(audio_chunk, 'cpu') and hasattr(audio_chunk, 'detach'):
                audio_chunk = audio_chunk.cpu().detach().numpy()

            if audio_chunk.dtype != np.float32:
                audio_chunk = audio_chunk.astype(np.float32)
            if len(audio_chunk.shape) > 1:
                audio_chunk = audio_chunk.flatten()

            chunk_queue.append(audio_chunk)
            total_samples += len(audio_chunk)

            if len(chunk_queue) >= PREFETCH_CHUNKS and is_first_batch:
                is_first_batch = False
                print(f"🔊 预缓存 {len(chunk_queue)} 块，开始流式播放...")

                first_batch = np.concatenate(chunk_queue)
                chunk_queue = []

                try:
                    stream_out = sd.OutputStream(
                        samplerate=sample_rate,
                        channels=1,
                        dtype='float32',
                        device=OUTPUT_DEVICE_INDEX if OUTPUT_DEVICE_INDEX is not None else None,
                        blocksize=BLOCKSIZE,
                        latency=LATENCY,
                    )
                    stream_out.start()
                    stream_out.write(first_batch)
                    print(f"✅ 首批数据已写入，{len(first_batch)} 采样点")
                except Exception as e:
                    print(f"❌ 播放流初始化失败: {e}，尝试默认设备")
                    stream_out = sd.OutputStream(
                        samplerate=sample_rate,
                        channels=1,
                        dtype='float32',
                        blocksize=BLOCKSIZE,
                        latency='low'
                    )
                    stream_out.start()
                    stream_out.write(first_batch)

            elif not is_first_batch and chunk_queue:
                if len(chunk_queue) >= 3: # 增加此值也可以提升流畅度
                    batch = np.concatenate(chunk_queue)
                    chunk_queue = []
                    try:
                        if stream_out and stream_out.active:
                            stream_out.write(batch)
                        else:
                            print("⚠️ 输出流异常，尝试重新启动...")
                            if stream_out:
                                try:
                                    stream_out.stop()
                                    stream_out.close()
                                except:
                                    pass
                            stream_out = sd.OutputStream(
                                samplerate=sample_rate,
                                channels=1,
                                dtype='float32',
                                blocksize=BLOCKSIZE,
                                latency='low'
                            )
                            stream_out.start()
                            stream_out.write(batch)
                    except Exception as e:
                        print(f"❌ 写入失败: {e}")

        # 剩余块处理
        if chunk_queue:
            print(f"📦 写入剩余 {len(chunk_queue)} 块...")
            remaining = np.concatenate(chunk_queue)
            chunk_queue = []
            try:
                if stream_out and stream_out.active:
                    stream_out.write(remaining)
                else:
                    print("⚠️ 输出流不可用，使用非流式播放剩余...")
                    sd.play(remaining, samplerate=sample_rate, device=OUTPUT_DEVICE_INDEX)
                    sd.wait()
            except Exception as e:
                print(f"❌ 写入剩余块失败: {e}")

        if stream_out is not None:
            time.sleep(0.1)
            stream_out.stop()
            stream_out.close()
            total_duration = total_samples / sample_rate
            print(f"✅ 播放完成，总计: {total_samples} 采样点, {total_duration:.2f} 秒")
        else:
            print("❌ 未生成任何音频数据")

    except Exception as e:
        print(f"❌ 流式合成异常: {e}")
        import traceback
        traceback.print_exc()
    finally:
        # 清理资源
        if stream_out is not None:
            try:
                stream_out.stop()
                stream_out.close()
            except:
                pass

        # 显存清理
        if torch.cuda.is_available():
            torch.cuda.empty_cache()

        time.sleep(0.2)
        with _play_lock:
            is_playing = False
        print("🔓 播放模式已关闭（VAD 线程恢复）")
        print("✅ 播放资源已释放")

# ==================== DeepSeek 调用（带显存清理） ====================
def ask_deepseek(prompt):
    print(f"🤖 正在询问 DeepSeek: {prompt}")
    try:
        torch.cuda.empty_cache()
        response = ollama.chat(
            model=DEEPSEEK_MODEL,
            messages=[{'role': 'user', 'content': prompt}],
            options={'think': False}  # 隐藏思考过程
        )
        reply = response['message']['content'].strip()
        print(f"💬 DeepSeek 回复: {reply}")
        return reply
    except Exception as e:
        print(f"❌ 调用 DeepSeek 失败: {e}")
        return "抱歉，我现在无法思考。"
    finally:
        torch.cuda.empty_cache()

# ==================== 文本有效性判断 ====================
def is_valid_utterance(text):
    if not text:
        return False
    cleaned = text.strip('。，、！？；：,.!?;: ')
    chars = re.findall(r'[\u4e00-\u9fff]|[a-zA-Z]', cleaned)
    return len(chars) >= 3

# ==================== 主程序 ====================
if __name__ == "__main__":
    # 唤醒词和结束词
    WAKE_WORD = "打开我的小宝贝"
    EXIT_WORD = "关闭再见"
    conversation_active = False  # 对话模式标志

    print("\n" + "=" * 50)
    print("语音对话助手启动 (优化版)")
    if ENABLE_LLM:
        print("模式: 智能问答 (ASR + DeepSeek + TTS)")
    else:
        print("模式: 语音回环 (ASR + TTS)")
    print("按 Ctrl+C 退出")
    print("=" * 50 + "\n")

    asr_thread = threading.Thread(target=audio_and_asr_worker, daemon=True)
    asr_thread.start()

    try:
        while True:
            user_text = result_queue.get()
            if not user_text:
                continue
            if not is_valid_utterance(user_text):
                # print(f"⚠️ 忽略无效输入: '{user_text}'")
                continue

            # 检查唤醒词（忽略大小写，去除空格后判断）
            cleaned = re.sub(r'[，。、！？；：,.!?;: ]', '', user_text)  # 移除常见标点和空格
            if WAKE_WORD in cleaned:
                conversation_active = True
                print(f"🔊 检测到唤醒词 '{WAKE_WORD}'，进入对话模式")
                continue  # 不处理本次输入

            # 检查结束词
            if EXIT_WORD in cleaned:
                conversation_active = False
                print(f"🔇 检测到结束词 '{EXIT_WORD}'，退出对话模式")
                continue

            # 如果未激活，忽略
            if not conversation_active:
                # print(f"⏳ 未唤醒，忽略输入: '{user_text}'")
                continue

            if ENABLE_LLM & conversation_active:
                print(f"👤 你说: {user_text}")
                reply_text = ask_deepseek(user_text)
            else:
                reply_text = f"你说的是：{user_text}"
                # print(f"🔄 回环回复: {reply_text}")

            text_to_speech_and_play(reply_text)

    except KeyboardInterrupt:
        print("\n👋 程序正在退出...")
    finally:
        time.sleep(0.5)
        is_running = False
        asr_thread.join(timeout=1)
        print("程序已退出。")

优化后的智能语音对话助手演示视频