1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
| import networkx as nx
import matplotlib.pyplot as plt
from typing import Dict, List, Tuple, Any
import json
import numpy as np
from dataclasses import dataclass
@dataclass
class Entity:
"""实体类"""
id: str
name: str
type: str
properties: Dict[str, Any]
def __hash__(self):
return hash(self.id)
@dataclass
class Relation:
"""关系类"""
id: str
name: str
source: str
target: str
properties: Dict[str, Any]
class KnowledgeGraph:
"""知识图谱类"""
def __init__(self):
self.entities: Dict[str, Entity] = {}
self.relations: Dict[str, Relation] = {}
self.graph = nx.MultiDiGraph()
def add_entity(self, entity: Entity):
"""添加实体"""
self.entities[entity.id] = entity
self.graph.add_node(entity.id, **entity.properties, name=entity.name, type=entity.type)
def add_relation(self, relation: Relation):
"""添加关系"""
self.relations[relation.id] = relation
self.graph.add_edge(
relation.source,
relation.target,
key=relation.id,
name=relation.name,
**relation.properties
)
def get_neighbors(self, entity_id: str, relation_type: str = None) -> List[str]:
"""获取邻居实体"""
neighbors = []
for neighbor in self.graph.neighbors(entity_id):
if relation_type is None:
neighbors.append(neighbor)
else:
edges = self.graph.get_edge_data(entity_id, neighbor)
for edge_data in edges.values():
if edge_data.get('name') == relation_type:
neighbors.append(neighbor)
break
return neighbors
def find_path(self, start_entity: str, end_entity: str, max_length: int = 3) -> List[List[str]]:
"""查找实体间的路径"""
try:
paths = list(nx.all_simple_paths(
self.graph, start_entity, end_entity, cutoff=max_length
))
return paths
except nx.NetworkXNoPath:
return []
def get_entity_info(self, entity_id: str) -> Dict[str, Any]:
"""获取实体详细信息"""
if entity_id not in self.entities:
return None
entity = self.entities[entity_id]
node_data = self.graph.nodes[entity_id]
outgoing_relations = []
incoming_relations = []
for neighbor in self.graph.neighbors(entity_id):
edges = self.graph.get_edge_data(entity_id, neighbor)
for edge_key, edge_data in edges.items():
outgoing_relations.append({
'relation': edge_data.get('name'),
'target': neighbor,
'target_name': self.entities.get(neighbor, {}).name if neighbor in self.entities else neighbor
})
for predecessor in self.graph.predecessors(entity_id):
edges = self.graph.get_edge_data(predecessor, entity_id)
for edge_key, edge_data in edges.items():
incoming_relations.append({
'relation': edge_data.get('name'),
'source': predecessor,
'source_name': self.entities.get(predecessor, {}).name if predecessor in self.entities else predecessor
})
return {
'entity': entity,
'properties': node_data,
'outgoing_relations': outgoing_relations,
'incoming_relations': incoming_relations
}
def visualize_subgraph(self, center_entity: str, depth: int = 2):
"""可视化子图"""
subgraph_nodes = set([center_entity])
current_level = set([center_entity])
for _ in range(depth):
next_level = set()
for node in current_level:
neighbors = list(self.graph.neighbors(node)) + list(self.graph.predecessors(node))
next_level.update(neighbors)
subgraph_nodes.update(next_level)
current_level = next_level
subgraph = self.graph.subgraph(subgraph_nodes)
plt.figure(figsize=(12, 8))
pos = nx.spring_layout(subgraph, k=2, iterations=50)
node_colors = ['red' if node == center_entity else 'lightblue' for node in subgraph.nodes()]
nx.draw_networkx_nodes(subgraph, pos, node_color=node_colors, node_size=1000, alpha=0.7)
nx.draw_networkx_edges(subgraph, pos, alpha=0.5, arrows=True, arrowsize=20)
labels = {node: self.entities[node].name if node in self.entities else node
for node in subgraph.nodes()}
nx.draw_networkx_labels(subgraph, pos, labels, font_size=8)
edge_labels = {}
for u, v, data in subgraph.edges(data=True):
edge_labels[(u, v)] = data.get('name', '')
nx.draw_networkx_edge_labels(subgraph, pos, edge_labels, font_size=6)
plt.title(f"Knowledge Graph Subgraph (Center: {self.entities[center_entity].name})")
plt.axis('off')
plt.tight_layout()
plt.show()
def build_sample_kg():
"""构建示例知识图谱"""
kg = KnowledgeGraph()
entities = [
Entity("person_1", "张三", "Person", {"age": 30, "occupation": "工程师"}),
Entity("person_2", "李四", "Person", {"age": 28, "occupation": "设计师"}),
Entity("company_1", "科技公司A", "Company", {"industry": "软件开发", "size": "大型"}),
Entity("project_1", "AI项目", "Project", {"status": "进行中", "budget": 1000000}),
Entity("tech_1", "机器学习", "Technology", {"category": "AI", "maturity": "成熟"})
]
for entity in entities:
kg.add_entity(entity)
relations = [
Relation("rel_1", "works_for", "person_1", "company_1", {"start_date": "2020-01-01"}),
Relation("rel_2", "works_for", "person_2", "company_1", {"start_date": "2021-06-01"}),
Relation("rel_3", "manages", "person_1", "project_1", {"role": "项目经理"}),
Relation("rel_4", "participates_in", "person_2", "project_1", {"role": "设计师"}),
Relation("rel_5", "uses_technology", "project_1", "tech_1", {"usage_level": "核心"}),
Relation("rel_6", "knows", "person_1", "tech_1", {"proficiency": "专家"})
]
for relation in relations:
kg.add_relation(relation)
return kg
if __name__ == "__main__":
kg = build_sample_kg()
print("张三的邻居实体：", kg.get_neighbors("person_1"))
print("张三工作的公司：", kg.get_neighbors("person_1", "works_for"))
info = kg.get_entity_info("person_1")
print("\n张三的详细信息：")
print(f"姓名: {info['entity'].name}")
print(f"类型: {info['entity'].type}")
print(f"属性: {info['entity'].properties}")
print(f"出度关系: {info['outgoing_relations']}")
print(f"入度关系: {info['incoming_relations']}")
|