给本地设备搭建一个云端语音助手

news2024/12/22 10:13:52

概述

本语音助手实现了从关键词唤醒 (KWS) 到语音识别 (ASR) 再到自然语言理解 (NLU) 的完整流程。该系统可以通过监听用户的音频输入，检测指定的关键词，并将用户的语音转换为文本，最后与预设的命令进行匹配，执行相应的操作（具体实现请参考main.py），为你的设备配置远程语音助手。

流程步骤

请添加图片描述

1. 初始化配置

在启动时，语音助手会加载以下配置：

KWS关键词列表 (words_kws.txt)：用于检测唤醒词，如 “小新小新” 或 “小爱同学”。
ASR热词及其权重 (words_asr.txt)：用于优化语音识别的热词列表及其权重，如 “阿里巴巴 20”。
NLU命令匹配句子 (words_nlu.txt)：预定义的命令句子列表，如 “打开灯”、“关闭灯” 等。

2. 建立WebSocket连接

系统通过以下两个WebSocket连接与KWS和ASR服务器进行交互：

KWS WebSocket连接：用于传输音频流并接收关键词检测结果。
ASR WebSocket连接：用于传输音频流并接收语音识别结果。

3. 关键词唤醒 (KWS)

当系统启动后，它首先处于关键词唤醒 (KWS) 模式。系统从麦克风获取音频数据，并将这些数据通过WebSocket发送到KWS服务器。服务器实时分析音频流，如果检测到预定义的唤醒词，系统将从KWS模式切换到ASR模式，并准备进行语音识别。

4. 语音识别 (ASR)

在ASR模式下，系统继续从麦克风获取音频数据，并将其发送到ASR服务器。ASR服务器会将音频数据转换为文本，并实时返回识别结果。识别结果分为两种：

在线识别结果：实时返回的识别文本片段。
离线识别结果：最终的完整句子识别结果。

5. 自然语言理解 (NLU)

当ASR返回完整的离线识别结果后，系统会将该文本发送到NLU服务器进行分析。NLU服务器会将该文本与预定义的命令进行匹配，并返回最匹配的命令及其得分。

6. 执行命令

系统根据NLU的匹配结果，输出相应的命令和匹配得分。如果匹配的命令足够明确，系统可以执行相应的操作，如控制设备的开关等。

7. 循环检测

处理完成后，系统重新进入KWS模式，等待下一个唤醒词的检测，整个流程循环进行。

结果展示：

请添加图片描述

远程服务搭建联系作者邮箱：605686962@qq.com

import sounddevice as sd
import websockets
import asyncio
import json
import ssl 
import queue 
import numpy as np 
import requests

class Speech_Assistant():
    """
    Speech Assistant class for handling Keyword Spotting (KWS), Automatic Speech Recognition (ASR), 
    and Natural Language Understanding (NLU) tasks using WebSocket connections.
    """

    def __init__(self,
                 uri_kws="ws://0.0.0.0:10094",
                 uri_asr="wss://0.0.0.0:10095",
                 uri_nlu="http://0.0.0.0:10096",
                 words_kws=['小新小新', '小爱同学'],
                 words_asr={
   "鸿合科技":20},
                 words_nlu=[]) -> None:
        """
        Initializes the Speech Assistant with URLs and keyword configurations.
        
        Args:
            uri_kws (str): URL for the KWS WebSocket server.
            uri_asr (str): URL for the ASR WebSocket server.
            uri_nlu (str): URL for the NLU HTTP server.
            words_kws (list): List of keywords for KWS.
            words_asr (dict): Dictionary of hotwords and their weights for ASR.
            words_nlu (list): List of sentences to compare for NLU.
        """
        self.uri_kws = uri_kws
        self.uri_asr = uri_asr
        self.uri_nlu = uri_nlu
        self.words_kws = words_kws
        self.words_asr = words_asr
        self.words_nlu = words_nlu
        self.sample_rate = 16000
        self.state = 'kws'
        self.assistant = "unknown"

    async def init_websocket_kws(self):
        """Initialize WebSocket connection for KWS."""
        self.ws_session_kws = await websockets.connect(self.uri_kws)
        print("Connected to websocket_kws")

    async def init_websocket_asr(self):
        """Initialize WebSocket connection for ASR."""
        ssl_context = ssl.SSLContext()
        ssl_context.check_hostname = False
        ssl_context.verify_mode = ssl.CERT_NONE
        self.ws_session_asr = await websockets.connect(self.uri_asr, subprotocols=["binary"], ping_interval=None, ssl=ssl_context)
        print("Connected to websocket_asr")

    async def close_websockets(self):
        """Close WebSockets ."""
        if self.ws_session_kws:
            await self.ws_session_kws.close()
            print("Closed websocket_kws")
        if</