from __future__ import unicode_literals
import json
import logging
from builtins import str
from collections import defaultdict
from itertools import combinations
from pathlib import Path
from future.utils import iteritems, itervalues
from snips_nlu_utils import normalize, hash_str
from snips_inference_agl.common.log_utils import log_elapsed_time, log_result
from snips_inference_agl.common.utils import (
check_persisted_path, deduplicate_overlapping_entities, fitted_required,
json_string)
from snips_inference_agl.constants import (
DATA, END, ENTITIES, ENTITY, ENTITY_KIND, INTENTS, LANGUAGE, RES_INTENT,
RES_INTENT_NAME, RES_MATCH_RANGE, RES_SLOTS, SLOT_NAME, START, TEXT,
UTTERANCES, RES_PROBA)
from snips_inference_agl.dataset import (
validate_and_format_dataset, extract_intent_entities)
from snips_inference_agl.dataset.utils import get_stop_words_whitelist
from snips_inference_agl.entity_parser.builtin_entity_parser import is_builtin_entity
from snips_inference_agl.exceptions import IntentNotFoundError, LoadingError
from snips_inference_agl.intent_parser.intent_parser import IntentParser
from snips_inference_agl.pipeline.configs import LookupIntentParserConfig
from snips_inference_agl.preprocessing import tokenize_light
from snips_inference_agl.resources import get_stop_words
from snips_inference_agl.result import (
empty_result, intent_classification_result, parsing_result,
unresolved_slot, extraction_result)
logger = logging.getLogger(__name__)
@IntentParser.register("lookup_intent_parser")
class LookupIntentParser(IntentParser):
"""A deterministic Intent parser implementation based on a dictionary
This intent parser is very strict by nature, and tends to have a very good
precision but a low recall. For this reason, it is interesting to use it
first before potentially falling back to another parser.
"""
config_type = LookupIntentParserConfig
def __init__(self, config=None, **shared):
"""The lookup intent parser can be configured by passing a
:class:`.LookupIntentParserConfig`"""
super(LookupIntentParser, self).__init__(config, **shared)
self._language = None
self._stop_words = None
self._stop_words_whitelist = None
self._map = None
self._intents_names = []
self._slots_names = []
self._intents_mapping = dict()
self._slots_mapping = dict()
self._entity_scopes = None
@property
def language(self):
return self._language
@language.setter
def language(self, value):
self._language = value
if value is None:
self._stop_words = None
else:
if self.config.ignore_stop_words:
self._stop_words = get_stop_words(self.resources)
else:
self._stop_words = set()
@property
def fitted(self):
"""Whether or not the intent parser has already been trained"""
return self._map is not None
@log_elapsed_time(
logger, logging.INFO, "Fitted lookup intent parser in {elapsed_time}")
def fit(self, dataset, force_retrain=True):
"""Fits the intent parser with a valid Snips dataset"""
logger.info("Fitting lookup intent parser...")
dataset = validate_and_format_dataset(dataset)
self.load_resources_if_needed(dataset[LANGUAGE])
self.fit_builtin_entity_parser_if_needed(dataset)
self.fit_custom_entity_parser_if_needed(dataset)
self.language = dataset[LANGUAGE]
self._entity_scopes = _get_entity_scopes(dataset)
self._map = dict()
self._stop_words_whitelist = get_stop_words_whitelist(
dataset, self._stop_words)
entity_placeholders = _get_entity_placeholders(dataset, self.language)
ambiguous_keys = set()
for (key, val) in self._generate_io_mapping(dataset[INTENTS],
entity_placeholders):
key = hash_str(key)
# handle key collisions -*- flag ambiguous entries -*-
if key in self._map and self._map[key] != val:
ambiguous_keys.add(key)
else:
self._map[key] = val
# delete ambiguous keys
for key in ambiguous_keys:
self._map.pop(key)
return self
@log_result(logger, logging.DEBUG, "LookupIntentParser result -> {result}")
@log_elapsed_time(logger, logging.DEBUG, "Parsed in {elapsed_time}.")
@fitted_required
def parse(self, text, intents=None, top_n=None):
"""Performs intent parsing on the provided *text*
Intent and slots are extracted simultaneously through pattern matching
Args:
text (str): input
intents (str or list of str): if provided, reduces the scope of
intent parsing to the provided list of intents
top_n (int, optional): when provided, this method will return a
list of at most top_n most likely intents, instead of a single
parsing result.
Note that the returned list can contain less than ``top_n``
elements, for instance when the parameter ``intents`` is not
None, or when ``top_n`` is greater than the total number of
intents.
Returns:
dict or list: the most likely intent(s) along with the extracted
slots. See :func:`.parsing_result` and :func:`.extraction_result`
for the output format.
Raises:
NotTrained: when the intent parser is not fitted
"""
if top_n is None:
top_intents = self._parse_top_intents(text, top_n=1,
intents=intents)
if top_intents:
intent = top_intents[0][RES_INTENT]
slots = top_intents[0][RES_SLOTS]
if intent[RES_PROBA] <= 0.5:
# return None in case of ambiguity
return empty_result(text, probability=1.0)
return parsing_result(text, intent, slots)
return empty_result(text, probability=1.0)
return self._parse_top_intents(text, top_n=top_n, intents=intents)
def _parse_top_intents(self, text, top_n, intents=None):
if isinstance(intents, str):
intents = {intents}
elif isinstance(intents, list):
intents = set(intents)
if top_n < 1:
raise ValueError(
"top_n argument must be greater or equal to 1, but got: %s"
% top_n)
results_per_intent = defaultdict(list)
for text_candidate, entities in self._get_candidates(text, intents):
val = self._map.get(hash_str(text_candidate))
if val is not None:
result = self._parse_map_output(text, val, entities, intents)
if result:
intent_name = result[RES_INTENT][RES_INTENT_NAME]
results_per_intent[intent_name].append(result)
results = []
for intent_results in itervalues(results_per_intent):
sorted_results = sorted(intent_results,
key=lambda res: len(res[RES_SLOTS]))
results.append(sorted_results[0])
# In some rare cases there can be multiple ambiguous intents
# In such cases, priority is given to results containing fewer slots
weights = [1.0 / (1.0 + len(res[RES_SLOTS])) for res in results]
total_weight = sum(weights)
for res, weight in zip(results, weights):
res[RES_INTENT][RES_PROBA] = weight / total_weight
results = sorted(results, key=lambda r: -r[RES_INTENT][RES_PROBA])
return results[:top_n]
def _get_candidates(self, text, intents):
candidates = defaultdict(list)
for grouped_entity_scope in self._entity_scopes:
entity_scope = grouped_entity_scope["entity_scope"]
intent_group = grouped_entity_scope["intent_group"]
intent_group = [intent_ for intent_ in intent_group
if intents is None or intent_ in intents]
if not intent_group:
continue
builtin_entities = self.builtin_entity_parser.parse(
text, scope=entity_scope["builtin"], use_cache=True)
custom_entities = self.custom_entity_parser.parse(
text, scope=entity_scope["custom"], use_cache=True)
all_entities = builtin_entities + custom_entities
all_entities = deduplicate_overlapping_entities(all_entities)
# We generate all subsets of entities to match utterances
# containing ambivalent words which can be both entity values or
# random words
for entities in _get_entities_combinations(all_entities):
processed_text = self._replace_entities_with_placeholders(
text, entities)
for intent in intent_group:
cleaned_text = self._preprocess_text(text, intent)
cleaned_processed_text = self._preprocess_text(
processed_text, intent)
raw_candidate = cleaned_text, []
placeholder_candidate = cleaned_processed_text, entities
intent_candidates = [raw_candidate, placeholder_candidate]
for text_input, text_entities in intent_candidates:
if text_input not in candidates \
or text_entities not in candidates[text_input]:
candidates[text_input].append(text_entities)
yield text_input, text_entities
def _parse_map_output(self, text, output, entities, intents):
"""Parse the map output to the parser's result format"""
intent_id, slot_ids = output
intent_name = self._intents_names[intent_id]
if intents is not None and intent_name not in intents:
return None
parsed_intent = intent_classification_result(
intent_name=intent_name, probability=1.0)
slots = []
# assert invariant
assert len(slot_ids) == len(entities)
for slot_id, entity in zip(slot_ids, entities):
slot_name = self._slots_names[slot_id]
rng_start = entity[RES_MATCH_RANGE][START]
rng_end = entity[RES_MATCH_RANGE][END]
slot_value = text[rng_start:rng_end]
entity_name = entity[ENTITY_KIND]
slot = unresolved_slot(
[rng_start, rng_end], slot_value, entity_name, slot_name)
slots.append(slot)
return extraction_result(parsed_intent, slots)
@fitted_required
def get_intents(self, text):
"""Returns the list of intents ordered by decreasing probability
The length of the returned list is exactly the number of intents in the
dataset + 1 for the None intent
"""
nb_intents = len(self._intents_names)
top_intents = [intent_result[RES_INTENT] for intent_result in
self._parse_top_intents(text, top_n=nb_intents)]
matched_intents = {res[RES_INTENT_NAME] for res in top_intents}
for intent in self._intents_names:
if intent not in matched_intents:
top_intents.append(intent_classification_result(intent, 0.0))
# The None intent is not included in the lookup table and is thus
# never matched by the lookup parser
top_intents.append(intent_classification_result(None, 0.0))
return top_intents
@fitted_required
def get_slots(self, text, intent):
"""Extracts slots from a text input, with the knowledge of the intent
Args:
text (str): input
intent (str): the intent which the input corresponds to
Returns:
list: the list of extracted slots
Raises:
IntentNotFoundError: When the intent was not part of the training
data
"""
if intent is None:
return []
if intent not in self._intents_names:
raise IntentNotFoundError(intent)
slots = self.parse(text, intents=[intent])[RES_SLOTS]
if slots is None:
slots = []
return slots
def _get_intent_stop_words(self, intent):
whitelist = self._stop_words_whitelist.get(intent, set())
return self._stop_words.difference(whitelist)
def _get_intent_id(self, intent_name):
"""generate a numeric id for an intent
Args:
intent_name (str): intent name
Returns:
int: numeric id
"""
intent_id = self._intents_mapping.get(intent_name)
if intent_id is None:
intent_id = len(self._intents_names)
self._intents_names.append(intent_name)
self._intents_mapping[intent_name] = intent_id
return intent_id
def _get_slot_id(self, slot_name):
"""generate a numeric id for a slot
Args:
slot_name (str): intent name
Returns:
int: numeric id
"""
slot_id = self._slots_mapping.get(slot_name)
if slot_id is None:
slot_id = len(self._slots_names)
self._slots_names.append(slot_name)
self._slots_mapping[slot_name] = slot_id
return slot_id
def _preprocess_text(self, txt, intent):
"""Replaces stop words and characters that are tokenized out by
whitespaces"""
stop_words = self._get_intent_stop_words(intent)
tokens = tokenize_light(txt, self.language)
cleaned_string = " ".join(
[tkn for tkn in tokens if normalize(tkn) not in stop_words])
return cleaned_string.lower()
def _generate_io_mapping(self, intents, entity_placeholders):
"""Generate input-output pairs"""
for intent_name, intent in sorted(iteritems(intents)):
intent_id = self._get_intent_id(intent_name)
for entry in intent[UTTERANCES]:
yield self._build_io_mapping(
intent_id, entry, entity_placeholders)
def _build_io_mapping(self, intent_id, utterance, entity_placeholders):
input_ = []
output = [intent_id]
slots = []
for chunk in utterance[DATA]:
if SLOT_NAME in chunk:
slot_name = chunk[SLOT_NAME]
slot_id = self._get_slot_id(slot_name)
entity_name = chunk[ENTITY]
placeholder = entity_placeholders[entity_name]
input_.append(placeholder)
slots.append(slot_id)
else:
input_.append(chunk[TEXT])
output.append(slots)
intent = self._intents_names[intent_id]
key = self._preprocess_text(" ".join(input_), intent)
return key, output
def _replace_entities_with_placeholders(self, text, entities):
if not entities:
return text
entities = sorted(entities, key=lambda e: e[RES_MATCH_RANGE][START])
processed_text = ""
current_idx = 0
for ent in entities:
start = ent[RES_MATCH_RANGE][START]
end = ent[RES_MATCH_RANGE][END]
processed_text += text[current_idx:start]
place_holder = _get_entity_name_placeholder(
ent[ENTITY_KIND], self.language)
processed_text += place_holder
current_idx = end
processed_text += text[current_idx:]
return processed_text
@check_persisted_path
def persist(self, path):
"""Persists the object at the given path"""
path.mkdir()
parser_json = json_string(self.to_dict())
parser_path = path / "intent_parser.json"
with parser_path.open(mode="w", encoding="utf8") as pfile:
pfile.write(parser_json)
self.persist_metadata(path)
@classmethod
def from_path(cls, path, **shared):
"""Loads a :class:`LookupIntentParser` instance from a path
The data at the given path must have been generated using
:func:`~LookupIntentParser.persist`
"""
path = Path(path)
model_path = path / "intent_parser.json"
if not model_path.exists():
raise LoadingError(
"Missing lookup intent parser metadata file: %s"
% model_path.name)
with model_path.open(encoding="utf8") as pfile:
metadata = json.load(pfile)
return cls.from_dict(metadata, **shared)
def to_dict(self):
"""Returns a json-serializable dict"""
stop_words_whitelist = None
if self._stop_words_whitelist is not None:
stop_words_whitelist = {
intent: sorted(values)
for intent, values in iteritems(self._stop_words_whitelist)}
return {
"config": self.config.to_dict(),
"language_code": self.language,
"map": self._map,
"slots_names": self._slots_names,
"intents_names": self._intents_names,
"entity_scopes": self._entity_scopes,
"stop_words_whitelist": stop_words_whitelist,
}
@classmethod
def from_dict(cls, unit_dict, **shared):
"""Creates a :class:`LookupIntentParser` instance from a dict
The dict must have been generated with
:func:`~LookupIntentParser.to_dict`
"""
config = cls.config_type.from_dict(unit_dict["config"])
parser = cls(config=config, **shared)
parser.language = unit_dict["language_code"]
# pylint:disable=protected-access
parser._map = _convert_dict_keys_to_int(unit_dict["map"])
parser._slots_names = unit_dict["slots_names"]
parser._intents_names = unit_dict["intents_names"]
parser._entity_scopes = unit_dict["entity_scopes"]
if parser.fitted:
whitelist = unit_dict["stop_words_whitelist"]
parser._stop_words_whitelist = {
intent: set(values) for intent, values in iteritems(whitelist)}
# pylint:enable=protected-access
return parser
def _get_entity_scopes(dataset):
intent_entities = extract_intent_entities(dataset)
intent_groups = []
entity_scopes = []
for intent, entities in sorted(iteritems(intent_entities)):
scope = {
"builtin": list(
{ent for ent in entities if is_builtin_entity(ent)}),
"custom": list(
{ent for ent in entities if not is_builtin_entity(ent)})
}
if scope in entity_scopes:
group_idx = entity_scopes.index(scope)
intent_groups[group_idx].append(intent)
else:
entity_scopes.append(scope)
intent_groups.append([intent])
return [
{
"intent_group": intent_group,
"entity_scope": entity_scope
} for intent_group, entity_scope in zip(intent_groups, entity_scopes)
]
def _get_entity_placeholders(dataset, language):
return {
e: _get_entity_name_placeholder(e, language) for e in dataset[ENTITIES]
}
def _get_entity_name_placeholder(entity_label, language):
return "%%%s%%" % "".join(tokenize_light(entity_label, language)).upper()
def _convert_dict_keys_to_int(dct):
if isinstance(dct, dict):
return {int(k): v for k, v in iteritems(dct)}
return dct
def _get_entities_combinations(entities):
yield ()
for nb_entities in reversed(range(1, len(entities) + 1)):
for combination in combinations(entities, nb_entities):
yield combination