How to Query Historical Graph States with Point-in-Time Queries in Semantica
Use TemporalGraphQuery with query_at_time() to reconstruct any past state of a Semantica knowledge graph by filtering entities and relationships against their valid_from/valid_until timestamps.
Semantica provides native support for temporal knowledge graphs, storing validity periods directly on nodes and edges. The TemporalGraphQuery engine in semantica/kg/temporal_query.py enables precise point-in-time queries to retrieve exact graph snapshots as they existed at any historical moment.
How TemporalGraphQuery Works
The query engine operates in four stages when processing a point-in-time request:
- Initialize — Configure logging, progress tracking, and optional temporal reasoning
- Parse timestamp — Normalize ISO strings, Unix timestamps, or
datetimeobjects - Reconstruct subgraph — Deep-copy the graph and filter inactive elements via
reconstruct_at_time() - Return snapshot — Package the result with metadata, active entities, and valid relationships
Core Methods in semantica/kg/temporal_query.py
| Method | Location | Purpose |
|---|---|---|
__init__() |
Lines 45-72 | Sets up the engine with optional temporal reasoning and progress tracking |
query_at_time() |
Lines 107-144 | Public entry point: parses time, calls reconstruction, builds result dictionary |
reconstruct_at_time() |
Lines 170-209 | Deep-copies graph and applies temporal filters using _entity_active_at_time() and _relationship_active_at_time() |
Basic Point-in-Time Query Example
from semantica.kg.temporal_query import TemporalGraphQuery
# Define a temporal knowledge graph
graph = {
"entities": [
{"id": "e1", "name": "Alice"},
{"id": "e2", "name": "Bob"},
],
"relationships": [
{
"id": "r1",
"source": "e1",
"target": "e2",
"type": "knows",
"valid_from": "2022-01-01T00:00:00Z",
"valid_until": "2023-06-30T23:59:59Z",
},
{
"id": "r2",
"source": "e2",
"target": "e1",
"type": "likes",
"valid_from": "2023-07-01T00:00:00Z",
"valid_until": None, # still active
},
],
}
# Initialize query engine
tgq = TemporalGraphQuery(enable_temporal_reasoning=False)
# Query graph state as of March 15, 2023
snapshot = tgq.query_at_time(
graph=graph,
query="snapshot-2023-03-15",
at_time="2023-03-15T12:00:00Z",
)
print(f"Timestamp: {snapshot['at_time']}")
print(f"Active relationships: {snapshot['num_relationships']}")
print(snapshot["relationships"])
Output:
Timestamp: 2023-03-15 12:00:00+00:00
Active relationships: 1
[{'id': 'r1', 'source': 'e1', 'target': 'e2', 'type': 'knows', ...}]
Only r1 appears because r2 had not yet become valid. The reconstruct_at_time() method filters edges by checking valid_from <= query_time < valid_until (or valid_until is None for unbounded validity).
Configuring Temporal Axes and Granularity
Time Axis Selection
By default, query_at_time() uses the valid time axis (valid_from/valid_until). Switch to transaction time for audit-style queries:
# Use transaction timestamps instead of validity periods
snapshot = tgq.query_at_time(
graph=graph,
query="audit-snapshot",
at_time="2023-03-15T12:00:00Z",
time_axis="transaction", # default: "valid"
)
Temporal Granularity
The engine respects granularity settings when normalizing timestamps. Available levels include second, minute, hour, day, week, month, year. Configure via the graph's temporal model or pass explicitly to parsing utilities in semantica/kg/temporal_model.py.
Advanced Query Options
Include Historical Metadata
To retrieve inactive relationships alongside active ones for audit purposes:
snapshot = tgq.query_at_time(
graph=graph,
query="full-history",
at_time="2023-03-15T12:00:00Z",
include_history=True, # includes edges valid before/after timestamp
)
Batch Historical Snapshots
Generate time-series graph states for analysis or visualization:
from datetime import datetime, timedelta
timestamps = [datetime(2023, 1, 1) + timedelta(days=30*i) for i in range(12)]
snapshots = [
tgq.query_at_time(graph, f"month-{i}", ts)
for i, ts in enumerate(timestamps)
]
Combine with Pattern Detection
Feed point-in-time results into temporal pattern analysis:
# First obtain a historical snapshot
snapshot = tgq.query_at_time(graph, "pattern-base", "2023-06-01")
# Then detect patterns existing at that moment
patterns = tgq.query_temporal_pattern(
graph=snapshot,
pattern_type="causal_chain",
time_window=timedelta(days=30),
)
Key Supporting Files
| File | Role |
|---|---|
semantica/kg/temporal_query.py |
Core query engine with TemporalGraphQuery class |
semantica/kg/temporal_model.py |
Temporal value parsing (parse_temporal_value), TemporalBound dataclass |
semantica/utils/progress_tracker.py |
Progress monitoring for long-running reconstructions |
semantica/visualization/temporal_visualizer.py |
Rendering point-in-time snapshots and evolution timelines |
Summary
- Point-in-time queries use
TemporalGraphQuery.query_at_time()insemantica/kg/temporal_query.py - The engine filters entities and relationships by their
valid_from/valid_untiltimestamps viareconstruct_at_time() - Support multiple time axes (
valid,transaction) and granularity levels - Enable
include_history=Trueto retrieve inactive elements for auditing - Combine with pattern detection and batch processing for advanced temporal analytics
Frequently Asked Questions
What timestamp formats does Semantica accept for point-in-time queries?
Semantica accepts ISO 8601 strings (e.g., "2023-03-15T12:00:00Z"), Unix timestamps (integer or float seconds since epoch), and Python datetime objects. The parse_temporal_value() function in semantica/kg/temporal_model.py normalizes all inputs to timezone-aware datetimes before filtering.
How does Semantica handle unbounded or currently valid relationships?
Relationships with valid_until=None are treated as permanently valid from their valid_from timestamp forward. The _relationship_active_at_time() helper checks valid_until is None or query_time < valid_until, so open-ended edges remain active in all future point-in-time queries until explicitly closed.
Can I query historical states using transaction time instead of validity time?
Yes. Pass time_axis="transaction" to query_at_time() or reconstruct_at_time(). This switches the filtering logic to use transaction_start and transaction_end fields instead of valid_from/valid_until, enabling audit trails of when records were physically added or modified in the system.
Is progress tracking available for large graph reconstructions?
Yes. TemporalGraphQuery automatically initializes a progress tracker from semantica/utils/progress_tracker.py. For long-running operations on large graphs, monitor the tracker to observe filtering progress through entities and relationships during the reconstruction phase.
Have a question about this repo?
These articles cover the highlights, but your codebase questions are specific. Give your agent direct access to the source. Share this with your agent to get started:
curl -s "https://instagit.com/install.md" Maintain an open-source project? Get it listed too →