本次我們基於浙江大學在openKG上提供的 基於REfO的KBQA實現及示例，在自己的知識圖譜上實現簡單的知識問答系統。

簡介

基於浙江大學在openKG上提供的 基於REfO的KBQA實現及示例。代碼部分浙大方面已經完成絕大部分，這裡主要將其應用到自己的知識圖譜上。在運行KBQA代碼前，應按照前面的教程將電影類知識圖譜導入到Jena的TDB資料庫中，並運行fuseki伺服器，這樣我們才能進行訪問查詢。

從零開始構建知識圖譜?

github.com

代碼結構

代碼結構為

.:

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}.

?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
")

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"]

推薦閱讀：

相关文章