How Semantica Implements Provenance Tracking with PROV-O: A Complete Technical Guide

Semantica implements provenance tracking using the W3C PROV-O ontology through a unified three-module architecture: provenance/schemas.py defines the data model, provenance/manager.py orchestrates creation and queries, and provenance/storage.py handles persistence with hash-chain integrity and graph traversal capabilities.

Semantica is an open-source knowledge extraction system that requires rigorous auditability for every fact it generates. Its provenance subsystem provides complete W3C PROV-O compliance, enabling downstream systems to trace who did what, when, and why. This article examines how Semantica maps PROV-O concepts to code, maintains data integrity through cryptographic chaining, and exposes lineage queries for forensic analysis.

PROV-O Data Model: The ProvenanceEntry Schema

The foundation of Semantica's provenance system lives in semantica/provenance/schemas.py. The ProvenanceEntry dataclass provides a one-to-one mapping between Python fields and PROV-O ontology terms.

Field	PROV-O Equivalent	Purpose
entity_id	prov:Entity	Identifies the subject of provenance
activity_id	prov:Activity	Names the process that created/used the entity
agent_id, agent_type	prov:Agent	Records who performed the activity (Person, SoftwareAgent, or Organization)
derived_from_id	prov:wasDerivedFrom	Cross-source derivation
used_entities	prov:used	Entities consumed as input
timestamp	prov:generatedAtTime	Instant of generation
invalidated	prov:Invalidation	Tombstone marker for retractions

The schema also captures temporal bounds (activity_started_at_time, activity_ended_at_time), delegation (acted_on_behalf_of), and provenance bundles (bundle_id). All fields are optional, allowing graceful degradation when metadata is incomplete.

@dataclass
class ProvenanceEntry:
    entity_id: str
    entity_type: str
    activity_id: str
    agent_id: str = "semantica"
    agent_type: str = "software_agent"  # Maps via _AGENT_TYPE_PROV_CLASS in manager.py

    timestamp: str = field(default_factory=lambda: utc_now_iso())
    parent_entity_id: Optional[str] = None      # prov:wasDerivedFrom (legacy)

    derived_from_id: Optional[str] = None       # prov:wasDerivedFrom (true derivation)

    used_entities: List[str] = field(default_factory=list)  # prov:used

    # ... additional fields for versioning, invalidation, bundles

The ProvenanceManager: Unified API for PROV-O Operations

semantica/provenance/manager.py contains the ProvenanceManager class—the primary interface for creating, querying, and exporting provenance records.

Core Tracking Methods

• track_entity() – Records provenance for extracted entities
• track_relationship() – Captures relationship derivations
• track_chunk() – Documents text segmentation operations
• track_property_source() – Links properties to their origins

Each method resolves agent and activity parameters, constructs a ProvenanceEntry, computes integrity checksums, and delegates persistence to the configured storage backend.

def track_entity(self, entity_id: str, source: str,
                 metadata: Optional[Dict[str, Any]] = None,
                 _conn: Optional[Any] = None, **kwargs) -> Optional[ProvenanceEntry]:
    # Resolve agent/activity, compute provenance entry, store with checksum

    entry = ProvenanceEntry(
        entity_id=entity_id,
        entity_type=kwargs.get("entity_type", "entity"),
        activity_id=activity_info["activity_id"],
        agent_id=agent_info["agent_id"],
        source_document=source,
        metadata=metadata or {},
        first_seen=existing.first_seen if existing else utc_now_iso(),
        last_updated=utc_now_iso(),
        parent_entity_id=parent_id,
        used_entities=list(kwargs.get("used_entities", [])),
        activity_started_at_time=activity_info["activity_started_at_time"],
        activity_ended_at_time=activity_info["activity_ended_at_time"],
    )
    self._save_entry(entry, _conn=conn, _raise_on_error=True)
    return entry

Hash-Chain Integrity and Versioning

Semantica implements two distinct lineage concepts:

• previous_version_id – Identifies a correction of the same fact (versioning)
• derived_from_id – Records genuine derivation from a different source

Every entry carries a SHA-256 checksum of its contents and references the previous entry's checksum (previous_checksum) plus a sequential sequence_id. The verify_checksum() method detects any deletion or tampering by validating this chain.

Batch Operations for Performance

Large-scale extraction pipelines use track_entities_batch() and track_chunks_batch(), which partition inputs into 1,000-item transactions to minimize SQLite lock contention and improve throughput.

Storage Backends: In-Memory and SQLite

semantica/provenance/storage.py provides two implementations of the same interface:

• InMemoryStorage – Fast, non-persistent storage for testing
• SQLiteStorage – File-backed persistence for production

Both expose store(), retrieve(), trace_lineage(), trace_descendants(), and transaction management. The storage layer also implements get_chain_head(), used by the manager to compute sequence_id and previous_checksum for new entries.

Lineage Queries: Upstream and Downstream Traversal

Semantica's PROV-O implementation supports bidirectional graph traversal for forensic analysis.

get_lineage(entity_id)

Returns the complete ancestor chain including aggregated metadata, source documents, and checksum verification status.

trace_descendants(entity_id)

Returns all downstream entries that depend on the given entity—critical for incident response ("who used this contaminated fact?").

Both methods follow parent_entity_id, derived_from_id, and used_entities edges through the storage layer's graph utilities.

Invalidation: Tombstone Pattern

Instead of hard deletion, invalidate(entity_id, agent_id, ...) creates a new entry marking the original as retracted. This preserves the complete audit trail per the PROV-O prov:Invalidation pattern.

PROV-O RDF Export

The export_prov(format="turtle", base_uri=None) method serializes all stored entries to standard RDF formats (Turtle, JSON-LD, N-Triples). It binds the prov namespace and emits fully compliant PROV-O triples:

PROV = Namespace("http://www.w3.org/ns/prov#")
EX = Namespace(base_uri or DEFAULT_BASE_URI)

g = Graph()
g.bind("prov", PROV)
g.bind("ex", EX)

for e in self.storage.retrieve_all():
    ent_uri = URIRef(EX[str(e.entity_id)])
    g.add((ent_uri, RDF.type, PROV.Entity))
    if e.timestamp:
        g.add((ent_uri, PROV.generatedAtTime,
               Literal(e.timestamp, datatype=XSD.dateTime)))
    # Additional triples for agents, activities, relationships...

Complete Working Example

from semantica.provenance.manager import ProvenanceManager

# Initialise with SQLite persistence

prov = ProvenanceManager(storage_path="my_provenance.db")

# Track entity extracted by NER pipeline

prov.track_entity(
    entity_id="gene_ABC1",
    source="doi:10.1234/example.pdf",
    metadata={"label": "Gene", "score": 0.97},
    agent_id="ner_extractor",
    agent_type="software_agent",
    role="generator",
    activity_id="ner_extraction",
    activity_started_at_time="2024-01-15T09:00:00Z",
    activity_ended_at_time="2024-01-15T09:00:05Z",
)

# Track source document chunk

prov.track_chunk(
    chunk_id="chunk_001",
    source_document="doi:10.1234/example.pdf",
    source_path="/tmp/example.pdf",
    start_index=120,
    end_index=250,
    parent_chunk_id=None,
    agent_id="chunker",
    activity_id="text_split",
)

# Retrieve and verify upstream lineage

lineage = prov.get_lineage("gene_ABC1")
print(f"Sources: {lineage['source_documents']}")
print(f"Verified: {lineage['checksum_valid']}")

# Export for external provenance tools

turtle = prov.export_prov(format="turtle")
print(turtle)

Summary

• Three-module architecture: schemas.py (data model), manager.py (API/integrity), storage.py (persistence/traversal)
• Full PROV-O compliance: ProvenanceEntry maps directly to prov:Entity, prov:Activity, prov:Agent, and their relations
• Cryptographic integrity: SHA-256 hash chaining with previous_checksum and sequence_id enables tamper detection
• Distinguishes versioning from derivation: previous_version_id for corrections, derived_from_id for cross-source origins
• Bidirectional lineage: get_lineage() for ancestors, trace_descendants() for downstream dependency analysis
• Safe invalidation: Tombstone pattern preserves complete audit trails per prov:Invalidation
• Standard export: Turtle/JSON-LD/N-Triples output for PROV-O-aware visualization and compliance tools

Frequently Asked Questions

What PROV-O classes does Semantica support?

Semantica implements prov:Entity, prov:Activity, and prov:Agent (with typed subclasses Person, SoftwareAgent, Organization via _AGENT_TYPE_PROV_CLASS in manager.py). It also supports prov:wasDerivedFrom, prov:used, prov:generatedAtTime, prov:Invalidation, and provenance bundles.

How does Semantica prevent provenance record tampering?

The ProvenanceManager computes a SHA-256 checksum for each ProvenanceEntry and stores it with a reference to the previous entry's checksum (previous_checksum) and a monotonic sequence_id. The verify_checksum() method detects any deletion or modification by validating this chain.

Can Semantica export provenance for external tools?

Yes. The export_prov(format="turtle") method serializes all entries to standard RDF formats including Turtle, JSON-LD, and N-Triples, using the http://www.w3.org/ns/prov# namespace. This enables integration with PROV-O visualizers, compliance systems, and knowledge graphs.

What is the difference between parent_entity_id and derived_from_id?

parent_entity_id is the legacy field for prov:wasDerivedFrom relationships, while derived_from_id explicitly records genuine derivations from different sources. Semantica also uses previous_version_id to track corrections of the same fact—distinguishing versioning from derivation in the audit trail.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}