从零手搓“贾维斯”：构建超低延迟的 LLM 实时语音对话系统 (附全套开源代码)

在当前的大模型时代，做一个文本对话机器人很简单，但要想做一个像人类一样实时听、实时想、实时说，且不抢话、没有回音死循环的语音助手，工程难度将呈指数级上升。

本文将详细复盘如何利用 Qwen3-ASR、SiliconFlow (Qwen2.5-7B) 和 CosyVoice，在本地从零构建一套工业级的实时语音流式交互架构，并重点解决异步并发死锁与声学回声套娃这两个终极难题。

🏗️ 一、 架构设计：微服务与环境隔离

在引入开源语音模型时，最大的忌讳就是把 ASR（语音识别）和 TTS（语音合成）装在同一个 Python 环境中。底层 PyTorch 和 CUDA 版本的冲突会让你陷入无尽的依赖地狱 (Dependency Hell)。

因此，我们采用了基于 WebSocket 的微服务架构：

1. Windows 客户端 (client_vad.py)：负责麦克风录音、VAD (静音检测)、以及扬声器播放。
2. ASR + LLM 服务端 (server_asr.py)：运行在独立的 Conda 环境 A 中。负责将音频转文字，并流式请求大模型。
3. TTS 服务端 (server_tts.py)：运行在独立的 Conda 环境 B 中。负责接收文字，使用 CosyVoice 进行 3s 极速音色克隆 (Zero-Shot)，并流式吐出音频。

[ Windows 客户端 (录音/播放/VAD) ] 
        |  (WebSocket 端口 8765)
[ WSL2 环境 A: Qwen3-ASR + LLM 服务端 ]
        |  (WebSocket 端口 8766)
[ WSL2 环境 B: CosyVoice TTS 服务端 ]

---

🎙️ 二、 TTS 服务端：打破并发死锁，榨干 GPU 性能

CosyVoice 的 3s 极速复刻 (Zero-Shot) 效果惊艳，但深度学习模型的推理是同步阻塞的。如果直接在 WebSocket 的 async for 循环中跑推理，会直接卡死整个网络事件循环，导致音频无法实时发送。

核心优化：引入多线程 (threading) 与线程安全队列 (asyncio.Queue)。让 GPU 在后台线程全力推理，主协程只负责把队列里的音频数据瞬间通过网络发走。

server_tts.py (CosyVoice 环境)

运行前请准备一段 3~5 秒的 prompt.wav 人声文件放在同级目录，并修改 PROMPT_TEXT。

import asyncio
import websockets
import argparse
import numpy as np
import os
import sys
import threading

# 将第三方的 Matcha-TTS 加入系统路径，防止找不到模块
ROOT_DIR = os.path.dirname(os.path.abspath(__file__))
sys.path.append('{}/third_party/Matcha-TTS'.format(ROOT_DIR))

from cosyvoice.cli.cosyvoice import AutoModel
from cosyvoice.utils.common import set_all_random_seed

# ================= 极速复刻配置 =================
PROMPT_WAV_PATH = "prompt.wav"
PROMPT_TEXT = "希望你以后能够做的比我还好呢。"  # 替换为音频中实际说的文字

async def tts_handler(websocket, cosyvoice):
    client_ip = websocket.remote_address
    print(f"🔗 ASR/LLM 服务端已连接: {client_ip}")

    try:
        async for message in websocket:
            if isinstance(message, bytes):
                text = message.decode('utf-8').strip()
            else:
                text = message.strip()

            if not text:
                continue

            print(f"🔊 开始极速合成: {text}")
            set_all_random_seed(0)

            # 建立线程安全的队列
            queue = asyncio.Queue()
            loop = asyncio.get_running_loop()

            def run_inference():
                try:
                    # 使用绝对/相对文件路径，防止底层 C++ 解码器崩溃
                    for i in cosyvoice.inference_zero_shot(text, PROMPT_TEXT, PROMPT_WAV_PATH, stream=True):
                        tts_audio_tensor = i['tts_speech'].numpy().flatten()
                        audio_np = (tts_audio_tensor * 32768).astype(np.int16)
                        loop.call_soon_threadsafe(queue.put_nowait, audio_np.tobytes())
                    loop.call_soon_threadsafe(queue.put_nowait, b"TTS_END")
                except Exception as e:
                    loop.call_soon_threadsafe(queue.put_nowait, e)

            # 启动后台独立推理线程，释放主协程网络传输能力
            threading.Thread(target=run_inference, daemon=True).start()

            # 主协程：拿到数据瞬间通过网络发走
            while True:
                chunk = await queue.get()
                if isinstance(chunk, Exception):
                    print(f"⚠️ Zero-Shot 推理报错: {repr(chunk)}")
                    await websocket.send(b"TTS_END")
                    break
                
                await websocket.send(chunk)
                if chunk == b"TTS_END":
                    print(f"✅ 合成完毕")
                    break

    except websockets.exceptions.ConnectionClosed:
        print(f"❌ 客户端断开连接")
    except Exception as e:
        print(f"⚠️ TTS WebSocket 出错: {repr(e)}")

async def main(args):
    print(f"⏳ 正在通过 AutoModel 加载模型: {args.model_dir} ...")
    cosyvoice = AutoModel(model_dir=args.model_dir)
    print(f"✅ 模型加载成功！采样率: {cosyvoice.sample_rate}Hz")

    if not os.path.exists(PROMPT_WAV_PATH):
        print(f"\n❌ 致命错误：找不到参考音频文件 '{PROMPT_WAV_PATH}'！")
        return

    async with websockets.serve(lambda ws: tts_handler(ws, cosyvoice), "0.0.0.0", args.port):
        print(f"🚀 CosyVoice 【多线程极速版】已启动: ws://0.0.0.0:{args.port}")
        await asyncio.Future()

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument('--port', type=int, default=8766)
    parser.add_argument('--model_dir', type=str, default='pretrained_models/CosyVoice2-0.5B')
    args = parser.parse_args()
    asyncio.run(main(args))

---

🧠 三、 ASR + LLM 服务端：构建完美的“生产者-消费者”流水线

最影响语音对话体验的，是 LLM 和 TTS 之间的配合。如果等 LLM 生成完一整段话再去合成语音，延迟会高得无法忍受。

核心优化：

1. 标点符号断句：只要 LLM 吐出逗号、句号等标点，立刻将这半句话截断。
2. 生产者-消费者队列：LLM（生产者）将半句话扔进 text_queue 后，一秒都不等，立刻去生成下一句话。后台专属协程（消费者）盯着队列，拿到半句话就发给 TTS 并将返回的音频透传给客户端。真正实现了“边想、边合成、边播报”。

server_asr.py (Qwen3-ASR 环境)

import asyncio
import websockets
import json
import logging
import numpy as np
import re
from qwen_asr import Qwen3ASRModel
from openai import AsyncOpenAI

logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")

# ================= 1. 配置参数 =================
ASR_MODEL_PATH = "/mnt/d/develop/asr/Qwen3-ASR-1.7B"  # 替换为真实的本地绝对路径
CHUNK_BYTES_THRESHOLD = 16000      

SILICONFLOW_API_KEY = "你的_SiliconFlow_API_KEY"
LLM_MODEL_NAME = "Qwen/Qwen2.5-7B-Instruct"  # 推荐使用 7B 模型，极致响应速度

llm_client = AsyncOpenAI(
    api_key=SILICONFLOW_API_KEY,
    base_url="https://api.siliconflow.cn/v1"
)

TTS_WS_URL = "ws://127.0.0.1:8766"

# ================= 2. 核心服务端逻辑 =================
async def main():
    logging.info(f"正在加载 ASR 模型 {ASR_MODEL_PATH} ...")
    asr = Qwen3ASRModel.LLM(
        model=ASR_MODEL_PATH,
        gpu_memory_utilization=0.5,
        max_new_tokens=32,
        max_model_len=4096
    )
    logging.info("✅ ASR 模型加载完成。等待连接...")

    async def handle_client(websocket):
        client_ip = websocket.remote_address
        logging.info(f"🔗 客户端已连接: {client_ip}")

        state = None
        audio_buffer = bytearray()

        chat_history = [
            {"role": "system",
             "content": "你是一个聪明、幽默的语音助手。由于是以语音形式交互，你的回答必须尽量简短、口语化，不要使用复杂的排版或长篇大论。"}
        ]

        try:
            async for message in websocket:
                if isinstance(message, str):
                    data = json.loads(message)
                    
                    if data.get("event") == "start":
                        state = await asyncio.to_thread(
                            asr.init_streaming_state, unfixed_chunk_num=2, unfixed_token_num=5, chunk_size_sec=2.0
                        )
                        audio_buffer.clear()

                    elif data.get("event") == "stop":
                        if state:
                            logging.info("【探针】收到 VAD 停止信号，处理尾部音频...")
                            if len(audio_buffer) > 0:
                                wav16k = np.frombuffer(audio_buffer, dtype=np.int16).astype(np.float32) / 32768.0
                                await asyncio.to_thread(asr.streaming_transcribe, wav16k, state)
                                audio_buffer.clear()

                            await asyncio.to_thread(asr.finish_streaming_transcribe, state)
                            final_text = state.text
                            state = None

                            logging.info(f"【探针】ASR 最终识别结果: '{final_text}'")
                            await websocket.send(json.dumps({"type": "final", "text": final_text}))

                            if final_text and final_text.strip():
                                chat_history.append({"role": "user", "content": final_text})

                                try:
                                    await websocket.send(json.dumps({"type": "llm_start"}))

                                    stream = await llm_client.chat.completions.create(
                                        model=LLM_MODEL_NAME,
                                        messages=chat_history,
                                        stream=True,
                                        max_tokens=512,
                                        temperature=0.7
                                    )

                                    llm_full_response = ""
                                    sentence_buffer = ""

                                    async with websockets.connect(TTS_WS_URL) as tts_ws:
                                        
                                        # 【核心重构：生产者-消费者模型】
                                        text_queue = asyncio.Queue()

                                        async def tts_worker():
                                            while True:
                                                text = await text_queue.get()
                                                if text is None:  
                                                    break
                                                
                                                await tts_ws.send(text)
                                                
                                                while True:
                                                    audio_bytes = await tts_ws.recv()
                                                    if audio_bytes == b"TTS_END":
                                                        break
                                                    await websocket.send(audio_bytes)

                                        tts_task = asyncio.create_task(tts_worker())

                                        async for chunk in stream:
                                            if chunk.choices and chunk.choices[0].delta.content:
                                                text_chunk = chunk.choices[0].delta.content
                                                llm_full_response += text_chunk
                                                sentence_buffer += text_chunk

                                                await websocket.send(json.dumps({"type": "llm_partial", "text": text_chunk}))

                                                # 遇到逗号也断句，让 TTS 能最快拿到首句进行合成
                                                if re.search(r'[，。！？、,!?\n]', text_chunk):
                                                    if sentence_buffer.strip():
                                                        text_queue.put_nowait(sentence_buffer.strip())
                                                        sentence_buffer = ""

                                        if sentence_buffer.strip():
                                            text_queue.put_nowait(sentence_buffer.strip())

                                        text_queue.put_nowait(None)
                                        # 等待最后一句音频发送完毕
                                        await tts_task

                                    chat_history.append({"role": "assistant", "content": llm_full_response})
                                    await websocket.send(json.dumps({"type": "llm_stop"}))
                                    logging.info(f"【探针】整轮交互完成: {llm_full_response}")

                                except Exception as e:
                                    logging.error(f"调用失败: {e}")
                                    await websocket.send(json.dumps({"type": "llm_partial", "text": f"\n[系统异常]\n"}))
                                    await websocket.send(json.dumps({"type": "llm_stop"}))

                elif isinstance(message, bytes):
                    if state:
                        audio_buffer.extend(message)
                        if len(audio_buffer) >= CHUNK_BYTES_THRESHOLD:
                            wav16k = np.frombuffer(audio_buffer, dtype=np.int16).astype(np.float32) / 32768.0
                            audio_buffer.clear()
                            await asyncio.to_thread(asr.streaming_transcribe, wav16k, state)
                            if state.text:
                                await websocket.send(json.dumps({"type": "partial", "text": state.text}))

        except websockets.exceptions.ConnectionClosed:
            logging.warning(f"❌ 客户端断开连接")
        finally:
            if state:
                await asyncio.to_thread(asr.finish_streaming_transcribe, state)

    async with websockets.serve(handle_client, "0.0.0.0", 8765, ping_interval=None):
        logging.info("🚀 ASR+LLM 语音后端已启动: ws://0.0.0.0:8765")
        await asyncio.Future()

if __name__ == "__main__":
    asyncio.run(main())

---

🎧 四、 客户端：消灭“回音套娃”与声学物理延迟

当使用外放喇叭时，麦克风会把喇叭的声音重新录进去，导致 AI 不断跟自己说话（回音死循环）。
你可能会想到在代码里加一个 is_ai_speaking 的标志位来闭麦。但实际上，网络传输极快，当代码收到 llm_stop 标志时，喇叭物理层面的振膜可能还需要 1~2 秒才能把缓存里的声音播放完！ 此时开启麦克风，依然会录入尾音。

核心优化：引入专用的同步播放队列 (play_queue)，并在播放结束指令到来时，强制加入 1.5 秒的声学垫片 (Acoustic Tail Pad)，等待声卡缓存物理播放完毕，同时等待房间内的声波回音彻底消散，再开启麦克风。

client_vad.py (Windows 环境)

import pyaudio
import numpy as np
import torch
import asyncio
import websockets
import json
import collections

# ================= 1. 配置参数 =================
WS_URL = "ws://172.27.27.243:8765" # 替换为你的服务器IP
SAMPLE_RATE = 16000
CHUNK_SIZE = 512
FORMAT = pyaudio.paInt16
CHANNELS = 1

PRE_SPEECH_PAD_FRAMES = 15  

# ================= 2. 加载 VAD 模型 =================
print("⏳ 正在从本地加载 Silero VAD 模型...")

VAD_LOCAL_PATH = r"D:\develop\asr\silero-vad-master"

vad_model, utils = torch.hub.load(
    repo_or_dir=VAD_LOCAL_PATH,  
    model='silero_vad',
    source='local',              
    force_reload=False,
    onnx=False
)
(get_speech_timestamps, save_audio, read_audio, VADIterator, collect_chunks) = utils
vad_iterator = VADIterator(
    vad_model,
    threshold=0.35,                
    min_silence_duration_ms=800,   
    speech_pad_ms=100              
)

TTS_SAMPLE_RATE = 22050

# ================= 3. 核心客户端逻辑 =================
async def mic_client():
    p = pyaudio.PyAudio()
    player_stream = p.open(format=pyaudio.paInt16, channels=1, rate=TTS_SAMPLE_RATE, output=True)

    try:
        async with websockets.connect(WS_URL, ping_interval=None) as websocket:
            print("✅ 成功连接到服务端！")

            audio_queue = asyncio.Queue()  # 录音队列
            play_queue = asyncio.Queue()   # 专门用于同步播放的队列
            
            loop = asyncio.get_running_loop()
            ring_buffer = collections.deque(maxlen=PRE_SPEECH_PAD_FRAMES)

            # 状态标志，用于判断 AI 是否正在说话
            state_flags = {"is_ai_speaking": False}

            def audio_callback(in_data, frame_count, time_info, status):
                loop.call_soon_threadsafe(audio_queue.put_nowait, in_data)
                return (None, pyaudio.paContinue)

            p2 = pyaudio.PyAudio()
            stream = p2.open(format=FORMAT, channels=CHANNELS, rate=SAMPLE_RATE, input=True,
                            frames_per_buffer=CHUNK_SIZE, stream_callback=audio_callback)
            stream.start_stream()
            print("\n🎤 麦克风已激活，请开始说话...\n")

            # ---------- 1. 录音与发送协程 ----------
            async def capture_and_send():
                is_speaking = False
                while True:
                    data = await audio_queue.get()

                    # 如果 AI 正在说话，直接丢弃麦克风声音（硬件级闭麦）
                    if state_flags["is_ai_speaking"]:
                        ring_buffer.clear()  
                        if is_speaking:
                            is_speaking = False
                            await websocket.send(json.dumps({"event": "stop"}))
                        continue  

                    audio_float32 = torch.from_numpy(np.frombuffer(data, dtype=np.int16).copy()).float() / 32768.0
                    speech_dict = vad_iterator(audio_float32, return_seconds=False)

                    if speech_dict and 'start' in speech_dict and not is_speaking:
                        is_speaking = True
                        await websocket.send(json.dumps({"event": "start"}))
                        for buffered_data in ring_buffer:
                            await websocket.send(buffered_data)
                        ring_buffer.clear()

                    elif speech_dict and 'end' in speech_dict and is_speaking:
                        is_speaking = False
                        await websocket.send(data)
                        await websocket.send(json.dumps({"event": "stop"}))

                    if is_speaking:
                        await websocket.send(data)
                    else:
                        ring_buffer.append(data)

            # ---------- 2. 接收网络数据协程 ----------
            async def receive_results():
                async for message in websocket:
                    if isinstance(message, str):
                        res = json.loads(message)
                        if res["type"] == "partial":
                            print(f"\r🗣️ [你]: {res['text']}    ", end="", flush=True)
                        elif res["type"] == "final":
                            print(f"\r🎯 [你]: {res['text']}    \n", flush=True)
                        elif res["type"] == "llm_start":
                            state_flags["is_ai_speaking"] = True
                            print(f"🤖 [AI]: ", end="", flush=True)
                        elif res["type"] == "llm_partial":
                            print(res["text"], end="", flush=True)
                        elif res["type"] == "llm_stop":
                            # 往播放队列塞入结束符
                            await play_queue.put(b"STOP_PLAYBACK")

                    elif isinstance(message, bytes):
                        await play_queue.put(message)

            # ---------- 3. 绝对同步的音频播放协程 ----------
            async def play_audio_task():
                while True:
                    chunk = await play_queue.get()
                    if chunk == b"STOP_PLAYBACK":
                        # ========================================================
                        # 【终极防套娃核心】：增加 1.5 秒的硬件级声学垫片
                        # 强行等待声卡底层物理播放完毕，并等待房间回声消散
                        # ========================================================
                        await asyncio.sleep(1.5) 
                        
                        # 清空在这 1.5 秒内麦克风偷偷录入的废料音频
                        while not audio_queue.empty():
                            audio_queue.get_nowait()
                        ring_buffer.clear() 
                        
                        state_flags["is_ai_speaking"] = False
                        print("\n\n🎤 请继续说话...", flush=True)
                    else:
                        # 阻塞式顺序播放
                        await asyncio.to_thread(player_stream.write, chunk)

            await asyncio.gather(capture_and_send(), receive_results(), play_audio_task())

    except Exception as e:
        print(f"\n⚠️ 发生异常: {e}")
    finally:
        if 'stream' in locals() and stream.is_active():
            stream.stop_stream()
            stream.close()
        if 'p2' in locals():
            p2.terminate()
        if 'player_stream' in locals() and player_stream.is_active():
            player_stream.stop_stream()
            player_stream.close()
        if 'p' in locals():
            p.terminate()

if __name__ == "__main__":
    try:
        asyncio.run(mic_client())
    except KeyboardInterrupt:
        print("\n👋 客户端已关闭。")

总结

至此，一套完整的流式语音大模型底座已搭建完毕。它不仅拥有毫秒级的 VAD 响应断句，还实现了多并发异步的文本/音频吞吐，并且彻底杜绝了外放设备的声学回声。

这套工程架构可以说是目前搭建本地 LLM 语音助手的最优解之一。无论是替换更强大的 ASR，还是接入更聪明的云端大模型，这套极低耦合度的微服务底座都能完美兼容！