rasa算法_智能问答rasa框架02自定义action

class FormAction(Action):def name(self) -> Text:"""Unique identifier of the form"""raise NotImplementedError("A form must implement a name")&＃64;staticmethoddef required_slots(tracker: "Tracker") -> List[Text]:"""A list of required slots that the form has to fill.Use &＃96;tracker&＃96; to request different list of slotsdepending on the state of the dialogue"""raise NotImplementedError("A form must implement required slots that it has to fill")

class SalesForm(FormAction):"""Collects sales information and adds it to the spreadsheet"""def name(self) -> Text:return "sales_form"&＃64;staticmethoddef required_slots(tracker) -> List[Text]:return ["job_function","use_case","budget","person_name","company","business_email",]

## sales form
* contact_sales- sales_form - form{"name": "sales_form"} - form{"name": null}

forms:- sales_form

python ../start.py run actions
2020-09-07 14:03:54 INFO rasa_sdk.endpoint - Starting action endpoint server...
2020-09-07 14:03:54 INFO rasa_sdk.executor - Registered function for &＃39;sales_form&＃39;.
2020-09-07 14:03:54 INFO rasa_sdk.endpoint - Action endpoint is up and running on http://localhost:5055

def request_next_slot(self,dispatcher: "CollectingDispatcher",tracker: "Tracker",domain: Dict[Text, Any],) -> Optional[List[EventType]]:"""Request the next slot and utter template if needed,else return None"""for slot in self.required_slots(tracker):if self._should_request_slot(tracker, slot):logger.debug(f"Request next slot &＃39;{slot}&＃39;")dispatcher.utter_message(template&＃61;f"utter_ask_{slot}", **tracker.slots)return [SlotSet(REQUESTED_SLOT, slot)]# no more required slots to fillreturn None

def validate_business_email(self, value, dispatcher, tracker, domain) -> Dict[Text, Any]:"""Check to see if an email entity was actually picked up by duckling."""if any(tracker.get_latest_entity_values("email")):# entity was picked up, validate slotreturn {"business_email": value}else:# no entity was picked up, we want to ask againdispatcher.utter_message(template&＃61;"utter_no_email")return {"business_email": None}

async def validate_slots(self,slot_dict: Dict[Text, Any],dispatcher: "CollectingDispatcher",tracker: "Tracker",domain: Dict[Text, Any],) -> List[EventType]:"""Validate slots using helper validation functions.Call validate_{slot} function for each slot, value pair to be validated.If this function is not implemented, set the slot to the value."""for slot, value in list(slot_dict.items()):validate_func &＃61; getattr(self, f"validate_{slot}", lambda *x: {slot: value})if utils.is_coroutine_action(validate_func):validation_output &＃61; await validate_func(value, dispatcher, tracker, domain)else:validation_output &＃61; validate_func(value, dispatcher, tracker, domain)if not isinstance(validation_output, dict):logger.warning("Returning values in helper validation methods is deprecated. "&＃43; f"Your &＃96;validate_{slot}()&＃96; method should return "&＃43; "a dict of {&＃39;slot_name&＃39;: value} instead.")validation_output &＃61; {slot: validation_output}slot_dict.update(validation_output)# validation succeed, set slots to extracted valuesreturn [SlotSet(slot, value) for slot, value in slot_dict.items()]

def request_next_slot(self,dispatcher: "CollectingDispatcher",tracker: "Tracker",domain: Dict[Text, Any],) -> Optional[List[EventType]]:"""Request the next slot and utter template if needed,else return None"""for slot in self.required_slots(tracker):print(slot)if slot &＃61;&＃61; "person_name" and tracker.get_slot("person_name") &＃61;&＃61; "yxp":dispatcher.utter_message(text&＃61;"对不起&＃xff0c;不欢迎你&＃xff0c;走开&＃xff01;&＃xff01;")return self.deactivate()if self._should_request_slot(tracker, slot):print("_should_request_slot")## Condition of validated slot that triggers deactivationif slot &＃61;&＃61; "person_name" and tracker.get_slot("person_name") &＃61;&＃61; "yxp":dispatcher.utter_message(text&＃61;"对不起&＃xff0c;不欢迎你&＃xff0c;走开&＃xff01;&＃xff01;")return self.deactivate()## For all other slots, continue as usuallogger.debug(f"Request next slot &＃39;{slot}&＃39;")dispatcher.utter_message(template&＃61;f"utter_ask_{slot}", ** tracker.slots)REQUESTED_SLOT &＃61; "requested_slot"return [SlotSet(REQUESTED_SLOT, slot)]print("not _should_request_slot")return None

def slot_mappings(self) -> Dict[Text, Union[Dict, List[Dict]]]:"""A dictionary to map required slots to- an extracted entity- intent: value pairs- a whole messageor a list of them, where a first match will be picked"""return {"job_function": [self.from_entity(entity&＃61;"job_function"),self.from_text(intent&＃61;"enter_data"),],"use_case": [self.from_text(intent&＃61;"enter_data")],"budget": [self.from_entity(entity&＃61;"amount-of-money"),self.from_entity(entity&＃61;"number"),self.from_text(intent&＃61;"enter_data"),],"person_name": [self.from_entity(entity&＃61;"name"),self.from_text(intent&＃61;"enter_data"),],"company": [self.from_entity(entity&＃61;"company"),self.from_text(intent&＃61;"enter_data"),],"business_email": [self.from_entity(entity&＃61;"email"),self.from_text(intent&＃61;"enter_data"),],}

def submit(self,dispatcher: CollectingDispatcher,tracker: Tracker,domain: Dict[Text, Any],) -> List[Dict]:person_name &＃61; tracker.get_slot("person_name")import datetimebudget &＃61; tracker.get_slot("budget")company &＃61; tracker.get_slot("company")email &＃61; tracker.get_slot("business_email")job_function &＃61; tracker.get_slot("job_function")person_name &＃61; tracker.get_slot("person_name")use_case &＃61; tracker.get_slot("use_case")date &＃61; datetime.datetime.now().strftime("%d/%m/%Y")sales_info &＃61; [company, use_case, budget, date, person_name, job_function, email]print(sales_info)dispatcher.utter_message(template&＃61;"utter_confirm_salesrequest", user_name&＃61;person_name)return []