Source code for rasa_core.interpreter

importaiohttpimportjsonimportloggingimportreimportosfromtypingimportText,List,Dict,Anyfromrasa.coreimportconstantsfromrasa.core.utilsimportEndpointConfigfromrasa.core.constantsimportINTENT_MESSAGE_PREFIXlogger=logging.getLogger(__name__)classNaturalLanguageInterpreter(object):asyncdefparse(self,text,message_id=None):raiseNotImplementedError("Interpreter needs to be able to parse ""messages into structured output.")@staticmethoddefcreate(obj,endpoint=None):ifisinstance(obj,NaturalLanguageInterpreter):returnobjifnotisinstance(obj,str):ifobjisnotNone:logger.warning("Tried to create NLU interpreter ""from '{}', which is not possible.""Using RegexInterpreter instead.""".format(obj))returnRegexInterpreter()# default interpreterifnotendpoint:returnRasaNLUInterpreter(model_directory=obj)name_parts=os.path.split(obj)iflen(name_parts)==1:# using the default project namereturnRasaNLUHttpInterpreter(name_parts[0],endpoint)eliflen(name_parts)==2:returnRasaNLUHttpInterpreter(name_parts[1],endpoint,name_parts[0])else:raiseException("You have configured an endpoint to use for ""the NLU model. To use it, you need to ""specify the model to use with ""`--nlu project/model`.")classRegexInterpreter(NaturalLanguageInterpreter):@staticmethoddefallowed_prefixes():returnINTENT_MESSAGE_PREFIX@staticmethoddef_create_entities(parsed_entities,sidx,eidx):entities=[]fork,vsinparsed_entities.items():ifnotisinstance(vs,list):vs=[vs]forvalueinvs:entities.append({"entity":k,"start":sidx,"end":eidx,# can't be more specific"value":value})returnentities@staticmethoddef_parse_parameters(entitiy_str:Text,sidx:int,eidx:int,user_input:Text)->List[Dict[Text,Any]]:ifentitiy_strisNoneornotentitiy_str.strip():# if there is nothing to parse we will directly exitreturn[]try:parsed_entities=json.loads(entitiy_str)ifisinstance(parsed_entities,dict):returnRegexInterpreter._create_entities(parsed_entities,sidx,eidx)else:raiseException("Parsed value isn't a json object ""(instead parser found '{}')"".".format(type(parsed_entities)))exceptExceptionase:logger.warning("Invalid to parse arguments in line ""'{}'. Failed to decode parameters""as a json object. Make sure the intent""followed by a proper json object. ""Error: {}".format(user_input,e))return[]@staticmethoddef_parse_confidence(confidence_str:Text)->float:ifconfidence_strisNone:return1.0try:returnfloat(confidence_str.strip()[1:])exceptExceptionase:logger.warning("Invalid to parse confidence value in line ""'{}'. Make sure the intent confidence is an ""@ followed by a decimal number. ""Error: {}".format(confidence_str,e))return0.0def_starts_with_intent_prefix(self,text):forcinself.allowed_prefixes():iftext.startswith(c):returnTruereturnFalse@staticmethoddefextract_intent_and_entities(user_input:Text)->object:"""Parse the user input using regexes to extract intent & entities."""prefixes=re.escape(RegexInterpreter.allowed_prefixes())# the regex matches "slot{"a": 1}"m=re.search('^['+prefixes+']?([^{@]+)(@[0-9.]+)?([{].+)?',user_input)ifmisnotNone:event_name=m.group(1).strip()confidence=RegexInterpreter._parse_confidence(m.group(2))entities=RegexInterpreter._parse_parameters(m.group(3),m.start(3),m.end(3),user_input)returnevent_name,confidence,entitieselse:logger.warning("Failed to parse intent end entities from ""'{}'. ".format(user_input))returnNone,0.0,[]asyncdefparse(self,text,message_id=None):"""Parse a text message."""intent,confidence,entities=self.extract_intent_and_entities(text)ifself._starts_with_intent_prefix(text):message_text=textelse:message_text=INTENT_MESSAGE_PREFIX+textreturn{'text':message_text,'intent':{'name':intent,'confidence':confidence,},'intent_ranking':[{'name':intent,'confidence':confidence,}],'entities':entities,}

[docs]asyncdefparse(self,text,message_id=None):"""Parse a text message. Return a default value if the parsing of the text failed."""default_return={"intent":{"name":"","confidence":0.0},"entities":[],"text":""}result=awaitself._rasa_http_parse(text,message_id)returnresultifresultisnotNoneelsedefault_return

[docs]asyncdefparse(self,text,message_id=None):"""Parse a text message. Return a default value if the parsing of the text failed."""ifself.lazy_initandself.interpreterisNone:self._load_interpreter()result=self.interpreter.parse(text)# TODO: hotfix to append attributes that NLU is adding as a server# but where the interpreter does not add themifresult:result["model"]="current"result["project"]="default"returnresult