概述
本语音助手实现了从关键词唤醒 (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 requestsclass 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_kwsself.uri_asr = uri_asrself.uri_nlu = uri_nluself.words_kws = words_kwsself.words_asr = words_asrself.words_nlu = words_nluself.sample_rate = 16000self.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 = Falsessl_context.verify_mode = ssl.CERT_NONEself.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</
高可用集群)