今年五月份参加Oracle开发者大会,在会议上看到智能AI在运维方面的应用场景;讲师现场展现了一款能够结合上下文对话的智能AI,通过聊天方式完成运维工作。
会议后对该款智能AI机器人念念不忘,由于人工智能AI学习成本较高,寻思着是否能够写一套低配版运维交互机器人;
思考
初期期望该机器人能够：
通过手机能够处理简单的故障
不智能但至少配置能够灵活变更
有了具体的目标, 再考虑具体实现方案, 主要思考几个点：
应用载体
我期望这个载体是一款常用的手机APP;现有环境中微信企业号适合干这个事情, 且官网有各种API文档, 实施起来不是个什么巨大挑战.
安全性
涉及到运维平台,控制了运维平台就相当于控制了所有服务器;所以关系到运维平台的安全问题不可小窥,得确保在交互过程中的安全,在交互过程中需要加密,对不信任服务器进行策略管控.
灵活性
可以通过配置文件方式进行配置,后续随着功能模块增加可以随时进行更改,考虑到使用配置文件方式可能太过单一,花里胡哨的功能可能无法满足实现,尽量考虑又能花里胡哨,又能灵活管理配置的方案.
对话上下文
一般而言,通讯都需要一个长连接保证通信期间双方可以收发数据包; 考虑到一个对话就得专门起一个线程进行通信,这样不但增加开发难度,且更消耗资源, 权衡利弊后,对于上下文管理这一部分尽量选用非实时性方案去做.
架构
列出思考的几个关键点后,对整体的设计进行深入思考,几经思考后:
采用微信企业号作为应用载体
有关于企业号的开发传送门.安全加固
接口平台只放通腾讯服务器IP访问.运维平台开放接口平台白名单访问,并且采用Python itsdangerous生成安全令牌进行通信交互.程序设计思想
采用树结构设计模式,每个分叉为一个功能.这样就不必担心无法完成花里胡哨的操作,又能够灵活变更.持久化存储接收信息
对每个用户发送的信息进行存储,并作出快速响应.Redis对于这个场景非常适用,既能够存储信息又十分高效.架构图看起来大概是这样:
实现
接收企业号信息API代码片段展示
# 引用企业微信JDK from WXcrypt.WXBizMsgCrypt import WXBizMsgCrypt def 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
接着,编写一个功能树的类