概述
本语音助手实现了从关键词唤醒 (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</