from convlab2.dst.dst import DST from convlab2.dst.rule.multiwoz.dst_util import normalize_value from collections import defaultdict from convlab2.policy.policy import Policy from convlab2.util.multiwoz.dbquery import Database import copy from copy import deepcopy import json import os import jsgf # Natural Language Understanding class NLU: def __init__(self): self.grammars = [ jsgf.parse_grammar_file(f"JSGFs/{file_name}") for file_name in os.listdir("JSGFs") ] def get_dialog_act(self, rule): slots = [] self.get_slots(rule.expansion, slots) return {"act": rule.grammar.name, "slots": slots} def get_slots(self, expansion, slots): if expansion.tag != "": slots.append((expansion.tag, expansion.current_match)) return for child in expansion.children: self.get_slots(child, slots) if not expansion.children and isinstance(expansion, jsgf.NamedRuleRef): self.get_slots(expansion.referenced_rule.expansion, slots) def match(self, utterance): list_of_illegal_character = [",", ".", "'", "?", "!", ":", "-", "/"] for illegal_character in list_of_illegal_character[:-2]: utterance = utterance.replace(f"{illegal_character}", "") for illegal_character in list_of_illegal_character[-2:]: utterance = utterance.replace(f"{illegal_character}", " ") for grammar in self.grammars: matched = grammar.find_matching_rules(utterance) if matched: return self.get_dialog_act(matched[0]) return {"act": "null", "slots": []} class DP(Policy): def __init__(self): Policy.__init__(self) self.db = Database() def predict(self, state): self.results = [] system_action = defaultdict(list) user_action = defaultdict(list) for intent, domain, slot, value in state["user_action"]: user_action[(domain, intent)].append((slot, value)) for user_act in user_action: self.update_system_action(user_act, user_action, state, system_action) system_acts = [ [intent, domain, slot, value] for (domain, intent), slots in system_action.items() for slot, value in slots ] state["system_action"] = system_acts return system_acts def update_system_action(self, user_act, user_action, state, system_action): domain, intent = user_act constraints = [ (slot, value) for slot, value in state["belief_state"][domain.lower()]["semi"].items() if value != "" ] self.db.dbs = { "books": [ { "author": "autor", "title": "krew", "edition": "2020", "lang": "polski", }, { "author": "Marcin Bruczkowski", "title": "Bezsenność w Tokio", "genre": "reportaż", "publisher": "Społeczny Instytut Wydawniczy Znak", "edition": "2004", "lang": "polski", }, { "author": "Harari Yuval Noah", "title": "Sapiens Od zwierząt do bogów", "edition": "2011", "lang": "polski", }, { "author": "Haruki Murakami", "title": "1Q84", "edition": "2009", "lang": "polski", }, { "author": "Fiodor Dostojewski", "title": "Zbrodnia i Kara", "publisher": "Wydawnictwo Mg", "edition": "2015", "lang": "polski", }, ] } self.results = deepcopy(self.db.query(domain.lower(), constraints)) # Reguła 1 if intent == "Request": if len(self.results) == 0: system_action[(domain, "NoOffer")] = [] else: for slot in user_action[user_act]: kb_slot_name = ref[domain].get(slot[0], slot[0]) if kb_slot_name in self.results[0]: system_action[(domain, "Inform")].append( [slot[0], self.results[0].get(kb_slot_name, "unknown")] ) # Reguła 2 elif intent == "Inform": if len(self.results) == 0: system_action[(domain, "NoOffer")] = [] else: system_action[(domain, "Inform")].append( ["Choice", str(len(self.results))] ) choice = self.results[0] if domain in ["Book"]: system_action[(domain, "Recommend")].append( ["Title", choice["title"]] ) # Dialogue State Tracker class SDST(DST): def __init__(self): DST.__init__(self) self.state = { "user_action": [], "system_action": [], "belief_state": { "books": { "reserve": {"reservation": []}, "semi": { "title": "", "author": "", "genre": "", "publisher": "", "edition": "", "lang": "", }, }, "library": { "semi": { "location": "", "status": "", "events": "", "days": "", "phone number": "", } }, "card": {"semi": {"lost": "", "destroyed": "", "new": ""}}, "date": {"semi": {"day": "", "month": "", "year": ""}}, }, "request_state": { "reserve": {"reservation": []}, }, "reqmore_state": { "books": { "reserve": {"reservation": []}, "semi": { "title": "", "author": "", "genre": "", "publisher": "", "edition": "", "lang": "", }, }, "library": { "semi": { "location": "", "status": "", "events": "", "days": "", "phone number": "", } }, "card": {"semi": {"lost": "", "destroyed": "", "new": ""}}, "date": {"semi": {"day": "", "month": "", "year": ""}}, }, "terminated": False, "history": [], } self.ref = { "Books": { "Title": "title", "Author": "author", "Genre": "genre", "Publisher": "publisher", "Edition": "edition", "Lang": "lang", "None": "none", }, "Library": { "Location": "location", "Status": "status", "Events": "events", "Days": "days", "Phone number": "phone number", "None": "none", }, "Card": { "Lost": "lost", "Destroyed": "destroyed", "New": "new", "None": "none", }, "Date": {"Day": "day", "Month": "month", "Year": "year", "None": "none"}, } self.value_dict = json.load(open("value_dict.json")) def update(self, user_act=None): for intent, domain, slot, value in user_act: domain = domain.lower() intent = intent.lower() if domain in ["bye", "thankyou", "hello"]: continue if intent == "inform": k = self.ref[domain.capitalize()].get(slot, slot) if k is None: continue domain_dic = self.state["belief_state"][domain] if k in domain_dic["semi"]: nvalue = normalize_value(self.value_dict, domain, k, value) self.state["belief_state"][domain]["semi"][k] = value elif k in domain_dic["books"]: self.state["belief_state"][domain]["books"][k] = value elif k.lower() in domain_dic["books"]: self.state["belief_state"][domain]["books"][k.lower()] = value elif intent == "request": k = self.ref[domain.capitalize()].get(slot, slot) if k is None: continue if domain not in self.state["request_state"]: self.state["request_state"][domain] = {} self.state["request_state"]["reserve"]["reservation"].append(value) else: if ( value not in self.state["request_state"]["reserve"]["reservation"] ): self.state["request_state"]["reserve"]["reservation"].append( value ) if k not in self.state["request_state"][domain]: self.state["request_state"][domain][k] = value else: self.state["request_state"][domain][k] = value self.state["history"].append( self.state["request_state"]["reserve"]["reservation"][-1] ) elif intent == "reqmore": k = self.ref[domain.capitalize()].get(slot, slot) if k is None: continue domain_dic = self.state["reqmore_state"][domain] if k in domain_dic["semi"]: nvalue = normalize_value(self.value_dict, domain, k, value) self.state["reqmore_state"][domain]["semi"][k] = value elif k in domain_dic["books"]: self.state["reqmore_state"][domain]["books"][k] = value elif k.lower() in domain_dic["books"]: self.state["reqmore_state"][domain]["books"][k.lower()] = value elif intent == "negate": try: self.state["request_state"]["reserve"]["reservation"].remove( self.state["history"][-1] ) except Exception: pass elif intent == "affirm": self.state["belief_state"]["books"]["reserve"][ "reservation" ] = self.state["request_state"]["reserve"]["reservation"] else: continue return self.state def init_session(self): self.state = self.state # Natural Language Generator class NLG: def __init__(self, acts, arguments): self.acts = acts self.arguments = arguments def vectorToText(self, actVector): if actVector == [0, 0]: return "Witaj, nazywam się Mateusz." else: return "Przykro mi, nie zrozumiałem Cię" class Run: def __init__(self): self.acts = { 0: "hello", 1: "request", } self.arguments = {0: "name"} self.nlu = NLU() self.dp = DP(self.acts, self.arguments) self.nlg = NLG(self.acts, self.arguments) self.dst = DST(self.acts, self.arguments) def inputProcessing(self, command): act = self.nlu.analyze(command) self.dst.store(act) basic_act = self.dp.tacticChoice(self.dst.transfer()) return self.nlg.vectorToText(basic_act) # run = Run() # while(1): # message = input("Napisz coś: ") # print(run.inputProcessing(message))