Python脚本数据知识图谱如何构建

wen 实用脚本 2026-07-18 177

本文目录导读：

Python脚本数据知识图谱如何构建

知识图谱核心概念
Python构建知识图谱方案
数据存储方案
实用建议

我来详细介绍一下如何使用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

选择哪种方案取决于你的具体需求：数据规模、查询复杂度、实时性要求等。

本文地址： https://www.bmjup.com/post/24500.html