初识知识图谱

第一天学习打卡

知识图谱基本介绍

前身:语义网络-->对文本进行语义的分析

知识图谱是一种揭示实体之间关系的语义网络,可以对现实世界的事物及其相互关系进行形式化地描述.

现在的知识图谱被用来泛指各种大规模的知识库

知识图谱的构建

通用表示方式:三元组

两种基本形态:

实体1—关系—实体2

实体—属性—属性值

采用哪种形态取决于图谱的使用方式和想要完成的任务 构建图谱时一定要想清楚图谱建成后用来做什么,正确确定采用属性还是实体。

知识图谱构建关键技术

• 知识抽取 非结构化-->结构化
• 知识融合 清洗提升数据质量
• 知识推理 挖掘扩充或补全数据，通过算法挖掘出一些图中没有的关系等
• 知识表示 为应用做准备（向量化处理(h,r,t)）

知识抽取：目的：抽取三元组

1.实体抽取 命名实体识别

• 基于规则和词典的方法
• 基于机器学习的模型预测方法(序列标注问题BIO)

2.关系抽取

• 限定领域关系抽取

从预设好的数个关系中判断，给定的实体对是否满足某一个关系：F(实体1，实体2，文本)-->关系（有监督分类任务）

两种主要训练方式：pipeline分类 or 联合训练

• 开放领域的关系抽取

3.属性抽取

Neo4j python接口：

pip install py2neo

数据库基本语言

# 创建节点
CREATE (Zhh:Person{name:'Zhh String',born:2002})

# 创建关系
CREATE (Zhh)-[:ACTED_IN{roles:['Student']}]->(UESTC)

# 查找节点
MATCH (smy{name:'smy Love'}) RETURN smy
MATCH (people:Person) RETURN people.name LIMIT 10
MATCH (nineties:Movies) WHERE nineties.released >= 1990 AND nineties.released < 2000 RETURN nineties.title

import re
import json
from py2neo import Graph
from collections import defaultdict
#读取三元组

#连接图数据库
graph = Graph("http://localhost:7474",auth=("username","pwd"))

attribute_data = defaultdict(dict)
relation_data = defaultdict(dict)
label_data = {}

#有的实体后面会有括号，里面的内容可以作为标签
#提取到标签后，将括号删除
def get_label(x,label_data):
    if re.search("(.+)",x):
        label_string = re.search("(.+)",x).group()
        for label in ["歌曲","专辑","电影","电视剧"]:
            if label in label_string:
                x = re.sub("(.+)","",x)
                label_data[x] = label
            else:
                x = re.sub("(.+)","",x)
    return x

#读取三元组文件
#实体-关系-实体
with open("file.txt",encoding = "utf-8") as f:
    for line in f:
        head,relation,tail = line.strip().split("\t")
        head = get_label(head,label_data)
        relation_data[head][relation] = tail
with open("file.txt",encoding="utf-8") as f:
    for line in f:
        entity,attribute,value = line.strip().split("\t")
        entity = get_label(entity,label_data)
        attribute_data[entity][attribute] = value
  
 #构建cyper语句
 cypher = ""
 in_graph_entity = set()
 for i,entity in enumerate(attribute_date):
     #为所有实体增加一个名字属性
     attribute_date[entity]["NAME"] = entity
     #将一个实体的所有属性拼接成一个类似字典的表达式
     text = "{"
     for attribute,value in attribute_data[entity].items():
         text += "%s:\'%s\',"%(attribute,value)
     text = text[:-1] + "}"
     if entity in label_data:
         label = label_data[entity]
         #带标签的实体构建语句
         cypher += "CREATE (%s:%s %s)" % (entity,label,text) + "\n"
     else:
         #不带标签的实体构建语句
         cypher += "CREATE (%s %s)" % (entity,text) + "\n"
     in_graph_entity.add(entity)
     
 #构建关系语句
 for i,head in enumerate(relation_data):
     if head not in in_graph_entity:
         cypher += "CREATE (%s {NAME:'%s'})" % (head,head) + "\n"
         in_graph_entity.add(head)
     for relation,tail in relation_data[head].items():
         if tail not in in_graph_entity:
             cypher += "CREATE (%s {NAME:'%s'})" % (tail,tail) + "\n"
             in_graph_entity.add(tail)
         cypher += "CREATE (%s)-[:%s]->(%s)" % (head,relation,tail) + "\n"
 
 graph.run(cyper)
 
 #将所有实体存储在一个json文件中去
 data = defaultdict(set)
 for head in relation_data:
     data["entitys"].add(head)
     for relation,tail in relation_data[head].items():
         data["relations"].add(relation)
         data["entitys"].add(tail)
 
 for entity,label in label_data.items():
     data["entitys"].add(entity)
     data["labels"].add(label)
     
 for entity in attribute_data:
     for attr,value in attribute_data[enti].items():
         data["entitys"].add(entity)
         data["attributes"].add(attr)
         
 data = dict((x,list(y)) for x,y in data.items())
 
 with open("schema.json","w",encoding="utf-8") as f:
     f.write(json,dumps(data,ensure_ascii=False,indent=2))

将文本转化成查询语句

NL2SQL

基于模板+文本匹配

在一个知识库内,有很多Q-SQL对,对于用户提出一个真实问题后,与Q-SQL库进行分文匹配[技能点:文本匹配算法],计算两个文本间的相似度,选择最接近的问题,将对应的答案返回。 1.问题会有很多，怎样让匹配次数减少？-->使用模板：关键词->{Q1,Q2,Qn}

问题模板	cypher	answer	check
%ENT%的年龄是多少	Match (n)<-[:年龄]-(m{AGE:"%ENT%"}) return n.AGE	n.AGE	{"%ENT%":1}
谁创作的%ENT%	Match (n)<-[:创作]-(m{NAME:"%ENT%"}) return n.NAME	%ENT%的创作者是n.NAME	{"%ENT%":1}
%ENT%的%ATT%是什么	Match (n) where n.NAME="%ENT%" return n.%ATT%	n.%ATT%	{"%ENT%":1,"%ATT%":1}
%ENT0%和%ENT1%的关系是什么	Match (n{NAME="%ENT0%"})-[REL]->(m{NAME="%ENT1%"}) RETURN REL	REL	{"%ENT%":2}

先判断实体属量(槽位)的个数是否匹配(check列)

import re
import json
import pandas
import itertools
from pt2neo import Graph
from collection import defaultdict

class GraphQA:
    def __init__(self):
        self.graph = Graph("http://localhost:7474",auth=("neo4j","demo"))
        #加载知识图谱中所有实体、关系、属性的json文件,便于寻找槽位
        schema_path = "schema.json"
        #类似上表的模板excel文件
        templet_path = "templet.xlsx"
        self.load(schema_path,templet_path)
        print("知识图谱问答系统加载完毕!\n")
        
    def load(self,schema_path,templet_path):
        self.load_schema(schema_path)
        self.load_templet(templet_path)
        return
   #获取问题中的实体：1.基于词表的方式 2.NER模型3.正则表达式
   def get_mention+entitys(self,sentence):
       return re.findall("|".join(self.entity_set),sentence)
   
   #获取问题中谈到的关系，也可以使用各种文本分类模型
   def get_mention_relations(self,sentence):
       return re.fiindall("|".join(self.relation_set),sentence)
       
   # 获取问题中谈到的属性
   def get_mention_attributes(self,sentence):
       return re.findall("|".join(self.attrubute_set),sentence)
       
   #获取问题中谈到的标签
   def get_mention_labels(self,sentence):
       return re.findall("|".join(self.label_set),sentence)
       
   #提取问题中需要的信息
   def parse_sentence(self,sentence):
       entitys = self.get_mention_entitys(sentence)
       relations = self.get_mention_relations(sentence)
       labels = self.get_mention_labels(sentence)
       attributes = self.get_mention_sttributes(sentence)
       return {"%ENT%":entitys,
               "%REL%":relations,
               "%LAB%":labels,
               "%ATT%":attributes}
               
   #将提取到的值分配到键上
   def decode_value_combiantion(self,value_combination,cypher_check):
       res = {}
       for index,(key,required_count) in enumerate(cypher_check.items()):
           if required_count == 1:
               res[key] = value_combination[index][0]
           else:
               for i in range(required_count):
                   key_num = key[:-1] + str(i) + "%"
                   res[key_num] = value_combination[index][i]
       return res
   
   #对于找到了超过模板中需求的实体属量的情况，需要进行排列组合
   def get_combination(self,cypher_check,info):
       slot_values = []
       for key,required_count in cypher_check.items():
           slot_values.append(itertools.combinations(info[key],required_count))
       value_combinations = itertools.product(*slot_values)
       combinations = []
       for value_combination in value_combinations:
          combinations.append(self.decode_value_combination(value_combination,cypher_check))
       return combinations
   
   #将带有模板的token替换成真实值
   def expand_templet(self,templet,cypher,cypher_check,info,answer):
       combinations = self.get_conbination(cypher_check,info)
       templet_cpyher_pair = []
       for combination in combinations:
           replaced_templet = self.replace_token_in_string(templet,combination)
           replaced_cypher = self.replace_token_in_string(cypher,combination)
           replaced_answer = self.replace_token_in_string(answer,combination)
           templet_cypher_pair.append([replace_templet,reolaced_cypher,replaces_answer])
       return templet_cypher_pair
           
   #计算分值
   zl
   #解析得分
        
   #对外提供问答接口
   #参数：用户输入的问题：sentence
   def query(self,sentence):
       #扣取槽位
       info = self.parse_sentence(sentence)
       #匹配，得分数
       templet_cypher_score = self.cypher_match(sentence,info)
       for templet,cypher,score,answer in templet_cypher_score:
           graph_search_result = self.graph.run(cypher).data()
           if graph_search_result:
               break
       answer = self.parse_result(graph_search_result,answer,info)
       return answer