从零开始构建知识图谱(三)

基于REfO的简单知识问答

Posted by Pelhans on September 3, 2018

本次我们基于浙江大学在openKG上提供的 基于REfO的KBQA实现及示例,在自己的知识图谱上实现简单的知识问答系统。

简介

基于浙江大学在openKG上提供的 基于REfO的KBQA实现及示例。代码部分浙大方面已经完成绝大部分,这里主要将其应用到自己的知识图谱上。在运行KBQA代码前,应按照前面的教程将电影类知识图谱导入到Jena的TDB数据库中,并运行fuseki服务器,这样我们才能进行访问查询。

代码结构

代码结构为
.:
data/ get_dict.sh query.py utils/
./data:
actorName.txt get_dict.txt movieName.txt
./utils:
init.py init.pyc rules.py rules.pyc word_tagging.py word_tagging.pyc

  • 其中data 目录存放由数据库倒出生成的字典文件,用于扩展jieba分词,由 get_dict.sh 生成。
  • utils/ 内存放查询预处理的模块。word_tagging.py 用于将词的文本和词性打包,视为词对象,对应:class:Word(token, pos)。rules.py 内定义各种规则并将自然语言转换为SPARQL查询语言,最终以JSON返回结果。
  • query.py 为程序入口,运行它来进行简单的KBQA。

示例


基于REfO的KBQA

具体实现

基于REfO的简单知识问答的原理很简单,就是通过REfo提供的匹配能力,在输入的自然语言问题中进行匹配查找。如果找到我们预先设定的词或词性组合,那么就认为该问题与这个词或词性组合匹配。而一个词或词性的组合又对应着一个SPARQL查询模板,这样我们就借助REfO完成了自然语言到查询模板的转换。得到查询模板后,我们就利用Jena fuseki 服务器提供的端口进行查询得到返回的结果。

模块一 word_tagging部分

该部分利用jieba分词对中文句子进行分词和词性标注。将词的文本和词性进行打包,视为词对象,对应 :class:Word(token, pos)。

class Word(object):    
    def __init__(self, token, pos):
        self.token = token
        self.pos = pos 
                       
                       
class Tagger:          
    def __init__(self, dict_paths):
        # TODO 加载外部词典
        for p in dict_paths:
            jieba.load_userdict(p)
                       
    def get_word_objects(self, sentence):
        """            
        Get :class:WOrd(token, pos)
        """            
        return [Word(word.encode('utf-8'), tag) for word, tag in pseg.cut(sentence)]

模块二 rules 部分

该部分为程序核心,负责将自然语言转换为SPARQL模板。

下面为rules的程序入口,customize_rules 函数:

def customize_rules():
    # some rules for matching
    # TODO: customize your own rules here
    person = (W(pos="nr") | W(pos="x") | W(pos="nrt"))
    movie = (W(pos="nz"))
    place = (W("出生地") | W("出生"))
    intro = (W("简介") | W(pos="介绍"))
                                
    rules = [      
                                
        Rule(condition=W(pos="r") + W("是") + person | \ 
                       person + W("是") + W(pos="r"),
             action=who_is_question),
                            
        Rule(condition=person + Star(Any(), greedy=False) + place + Star(Any(), greedy=False),
             action=where_is_from_question),
                            
        Rule(condition=movie + Star(Any(), greedy=False) + intro + Star(Any(), greedy=False) ,
             action=movie_intro_question)
                            
    ]         
    return rules

该函数中我们设置了一些简单的匹配规则,例如我们设置 ‘'’movie = (W(pos=”nz”))’’‘,即movie 的词性应该是nz。其中的W()是我们在继承REfO的Predicate方法的基础上扩展更新了match方法。您可以简单的把它理解为re中compile后的match,只不过多个W()间出现的顺序可以变化。这样通过多个定制的W()和Star(Any(), greedy=False)(相当于.*?)这种通配符的组合,我们就定义了一组匹配规则,当遇到符合该规则的句子时,就选取该规则后action对应的查询模板。

例如当输入为“周星驰是谁”这样的问题时,会匹配到rules 中的 第一条规则。而后执行该规则后对应的action, who_is_question。而who_is_question对应的查询模板为:

def who_is_question(x):
    select = u"?x0"               
         
    sparql = None
    for w in x:
        if w.pos == "nr" or w.pos == "x":
            e = u" ?a :actor_chName '{person}'. \n \
            ?a :actor_bio ?x0".format(person=w.token.decode("utf-8"))
         
            sparql = SPARQL_TEM.format(preamble=SPARQL_PREAMBLE,
                                       select=select,
                                       expression=INDENT + e)
            break
    return sparql

有了查询模板后,我们通过SPARQLWrapper 模块的SPARQLWrapper 执行该查询,并对返回的结果进行转换得到回答。对应的代码如下:

from SPARQLWrapper import SPARQLWrapper, JSON    
from utils.word_tagging import Tagger
from utils.rules import customize_rules                   
                 
if __name__ == "__main__":    
    print("init...........")    
    sparql_base = SPARQLWrapper("http://localhost:3030/kg_demo_movie/query")
    #加载外部词典,提升分词准确性和词性标注准确性
    tagger = Tagger(['data/actorName.txt', 'data/movieName.txt'])
    #初始化并获取规则列表
    rules = customize_rules()   
    print("done \n")    
                 
    while True:    
        print("Please input your question: ")    
        default_question = raw_input()    
        # 获取wordclass
        seg_list = tagger.get_word_objects(default_question)
                 
        for rule in rules:    
            # 将规则列表应用到问题上得到查询模板
            query = rule.apply(seg_list)    
            if query:    
                # 设置查询相关
                sparql_base.setQuery(query)    
                sparql_base.setReturnFormat(JSON)         
                # 得到返回结果并做转换
                results = sparql_base.query().convert()   
                 
                if not results["results"]["bindings"]:    
                    print("No answer found :(")    
                    continue    
                for result in results["results"]["bindings"]:
                    print "Result: ", result["x0"]["value"]