鑽石級贊助商 - KKBOX 帶你打造具備 NLP 功能的 Telegram Bot（下）

打造具備 NLP 功能的 Telegram Bot（下）

上一篇文章已經讓 Chatbot 有了許多自然應答的功能，透過 OLAMI 預設的 IDS 對話模組也能處理多數詢問，但如果想要讓內容更多元，就需要仰賴內容供應商的資料來豐富對話內容，例如：希望它幫我找動漫歌曲，同時列出漂亮的專輯封面圖供選擇

這一篇內容會介紹要怎麼整合更多內容進 Chatbot，同時也會介紹如何將 Chatbot 部署到 Heroku 上，正式在網路上提供服務給所有使用者～

使用的工具及服務：

1. Python 3（for develop）
   
2. pipenv（for dependency management）
   
3. OLAMI（for NLP）
   
4. KKBOX Open API（for third-party skill import）
   
5. ngrok（for testing）
   
6. Heroku（for deploying our chatbot）
   

Step 7. Add custom skill into chatbot

OLAMI NLI 系統提供方法，讓我們可以定義 Intent、符合該 Intent 的句型和句型中的 slot。舉例來說，我可以設定 播放<keyword>類型的歌 的句型屬於 music_play_playlist Intent； <keyword> 則是句型中重要的 slot，讓 Chatbot 根據 <keyword> fetch 對應的 data 提供給使用者。
音樂資料從 KKBOX Open API 取得，因為 KKBOX 日本歌曲比較多。API 裡面的 search method 提供搜尋功能，把音樂資料分為 track、album、artist、playlist 四種類型。可以根據這四種類型定義四種 Intent 與符合的句型，然後用句型中的 slot 作為 request KKBOX search API 的 query parameter，最後將搜尋結果回傳給使用者。
從 OLAMI 我的應用介面 => 進入 NLI 系統 => 新增模組 => 名稱命名：music_kkbox => 提交

點選左側選單中的我的語法 => +新增語法
在這裡新增四種 Intent

1. music_play_track
   
2. music_play_album
   
3. music_play_artist
   
4. music_play_playlist
   

四種 slot（類型選擇 ext）

1. track_name
   
2. album_name
   
3. artist_name
   
4. keyword
   

最後組合成 4 種句型

1. 播放<track_name><{@=music_play_track}>
   
2. 播放<album_name>專輯的歌<{@=music_play_album}>
   
3. 播放<artist_name>的歌<{@=music_play_artist}>
   
4. 播放<keyword>類型的歌<{@=music_play_playlist}>
   

OSL（OLAMI Syntax Language） 語法教學文件

完成後的結果


點選畫面右上角發佈 => 回到 OLAMI 我的應用介面 => 對 App 點選變更設定 => 將 NLI 模組 => music_kkbox 模組打勾 => 儲存設定

在 OLAMI 我的應用介面點選測試，測試語句打 播放動漫歌曲類型的歌，會得到 Response

{
   "nli":[
      {
         "desc_obj":{
            "status":0
         },
         "semantic":[
            {
               "app":"music_kkbox",
               "input":"播放動漫歌曲類型的歌",
               "slots":[
                  {
                     "name":"keyword",
                     "value":"動漫歌曲"
                  }
               ],
               "modifier":[
                  "music_play_playlist"
               ],
               "customer":"59e031f7e4b0a8057efdce99"
            }
         ],
         "type":"music_kkbox"
      }
   ]
}

當使用者告訴 Chatbot：「播放動漫歌曲類型的歌」，由於有設定 webhook，Telegram 會把訊息傳送到 web server。程式收到訊息後，再把訊息傳給 OLAMI NLI API，就會得到上面的 Response。程式就可根據 Response 中的 type value 判斷「播放動漫歌曲類型的歌」這句話屬於 music_kkbox NLI 模組，再由 modifier 中的 element 暸解使用者的 Intent 是 music_play_playlist，最後用 slots 中的 keyword value 動漫歌曲 作為 request KKBOX Open API search method 的 query parameter，取得動漫歌曲的 playlist。
先註冊 KKBOX Developer 帳號，在 My Apps 頁面 Create new app，得到 App ID 及 Secret，再把它們填入專案目錄中的 config.ini 檔案

[KKBOX]
ID = your_app_id
SECRET = your_app_secret

接著專案目錄中，新增資料夾，名字叫做 api

$ mkdir api

進入 api 資料夾，新增兩個檔案， __init__.py 及 kkbox.py

$ cd api
$ touch __init__.py
$ touch kkbox.py

做完上述動作後的專案目錄結構

Project Directory
├── api
|   ├── __init__.py
|   └── kkbox.py
├── nlp
|   ├── __init__.py
|   └── olami.py
├── config.ini
├── main.py
├── Pipfile
└── Pipfile.lock

新增 __init__.py 是為了讓 olami.py import api 的時候認定 api 是一個 Module。
編輯 api/__init__.py

from . import kkbox

編輯 kkbox.py

import configparser
import logging

import requests

config = configparser.ConfigParser()
config.read('config.ini')

logger = logging.getLogger(__name__)


class KKBOX:
    AUTH_URL = 'https://account.kkbox.com/oauth2/token'
    API_BASE_URL = 'https://api.kkbox.com/v1.1/'

    def __init__(self, id=config['KKBOX']['ID'], secret=config['KKBOX']['SECRET']):
        self.id = id
        self.secret = secret
        self.token = self._get_token()

    def _get_token(self):
        response = requests.post(self.AUTH_URL, data={'grant_type': 'client_credentials'}, auth=(self.id, self.secret))
        response.raise_for_status()
        return response.json()['access_token']

    def search(self, type, q, territory='TW'):
        response = requests.get(self.API_BASE_URL + 'search', params={'type': type, 'q': q, 'territory': territory},
                                headers={'Authorization': 'Bearer ' + self.token})
        response.raise_for_status()
        response_json = response.json()
        result = {
            'artist': lambda: response_json['artists']['data'][0]['url'],
            'album': lambda: response_json['albums']['data'][0]['url'],
            'track': lambda: response_json['tracks']['data'][0]['url'],
            'playlist': lambda: response_json['playlists']['data'][0]['url']
        }[type]()
        return result

kkbox.py 實作了 KKBOX class init 時會利用 App ID 及 App Secret 走 Basic Authentication 取得 access_token。還有實作 request Search API method，根據期望的 type 及 q（keyword） 搜尋音樂資料，當搜尋有結果時，會 return 第一筆資料的 url。

KKBOX Open API access token 取得方法可參考官方 Tutorial
Search API 詳細說明文件

完成後，編輯 nlp/olami.py，我們要讓程式可以處理新的 music_kkbox Intent

+from api.kkbox import KKBOX

class Olami:
    def intent_detection(self, nli_obj):
+       def handle_music_kkbox_type(semantic):
+           type = semantic['modifier'][0].split('_')[2]
+           slots = semantic['slots']
+           kkbox = KKBOX()
+
+           def get_slot_value(key):
+               return next(filter(lambda el: el['name'] == key, slots))['value']
+
+           _reply = {
+               'artist': lambda: kkbox.search(type, get_slot_value('artist_name')),
+               'album': lambda: kkbox.search(type, get_slot_value('album_name')),
+               'track': lambda: kkbox.search(type, get_slot_value('track_name')),
+               'playlist': lambda: kkbox.search(type, get_slot_value('keyword'))
+           }[type]()
+           return _reply

        type = nli_obj['type']
        desc = nli_obj['desc_obj']
        data = nli_obj.get('data_obj', [])

        reply = {
            'kkbox': lambda: data[0]['url'] if len(data) > 0 else desc['result'],
            'baike': lambda: data[0]['description'],
            'news': lambda: data[0]['detail'],
            'joke': lambda: data[0]['content'],
            'cooking': lambda: data[0]['content'],
            'selection': lambda: handle_selection_type(desc['type']),
            'ds': lambda: desc['result'] + '\n請用 /help 指令看看我能怎麼幫助您',
+           'music_kkbox': lambda: handle_music_kkbox_type(nli_obj['semantic'][0])
        }.get(type, lambda: desc['result'])()

        return reply

修改後的完整 olami.py

import configparser
import json
import logging
import time
from hashlib import md5
from api.kkbox import KKBOX

import requests

config = configparser.ConfigParser()
config.read('config.ini')

logger = logging.getLogger(__name__)


class NliStatusError(Exception):
    """The NLI result status is not 'ok'"""


class Olami:
    URL = 'https://tw.olami.ai/cloudservice/api'

    def __init__(self, app_key=config['OLAMI']['APP_KEY'], app_secret=config['OLAMI']['APP_SECRET'], input_type=1):
        self.app_key = app_key
        self.app_secret = app_secret
        self.input_type = input_type

    def nli(self, text, cusid=None):
        response = requests.post(self.URL, params=self._gen_parameters('nli', text, cusid))
        response.raise_for_status()
        response_json = response.json()
        if response_json['status'] != 'ok':
            raise NliStatusError(
                "NLI responded status != 'ok': {}".format(response_json['status']))
        else:
            nli_obj = response_json['data']['nli'][0]
            return self.intent_detection(nli_obj)

    def _gen_parameters(self, api, text, cusid):
        timestamp_ms = (int(time.time() * 1000))
        params = {'appkey': self.app_key,
                  'api': api,
                  'timestamp': timestamp_ms,
                  'sign': self._gen_sign(api, timestamp_ms),
                  'rq': self._gen_rq(text)}
        if cusid is not None:
            params.update(cusid=cusid)
        return params

    def _gen_sign(self, api, timestamp_ms):
        data = self.app_secret + 'api=' + api + 'appkey=' + self.app_key + \
               'timestamp=' + str(timestamp_ms) + self.app_secret
        return md5(data.encode('ascii')).hexdigest()

    def _gen_rq(self, text):
        obj = {'data_type': 'stt', 'data': {'input_type': self.input_type, 'text': text}}
        return json.dumps(obj)

    def intent_detection(self, nli_obj):
        def handle_selection_type(type):
            reply = {
                'news': lambda: desc['result'] + '\n\n' + '\n'.join(
                    str(index + 1) + '. ' + el['title'] for index, el in enumerate(data)),
                'poem': lambda: desc['result'] + '\n\n' + '\n'.join(
                    str(index + 1) + '. ' + el['poem_name'] + '，作者：' + el['author'] for index, el in
                    enumerate(data)),
                'cooking': lambda: desc['result'] + '\n\n' + '\n'.join(
                    str(index + 1) + '. ' + el['name'] for index, el in
                    enumerate(data))
            }.get(type, lambda: '對不起，你說的我還不懂，能換個說法嗎？')()
            return reply

        def handle_music_kkbox_type(semantic):
            type = semantic['modifier'][0].split('_')[2]
            slots = semantic['slots']
            kkbox = KKBOX()

            def get_slot_value(key):
                return next(filter(lambda el: el['name'] == key, slots))['value']

            _reply = {
                'artist': lambda: kkbox.search(type, get_slot_value('artist_name')),
                'album': lambda: kkbox.search(type, get_slot_value('album_name')),
                'track': lambda: kkbox.search(type, get_slot_value('track_name')),
                'playlist': lambda: kkbox.search(type, get_slot_value('keyword'))
            }[type]()
            return _reply

        type = nli_obj['type']
        desc = nli_obj['desc_obj']
        data = nli_obj.get('data_obj', [])

        reply = {
            'kkbox': lambda: data[0]['url'] if len(data) > 0 else desc['result'],
            'baike': lambda: data[0]['description'],
            'news': lambda: data[0]['detail'],
            'joke': lambda: data[0]['content'],
            'cooking': lambda: data[0]['content'],
            'selection': lambda: handle_selection_type(desc['type']),
            'ds': lambda: desc['result'] + '\n請用 /help 指令看看我能怎麼幫助您',
            'music_kkbox': lambda: handle_music_kkbox_type(nli_obj['semantic'][0])
        }.get(type, lambda: desc['result'])()

        return reply

測試 Telegram Bot

Yeah~ Chatbot 有新的 Music-KKBOX 技能了！如果手機有裝 KKBOX App，從手機點選連結就會啟動 KKBOX App 播放歌曲。

Step 8. User-friendly chatbot design

還有幾個讓 Chatbot 變得更佳 user-friendly 的方法，分享給大家。

1. Welcome message

在使用者一加入 Chatbot 時出現，馬上讓使用者暸解 Chatbot 提供的功能，就像 Telegram 的 BotFather ㄧ樣。

2. Reply keyboard markup

對於制式回答，提供 keyboard markup 讓使用者可以直接點選，不用自己打字。

3. Help message

在使用者傳送的訊息 Chatbot 無法理解時出現，提示使用者這個 Chatbot 的使用方法。

4. Error handling

錯誤發生時的處理，Chatbot 如果發生錯誤而沒有回應，使用者就會覺得很困惑。

實作，編輯 main.py

+from telegram import ReplyKeyboardMarkup
+from telegram.ext import Dispatcher, CommandHandler, MessageHandler, Filters
+
+welcome_message = '親愛的主人，您可以問我\n' \
+                  '天氣，例如：「高雄天氣如何」\n' \
+                  '百科，例如：「川普是誰」\n' \
+                  '新聞，例如：「今日新聞」\n' \
+                  '音樂，例如：「我想聽周杰倫的等你下課」\n' \
+                  '日曆，例如：「現在時間」\n' \
+                  '詩詞，例如：「我想聽水調歌頭這首詩」\n' \
+                  '笑話，例如：「講個笑話」\n' \
+                  '故事，例如：「說個故事」\n' \
+                  '股票，例如：「台積電的股價」\n' \
+                  '食譜，例如：「蛋炒飯怎麼做」\n' \
+                  '聊天，例如：「你好嗎」'
+reply_keyboard_markup = ReplyKeyboardMarkup([['高雄天氣如何'],
+                                             ['川普是誰'],
+                                             ['今日新聞'],
+                                             ['我想聽周杰倫的等你下課'],
+                                             ['現在時間'],
+                                             ['我想聽水調歌頭這首詩'],
+                                             ['講個笑話'],
+                                             ['說個故事'],
+                                             ['台積電的股價'],
+                                             ['蛋炒飯怎麼做'],
+                                             ['你好嗎']])


+def start_handler(bot, update):
+    """Send a message when the command /start is issued."""
+    update.message.reply_text(welcome_message, reply_markup=reply_keyboard_markup)
+
+
+def help_handler(bot, update):
+    """Send a message when the command /help is issued."""
+    update.message.reply_text(welcome_message, reply_markup=reply_keyboard_markup)
+
+
+def error_handler(bot, update, error):
+    """Log Errors caused by Updates."""
+    logger.error('Update "%s" caused error "%s"', update, error)
+    update.message.reply_text('對不起主人，我需要多一點時間來處理 Q_Q')

+dispatcher.add_handler(CommandHandler('start', start_handler))
+dispatcher.add_handler(CommandHandler('help', help_handler))
+dispatcher.add_error_handler(error_handler)

修改後的完整 main.py

import configparser
import logging

import telegram
from flask import Flask, request
from telegram import ReplyKeyboardMarkup
from telegram.ext import Dispatcher, CommandHandler, MessageHandler, Filters

from nlp.olami import Olami

# Load data from config.ini file
config = configparser.ConfigParser()
config.read('config.ini')

# Enable logging
logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
                    level=logging.INFO)
logger = logging.getLogger(__name__)

# Initial Flask app
app = Flask(__name__)

# Initial bot by Telegram access token
bot = telegram.Bot(token=(config['TELEGRAM']['ACCESS_TOKEN']))

welcome_message = '親愛的主人，您可以問我\n' \
                  '天氣，例如：「高雄天氣如何」\n' \
                  '百科，例如：「川普是誰」\n' \
                  '新聞，例如：「今日新聞」\n' \
                  '音樂，例如：「我想聽周杰倫的等你下課」\n' \
                  '日曆，例如：「現在時間」\n' \
                  '詩詞，例如：「我想聽水調歌頭這首詩」\n' \
                  '笑話，例如：「講個笑話」\n' \
                  '故事，例如：「說個故事」\n' \
                  '股票，例如：「台積電的股價」\n' \
                  '食譜，例如：「蛋炒飯怎麼做」\n' \
                  '聊天，例如：「你好嗎」'
reply_keyboard_markup = ReplyKeyboardMarkup([['高雄天氣如何'],
                                             ['川普是誰'],
                                             ['今日新聞'],
                                             ['我想聽周杰倫的等你下課'],
                                             ['現在時間'],
                                             ['我想聽水調歌頭這首詩'],
                                             ['講個笑話'],
                                             ['說個故事'],
                                             ['台積電的股價'],
                                             ['蛋炒飯怎麼做'],
                                             ['你好嗎']])


@app.route('/hook', methods=['POST'])
def webhook_handler():
    """Set route /hook with POST method will trigger this method."""
    if request.method == "POST":
        update = telegram.Update.de_json(request.get_json(force=True), bot)
        dispatcher.process_update(update)
    return 'ok'


def start_handler(bot, update):
    """Send a message when the command /start is issued."""
    update.message.reply_text(welcome_message, reply_markup=reply_keyboard_markup)


def help_handler(bot, update):
    """Send a message when the command /help is issued."""
    update.message.reply_text(welcome_message, reply_markup=reply_keyboard_markup)


def reply_handler(bot, update):
    """Reply message."""
    text = update.message.text
    reply = Olami().nli(text)
    update.message.reply_text(reply)


def error_handler(bot, update, error):
    """Log Errors caused by Updates."""
    logger.error('Update "%s" caused error "%s"', update, error)
    update.message.reply_text('對不起主人，我需要多一點時間來處理 Q_Q')


# New a dispatcher for bot
dispatcher = Dispatcher(bot, None)

# Add handler for handling message, there are many kinds of message. For this handler, it particular handle text
# message.
dispatcher.add_handler(MessageHandler(Filters.text, reply_handler))
dispatcher.add_handler(CommandHandler('start', start_handler))
dispatcher.add_handler(CommandHandler('help', help_handler))
dispatcher.add_error_handler(error_handler)

if __name__ == "__main__":
    # Running server
    app.run(debug=True)

Last Step - Deployment

最後，我們要把 Chatbot web server 程式 deploy 到 production 環境上。選擇的是 Heroku，它的 Free pricing 方案不用綁定信用卡就可以使用，部署方法也相當簡單。確認已經有註冊 Heroku Account，而且電腦有安裝 Heroku CLI。
進入 Heroku Dashboard，Create new app
回到專案目錄，新增 Procfile 檔，編輯成如下

web: gunicorn main:app --log-file -

Procfile 中的指令在程式成功 deploy 上 Heroku 後就會被執行。裡面的指令代表要讓 Heroku running 一個 web process，用 gunicorn 部署 main module 中的 Flask App。

Procfile 說明文件
Gunicorn 官網

完成後的專案目錄結構

Project Directory
├── api
|   ├── __init__.py
|   └── kkbox.py
├── nlp
|   ├── __init__.py
|   └── olami.py
├── config.ini
├── main.py
├── Pipfile
├── Pipfile.lock
└── Procfile

在專案目錄初始化 git repository

$ git init

新增 production 分支、切換到該分支

$ git checkout -b production

把專案中的所有檔案加入至 git

$ git add .

commit

$ git commit -m "Deploying to Heroku"

Log in Heroku account

$ heroku login

Add remote Heroku repository

$ heroku git:remote -a {your_heroku_app_name}

Push to Heroku repository

$ git push heroku production:master

順利的話，程式就會成功部署上 Heroku。從 console 的輸出訊息、Heroku Dashboard App 的 Settings 頁面中，及右上角 Open app 都可以找到 App 的 url。它會是

https://{$your_heroku_app_name}.herokuapp.com/

例如：

https://kkbox-telegram-bot.herokuapp.com/

最後將這個 url 設定為 Telegram Bot 的 webhook url，就大功告成了。

https://api.telegram.org/bot{$token}/setWebhook?url={$webhook_url}

$token 和 $webhook_url 請換成你在 Step 1 中申請到的，例如：

https://api.telegram.org/bot606248605:AAGv_TOJdNNMc_v3toHK_X6M-dev_1tG-JA/setWebhook?url=https://kkbox-telegram-bot.herokuapp.com/hook

Getting Started on Heroku with Python Document
如果要將專案 push 到 GitHub，先 checkout 回 master branch、新增 .gitignore 檔案，ignore config.ini（因為裡面有 credential 資訊），再進行後續動作

總結

用上下兩篇文章介紹如何打造具備 NLP 功能的 Telegram Bot，共 9 個步驟實作上圖 Chatbot 流程中的每個環節，包含導入 NLP service、Intent detection、Add custom skill、優化使用者體驗等。
完整程式碼放在 GitHub repository，按照 README 的步驟，就可以 deploy 和範例具備一樣技能的 Telegram Bot，歡迎 pull request。

回到第一篇文章