###############################################################################
# PyDial: Multi-domain Statistical Spoken Dialogue System Software
###############################################################################
#
# Copyright 2015 - 2019
# Cambridge University Engineering Department Dialogue Systems Group
#
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
###############################################################################
'''
SimulatedUsersManager.py - combines simulated components into functional simulator
==================================================================================
Copyright CUED Dialogue Systems Group 2015 - 2017
.. seealso:: CUED Imports/Dependencies:
import :mod:`usersimulator.UserModel` |.|
import :mod:`usersimulator.ErrorSimulator` |.|
import :mod:`ontology.Ontology` |.|
import :mod:`ontology.OntologyUtils` |.|
import :mod:`utils.Settings` |.|
import :mod:`utils.DiaAct` |.|
import :mod:`utils.ContextLogger`
************************
'''
__author__ = "cued_dialogue_systems_group"
from usersimulator import UserModel
from usersimulator import ErrorModel
from utils import DiaAct, Settings, ContextLogger
from ontology import Ontology, OntologyUtils
logger = ContextLogger.getLogger('')
class DomainsSimulatedUser(object):
'''User Simulator for a single domain. Comprised of a behaviour component: :class:`UserModel` to produce a semantic act
and an error simulator to produce from the semantic act a list of semantic hypotheses.
:param str: domain string
'''
def __init__(self, domainString, error_rate):
'''
'''
self.um = UserModel.UM(domainString)
self.error_simulator = ErrorModel.DomainsErrorSimulator(domainString, error_rate)
self.randomLearning = False
if Settings.config.has_option("mogp_"+domainString, "randomweightlearning"):
self.randomLearning = Settings.config.getboolean("mogp_"+domainString, "randomweightlearning")
if self.randomLearning:
import policy.morl.WeightGenerator as wg
self.weightGen = wg.WeightGenerator(domainString)
def restart(self, otherDomainsConstraints):
'''Resets all components (**User Model**) that are statefull.
:param otherDomainsConstraints: of domain goal tuples (slot=val)
:type otherDomainsConstraints: list
:returns: None
'''
self.um.init(otherDomainsConstraints)
if self.randomLearning:
self.weightGen.updateWeights()
def act_on(self, sys_act_string):
'''Thru the UserModel member, receives the system action and then responds.
:param sys_act_string: system action
:type sys_act_string: unicode str
:returns: (str) user action
'''
sys_act = DiaAct.DiaAct(sys_act_string)
self.um.receive(sys_act)
user_act = self.um.respond()
return user_act
class SimulatedUsersManager(object):
"""
The multidomain simulated user, which is made up of a dictionary of simulated users indexed by domain.
:param (list): of domain strings
"""
def __init__(self, error_rate):
self.possible_domains = Ontology.global_ontology.possible_domains
if Settings.config.has_option("GENERAL", "testdomains"):
self.possible_domains = Settings.config.get("GENERAL", "testdomains").split(',')
if Settings.config.has_option("GENERAL", "traindomains"):
self.possible_domains = Settings.config.get("GENERAL", "traindomains").split(',')
self.error_rate = error_rate
logger.info('Simulating with error rate: ' + str(error_rate))
self.simUserManagers = dict.fromkeys(OntologyUtils.available_domains, None)
# DEFAULTS:
self.MIN_DOMAINS_PER_DIALOG = 1
self.MAX_DOMAINS_PER_DIALOG = 3
self.INCLUDE_DOMAIN_PROB = 0.6
self.CONDITIONAL_BEHAVIOUR = False
self.forceNullPositive = False
self.traceDialog = 2
self.temp_domains = []
# CONFIG OPTIONS:
if Settings.config.has_option("simulate", "forcenullpositive"):
self.forceNullPositive = Settings.config.getboolean("simulate", "forcenullpositive")
if Settings.config.has_option("simulate","includedomainprob"):
self.INCLUDE_DOMAIN_PROB = Settings.config.getfloat("simulate","includedomainprob")
assert(self.INCLUDE_DOMAIN_PROB <= 1.0 and self.INCLUDE_DOMAIN_PROB > 0)
if Settings.config.has_option("simulate","maxdomainsperdialog"):
self.MAX_DOMAINS_PER_DIALOG = Settings.config.getint("simulate","maxdomainsperdialog")
if Settings.config.has_option("simulate","mindomainsperdialog"):
self.MIN_DOMAINS_PER_DIALOG = Settings.config.getint("simulate","mindomainsperdialog")
assert(self.MIN_DOMAINS_PER_DIALOG <= self.MAX_DOMAINS_PER_DIALOG)
assert(self.MIN_DOMAINS_PER_DIALOG <= len(Ontology.global_ontology.possible_domains))
if Settings.config.has_option("conditional","conditionalsimuser"):
self.CONDITIONAL_BEHAVIOUR = Settings.config.getboolean("conditional","conditionalsimuser")
if Settings.config.has_option("GENERAL", "tracedialog"):
self.traceDialog = Settings.config.getint("GENERAL", "tracedialog")
if Settings.config.has_option("simulate", "domainsampling"):
self.domainSampling = Settings.config.get("simulate", "domainsampling")
else:
self.domainSampling = "random"
self.sim_level = 'dial_act'
if Settings.config.has_option("usermodel", "simlevel"):
self.sim_level = Settings.config.get("usermodel", "simlevel")
def set_allowed_codomains(self, ROOTDOMAIN):
"""
Sets member (list) *allowed_codomains* given a root domain name (ie the domain of the first constraint)
Uses the hardcoded rules in Settings.py to do so. Also, based on determined allowed_codomains, sets
the probability of each being included, independently.
:param ROOTDOMAIN: domain tag
:type ROOTDOMAIN: str
:returns: None
"""
if self.CONDITIONAL_BEHAVIOUR and ROOTDOMAIN in OntologyUtils.ALLOWED_CODOMAIN_RULES:
#NB: These next few lines depend on hard coding of OntologyUtils.ALLOWED_CODOMAIN_RULES
self.allowed_codomains = OntologyUtils.ALLOWED_CODOMAIN_RULES[ROOTDOMAIN]
if self.allowed_codomains is not None:
if len(self.allowed_codomains) > 1:
if isinstance(self.allowed_codomains[1],list):
randindex = Settings.random.randint(0,len(OntologyUtils.ALLOWED_CODOMAIN_RULES[ROOTDOMAIN]))
self.allowed_codomains = OntologyUtils.ALLOWED_CODOMAIN_RULES[ROOTDOMAIN][randindex]
else:
self.allowed_codomains = []
# based on the allowed_codomains, set the prob of each one independently being in the dialog:
#self.INCLUDE_DOMAIN_PROB = min(0.3,1.0/len(self.allowed_codomains))
else:
self.allowed_codomains = list(self.possible_domains)
self.allowed_codomains.remove(ROOTDOMAIN)
return
def sample_domains(self):
"""Randomly select a set of domains from those available.
The selected domains will be used by the simulated user over a single dialog.
:param None:
:returns: None
"""
if self.domainSampling == "random":
# sample from possible_domains
self.using_domains = []
self.using_domains.append(Settings.random.choice(self.possible_domains)) # must have at least 1 element
root_domain = self.using_domains[0] # the first chosen domain - will affect which codomains can be partnered with
self.set_allowed_codomains(ROOTDOMAIN=root_domain)
shuffled_possible_domains = list(self.possible_domains)
Settings.random.shuffle(shuffled_possible_domains)
for dstring in shuffled_possible_domains:
if len(self.using_domains) == self.MAX_DOMAINS_PER_DIALOG:
break
if dstring not in self.allowed_codomains:
continue
if dstring in self.using_domains:
continue
if len(self.using_domains) < self.MIN_DOMAINS_PER_DIALOG:
self.using_domains.append(dstring)
elif Settings.random.rand() < self.INCLUDE_DOMAIN_PROB:
self.using_domains.append(dstring)
# Note - we may not have met the Min domains limit at this point - for example the allowed codomains for the root_domain
# may have been too low.
if len(self.using_domains) < self.MIN_DOMAINS_PER_DIALOG:
logger.warning("Found {} domains only, which is less than the config set minimum domains per dialog of {}".format(\
len(self.using_domains), self.MIN_DOMAINS_PER_DIALOG))
Settings.random.shuffle(self.using_domains) # list order is persistent. Simulated user will act in this order now.
logger.info('Order sim user will execute goals:'+str(self.using_domains))
elif self.domainSampling == "roundrobin":
if self.MIN_DOMAINS_PER_DIALOG > 1:
logger.warning('With "roundrobin" domains ampling "mindomainsperdialog" cannot be larger than 1, setting it to 1')
self.MIN_DOMAINS_PER_DIALOG = 1
if self.MAX_DOMAINS_PER_DIALOG > 1:
logger.warning('With "roundrobin" domain sampling "maxdomainsperdialog" cannot be larger than 1, setting it to 1')
self.MAX_DOMAINS_PER_DIALOG = 1
if self.temp_domains == []:
self.temp_domains = list(self.possible_domains)
Settings.random.shuffle(self.temp_domains)
self.using_domains=[self.temp_domains.pop()]
self.number_domains_this_dialog = len(self.using_domains)
self.uncompleted_domains = list(self.using_domains)
return
def restart(self):
"""Restarts/boots up the selected domains simulated user components. Shuts down those running and not needed for
the next dialog.
:param None:
:returns: None
"""
# sample domain for this dialog and randomise order:
self.sample_domains()
# reset domain simulators:
otherDomainsConstraints = [] # used to conditionally generate domain goals
for dstring in self.using_domains: # doing this way to generate goals/behaviour in an order.
# fire up or check if it is running
if self.sim_level == 'sys2text':
if self.simUserManagers[dstring] is None:
from usersimulator.US_sys2text.Sys2TextSimulatedUser import Sys2TextSimulatedUser
self.simUserManagers[dstring] = Sys2TextSimulatedUser(dstring)
self.simUserManagers[dstring].restart()
else:
if self.simUserManagers[dstring] is None:
self.simUserManagers[dstring] = DomainsSimulatedUser(dstring, self.error_rate)
self.simUserManagers[dstring].restart(otherDomainsConstraints)
# DEBUG prints to inspect goals we have generated:
if self.sim_level != 'sys2text':
logger.debug(str(self.simUserManagers[dstring].um.goal))
logger.debug(str(self.simUserManagers[dstring].um.goal.copied_constraints))
logger.debug(str(self.simUserManagers[dstring].um.hdcSim.agenda.agenda_items))
logger.debug("DOMAIN-----"+dstring)
#raw_input('goal and agenda for domain '+dstring)
if self.CONDITIONAL_BEHAVIOUR and self.sim_level != 'sys2text':
otherDomainsConstraints += self.simUserManagers[dstring].um.goal.constraints
for dstring in self.possible_domains: #STATELESS, no memory here. Fine to actually `kill' domains not using
if dstring not in self.using_domains:
self.simUserManagers[dstring] = None
return
def act_on(self, sys_act):
""" First produce a single semantic act from the simulated user. Then create from it a list of semantic hypotheses which
include simulated errors.
"""
user_act, user_actsDomain = self._user_act(sys_act)
if self.sim_level != 'sys2text':
hyps = self._confuse_user_act_and_enforce_null(user_act, user_actsDomain)
else:
hyps = None
return user_act, user_actsDomain, hyps
def _user_act(self, sys_act):
'''Produces the next user semantic act from the simulated user. Also returns the domain that the act came from
--> which avoids doing topictracking during simulate
'''
# TODO - this is just a start. lots needs thinking about here.
# -- needs to return the current simulation domain explictly for now
# return dstring too - use this information for error simulation.
logger.debug('simulated users uncompleted domains:'+str(self.uncompleted_domains))
for dstring in self.using_domains:
if dstring in self.uncompleted_domains:
user_act = self.simUserManagers[dstring].act_on(sys_act)
if self.sim_level != 'sys2text':
if 'bye(' in user_act.to_string():
sys_act = 'hello()'
self.uncompleted_domains.remove(dstring)
if len(self.uncompleted_domains):
continue
else:
break
else:
break
return user_act, dstring
def _confuse_user_act_and_enforce_null(self, user_act, user_actsDomain):
'''Simulate errors in the semantic parses. Returns a set of confused hypotheses.
Also enforces a null() act if config set to do so.
'''
# Confused user act.
hyps = self.simUserManagers[user_actsDomain].error_simulator.confuse_act(user_act)
null_prob = 0.0
for h in hyps:
act = h.to_string()
prob = h.P_Au_O
if act == 'null()':
null_prob += prob
if self.traceDialog>1:
print(' Semi >', act, '[%.6f]' % prob)
logger.info('| Semi > %s [%.6f]' % (act,prob))
if self.forceNullPositive and null_prob < 0.001:
nullAct = DiaAct.DiaActWithProb('null()')
nullAct.P_Au_O = 0.001
hyps.append(nullAct)
if self.traceDialog>1:
print('| Semi > null() [0.001]')
logger.info(' Semi > null() [0.001]')
return hyps
# END OF FILE