mengzhx/semantica
1
0
Fork 0
mirror of https://github.com/semantica-agi/semantica.git synced 2026-08-29 04:26:20 +00:00
Code Issues Packages Projects Releases Wiki Activity

docs: add Temporal & Distance Intelligence reference pages with accurate API (#650)

Browse Source 
- Add docs/reference/temporal.md: full Temporal Intelligence reference covering
  bi-temporal model (TemporalBound.OPEN sentinel, BiTemporalFact.from_relationship()
  factory), TemporalGraphQuery (query_at_time, reconstruct_at_time, query_time_range,
  find_temporal_paths, analyze_evolution, validate_temporal_consistency),
  TemporalPatternDetector, TemporalReasoningEngine with all 13 Allen interval
  relations over TemporalInterval objects, TemporalNormalizer (returns
  Optional[Tuple[datetime, datetime]]), TemporalQueryRewriter.rewrite() returning
  TemporalQueryResult, and TemporalVersionManager with SQLite storage and correct
  method names (list_versions, compare_versions, get_version, apply_revision,
  validate_snapshot, verify_checksum)

- Add docs/reference/distance.md: Distance Intelligence reference with corrected
  SimilarityCalculator API (pairwise_similarity, batch_similarity, find_most_similar)
  and semantic neighborhood / proximity-blended retrieval patterns

- Update docs/reference/kg.md: expand Exported Classes table to include all
  TemporalPatternDetector, TemporalInterval, IntervalRelation, TemporalQueryResult,
  AlgorithmTrackerWithProvenance, AlgorithmRegistry, ProvenanceTracker, SeedManager,
  KGConfig; fix all temporal code examples to use correct constructors and method names

- Update docs/reference/context.md: add Distance Intelligence section

- Update docs/index.md: add v0.3.0 release accordion with feature highlights

- Update docs/docs.json: wire temporal and distance pages into Modules navigation
This commit is contained in:
Mohd Kaif
2026-06-18 13:37:24 +05:30
committed by GitHub
parent 0765cfea77
commit 12d61b92df
6 changed files with 1798 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/docs.json
+2
View File
@@ -120,6 +120,8 @@
"pages": [
"reference/context",
"reference/kg",
"reference/temporal",
"reference/distance",
"reference/semantic_extract",
"reference/reasoning",
"reference/ontology"
docs/index.md
+34 -6
View File
@@ -15,10 +15,12 @@ If your stack can't answer those questions with a traceable record, you have a g
**Semantica closes that gap.** It's the context and accountability layer that sits beneath your existing agent framework: not a replacement for LangChain or LlamaIndex, but the infrastructure that makes their outputs trustworthy.
<div style={{display:"flex",flexWrap:"wrap",gap:"3rem",margin:"2rem 0",padding:"1.5rem 2rem",borderRadius:"10px",border:"1px solid rgba(16,185,129,0.2)",background:"rgba(16,185,129,0.03)"}}>
<div style={{display:"flex",flexWrap:"wrap",gap:"2rem",margin:"2rem 0",padding:"1.5rem 2rem",borderRadius:"10px",border:"1px solid rgba(16,185,129,0.2)",background:"rgba(16,185,129,0.03)"}}>
<div><div style={{fontSize:"1.75rem",fontWeight:700,color:"#10B981",lineHeight:1.1}}>1,000+</div><div style={{fontSize:"0.8rem",color:"rgba(255,255,255,0.5)",marginTop:"4px"}}>passing tests</div></div>
<div><div style={{fontSize:"1.75rem",fontWeight:700,color:"#10B981",lineHeight:1.1}}>25+</div><div style={{fontSize:"0.8rem",color:"rgba(255,255,255,0.5)",marginTop:"4px"}}>modules</div></div>
<div><div style={{fontSize:"1.75rem",fontWeight:700,color:"#10B981",lineHeight:1.1}}>12</div><div style={{fontSize:"0.8rem",color:"rgba(255,255,255,0.5)",marginTop:"4px"}}>LLM providers</div></div>
<div><div style={{fontSize:"1.75rem",fontWeight:700,color:"#10B981",lineHeight:1.1}}>29</div><div style={{fontSize:"0.8rem",color:"rgba(255,255,255,0.5)",marginTop:"4px"}}>modules</div></div>
<div><div style={{fontSize:"1.75rem",fontWeight:700,color:"#10B981",lineHeight:1.1}}>13</div><div style={{fontSize:"0.8rem",color:"rgba(255,255,255,0.5)",marginTop:"4px"}}>LLM providers</div></div>
<div><div style={{fontSize:"1.75rem",fontWeight:700,color:"#10B981",lineHeight:1.1}}>6,000×</div><div style={{fontSize:"0.8rem",color:"rgba(255,255,255,0.5)",marginTop:"4px"}}>search speedup</div></div>
<div><div style={{fontSize:"1.75rem",fontWeight:700,color:"#10B981",lineHeight:1.1}}>109</div><div style={{fontSize:"0.8rem",color:"rgba(255,255,255,0.5)",marginTop:"4px"}}>REST endpoints</div></div>
<div><div style={{fontSize:"1.75rem",fontWeight:700,color:"#10B981",lineHeight:1.1}}>MIT</div><div style={{fontSize:"0.8rem",color:"rgba(255,255,255,0.5)",marginTop:"4px"}}>open source</div></div>
</div>
@@ -300,11 +302,37 @@ pip install semantica==0.5.0
<Accordion title="v0.4.0: Temporal Intelligence & Knowledge Explorer" icon="clock">
- **Temporal Intelligence** — 6-PR system: temporal data model, point-in-time queries, Allen interval algebra (all 13 relations), OWL-Time export
- **Knowledge Explorer API** — Full FastAPI backend: 99 tests, 12 export formats, WebSocket progress, thread-safe sessions, audit trail
- **Ontology Foundations** — SHACL generation/validation, SKOS vocabulary, ontology alignment API, diff & migration tooling
Released **April 8, 2026**
- **Temporal Intelligence** — Bi-temporal data model (`BiTemporalFact`, `TemporalBound`), point-in-time query engine, Allen interval algebra (all 13 relations), `TemporalNormalizer` (zero LLM calls), `TemporalQueryRewriter`, OWL-Time export
- **Knowledge Explorer API** — Full FastAPI backend: 99 integration tests, 12 export formats, WebSocket progress, thread-safe `RLock` sessions, audit trail, rollback protection
- **Ontology Foundations** — SHACL generation/validation (basic/standard/strict tiers), SKOS vocabulary, ontology alignment API, diff & migration with change impact analysis
- **Datalog Reasoning** — Pure-Python bottom-up semi-naive fixpoint, recursive Horn clause rules, guaranteed termination
- **Agno Integration** — 5 components: graph-backed memory, multi-hop GraphRAG, decision toolkit, KG toolkit, shared team context; 110 tests
- **Novita AI provider** — OpenAI-compatible provider for 200+ open-weight models
```bash
pip install semantica==0.4.0
```
</Accordion>
<Accordion title="v0.3.0: Context Graph Features & Decision Intelligence" icon="brain">
Released **March 10, 2026**
- **Context Graph completeness** — `valid_from`/`valid_until` on every node, `find_active_nodes()` temporal filtering, `get_neighbors(min_weight)` confidence-filtered traversal
- **Cross-graph navigation** — `link_graph()`, `navigate_to()`, `resolve_links()`, `cross_graph_path()` for multi-graph agent architectures
- **Decision tracking system** — Full lifecycle: `record_decision()` → `add_causal_relationship()` → `find_similar_decisions()` → `trace_decision_chain()` → `analyze_decision_impact()` → `check_decision_rules()` → `get_decision_insights()`
- **Advanced KG algorithms** — Node2Vec embeddings, betweenness / closeness / eigenvector centrality, Louvain community detection
- **Policy management** — Versioned policy storage with policy exception tracking and approver audit trail
- **Vector store** — Hybrid search (dense + sparse), PgVector HNSW/IVFFlat indexing with JSONB metadata filtering
- **Apache AGE backend** — Full graph store support for PostgreSQL-native AGE graphs
- **ResourceScheduler fix** — `RLock` deadlock fix for concurrent agent pipelines
```bash
pip install semantica==0.3.0
```
</Accordion>
docs/reference/context.md
+75 -1
View File
@@ -443,7 +443,81 @@ print("Nodes: {}, Edges: {}".format(stats["node_count"], stats["edge_count"]))
| `cross_graph_path(source_node_id, target_graph, target_node_id, max_hops)` | `Dict` | Shortest path across linked graphs |
| `clear()` | `None` | Reset graph state and all indexes |
### Cross-Graph Navigation
### Distance Intelligence (v0.5.0)
`ContextGraph` exposes a full Distance Intelligence API for exploring semantic neighborhoods and blending proximity into retrieval.
<Info>
Full Distance Intelligence reference — distance matrices, API endpoints, embedding cache, Explorer UI — is covered in the dedicated [Distance Intelligence](distance) page. This section documents the context-layer API.
</Info>
### Neighbors with Distance Metadata
Pass `include_distance_metadata=True` to `get_neighbors()` to receive distance band, confidence decay, and path information alongside every neighbor:
```python
graph = ContextGraph(advanced_analytics=True)
# ... populate graph ...
neighbors = graph.get_neighbors(
"python",
hops=3,
include_distance_metadata=True,
min_weight=0.3, # exclude low-confidence edges
)
for n in neighbors:
print(
f"{n['node_id']:15s} "
f"band={n['distance_band']:10s} "
f"decay={n['confidence_decay']:.3f} "
f"hops={n['hop_count']}"
)
```
| Added field | Type | Description |
| :---------- | :---- | :----------- |
| `distance_band` | `str` | `"direct"` (1 hop) / `"near"` (2) / `"mid-range"` (3–4) / `"distant"` (5+) |
| `confidence_decay` | `float` | `edge_weight ^ hop_count` — decays with each hop |
| `path_to_anchor` | `List[str]` | Shortest path from anchor node to this neighbor |
| `hop_count` | `int` | BFS depth from anchor |
### Proximity-Blended Retrieval
Set `proximity_weight` on `AgentContext` to blend graph proximity into every `retrieve()` and `find_precedents()` call:
```python
context = AgentContext(
vector_store=VectorStore(backend="faiss", dimension=768),
knowledge_graph=ContextGraph(advanced_analytics=True),
proximity_weight=0.3, # 0.7×semantic + 0.3×proximity
)
# combined_score is returned alongside semantic_score and proximity_score
results = context.retrieve("web API frameworks", max_results=10)
for r in results:
print(
f"[{r['combined_score']:.3f}] "
f"semantic={r['semantic_score']:.3f} "
f"proximity={r['proximity_score']:.3f} "
f"{r['content'][:60]}"
)
# Override weight per-call
precedents = context.find_precedents(
"infrastructure scaling decisions",
proximity_weight=0.5,
limit=5,
)
```
<Tip>
`proximity_weight=0.0` disables proximity blending entirely (pure semantic). `proximity_weight=1.0` returns results ranked purely by graph proximity to the query anchor. Values between `0.2`–`0.4` work well for most production use cases.
</Tip>
## Cross-Graph Navigation
Link multiple independent `ContextGraph` instances so agents can traverse across problem spaces:
docs/reference/distance.md
+615
View File
@@ -0,0 +1,615 @@
---
title: "Distance Intelligence"
description: "Semantic neighborhoods, N×N distance matrices, ego-mode exploration, proximity-blended retrieval, and embedding cache optimization."
icon: "radar"
---
Distance Intelligence gives every node in your knowledge graph a **semantic neighborhood** — making it possible to answer not just "is A connected to B?" but "how semantically close is A to B, and what lies in between?"
Introduced in **v0.5.0**, Distance Intelligence operates across three layers:
<div style={{display:"flex",flexWrap:"wrap",gap:"1.5rem",margin:"1.5rem 0"}}>
<div style={{flex:"1 1 200px",padding:"1.25rem 1.5rem",borderRadius:"10px",border:"1px solid rgba(16,185,129,0.25)",background:"rgba(16,185,129,0.04)"}}>
<div style={{fontSize:"1.1rem",fontWeight:700,color:"#10B981",marginBottom:"6px"}}>Distance Matrices</div>
<div style={{fontSize:"0.82rem",color:"rgba(255,255,255,0.6)",lineHeight:1.5}}>N×N upper-triangle semantic distance between any node set</div>
</div>
<div style={{flex:"1 1 200px",padding:"1.25rem 1.5rem",borderRadius:"10px",border:"1px solid rgba(16,185,129,0.25)",background:"rgba(16,185,129,0.04)"}}>
<div style={{fontSize:"1.1rem",fontWeight:700,color:"#10B981",marginBottom:"6px"}}>Semantic Neighborhoods</div>
<div style={{fontSize:"0.82rem",color:"rgba(255,255,255,0.6)",lineHeight:1.5}}>BFS ego-graphs with confidence decay and distance band classification</div>
</div>
<div style={{flex:"1 1 200px",padding:"1.25rem 1.5rem",borderRadius:"10px",border:"1px solid rgba(16,185,129,0.25)",background:"rgba(16,185,129,0.04)"}}>
<div style={{fontSize:"1.1rem",fontWeight:700,color:"#10B981",marginBottom:"6px"}}>Proximity Blending</div>
<div style={{fontSize:"0.82rem",color:"rgba(255,255,255,0.6)",lineHeight:1.5}}>Combine semantic similarity with graph proximity in retrieval</div>
</div>
<div style={{flex:"1 1 200px",padding:"1.25rem 1.5rem",borderRadius:"10px",border:"1px solid rgba(16,185,129,0.25)",background:"rgba(16,185,129,0.04)"}}>
<div style={{fontSize:"1.1rem",fontWeight:700,color:"#10B981",marginBottom:"6px"}}>10× Cache</div>
<div style={{fontSize:"0.82rem",color:"rgba(255,255,255,0.6)",lineHeight:1.5}}>Graph revision–based embedding cache avoids redundant re-computation</div>
</div>
</div>
## Distance Bands
Every neighbor result is classified into one of four distance bands based on hop count and semantic similarity:
| Band | Hop count | Meaning | Explorer color |
| :---- | :-------- | :------- | :------------- |
| `direct` | 1 | Immediate neighbor — strong semantic overlap | Green |
| `near` | 2 | One-hop away — closely related concept | Teal |
| `mid-range` | 3–4 | Conceptually related but some separation | Yellow |
| `distant` | 5+ | Weak structural connection | Red |
Distance bands flow through the entire system: retrieval results, path responses, API endpoints, and the Explorer Ego Mode visualization all use the same four-tier classification.
## Quick Start
<Steps>
<Step title="Get neighbors with distance metadata">
The simplest entry point: call `get_neighbors()` with `include_distance_metadata=True`:
```python
from semantica.context import ContextGraph
graph = ContextGraph(advanced_analytics=True)
graph.add_node("python", "language", properties={"paradigm": "multi"})
graph.add_node("fastapi", "framework", properties={"language": "Python"})
graph.add_node("django", "framework", properties={"language": "Python"})
graph.add_node("sqlmodel", "library", properties={"orm": True})
graph.add_edge("python", "fastapi", "enables")
graph.add_edge("python", "django", "enables")
graph.add_edge("fastapi", "sqlmodel", "uses")
neighbors = graph.get_neighbors(
"python",
hops=3,
include_distance_metadata=True,
)
for n in neighbors:
print(f"{n['node_id']:12s} band={n['distance_band']:10s} "
f"decay={n['confidence_decay']:.3f} "
f"path={n['path_to_anchor']}")
```
```
fastapi band=direct decay=1.000 path=['python', 'fastapi']
django band=direct decay=1.000 path=['python', 'django']
sqlmodel band=near decay=0.750 path=['python', 'fastapi', 'sqlmodel']
```
</Step>
<Step title="Compute a semantic distance matrix">
```python
from semantica.kg import SimilarityCalculator, NodeEmbedder
# Generate structural embeddings first
embedder = NodeEmbedder(method="node2vec", embedding_dimension=128)
embeddings = embedder.compute_embeddings(kg, ["language", "framework", "library"], ["enables", "uses"])
# N×N upper-triangle distance matrix
calc = SimilarityCalculator()
matrix = calc.compute_distance_matrix(embeddings)
# matrix["distances"] is an upper-triangle dict: {(node_a, node_b): distance}
for (a, b), dist in sorted(matrix["distances"].items(), key=lambda x: x[1]):
print(f"{a:15s} ↔ {b:15s} distance={dist:.4f}")
```
</Step>
<Step title="Blend proximity into retrieval">
Set `proximity_weight` on `AgentContext` to blend graph proximity into every semantic retrieval call:
```python
from semantica.context import AgentContext, ContextGraph
from semantica.vector_store import VectorStore
context = AgentContext(
vector_store=VectorStore(backend="faiss", dimension=768),
knowledge_graph=ContextGraph(advanced_analytics=True),
decision_tracking=True,
proximity_weight=0.3, # combined = 0.7×semantic + 0.3×proximity
)
# retrieve() and find_precedents() both use the blended score
results = context.retrieve("web API frameworks", max_results=10)
for r in results:
print(f"[{r['combined_score']:.3f}] semantic={r['semantic_score']:.3f} "
f"proximity={r['proximity_score']:.3f} {r['content'][:60]}")
```
</Step>
</Steps>
## ContextGraph Distance API
### `get_neighbors()`
Returns BFS neighbors enriched with distance metadata when `include_distance_metadata=True`:
```python
neighbors = graph.get_neighbors(
node_id="python",
hops=4,
include_distance_metadata=True,
min_weight=0.3, # exclude low-confidence edges
)
```
| Field | Type | Description |
| :---- | :---- | :----------- |
| `node_id` | `str` | Node identifier |
| `node_type` | `str` | Node type label |
| `properties` | `Dict` | Node property dict |
| `hop_count` | `int` | BFS hops from anchor |
| `distance_band` | `str` | `"direct"` / `"near"` / `"mid-range"` / `"distant"` |
| `confidence_decay` | `float` | Confidence score after hop-based decay: `weight^hop_count` |
| `path_to_anchor` | `List[str]` | Shortest path from anchor to this node |
| `edge_weight` | `float` | Weight of the direct edge (if hop=1) |
### `get_neighbor_distances()`
Returns a sorted list of neighbors ranked by combined confidence-decay distance score:
```python
distances = graph.get_neighbor_distances("fastapi", hops=3)
for d in distances:
print(f"{d['node_id']:15s} score={d['combined_distance_score']:.4f} "
f"band={d['distance_band']}")
```
## SimilarityCalculator — Pairwise Similarity
`SimilarityCalculator` computes similarity between node embeddings using four metrics.
```python
from semantica.kg import SimilarityCalculator
calc = SimilarityCalculator(method="cosine", normalize=True)
# method: "cosine" | "euclidean" | "manhattan" | "correlation"
```
### Constructor
| Parameter | Type | Default | Description |
| :--------- | :---- | :------- | :----------- |
| `method` | `str` | `"cosine"` | Default metric: `"cosine"`, `"euclidean"`, `"manhattan"`, `"correlation"` |
| `normalize` | `bool` | `True` | Normalize vectors before calculation |
### Methods
| Method | Returns | Description |
| :------ | :------- | :----------- |
| `cosine_similarity(vector1, vector2)` | `float` | Cosine similarity `[-1, 1]` between two vectors |
| `euclidean_distance(embedding1, embedding2)` | `float` | L2 distance (non-negative) between two vectors |
| `manhattan_distance(embedding1, embedding2)` | `float` | L1 distance (non-negative) between two vectors |
| `correlation_similarity(embedding1, embedding2)` | `float` | Pearson correlation `[-1, 1]` between two vectors |
| `batch_similarity(embeddings, query_embedding, method=None, top_k=None, chunk_size=1000)` | `Dict[str, float]` | Similarity of all nodes against a query vector. Returns `{node_id: score}` |
| `pairwise_similarity(embeddings, method=None)` | `Dict[Tuple[str,str], float]` | Upper-triangle N×N pairwise similarity matrix for all node pairs |
| `find_most_similar(embeddings, query_embedding, top_k=10, method=None)` | `List[Tuple[str, float]]` | Top-k `(node_id, score)` pairs sorted by similarity |
### Pairwise Similarity Matrix
`pairwise_similarity()` returns the upper triangle of the N×N matrix — each key is a `(node_id_a, node_id_b)` tuple:
```python
from semantica.kg import NodeEmbedder, SimilarityCalculator
embedder = NodeEmbedder(method="node2vec", embedding_dimension=128)
embeddings = embedder.compute_embeddings(kg, ["language", "framework"], ["enables", "uses"])
calc = SimilarityCalculator(method="cosine")
# N×N upper-triangle: Dict[(node_a, node_b), similarity_score]
matrix = calc.pairwise_similarity(embeddings)
# Sort by similarity (most similar first)
for (a, b), score in sorted(matrix.items(), key=lambda x: x[1], reverse=True)[:5]:
print(f"{a:15s} ↔ {b:15s} similarity={score:.4f}")
# Find most similar pair
best_pair = max(matrix.items(), key=lambda x: x[1])
print(f"Most similar: {best_pair[0]} score={best_pair[1]:.4f}")
# Find most dissimilar pair
worst_pair = min(matrix.items(), key=lambda x: x[1])
print(f"Most distant: {worst_pair[0]} score={worst_pair[1]:.4f}")
```
<Note>
The matrix is upper-triangle only — `(a, b)` is stored but `(b, a)` is not. To look up either direction: `matrix.get((a, b)) or matrix.get((b, a))`.
</Note>
### Batch Similarity
Efficiently compare a query vector against all nodes using chunked vectorized ops:
```python
# Query vector against all nodes
scores = calc.batch_similarity(
embeddings,
query_embedding=my_query_vec,
method="cosine", # override default
top_k=10, # return only top 10 (None = all)
chunk_size=1000, # chunk size for memory efficiency
)
for node_id, score in sorted(scores.items(), key=lambda x: x[1], reverse=True):
print(f"{node_id:15s} {score:.4f}")
```
### Find Most Similar
```python
# Top-k (node_id, score) tuples sorted descending
similar = calc.find_most_similar(
embeddings,
query_embedding=embeddings["python"],
top_k=5,
method="cosine",
)
for node_id, score in similar:
print(f"{node_id:15s} similarity={score:.4f}")
```
### Individual Metrics
```python
vec_a = embeddings["fastapi"]
vec_b = embeddings["django"]
cosine = calc.cosine_similarity(vec_a, vec_b)
l2 = calc.euclidean_distance(vec_a, vec_b)
l1 = calc.manhattan_distance(vec_a, vec_b)
pearson = calc.correlation_similarity(vec_a, vec_b)
print(f"Cosine: {cosine:.4f}")
print(f"Euclidean: {l2:.4f}")
print(f"Manhattan: {l1:.4f}")
print(f"Correlation: {pearson:.4f}")
```
## Proximity-Blended Retrieval
`AgentContext.retrieve()` and `find_precedents()` both support a `proximity_weight` parameter that blends graph proximity into the semantic similarity score:
```
combined_score = (1 − proximity_weight) × semantic_score
+ proximity_weight × proximity_score
```
Where `proximity_score` is derived from hop count and edge weights from the query anchor node.
```python
from semantica.context import AgentContext, ContextGraph
from semantica.vector_store import VectorStore
context = AgentContext(
vector_store=VectorStore(backend="faiss", dimension=768),
knowledge_graph=ContextGraph(advanced_analytics=True),
proximity_weight=0.3,
)
# Standard retrieval — proximity blended automatically
results = context.retrieve("model deployment strategies", max_results=10)
# Override weight per-call
results = context.retrieve(
"model deployment strategies",
max_results=10,
proximity_weight=0.5, # stronger proximity weight for this query
)
# find_precedents also blends proximity
precedents = context.find_precedents(
"infrastructure scaling decisions",
proximity_weight=0.4,
limit=5,
)
for p in precedents:
print(f"[{p.combined_score:.3f}] {p.outcome} (confidence: {p.confidence:.2f})")
```
## Embedding Cache
The embedding cache avoids re-computing embeddings for nodes that haven't changed since the last call — delivering up to **10× throughput improvement** on large graphs.
### How It Works
Each `GraphSession` tracks a **graph revision hash** derived from the current node and edge state. When a distance matrix or neighborhood request arrives:
1. The revision hash is compared to the cached hash
2. If unchanged: the cached embeddings are returned directly
3. If changed (nodes/edges added or modified): the cache is invalidated and embeddings are recomputed
```python
from semantica.explorer import GraphSession
session = GraphSession(graph=kg)
# First call: computes embeddings, stores in cache
embeddings = session.get_cached_embeddings()
# Second call (graph unchanged): returns cache instantly
embeddings = session.get_cached_embeddings()
# After graph modification: cache is automatically invalidated
session.graph.add_node("new_node", "concept", properties={})
embeddings = session.get_cached_embeddings() # recomputes
```
| Parameter | Type | Default | Description |
| :--------- | :---- | :------- | :----------- |
| `force_refresh` | `bool` | `False` | Force cache invalidation even if the graph is unchanged |
| Cache invalidation | Automatic | — | Triggered by `add_nodes()`, `add_edges()`, or any mutation |
| Cache scope | Per-session | — | Each `GraphSession` maintains its own independent cache |
<Tip>
The cache is most effective in Explorer deployments where the same graph is queried repeatedly for distance matrices and ego-mode neighborhoods. In batch pipeline contexts, set `force_refresh=True` to ensure the latest graph state is always used.
</Tip>
## REST API Endpoints
Five new endpoints were added in v0.5.0 for programmatic distance intelligence access:
### `POST /api/graph/distance-matrix`
Compute N×N semantic distance matrix for a set of node IDs:
```bash
curl -X POST http://localhost:8000/api/graph/distance-matrix \
-H "Content-Type: application/json" \
-d '{
"node_ids": ["alice", "bob", "acme_corp", "beta_ltd"],
"embedding_model": "all-MiniLM-L6-v2",
"include_band_classification": true
}'
```
```json
{
"matrix": {
"alice,bob": 0.312,
"alice,acme_corp": 0.087,
"alice,beta_ltd": 0.154,
"bob,acme_corp": 0.401,
"bob,beta_ltd": 0.233,
"acme_corp,beta_ltd": 0.198
},
"most_similar": ["alice", "acme_corp"],
"most_distant": ["bob", "acme_corp"],
"mean_distance": 0.231
}
```
### `GET /api/graph/node/{id}/semantic-neighborhood`
Retrieve the ego-graph (BFS neighborhood) of a node with distance metadata:
```bash
curl "http://localhost:8000/api/graph/node/alice/semantic-neighborhood?depth=3&include_distance_metadata=true"
```
```json
{
"anchor_node": "alice",
"neighbors": [
{"node_id": "acme_corp", "distance_band": "direct", "confidence_decay": 1.0, "hop_count": 1},
{"node_id": "ceo_role", "distance_band": "direct", "confidence_decay": 1.0, "hop_count": 1},
{"node_id": "beta_ltd", "distance_band": "near", "confidence_decay": 0.75, "hop_count": 2},
{"node_id": "london_hq", "distance_band": "mid-range","confidence_decay": 0.56, "hop_count": 3}
],
"total_neighbors": 4,
"depth": 3
}
```
### `GET /api/decisions/causal-distance`
Return causal distance (hop count through causal edges) between two decision nodes:
```bash
curl "http://localhost:8000/api/decisions/causal-distance?source=dec_001&target=dec_005"
```
```json
{
"source": "dec_001",
"target": "dec_005",
"causal_hops": 3,
"causal_path": ["dec_001", "dec_002", "dec_004", "dec_005"],
"distance_band": "near"
}
```
### `GET /api/temporal/distance-history`
Track how the semantic distance between two nodes has evolved over time:
```bash
curl "http://localhost:8000/api/temporal/distance-history?node_a=alice&node_b=acme_corp&snapshots=2021-01-01,2022-01-01,2023-01-01"
```
```json
{
"node_a": "alice",
"node_b": "acme_corp",
"history": [
{"timestamp": "2021-01-01", "distance": 0.08, "band": "direct"},
{"timestamp": "2022-01-01", "distance": 0.09, "band": "direct"},
{"timestamp": "2023-01-01", "distance": 0.54, "band": "mid-range"}
]
}
```
### `POST /api/export/distance-enriched`
Export graph data enriched with distance metadata (CSV or JSONL, capped at 200 nodes):
```bash
curl -X POST http://localhost:8000/api/export/distance-enriched \
-H "Content-Type: application/json" \
-d '{"anchor_node": "alice", "depth": 4, "format": "csv"}'
```
## Explorer Distance Intelligence UI
The Knowledge Explorer embeds Distance Intelligence directly in the browser dashboard:
<AccordionGroup>
<Accordion title="Ego Mode" icon="circle-nodes">
Ego Mode centers the visualization on a selected node and renders its semantic neighborhood with **BFS depth-of-field fading** — nodes further from the anchor become progressively dimmer, revealing the "shape" of conceptual proximity.
- **Depth slider (1–8)**: controls the BFS radius of the neighborhood
- **Confidence decay visualization**: edge opacity maps to `confidence_decay` score
- **Distance band color coding**: green (direct) → teal (near) → yellow (mid-range) → red (distant)
- **Bottleneck highlighting**: bridge nodes that connect otherwise separate clusters are highlighted in the path inspector
Activate via the Explorer toolbar: **View → Ego Mode**, then click any node to set it as anchor.
</Accordion>
<Accordion title="Distance Heatmap" icon="table-cells">
The heatmap renders an N×N distance matrix as a color-coded grid — instantly revealing which clusters of nodes are semantically cohesive and which are isolated.
- **Color scale**: green (near, distance → 0) through yellow to red (distant, distance → 1)
- **Hover**: shows exact distance value and distance band for each cell
- **Sort options**: sort rows/columns by node type, community membership, or alphabetical
Access via **View → Distance Heatmap** in the Explorer sidebar.
</Accordion>
<Accordion title="Semantic Overlay" icon="layer-group">
Overlay semantic similarity on the standard force-directed graph layout without switching modes:
- **Semantic overlay**: edge thickness scaled by semantic similarity score
- **Structural overlay**: edge thickness scaled by graph centrality
- Both overlays can be toggled independently
Access via the **Overlay** toggle in the Explorer toolbar.
</Accordion>
<Accordion title="Path Inspector" icon="route">
Click any two nodes to inspect the shortest path between them. The Path Inspector shows:
- **Distance band chip**: classifies the overall path as direct / near / mid-range / distant
- **Metric cards**: hop count, mean edge weight, path confidence decay
- **Bottleneck node highlight**: the single node whose removal would disconnect the path
- **Distance history**: timeline of how the distance between the two nodes has changed across graph snapshots
Access via **right-click → Inspect Path** on any two selected nodes.
</Accordion>
</AccordionGroup>
## Real-World Patterns
<Tabs>
<Tab title="Knowledge Cluster Discovery">
Find semantically cohesive topic clusters in a large knowledge graph without running community detection:
```python
from semantica.kg import NodeEmbedder, SimilarityCalculator
embedder = NodeEmbedder(method="node2vec", embedding_dimension=128)
embeddings = embedder.compute_embeddings(kg, node_types=["Concept", "Topic"])
calc = SimilarityCalculator()
# Cluster nodes where pairwise distance < 0.2
clusters = calc.cluster_by_distance(embeddings, threshold=0.2)
for i, cluster in enumerate(clusters):
print(f"Cluster {i+1} ({len(cluster)} nodes): {cluster[:5]}")
```
</Tab>
<Tab title="Anomaly Detection">
Flag nodes that are unexpectedly distant from their structural neighbors — potential data quality issues or genuine anomalies:
```python
from semantica.context import ContextGraph
from semantica.kg import NodeEmbedder, SimilarityCalculator
graph = ContextGraph(advanced_analytics=True)
# ... build graph ...
embedder = NodeEmbedder(method="node2vec", embedding_dimension=128)
embeddings = embedder.compute_embeddings(graph._graph, ["entity"], ["RELATED_TO"])
calc = SimilarityCalculator()
for node_id in graph._graph.nodes():
neighbors = graph.get_neighbors(node_id, hops=1, include_distance_metadata=True)
for n in neighbors:
# Node connected by edge but semantically very distant → anomaly candidate
structural_dist = 1.0 - n["edge_weight"]
semantic_dist = calc.euclidean_distance(
embeddings[node_id], embeddings[n["node_id"]]
)
if semantic_dist > 0.7 and structural_dist < 0.3:
print(f"Anomaly: {node_id} → {n['node_id']} "
f"(structural={structural_dist:.2f}, semantic={semantic_dist:.2f})")
```
</Tab>
<Tab title="Decision Consistency Audit">
Verify that similar decisions (low semantic distance) reached similar outcomes — flag inconsistencies for review:
```python
from semantica.context import AgentContext, ContextGraph
from semantica.vector_store import VectorStore
context = AgentContext(
vector_store=VectorStore(backend="faiss", dimension=768),
knowledge_graph=ContextGraph(advanced_analytics=True),
decision_tracking=True,
proximity_weight=0.4,
)
# ... populate with historical decisions ...
# Find pairs of semantically close decisions with different outcomes
all_decisions = context.query_decisions("", max_hops=0)
for i, d1 in enumerate(all_decisions):
for d2 in all_decisions[i+1:]:
precedents = context.find_precedents(
d1.scenario, limit=5, proximity_weight=0.4
)
for p in precedents:
if p.source_decision_id == d2.decision_id:
if p.similarity_score > 0.85 and d1.outcome != d2.outcome:
print(f"INCONSISTENCY: {d1.scenario}")
print(f" Decision A: {d1.outcome} (confidence {d1.confidence:.2f})")
print(f" Decision B: {d2.outcome} (confidence {d2.confidence:.2f})")
print(f" Similarity: {p.similarity_score:.3f}")
```
</Tab>
</Tabs>
## Performance
| Operation | Without cache | With cache | Improvement |
| :--------- | :------------ | :--------- | :---------- |
| Distance matrix (118k nodes) | ~48s | ~4.8s | **10×** |
| Semantic neighborhood (depth 4) | ~2.1s | ~0.21s | **10×** |
| Node search (indexed) | 24 ms | 0.004 ms | **6,000×** |
| Semantic deduplication | baseline | — | **6.98×** (v2 algorithms) |
<Note>
The 10× cache improvement applies when the graph is unchanged between requests. In write-heavy pipelines where nodes are added continuously, cache hit rates will be lower. Use `force_refresh=False` (default) for read-heavy Explorer usage and `force_refresh=True` for batch pipeline contexts.
</Note>
- [Context Module](context) — `ContextGraph.get_neighbors()` and proximity-blended retrieval.
- [Knowledge Graph Module](kg) — `NodeEmbedder`, `SimilarityCalculator`, and graph analytics.
- [Visualization](visualization) — Programmatic distance heatmaps and ego-mode graph renders.
- [Explorer](explorer) — Knowledge Explorer with built-in Distance Intelligence dashboard.
- [Distance Intelligence](https://github.com/semantica-agi/semantica/blob/main/cookbook/advanced/12_Distance_Intelligence.ipynb) — Semantic neighborhoods and distance matrices · Advanced
docs/reference/kg.md
+189 -30
View File
@@ -23,7 +23,17 @@ icon: "diagram-project"
| `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, temporal diffs, and all 13 Allen interval queries |
| `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 |
@@ -31,6 +41,11 @@ icon: "diagram-project"
| `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 |
<Tip>
@@ -57,56 +72,200 @@ kg = builder.build({"entities": entities, "relationships": relationships})
| `build_single_source(data)` | `dict` | Build graph from a single data source dict |
## Temporal Knowledge Graphs (v0.4.0)
## Temporal Knowledge Graphs (v0.4.0+)
Use **`TemporalGraphQuery`** to attach `valid_from`/`valid_until` windows and query **point-in-time snapshots** of any graph:
<Info>
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.
</Info>
The temporal stack — see the [Temporal Intelligence](temporal) page for the full reference.
### Building a Temporal Graph
```python
from semantica.kg import GraphBuilder, TemporalGraphQuery, TemporalVersionManager
from datetime import datetime
# Build a time-aware graph
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": "2020-01-01",
"valid_until": "2023-06-01",
}
]
"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 snapshot: TemporalGraphQuery takes no positional graph arg;
# pass the graph into each query method instead.
query = TemporalGraphQuery()
snapshot_2021 = query.reconstruct_at_time(kg, "2021-06-15")
snapshot_2023 = query.reconstruct_at_time(kg, "2023-01-01")
# Relationships active within a date range
range_result = query.query_time_range(kg, "", "2020-01-01", "2023-01-01")
print(f"Relationships in range: {range_result['num_relationships']}")
# Versioned snapshots: author and description are required
versioner = TemporalVersionManager()
versioner.create_snapshot(kg, version_label="2024-Q1",
author="user@example.com",
description="Q1 2024 snapshot")
```
Supports all 13 Allen interval algebra relations:
### Point-in-Time Queries
- before, after, meets, met_by
- overlaps, overlapped_by
- during, contains, starts, started_by, finishes, finished_by, equals
`TemporalGraphQuery` accepts optional constructor args; pass the graph into each query call:
OWL-Time export available.
```python
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:
```python
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):
```python
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
```python
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
```python
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)
```
<Tip>
See the [Temporal Intelligence](temporal) reference for the full class API, domain examples (personnel changes, policy evolution, financial timelines), and configuration options.
</Tip>
## Similarity Scoring
docs/reference/temporal.md
+883
View File
@@ -0,0 +1,883 @@
---
title: "Temporal Intelligence"
description: "Bi-temporal facts, point-in-time snapshots, Allen interval algebra, temporal pattern detection, and natural-language temporal parsing for time-aware knowledge graphs."
icon: "clock"
---
Temporal Intelligence gives your knowledge graph a complete understanding of *when* — not just what is true, but when it was true in the real world, when it was recorded, and how facts have evolved over time.
Shipped across **v0.3.0** (context temporal validity) and **v0.4.0** (full temporal stack), the system covers five layers:
<div style={{display:"flex",flexWrap:"wrap",gap:"1.5rem",margin:"1.5rem 0"}}>
<div style={{flex:"1 1 180px",padding:"1.25rem 1.5rem",borderRadius:"10px",border:"1px solid rgba(16,185,129,0.25)",background:"rgba(16,185,129,0.04)"}}>
<div style={{fontSize:"1.1rem",fontWeight:700,color:"#10B981",marginBottom:"6px"}}>Bi-temporal model</div>
<div style={{fontSize:"0.82rem",color:"rgba(255,255,255,0.6)",lineHeight:1.5}}>Valid time + transaction time on every fact</div>
</div>
<div style={{flex:"1 1 180px",padding:"1.25rem 1.5rem",borderRadius:"10px",border:"1px solid rgba(16,185,129,0.25)",background:"rgba(16,185,129,0.04)"}}>
<div style={{fontSize:"1.1rem",fontWeight:700,color:"#10B981",marginBottom:"6px"}}>Point-in-time queries</div>
<div style={{fontSize:"0.82rem",color:"rgba(255,255,255,0.6)",lineHeight:1.5}}>Reconstruct any historical graph state in one call</div>
</div>
<div style={{flex:"1 1 180px",padding:"1.25rem 1.5rem",borderRadius:"10px",border:"1px solid rgba(16,185,129,0.25)",background:"rgba(16,185,129,0.04)"}}>
<div style={{fontSize:"1.1rem",fontWeight:700,color:"#10B981",marginBottom:"6px"}}>Allen interval algebra</div>
<div style={{fontSize:"0.82rem",color:"rgba(255,255,255,0.6)",lineHeight:1.5}}>All 13 temporal relations, deterministic reasoning</div>
</div>
<div style={{flex:"1 1 180px",padding:"1.25rem 1.5rem",borderRadius:"10px",border:"1px solid rgba(16,185,129,0.25)",background:"rgba(16,185,129,0.04)"}}>
<div style={{fontSize:"1.1rem",fontWeight:700,color:"#10B981",marginBottom:"6px"}}>NL temporal parsing</div>
<div style={{fontSize:"0.82rem",color:"rgba(255,255,255,0.6)",lineHeight:1.5}}>Zero LLM calls — pure regex + dateutil</div>
</div>
</div>
## Exported Classes
| Class | Role |
| :---- | :---- |
| `BiTemporalFact` | Dataclass wrapping `valid_from`, `valid_until`, `recorded_at`, `superseded_at`. Factory: `BiTemporalFact.from_relationship(rel_dict)` |
| `TemporalBound` | Enum sentinel for open-ended intervals. Single value: `TemporalBound.OPEN` |
| `TemporalInterval` | Frozen dataclass `(start: datetime, end: datetime \| TemporalBound, label?)` used by `TemporalReasoningEngine` |
| `IntervalRelation` | Enum of all 13 Allen relation labels (`BEFORE`, `AFTER`, `MEETS`, etc.) |
| `TemporalGraphQuery` | Point-in-time snapshots, range queries, pattern detection, evolution analysis, temporal path finding |
| `TemporalPatternDetector` | Sequence and cycle pattern detection over temporal edges |
| `TemporalReasoningEngine` | Allen interval algebra over `TemporalInterval` objects — pure Python, deterministic |
| `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` | Create, list, compare, and apply revisions to versioned graph snapshots |
## Quick Start
<Steps>
<Step title="Build a time-aware graph">
Attach `valid_from` / `valid_until` to any relationship at construction time:
```python
from semantica.kg import GraphBuilder
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)
},
],
}])
```
</Step>
<Step title="Query the graph at a point in time">
`TemporalGraphQuery` takes constructor args; pass the graph into each query call:
```python
from semantica.kg import TemporalGraphQuery
query = TemporalGraphQuery(temporal_granularity="day")
# query_at_time is the primary public API
result_2020 = query.query_at_time(kg, query="", at_time="2020-06-15")
result_2023 = query.query_at_time(kg, query="", at_time="2023-01-01")
print(f"Rels active in 2020: {result_2020['num_relationships']}")
print(f"Rels active in 2023: {result_2023['num_relationships']}")
```
</Step>
<Step title="Reconstruct a subgraph at a specific timestamp">
`reconstruct_at_time()` is the low-level primitive — returns a full graph dict
with only nodes and edges that were valid at the given moment:
```python
snapshot = query.reconstruct_at_time(kg, "2021-06-15")
# snapshot has "entities" and "relationships" keys
# usable with all GraphAnalyzer, PathFinder, CommunityDetector calls
```
</Step>
<Step title="Create versioned snapshots">
```python
from semantica.kg import TemporalVersionManager
versioner = TemporalVersionManager() # in-memory storage
# versioner = TemporalVersionManager(storage_path="versions.db") # SQLite
versioner.create_snapshot(
kg,
version_label="2024-Q1",
author="user@example.com",
description="Q1 2024 snapshot after board restructure",
)
for v in versioner.list_versions():
print(f"{v['label']:12s} {v['author']} {v['timestamp']}")
```
</Step>
</Steps>
## The Bi-Temporal Model
Most systems track only one timeline: when something is currently true. Bi-temporal graphs track **two independent timelines** simultaneously:
<Tabs>
<Tab title="Valid Time">
*When was the fact true in the real world?*
- `valid_from` — date the fact became true
- `valid_until` — date the fact ceased to be true. Omit (or use `TemporalBound.OPEN`) for currently-active facts
```python
from semantica.kg import BiTemporalFact, TemporalBound
# Create from an existing relationship dict
rel = {
"source": "alice", "target": "acme_corp", "type": "ceo_of",
"valid_from": "2018-01-01",
"valid_until": "2022-06-01",
}
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)
# Serialize back to dict fields
fields = fact.to_relationship_fields()
print(fields["valid_from"]) # "2018-01-01T00:00:00Z"
print(fields["valid_until"]) # "2022-06-01T00:00:00Z"
```
</Tab>
<Tab title="Transaction Time">
*When did we record this fact in the system?*
- `recorded_at` — auto-stamped at ingestion time (defaults to `datetime.now(utc)`)
- `superseded_at` — set when a later version replaces this record. `TemporalBound.OPEN` means still current
```python
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, # still the current record
}
fact = BiTemporalFact.from_relationship(rel)
print(fact.recorded_at) # datetime(2018, 1, 5, 9, 32, tzinfo=utc)
print(fact.superseded_at) # TemporalBound.OPEN
```
</Tab>
<Tab title="TemporalBound.OPEN">
`TemporalBound.OPEN` is the single sentinel that represents an open-ended interval — a fact with no defined end date:
```python
from semantica.kg import TemporalBound
print(TemporalBound.OPEN) # TemporalBound.OPEN
print(TemporalBound.OPEN.value) # "OPEN"
# A relationship with no valid_until gets TemporalBound.OPEN automatically
rel = {"source": "alice", "target": "beta_ltd", "type": "ceo_of",
"valid_from": "2022-06-01"}
fact = BiTemporalFact.from_relationship(rel)
print(fact.valid_until) # TemporalBound.OPEN
```
<Note>
`TemporalBound.OPEN` replaces both the start and end sentinels — there is only one value. The reasoning engine treats `OPEN` as `datetime.max` (far future) when comparing end bounds, and as `datetime.min` (far past) when used for `superseded_at`.
</Note>
</Tab>
</Tabs>
## TemporalGraphQuery — Reference
Constructed once; the graph is passed into each method call:
```python
from semantica.kg import TemporalGraphQuery
query = TemporalGraphQuery(
enable_temporal_reasoning=True, # default
temporal_granularity="day", # second|minute|hour|day|week|month|year
max_temporal_depth=None, # optional max depth
)
```
### Core Methods
| Method | Returns | Description |
| :------ | :------- | :----------- |
| `query_at_time(graph, query, at_time, include_history=False, time_axis="valid")` | `Dict` | Primary API — filter graph to facts valid at `at_time`. Returns `entities`, `relationships`, `num_entities`, `num_relationships` |
| `reconstruct_at_time(graph, at_time, *, time_axis="valid")` | `Dict` | Low-level — returns a deep-copied subgraph valid at `at_time`. Usable with all analytics tools |
| `query_time_range(graph, query, start_time, end_time, temporal_aggregation="union", include_intervals=True, time_axis="valid")` | `Dict` | All relationships active during `[start, end]`. `temporal_aggregation`: `"union"` / `"intersection"` / `"evolution"` |
| `validate_temporal_consistency(graph)` | `TemporalConsistencyReport` | Detect inverted intervals, overlapping same-edge facts, and entity lifetime violations |
| `query_temporal_pattern(graph, pattern, time_window=None, min_support=1)` | `Dict` | Detect `"sequence"` or `"cycle"` patterns. Delegates to `TemporalPatternDetector` |
| `analyze_evolution(graph, entity=None, relationship=None, start_time=None, end_time=None, metrics=None)` | `Dict` | Track evolution metrics (`"count"`, `"diversity"`, `"stability"`) over time |
| `find_temporal_paths(graph, source, target, start_time=None, end_time=None, max_path_length=None, enforce_causal_ordering=True, ordering_strategy="strict")` | `Dict` | BFS paths respecting temporal validity. `ordering_strategy`: `"strict"` / `"overlap"` / `"loose"` |
### `time_axis` Parameter
All query methods accept a `time_axis` parameter controlling which timestamps are used for filtering:
| Value | Effect |
| :---- | :----- |
| `"valid"` (default) | Filter by `valid_from` / `valid_until` — when the fact was true |
| `"transaction"` | Filter by `recorded_at` / `superseded_at` — when we recorded it |
| `"both"` | Fact must be active on both axes simultaneously |
### Range Query Example
```python
# All relationships active at any point in 2021
result = query.query_time_range(kg, "", "2021-01-01", "2021-12-31")
for rel in result["relationships"]:
print(f" {rel['source']} --[{rel['type']}]--> {rel['target']}")
# Only relationships valid throughout the entire range (stricter)
result = query.query_time_range(
kg, "", "2021-01-01", "2021-12-31",
temporal_aggregation="intersection",
)
# Grouped by calendar period
result = query.query_time_range(
kg, "", "2021-01-01", "2021-12-31",
temporal_aggregation="evolution",
)
for period, rels in result["relationship_buckets"].items():
print(f" {period}: {len(rels)} relationships active")
```
### Evolution Analysis
```python
evolution = query.analyze_evolution(
kg,
entity="alice", # track a specific entity (None = whole graph)
relationship="ceo_of", # track a specific edge type (None = all)
start_time="2018-01-01",
end_time="2024-12-31",
metrics=["count", "diversity", "stability"],
)
print(f"Relationship count: {evolution['count']}")
print(f"Relationship types: {evolution['diversity']}")
```
### Temporal Path Finding
```python
paths = query.find_temporal_paths(
kg,
source="alice",
target="beta_ltd",
start_time="2022-01-01",
end_time="2024-12-31",
max_path_length=5,
enforce_causal_ordering=True,
ordering_strategy="strict", # strict|overlap|loose
)
for p in paths["paths"]:
print(f" {' → '.join(p['path'])} (length={p['length']})")
```
### Consistency Validation
```python
from semantica.kg import TemporalGraphQuery
report = TemporalGraphQuery().validate_temporal_consistency(kg)
print(f"Errors: {len(report.errors)}")
print(f"Warnings: {len(report.warnings)}")
for err in report.errors:
print(f" [{err['issue_type']}] fact_id={err['fact_id']}: {err['message']}")
```
Error types reported: `inverted_interval`, `invalid_temporal_fields`, `missing_source_entity`, `missing_target_entity`, `source_lifetime_mismatch`, `target_lifetime_mismatch`.
Warning types: `overlapping_same_edge`, `gap_after_restart`.
## TemporalPatternDetector
Detect recurring temporal patterns across graph edges. Accessed directly or via `TemporalGraphQuery.query_temporal_pattern()`:
```python
from semantica.kg import TemporalPatternDetector
detector = TemporalPatternDetector()
# Find sequential edge patterns (A→B→C where edges are back-to-back)
sequences = detector.detect_temporal_patterns(
kg,
pattern_type="sequence",
min_frequency=2,
time_window=None,
)
for seq in sequences:
print(f"Sequence: {seq['signature']} (occurs {seq['frequency']} times)")
for occ in seq["occurrences"]:
print(f" nodes={occ['nodes']} {occ['start_time']} → {occ['end_time']}")
# Find cyclic patterns (A→B→C→A)
cycles = detector.detect_temporal_patterns(
kg,
pattern_type="cycle",
min_frequency=1,
)
```
| Parameter | Type | Default | Description |
| :--------- | :---- | :------- | :----------- |
| `pattern_type` | `str` | `"sequence"` | `"sequence"` or `"cycle"` |
| `min_frequency` | `int` | `2` | Minimum occurrences for a pattern to be returned |
| `time_window` | `Any` | `None` | Optional time constraint on pattern window |
Each pattern dict has: `pattern_type`, `signature` (tuple of node IDs), `frequency`, `occurrences` (list with `nodes`, `edges`, `start_time`, `end_time`).
## Allen Interval Algebra
`TemporalReasoningEngine` operates on `TemporalInterval` objects — a frozen dataclass with `start: datetime` and `end: datetime | TemporalBound`:
```python
from semantica.kg import (
TemporalReasoningEngine, TemporalInterval, IntervalRelation, TemporalBound
)
from datetime import datetime, timezone
def dt(year, month, day):
return datetime(year, month, day, 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)
print(relation) # IntervalRelation.OVERLAPS
print(relation.value) # "overlaps"
print(engine.overlaps(h1_2020, q2_q4)) # True
print(engine.contains(q2_q4, h1_2020)) # False
```
### All 13 Relations
| `IntervalRelation` | `.value` | Inverse | Description |
| :--- | :--- | :--- | :--- |
| `BEFORE` | `"before"` | `AFTER` | A ends strictly before B starts |
| `AFTER` | `"after"` | `BEFORE` | A starts strictly after B ends |
| `MEETS` | `"meets"` | `MET_BY` | A ends exactly when B starts |
| `MET_BY` | `"met_by"` | `MEETS` | A starts exactly when B ends |
| `OVERLAPS` | `"overlaps"` | `OVERLAPPED_BY` | A and B share a period; A starts and ends first |
| `OVERLAPPED_BY` | `"overlapped_by"` | `OVERLAPS` | B starts and ends before A, they share a period |
| `STARTS` | `"starts"` | `STARTED_BY` | Same start time; A ends before B |
| `STARTED_BY` | `"started_by"` | `STARTS` | Same start time; B ends before A |
| `DURING` | `"during"` | `CONTAINS` | A is entirely inside B |
| `CONTAINS` | `"contains"` | `DURING` | B is entirely inside A |
| `FINISHES` | `"finishes"` | `FINISHED_BY` | Same end time; A started after B |
| `FINISHED_BY` | `"finished_by"` | `FINISHES` | Same end time; B started after A |
| `EQUALS` | `"equals"` | *(self-inverse)* | Identical interval |
### Additional Engine Methods
| Method | Returns | Description |
| :------ | :------- | :----------- |
| `active_at(interval, timestamp, granularity=None)` | `bool` | Is `timestamp` within `interval`? |
| `merge_intervals(intervals)` | `List[TemporalInterval]` | Merge overlapping/touching intervals |
| `gap_analysis(intervals, domain_start, domain_end)` | `List[TemporalInterval]` | Find uncovered gaps within a domain |
| `coverage_percentage(intervals, domain_start, domain_end)` | `float` | Fraction of domain covered by intervals |
| `timeline_of(entity_id, graph)` | `List[Dict]` | Sorted event timeline for an entity |
| `retroactive_coverage(revision, original_facts)` | `Dict` | Classify facts as `affected`, `partial`, or `unaffected` by a revision |
| `normalize_timestamp(timestamp, granularity)` | `datetime` | Truncate timestamp to granularity |
| `normalize_interval(start, end, granularity)` | `TemporalInterval` | Parse and expand interval to granularity boundaries |
### Advanced: Interval Operations
```python
from datetime import datetime, timezone
def dt(y, m, d): return datetime(y, m, d, tzinfo=timezone.utc)
intervals = [
TemporalInterval(start=dt(2020, 1, 1), end=dt(2020, 6, 30)),
TemporalInterval(start=dt(2020, 4, 1), end=dt(2020, 12, 31)),
TemporalInterval(start=dt(2021, 3, 1), end=TemporalBound.OPEN),
]
# Merge overlapping intervals
merged = engine.merge_intervals(intervals)
print(f"Merged into {len(merged)} intervals")
# Find gaps in coverage across 2020
gaps = engine.gap_analysis(intervals, dt(2020, 1, 1), dt(2020, 12, 31))
print(f"Uncovered gaps: {len(gaps)}")
# Coverage fraction
pct = engine.coverage_percentage(intervals, dt(2020, 1, 1), dt(2021, 12, 31))
print(f"Coverage: {pct:.1%}")
# Entity timeline (all add/modify/remove events sorted by time)
timeline = engine.timeline_of("alice", kg)
for event in timeline:
print(f" {event['timestamp'].date()} {event['change_type']}")
```
## TemporalNormalizer — NL Temporal Parsing
Converts natural-language temporal phrases into `(valid_from, valid_until)` datetime tuples. **Zero LLM calls.** Pure regex + `dateutil.relativedelta`.
```python
from semantica.kg import TemporalNormalizer
from datetime import datetime, timezone
norm = TemporalNormalizer(
reference_date=datetime(2024, 6, 15, tzinfo=timezone.utc)
)
```
### `normalize(value)` → `Optional[Tuple[datetime, datetime]]`
```python
# ISO 8601 → point interval
result = norm.normalize("2022-03-15")
print(result)
# (datetime(2022, 3, 15, tzinfo=utc), datetime(2022, 3, 15, tzinfo=utc))
# Year → full year span
result = norm.normalize("2022")
print(result)
# (datetime(2022, 1, 1, tzinfo=utc), datetime(2022, 12, 31, tzinfo=utc))
# Quarter → quarter span
result = norm.normalize("Q2 2021")
print(result)
# (datetime(2021, 4, 1, tzinfo=utc), datetime(2021, 6, 30, tzinfo=utc))
# Month + year
result = norm.normalize("January 2022")
print(result)
# (datetime(2022, 1, 1, tzinfo=utc), datetime(2022, 1, 31, tzinfo=utc))
# YYYY-MM (ISO partial)
result = norm.normalize("2022-03")
print(result)
# (datetime(2022, 3, 1, tzinfo=utc), datetime(2022, 3, 31, tzinfo=utc))
# Relative phrases (requires reference_date)
result = norm.normalize("last quarter")
print(result)
# (datetime(2024, 1, 1, tzinfo=utc), datetime(2024, 3, 31, tzinfo=utc))
result = norm.normalize("last year")
# (datetime(2023, 1, 1, tzinfo=utc), datetime(2023, 12, 31, tzinfo=utc))
# Unparseable → None (never raises, logs debug)
result = norm.normalize("recently")
print(result) # None
```
<Warning>
`normalize()` returns `None` for unparseable input — it **never raises** an exception. For relative phrases (`"last quarter"`, `"this year"`, etc.), `reference_date` **must** be set at construction time, otherwise `ValueError` is raised at call time.
</Warning>
### `normalize_phrase(phrase)` → `Optional[Dict]`
Look up a domain-specific temporal phrase in the phrase map:
```python
meta = norm.normalize_phrase("expiry date")
print(meta)
# {"maps_to": "valid_until", "type": "end", "domain": ["Healthcare", "Supply Chain"]}
meta = norm.normalize_phrase("retroactive to")
print(meta)
# {"maps_to": "valid_from", "type": "start", "retroactive": True, "domain": ["Regulatory", "Finance"]}
meta = norm.normalize_phrase("unknown phrase")
print(meta) # None
```
Built-in domain phrases cover: General/Policy, Healthcare, Cybersecurity, Supply Chain, Finance, and Energy.
### Custom Phrase Map
```python
from datetime import datetime, timezone
def my_grant_window(ref: datetime):
return (
datetime(ref.year, 10, 1, tzinfo=timezone.utc),
datetime(ref.year, 10, 31, tzinfo=timezone.utc),
)
norm = TemporalNormalizer(
reference_date=datetime(2024, 1, 1, tzinfo=timezone.utc),
phrase_map={"grant application window": my_grant_window},
)
start, end = norm.normalize("grant application window")
```
### Supported Expressions
| Pattern | Example | Return type |
| :------- | :------- | :---------- |
| ISO 8601 full date/datetime | `"2022-03-15"`, `"2022-03-15T10:00:00Z"` | Point interval |
| Year only | `"2022"` | Full year span |
| Month + year (word) | `"January 2022"`, `"Jan 2022"` | Full month span |
| YYYY-MM (ISO partial) | `"2022-03"` | Full month span |
| Quarter + year | `"Q2 2021"` | Quarter span |
| Relative (built-in) | `"last year"`, `"last quarter"`, `"this month"`, `"three months ago"`, `"six months ago"`, `"two years ago"` | Computed span |
| Ambiguous slash date | `"03/04/2022"` | `None` + `TemporalAmbiguityWarning` |
| Domain phrase | `"expiry date"`, `"retroactive to"` | Only via `normalize_phrase()` |
## TemporalQueryRewriter
Extract temporal intent from a natural-language query so downstream retrieval can apply deterministic temporal filtering.
**Two modes:** regex-only (no LLM) or LLM-assisted for free-form phrasing.
```python
from semantica.kg import TemporalQueryRewriter
# Regex-only (default — no dependencies beyond standard library)
rewriter = TemporalQueryRewriter()
# LLM-assisted for more complex phrasings
from semantica.llms import Groq
rewriter = TemporalQueryRewriter(
llm_provider=Groq(model="llama-3.1-8b-instant"),
reference_date=datetime.now(timezone.utc),
)
```
### `rewrite(query, context=None)` → `TemporalQueryResult`
```python
# "before" intent
r = rewriter.rewrite("which suppliers were certified before 2021?")
print(r.temporal_intent) # "before"
print(r.at_time.year) # 2021
print(r.rewritten_query) # "which suppliers were certified?"
print(r.confidence) # 0.85
# "between" intent
r = rewriter.rewrite("revenue between Q1 2022 and Q3 2022")
print(r.temporal_intent) # "between"
print(r.start_time) # datetime(2022, 1, 1, tzinfo=utc)
print(r.end_time) # datetime(2022, 9, 30, tzinfo=utc)
# "during" intent
r = rewriter.rewrite("what decisions were made during Q2 2023?")
print(r.temporal_intent) # "during"
print(r.at_time) # datetime(2023, 4, 1, tzinfo=utc)
# No temporal phrase
r = rewriter.rewrite("list all active suppliers")
print(r.temporal_intent) # None
print(r.rewritten_query) # "list all active suppliers"
print(r.has_temporal_context()) # False
```
### `TemporalQueryResult` Fields
| Field | Type | Description |
| :---- | :---- | :----------- |
| `rewritten_query` | `str` | Original query with the temporal phrase stripped and whitespace normalised |
| `at_time` | `Optional[datetime]` | Point-in-time bound for `before`, `after`, `at`, `during` intents |
| `start_time` | `Optional[datetime]` | Lower bound for `between` queries |
| `end_time` | `Optional[datetime]` | Upper bound for `between` queries |
| `temporal_intent` | `Optional[str]` | One of `"before"`, `"after"`, `"at"`, `"during"`, `"between"`, or `None` |
| `confidence` | `float` | `0.85` for regex extraction; LLM-propagated confidence or `0.75` fallback |
| Method | Returns | Description |
| :------ | :------- | :----------- |
| `has_temporal_context()` | `bool` | `True` if any temporal parameter was extracted |
Supported intent keywords: `before` / `prior to` / `until` / `up to`, `after` / `since` / `following`, `during` / `in` / `within`, `as of` / `at` / `on`, `between … and …`.
## TemporalVersionManager
Create and manage versioned graph snapshots with SHA-256 integrity checking. Supports both **in-memory** (default) and **SQLite persistent** storage.
```python
from semantica.kg import TemporalVersionManager
# In-memory (default)
versioner = TemporalVersionManager()
# SQLite-backed (persists across process restarts)
versioner = TemporalVersionManager(
storage_path="graph_versions.db",
version_strategy="timestamp", # timestamp | incremental | semantic
)
```
### Methods
| Method | Returns | Description |
| :------ | :------- | :----------- |
| `create_snapshot(graph, version_label, author, description)` | `Dict` | Create snapshot with SHA-256 checksum. `author` and `description` are required |
| `create_version(graph, version_label=None, timestamp=None, metadata=None)` | `Dict` | Lightweight version without checksum or mandatory author |
| `list_versions()` | `List[Dict]` | List all stored snapshots |
| `get_version(label)` | `Optional[Dict]` | Retrieve snapshot by label |
| `compare_versions(v1, v2, comparison_metrics=None)` | `Dict` | Detailed entity + relationship diff between two versions or labels |
| `apply_revision(snapshot, revision)` | `Dict` | Temporal revision: supersede matching facts without deleting originals |
| `validate_snapshot(snapshot)` | `bool` | Validate against v1.0 schema (required fields + types) |
| `migrate_snapshot(snapshot)` | `Dict` | Upgrade old-format snapshot to v1.0 |
| `verify_checksum(snapshot)` | `bool` | Integrity check via SHA-256 |
### Snapshot & Diff Example
```python
# Create a snapshot (author and description are required)
snap = versioner.create_snapshot(
kg,
version_label="v1.0",
author="analyst@example.com",
description="Initial baseline",
)
print(snap["checksum"]) # SHA-256 hex string
# List versions
for v in versioner.list_versions():
print(f"{v['label']:12s} {v['author']} {v['timestamp']}")
# Get a specific version
past = versioner.get_version("v1.0")
# Diff: compare two versions (pass labels or snapshot dicts)
diff = versioner.compare_versions("v1.0", "v2.0")
print(f"Entities added: {diff['summary']['entities_added']}")
print(f"Entities removed: {diff['summary']['entities_removed']}")
print(f"Relationships added: {diff['summary']['relationships_added']}")
print(f"Relationships removed: {diff['summary']['relationships_removed']}")
# Field-level changes on each modified entity
for change in diff["entities_modified"]:
print(f" {change['id']}: {change['changes']}")
```
### Temporal Revision
Apply a revision to specific fact IDs — the originals are **superseded** (not deleted), preserving full audit history:
```python
revision = {
"fact_ids": ["alice|ceo_of|acme_corp"], # relationship key: src|type|target
"new_valid_from": "2018-03-01",
"new_valid_until": None, # None = TemporalBound.OPEN
"revision_type": "correction", # correction | retroactive
"author": "analyst@example.com",
"reason": "Original start date was incorrect",
}
revised_snapshot = versioner.apply_revision(snap, revision)
# original fact is preserved with superseded_at set
# replacement fact has new_valid_from, superseded_at = OPEN
```
### Integrity & Migration
```python
# Validate snapshot schema
is_valid = versioner.validate_snapshot(snap)
# Verify checksum integrity
is_intact = versioner.verify_checksum(snap)
# Upgrade old-format snapshot (no format_version field)
upgraded = versioner.migrate_snapshot(old_snap)
```
## Context Graph Temporal Features (v0.3.0)
The `ContextGraph` exposes temporal awareness directly on graph nodes and decisions, available since v0.3.0:
```python
from semantica.context import ContextGraph
from datetime import datetime, timezone
graph = ContextGraph(advanced_analytics=True)
# Add time-bounded nodes
graph.add_node("policy_v1", "policy",
properties={"text": "All transactions require dual approval"},
valid_from="2021-01-01",
valid_until="2023-06-30")
graph.add_node("policy_v2", "policy",
properties={"text": "Transactions > $50k require dual approval"},
valid_from="2023-07-01")
# Find nodes active at a specific timestamp
current_policies = graph.find_active_nodes(
node_type="policy",
at_time=datetime.now(timezone.utc),
)
for p in current_policies:
print(p["properties"]["text"])
# → "Transactions > $50k require dual approval"
# Historical query
past_policies = graph.find_active_nodes(
node_type="policy",
at_time=datetime(2022, 6, 1, tzinfo=timezone.utc),
)
for p in past_policies:
print(p["properties"]["text"])
# → "All transactions require dual approval"
```
### Temporal Decision Windows
```python
from semantica.context import AgentContext, ContextGraph
from semantica.vector_store import VectorStore
context = AgentContext(
vector_store=VectorStore(backend="faiss", dimension=768),
knowledge_graph=ContextGraph(),
decision_tracking=True,
)
# Decision superseded after policy change
old_id = context.record_decision(
category="data_retention", scenario="Set retention window for user PII",
reasoning="GDPR Article 5(1)(e) limits storage",
outcome="retain_90_days", confidence=0.98,
valid_from="2023-01-01", valid_until="2023-06-30",
)
new_id = context.record_decision(
category="data_retention", scenario="Set retention window for user PII",
reasoning="Legal confirmed 60-day window after new DPA amendment",
outcome="retain_60_days", confidence=0.99,
valid_from="2023-07-01",
)
# Temporal precedent search
old_prec = context.find_precedents("data retention PII", as_of="2023-03-01", limit=3)
new_prec = context.find_precedents("data retention PII", as_of="2024-01-01", limit=3)
```
## Real-World Patterns
<Tabs>
<Tab title="Personnel & Org Structure">
```python
from semantica.kg import GraphBuilder, TemporalGraphQuery
builder = GraphBuilder()
kg = builder.build(sources=[{
"entities": [
{"id": "alice", "type": "Person"},
{"id": "finteam", "type": "Team"},
],
"relationships": [
{"source": "alice", "target": "finteam", "type": "leads",
"valid_from": "2020-01-01", "valid_until": "2022-12-31"},
],
}])
query = TemporalGraphQuery()
# Incident in Nov 2022 → who was responsible?
result = query.query_at_time(kg, "", "2022-11-15")
leads = [r for r in result["relationships"] if r["type"] == "leads"]
print(f"Team lead at incident: {leads[0]['source']}")
```
</Tab>
<Tab title="Policy Evolution">
```python
from semantica.kg import TemporalVersionManager, TemporalGraphQuery
versioner = TemporalVersionManager(storage_path="policy_history.db")
versioner.create_snapshot(kg_before, version_label="2023-H1",
author="compliance@org.com",
description="Pre-July policy baseline")
versioner.create_snapshot(kg_after, version_label="2023-H2",
author="compliance@org.com",
description="Post-July amendment")
diff = versioner.compare_versions("2023-H1", "2023-H2")
print(f"Policy changes: {diff['summary']['relationships_modified']}")
```
</Tab>
<Tab title="Consistency Audit">
```python
from semantica.kg import TemporalGraphQuery
report = TemporalGraphQuery().validate_temporal_consistency(kg)
if report.errors:
print("ERRORS (must fix):")
for e in report.errors:
print(f" [{e['issue_type']}] {e['message']} (fact: {e['fact_id']})")
if report.warnings:
print("WARNINGS (review):")
for w in report.warnings:
print(f" [{w['issue_type']}] {w['message']} (fact: {w['fact_id']})")
```
</Tab>
<Tab title="NL Query Rewriting">
```python
from semantica.kg import TemporalQueryRewriter, TemporalGraphQuery
rewriter = TemporalQueryRewriter()
query = TemporalGraphQuery()
user_query = "Who was responsible for compliance before the 2022 audit?"
result = rewriter.rewrite(user_query)
if result.has_temporal_context():
# Use point-in-time filtering
snapshot = query.reconstruct_at_time(kg, result.at_time)
else:
snapshot = kg
# Now run your retrieval over snapshot with result.rewritten_query
print(f"Intent: {result.temporal_intent}")
print(f"Query: {result.rewritten_query}")
```
</Tab>
</Tabs>
## Configuration
```yaml
kg:
temporal:
enabled: true
default_validity: infinite # OPEN when valid_until is omitted
recorded_at_auto_stamp: true # auto-fill recorded_at on every ingested fact
reasoning:
enabled: true
granularity: day # second|minute|hour|day|week|month|year
engine: allen # allen | point_in_time_only
```
- [Knowledge Graph Module](kg) — Core graph construction, `GraphBuilder`, analytics.
- [Context Module](context) — Decision temporal windows and `find_active_nodes()`.
- [Provenance](provenance) — W3C PROV-O lineage stamped alongside temporal metadata.
- [Export](export) — OWL, Turtle, JSON-LD, and Parquet export with temporal annotations.
- [Temporal Knowledge Graphs](https://github.com/semantica-agi/semantica/blob/main/cookbook/advanced/10_Temporal_Knowledge_Graphs.ipynb) — Temporal reasoning and Allen algebra · Advanced
- [Context Module](https://github.com/semantica-agi/semantica/blob/main/cookbook/introduction/19_Context_Module.ipynb) — Including temporal decision windows · Intermediate