import jsgf from tabulate import tabulate from flair.data import Sentence, Token from flair.datasets import SentenceDataset from flair.models import SequenceTagger import random import torch random.seed(42) torch.manual_seed(42) if torch.cuda.is_available(): torch.cuda.manual_seed(0) torch.cuda.manual_seed_all(0) torch.backends.cudnn.enabled = False torch.backends.cudnn.benchmark = False torch.backends.cudnn.deterministic = True class ML_NLU: def __init__(self, acts, arguments): self.acts = acts self.arguments = arguments self.slot_model, self.frame_model = self.setup() def nolabel2o(self, line, i): return 'O' if line[i] == 'NoLabel' else line[i] def conllu2flair(self, sentences, label=None): fsentences = [] for sentence in sentences: fsentence = Sentence() for token in sentence: ftoken = Token(token['form']) if label: ftoken.add_tag(label, token[label]) fsentence.add_token(ftoken) fsentences.append(fsentence) return SentenceDataset(fsentences) def predict(self, sentence): csentence = [{'form': word} for word in sentence] fsentence = self.conllu2flair([csentence])[0] self.slot_model.predict(fsentence) self.frame_model.predict(fsentence) possible_intents = {} for token in fsentence: for intent in token.annotation_layers["frame"]: if(intent.value in possible_intents): possible_intents[intent.value] += intent.score else: possible_intents[intent.value] = intent.score return [(token, ftoken.get_tag('slot').value) for token, ftoken in zip(sentence, fsentence)], max(possible_intents) def setup(self): slot_model = SequenceTagger.load('slot-model/final-model.pt') frame_model = SequenceTagger.load('frame-model/final-model.pt') return slot_model, frame_model def test_nlu(self, utterance): if utterance: slots, act = self.predict(utterance.split()) slots = [x for x in slots if x[1] != 'O'] arguments = self.convert_slot_to_argument(slots) return {'act': act, 'slots': arguments} else: return 'Critical Error' def convert_slot_to_argument(self, slots): arguments = [] candidate = None for slot in slots: if slot[1].startswith("B-"): if(candidate != None): arguments.append(candidate) candidate = [slot[1].replace("B-", ""), slot[0]] if slot[1].startswith("I-") and candidate != None and slot[1].endswith(candidate[0]): candidate[1] += " " + slot[0] if(candidate != None): arguments.append(candidate) return [(x[0], x[1]) for x in arguments] class Book_NLU: #Natural Language Understanding """ Moduł odpowiedzialny za analizę tekstu. W wyniku jego działania tekstowa reprezentacja wypowiedzi użytkownika zostaje zamieniona na jej reprezentację semantyczną, najczęściej w postaci ramy. Wejście: Tekst Wyjście: Akt użytkownika (rama) """ def __init__(self, acts, arguments, book_grammar): self.acts = acts self.arguments = arguments self.book_grammar = book_grammar 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 analyze(self, text): """ Analiza Tekstu wprowadzonego przez użytkownika i zamiana na akt (rama) """ print("Analiza Tekstu: " + text) act = "(greetings()&request(name))" print("Akt to: " + act) #przerobienie na wektor act_vector = [[0],[1,0]] #1 wektor to greetings, a 2 wektor to request z argumentem "name" print("Zamiana na: ") print(act_vector) return act_vector def test_nlu(self, utterance): matched = self.book_grammar.find_matching_rules(utterance) print(matched) if matched: return self.get_dialog_act(matched[0]) else: return {'act': 'null', 'slots': []} class DST: #Dialogue State Tracker """ Moduł odpowiedzialny za śledzenie stanu dialogu. Przechowuje informacje o tym jakie dane zostały uzyskane od użytkownika w toku prowadzonej konwersacji. Wejście: Akt użytkownika (rama) Wyjście: Reprezentacja stanu dialogu (rama) """ def __init__(self, acts, arguments): self.acts = acts self.arguments = arguments self.frame_list= [] def store(self, rama): """ Dodanie nowego aktu do listy """ print("\nDodanie do listy nowej ramy: ") print(rama) self.frame_list.append(rama) def transfer(self): print("Przekazanie dalej listy ram: ") print(self.frame_list) return self.frame_list class DP: """ Moduł decydujący o wyborze kolejnego aktu, który ma podjąć system prowadząc rozmowę. Wejście: Reprezentacja stanu dialogu (rama) Wyjście: Akt systemu (rama) """ def __init__(self, acts, arguments): self.acts = acts self.arguments = arguments def choose_tactic(self, frame_list): """ Obieranie taktyki na podstawie aktów usera. Bardzo ważna jest kolejność dodawanych do frame_list wartości. """ act_vector = [0, 0] return act_vector class NLG: """ Moduł, który tworzy reprezentację tekstową aktu systemowego wybranego przez taktykę dialogu. Wejście: Akt systemu (rama) Wyjście: Tekst """ def __init__(self, acts, arguments): self.acts = acts self.arguments = arguments def change_to_text(self, act_vector): """ Funkcja zamieniająca akt systemu na tekst rozumiany przez użytkownika. """ if(act_vector == [0, 0]): return "Cześć, mam na imię Janet" return "Nie rozumiem" class Janet: def __init__(self): self.acts={ 0: "greetings", 1: "request", } self.arguments={ 0: "name" } self.nlg = NLG(self.acts, self.arguments) self.dp = DP(self.acts, self.arguments) self.dst = DST(self.acts, self.arguments) self.nlu = Book_NLU(self.acts, self.arguments, jsgf.parse_grammar_file('book.jsgf')) self.nlu_v2 = ML_NLU(self.acts, self.arguments) def test(self, command): out = self.nlu_v2.test_nlu(command) return out def process(self, command): act = self.nlu.analyze(command) self.dst.store(act) dest_act = self.dp.choose_tactic(self.dst.transfer()) return self.nlg.change_to_text(dest_act) janet = Janet() print(janet.test('chciałbym się umówić na wizytę do Piotra Pająka na jutro')) #Testowy print na start while(1): print('\n') text = input("Wpisz tekst: ") print(janet.test(text))