Knowledge Graphs: структурирование знаний для LLM
opensourceaillmknowledge-graphkgrdfit
Введение: Knowledge Graphs
Что такое Knowledge Graph?
Knowledge Graph = структурированное знание
Структура:
(Subject) --[Predicate]--> (Object)
Примеры:
(Москва) --[является_столицей]--> (Россия)
(Python) --[тип]--> (Язык_программирования)
(Python) --[создан]--> (Guido_van_Rossum)
(Python) --[поддерживает]--> (ООП)
Факт: Google Knowledge Graph содержит более 3.5 миллиарда сущностей и 63 миллиарда связей.
Основы Knowledge Graphs
RDF (Resource Description Framework)
RDF Triple:
Subject | Predicate | Object
<http://example.org/Москва> | <http://example.org/столица> | <http://example.org/Россия>
Можно представить как граф:
Москва --столица--> Россия
RDF Store:
- Вершины = ресурсы (URI)
- Рёбра = предикаты (URI)
- Направленные рёбра
OWL (Web Ontology Language)
OWL определяет:
1. Классы (Classes)
2. Свойства (Properties)
3. Индивиды (Individuals)
4. Ограничения (Constraints)
Пример онтологии:
Class: Person
SubclassOf: hasParent Person
DisjointWith: Animal
Property: hasParent
Domain: Person
Range: Person
Inverse: hasChild
# Representing OWL ontology in Python
class Ontology:
def __init__(self):
self.classes = {}
self.properties = {}
self.instances = {}
def define_class(self, name, superclass=None, constraints=None):
self.classes[name] = {
'superclass': superclass,
'constraints': constraints or [],
'subclasses': []
}
if superclass:
self.classes[superclass]['subclasses'].append(name)
def define_property(self, name, domain, range_type,
inverse=None, cardinality=None):
self.properties[name] = {
'domain': domain,
'range': range_type,
'inverse': inverse,
'cardinality': cardinality
}
def add_instance(self, class_name, instance_id, properties=None):
if class_name not in self.instances:
self.instances[class_name] = {}
self.instances[class_name][instance_id] = properties or {}
Building Knowledge Graphs
Schema Design
Schema (TBox) vs Data (ABox)
Schema:
Class: Article
Properties: hasAuthor, hasTopic, publishedDate
Class: Author
Properties: wrote, affiliation
Property: wrote
Domain: Author
Range: Article
Data:
(Alice) --wrote--> (Paper1)
(Alice) --affiliation--> (MIT)
(Paper1) --hasTopic--> (ML)
class KnowledgeGraph:
def __init__(self):
self.nodes = {} # id -> {label, properties}
self.edges = [] # [(src, pred, dst, properties)]
self.indexes = {
'by_label': {}, # label -> [ids]
'by_type': {}, # type -> [ids]
}
def add_node(self, id, label, properties=None):
self.nodes[id] = {
'label': label,
'properties': properties or {}
}
if label not in self.indexes['by_label']:
self.indexes['by_label'][label] = []
self.indexes['by_label'][label].append(id)
def add_edge(self, src, pred, dst, properties=None):
self.edges.append({
'src': src,
'pred': pred,
'dst': dst,
'properties': properties or {}
})
def get_neighbors(self, node_id, direction='both'):
"""Get connected nodes"""
neighbors = []
for edge in self.edges:
if direction in ['in', 'both'] and edge['dst'] == node_id:
neighbors.append((edge['src'], edge['pred'], 'in'))
if direction in ['out', 'both'] and edge['src'] == node_id:
neighbors.append((edge['dst'], edge['pred'], 'out'))
return neighbors
Entity Extraction
class EntityExtractor:
"""
Extract entities and relations from text
to build Knowledge Graph triples
"""
def __init__(self, llm_client):
self.llm = llm_client
self.entity_types = [
'Person', 'Organization', 'Location',
'Date', 'Number', 'Product', 'Concept'
]
def extract(self, text):
"""Extract triples from text"""
prompt = f"""
Extract knowledge triples from the following text.
Use these entity types: {', '.join(self.entity_types)}
Text: {text}
Return format:
(Entity1, Relation, Entity2, Type1, Type2)
"""
response = self.llm.generate(prompt, max_tokens=500)
triples = self._parse_triples(response)
return triples
def _parse_triples(self, text):
"""Parse triple format from LLM response"""
triples = []
for line in text.strip().split('\n'):
line = line.strip().strip('()')
parts = [p.strip() for p in line.split(',')]
if len(parts) >= 5:
triples.append({
'subject': parts[0],
'predicate': parts[1],
'object': parts[2],
'subject_type': parts[3],
'object_type': parts[4]
})
return triples
def build_graph(self, texts):
"""Build KG from multiple texts"""
kg = KnowledgeGraph()
for text in texts:
triples = self.extract(text)
for t in triples:
# Add entities as nodes
if t['subject'] not in kg.nodes:
kg.add_node(t['subject'], t['subject_type'])
if t['object'] not in kg.nodes:
kg.add_node(t['object'], t['object_type'])
# Add relation as edge
kg.add_edge(t['subject'], t['predicate'], t['object'])
return kg
Knowledge Graph Embeddings
TransE
TransE: Embed entities and relations in vector space
Core assumption:
h + r ≈ t
Where:
h = head entity embedding
r = relation embedding
t = tail entity embedding
Loss function:
L = Σ γ - ||h + r - t|| + ||h + r - t'||
γ = margin
(h, r, t) = valid triple
(h, r, t') = corrupted triple
class TransE(torch.nn.Module):
def __init__(self, num_entities, num_relations, dim=100, margin=1.0):
super().__init__()
self.num_entities = num_entities
self.num_relations = num_relations
self.dim = dim
self.margin = margin
# Embeddings
self.entity_embeddings = torch.nn.Embedding(
num_entities, dim, sparse=True
)
self.relation_embeddings = torch.nn.Embedding(
num_relations, dim, sparse=True
)
# Initialize
torch.nn.init.xavier_uniform_(self.entity_embeddings.weight)
torch.nn.init.xavier_uniform_(self.relation_embeddings.weight)
def forward(self, pos_head, pos_tail, pos_rel,
neg_head, neg_tail, neg_rel):
"""
pos: positive (valid) triple
neg: negative (corrupted) triple
"""
# Get embeddings
h = self.entity_embeddings(pos_head)
t = self.entity_embeddings(pos_tail)
r = self.relation_embeddings(pos_rel)
nh = self.entity_embeddings(neg_head)
nt = self.entity_embeddings(neg_tail)
nr = self.relation_embeddings(neg_rel)
# Compute distances
pos_dist = torch.norm(h + r - t, p=1, dim=1)
neg_dist = torch.norm(nh + nr - nt, p=1, dim=1)
# Margin-based loss
loss = torch.relu(self.margin - pos_dist + neg_dist)
return loss.sum()
def score_triples(self, heads, tails, rels):
"""Score triples (lower = more valid)"""
h = self.entity_embeddings(heads)
t = self.entity_embeddings(tails)
r = self.relation_embeddings(rels)
return -torch.norm(h + r - t, p=1, dim=1)
RotatE
RotatE: Entities in complex space, relations as rotations
Key idea:
h + r = t → h ⊙ r = t
Where:
h, t ∈ ℂ^d (complex vectors)
r = rotation matrix (|r_i| = 1)
⊙ = element-wise product
Property:
If (a, r, b) and (b, r, c), then (a, r, c)
→ Can model composition!
class RotatE(torch.nn.Module):
def __init__(self, num_entities, num_relations, dim=200, margin=1.0):
super().__init__()
self.num_entities = num_entities
self.num_relations = num_relations
self.dim = dim # must be even
self.margin = margin
self.entity_embeddings = torch.nn.Embedding(
num_entities, dim
)
self.relation_embeddings = torch.nn.Embedding(
num_relations, dim
)
torch.nn.init.uniform_(self.entity_embeddings.weight, -1, 1)
torch.nn.init.uniform_(self.relation_embeddings.weight, -1, 1)
def _compute_score(self, h, r, t):
"""Compute score for (h, r, t)"""
# Convert to complex
h_real, h_imag = self._to_complex(h)
r_real, r_imag = self._to_complex(r)
t_real, t_imag = self._to_complex(t)
# Rotation: h ⊙ r = t
# (a+bi)(c+di) = (ac-bd) + (ad+bc)i
hr = h_real * r_real - h_imag * r_imag
hi = h_real * r_imag + h_imag * r_real
# Distance
dist = torch.sqrt((hr - t_real)**2 + (hi - t_imag)**2 + 1e-6)
return dist
def forward(self, pos_h, pos_t, pos_r, neg_h, neg_t, neg_r):
pos_dist = self._compute_score(pos_h, pos_r, pos_t)
neg_dist = self._compute_score(neg_h, neg_r, neg_t)
loss = torch.relu(self.margin - pos_dist + neg_dist)
return loss.sum()
def _to_complex(self, x):
"""Normalize to unit circle"""
x = torch.nn.functional.normalize(x, p=1, dim=-1)
angle = x * torch.pi # Map to [-π, π]
return torch.cos(angle), torch.sin(angle)
Knowledge Graphs with LLMs
KG-Enhanced Prompting
Prompt template with KG context:
Question: "Who directed Inception?"
KG retrieval:
(Inception) --[director]--> (Christopher_Nolan)
(Christopher_Nolan) --[born]--> (1970)
(Christopher_Nolan) --[worked_with]--> (Christian_Bale)
Prompt:
"Based on the knowledge:
- Inception was directed by Christopher Nolan
- Christopher Nolan was born in 1970
- Christopher Nolan worked with Christian Bale
Question: Who directed Inception?
Answer: Christopher Nolan"
class KGEnhancedLLM:
"""
LLM with Knowledge Graph context
"""
def __init__(self, llm_client, kg):
self.llm = llm_client
self.kg = kg
def answer(self, question):
"""Answer question using KG + LLM"""
# 1. Extract entities from question
entities = self._extract_entities(question)
# 2. Query KG
kg_context = self._query_kg(entities, question)
# 3. Build prompt
prompt = f"""
Knowledge Graph context:
{kg_context}
Question: {question}
Answer based on the knowledge above.
"""
# 4. Get answer
answer = self.llm.generate(prompt)
return {
'question': question,
'kg_context': kg_context,
'answer': answer
}
def _query_kg(self, entities, question):
"""Query KG for relevant triples"""
triples = []
for entity in entities:
neighbors = self.kg.get_neighbors(entity)
for neighbor, pred, direction in neighbors:
triples.append(f"({entity}, {pred}, {neighbor})")
return '\n'.join(triples[:20]) # Limit context
Graph RAG
Graph RAG = Knowledge Graph + RAG
Flow:
1. Build KG from documents
2. Embed entities and relations
3. For query:
a. Retrieve relevant subgraph
b. Generate summary of subgraph
c. Use LLM to answer with summaries
class GraphRAG:
def __init__(self, kg, llm_client, embedding_model):
self.kg = kg
self.llm = llm_client
self.embeddings = embedding_model
def index(self, documents):
"""Build KG index from documents"""
for doc in documents:
# Extract entities and relations
triples = self._extract_from_text(doc.text)
# Add to KG
for t in triples:
self.kg.add_node(t['subject'], t['type'])
self.kg.add_edge(
t['subject'], t['relation'], t['object']
)
def query(self, question):
"""Answer question using Graph RAG"""
# 1. Get entity embeddings
query_emb = self.embeddings.encode(question)
# 2. Find similar entities in KG
similar = self._find_similar_entities(query_emb)
# 3. Extract subgraph
subgraph = self._extract_subgraph(similar, max_nodes=50)
# 4. Generate subgraph summary
summary = self._summarize_subgraph(subgraph)
# 5. Answer with LLM
answer = self.llm.generate(
f"Subgraph summary: {summary}\n\n"
f"Question: {question}\n\nAnswer:"
)
return answer
def _summarize_subgraph(self, subgraph):
"""Generate natural language summary of subgraph"""
triples_text = '\n'.join([
f"({s}, {p}, {o})" for s, p, o in subgraph
])
prompt = f"""
Summarize the following knowledge graph:
{triples_text}
Provide a concise summary of the key facts.
"""
return self.llm.generate(prompt, max_tokens=200)
Knowledge Graph Databases
Graph Query Languages
Cypher (Neo4j):
MATCH (person:Person {name: 'Alice'})-[:FRIEND]->(friend)
RETURN friend.name, friend.age
CREATE (p:Person {name: 'Bob', age: 30})
CREATE (p)-[:LIKES]->(:Topic {name: 'AI'})
SPARQL (RDF):
SELECT ?person ?age
WHERE {
?person a :Person .
?person :name "Alice" .
?person :age ?age .
}
Gremlin:
g.V().has('name', 'Alice')
.out('FRIEND').values('name')
# Neo4j with Python
from neo4j import GraphDatabase
class Neo4jKG:
def __init__(self, uri, user, password):
self.driver = GraphDatabase.driver(uri, auth=(user, password))
def close(self):
self.driver.close()
def add_entity(self, name, label, properties=None):
"""Add or merge entity"""
with self.driver.session() as session:
session.execute_write(
self._add_entity_tx, name, label, properties
)
def _add_entity_tx(self, tx, name, label, props):
query = f"""
MERGE (e:{label} {{name: $name}})
SET e += $properties
"""
tx.run(query, name=name, properties=props or {})
def add_relation(self, from_name, to_name, relation):
"""Add relation between entities"""
with self.driver.session() as session:
session.execute_write(
self._add_relation_tx, from_name, to_name, relation
)
def _add_relation_tx(self, tx, from_name, to_name, relation):
tx.run("""
MATCH (a {name: $from})
WITH a
MATCH (b {name: $to})
CREATE (a)-[r:`{relation}`]->(b)
RETURN r
""".format(relation=relation),
from=from_name, to=to_name
)
def query(self, cypher, params=None):
"""Execute Cypher query"""
with self.driver.session() as session:
result = session.execute_read(
self._query_tx, cypher, params
)
return list(result)
def _query_tx(self, tx, cypher, params):
return tx.run(cypher, params or {})
Applications
Semantic Search
KG-enhanced search:
Query: "Italian restaurants near me"
1. Parse query → entities + constraints
2. Query KG for related entities
3. Expand with synonyms, categories
4. Rank results using graph proximity
Question Answering
KG-based QA:
Q: "What is Christopher Nolan's birthplace?"
1. Entity link: "Christopher Nolan" → node
2. Relation extraction: "birthplace" → bornIn
3. Query: (Christopher_Nolan)-[:bornIn]->(?)
4. Answer: London
Recommendation
KG-based recommendation:
User: Alice likes Sci-Fi movies
Path: Alice → likes → Sci-Fi ← likes → Interstellar
← likes → Inception
Score: Path strength × entity importance
Recommend: Interstellar, Inception
Заключение
Knowledge Graphs provide:
- Structured knowledge: Machine-readable facts
- Reasoning: Inference over relationships
- Interpretability: Human-readable structure
- Integration: Combine multiple data sources
Key technologies:
- RDF/OWL for representation
- Graph databases (Neo4j, Amazon Neptune)
- Embeddings (TransE, RotatE, ComplEx)
- KG-LLM integration (Graph RAG)
Applications:
- Semantic search
- Question answering
- Recommendation systems
- Data integration