Hi everyone. I’m a newbie. I’m using Rasa 3.6.18 and Rasa SDK 3.6.2, and I don’t know why the form validation action is interrupted when I add a custom tracker store to the endpoint.yml file.
**endpoint.yml**
tracker_store:
type: custom_tracker_store.CustomTrackerStore
url: localhost
port: 3306
username: root
password: 123456
dbname: responsumtest
**custom_tracker_store.py**
class ConversationTracker(Base):
__tablename__ = 'rasa_conversation_trackers'
id = Column(Integer, primary_key=True)
sender_id = Column('sender_id',String(255))
conversation_data = Column(Text)
message = Column(Text)
created = Column(DateTime, default=datetime.datetime.utcnow)
updated = Column(DateTime, default=datetime.datetime.utcnow)
deleted = Column(Boolean, unique=False, default=True)
class CustomTrackerStore(TrackerStore):
def __init__(self, domain, event_broker, **kwargs: Dict[Text, Any]) -> None:
super().__init__(domain, event_broker, **kwargs)
host = kwargs.get('host')
port = kwargs.get('port')
username = kwargs.get('username')
password = kwargs.get('password')
dbname = kwargs.get('dbname')
url = "mysql+pymysql://{0}:{1}@{2}:{3}/{4}".format(username, password, host, port, dbname)
self.engine = create_engine(url)
Base.metadata.create_all(self.engine)
self.Session = sessionmaker(bind=self.engine)
async def save(self, tracker: DialogueStateTracker) -> None:
latest_message = tracker.latest_message.text;
if self.event_broker:
self.stream_events(tracker)
conversation_data = tracker.as_dialogue().as_dict()
conversation_data = json.dumps(conversation_data)
conversation_tracker = ConversationTracker(sender_id=tracker.sender_id, conversation_data=conversation_data, message=latest_message)
with self.Session() as session:
session.add(conversation_tracker)
session.commit()
async def retrieve(self, sender_id) -> Optional[DialogueStateTracker]:
with self.Session() as session:
conversation = session.query(ConversationTracker).filter_by(sender_id=sender_id).order_by(ConversationTracker.id.desc()).first()
if conversation:
try:
events_json = json.loads(conversation.conversation_data)
tracker = DialogueStateTracker.from_dict(sender_id, [events_json], self.domain.slots)
return tracker
except Exception as e:
logging.error(traceback.format_exc())
return None
else:
return None
**stories.yml**
- story: request_find_salary
steps:
- intent: salary
- action: find_salary_form
- active_loop: find_salary_form
- slot_was_set:
- requested_slot: employee_name
- active_loop: null
- slot_was_set:
- employee_name: null
- action: action_find_salary_form
- intent: goodbye
- action: utter_goodbye
**rules.yml**
- rule: Activate Salary Form
steps:
- intent: salary
- action: find_salary_form
- active_loop: find_salary_form
- rule: Submit Salary Form
condition:
- active_loop: find_salary_form
steps:
- action: find_salary_form
- active_loop: null
- slot_was_set:
- requested_slot: null
- action: action_find_salary_form
**actions.py**
class ActionFindSalaryForm(Action):
def name(self) -> Text:
return "action_find_salary_form"
@staticmethod
def salary_data(key):
headers = {
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10.12; rv:55.0) Gecko/20100101 Firefox/55.0',
}
response_API = requests.get('https://reqres.in/api/users?page=0', headers=headers)
status = response_API.status_code
if status==200:
data = response_API.content
return json.loads(data)
return json.loads({})
def run(self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: Dict[Text, Any]) -> List[Dict[Text, Any]]:
employee_name = tracker.get_slot("employee_name")
if employee_name:
parse_json = self.salary_data()
salary = 0
for em in parse_json['data']:
if em['email']==employee_name:
salary = em['first_name'] + " " + em['last_name']
dispatcher.utter_message(text="Full name: " + str(salary))
else:
dispatcher.utter_message(text="Sorry! Not found your salary.")
return []
class ValidateFindSalaryForm(FormValidationAction):
def name(self) -> Text:
return "validate_find_salary_form"
def extract_employee_name(
self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: DomainDict
) -> Dict[Text, Any]:
employee_name = tracker.latest_message.get("text")
slot_name = tracker.slots["requested_slot"]
if slot_name=="employee_name":
return {"employee_name": employee_name}
return {"employee_name": None}
def validate_employee_name(
self,
slot_value: Any,
dispatcher: CollectingDispatcher,
tracker: Tracker,
domain: DomainDict,
) -> Dict[Text, Any]:
return {"employee_name": slot_value}