# encoding:utf-8 import json import os import uuid from uuid import getnode as get_mac import requests import plugins from bridge.context import ContextType from bridge.reply import Reply, ReplyType from common.log import logger from plugins import * """利用百度UNIT实现智能对话 如果命中意图，返回意图对应的回复，否则返回继续交付给下个插件处理 """ @plugins.register( name="BDunit", desire_priority=0, hidden=True, desc="Baidu unit bot system", version="0.1", author="jackson", ) class BDunit(Plugin): def __init__(self): super().__init__() try: conf = super().load_config() if not conf: raise Exception("config.json not found") self.service_id = conf["service_id"] self.api_key = conf["api_key"] self.secret_key = conf["secret_key"] self.access_token = self.get_token() self.handlers[Event.ON_HANDLE_CONTEXT] = self.on_handle_context logger.info("[BDunit] inited") except Exception as e: logger.warn("[BDunit] init failed, ignore ") raise e def on_handle_context(self, e_context: EventContext): if e_context["context"].type != ContextType.TEXT: return content = e_context["context"].content logger.debug("[BDunit] on_handle_context. content: %s" % content) parsed = self.getUnit2(content) intent = self.getIntent(parsed) if intent: # 找到意图 logger.debug("[BDunit] Baidu_AI Intent= %s", intent) reply = Reply() reply.type = ReplyType.TEXT reply.content = self.getSay(parsed) e_context["reply"] = reply e_context.action = EventAction.BREAK_PASS # 事件结束，并跳过处理context的默认逻辑 else: e_context.action = EventAction.CONTINUE # 事件继续，交付给下个插件或默认逻辑 def get_help_text(self, **kwargs): help_text = "本插件会处理询问实时日期时间，天气，数学运算等问题，这些技能由您的百度智能对话UNIT决定\n" return help_text def get_token(self): """获取访问百度UUNIT 的access_token #param api_key: UNIT apk_key #param secret_key: UNIT secret_key Returns: string: access_token """ url = "https://aip.baidubce.com/oauth/2.0/token?client_id={}&client_secret={}&grant_type=client_credentials".format(self.api_key, self.secret_key) payload = "" headers = {"Content-Type": "application/json", "Accept": "application/json"} response = requests.request("POST", url, headers=headers, data=payload) # print(response.text) return response.json()["access_token"] def getUnit(self, query): """ NLU 解析version 3.0 :param query: 用户的指令字符串 :returns: UNIT 解析结果。如果解析失败，返回 None """ url = "https://aip.baidubce.com/rpc/2.0/unit/service/v3/chat?access_token=" + self.access_token request = { "query": query, "user_id": str(get_mac())[:32], "terminal_id": "88888", } body = { "log_id": str(uuid.uuid1()), "version": "3.0", "service_id": self.service_id, "session_id": str(uuid.uuid1()), "request": request, } try: headers = {"Content-Type": "application/json"} response = requests.post(url, json=body, headers=headers) return json.loads(response.text) except Exception: return None def getUnit2(self, query): """ NLU 解析 version 2.0 :param query: 用户的指令字符串 :returns: UNIT 解析结果。如果解析失败，返回 None """ url = "https://aip.baidubce.com/rpc/2.0/unit/service/chat?access_token=" + self.access_token request = {"query": query, "user_id": str(get_mac())[:32]} body = { "log_id": str(uuid.uuid1()), "version": "2.0", "service_id": self.service_id, "session_id": str(uuid.uuid1()), "request": request, } try: headers = {"Content-Type": "application/json"} response = requests.post(url, json=body, headers=headers) return json.loads(response.text) except Exception: return None def getIntent(self, parsed): """ 提取意图 :param parsed: UNIT 解析结果 :returns: 意图数组 """ if parsed and "result" in parsed and "response_list" in parsed["result"]: try: return parsed["result"]["response_list"][0]["schema"]["intent"] except Exception as e: logger.warning(e) return "" else: return "" def hasIntent(self, parsed, intent): """ 判断是否包含某个意图 :param parsed: UNIT 解析结果 :param intent: 意图的名称 :returns: True: 包含; False: 不包含 """ if parsed and "result" in parsed and "response_list" in parsed["result"]: response_list = parsed["result"]["response_list"] for response in response_list: if "schema" in response and "intent" in response["schema"] and response["schema"]["intent"] == intent: return True return False else: return False def getSlots(self, parsed, intent=""): """ 提取某个意图的所有词槽 :param parsed: UNIT 解析结果 :param intent: 意图的名称 :returns: 词槽列表。你可以通过 name 属性筛选词槽， 再通过 normalized_word 属性取出相应的值 """ if parsed and "result" in parsed and "response_list" in parsed["result"]: response_list = parsed["result"]["response_list"] if intent == "": try: return parsed["result"]["response_list"][0]["schema"]["slots"] except Exception as e: logger.warning(e) return [] for response in response_list: if "schema" in response and "intent" in response["schema"] and "slots" in response["schema"] and response["schema"]["intent"] == intent: return response["schema"]["slots"] return [] else: return [] def getSlotWords(self, parsed, intent, name): """ 找出命中某个词槽的内容 :param parsed: UNIT 解析结果 :param intent: 意图的名称 :param name: 词槽名 :returns: 命中该词槽的值的列表。 """ slots = self.getSlots(parsed, intent) words = [] for slot in slots: if slot["name"] == name: words.append(slot["normalized_word"]) return words def getSayByConfidence(self, parsed): """ 提取 UNIT 置信度最高的回复文本 :param parsed: UNIT 解析结果 :returns: UNIT 的回复文本 """ if parsed and "result" in parsed and "response_list" in parsed["result"]: response_list = parsed["result"]["response_list"] answer = {} for response in response_list: if ( "schema" in response and "intent_confidence" in response["schema"] and (not answer or response["schema"]["intent_confidence"] > answer["schema"]["intent_confidence"]) ): answer = response return answer["action_list"][0]["say"] else: return "" def getSay(self, parsed, intent=""): """ 提取 UNIT 的回复文本 :param parsed: UNIT 解析结果 :param intent: 意图的名称 :returns: UNIT 的回复文本 """ if parsed and "result" in parsed and "response_list" in parsed["result"]: response_list = parsed["result"]["response_list"] if intent == "": try: return response_list[0]["action_list"][0]["say"] except Exception as e: logger.warning(e) return "" for response in response_list: if "schema" in response and "intent" in response["schema"] and response["schema"]["intent"] == intent: try: return response["action_list"][0]["say"] except Exception as e: logger.warning(e) return "" return "" else: return ""