370 lines
12 KiB
Python
370 lines
12 KiB
Python
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
|
|
from convlab2.util.multiwoz.multiwoz_slot_trans import REF_SYS_DA, REF_USR_DA
|
|
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_USR_DA[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))
|