手把手教你做一個天貓精靈(一)

語言: CN / TW / HK

theme: github

如今,智能家居的話題越來越火,物聯網已經融入了我們生活。最近閒在家裏瞭解了一下這方面的背景知識,自己動手做了一個類似天貓精靈的物聯網智能終端。於是打算出一個教程分享一下我的研究成果。

硬件準備

  • 一台能連上網的電腦(系統最好是Windows 10)

背景知識

物聯網智能終端

類似於天貓精靈這類產品我們都把它叫做智能終端(或者智能音箱)。它能通過用户之間的語音互動對智能家居進行命令控制,也能提供一些內置的軟件服務,比如語音提醒和QQ音樂等。因此,我把它總結出以下幾個功能模塊。

功能模塊

目前,市面上除了天貓精靈,還有百度小度、小米小愛、Google Home和比亞迪小迪等。

語音處理術語

智能終端運用了大量的語音處理方面的人工智能算法,這些算法可軟件實現也可以硬件實現,主要包括語音識別(Automated Speech Recognition,ASR)、語音合成(Text To Speech,TTS)和語音喚醒(Keyword Spotting,KWS)。

一般的流程是這樣的,先通過語音喚醒捕捉用户的Keyword,當捕捉到了才給用户錄音並作語音識別,然後根據用户的意圖調用相應的API並通過語音合成給用户適當的反饋。據我觀察至少有三種API調用方式,或者説是對外調用的接口,總結如下:

  • MQTT消息:智能終端操作硬件一般採用IBM定義的MQTT協議,這種消息協議類似 Kafka,但是它更適應網絡環境不穩定的場景,比如多了服務質量(QoS)和遺言機制(Last Will),這部分內容以後介紹。
  • 藍牙接口:還有一種是通過藍牙Mesh直接和硬件相連,不需要聯網就能直接操作硬件。
  • HTTP接口:這種接口一般是提供軟件服務的,比如QQ音樂、墨跡天氣和喜馬拉雅廣播

快速開始

考慮到可能會接入很多應用功能,所以本項目是通過Python開發的。新建一個工程,並設置好環境,然後下載這個包:

shell pip install fubuki-iot

提醒:如果下載不了可以換源,也可以在GitHub上下載,倉庫地址點這裏 。同時也歡迎PR。

然後創建一個程序入口,我把它命名為 app.py, 然後調用 Terminalrun函數,如下:

```python from iot import Terminal

if name == 'main': Terminal.run() ```

當然現在這個程序是跑不起來的,因為沒有相應的配置。新建一個目錄叫 resources ,然後再創建一個配置文件 .env ,注意前面有一個點。接着,把剛剛新建的 resource 文件的路徑寫到這個配置文件中,如下:

text RESOURCE_PATH=你的resources的路徑

這時候,你的項目目錄應該是這樣的:

目錄結構

原則上就可以運行了,但是像上文所説需要語音處理相關的能力,所以我先暫時借用了百度智能雲的AI能力。這個平台可以免費提供半年的人工智能服務,所以先去申請以下相關服務。

進入百度智能雲的官網,依次點擊“產品”-“人工智能”-“語音技術”,然後申請這個AI服務。之後就可以在控制枱中找到自己的API Key和Secret Key,如圖所示:

控制枱

然後在剛才的 .env文件中追加這兩條信息,並預留一個BAIDU_ACCESS_TOKEN字段,如下所示:

text BAIDU_API_KEY=你的API Key BAIDU_SECRET_KEY=你的Secret Key BAIDU_ACCESS_TOKEN=

提醒:由於百度API是要生成access token使用的,這個token是有時效的,所以這裏預留可以方便以後不用重複申請,但是過了時效還是要把這個字段置空,就像現在這個樣子。

當然,百度API試用結束後還是會收費,如果你不想收費可以自己訓練模型來替代它,這個以後會介紹。

現在就可以運行這個程序了。運行結果如圖所示:

運行

按下“F”鍵就可以喚醒終端,它會迴應“我在,你説”。然後等它開始錄音就可以説“你好”,它會迴應“在的”。至此,一個極簡的智能終端就做好了。

開啟語音喚醒

雖然智能終端的樣子已經有了,但是距離天貓精靈還差遠了。現在,就讓我們開啟它的語音喚醒功能。

這個終端內置了一個PocketSphinx的語音喚醒功能。這個工具是C寫的,因此要在Windows環境下需要對應的編譯器,官方推薦的是Visual C++。

首先點這裏下載Visual Studio Installer,然後安裝,接着打開以後選擇安裝“單個組件”,然後勾選這三個組件“Windows 10 SDK”、“Windows 通用 CRT SDK”以及“MSVC v140 - VS 2015 C++ 生成工具(v14.00)”,下載完即可。這個環境變量無需配置。

安裝目錄

注意:有可能碰到“無法打開包括文件: “windows.h”: No such file or directory”的問題,這個需要將<windows.h> 頭文件放在指定目錄,具體自行百度或者谷歌。

然後下載swigwin,這個軟件是讓python調用C/C++編譯的軟件的,點這裏下載,下載完成後解壓,並配置環境變量,然後重啟生效。

最後,安裝PocketSphinx。

shell pip install pocketsphinx 如果沒有報錯就説明成功了,有報錯可能和C語言環境有關,這個需要具體問題具體分析了。

然後在 .env 文件中追加兩行配置:

TERMINAL_MODE=0 DEVICE_REC=PocketsphinxRecorder 接着啟動程序可以通過對他説“hello”或者“hi”喚醒它。

編寫第一個程序

接下來,我們要開始向終端裏面添加功能。類似天貓精靈的定時提醒功能,我們也做一個提醒功能。

首先創建一個包,命名為 mods,我們接下來寫的模組都將放在這個包裏。新建一個文件命名為timer.py,然後在文件裏新建一個語義模型,這個模型繼承了SemanticsModel,其目的是為了匹配用户的命令,如下所示:

python @SemanticsGroup.add_model class TimerSemanticsModel(SemanticsModel): code = 'timer' frm = SemanticsFromEnum.USER topic = '' regex = '(.*)後提醒我(.*)' regex_num = 3 redirect = SemanticsRedirectEnum.ACOUSTICS func: SemanticsFunc = timer_semantics_func 字段説明如下: - code:語義模型的標識,這裏隨便填,只用來區分。 - frm:語義來源,這裏是用户 - topic:因為來自用户,所以這裏不需要填 - regex:正則表達式,用來匹配用户的命令,如果匹配命中則使用這個語義模型,否則繼續匹配下一個語義模型,如果都沒有命中則發送兜底的語音迴應用户 - regex_num:上面正則表達式的groups()個數,至少為1,表示全量,具體參考Python正則表達式相關文檔,這個是用來抽取參數的 - redirect:重定向,這裏不需要處理硬件,只是返回語音,所以是 ACOUSTICS - func:用來處理這個命令的回調函數

重點就在這個回調函數,它是處理語義的關鍵。在當前環境下,它接受一個列表,即正則表達式匹配的groups,我們可以通過數組下標拿到想要的參數,並作後續處理。比如:

python def timer_semantics_func(*args) -> FunctionDeviceModel: sentence = args[0] # 獲取全文 timespan: str = args[1] # 第一個參數,時間 content = args[2] # 第二個參數,內容 ...

此外它返回一個功能設備模型FunctionDeviceModel,它告知終端要如何處理,即如何返回給用户信息。其字段如下。 ```python class FunctionDeviceModel(BaseModel): """ 功能設備模型,既是用户發送指令的封裝,也是語義轉換模塊處理後的產物。 """

# 對應的semantics_model的code
smt_code: Optional[str]

# raw標誌位,如果為True則data為str,否則為dict
is_raw: bool

# 執行成功後返回的語音提示,如果重定向到MESSAGE則作為語音提示,
# 如果是重定向到ACOUSTICS則也可以表示返回內容
acoustics: str

# 數據,只有當is_raw為False才為字典,用於展示詳細內容
data: Union[str, Dict[str, str]]

```

因此,做一個提醒功能就是判斷用户的語句是否正確,如果正確則返回正確的語音信息,否則告訴用户聽不懂,具體如下: python def timer_semantics_func(*args) -> FunctionDeviceModel: ... if not is_ok: return FunctionDeviceModel( smt_code='timer', is_raw=True, acoustics="抱歉我沒聽清", data="" ) else: scheduler.start() return FunctionDeviceModel( smt_code='timer', is_raw=True, acoustics=f"好的,我會在{timespan}後提醒你{content}", data="" )

有正確的處理和返回還不夠,還需要在時間到了的時候給用户語音提醒,這實現也很簡單,只要從Context中獲取相應的API就可以了,具體如下:

python path = Context.tts_processor.tts("您好,時間到了,請" + content) Context.player.play(path)

實現定時的方案有很多種,我採用了APScheduler實現了這個功能,下面僅供參考:

```python def ch2i(ch: str): try: # 嘗試直接轉換 return int(ch) except Exception: pass

_dict = {
    "一": 1,
    "二": 2,
    "三": 3,
    "四": 4,
    "五": 5,
    "六": 6,
    "七": 7,
    "八": 8,
    "九": 9,
    "十": 10,
    "百": 100
}
if ch.startswith("十") or ch.startswith("百"):
    ch = "一" + ch
_buf = list()
_res = 0
for i in range(len(ch)):
    if len(_buf) == 0: 
        _buf.append(_dict[ch[i]]) 
    else: 
        _res += _buf.pop() * _dict[ch[i]]
return _res + (_buf[0] if len(_buf) else 0)

def timer_semantics_func(*args) -> FunctionDeviceModel: sentence = args[0] # 獲取全文 timespan: str = args[1] # 獲取時間 content = args[2] # 獲取內容 is_ok = True scheduler = BackgroundScheduler()

def _job():
    path = Context.tts_processor.tts("您好,時間到了,請" + content)
    Context.player.play(path)
try:
    if timespan.endswith("秒"):
        target = ch2i(timespan.split("秒")[0])
        scheduler.add_job(_job, 'date', run_date=datetime.datetime.now() + datetime.timedelta(seconds=target))
    elif timespan.endswith("分鐘"):
        target = ch2i(timespan.split("分鐘")[0])
        scheduler.add_job(_job, 'date', run_date=datetime.datetime.now() + datetime.timedelta(minutes=target))
    elif timespan.endswith("小時"):
        target = ch2i(timespan.split("小時")[0])
        scheduler.add_job(_job, 'date', run_date=datetime.datetime.now() + datetime.timedelta(hours=target))
    else:
        # 只處理到小時,更大的單位不處理了
        raise RuntimeError("Fail to match")
except Exception as e:
    logger.error("Fail to translate " + e.__str__())
    is_ok = False
if not is_ok:
    return FunctionDeviceModel(
        smt_code='timer',
        is_raw=True,
        acoustics="抱歉我沒聽清",
        data=""
    )
else:
    scheduler.start()
    return FunctionDeviceModel(
        smt_code='timer',
        is_raw=True,
        acoustics=f"好的,我會在{timespan}後提醒你{content}",
        data=""
    )

@SemanticsGroup.add_model class TimerSemanticsModel(SemanticsModel): code = 'timer' frm = SemanticsFromEnum.USER topic = '' regex = '(.)後提醒我(.)' regex_num = 3 redirect = SemanticsRedirectEnum.ACOUSTICS func: SemanticsFunc = timer_semantics_func 最後,在程序入口處加載這個包: from iot import Terminal Terminal.load_models('mods.timer')

if name == 'main': Terminal.run() ```

現在,再啟動程序,對它説“五分鐘後提醒我打掃衞生”它就會迴應,並在五分鐘後通過語音播報提醒我們。

本章初步介紹了物聯網領域的相關概念,然後跑通了一個簡單的智能終端的程序。下一章我們會進一步完善這個程序,並把它做成硬件,離我們的目標更近一步!