本文目录导读：
我来详细介绍一下如何使用Python构建数据知识图谱。
知识图谱核心概念
知识图谱由实体（节点）和关系（边）组成，使用三元组表示：(实体1, 关系, 实体2)
Python构建知识图谱方案
使用NetworkX（轻量级）
import networkx as nx
import matplotlib.pyplot as plt
from typing import List, Tuple, Dict
class SimpleKnowledgeGraph:
def __init__(self):
self.graph = nx.MultiDiGraph() # 有向多重图
def add_entity(self, entity_id: str, properties: dict = None):
"""添加实体节点"""
self.graph.add_node(entity_id, **properties if properties else {})
def add_relationship(self, entity1: str, entity2: str,
relation: str, properties: dict = None):
"""添加关系边"""
self.graph.add_edge(entity1, entity2,
relation=relation,
**properties if properties else {})
def query_entity(self, entity_id: str):
"""查询实体信息"""
if entity_id in self.graph:
return dict(self.graph.nodes[entity_id])
return None
def query_relationships(self, entity_id: str):
"""查询实体的所有关系"""
relationships = []
for _, neighbor, data in self.graph.edges(entity_id, data=True):
relationships.append({
'entity1': entity_id,
'entity2': neighbor,
'relation': data['relation']
})
return relationships
def visualize(self, figsize=(12, 8)):
"""可视化知识图谱"""
pos = nx.spring_layout(self.graph)
plt.figure(figsize=figsize)
# 绘制节点
nx.draw_networkx_nodes(self.graph, pos, node_color='lightblue',
node_size=2000)
# 绘制边
nx.draw_networkx_edges(self.graph, pos, arrows=True)
# 添加标签
nx.draw_networkx_labels(self.graph, pos)
# 添加关系标签
edge_labels = {(u, v): d['relation']
for u, v, d in self.graph.edges(data=True)}
nx.draw_networkx_edge_labels(self.graph, pos, edge_labels=edge_labels)
plt.title("知识图谱")
plt.axis('off')
plt.show()
# 使用示例
kg = SimpleKnowledgeGraph()
# 添加实体
kg.add_entity("Python", {"type": "编程语言", "creator": "Guido van Rossum"})
kg.add_entity("Django", {"type": "Web框架", "language": "Python"})
kg.add_entity("Guido", {"type": "程序员", "nationality": "荷兰"})
# 添加关系
kg.add_relationship("Guido", "Python", "创造了")
kg.add_relationship("Python", "Django", "用于开发")
kg.add_relationship("Guido", "Python", "开发了")
# 查询
print(kg.query_entity("Python"))
print(kg.query_relationships("Python"))
# 可视化
kg.visualize()
使用Neo4j（企业级）
from neo4j import GraphDatabase
import pandas as pd
class Neo4jKnowledgeGraph:
def __init__(self, uri, user, password):
self.driver = GraphDatabase.driver(uri, auth=(user, password))
def close(self):
self.driver.close()
def create_entity(self, tx, entity_id, properties):
"""创建实体节点"""
query = """
CREATE (e:Entity {id: $entity_id})
SET e += $properties
RETURN e
"""
result = tx.run(query, entity_id=entity_id, properties=properties)
return result.single()
def create_relationship(self, tx, entity1_id, entity2_id, relation, properties):
"""创建关系"""
query = """
MATCH (e1:Entity {id: $entity1_id})
MATCH (e2:Entity {id: $entity2_id})
CREATE (e1)-[r:RELATION {type: $relation}]->(e2)
SET r += $properties
RETURN r
"""
result = tx.run(query, entity1_id=entity1_id, entity2_id=entity2_id,
relation=relation, properties=properties)
return result.single()
def load_from_dataframe(self, df: pd.DataFrame,
entity_col: str = 'entity',
relation_col: str = 'relation',
target_col: str = 'target'):
"""从DataFrame加载知识图谱"""
with self.driver.session() as session:
for _, row in df.iterrows():
session.execute_write(self.create_entity,
row[entity_col],
{'source': row[entity_col]})
session.execute_write(self.create_entity,
row[target_col],
{'target': row[target_col]})
session.execute_write(self.create_relationship,
row[entity_col],
row[target_col],
row[relation_col],
{})
def query_entity_relations(self, entity_id):
"""查询实体关系"""
with self.driver.session() as session:
result = session.run("""
MATCH (e:Entity {id: $entity_id})-[r]-(connected)
RETURN e.id as entity, type(r) as relation, connected.id as connected_entity
""", entity_id=entity_id)
return pd.DataFrame([r for r in result])
# 使用示例
# kg = Neo4jKnowledgeGraph("bolt://localhost:7687", "neo4j", "password")
# 准备数据
# data = {
# 'entity': ['Python', 'Python', 'Guido'],
# 'relation': ['用于开发', '有包管理器', '创造了'],
# 'target': ['Django', 'pip', 'Python']
# }
# df = pd.DataFrame(data)
# kg.load_from_dataframe(df)
从结构化数据构建
import json
import numpy as np
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
class DataDrivenKnowledgeGraph:
def __init__(self):
self.entities = {}
self.relationships = []
def extract_entities_from_data(self, data: list, key_fields: list):
"""从数据中提取实体"""
for item in data:
entity_id = str(item.get(key_fields[0], ''))
properties = {k: v for k, v in item.items() if k in key_fields}
self.entities[entity_id] = properties
def infer_relationships(self, threshold=0.7):
"""基于相似度推断实体关系"""
# 将实体属性转换为向量
texts = []
entity_ids = []
for entity_id, properties in self.entities.items():
text = ' '.join([str(v) for v in properties.values()])
texts.append(text)
entity_ids.append(entity_id)
# 计算相似度
vectorizer = TfidfVectorizer()
tfidf_matrix = vectorizer.fit_transform(texts)
similarity_matrix = cosine_similarity(tfidf_matrix)
# 建立关系
for i in range(len(entity_ids)):
for j in range(i+1, len(entity_ids)):
if similarity_matrix[i][j] > threshold:
self.relationships.append({
'source': entity_ids[i],
'target': entity_ids[j],
'relation': 'similar',
'weight': similarity_matrix[i][j]
})
def export_to_json(self, filepath):
"""导出为JSON格式"""
output = {
'entities': self.entities,
'relationships': self.relationships
}
with open(filepath, 'w', encoding='utf-8') as f:
json.dump(output, f, ensure_ascii=False, indent=2)
# 使用示例
data = [
{'id': 'Python', 'type': 'language', 'year': 1991, 'creator': 'Guido'},
{'id': 'JavaScript', 'type': 'language', 'year': 1995, 'creator': 'Brendan'},
{'id': 'Django', 'type': 'framework', 'year': 2005, 'language': 'Python'},
{'id': 'React', 'type': 'framework', 'year': 2013, 'language': 'JavaScript'}
]
kg = DataDrivenKnowledgeGraph()
kg.extract_entities_from_data(data, ['id', 'type', 'year'])
kg.infer_relationships(threshold=0.5)
print("Relationships found:", kg.relationships)
kg.export_to_json('knowledge_graph.json')
从文本中抽取（NLP方法）
import spacy
from collections import defaultdict
class TextToKnowledgeGraph:
def __init__(self):
self.nlp = spacy.load("zh_core_web_sm")
self.entities = set()
self.relationships = []
def extract_triples(self, text: str):
"""从文本中提取三元组"""
doc = self.nlp(text)
triples = []
# 提取命名实体
entities = [(ent.text, ent.label_) for ent in doc.ents]
# 简单的三元组提取规则
for token in doc:
if token.dep_ == "nsubj" and token.head.pos_ == "VERB":
subject = token.text
predicate = token.head.text
objects = [child.text for child in token.head.children
if child.dep_ == "dobj"]
for obj in objects:
triples.append((subject, predicate, obj))
return triples
def build_from_texts(self, texts: list):
"""从文本列表构建知识图谱"""
for text in texts:
triples = self.extract_triples(text)
for subj, pred, obj in triples:
self.entities.add(subj)
self.entities.add(obj)
self.relationships.append({
'subject': subj,
'predicate': pred,
'object': obj
})
def display_statistics(self):
"""显示统计信息"""
print(f"实体数量: {len(self.entities)}")
print(f"关系数量: {len(self.relationships)}")
# 统计关系类型
relation_types = defaultdict(int)
for rel in self.relationships:
relation_types[rel['predicate']] += 1
print("\n关系类型分布:")
for rel_type, count in sorted(relation_types.items(), key=lambda x: -x[1]):
print(f" {rel_type}: {count}")
# 使用示例
# 需要先下载中文模型: python -m spacy download zh_core_web_sm
# texts = [
# "Python是由Guido van Rossum创建的编程语言",
# "Django是用Python开发的Web框架",
# "TensorFlow是Google开发的机器学习框架"
# ]
# kg = TextToKnowledgeGraph()
# kg.build_from_texts(texts)
# kg.display_statistics()
数据存储方案
RDF格式存储
from rdflib import Graph, Literal, RDF, URIRef, Namespace
from rdflib.namespace import FOAF, XSD
# 创建RDF图
g = Graph()
ex = Namespace("http://example.org/")
# 添加三元组
python = URIRef(ex + "Python")
guido = URIRef(ex + "Guido")
g.add((guido, RDF.type, FOAF.Person))
g.add((guido, FOAF.name, Literal("Guido van Rossum", lang="en")))
g.add((python, ex.createdBy, guido))
# 序列化
print(g.serialize(format="turtle"))
g.serialize("knowledge_graph.rdf", format="xml")
实用建议
- 数据质量：确保实体ID唯一，关系明确
- 存储选择：
- 小规模（<10万节点）：NetworkX
- 中规模：Neo4j Community
- 大规模：Neo4j Enterprise / JanusGraph
- 查询优化：使用索引，设计合适的图模式
- 可视化：D3.js, Cytoscape.js, vis.js
选择哪种方案取决于你的具体需求：数据规模、查询复杂度、实时性要求等。
