ML: Чатбот RASA: Действия

Введение

В ряде случаев отклики чатбота RASA требуют некоторого логического анализа или работы с внешними данными. Всё это можно запрограммировать на Python. Созданию таких откликов, называемых action (действие), посвящён этот документ.

Подготовка

Для работы с действиями, необходимо сделать несколько предварительных настроек. Прежде всего, в файле endpoints.yml раскомментируем следующие две линии: 

action_endpoint:                            # endpoints.yml
  url: "http://localhost:5055/webhook"
Затем, в разделе actions файла domain.yml перечислим используемые действия. Пока это будет единственное действие с именем action_show_time, сообщающее пользователю текущее время: 
actions:                                    # domain.yml
  - action_show_time

Чтобы тестирование действия было более осмысленным, в файле data/nlu.yml добавим примеры, для намерения пользователя узнать текущее время: 

nlu:                                        # data/nlu.yml
- intent: what_time_is_it
  examples: |
    - Скажи мне сколько времени?
    - Сколько времени?    
    - Который сейчас час?
    - Который час?

Для простоты, в data/rues.yml добавим жёсткое правило, обязующее бот на намерение what_time_is_it всегда вызывать действие action_show_time: 

rules:                                      # data/rues.yml
- rule: Скажи мне сколько времени
  steps:
  - intent: what_time_is_it
  - action: action_show_time
Естественно, действие можно вызывать и в историях наравне с откликами utter_... В отличии от простых откликов, имена откликов действий начинаются с преффикса action_...

Скрипт действия

Теперь запрограммируем собственно действие action_show_time. Для этого в actions/actions.py напишем следующий код: 


from rasa_sdk          import Action, Tracker
from rasa_sdk.events   import SlotSet
from rasa_sdk.executor import CollectingDispatcher

from datetime import datetime as dt
from typing import Any, Text, Dict, List

class ActionShowTime(Action):

    def name(self) -> Text:              # регистрируем имя действия
        return "action_show_time"

    def run(self, dispatcher:CollectingDispatcher, tracker:Tracker, domain:Dict[Text,Any])
        -> List[Dict[Text, Any]]:

        # при вызове действия возвращаться ответ с текущим временем: 
        dispatcher.utter_message(text=f'Сейчас {dt.now().strftime("%H:%M")}')

        return []
В классе ActionShowTime метод name должен возвращать имя действия, а в методе run происходит генерация отклика бота, вызовом функции utter_message. Вместо аргумента text, можно вызвать стандартный отклик, присвоив его в аргумент response (можно совмещать text и response): 
dispatcher.utter_message(response="utter_goodbye")

Такой бот обучается как обычно (rasa train). Перед его тестированием в корне проекта необходимо запустить сервер RASA командой "rasa run actions". Затем запускается диалог с ботом (rasa shell или !agent.py).

Если скрипт действия меняется, повторно вызывать обучение не надо, но необходимо перезапустить сервер (rasa run actions), прервав работу предыдущего. Возникающие в скрипте ошибки видны в окне сервера. Там же видны отладочные вызовы функции print.

Усложняем действие

Метод run получает переменную tracker, из которой можно извлечь последнее сообщение от клиента. Например, перед сообщением времени, в классе ActionShowTime глубокомысленно повторим вопрос человека: 

    def run(self, dispatcher:CollectingDispatcher, tracker:Tracker, domain:Dict[Text,Any])
        -> List[Dict[Text, Any]]:
        
        text  = tracker.latest_message['text']
        utter = f'На ваш вопрос "{text}" отвечу: {dt.now().strftime("%H:%M")}'
        dispatcher.utter_message(text=utter)

        return []
Естественно, переменную text можно использовать с большей пользой, например, для углублённого (по сравнению с классификацией намерения) семантического анализа.

Метод run возвращает список, в который можно поместить значение слотов (ячеек памяти). Слоты должны быть перечислены в файле domain.yml, поэтому добавляем в него слот времени последнего вызова действия: 

slots:
  LAST_TIME_ASKING:
    type: text
После этого необходимо запустить повторное обучение (rasa train), иначе при попытке задать значение слота, будет возникать предупреждение о его отсутствии.

Теперь модифицируем метод ActionShowTime.run так, чтобы в слоте сохранялось (в формате dt_frm как текст) время последнего вызова (оператор return), которое (функция get_slot) будет влиять на ответ действия: 

    def run(self, dispatcher:CollectingDispatcher, tracker:Tracker, domain:Dict[Text,Any])
        -> List[Dict[Text, Any]]:
        
        dt_frm = "%Y-%m-%d %H:%M:%S.%f"                  # формат даты и времени
        
        prev  = tracker.get_slot("LAST_TIME_ASKING")     # значение слота "LAST_TIME_ASKING"
        
        utter = f'Сейчас: {dt.now().strftime("%H:%M")}.'
        if prev:                                         # если не None (уже заполнили), то:
            prev = dt.strptime(prev, dt_frm)             # восстанавливаем время
            secs = (dt.now() - prev).total_seconds()     # разница в секундах с текущим
            
            utter += " Об этом ты спрашивал %d секунд назад." % (secs)
            
        dispatcher.utter_message(text=utter)

        return [SlotSet("LAST_TIME_ASKING", dt.now().strftime(dt_frm))] # заполняем слот
Пример бота, сообщающего текущее время, находится в файле Bot04_Action.zip.

Dispatcher, Tracker и domain

Объект dispatcher (диспетчер), доступный в действии, при помощи метода utter_message отправляет сообщения пользователю. Он имеет следующие необязательные аргументы:

Например, меню в виде кнопок можно послать следующим образом: 
dispatcher.utter_message(buttons = [
                {"payload": "/affirm", "title": "Yes"},
                {"payload": "/deny", "title": "No"},
            ])

Объект tracker обладает следующими свойствами

В словаре domain находятся все свойства из файла domain.yml.

Приведём пример получения из latest_message текста последнего сообщения, его намерения и извлечённых из него сущностей: 

print(tracker.latest_message['text'])
print(tracker.latest_message['intent']['name'], "[%.2f]" % 
              (tracker.latest_message['intent']['confidence']))
if 'entities' in tracker.latest_message:
    for e in tracker.latest_message['entities']:
        print("- [%.2f] entity:%s, value:%s pos:[%d,%d]" %  
             (e['confidence_entity'],  e['entity'],  e['value'], e['start'], e['end']))

Полезная информация