1. 问答机器人的组成-基于知识图谱的搜索
- 在教育场景下，若学生有关于学习内容的提问，或业务层面的提问，则要求问答机器人的回答必须精准，来满足业务的要求
- 因此需要通过知识图谱来快速检索，所提内容的相关信息，并针对这些相关信息进行处理，得到精准的回答
- 本次内容中直接应用了知识图谱基本功能，关系寻找及关系提取
- 后续学习中会提供更多的应用
2. 伪代码实现
from py2neo import Graph #导入neo4j
from pyhanlp import * #导入nlp的工具包
from random import choice
#知识图谱的初步应用
class GraphSearch():
def __init__(self):
self.graph = Graph("http://localhost:7474", username="graph.db", password="lesson1")
self.iswho_sql = "profile match p=(n)<-[r]-(b) where n.name='%s' return n.name,r.name,b.name"
self.isrelation_sql = "profile match p=(n)<-[r]-(b) where n.name=~'%s' and b.name=~'%s' return n.name,r.name,b.name"
def search_answer(self,question):
#使用HanLP进行词性判断
sentence = HanLP.parseDependency(question)
#后续可以替换成自己训练的模块首先针对句意进行分析，其次针对目标实体进行提取；但主要也是针对业务场景进行分析和处理
seg = {}
res_combine = ''
for word in sentence.iterator():
##只处理nr名词：人，v动词，n名词，针对进行提问进行词性分析
if word.POSTAG[0] == 'n' or word.POSTAG in ['v','r']:
if word.POSTAG not in seg:
seg[word.POSTAG] = [word.LEMMA]
else:
seg[word.POSTAG].append(word.LEMMA)
#简单基于词性和内容判断是否为目标句式'A是谁'以此使用知识图谱进行回答
if 'v' in seg and '是' in seg['v']:
if 'r' in seg and 'nr' in seg and '谁' in seg['r']:
for person in seg['nr']:
res = self.graph.run(self.iswho_sql%(person)).data()
res_combine = []
for i in res[:10]:
res_combine.append('%s是:%s%s'%(i['n.name'],i['b.name'],i['r.name']))
return choice(res_combine)
#基于词性和内容判断是否为目标句式'A和B的关系'以此使用知识图谱进行回答
if 'n' in seg and '关系' in seg['n']:
if len(seg['nr']) == 2:
res1 = self.graph.run(self.isrelation_sql%(seg['nr'][1],seg['nr'][0])).data()
if res1 != []:
res_combine = seg['nr'][0]+'的'+res2[0]['r.name']+'是'+seg['nr'][1]
return res_combine
res2 = self.graph.run(self.isrelation_sql%(seg['nr'][0],seg['nr'][1])).data()
if res2 != []:
res_combine = seg['nr'][1]+'的'+res2[0]['r.name']+'是'+seg['nr'][0]
return res_combine
if res_combine == '':
return None
后续在完成业务场景下的实体提取和意图识别，学习及模型训练后，丰富该功能的问答
3. 问答机器人的组成-基于网络的回答
- 在对话机器人构建的初期，常常面临数据不足，导致机器人无法进行准确回答，因此在前期会适当调用第三方对话的接口，来进行回答
- 以此防止突然冷场或机器人对话失灵
- 后续使用生成式的对话机器人，来解决此类问题
import requests
class InterNet():
def __init__(self):
pass
def search_answer(self,question):
url = 'https://api.ownthink.com/bot?appid=xiaosi&userid=user&spoken='
try:
text = requests.post(url+question).json()
if 'message' in text and text['message'] == 'success':
return text['data']['info']['text']
else:
return None
except:
return None
4. 服务的构建-什么是flask
5. 服务的构建-我的第一个flask
'''
我的第一个flask_service.py
'''
from flask_cors import cross_origin
from flask import Flask,request,redirect,url_for
import requests,json
#初始化一个flask
app = Flask(__name__)
@app.route('/test', methods=['GET', 'POST'])
@cross_origin()
def myfirst_service():
if request.method == "POST":
data = request.get_data().decode()
data = json.loads(data)
return json.dumps(data['question'],ensure_ascii=False)
if __name__ == "__main__":
app.run(host='0.0.0.0',port=8080,threaded=True)
'''
我的第一个send.py
'''
import requests
import json
url = 'http://127.0.0.1:8080/test'
question = '你好啊，我的第一个flask'
data = {
'question':question
}
print(requests.post(url,data=json.dumps(data)).json())
#service
from flask_cors import cross_origin
from flask import Flask,request,redirect,url_for
import requests,json
from mychatbot import template,CorpusSearch,GraphSearch,InterNet
#global
app = Flask(__name__)
#init the chatbot
template_model = template()
CorpusSearch_model = CorpusSearch()
GraphSearch_model = GraphSearch()
InterNet_model = InterNet()
@app.route('/test', methods=['GET', 'POST'])
@cross_origin()
def myfirst_service():
if request.method == "POST":
#sta_post = time.time()
data = request.get_data().decode()
data = json.loads(data)
return json.dumps('1',ensure_ascii=False)
@app.route('/template', methods=['GET', 'POST'])
@cross_origin()
def test_template():
if request.method == "POST":
#sta_post = time.time()
data = request.get_data().decode()
data = json.loads(data)
question = data['question']
answer = template_model.search_answer(question)
return json.dumps(answer,ensure_ascii=False)
@app.route('/CorpusSearch', methods=['GET', 'POST'])
@cross_origin()
def test_CorpusSearch():
if request.method == "POST":
#sta_post = time.time()
data = request.get_data().decode()
data = json.loads(data)
question = data['question']
answer = CorpusSearch_model.search_answer(question)
return json.dumps(answer,ensure_ascii=False)
@app.route('/GraphSearch', methods=['GET', 'POST'])
@cross_origin()
def test_GraphSearch():
if request.method == "POST":
#sta_post = time.time()
data = request.get_data().decode()
data = json.loads(data)
question = data['question']
answer = GraphSearch_model.search_answer(question)
return json.dumps(answer,ensure_ascii=False)
@app.route('/InterNet', methods=['GET', 'POST'])
@cross_origin()
def test_InterNet():
if request.method == "POST":
#sta_post = time.time()
data = request.get_data().decode()
data = json.loads(data)
question = data['question']
if '是谁' in question or '关系' in question:
return json.dumps(None,ensure_ascii=False)
try:
answer = InterNet_model.search_answer(question)
except:
answer = None
# except:
# answer = '对不起啊，小智无法解决这个问题'
return json.dumps(answer,ensure_ascii=False)
if __name__ == "__main__":
app.run(host='0.0.0.0',port=8080,threaded=True)
