问答机器人框架中非常重要的几个概念为意图识别、槽位填充
通常使用机器学习算法进行意图的分类&＃xff0c;但最终我们是让将用户的输入提取出我们想要的核心信息。--槽位提取
因此slot在rasa中极为重要, 如何我们想要深入自定义slot&＃xff0c;以及solt前后逻辑&＃xff0c;那么FormAction是必须要了解清楚的。
位于rasa_sdk/forms.py
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")
源码中可以看到name&＃xff0c;required_slots&＃xff0c;是必须实现的函数&＃xff0c;作用是告诉rasa 此action都有哪些slot&＃xff0c;此aciton名为什么&＃xff0c;rasa官方demo中也如此进行了简介
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",]
那么rasa将按照required_slots list顺序进行slot填充
流程是一致的&＃xff0c;我们在story中定义意图以及匹配此意图是的机器人行为&＃xff0c;
我们想要使用自定义的action&＃xff0c;需要在 domain.yml 和stories.md中进行配置
## sales form
* contact_sales- sales_form - form{"name": "sales_form"} - form{"name": null}
domain中定义fromaction
当我们运行 rasa run&＃xff0c; 和 新开一个窗口 rasa run actions时&＃xff0c;会进行sales_form注册到rasa中
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
重点知识-story中的定义的何时触发action&＃xff0c; domain.yml 中定义。
运行之后&＃xff0c;发现会自动识别每一个slot槽位的&＃xff0c;向用户提问。这里是因为formaction会自动循环required_slots中的slot&＃xff0c;同时自动匹配 domain.yml utter_ask_&＃43; 意图。
源码中此处进行了定义
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
因此我们可以重写request_next_slot&＃xff0c;做前后slot信息验证
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}
函数定义以 validate_ &＃43; slot rasa将自动进行输入校验
源码定义如下&＃xff1a;
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()]
定义特定槽位的用户输入验证&＃xff0c;有时候还是无法满足我们需求&＃xff0c;我们得自定义request_next_slot&＃xff0c; 方法中放飞自我
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
最后两个重要方法为 slot_mappings&＃xff0c; 和 submit
slot_mappings 正如函数名称定义的那样&＃xff0c;允许用户进行slot槽位提取时的复杂方式。
比如用户输入 确认意图&＃xff0c;填充true&＃xff0c;输入否定意图&＃xff0c;填充false等
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"),],}
最后一个action结束时候&＃xff0c;进行最后处理 submit函数为&＃xff0c;当收集到所有的意图slot&＃xff0c;最终处理方法。
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 []
以上&＃xff0c;我们完成了FormAction 自定义处理&＃xff0c; 可以帮助我们实现复杂的多轮次对话。
京公网安备 11010802041100号