前言
今年五月份參加Oracle開發者大會,在會議上看到智能AI在運維方面的應用場景;講師現場展現了一款能夠結合上下文對話的智能AI,通過聊天方式完成運維工作。
會議后對該款智能AI機器人念念不忘,由于人工智能AI學習成本較高,尋思著是否能夠寫一套低配版運維交互機器人;
思考
初期期望該機器人能夠：
有了具體的目標, 再考慮具體實現方案, 主要思考幾個點：
應用載體
我期望這個載體是一款常用的手機APP;現有環境中微信企業號適合干這個事情, 且官網有各種API文檔, 實施起來不是個什么巨大挑戰.
安全性
涉及到運維平臺,控制了運維平臺就相當于控制了所有服務器;所以關系到運維平臺的安全問題不可小窺,得確保在交互過程中的安全,在交互過程中需要加密,對不信任服務器進行策略管控.
靈活性
可以通過配置文件方式進行配置,后續隨著功能模塊增加可以隨時進行更改,考慮到使用配置文件方式可能太過單一,花里胡哨的功能可能無法滿足實現,盡量考慮又能花里胡哨,又能靈活管理配置的方案.
對話上下文
一般而言,通訊都需要一個長連接保證通信期間雙方可以收發數據包; 考慮到一個對話就得專門起一個線程進行通信,這樣不但增加開發難度,且更消耗資源, 權衡利弊后,對于上下文管理這一部分盡量選用非實時性方案去做.
架構
列出思考的幾個關鍵點后,對整體的設計進行深入思考,幾經思考后:
采用微信企業號作為應用載體
有關于企業號的開發傳送門.
安全加固
接口平臺只放通騰訊服務器IP訪問.運維平臺開放接口平臺白名單訪問,并且采用Python itsdangerous生成安全令牌進行通信交互.
程序設計思想
采用樹結構設計模式,每個分叉為一個功能.這樣就不必擔心無法完成花里胡哨的操作,又能夠靈活變更.
持久化存儲接收信息
對每個用戶發送的信息進行存儲,并作出快速響應.Redis對于這個場景非常適用,既能夠存儲信息又十分高效.
架構圖看起來大概是這樣:
實現
接收企業號信息API代碼片段展示
# 引用企業微信JDKfrom WXcrypt.WXBizMsgCrypt import WXBizMsgCryptdef work_weixin_api(request): # 獲取微信Post參數 msg_signature = request.GET.get('msg_signature', '') timestamp = request.GET.get('timestamp', '') nonce = request.GET.get('nonce', '') echostr = request.GET.get('echostr', '') # 構造微信信息解析方法 wxcpt = WXBizMsgCrypt(WXTOKEN, WXENCODINGAESKEY, WXCROPID) if request.method == 'POST': eagle_branch = request.POST.get('eagle_branch', 'master') if eagle_branch == "master": request_data = request.body # 解析接收到的文本 ret, msg = wxcpt.DecryptMsg(request_data, msg_signature, timestamp, nonce) request_xml = ET.fromstring(msg) # 獲取信息內容 content = request_xml.find("Content").text # 獲取信息類型 msg_type = request_xml.find("MsgType").text # 獲取發送人 from_user = request_xml.find("FromUserName").text else: content = request.POST.get('content', '') from_user = request.POST.get('from_user', '')安全令牌生成
# 加密def enc_dict(d): # 加密 s = URLSafeSerializer('1234') st = s.dumps(d) # 加密后再生成基于時間戳的令牌 t = TimestampSigner('4567') ts = t.sign(st) return ts功能樹設計代碼片段展示
先定義一個功能樹基類
# 菜單功能的基類class Function: def __init__(self, data): self._data = data self._functions = [] # 傳入的方法的描述 def __str__(self): return str(self._data()) # 返回當前對象類型 def f_type(self): return self._data.f_type # 返回當前對象 def getData(self): return self._data # 返回所有子菜單 def getFunctions(self): return self._functions # 新增子菜單 def add(self, function): self._functions.append(function) # 遞歸搜索 def go(self, num): for _, i in enumerate(self._functions): if int(num) == _ : return i return None
由于是在手機上操作, 那么交互內容盡可能簡單,所以采用全數字交互方式.
在樹結構設計模式下,所有操作都是在遞歸搜尋,對于其他特殊的輸入,例如端口 確認驗證碼之類的無法實現.
在這里需要有小小的改動
# 新增一個類型屬性 def f_type(self): return self._data.f_type # 遞歸搜索 def go(self, num): for _, i in enumerate(self._functions): f_type = i._data().f_type # 如果類型是默認且存在列表中，或動態生成類型的，直接返回 if f_type == "default" and int(num) == _ or f_type == "dynamic": return i return None
接著,編寫一個功能樹的類
class Menu: def __init__(self): self._head = Function(FunctionNodeBase()) self.input_text = None # 鏈接 def linkToHead(self, function): self._head.add(function) # 搜索 def search(self, text): cur = self._head for i in text.split('-'): if cur.go(i) == None: return None else: self.input_text = i cur = cur.go(i) return cur葉子 跟 樹的主體都有了,下面來創建樹頂
展示： 基礎功能葉 動態功能葉 靜態功能葉
# 空的功能Nodeclass FunctionNodeBase: __metaclass__ = ABCMeta def __init__(self, user=None, f_type="default", input_text=None, sub_text=None): self.user = user self.sub_text = sub_text self.input_text = input_text self.f_type = f_type self.f_mark = [] # 菜單通過run方法執行與生成文本 @abstractmethod def run(self): return self.__str__() # 描述 @abstractmethod def __str__(self): return "菜單樹頂層"# 動態生成class SelectDeploymentTop(FunctionNodeBase): # 動態生成的菜單需要聲明f_type def __init__(self): super().__init__() self.f_type = "dynamic" def run(self): text = "請選擇事業部/n/n" deployment_list = [i for i in FunctionList.keys()] for _, i in enumerate(deployment_list): self.f_mark.append(_) text += "%s %s/n" % (_, i) return text # 微信顯示的文本信息 def __str__(self): return "選擇事業部"# 靜態class MySQLFunctionTop(FunctionNodeBase): def __init__(self): super().__init__() def run(self): text = "您選擇的是%s,請選擇您想要操作:/n" % str(self.__str__()) text += "%s/n" % self.sub_text return text def __str__(self): return "MySQL操作"
效果圖,第一層功能展示
將需要的功能逐一寫好后需要進行注冊
def api(tid,user): # 實例化 menu = Menu() top = Function(SelectDeploymentTop) function_top = Function(FunctionTop) mysql_top = Function(MySQLFunctionTop) # 鏈接 top.add(function_top function_top.add(mysql_top) # 關聯菜單樹 menu.linkToHead(top) # 遞歸搜索 function = menu.search(tid)
Redis存儲對話代碼片段
class redis_db: def __init__(self): # 按符號隔開 self.mark = '-' self.redis_db = redis.StrictRedis( host = host, port=6379, db=1, decode_responses=True) # 默認回話過期600秒,每次存儲 '-'隔開 def add(self,key,text,Timeout=600): if key not in self.keys(): self.redis_db.set(key,'',ex=Timeout) if self.get(key): self.redis_db.append(key,self.mark) self.redis_db.append(key,text)
同理,返回上層就刪除一格; 退出即刪除該KEY的值.
成果
下圖為：通過交互機器人連接k8s增加POD數的應用場景
后記
該系統已經在平臺上穩定運行大半年, 上線后使運維人員能夠更高效快速解決日常中遇到的一些故障.
以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支持武林網。
新聞熱點
疑難解答
