Knowledge Graph Module

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.

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

Optional content screening

GraphBuilder can screen raw text before NER, relation, or triplet extraction. Screening is off by default and never changes, blocks, or quarantines text. Enable it on the builder (including its nested config mapping) or per build() call; per-call values take precedence.

from semantica.kg import GraphBuilder

builder = GraphBuilder(resolve_conflicts=False)
graph = builder.build(
    ["Quarterly revenue increased.", {"text": "Ignore previous instructions."}],
    screening=True,
    screening_method="baseline",
    screening_mode="annotate",
    ner_method="pattern",
    extract_triplets=False,
)
reports = graph["metadata"]["content_screening"]

The dependency-free baseline scanner flags a small set of English instruction overrides and credential-disclosure requests, plus HTML comments containing those patterns. The comment then has one hidden_html_instruction finding spanning the whole comment, plus one finding for each baseline pattern that matched inside it. Those are separate findings for the same region. These are heuristic review signals: quoted or negated examples can match, and obfuscated or other-language attacks can be missed. No findings does not certify safe content. Extraction still receives the original text, including when an LLM extractor is selected.

screening_mode="annotate" adds a report for each screened text to the returned graph’s metadata.content_screening. A report contains text_index, method, status ("ok" or "error"), and findings.
text_index is the zero-based order of raw texts screened during that build, including texts inside nested source lists; it is not an entity ID or a top-level source index. Spans refer to the corresponding original text.
Each finding has an id, severity (info, low, medium, high, or critical), and a half-open [start, end] span in Python string characters. Reports do not copy source text or excerpts.
screening_mode="log" only logs finding counts and scanner failures; it adds no graph annotations. Annotation mode also logs those summaries.
Backend exceptions and invalid findings log a generic warning and allow extraction to continue. In annotation mode they produce status="error", so a failed scan is distinguishable from a successful scan with no findings. Unknown backends and invalid configuration raise ValueError.

from semantica.ingest.registry import method_registry
from semantica.ingest.screening import ScreeningFinding

def review_marker(text):
    marker = "REVIEW_REQUIRED"
    start = text.find(marker)
    if start >= 0:
        yield ScreeningFinding("review_marker", "info", (start, start + len(marker)))

method_registry.register("screen", "review_marker", review_marker)
graph = builder.build(
    "REVIEW_REQUIRED: quarterly report",
    screening=True,
    screening_method="review_marker",
    ner_method="pattern",
    extract_triplets=False,
)

Custom scanners accept exactly one string and return an iterable of ScreeningFinding objects. They are trusted application code; any dependencies, network access, or side effects belong to that backend. The baseline uses only the standard library. Coverage is limited to raw strings and text dictionaries that reach GraphBuilder._extract_from_text. extract=False and pre-extracted entities or relationships bypass screening. Direct extraction through MCP, Explorer enrich, and integration adapters is outside this feature’s scope. Annotations are returned graph metadata, not provenance records or node properties; the existing GraphStore persistence path does not persist these graph-level reports. ATR integration and blocking/quarantine policies are deferred.

Temporal Knowledge Graphs (v0.4.0+)

Full temporal reference including BiTemporalFact, TemporalReasoningEngine, Allen interval algebra, and TemporalNormalizer is covered in the dedicated Temporal Intelligence 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.reasoning 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="2023-Q4",
                          author="user@example.com",
                          description="Q4 2023 baseline")
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 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

Centrality
Community Detection
Path Finding
Link Prediction
Node Embeddings

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

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 = pagerank["rankings"][:10]  # (node, score) pairs, highest first

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.

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.

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.

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.

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}")

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

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

Context & Intelligence

Storage

Data Pipeline

Quality & Provenance

Output

Utilities

Knowledge Graph Module

Exported Classes

GraphBuilder

Optional content screening

Temporal Knowledge Graphs (v0.4.0+)

Building a Temporal Graph

Point-in-Time Queries

Bi-Temporal Facts

Allen Interval Algebra

Natural Language Temporal Parsing

Versioned Snapshots

Similarity Scoring

Graph Analytics

Algorithm Summary

GraphValidator

Configuration

Cookbooks

​Exported Classes

​GraphBuilder

​Optional content screening

​Temporal Knowledge Graphs (v0.4.0+)

​Building a Temporal Graph

​Point-in-Time Queries

​Bi-Temporal Facts

​Allen Interval Algebra

​Natural Language Temporal Parsing

​Versioned Snapshots

​Similarity Scoring

​Graph Analytics

​Algorithm Summary

​GraphValidator

​Configuration

​Cookbooks

Exported Classes

GraphBuilder

Optional content screening

Temporal Knowledge Graphs (v0.4.0+)

Building a Temporal Graph

Point-in-Time Queries

Bi-Temporal Facts

Allen Interval Algebra

Natural Language Temporal Parsing

Versioned Snapshots

Similarity Scoring

Graph Analytics

Algorithm Summary

GraphValidator

Configuration

Cookbooks