Source code for deeppavlov.skills.aiml_skill.aiml_skill

import uuid
from pathlib import Path
from typing import Tuple, Optional, List, Iterable
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()"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[str], history_batch: Optional[List]=None, states_batch: Optional[List]=None) -> Tuple[List[str], List[float], 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 = [] if states_batch is None: # generate states batch matching batch of utterances: states_batch = [None] * len(utterances_batch) 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