Skip to content

Run temporal queries

Track how threat intelligence changes over time. ZettelForge stores a temporal graph index that records supersession events and date-stamped observations as SUPERSEDES and OCCURRED_ON edges. You can query this index directly, or phrase your recall() queries to trigger temporal retrieval.

Prerequisites

  • ZettelForge installed and initialized (pip install zettelforge)
  • At least some notes in the store (run mm.remember() first if the store is empty)

No LLM provider is required for the graph-based steps (Steps 2 and 3). Steps 4 and 5 need a configured LLM; see Configuration.

Steps

1. Store notes at different points in time

from zettelforge.memory_manager import MemoryManager

mm = MemoryManager()

# Earlier intelligence
note1, _ = mm.remember(
    content="APT28 used Cobalt Strike for C2 in January 2026 campaigns targeting EU governments.",
    source_type="report",
    source_ref="vendor-report-jan-2026",
    domain="cti",
)

# Later intelligence about the same group and campaign
note2, _ = mm.remember(
    content="APT28 shifted to Sliver C2 framework in March 2026, replacing Cobalt Strike in EU operations.",
    source_type="report",
    source_ref="vendor-report-mar-2026",
    domain="cti",
)

When note2 shares entities with note1 and the overlap score exceeds the supersession threshold, ZettelForge marks note1.links.superseded_by = note2.id and adds a SUPERSEDES edge to the knowledge graph. The supersession check is automatic; you do not configure it per call.

2. Get an entity's temporal timeline

get_entity_timeline() returns the sequence of temporal edges recorded for one entity node. The edges include SUPERSEDES, TEMPORAL_BEFORE, and TEMPORAL_AFTER relationships.

from zettelforge.knowledge_graph import get_knowledge_graph

kg = get_knowledge_graph()

# Use the correct entity type: APT-numbered groups are indexed as 'intrusion_set'
timeline = kg.get_entity_timeline(
    entity_type="intrusion_set",
    entity_value="apt28",
)

for entry in timeline:
    print(f"  [{entry['timestamp']}] {entry['edge']['relationship']} -> {entry['to_entity']}")

Return shape — each item in the list:

Key Type Description
timestamp str ISO-8601 timestamp from the edge properties
edge dict Full edge dict (relationship, from/to node IDs, properties)
to_entity str "type:value" string for the target node

Returns [] if no entity node of that type and value exists in the graph, or if no temporal edges have been recorded for it.

Note

Entity types follow the extractor's normalized values. APT-numbered groups (APT28, APT41) are stored as intrusion_set. Named tools are stored as tool. Use the types from KG Edge Schema.

3. Get all changes since a timestamp

get_changes_since() scans the temporal index and returns every SUPERSEDES, TEMPORAL_BEFORE, and TEMPORAL_AFTER edge recorded on or after the given timestamp.

changes = kg.get_changes_since("2026-03-01T00:00:00")

for change in changes:
    print(
        f"  [{change['timestamp']}] "
        f"{change['from']} --{change['relationship']}--> "
        f"{change['to']}"
    )

Return shape — each item:

Key Type Description
timestamp str ISO-8601 timestamp from the edge
from str "type:value" for the source node
relationship str SUPERSEDES, TEMPORAL_BEFORE, or TEMPORAL_AFTER
to str "type:value" for the target node

Warning

Pass full ISO-8601 timestamps (2026-03-01T00:00:00), not partial dates (2026-03-01). The comparison is a string lexicographic sort on the created_at column; partial dates sort correctly for midnight boundaries but can misorder intra-day edges on older SQLite deployments.

You can also access these methods through the store directly: mm.store.get_changes_since("2026-03-01T00:00:00") returns the same shape from the SQLite backend.

4. Use temporal query phrasing with recall()

recall() detects temporal intent from your query wording and adjusts retrieval weights to favor notes containing matching date strings.

notes = mm.recall(
    query="How has APT28 tooling changed since January 2026?",
    domain="cti",
    k=10,
)

for note in notes:
    status = "SUPERSEDED" if note.links.superseded_by else "CURRENT"
    print(f"  [{status}] {note.created_at[:10]}: {note.content.raw[:80]}")

Queries containing words like when, since, before, after, changed, history, timeline, recent, latest, or previously trigger the temporal retrieval path (retriever weight 0.2, temporal weight 0.5, top_k 5 by default).

By default, recall() hides superseded notes (exclude_superseded=True). Pass exclude_superseded=False to include historical versions alongside current ones — useful when reconstructing how an assessment evolved.

If dateparser is installed, the retriever also extracts date strings from the query and boosts notes whose content contains those dates. Without dateparser, temporal intent still fires but no date-boost is applied.

5. Synthesize an answer about temporal changes

synthesize() retrieves notes and passes them to an LLM for a structured response.

result = mm.synthesize(
    query="Timeline of APT28 tool changes in 2026",
    format="direct_answer",
    k=10,
)

print(result["synthesis"]["answer"])

ThreatRecall.ai SaaS

The format="timeline_analysis" extended format is available on ThreatRecall.ai. In ZettelForge OSS, requesting timeline_analysis silently falls back to direct_answer. Use format="direct_answer" in OSS code so the behavior is explicit and the result key (result["synthesis"]["answer"]) is correct.

6. Check the supersession status of a specific note

note = mm.store.get_note_by_id(note1.id)

if note.links.superseded_by:
    print(f"Note {note.id} was superseded by {note.links.superseded_by}")
    newer = mm.store.get_note_by_id(note.links.superseded_by)
    print(f"  Newer content: {newer.content.raw[:100]}")
else:
    print(f"Note {note.id} is current (not superseded)")

7. Anchor report ingestion to a publication date

Pass published_date when ingesting reports to give the LLM extraction phase a temporal anchor. The date is included in the extraction context so the LLM can correctly associate facts with the right time period.

results = mm.remember_report(
    content="Lazarus Group deployed a new DTrack variant in February 2026 against South Korean targets...",
    source_url="https://example.com/lazarus-feb-2026",
    published_date="2026-02-15",
    domain="cti",
)

published_date requires a configured LLM provider. Without one, remember_report() completes with 0 extracted operations (fact extraction is LLM-dependent).

Reference

Concept API
Entity temporal history kg.get_entity_timeline(entity_type, entity_value)
All changes since a time kg.get_changes_since(iso_timestamp)
Same via store mm.store.get_changes_since(iso_timestamp)
Temporal query retrieval mm.recall(query, ...) with temporal keywords
Exclude superseded notes mm.recall(..., exclude_superseded=True) (default)
Include historical notes mm.recall(..., exclude_superseded=False)
OSS synthesis mm.synthesize(query, format="direct_answer")
Extended synthesis mm.synthesize(query, format="timeline_analysis") — ThreatRecall.ai SaaS
Supersession check note.links.superseded_by — non-empty if note is outdated
Temporal edge types SUPERSEDES, TEMPORAL_BEFORE, TEMPORAL_AFTER, OCCURRED_ON