基于 LLM 与 ASR 从零搭建 Windows 智能语音助手

基于 LLM 与 ASR 从零搭建 Windows 智能语音助手

项目背景与架构设计

随着大语言模型（LLM）技术的快速发展，构建本地化的智能助手已成为可能。传统的命令行工具或简单的脚本已无法满足用户对自然交互的需求。本项目旨在利用 Python 生态中的成熟组件，在 Windows 系统上搭建一个具备语音输入、语义理解能力的智能体（Agent）。

核心架构

系统主要由以下三个模块组成：

1. 语音识别模块 (ASR)：负责将麦克风采集的音频流实时转换为文本。本方案选用 Sherpa-onnx，因其支持 ONNX Runtime，在 CPU 环境下也能保持高效的推理速度。
2. 大语言模型模块 (LLM)：负责接收识别后的文本，进行语义分析并生成回复。本方案采用 Agently 框架来管理 Agent 状态及 Workflow。
3. 交互控制模块：负责协调 ASR 与 LLM 的数据流转，实现'语音 - 文本 - 语音'的闭环。

环境准备

在开始编码之前，请确保您的开发环境满足以下要求：

• 操作系统：Windows 10/11
• Python 版本：建议 Python 3.9 及以上
• 依赖管理：推荐使用 venv 创建虚拟环境以隔离依赖

安装基础依赖库：

pip install sounddevice sherpa_onnx Agently python-dotenv

一、接入语音识别 (ASR)

语音识别是智能助手的耳朵。我们使用 Sherpa-onnx 提供的在线识别接口，它支持端点检测（Endpoint Detection），能够自动判断用户何时说完一句话。

1. 模型下载

Sherpa-onnx 需要预训练模型文件。对于中文场景，推荐使用 Paraformer 模型。您可以从官方仓库或镜像站下载模型包，解压后需包含 encoder.int8.onnx, decoder.int8.onnx, tokens.txt 等文件。

2. 代码实现

创建一个名为 microphone_asr.py 的文件，用于封装语音处理逻辑。该类负责初始化识别器、监听麦克风流并进行实时转写。

#!/usr/bin/env python3
import os.path
import queue
from typing import Generator
import sounddevice as sd
import sys
import logging
import sherpa_onnx

logger = logging.getLogger(__name__)

class AsrHandler:
    def __init__(self, model_path, debug=False):
        .recognizer = 
        .sentence_q = queue.Queue()
        .init_recognizer(model_path)
        .debug = debug

     ():
        
        encoder = os.path.join(model_path, )
        decoder = os.path.join(model_path, )
        tokens = os.path.join(model_path, )
        
        .recognizer = sherpa_onnx.OnlineRecognizer.from_paraformer(
            tokens=tokens,
            encoder=encoder,
            decoder=decoder,
            num_threads=,
            sample_rate=,
            feature_dim=,
            enable_endpoint_detection=,
            rule1_min_trailing_silence=,
            rule2_min_trailing_silence=,
            rule3_min_utterance_length=,
        )


     ():
        devices = sd.query_devices()
         (devices) == :
            logger.info()
            sys.exit()
        logger.info()

     () -> Generator:
        logger.info()
        sample_rate = 
        samples_per_read = ( * sample_rate)
        stream = .recognizer.create_stream()
        last_result = 
        segment_id = 
        :
             sd.InputStream(channels=, dtype=, samplerate=sample_rate)  s:
                 :
                    samples, _ = s.read(samples_per_read)
                    samples = samples.reshape(-)
                    stream.accept_waveform(sample_rate, samples)
                     .recognizer.is_ready(stream):
                        .recognizer.decode_stream(stream)

                    is_endpoint = .recognizer.is_endpoint(stream)
                    result = .recognizer.get_result(stream)

                     result  (last_result != result):
                        last_result = result
                         .debug: logger.info()
                    
                     is_endpoint:
                         result:
                             .debug: logger.info()
                            segment_id += 
                             result
                        .recognizer.reset(stream)
         sd.PortAudioError  e:
            logger.exception()

 __name__ == :
    :
        asr_gen = AsrHandler(model_path=).handle()
         chunk  asr_gen:
            (, chunk)
     KeyboardInterrupt:
        ()