跳到主要內容

鑽石級贊助商 - 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__.pykkbox.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,根據期望的 typeq(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 ㄧ樣。

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

  1. Help message
在使用者傳送的訊息 Chatbot 無法理解時出現,提示使用者這個 Chatbot 的使用方法。

  1. 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。
回到第一篇文章

留言

這個網誌中的熱門文章

COSCUP 2024 財務報告

COSCUP 2024 年度財務收支摘要 COSCUP 2024 Annual Income and Expenditure Summary For the period 2024/01/01~2024/12/31 收入 Revenues 金額 總計 企業贊助收入 NT$1,644,863 個人贊助收入 129,099 義賣收入 141,555 利息收入 6,199 周邊活動收入 59,200 總收入 Total Revenues NT$1,980,916 支出 Expenses 金額 總計 場地費 NT$714,338 義務工作人員費用 592,639 擺攤出訪費用 471,205 行銷費 353,631 餐飲費用 230,079 議程交流費用 229,525 設備與器材費 220,106 周邊活動費用 75,500 倉儲及物流費用 45,510 線上系統費用 41,728 保險費 15,228 雜支 3,017 捐贈開源社群 43,000 金流手續費 6,412 稅費與金流服務費 174,693 總支出 Total Expenses NT$3,216,611 期間淨收支 Net Income 2024/01/01~2024/12/31 -NT$1,235,695 支出科目 歷年結餘 歷年結餘 2006/2008~2018 結餘 NT$5,146,663 2019 結餘 NT$1,128,514 2020 結餘 NT$86,842 2021 結餘 NT$744,271 2022 結餘 NT$1,015,462 2023 結餘 -NT$198,240 2024 結餘 -NT$1,235,695 結餘款總計 NT$6,806,648

Early Guide to Joining COSCUP 2025 

(Updated: Feb 13) Many people in the community are excited to find out how to participate in COSCUP 2025. We’ll share more updates soon, but here’s what we can share so far! You can join as a speaker, host a booth, organize a track, or become a sponsor. If you’re new to COSCUP, take a look at the article " COSCUP Unveiled " to learn more. Date and Location The event will take place at NTUST in Taipei, Taiwan, on a weekend in late July to early August, as usual. The planned dates for this year are August 9-10. Be a Speaker Our early bird Call for Proposals (CfP) is now opened!  If you have a presentation about open culture or open-source technology, we want to hear from you. Submitting early means you’ll hear back sooner about your proposal. Host a Booth Open-source communities can set up booths to display their work and interact with attendees. This is a great way to share your community’s projects and connect with others. Businesses that support open-source can also host a ...

COC 通報處理說明公告 - 20240811 通報事件

各位好, COSCUP COC 服務小組於 2024 年 8 月 11 日接獲一件通報,內容涉及在會期干擾議程進行;並於會後持續發送私訊予會中結識的講者;同時,該行為人亦被紀錄於活動當日干擾志工執行勤務。 有關此事件的處理過程,詳如下述: COC 服務小組接到通報後,於 8 月 15 日正式成立專案小組進行討論與檢視相關資料。經查,通報內容與 COC 條款「持續干擾議程或活動的正常進行,無視工作人員或與會者的制止」相符。同一行為人於大會期間,另有兩位會眾通報類似事件,COC 服務小組皆已明確指正其行為並重申 COC 規範和界線。綜合此次會後通報,行為人經提醒仍多次抵觸 COC 條例。 有鑒於上述行徑已明確影響 COSCUP 其他會眾之權益,COC 服務小組將依照 COSCUP COC 之辦法記錄事件處理過程及結果、行為人資料等,於籌備團隊組長群資料夾建立文件,以俾後續籌備團隊審慎思量該名行為人未來的參與形式與程度。 在此,感謝會眾願意信任 COC 和 COSCUP 團隊並且將其所遇到的事件於會後彙整提供予我們。另本次通報中,通報人所提及之部分事項,因非屬 COSCUP 大會參與期間和相關行為,已建議通報人另行循其他正規途徑處理。在此聲明, COSCUP 的 COC 落實並非要拒任何人於門外,而是希冀透過針對行為本身的評估,為無論志工、社群協調人、講者、廠商與所有會眾營造舒適與安全的交流環境。 我們在乎所有人於 COSCUP 大會的各種參與體驗與感受,如果您在大會和籌組期間有相關困擾,籌備團隊志工將會竭力協助釐清,希望一同打造友善的 COSCUP 與會環境。 COSCUP 2024 COC 服務小組