Hi, I am trying to train a bot having domain.yml
intents:
- alarm name
- impact
- probable cause
- general_info
- prerequisites
- Default Welcome Intent
- alarm id
- Default Fallback Intent
entities:
- alarm_id
- alarm_name
slots:
alarm_id:
type:list
alarm_name:
type:text
actions:
- utter_greet
- utter_ask_alarm_id
- utter_ask_alarm_name
- utter_default_fallback
- action_general_info
- action_impact
- action_probable_cause
- action_alarm_name
- action_prerequisites
- action_alarm_id
templates:
utter_greet:
- text: 'Hi, How can I help you!'
- text: 'Welcome, Please let me know how I could be a help today!'
- text: 'Hello, How can I help you!'
utter_ask_alarm_name:
- text: 'Please provide a valid alarm name!'
- text: 'Please share an alarm name for your query!'
- text: 'Please tell me alarm name!'
utter_ask_alarm_id:
- text: 'Please provide the alarm id!'
- text: 'Let me know the alarm id for your query!'
- text: 'Please help me with the alarm id for your query!'
utter_default_fallback:
- text: 'I am sorry for your query! I have been trained for the queries related Alarm Clearing like "Alarm Name", "Alarm ID", "Impact of an alarm", "Prerequsites for clearing an alarm"'
- text: 'I did not get that. Can you say it again? I can help you with Fault Management - Alarm Clearing for Release'
- text: 'I missed what you said. What was that? Please ask your queries related to Fault Management - Alarm Clearing for Release'
- text: 'Sorry, could you say that again? I can help you with Fault Management - Alarm Clearing'
and my action file looks like this:
from typing import Text, Dict, Any, List, Union
from rasa_sdk import Action
from rasa_sdk.events import SlotSet
import textwrap
from grakn.client import GraknClient
from alarm_check import dic_alarm_name
# Action Class for action_general_info
class ActionGeneralInfo(Action):
def name(self):
return "action_general_info"
def run(self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: Dict[Text, Any]) -> List[Dict[Text, Any]]:
try:
slot_alarm_name = tracker.get_slot('alarm_name')
# Storing the alarm name from alarm dictionary
slot_alarm_name = dic_alarm_name.get(slot_alarm_name, default = slot_alarm_name)
slot_alarm_id = tracker.get_slot('alarm_id')[0] if len(tracker.get_slot('alarm_id')) > 0 else '0'
except:
slot_alarm_name = ""
slot_alarm_id = "0"
general_info_query = 'match $a isa alarm,\
has alarm_name $alarm_name,\
has alarm_id $alarm_id,\
has filename $file_name,\
has page $page,\
has probable_cause $probabale_cause;\
{$a has alarm_name "' + slot_alarm_name + '";} or {$a has alarm_id "' + slot_alarm_id + '";};\
get $alarm_name,$alarm_id,$probabale_cause,$file_name,$page ; offset 0; limit 1;'
with GraknClient(uri="localhost:48555") as client:
with client.session(keyspace = "demo_db") as session:
with session.transaction().read() as read_transaction:
answer_iterator = read_transaction.query(general_info_query)
result = []
for structured_answer in answer_iterator:
var_map = structured_answer.map()
#print("######","\n",var_map, "\n", "#####")
var_value_dict = {var_name: var_map[var_name].value() for var_name in var_map}
result.append(var_value_dict)
if len(result) > 0:
return_result = "Alarm Name: " + result[0]['alarm_name'] +"\r\n"+ "Alarm ID: " + result[0]['alarm_id'] + "\r\n" + "Alarm Reason: " + textwrap.shorten(text = result[0]['probabale_cause'], width = 350) + '\n' +'Read More http://18.224.75.148/Ciena/Doc/' +\
result[0]['file_name'] + '#page=' + result[0]['page']
else:
return_result = "Sorry, could not find anything in Knowledge base"
dispatcher.utter_message(return_result)
return[]
#return [SlotSet("alarm_name", slot_alarm_name), SlotSet("alarm_id", slot_alarm_id)]
# Action Class for action_impact
class ActionImpact(Action):
def name(self):
return "action_impact"
def run(self,
dispatcher: CollectingDispatcher,
tracker: Tracker,
domain: Dict[Text, Any]) -> List[Dict[Text, Any]]:
try:
slot_alarm_name = tracker.get_slot('alarm_name')
# Storing the alarm name from alarm dictionary
slot_alarm_name = dic_alarm_name.get(slot_alarm_name, default = slot_alarm_name)
slot_alarm_id = tracker.get_slot('alarm_id')[0] if len(tracker.get_slot('alarm_id')) > 0 else '0'
except:
slot_alarm_name = ""
slot_alarm_id = "0"
# Impact query for Grakn
impact_query = 'match $a isa alarm,\
has alarm_name $an,\
has alarm_id $id,\
has filename $file_name,\
has page $page,\
has impact $impact;\
{$a has alarm_name "' + slot_alarm_name + '";} or {$a has alarm_id "' + slot_alarm_id + '";};\
get $impact,$file_name,$page ; offset 0; limit 1;'
with GraknClient(uri="localhost:48555") as client:
with client.session(keyspace = "demo_db") as session:
with session.transaction().read() as read_transaction:
answer_iterator = read_transaction.query(impact_query)
result = []
for structured_answer in answer_iterator:
var_map = structured_answer.map()
#print("######","\n",var_map, "\n", "#####")
var_value_dict = {var_name: var_map[var_name].value() for var_name in var_map}
result.append(var_value_dict)
if len(result) > 0:
return_result = textwrap.shorten(text = result[0]['impact'], width = 350) + '\n' +'Read More http://18.224.75.148/Ciena/Doc/' +\
result[0]['file_name'] + '#page=' + result[0]['page']
else:
return_result = "Sorry! Could not find anything for this alarm in Knowledge Graph. Try with a different alarm!"
dispatcher.utter_message(return_result)
return[]
#return [SlotSet("alarm_name", slot_alarm_name), SlotSet("alarm_id", slot_alarm_id)]
# Action Class for Probable Cause
class ActionProbableCause(Action):
def name(self):
return "action_probable_cause"
def run(self,
dispatcher: CollectingDispatcher,
tracker: Tracker,
domain: Dict[Text, Any]) -> List[Dict[Text, Any]]:
try:
slot_alarm_name = tracker.get_slot('alarm_name')
# Storing the alarm name from alarm dictionary
slot_alarm_name = dic_alarm_name.get(slot_alarm_name, default = slot_alarm_name)
slot_alarm_id = tracker.get_slot('alarm_id')[0] if len(tracker.get_slot('alarm_id')) > 0 else '0'
except:
slot_alarm_name = ""
slot_alarm_id = "0"
# Probable Cause Grakn Query
probable_cause_query = 'match $a isa alarm,\
has alarm_name $an,\
has alarm_id $id,\
has filename $file_name,\
has page $page,\
has probable_cause $probable_cause;\
{$a has alarm_name "' + slot_alarm_name + '";} or {$a has alarm_id "' + slot_alarm_id + '";};\
get $probable_cause, $file_name, $page; offset 0; limit 1;'
with GraknClient(uri="localhost:48555") as client:
with client.session(keyspace = "demo_db") as session:
with session.transaction().read() as read_transaction:
answer_iterator = read_transaction.query(probable_cause_query)
result = []
for structured_answer in answer_iterator:
var_map = structured_answer.map()
#print("######","\n",var_map, "\n", "#####")
var_value_dict = {var_name: var_map[var_name].value() for var_name in var_map}
result.append(var_value_dict)
if len(result) > 0:
return_result = textwrap.shorten(text = result[0]['probable_cause'], width = 350) + '\n' +'Read More http://18.224.75.148/Ciena/Doc/' +\
result[0]['file_name'] + '#page=' + result[0]['page']
else:
return_result = "Sorry, could not find anything in Knowledge base"
dispatcher.utter_message(return_result)
return[]
#return [SlotSet("alarm_name", slot_alarm_name), SlotSet("alarm_id", slot_alarm_id)]
# Action Class Alarm Name
class ActionAlarmName(Action):
def name(self):
return "action_alarm_name"
def run(self,
dispatcher: CollectingDispatcher,
tracker: Tracker,
domain: Dict[Text, Any]) -> List[Dict[Text, Any]]:
try:
slot_alarm_id = tracker.get_slot('alarm_id')[0] if len(tracker.get_slot('alarm_id')) > 0 else '0'
except:
slot_alarm_id = "0"
# Alarm Name Grakn Query
alarm_name_query = 'match $a isa alarm,\
has alarm_id $id,\
has filename $file_name,\
has page $page,\
has alarm_name $alarm_name;\
$id "' + slot_alarm_id + '";\
get $alarm_name, $file_name, $page;'
# Connecting to Grakn
with GraknClient(uri="localhost:48555") as client:
with client.session(keyspace = "demo_db") as session:
with session.transaction().read() as read_transaction:
answer_iterator = read_transaction.query(alarm_name_query)
result = []
for structured_answer in answer_iterator:
var_map = structured_answer.map()
#print("######","\n",var_map, "\n", "#####")
var_value_dict = {var_name: var_map[var_name].value() for var_name in var_map}
result.append(var_value_dict)
if len(result) > 0:
return_result = textwrap.shorten(text = result[0]['alarm_name'], width = 350) + '\n' +'Read More http://18.224.75.148/Ciena/Doc/' +\
result[0]['file_name'] + '#page=' + result[0]['page']
else:
return_result = "Sorry, could not find anything in Knowledge base"
dispatcher.utter_message(return_result)
return[]
#return [SlotSet("alarm_id", slot_alarm_id)]
# Action Class Alarm Id
class ActionAlarmId(Action):
def name(self):
return "action_alarm_id"
def run(self,
dispatcher: CollectingDispatcher,
tracker: Tracker,
domain: Dict[Text, Any]) -> List[Dict[Text, Any]]:
try:
slot_alarm_name = tracker.get_slot('alarm_name')
# Storing the alarm name from alarm dictionary
slot_alarm_name = dic_alarm_name.get(slot_alarm_name, default = slot_alarm_name)
except:
slot_alarm_name = ""
# Alarm Id Grakn Query
alarm_id_query = 'match $a isa alarm,\
has alarm_name $alarm_name,\
has filename $file_name,\
has page $page,\
has alarm_id $alarm_id;\
$alarm_name"' + slot_alarm_name + '";\
get $alarm_id, $file_name, $page;'
# Connecting to Grakn
with GraknClient(uri="localhost:48555") as client:
with client.session(keyspace = "demo_db") as session:
with session.transaction().read() as read_transaction:
answer_iterator = read_transaction.query(alarm_id_query)
result = []
for structured_answer in answer_iterator:
var_map = structured_answer.map()
#print("######","\n",var_map, "\n", "#####")
var_value_dict = {var_name: var_map[var_name].value() for var_name in var_map}
result.append(var_value_dict)
if len(result) > 0:
return_result = textwrap.shorten(text = (",".join(([i['alarm_id'] for i in result]))), width = 350) + '\n' +'Read More http://18.224.75.148/Ciena/Doc/' +\
result[0]['file_name'] + '#page=' + result[0]['page']
else:
return_result = "Sorry, could not find anything in Knowledge base"
dispatcher.utter_message(return_result)
return[]
#return [SlotSet("alarm_id", slot_alarm_id)]
# Action Class Prerequisites
class ActionPrerequisites(Action):
def name(self):
return action_prerequisites
def run(self,
dispatcher: CollectingDispatcher,
tracker: Tracker,
domain: Dict[Text, Any]) -> List[Dict[Text, Any]]:
try:
slot_alarm_name = tracker.get_slot('alarm_name')
# Storing the alarm name from alarm dictionary
slot_alarm_name = dic_alarm_name.get(slot_alarm_name, default = slot_alarm_name)
slot_alarm_id = tracker.get_slot('alarm_id')[0] if len(tracker.get_slot('alarm_id')) > 0 else '0'
except:
slot_alarm_name = ""
slot_alarm_id = "0"
# Prerequisites query for Grakn
prerequisites_query = 'match $a isa alarm,\
has alarm_name $an,\
has alarm_id $id,\
has filename $file_name,\
has page $page,\
has prerequisites $prerequisites;\
{$a has alarm_name "' + slot_alarm_name + '";} or {$a has alarm_id "' + slot_alarm_id + '";};\
get $prerequisites,$file_name,$page ; offset 0; limit 1;'
with GraknClient(uri="localhost:48555") as client:
with client.session(keyspace = "demo_db") as session:
with session.transaction().read() as read_transaction:
answer_iterator = read_transaction.query(prerequisites_query)
result = []
for structured_answer in answer_iterator:
var_map = structured_answer.map()
#print("######","\n",var_map, "\n", "#####")
var_value_dict = {var_name: var_map[var_name].value() for var_name in var_map}
result.append(var_value_dict)
if len(result) > 0:
return_result = textwrap.shorten(text = result[0]['prerequisites'], width = 350) + '\n' +'Read More http://18.224.75.148/Ciena/Doc/' +\
result[0]['file_name'] + '#page=' + result[0]['page']
else:
return_result = "Sorry, could not find anything in Knowledge base"
dispatcher.utter_message(return_result)
return[]
#return [SlotSet("alarm_name", slot_alarm_name), SlotSet("alarm_id", slot_alarm_id)]
After doing
rasa train
I am getting an error
Error:
(rasa) ubuntu@ip-172-31-4-130:~/C-Rasa$ rasa train
Traceback (most recent call last):
File "/home/ubuntu/anaconda3/envs/rasa/bin/rasa", line 10, in <module>
sys.exit(main())
File "/home/ubuntu/anaconda3/envs/rasa/lib/python3.6/site-packages/rasa/__main__.py", line 76, in main
cmdline_arguments.func(cmdline_arguments)
File "/home/ubuntu/anaconda3/envs/rasa/lib/python3.6/site-packages/rasa/cli/train.py", line 77, in train
kwargs=extract_additional_arguments(args),
File "/home/ubuntu/anaconda3/envs/rasa/lib/python3.6/site-packages/rasa/train.py", line 40, in train
kwargs=kwargs,
File "uvloop/loop.pyx", line 1417, in uvloop.loop.Loop.run_until_complete
File "/home/ubuntu/anaconda3/envs/rasa/lib/python3.6/site-packages/rasa/train.py", line 75, in train_async
domain = await file_importer.get_domain()
File "/home/ubuntu/anaconda3/envs/rasa/lib/python3.6/site-packages/rasa/importers/importer.py", line 242, in get_domain
domains = await asyncio.gather(*domains)
File "/home/ubuntu/anaconda3/envs/rasa/lib/python3.6/site-packages/rasa/importers/rasa.py", line 65, in get_domain
domain = Domain.load(self._domain_path)
File "/home/ubuntu/anaconda3/envs/rasa/lib/python3.6/site-packages/rasa/core/domain.py", line 62, in load
other = cls.from_path(path)
File "/home/ubuntu/anaconda3/envs/rasa/lib/python3.6/site-packages/rasa/core/domain.py", line 72, in from_path
domain = cls.from_file(path)
File "/home/ubuntu/anaconda3/envs/rasa/lib/python3.6/site-packages/rasa/core/domain.py", line 85, in from_file
return cls.from_yaml(rasa.utils.io.read_file(path))
File "/home/ubuntu/anaconda3/envs/rasa/lib/python3.6/site-packages/rasa/core/domain.py", line 95, in from_yaml
return cls.from_dict(data)
File "/home/ubuntu/anaconda3/envs/rasa/lib/python3.6/site-packages/rasa/core/domain.py", line 100, in from_dict
slots = cls.collect_slots(data.get("slots", {}))
File "/home/ubuntu/anaconda3/envs/rasa/lib/python3.6/site-packages/rasa/core/domain.py", line 187, in collect_slots
slot_class = Slot.resolve_by_type(slot_dict[slot_name].get("type"))
AttributeError: 'str' object has no attribute 'get'
Any help would be appreciated
Thanks