semantica/docs/reference/kg.md

title, description, icon

title	description	icon
Knowledge Graph Module	Graph construction, temporal models, analytics, similarity scoring, and structural embeddings.	diagram-project

semantica.kg transforms extracted entities and relationships into structured, queryable knowledge graphs:

• Temporal nodes and edges with valid_from / valid_until windows and all 13 Allen interval relations
• Full graph analytics suite: centrality, community detection, path finding, link prediction
• Node2Vec structural embeddings for downstream ML and similarity scoring
• OWL-Time export and versioned snapshots via TemporalVersionManager
• Schema and constraint validation before persistence

Exported Classes

Class	Role
KnowledgeGraph	Core graph data structure: nodes, edges, properties, temporal validity
GraphBuilder	Construct from entities + relationships; pass merge_entities=True to enable deduplication
GraphBuilderWithProvenance	Wraps GraphBuilder with optional provenance tracking; pass provenance=True to enable
EntityResolver	Entity deduplication and merging during graph construction
GraphAnalyzer	Unified analytics wrapper: runs centrality, community detection, and connectivity in one call
ConnectivityAnalyzer	Connected component detection, bridge identification, density, and degree statistics
TemporalGraphQuery	Point-in-time snapshots, range queries, evolution analysis, temporal path finding
TemporalPatternDetector	Sequence and cycle pattern detection over temporal edges
TemporalReasoningEngine	All 13 Allen interval algebra relations over TemporalInterval objects
TemporalInterval	Frozen dataclass (start: datetime, end: datetime | TemporalBound, label?)
IntervalRelation	Enum of all 13 Allen relation labels (BEFORE, AFTER, MEETS, …)
BiTemporalFact	Dataclass wrapping valid_from, valid_until, recorded_at, superseded_at. Factory: BiTemporalFact.from_relationship(rel_dict)
TemporalBound	Sentinel enum for open-ended intervals — single value: TemporalBound.OPEN
TemporalNormalizer	Parse NL temporal expressions to (datetime, datetime) tuples — zero LLM calls
TemporalQueryRewriter	Extract temporal intent from free-text queries; returns TemporalQueryResult
TemporalQueryResult	Dataclass output of TemporalQueryRewriter.rewrite()
TemporalVersionManager	Versioned snapshots with SHA-256 integrity, SQLite-backed persistent storage
CentralityCalculator	PageRank, degree, betweenness, closeness, eigenvector centrality
CommunityDetector	Louvain, Leiden, Label Propagation, and K-Clique community detection
PathFinder	Dijkstra, A*, BFS, and K-Shortest path algorithms
LinkPredictor	Preferential Attachment, Jaccard, Adamic-Adar link prediction
NodeEmbedder	Node2Vec structural embeddings for downstream ML
SimilarityCalculator	Cosine, Euclidean, Manhattan, and correlation similarity scoring
GraphValidator	Schema and constraint validation before persistence
AlgorithmTrackerWithProvenance	Algorithm execution tracking with provenance metadata
AlgorithmRegistry / algorithm_registry	Registry for registered algorithms; algorithm_registry is the shared singleton
ProvenanceTracker	W3C PROV-O provenance tracking for graph operations
SeedManager	Reproducible random seed management across algorithms
KGConfig / kg_config	Module-level configuration; kg_config is the shared singleton

For conflict detection and advanced entity resolution, use `semantica.conflicts` and `semantica.deduplication` alongside this module.

<img src="/assets/img/diagrams/kg-structure.svg" alt="Knowledge graph entity and relation structure: Person, Organization, Location, Date nodes with typed labeled edges" style={{ width: '100%', borderRadius: '12px', margin: '0 0 24px' }} />

GraphBuilder

GraphBuilder constructs knowledge graphs from extracted entities and relationships. merge_entities defaults to False: pass True to enable entity deduplication during construction:

from semantica.kg import GraphBuilder

# Pass a dict with "entities" and "relationships" keys
builder = GraphBuilder(merge_entities=True)
kg = builder.build({"entities": entities, "relationships": relationships})

Method	Returns	Description
build(sources)	dict	Build graph from a dict, list of dicts, or list of entity/relation objects
build_single_source(data)	dict	Build graph from a single data source dict

Temporal Knowledge Graphs (v0.4.0+)

Full temporal reference including `BiTemporalFact`, `TemporalReasoningEngine`, Allen interval algebra, and `TemporalNormalizer` is covered in the dedicated [Temporal Intelligence](temporal) page. This section documents the KG-layer temporal API.

The temporal stack — see the Temporal Intelligence page for the full reference.

Building a Temporal Graph

from semantica.kg import GraphBuilder, TemporalGraphQuery, TemporalVersionManager

builder = GraphBuilder()
kg = builder.build(sources=[
    {
        "entities": [
            {"id": "alice",     "type": "Person"},
            {"id": "acme_corp", "type": "Organization"},
            {"id": "beta_ltd",  "type": "Organization"},
        ],
        "relationships": [
            {
                "source": "alice", "target": "acme_corp", "type": "ceo_of",
                "valid_from":  "2018-01-01",
                "valid_until": "2022-06-01",
            },
            {
                "source": "alice", "target": "beta_ltd", "type": "ceo_of",
                "valid_from":  "2022-06-01",
                # No valid_until → open-ended (TemporalBound.OPEN)
            },
        ],
    }
])

Point-in-Time Queries

TemporalGraphQuery accepts optional constructor args; pass the graph into each query call:

from semantica.kg import TemporalGraphQuery

query = TemporalGraphQuery(
    temporal_granularity="day",        # second|minute|hour|day|week|month|year
    enable_temporal_reasoning=True,
)

# Primary API: query_at_time returns counts + filtered data
result_2020 = query.query_at_time(kg, "", at_time="2020-06-15")
result_2023 = query.query_at_time(kg, "", at_time="2023-01-01")
print(f"Rels in 2020: {result_2020['num_relationships']}")

# Low-level: reconstruct_at_time returns a deep-copied subgraph dict
snapshot = query.reconstruct_at_time(kg, "2020-06-15")

# Range query: all relationships active during any part of 2021
range_result = query.query_time_range(kg, "", "2021-01-01", "2021-12-31")

# Compare two snapshots: use TemporalVersionManager.compare_versions()
# (temporal_diff() does not exist — see TemporalVersionManager below)

Bi-Temporal Facts

BiTemporalFact is a dataclass — use the from_relationship() factory, not a positional constructor:

from semantica.kg import BiTemporalFact, TemporalBound

rel = {
    "source": "alice", "target": "acme_corp", "type": "ceo_of",
    "valid_from":    "2018-01-01",
    "valid_until":   "2022-06-01",
    "recorded_at":   "2018-01-05T09:32:00Z",
    "superseded_at": None,   # None → TemporalBound.OPEN (still current)
}
fact = BiTemporalFact.from_relationship(rel)

print(fact.valid_from)      # datetime(2018, 1, 1, tzinfo=utc)
print(fact.valid_until)     # datetime(2022, 6, 1, tzinfo=utc)
print(fact.superseded_at)   # TemporalBound.OPEN

# Open-ended fact (no valid_until → TemporalBound.OPEN)
open_rel = {"source": "alice", "target": "beta_ltd", "type": "ceo_of",
            "valid_from": "2022-06-01"}
open_fact = BiTemporalFact.from_relationship(open_rel)
print(open_fact.valid_until)   # TemporalBound.OPEN

# Serialize back to dict fields for storage
fields = fact.to_relationship_fields()

Allen Interval Algebra

TemporalReasoningEngine implements all 13 Allen relations deterministically — no LLM, no probability. It operates on TemporalInterval objects (not plain dicts):

from semantica.kg import (
    TemporalReasoningEngine, TemporalInterval, IntervalRelation
)
from datetime import datetime, timezone

def dt(y, m, d): return datetime(y, m, d, tzinfo=timezone.utc)

engine = TemporalReasoningEngine()

h1_2020 = TemporalInterval(start=dt(2020, 1, 1), end=dt(2020, 6, 30))
q2_q4   = TemporalInterval(start=dt(2020, 4, 1), end=dt(2020, 12, 31))

relation = engine.relation(h1_2020, q2_q4)   # primary method
print(relation)          # IntervalRelation.OVERLAPS
print(relation.value)    # "overlaps"

print(engine.overlaps(h1_2020, q2_q4))  # True
print(engine.contains(q2_q4, h1_2020))  # False
print(engine.active_at(h1_2020, dt(2020, 3, 15)))  # True

IntervalRelation	.value	Description
BEFORE	"before"	A ends strictly before B starts
MEETS	"meets"	A ends exactly when B starts
OVERLAPS	"overlaps"	A and B share a period; A starts and ends first
STARTS	"starts"	Same start; A ends before B
DURING	"during"	A is entirely within B
FINISHES	"finishes"	Same end; B started earlier
EQUALS	"equals"	Identical interval
AFTER, MET_BY, OVERLAPPED_BY, STARTED_BY, CONTAINS, FINISHED_BY	(inverses)	Mirror relations

Natural Language Temporal Parsing

from semantica.kg import TemporalNormalizer, TemporalQueryRewriter
from datetime import datetime, timezone

# reference_date set at construction time (required for relative phrases)
norm = TemporalNormalizer(reference_date=datetime(2024, 6, 15, tzinfo=timezone.utc))

# Returns Optional[Tuple[datetime, datetime]] — not a dict
result = norm.normalize("last quarter")
start, end = result
print(start)   # datetime(2024, 1, 1, tzinfo=utc)
print(end)     # datetime(2024, 3, 31, tzinfo=utc)

result = norm.normalize("2022")
# (datetime(2022, 1, 1, tzinfo=utc), datetime(2022, 12, 31, tzinfo=utc))

result = norm.normalize("unparseable phrase")
print(result)  # None

# TemporalQueryRewriter: primary method is rewrite(), returns TemporalQueryResult
rewriter = TemporalQueryRewriter()
result = rewriter.rewrite("Who was CEO before the 2022 restructuring?")
print(result.temporal_intent)    # "before"
print(result.at_time.year)       # 2022
print(result.rewritten_query)    # "Who was CEO"
print(result.confidence)         # 0.85
print(result.has_temporal_context())  # True

Versioned Snapshots

from semantica.kg import TemporalVersionManager

# In-memory (default); pass storage_path="versions.db" for SQLite persistence
versioner = TemporalVersionManager()

# author and description are required for create_snapshot
versioner.create_snapshot(kg, version_label="2024-Q1",
                          author="user@example.com",
                          description="Q1 2024 baseline")

# List versions (not list_snapshots)
for v in versioner.list_versions():
    print(f"{v['label']:12s}  {v['author']}")

# Compare two versions (not diff_versions)
diff = versioner.compare_versions("2023-Q4", "2024-Q1")
print(f"Entities added:      {diff['summary']['entities_added']}")
print(f"Relationships added: {diff['summary']['relationships_added']}")

# Retrieve a version (not restore_snapshot)
past_kg = versioner.get_version("2023-Q4")

# SHA-256 integrity check
versioner.verify_checksum(past_kg)

See the [Temporal Intelligence](temporal) reference for the full class API, domain examples (personnel changes, policy evolution, financial timelines), and configuration options.

Similarity Scoring

SimilarityCalculator computes cosine, Euclidean, Manhattan, and correlation similarity between node embeddings:

from semantica.kg import SimilarityCalculator, NodeEmbedder

# First compute structural embeddings
embedder   = NodeEmbedder(method="node2vec", embedding_dimension=128)
embeddings = embedder.compute_embeddings(kg, ["Person", "Organization"], ["RELATED_TO"])

# Then compare nodes by embedding similarity
calc  = SimilarityCalculator()
score = calc.cosine_similarity(embeddings["Apple Inc."], embeddings["Google"])
print(f"Apple–Google structural similarity: {score:.3f}")

# Find structurally similar nodes: returns List[str] of node IDs
similar = embedder.find_similar_nodes(kg, "Apple Inc.", top_k=5)
for node_id in similar:
    print(node_id)

Graph Analytics

Measure node importance across five algorithms. Use `calculate_all_centrality()` to run them all at once.

```python
from semantica.kg import CentralityCalculator

calculator = CentralityCalculator()

# Run all centrality measures at once
all_metrics = calculator.calculate_all_centrality(graph)

# Or run individually
pagerank    = calculator.calculate_pagerank(graph, damping_factor=0.85)
betweenness = calculator.calculate_betweenness_centrality(graph)
closeness   = calculator.calculate_closeness_centrality(graph)

# Get the top 10 most important nodes
top_nodes = calculator.get_top_nodes(pagerank, top_k=10)
```

| Method | Best for |
| :------ | :-------- |
| `calculate_degree_centrality()` | Most-connected nodes |
| `calculate_pagerank()` | Link-based influence (like Google PageRank) |
| `calculate_betweenness_centrality()` | Bottleneck / bridge nodes |
| `calculate_closeness_centrality()` | Nodes closest to all others |
| `calculate_eigenvector_centrality()` | Nodes connected to other high-influence nodes |

Discover clusters and communities within the graph. Louvain is the fastest; Leiden produces higher-quality partitions.

```python
from semantica.kg import CommunityDetector

detector = CommunityDetector()

# Louvain: fast, high quality (default)
communities = detector.detect_communities(graph, algorithm="louvain")

# Leiden: higher quality, slower
communities = detector.detect_communities_leiden(graph, resolution=1.2)

# Evaluate community quality
metrics = detector.calculate_community_metrics(graph, communities)
print(f"Modularity: {metrics['modularity']:.3f}")
print(f"Communities found: {metrics['num_communities']}")
```

| Algorithm | Strength |
| :--------- | :-------- |
| Louvain | Fast, good modularity: use for large graphs |
| Leiden | Best modularity: use when quality matters more than speed |
| Label Propagation | Near-linear time: use for very large graphs |
| K-Clique | Overlapping communities: nodes can belong to multiple groups |

Find shortest paths and route alternatives between any two nodes.

```python
from semantica.kg import PathFinder

finder = PathFinder()

# Dijkstra shortest path
path = finder.dijkstra_shortest_path(graph, "Alice", "Bob")
print(" → ".join(path["path"]))

# All shortest paths between two nodes
paths = finder.all_shortest_paths(graph, "source", "target")

# K-Shortest paths (alternative routes)
k_paths = finder.find_k_shortest_paths(graph, "source", "target", k=3)
```

| Algorithm | Use case |
| :--------- | :-------- |
| Dijkstra | Weighted shortest path: standard routing |
| A\* | Heuristic-guided search: faster on large sparse graphs |
| BFS | Unweighted shortest path: hop count only |
| K-Shortest | Multiple alternative routes |

Predict missing or future edges. Use to complete knowledge graphs or find implicit relationships.

```python
from semantica.kg import LinkPredictor

predictor = LinkPredictor(method="preferential_attachment")

# Predict the top 20 most likely missing edges
predicted = predictor.predict_links(graph, top_k=20)
for link in predicted:
    print(f"{link['source']} → {link['target']}  (score: {link['score']:.3f})")

# Score a specific pair
score = predictor.score_link(graph, "Alice", "CompanyX")
```

| Algorithm | Best for |
| :--------- | :-------- |
| Preferential Attachment | High-degree node connection prediction |
| Common Neighbors | Nodes with shared connections |
| Jaccard | Normalized common-neighbor overlap |
| Adamic-Adar | Weighted common neighbors (penalizes hubs) |
| Resource Allocation | Conservative: ignores high-degree intermediaries |

Compute structural embeddings with Node2Vec, then find similar nodes or feed into downstream ML.

```python
from semantica.kg import NodeEmbedder, SimilarityCalculator

# Compute Node2Vec embeddings
embedder = NodeEmbedder(method="node2vec", embedding_dimension=128)
embeddings = embedder.compute_embeddings(
    graph, ["Person", "Organization"], ["RELATED_TO"]
)

# Find structurally similar nodes
similar = embedder.find_similar_nodes(graph, "Apple Inc.", top_k=5)
for node_id in similar:
    print(node_id)

# Compare two specific nodes by embedding similarity
calc  = SimilarityCalculator()
score = calc.cosine_similarity(embeddings["Apple Inc."], embeddings["Google"])
print(f"Structural similarity: {score:.3f}")
```

<Note>
  `find_similar_nodes` returns `List[str]`: a list of node IDs, not node objects. Look up full node data via `graph["nodes"]`.
</Note>

Algorithm Summary

Category	Algorithms	Use Cases
Node Embeddings	Node2Vec	Structural similarity, node representation
Similarity	Cosine, Euclidean, Manhattan, Correlation	Node matching, recommendation
Path Finding	Dijkstra, A*, BFS, K-Shortest	Route planning, network analysis
Link Prediction	Preferential Attachment, Jaccard, Adamic-Adar	Network completion
Centrality	Degree, Betweenness, Closeness, PageRank	Influence analysis
Community Detection	Louvain, Leiden, Label Propagation	Social clustering
Connectivity	Components, Bridges, Density	Network robustness

GraphValidator

Validates graph structure: checks required fields, duplicate IDs, dangling edges, and optionally detects cycles and orphan nodes:

from semantica.kg import GraphValidator

validator = GraphValidator()
result    = validator.validate(kg)   # accepts the dict returned by GraphBuilder.build()

if result.is_valid:
    print("Graph is valid")
else:
    for issue in result.issues:
        print(f"{issue.severity.value}: {issue.message}")

Pass strict=True to treat warnings as errors. Pass a schema dict with "entity_types" and "relationship_types" keys to validate against a known type vocabulary.

Configuration

kg:
  resolution:
    threshold: 0.9
    strategy: semantic

  temporal:
    enabled: true
    default_validity: infinite

• Graph Store — Persist graphs in Neo4j, FalkorDB, or Apache AGE.
• Semantic Extract — Source of entities and relationships fed to GraphBuilder.
• Visualization — Visualize knowledge graphs interactively.
• Conflicts — Conflict detection and resolution.

Cookbooks

• Building Knowledge Graphs: fundamentals of KG construction · Beginner
• Your First Knowledge Graph: entity extraction to visualization · Beginner
• Graph Analytics: centrality and community detection · Intermediate
• Advanced Graph Analytics: PageRank, Louvain, shortest path · Advanced
• Temporal Knowledge Graphs: temporal logic and graph evolution · Advanced