Source code for deeppavlov.skills.aiml_skill.aiml_skill

import uuid
from pathlib import Path
from typing import Union
from typing import Tuple, Optional, List
from logging import getLogger

import aiml

from deeppavlov.core.common.registry import register
from deeppavlov.core.skill.skill import Skill

log = getLogger(__name__)


[docs]@register("aiml_skill") class AIMLSkill(Skill): """Skill wraps python-aiml library into DeepPavlov interfrace. AIML uses directory with AIML scripts which are loaded at initialization and used as patterns for answering at each step. """ def __init__(self, path_to_aiml_scripts: str, positive_confidence: float = 0.66, null_response: str = "I don't know what to answer you", null_confidence: float = 0.33, **kwargs ) -> None: """ Construct skill: read AIML scripts, load AIML kernel Args: path_to_aiml_scripts: string path to folder with AIML scripts null_response: Response string to answer if no AIML Patterns matched positive_confidence: The confidence of response if response was found in AIML scripts null_confidence: The confidence when AIML scripts has no rule for responding and system returns null_response """ # we need absolute path (expanded for user home and resolved if it relative path): self.path_to_aiml_scripts = Path(path_to_aiml_scripts).expanduser().resolve() log.info(f"path_to_aiml_scripts is: `{self.path_to_aiml_scripts}`") self.positive_confidence = positive_confidence self.null_confidence = null_confidence self.null_response = null_response self.kernel = aiml.Kernel() # to block AIML output: self.kernel._verboseMode = False self._load_scripts() def _load_scripts(self) -> None: """ Scripts are loaded recursively from files with extensions .xml and .aiml Returns: None """ # learn kernel to all aimls in directory tree: all_files = sorted(self.path_to_aiml_scripts.rglob('*.*')) learned_files = [] for each_file_path in all_files: if each_file_path.suffix in ['.aiml', '.xml']: # learn the script file self.kernel.learn(str(each_file_path)) learned_files.append(each_file_path) if not learned_files: log.warning(f"No .aiml or .xml files found for AIML Kernel in directory {self.path_to_aiml_scripts}") def process_step(self, utterance_str: str, user_id: any) -> Tuple[str, float]: response = self.kernel.respond(utterance_str, sessionID=user_id) # here put your estimation of confidence: if response: # print(f"AIML responds: {response}") confidence = self.positive_confidence else: # print("AIML responses silently...") response = self.null_response confidence = self.null_confidence return response, confidence def _generate_user_id(self) -> str: """ Here you put user id generative logic if you want to implement it in the skill. Although it is better to delegate user_id generation to Agent Layer Returns: str """ return uuid.uuid1().hex def __call__(self, utterances_batch: List, history_batch: List, states_batch: List) -> Tuple[List, ...]: """Returns skill inference result. Returns batches of skill inference results, estimated confidence levels and up to date states corresponding to incoming utterance batch. Args: utterances_batch: A batch of utterances of str type. history_batch: A batch of list typed histories for each utterance. states_batch: A batch of arbitrary typed states for each utterance. Returns: response: A batch of arbitrary typed skill inference results. confidence: A batch of float typed confidence levels for each of skill inference result. output_states_batch: A batch of arbitrary typed states for each utterance. """ # grasp user_ids from states batch. # We expect that skill receives None or dict of state for each utterance. # if state has user_id then skill uses it, otherwise it generates user_id and calls the # user with this name in further. # In this implementation we use current datetime for generating uniqe ids output_states_batch = [] user_ids = [] for state in states_batch: if not state: user_id = self._generate_user_id() new_state = {'user_id': user_id} elif 'user_id' not in state: new_state = state user_id = self._generate_user_id() new_state['user_id'] = self._generate_user_id() else: new_state = state user_id = new_state['user_id'] user_ids.append(user_id) output_states_batch.append(new_state) confident_responses = map(self.process_step, utterances_batch, user_ids) responses_batch, confidences_batch = zip(*confident_responses) return responses_batch, confidences_batch, output_states_batch