快速入门 KBQA问答系统的实现
KBQA问答系统的实现
首先默认你是已经学会了如何构建知识图谱,并且学会用sparql语言查询里面的知识库里面的知识。如果不会,请看下面的链接
使用D2RQ把关系数据库的信息转化为rdf文件。
使用jena 构建知识数据库tdb,然后学会如何查询相关知识
源码在GitHub这里
目录结构很简单:
KBQA
/kbqa 里面有四个文件,分别是
word_tag.py ## 这个主要是用来分词的
question_temp.py ## 这个是问题模板
question2sparql.py ## 这个是用来把问题转化为sparql 语句
query_main.py ## 显然这个是最终的查询文件
另一个文件是:lsy.nt 这个是知识rdf的三元组,问答系统的知识来源
一、 把句子切分成一个个单词
这个也就是word_tag.py 的工作
这里主要是利用了斯坦福大学写好的 stanfordcorenlp
在你能使用这个包之前,首先要下载好一些东西
Stanford CoreNLP 官网下载
stanford-corenlp-full-2018-10-05
1、先下载红色按钮的,解压后,再把下面下载的jar包放进里面。
2、使用指令 安装 pip install stanfordcorenlp
需要介绍的是这里实现了一个类,word,表示一个词汇,有两个属性,token (记号) 和pos (part of speech)词性的意思
nlp.pos_tag(sentence) 会返回一个列表,元素为元组,即单词和它的词性组成。
这样我们就可以把句子给切分好了。
二、问题模板的构建
首先基于知识库,我们可以有下面的问题
sentence_list = [
"What is the name of littlejun ?",
"What is the age of chacha ?" ,
"What is the username of scc ?" ,
"Whose age is larger than 18 ? " ,
"What is the phone number of chacha ? " ,
"What is the password of littlejun ? "
]
sparql 的模板,最后都是由下面的模式组成
# 问题模板
prefix_temp ="""
PREFIX ps:<http://solicucu/person/#>
PREFIX us:<http://solicucu/user/#>
PREFIX vocab: <http://solicucu/vocab/>
"""
sparql_select_temp = u"""
{prefix}
select distinct {select} where {{
{expression}
}}
"""
那注意到我们只需要完成select 的填充和 expression 的填充就好了。
这个跟确却的问题有关,所以,我们要做的事情就是,如何知道,传入来的问题,是属于哪个句子呢?
REFO (Regular Expressions for Objects)
基于对象级别的正则匹配,这个跟python的很像正则表达式很像
"ab" is Literal("a") + Literal("b")
"a*" is Star(Literal("a"))
"(ab)+|(bb)*?" is:
a = Literal("a")
b = Literal("b")
regex = Plus(a + b) | Star(b + b, greedy=False)
如上面所看见,Literal 是一个文字类,那些类支持一些基本符号:+ 连接的意思
| 为或的意思,
python 正则表达式的 + 号用Plus() 代替,表示1到多个的意思。
python 正则表达式的 * 号用Star() 代替,表达0 到多个的意思,可以选择是否采取贪婪模式。
所以,词汇的定义很重要:
# 定义一个词汇类,继承Predicate
class W(Predicate):
#token 词汇的字面符号 pos 词汇的属性
def __init__(self,token=".*",pos=".*"):
self.token = re.compile(token + "$")
self.pos = re.compile(pos+"$")
super(W, self).__init__(self.match) # 不可缺少
def match(self,word):
m1 = self.token.match(word.token)
m2 = self.pos.match(word.pos)
return m1 and m2
谓词的定义,这是一个继承了Predicate(来自refo 的类,定义另个正则匹配对象属性
match 函数,表面,对于传进来的word 要同时满足 记号和词性都符合才行。
# 定义一些规则,相当与正则表达式的某个模式
class Rule(object):
#匹配的条件数 和条件,以及action 回调函数
def __init__(self,condition_num,condition=None,action=None):
assert condition and action
self.condition = condition
self.action = action
self.condition_num = condition_num
def apply(self,word_list):
#因为可能满足条件的有多处,所以用matches列表存储
matches = []
# 用条件去找匹配的词汇,finditer 里面用到了yeild,就是每次找到一个结果返回一次,继续找
# 可以理解为finditer 返回的值可以迭代
for m in finditer(self.condition,word_list):
i,j = m.span()
matches.extend(word_list[i:j]) # 提取出被匹配的句子区间划出,其中可能有其他杂词汇
return self.action(matches),self.condition_num
规则class的形式,首先必须有两个参数condition 和 action 表示这个规则适用的条件(对象正则表达式) 和采取的行动(某个回调函数)
注意里面的finditer 返回的对象是一个Match 的对象,通过span() 获取匹配的范围
当谓词定义好了,规则定义好了,就可以写匹配规则了。
# 疑问代词关键字 who what
what = (W("what")|W("What"))
whose = (W("whose")|W("Whose"))
of = W("of")
number_entity = W(pos="CD")
# 属性关键字
username = W("username")
name = W("name")
phone = W("phone")
age = W("age")
password = W("password")
attr_noun = (username | name | phone | phone | age | password)
#普通名词
common_noun = W(pos = pos_common_noun)
看上面,定义的词汇,what 可以匹配大写或者小写 ,因为what = (W(“what”)|W(“What”))
所以一个谓词,可以是多个谓词的或,或者只有一个,就是匹配固定的单词
number_entity = W(pos=“CD”) 这里定义了 一个数字谓词,因为我们不关注它的值,所以指定属性为”CD“就好,至于为什么是”CD“
attr_noun :表示属性名词,在本次知识数据库中,主要涉及的属性名词如上。
规则定义
rules = [
# What is the name of sb-uname?
# What is the age of sb-uname ?
# What is the username of sb-uname?
# What is the phone number of sb-uname?
# What is the password of sb-name?
Rule(condition_num = 4 ,condition = what + Star(Any(),greedy = False) + attr_noun + Star(Any(),greedy = False) + of + common_noun + Star(Any(),greedy = False),action = QuestionSet.proccess_attr_noun),
# Whose age is larger than 18 ?
Rule(condition_num = 4,condition = whose + attr_noun + Star(Any(),greedy = False) + compare + Star(Any(),greedy = False) + number_entity + Star(Any(),greedy = False),action = QuestionSet.who_age_compare )
]
这里规则,只有两条:第一条匹配了上面5个问题,第二条匹配了它上面的问题
第一个参数 condition_num :这个是我们condition中关注的词汇数。比如第一个我们希望匹配到有what attr_noun(即name,age 等5个中的一个),common_noun(普通名词) 所以为3
第二个参数 condition:
这里可以看到 是几个谓词相加,即如前面所述,是连接的意思,所以该条件表达的意思是
任意的一个句子 模式为:what/What … name/age/username/password/phone … of comomn_noun …
也就是为什么会匹配上面的句子的原因
第三个参数 action:
回调函数,在rule 里面有这么一个属性,他是在调用了apply 后,如果匹配了,就会调用的一个函数
在这里,指向了一个QuestionSet.proccess_attr_noun 这个函数,可以继续完成确定是哪个规则。
提取词汇,填充模板
# 1 what is the name of sb-uname ?
def what_name(word_list):
# if(len(word_list)):
# print("成功匹配问题")
# for w in word_list:
# print(w.token,end = " ")
sparql = None
select = "?name"
for w in reversed(word_list):
# 找到第一个普通名词
if(w.pos == pos_common_noun):
e = " ps:{person} vocab:person_name ?name .".format(person = w.token)
sparql = sparql_select_temp.format(prefix = prefix_temp,
select = select,
expression = e)
break
return sparql
首先是对select 赋值,确定要查询的变量 ,比如这里就是?name 就是要查询的变量
最终要的是确定表达式的变量,这里是:
ps:{person} vocab:person_name ?name
表示某人的名字是什么,这个某人 就会是word_list 从后往前的一个名词,所以,找到之后,就可以break退出了。
然后返回,对应的sparql语句
三、把句子转化为sparql 语句
def get_sparql(sentence):
word_list = word_tag.get_word_list(sentence)
query = None
queries_dict = dict()
for rule in question_temp.rules:
query,num = rule.apply(word_list)
if(query is not None):
queries_dict[num] = query
if len(queries_dict) == 0 :
return None
elif len(queries_dict) == 1:
# 要转化为list才可以用索引访问
return list(queries_dict.values())[0]
else: # key 就是对多元组的排序指定列 item的名字随便,表示列表的元素,item[0] 表示那个值
sorted_dict = sorted(queries_dict.iteritems(),key=lambda item:item[0],reversed = True)
return sorted_dict[0][1]
上面,就是对句子去匹配每一个规则,放到一个字典里面。
如果长度为0,那么就是没有匹配,如果长度为1 那么就是只匹配到一个句子,直接取第0个值,但是注意要转化为列表的形式才可以用索引访问。
如果长度大于1的时候,就对字典排序,按照key的大小排序,从大到小,也就是取匹配到关键字最多的一个。
四、访问endpoint,显示查询结果
sparql = SPARQLWrapper("http://localhost:3030/db/query")
sparql.setReturnFormat(JSON)
if __name__ == "__main__":
# sentence = "What is the username of scc ?"
while True:
sentence = input("please input the question ? input quit to leave\n")
# print("question:",sentence)
if(sentence == "quit"):
break
str_sparql = q2s.get_sparql(sentence)
if(str_sparql is not None):
sparql.setQuery(str_sparql)
results = sparql.query().convert()
head = results["head"]["vars"]
values = results["results"]["bindings"] # 存储的结果
if(len(values)==0):
print("no relevant answer")
else:
print("the answer is :",end = " ")
for v in values: # 对于所有value ,通过varname 获取其值
for varname in head:
print(v[varname]["value"])
else:
print("sorry,I can't understand your means")
这里要注意的就是,我们查询的变量名存在results[“head”][“vars”]
对应的值存在 results[“results”][“bindings”] 结果是一个个字典