GOATS/reguly.ipynb
2024-05-27 20:53:07 +02:00

12 KiB

from convlab.base_models.t5.nlu import T5NLU
import requests


def translate_text(text, target_language='en'):
    url = 'https://translate.googleapis.com/translate_a/single?client=gtx&sl=auto&tl={}&dt=t&q={}'.format(
        target_language, text)
    response = requests.get(url)
    if response.status_code == 200:
        translated_text = response.json()[0][0][0]
        return translated_text
    else:
        return None


class NaturalLanguageAnalyzer: 
    def predict(self, text, context=None):
        # Inicjalizacja modelu NLU
        model_name = "ConvLab/t5-small-nlu-multiwoz21"
        nlu_model = T5NLU(speaker='user', context_window_size=0, model_name_or_path=model_name)

        # Automatyczne tłumaczenie na język angielski
        translated_input = translate_text(text)

        # Wygenerowanie odpowiedzi z modelu NLU
        nlu_output = nlu_model.predict(translated_input)

        return nlu_output

    def init_session(self):
        # Inicjalizacja sesji (jeśli konieczne)
        pass
import json
import os
from convlab.dst.dst import DST
from convlab.dst.rule.multiwoz.dst_util import normalize_value


def default_state():
    return {
        'belief_state': {
            'hotel': {
                'info': {
                    'name': '',
                    'area': '',
                    'parking': '',
                    'price range': '',
                    'stars': '',
                    'internet': '',
                    'type': ''
                },
                'booking': {
                    'book stay': '',
                    'book day': '',
                    'book people': ''
                }
            }
        },
        'request_state': {},
        'history': [],
        'user_action': [],
        'system_action': [],
        'terminated': False,
        'booked': []
    }


class DialogueStateTracker(DST):
    def __init__(self):
        DST.__init__(self)
        self.state = default_state()
        with open('./hotels_data.json') as f:
            self.value_dict = json.load(f)

    def update(self, user_act=None):
        for intent, domain, slot, value in user_act:
            domain = domain.lower()
            intent = intent.lower()
            slot = slot.lower()
            
            if domain not in self.state['belief_state']:
                continue

            if intent == 'inform':
                if slot == 'none' or slot == '' or value == 'dontcare':
                    continue

                domain_dic = self.state['belief_state'][domain]['info']

                if slot in domain_dic:
                    nvalue = self.normalize_value(self.value_dict, domain, slot, value)
                    self.state['belief_state'][domain]['info'][slot] = nvalue

            elif intent == 'request':
                if domain not in self.state['request_state']:
                    self.state['request_state'][domain] = {}
                if slot not in self.state['request_state'][domain]:
                    self.state['request_state'][domain][slot] = 0

        return self.state

    def normalize_value(self, value_dict, domain, slot, value):
        normalized_value = value.lower().strip()
        if domain in value_dict and slot in value_dict[domain]:
            possible_values = value_dict[domain][slot]
            if isinstance(possible_values, dict) and normalized_value in possible_values:
                return possible_values[normalized_value]
        return value

    def init_session(self):
        self.state = default_state()
from collections import defaultdict
import copy
import json
from copy import deepcopy

from convlab.policy.policy import Policy
from convlab.util.multiwoz.dbquery import Database

db_path = './hotels_data.json'

class DialoguePolicy(Policy):
    def __init__(self):
        Policy.__init__(self)
        self.db = self.load_database(db_path)

    def load_database(self, db_path):
        with open(db_path, 'r', encoding='utf-8') as f:
            return json.load(f)

    def query(self, domain, constraints):
        if domain != 'hotel':
            return []
        
        results = []
        for entry in self.db:
            match = all(entry.get(key) == value for key, value in constraints)
            if match:
                results.append(entry)
        return results

    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.lower(), intent.lower())].append((slot.lower(), value))

        for user_act in user_action:
            self.update_system_action(user_act, user_action, state, system_action)

        if any(True for slots in user_action.values() for (slot, _) in slots if slot in ['book stay', 'book day', 'book people']):
            if self.results:
                system_action = {('Booking', 'Book'): [["Ref", self.results[0].get('Ref', 'N/A')]]}

        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]['info'].items() if value != '']
        print(f"Constraints: {constraints}")
        self.results = deepcopy(self.query(domain.lower(), constraints))
        print(f"Query results: {self.results}")

        if intent == 'request':
            if len(self.results) == 0:
                system_action[(domain, 'NoOffer')] = []
            else:
                for slot in user_action[user_act]:                  
                    if slot[0] in self.results[0]:
                        system_action[(domain, 'Inform')].append([slot[0], self.results[0].get(slot[0], 'unknown')])

        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 ["hotel"]:
                    system_action[(domain, 'Recommend')].append(['Name', choice['name']])
                    for slot in state['belief_state'][domain]['info']:
                        if choice.get(slot):
                            state['belief_state'][domain]['info'][slot] = choice[slot]
from convlab.nlg.template.multiwoz import TemplateNLG
from convlab.dialog_agent import PipelineAgent

nlu = NaturalLanguageAnalyzer()
dst = DialogueStateTracker()
policy = DialoguePolicy()
nlg = TemplateNLG(is_user=False)

agent = PipelineAgent(nlu=nlu, dst=dst, policy=policy, nlg=nlg, name='sys')
WARNING:root:nlu info_dict is not initialized
WARNING:root:dst info_dict is not initialized
WARNING:root:policy info_dict is not initialized
WARNING:root:nlg info_dict is not initialized
NLG seed 0
# nla = NaturalLanguageAnalyzer()
# nla_response = nla.predict("chciałbym zarezerwować drogi hotel bez parkingu 1 stycznia w Warszawie w centrum")
# print(nla_response)
# response = agent.response(nla_response)
# print(response)
response = agent.response("chciałbym zarezerwować drogi hotel z parkingiem 1 stycznia w Warszawie w centrum")
print(response)
Constraints: [('area', 'centre'), ('parking', 'yes'), ('price range', 'expensive'), ('type', 'hotel')]
Query results: [{'name': 'Four Seasons Hotel', 'area': 'centre', 'parking': 'yes', 'price range': 'expensive', 'stars': '5', 'internet': 'yes', 'type': 'hotel'}, {'name': 'The Ritz Hotel', 'area': 'centre', 'parking': 'yes', 'price range': 'expensive', 'stars': '5', 'internet': 'yes', 'type': 'hotel'}, {'name': 'The Savoy Hotel', 'area': 'centre', 'parking': 'yes', 'price range': 'expensive', 'stars': '5', 'internet': 'yes', 'type': 'hotel'}, {'name': 'Shangri-La Hotel', 'area': 'centre', 'parking': 'yes', 'price range': 'expensive', 'stars': '5', 'internet': 'yes', 'type': 'hotel'}]
We have 4 such places . Four Seasons Hotel looks like it would be a good choice .