前言
今年五月份参加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
接着,编写一个功能树的类
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
叶子 跟 树的主体都有了,下面来创建树顶
展示： 基础功能叶 动态功能叶 静态功能叶
# 空的功能node
class 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数的应用场景
后记
该系统已经在平台上稳定运行大半年, 上线后使运维人员能够更高效快速解决日常中遇到的一些故障.
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持移动技术网。
如对本文有疑问， 点击进行留言回复！！
offset、client、scroll (width,height、left,top、X,Y)
网友评论