代码收藏家技术教程 2025-01-16

在树莓派5上本地部署端到端智能语音聊天项目

项目介绍

项目总体采用_STT模型+本地大语言模型+TTS模型_三步进行实现，为了能够实现在树莓派等算力资源有限的边缘端设备部署，并使项目延迟尽可能降低，只能选择轻量化的TTS模型与参数较少或者经过量化后的大语言模型进行实现。本项目中每一次问答的处理时间可以控制在十秒以内。

硬件及模型选择

硬件设备

Raspberry pi5(8GB内存)，带声卡的扬声器，带麦克风的摄像头，显示屏（其实可以不需要）

模型选择

功能	模型名称
语音转文字	Paraformer-large
大语言模型	Qwen2.5-0.5B-Instruct
文字转音频(TTS)	Piper-TTS

项目具体实现步骤

模型下载与环境搭建

语音转文本模型和大语言模型可以直接在魔搭社区找到并下载到本地一个名为chat的项目文件夹下

至于TTS模型，可以进入piper/VOICES.md at master · rhasspy/piper页面，下滑到最后找到

任意选择一个声音（这里我选择medium）下载model与config，一样保存到chat文件夹中

之后进行环境的搭建，新建一个虚拟环境，输入如下指令

pip install piper-tts modelscope transformers funasr torch torchvision torchaudio soundcard accelerate pynput wave\
pygame soundfile

可能会出现没安装全部环境的情况（因为我也不太记得了），后续根据终端输出进行剩余库的安装就可以了

后面运行程序的时候可能会出现一个关于datasets库的报错（具体报的啥我也不太记得了），原因是datasets库的版本不兼容，只需要卸载原来的datasets库安装2.16.0的datasets库就可以解决

pip uninstall datasets
pip install datasets==2.16.0

_Piper_调用函数的编写

由于原github项目中没有给出_Piper_模型在python中调用的示例代码，我在查看库函数后写了一个简单的_Piper_的调用函数，并封装成_Piper_tts.py_文件

# 保存为项目中的piper_tts.py文件
import wave
from piper import PiperVoice

# audio_path为生成语音的保存路径，text为生成语音的文本
def generate_audio(audio_path, text):
    # 初始化 Piper 实例
    pipers = PiperVoice.load('zh_CN-huayan-medium.onnx', 'zh_zh_CN_huayan_medium_zh_CN-huayan-medium.onnx.json')
 
    # 合成语音
    output_wav_path = audio_path
    with wave.open(output_wav_path, 'wb') as wav_file:
        wav_file.setnchannels(1)
        wav_file.setsampwidth(2)
        wav_file.setframerate(pipers.config.sample_rate)

        pipers.synthesize(text, wav_file=wav_file)

音频文件播放函数的编写

在保存TTS模型生成的音频文件后，我希望将它播放出来，从而形成语音问答的流程，因此借助pygame库进行音频文件播放函数的编写，并保存到项目文件下的_play_audio.py_文件

import pygame

def play_audio(audio_path):
    pygame.mixer.init()
    pygame.mixer.music.load(audio_path)
    pygame.mixer.music.play()
    while pygame.mixer.music.get_busy():
        continue

音频录制函数的编写

首先查看设备的麦克风接入情况，在终端中输入

arecord -l

如果返回消息如上说明已经检查到有麦克风接入

在有麦克风接入的情况下，我试着通过pyaudio和sounddevice库进行录制音频，但是均报错没有可用的麦克风，想了很多办法还是没有解决，后来想到以前用过soundcard库来录音，就试着通过这个库试试能不能检测到接入的麦克风，结果居然真能检测到！

于是决定用soundcard库来写音频录制函数。函数保存到项目文件夹下的_record.py_

import soundcard as sc
import numpy as np
import soundfile as sf
import time
import threading

class AudioRecorder:
    def __init__(self, sample_rate=16000, block_size=1024, recording_duration=3):
        self.sample_rate = sample_rate
        self.block_size = block_size
        self.recording_duration = recording_duration
        self.frames = []
        self.recording = False
        self.recorder_thread = None

    def start_recording(self):
        self.frames = []
        self.recording = True
        print("开始录音...")

        # 打开默认的输入设备
        input_device = sc.default_microphone()
        recorder = input_device.recorder(samplerate=self.sample_rate, channels=1, blocksize=self.block_size)

        with recorder:
            start_time = time.time()
            while self.recording and (time.time() - start_time) < self.recording_duration:
                # 开始录制并获取数据
                data = recorder.record(numframes=self.block_size)
                data = data.squeeze()
                self.frames.append(data)
                time.sleep(0.01)  # 防止CPU占用过高

        self.stop_recording()

    def stop_recording(self):
        self.recording = False
        print("停止录音...")

        # 将录制的数据合并成一个数组
        audio_data = np.concatenate(self.frames)

        # 保存音频文件
        file_path = 'recorded_audio.wav'
        sf.write(file_path, audio_data, self.sample_rate)
        print(f"录音已保存到 {file_path}")

    def run(self):
        # 启动录音
        self.recorder_thread = threading.Thread(target=self.start_recording)
        self.recorder_thread.start()

if __name__ == "__main__":
    recorder = AudioRecorder()
    recorder.run()

写完上述辅助函数，就可以开始编写主函数了

由于在项目中我想实现：先预加载好所有模型，再通过监听键盘来判断是否开始语音问答，进行一轮的语音问答后继续进入监听状态，当再次按下键盘特定按键后，开始新一轮的语音问答。因此需要pynput库进行监听键盘

最后编写的主函数如下

from modelscope.pipelines import pipeline
from modelscope.utils.constant import Tasks
from modelscope.utils.hf_util import AutoModelForCausalLM, AutoTokenizer
import time
from pynput.keyboard import Listener, Key
from play_audio import play_audio
from piper_tts import generate_audio
from record import AudioRecorder

# 预加载模型
inference_pipeline = pipeline(
    task=Tasks.auto_speech_recognition,
    model='speech_paraformer-large_asr_nat-zh-cn-16k-common-vocab8404-pytorch'
)

model_path = "Qwen2___5-0___5B-Instruct"
model = AutoModelForCausalLM.from_pretrained(
    model_path,
    torch_dtype="auto",
    device_map="auto"
)
tokenizer = AutoTokenizer.from_pretrained(model_path)

def process_audio():
    # 创建录音对象
    recorder = AudioRecorder()
    recorder.start_recording()

    # 等待录音结束
    time.sleep(3)  # 录音3秒
    recorder.stop_recording()

    start = time.time()
    
    # 语音识别
    rec_result = inference_pipeline(input='recorded_audio.wav')
    question = rec_result[0]['text']
    print(f"识别的问题: {question}")

    # 生成回答
    prompt = f"请根据以下内容给出一段简短的回答：{question}"
    messages = [
        {"role": "system", "content": "你是千问，一位说话简明扼要机器人，你要根据问题给出相应的回答"},
        {"role": "user", "content": prompt}
    ]
    text = tokenizer.apply_chat_template(
        messages,
        tokenize=False,
        add_generation_prompt=True
    )
    model_inputs = tokenizer([text], return_tensors="pt").to(model.device)

    generated_ids = model.generate(
        **model_inputs,
        max_new_tokens=512
    )
    generated_ids = [
        output_ids[len(input_ids):] for input_ids, output_ids in zip(model_inputs.input_ids, generated_ids)
    ]

    response = tokenizer.batch_decode(generated_ids, skip_special_tokens=True)[0]
    print(f"生成的回答: {response}")

    # 生成音频
    audio_path = 'output.wav'
    generate_audio(audio_path, response)
    end = time.time()
    print(f"处理时间: {end - start} 秒")

    # 播放音频
    play_audio(audio_path)

def on_press(key):
    if key == Key.enter:
        process_audio()
    # 如果按下esc键，则终止程序
    if key == Key.esc:
        exit()

def main():
    print("按回车键开始录音和处理...")
    with Listener(on_press=on_press) as listener:
        listener.join()

if __name__ == "__main__":
    main()