Knowledge Graphs

Knowledge graphs represent structured knowledge as entities and relationships. They power search engines, recommendation systems, and question answering. Unlike unstructured text, knowledge graphs make relationships explicit and queryable - enabling reasoning, entity disambiguation, and multi-hop inference.

Where It Is Used

Knowledge graphs enable intelligent applications:

• Search engines: Google Knowledge Graph, Bing Satori
• Virtual assistants: Alexa, Siri, Google Assistant
• E-commerce: Product graphs, catalog enrichment
• Healthcare: Drug interactions, medical ontologies
• Enterprise: Customer 360, supply chain visibility
• Fraud detection: Entity resolution, relationship analysis

Notable examples: Wikidata, DBpedia, YAGO, enterprise knowledge bases.

Typical Data

What You Do in Practice

1. Define scope - What entities and relations matter?
2. Design schema - Ontology, types, constraints
3. Extract entities - NER, entity linking
4. Extract relations - Pattern matching, ML extraction
5. Resolve entities - Deduplicate, merge records
6. Validate quality - Consistency, completeness
7. Enable querying - SPARQL, graph queries
8. Train embeddings - For ML integration
9. Build applications - Search, QA, recommendations

Baseline First

Start with rule-based extraction and simple matching:

import re
from collections import defaultdict

class SimpleKnowledgeGraph:
    """Basic knowledge graph with triple store."""

    def __init__(self):
        self.triples = set()
        self.entity_index = defaultdict(set)
        self.relation_index = defaultdict(set)

    def add_triple(self, head, relation, tail):
        """Add a triple to the graph."""
        triple = (head, relation, tail)
        self.triples.add(triple)

        self.entity_index[head].add(triple)
        self.entity_index[tail].add(triple)
        self.relation_index[relation].add(triple)

    def query(self, head=None, relation=None, tail=None):
        """Query triples by pattern."""
        results = self.triples

        if head is not None:
            results = results & self.entity_index[head]
        if tail is not None:
            results = results & self.entity_index[tail]
        if relation is not None:
            results = results & self.relation_index[relation]

        return list(results)

    def get_neighbors(self, entity, relation=None):
        """Get all entities connected to given entity."""
        neighbors = []

        for h, r, t in self.entity_index[entity]:
            if relation is None or r == relation:
                if h == entity:
                    neighbors.append((r, t, 'outgoing'))
                else:
                    neighbors.append((r, h, 'incoming'))

        return neighbors

    def path_exists(self, start, end, max_hops=3):
        """Check if path exists between entities."""
        visited = {start}
        frontier = {start}

        for hop in range(max_hops):
            next_frontier = set()

            for entity in frontier:
                for _, neighbor, _ in self.get_neighbors(entity):
                    if neighbor == end:
                        return True
                    if neighbor not in visited:
                        next_frontier.add(neighbor)
                        visited.add(neighbor)

            frontier = next_frontier
            if not frontier:
                break

        return False

class RuleBasedExtractor:
    """Extract relations using patterns."""

    def __init__(self):
        self.patterns = {
            'born_in': [
                r'(\w+) was born in (\w+)',
                r'(\w+), born in (\w+)',
            ],
            'works_at': [
                r'(\w+) works at (\w+)',
                r'(\w+) is.*employee of (\w+)',
            ],
            'located_in': [
                r'(\w+) is located in (\w+)',
                r'(\w+), (\w+)',  # City, Country format
            ]
        }

    def extract(self, text):
        """Extract triples from text."""
        triples = []

        for relation, patterns in self.patterns.items():
            for pattern in patterns:
                matches = re.findall(pattern, text, re.IGNORECASE)
                for match in matches:
                    if len(match) == 2:
                        triples.append((match[0], relation, match[1]))

        return triples

Knowledge Graph Embeddings

Learn vector representations for entities and relations:

import torch
import torch.nn as nn
import torch.nn.functional as F
import numpy as np

class TransE(nn.Module):
    """TransE embedding model: h + r ≈ t"""

    def __init__(self, num_entities, num_relations, embedding_dim=100):
        super().__init__()
        self.entity_embeddings = nn.Embedding(num_entities, embedding_dim)
        self.relation_embeddings = nn.Embedding(num_relations, embedding_dim)

        # Initialize
        nn.init.xavier_uniform_(self.entity_embeddings.weight)
        nn.init.xavier_uniform_(self.relation_embeddings.weight)

        # Normalize relation embeddings
        self.relation_embeddings.weight.data = F.normalize(
            self.relation_embeddings.weight.data, p=2, dim=1
        )

    def forward(self, heads, relations, tails):
        """Score triples."""
        h = self.entity_embeddings(heads)
        r = self.relation_embeddings(relations)
        t = self.entity_embeddings(tails)

        # Score: negative L2 distance
        score = -torch.norm(h + r - t, p=2, dim=1)
        return score

    def predict_tail(self, head, relation, top_k=10):
        """Predict most likely tail entities."""
        h = self.entity_embeddings(head)
        r = self.relation_embeddings(relation)

        # Score all entities as potential tails
        all_entities = self.entity_embeddings.weight
        scores = -torch.norm(h + r - all_entities, p=2, dim=1)

        # Get top-k
        top_scores, top_indices = torch.topk(scores, top_k)
        return top_indices, top_scores

class ComplEx(nn.Module):
    """ComplEx: Complex-valued embeddings for asymmetric relations."""

    def __init__(self, num_entities, num_relations, embedding_dim=100):
        super().__init__()
        self.entity_re = nn.Embedding(num_entities, embedding_dim)
        self.entity_im = nn.Embedding(num_entities, embedding_dim)
        self.relation_re = nn.Embedding(num_relations, embedding_dim)
        self.relation_im = nn.Embedding(num_relations, embedding_dim)

    def forward(self, heads, relations, tails):
        """Score triples using complex dot product."""
        h_re = self.entity_re(heads)
        h_im = self.entity_im(heads)
        r_re = self.relation_re(relations)
        r_im = self.relation_im(relations)
        t_re = self.entity_re(tails)
        t_im = self.entity_im(tails)

        # Complex trilinear product
        score = (h_re * r_re * t_re +
                 h_re * r_im * t_im +
                 h_im * r_re * t_im -
                 h_im * r_im * t_re).sum(dim=1)

        return score

def train_kg_embeddings(model, train_triples, num_entities, 
                        epochs=100, batch_size=1024, margin=1.0):
    """Train KG embeddings with negative sampling."""
    optimizer = torch.optim.Adam(model.parameters(), lr=0.001)

    for epoch in range(epochs):
        model.train()
        total_loss = 0

        # Shuffle triples
        np.random.shuffle(train_triples)

        for i in range(0, len(train_triples), batch_size):
            batch = train_triples[i:i+batch_size]

            heads = torch.tensor([t[0] for t in batch])
            relations = torch.tensor([t[1] for t in batch])
            tails = torch.tensor([t[2] for t in batch])

            # Generate negative samples (corrupt tails)
            neg_tails = torch.randint(0, num_entities, (len(batch),))

            # Positive and negative scores
            pos_scores = model(heads, relations, tails)
            neg_scores = model(heads, relations, neg_tails)

            # Margin ranking loss
            loss = F.relu(margin - pos_scores + neg_scores).mean()

            optimizer.zero_grad()
            loss.backward()
            optimizer.step()

            total_loss += loss.item()

        if epoch % 10 == 0:
            print(f'Epoch {epoch}: Loss {total_loss:.4f}')

    return model

Entity Resolution

Matching entities across sources:

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
import numpy as np

class EntityResolver:
    """Resolve and merge duplicate entities."""

    def __init__(self, similarity_threshold=0.8):
        self.threshold = similarity_threshold
        self.vectorizer = TfidfVectorizer(ngram_range=(1, 3))

    def find_matches(self, entities_a, entities_b):
        """Find matching entities between two sources."""
        # Create text representations
        texts_a = [self._entity_to_text(e) for e in entities_a]
        texts_b = [self._entity_to_text(e) for e in entities_b]

        # Vectorize
        all_texts = texts_a + texts_b
        vectors = self.vectorizer.fit_transform(all_texts)

        vectors_a = vectors[:len(texts_a)]
        vectors_b = vectors[len(texts_a):]

        # Compute similarities
        similarities = cosine_similarity(vectors_a, vectors_b)

        # Find matches above threshold
        matches = []
        for i, row in enumerate(similarities):
            best_j = np.argmax(row)
            if row[best_j] >= self.threshold:
                matches.append({
                    'entity_a': entities_a[i],
                    'entity_b': entities_b[best_j],
                    'similarity': row[best_j]
                })

        return matches

    def _entity_to_text(self, entity):
        """Convert entity to searchable text."""
        parts = [entity.get('name', '')]

        if 'aliases' in entity:
            parts.extend(entity['aliases'])
        if 'description' in entity:
            parts.append(entity['description'])

        return ' '.join(parts).lower()

    def merge_entities(self, entity_a, entity_b):
        """Merge two matched entities."""
        merged = {}

        # Take non-null values, prefer entity_a
        for key in set(entity_a.keys()) | set(entity_b.keys()):
            val_a = entity_a.get(key)
            val_b = entity_b.get(key)

            if val_a is not None and val_b is not None:
                # Merge lists
                if isinstance(val_a, list) and isinstance(val_b, list):
                    merged[key] = list(set(val_a + val_b))
                else:
                    merged[key] = val_a  # Prefer source A
            else:
                merged[key] = val_a or val_b

        return merged

Relation Extraction with ML

import torch
from transformers import BertTokenizer, BertModel

class NeuralRelationExtractor:
    """Extract relations using BERT."""

    def __init__(self, model_name='bert-base-uncased', num_relations=10):
        self.tokenizer = BertTokenizer.from_pretrained(model_name)
        self.bert = BertModel.from_pretrained(model_name)
        self.classifier = torch.nn.Linear(768 * 2, num_relations)

    def extract_relations(self, text, entity_pairs):
        """Extract relations between entity pairs."""
        results = []

        for head_span, tail_span in entity_pairs:
            # Mark entities in text
            marked_text = self._mark_entities(text, head_span, tail_span)

            # Encode
            inputs = self.tokenizer(marked_text, return_tensors='pt',
                                    truncation=True, max_length=512)

            with torch.no_grad():
                outputs = self.bert(**inputs)

            # Get entity representations
            head_repr = self._get_entity_repr(outputs, head_span)
            tail_repr = self._get_entity_repr(outputs, tail_span)

            # Classify relation
            combined = torch.cat([head_repr, tail_repr], dim=-1)
            logits = self.classifier(combined)
            relation = logits.argmax(dim=-1).item()
            confidence = torch.softmax(logits, dim=-1).max().item()

            results.append({
                'head': text[head_span[0]:head_span[1]],
                'tail': text[tail_span[0]:tail_span[1]],
                'relation': relation,
                'confidence': confidence
            })

        return results

    def _mark_entities(self, text, head_span, tail_span):
        """Add entity markers to text."""
        # Sort spans to handle overlapping correctly
        spans = [(head_span, '[HEAD]', '[/HEAD]'), 
                 (tail_span, '[TAIL]', '[/TAIL]')]
        spans.sort(key=lambda x: x[0][0], reverse=True)

        for (start, end), open_tag, close_tag in spans:
            text = text[:end] + close_tag + text[end:]
            text = text[:start] + open_tag + text[start:]

        return text

Evaluation Metrics

Failure Modes and Debugging Checklist

• [ ] Sparse knowledge: Too few triples for embedding quality
• [ ] Schema drift: Data doesn't match ontology
• [ ] Entity ambiguity: Same name, different entities
• [ ] Relation sparsity: Some relations have few examples
• [ ] Temporal issues: Knowledge changes over time
• [ ] Incompleteness: Open-world assumption violations
• [ ] Noise: Incorrect triples from extraction

Production Checklist

Before deployment: - [ ] Schema documented and validated - [ ] Entity resolution pipeline tested - [ ] Quality metrics established - [ ] Query patterns optimized - [ ] Embedding model evaluated

After deployment: - [ ] Continuous ingestion pipeline - [ ] Quality monitoring - [ ] Schema evolution process - [ ] User feedback integration - [ ] Retraining schedule for embeddings

Tools and Stack

Example Mini-Case

Scenario: E-commerce product knowledge graph.

Graph construction: - Entities: Products, brands, categories, attributes - Relations: made_by, belongs_to, has_attribute, similar_to - Sources: Catalog data, reviews, web scraping

Approach: 1. Extracted 2M products, 50K brands 2. Entity resolution merged duplicates (15% reduction) 3. Trained ComplEx embeddings for recommendations 4. Enabled "similar products" and "compatible accessories"

Results: - 23% increase in cross-sell revenue - Reduced catalog inconsistencies by 60% - Powered new "shop by attribute" feature - 50ms query latency for recommendations

Related Pages

• Index: ML Field Guide
• Node Classification
• Named Entity Recognition
• RAG Systems

Further Reading

• Knowledge Graphs - ACM Computing Surveys
• PyKEEN Documentation
• Neo4j Graph Academy

Try it live

Everything above runs in your browser — open Knowledge Graph Builder: Triples, Entities & Shortest-Path Inference and change the parameters while it is running. Nothing is installed, nothing is uploaded, the whole model lives in one tab.