Context Module - Semantica

semantica.context is the memory and decision layer for AI agents:

Stores facts with provenance and embedding-backed retrieval
Records decisions as first-class graph objects with full causal chains
Lets agents search their own history to stay consistent across runs
Answers complex queries via multi-hop GraphRAG traversal
Enforces versioned policies and tracks compliance exceptions

Exported Classes

Class	Role
`AgentContext`	Primary entry point: memory, retrieval, decisions, graph traversal, checkpoints
`ContextGraph`	In-memory knowledge graph with centrality, community detection, and decision tracking
`AgentMemory`	Vector-backed persistent memory: `store(text)`, `retrieve(query, max_results)`
`EntityLinker`	Link entity mentions to URIs; create typed edges between entity IDs
`ContextRetriever`	Hybrid vector + graph retrieval with min-score and graph expansion options
`DecisionRecorder`	Record decisions with embeddings, causal chains, and metadata
`PolicyEngine`	Policy management: `add_policy()`, `check_compliance()`, `get_applicable_policies()`
`CausalChainAnalyzer`	Trace how decisions influenced each other: `get_causal_chain(decision_id)`
`ErasureCoordinator`	Erase an entity across graph, memory, and vector store, returning an auditable `ErasureReceipt`

What You Get

AgentContext: memory, decision tracking, and graph-backed retrieval behind one API
- Conversation history and checkpoint diffing
- Persist and restore full context state to disk
ContextGraph: thread-safe in-memory knowledge graph
- PageRank, centrality, community detection, temporal validity
- Cross-graph navigation and link traversal
AgentMemory: embedding-backed memory with retention policy
- LRU eviction at configurable max_memory_size
- Per-conversation history isolation
DecisionRecorder: records decisions with causal chains and confidence scores
- Temporal validity windows (valid_from / valid_until)
- Cross-system context capture on every decision
PolicyEngine: versioned policy storage in the knowledge graph
- Compliance checking against recorded decisions
- Policy exception tracking with approver audit trail
EntityLinker: maps entity text to stable URIs
- Creates typed links between entity IDs
- Prevents “Apple”, “Apple Inc.”, “AAPL” becoming separate nodes
ContextRetriever: fuses vector similarity, graph traversal, and agent memory
- Richer context than pure vector search
- Configurable hybrid_alpha and expansion hops
CausalChainAnalyzer: traces upstream causes and downstream effects of any decision
- Explainability paths with relationship types
- Configurable depth and direction

Quick Start

Initialize the agent context

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,
    retention_days=90,
    max_memories=50000,
)

Store facts and retrieve by semantic similarity

memory_id = context.store(
    "GPT-4 outperforms GPT-3.5 on reasoning benchmarks by 40%",
    metadata={"source": "openai_blog", "date": "2024-01"}
)

results = context.retrieve("LLM benchmark comparisons", max_results=5)
for r in results:
    print("{} (score: {:.3f})".format(r["content"], r["score"]))

Record decisions with full provenance

decision_id = context.record_decision(
    category="model_selection",
    scenario="Choose LLM for production reasoning pipeline",
    reasoning="GPT-4 benchmark advantage justifies 3x cost increase",
    outcome="selected_gpt4",
    confidence=0.91,
    entities=["gpt-4", "gpt-3.5"],
    decision_maker="pipeline_agent",
)

Find precedents and trace causal chains

# Search past decisions: prevents contradictory choices across runs
precedents = context.find_precedents("model selection reasoning", limit=5)
for p in precedents:
    print("[{}] {}  (confidence: {:.2f})".format(p.category, p.outcome, p.confidence))
    print("  Reasoning: {}".format(p.reasoning))

# Trace downstream decisions influenced by this one
chain = context.get_causal_chain(decision_id, direction="downstream", max_depth=5)
print("Downstream decisions: {}".format(len(chain)))

# Full explainability
explanation = context.trace_decision_explainability(decision_id)
print("Total connections: {}".format(explanation["total_connections"]))

Usage Patterns

Vector Memory Only
Full Agent Context
GraphRAG Query
Policy Enforcement

Fastest setup: no knowledge graph. Best for agents that need semantic search over facts without graph traversal overhead.

from semantica.context import AgentContext
from semantica.vector_store import VectorStore

# Zero-graph setup: vector memory only
context = AgentContext(
    vector_store=VectorStore(backend="faiss", dimension=768),
)

context.store("User prefers concise responses with code examples")
context.store("Project uses Python 3.11 with FastAPI and PostgreSQL")

results = context.retrieve("user coding preferences", max_results=5)
for r in results:
    print("{:.3f}  {}".format(r["score"], r["content"]))

Swap backend="faiss" to backend="inmemory" for zero-dependency local development.

Production setup: graph + decisions + analytics. Use when you need explainability and contradiction-free decision history.

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,  # PageRank, centrality, community detection
        kg_algorithms=True,       # path-finding, link prediction
    ),
    decision_tracking=True,       # requires knowledge_graph
    retention_days=90,
    max_memories=50000,
)

decision_id = context.record_decision(
    category="model_selection",
    scenario="Choose LLM for production reasoning pipeline",
    reasoning="GPT-4 benchmark advantage justifies 3x cost",
    outcome="selected_gpt4",
    confidence=0.91,
    entities=["gpt-4", "gpt-3.5"],
)

# Prevent contradictions across runs
precedents = context.find_precedents("model selection", limit=5)

Load a pre-built knowledge graph and answer complex questions with multi-hop graph traversal.

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),
    hybrid_alpha=0.4,        # 0.0 = pure vector  →  1.0 = pure graph
    max_expansion_hops=3,
)

# Load a pre-built knowledge graph
context.load_graph("company_kg.json")

# Multi-hop GraphRAG retrieval
results = context.retrieve(
    "companies founded by Apple alumni",
    use_graph=True,
    max_results=10,
)
for r in results:
    print("[{:.3f}] {}".format(r["score"], r["content"]))

Increase max_expansion_hops for deeper traversal at the cost of latency. Start at 2 and tune upward.

Add versioned compliance policies and gate every decision against them before recording.

from datetime import datetime

from semantica.context import AgentContext, ContextGraph, PolicyEngine
from semantica.context.decision_models import Decision, Policy
from semantica.vector_store import VectorStore

context = AgentContext(
    vector_store=VectorStore(backend="faiss", dimension=768),
    knowledge_graph=ContextGraph(),
    decision_tracking=True,
)

engine = PolicyEngine(graph_store=context.knowledge_graph)

engine.add_policy(Policy(
    policy_id="data_privacy",
    name="Data Privacy",
    description="No PII stored without user consent flag",
    rules={"required_user_consent": True, "max_retention_days": 90},
    category="privacy",
    version="1.2",
    created_at=datetime(2024, 1, 1),
    updated_at=datetime(2024, 1, 1),
))

decision = Decision(
    decision_id="dec_store_email",
    category="data_storage",
    scenario="Store user profile",
    reasoning="User opted in to profile storage",
    outcome="stored",
    confidence=1.0,
    timestamp=datetime.now(),
    decision_maker="profile_agent",
    metadata={"user_consent": True, "retention_days": 30},
)

if engine.check_compliance(decision, "data_privacy"):
    context.record_decision(
        category=decision.category,
        scenario=decision.scenario,
        reasoning=decision.reasoning,
        outcome=decision.outcome,
        confidence=decision.confidence,
    )
else:
    print("Blocked by policy: data_privacy")

AgentContext

AgentContext is the main entry point. Wraps memory, graph, and decision tracking behind a single unified API.

Constructor Parameters

Parameter	Type	Default	Description
`vector_store`	`VectorStore`	required	Backend for embedding-based memory retrieval
`knowledge_graph`	`ContextGraph`	`None`	Enables graph-backed relationships and GraphRAG
`decision_tracking`	`bool`	`False`	Activates `DecisionRecorder`: requires `knowledge_graph` to also be set
`retention_days`	`Optional[int]`	`30`	Auto-expire memories older than N days; `None` = keep forever
`max_memories`	`int`	`10000`	Hard cap before LRU eviction
`graph_expansion`	`bool`	`True`	Auto-expands graph from stored memories
`max_expansion_hops`	`int`	`2`	Max hops for graph expansion during retrieval
`hybrid_alpha`	`float`	`0.5`	Balance between vector (`0.0`) and graph (`1.0`) retrieval
`advanced_analytics`	`bool`	`True`	Enables PageRank, centrality, and community analysis
`kg_algorithms`	`bool`	`True`	Adds path-finding and link prediction

Set retention_days to avoid memory bloat. The default of 30 prunes automatically. Compliance-critical agents may need retention_days=None with explicit archival via export().

Persist your context between runs. VectorStore does not auto-persist; passing index_path= to its constructor is a no-op. Call context.save("agent_state/") to write memory, the vector index, and the graph to disk, and context.load("agent_state/") on the next process to restore them. See the “Persist & Restore” tab under Real-World Patterns below.

Memory Methods

Method	Returns	Description
`store(content, metadata, conversation_id, user_id)`	`str` or `Dict`	Store a fact (str → memory ID) or list of documents (list → stats dict)
`batch_store(items)`	`List[str]`	Store multiple items at once: returns list of memory IDs
`retrieve(query, max_results, min_score, use_graph, conversation_id)`	`List[Dict]`	Semantic retrieval; auto-selects GraphRAG if `knowledge_graph` is set
`forget(memory_id, conversation_id, days_old)`	`int`	Delete memories by ID, conversation, or age
`update(memory_id, content, metadata)`	`bool`	Update content or metadata of a stored memory
`get_memory(memory_id)`	`Optional[Dict]`	Fetch a specific memory by ID
`stats()`	`Dict`	Memory counts, vector store status, graph stats
`health()`	`Dict`	System health: all backends, status flags
`save(path)`	`None`	Persist full context state (memory + graph) to disk
`load(path)`	`None`	Restore context state from disk
`export(conversation_id, format)`	`str \| Dict`	Export memories as JSON or dict
`import_data(data, format)`	`int`	Import memories from JSON or dict

retrieve() uses max_results=, not top_k=. The parameter is max_results (default 5). Pass use_graph=True to force GraphRAG or use_graph=False to force vector-only retrieval regardless of whether a knowledge_graph is configured.

Conversation Methods

# Store turns in a conversation thread
context.store("User asked about deployment options", conversation_id="conv_001")
context.store("Agent recommended Docker + Kubernetes", conversation_id="conv_001")

# Retrieve full conversation history
history = context.conversation("conv_001", max_items=50)
for turn in history:
    print("[{}] {}".format(turn["timestamp"], turn["content"]))

# Retrieve across all conversations with a query
results = context.retrieve(
    "deployment recommendations",
    conversation_id="conv_001",
    max_results=10,
)

Multi-Hop GraphRAG

Requires knowledge_graph to be set at construction: enables query_with_reasoning() for LLM-grounded multi-hop traversal:

import os
from semantica.llms import Groq

llm    = Groq(model="llama-3.3-70b-versatile", api_key=os.getenv("GROQ_API_KEY"))
result = context.query_with_reasoning(
    query="What technologies have we chosen and why?",
    llm_provider=llm,
    max_hops=2,
    max_results=10,
)

print(result["response"])
print("Confidence: {:.2f}".format(result["confidence"]))
print("Sources used: {}".format(result["num_sources"]))

Decision Methods

Method	Returns	Description
`record_decision(category, scenario, reasoning, outcome, confidence, entities, decision_maker, valid_from, valid_until)`	`str`	Record a decision; raises `RuntimeError` if `decision_tracking=False` or no `knowledge_graph`
`find_precedents(scenario, category, limit, use_hybrid_search, max_hops, as_of)`	`List[Decision]`	Find similar past decisions by semantic + structural similarity
`query_decisions(query, max_hops, use_hybrid_search)`	`List[Decision]`	Broad context-aware decision search
`get_causal_chain(decision_id, direction, max_depth)`	`List[Decision]`	Trace `"upstream"` causes or `"downstream"` effects
`trace_decision_explainability(decision_id)`	`Dict`	Full explainability: causes, effects, relationship paths
`get_policy_engine()`	`PolicyEngine`	Access the active `PolicyEngine` instance

decision_tracking=True requires knowledge_graph to also be set. Without it, record_decision() raises RuntimeError.

Use find_precedents() before every significant decision. This is how the context module prevents agents from making contradictory choices across runs. Surface precedents to the LLM as context: “we chose X for similar reasons before.”

Checkpoint Methods

Ideal for auditing reasoning loops: take a snapshot before and after a pass to see exactly what changed:

# Take a named snapshot of the current graph state
context.checkpoint("before_inference")

# ... run reasoning, record decisions ...

context.checkpoint("after_inference")

# See exactly what was added/removed
diff = context.diff_checkpoints("before_inference", "after_inference")
print("Decisions added: {}".format(len(diff["decisions_added"])))
print("Relationships added: {}".format(len(diff["relationships_added"])))

# Persist a checkpoint to disk via TemporalVersionManager
context.flush_checkpoint("after_inference")

ContextGraph

ContextGraph is the knowledge graph backing AgentContext. Can also be used standalone for relationship modelling without the full context layer.

from semantica.context import ContextGraph

graph = ContextGraph(advanced_analytics=True)

# Build the graph
graph.add_node("Python",  "language",  properties={"paradigm": "multi-paradigm"})
graph.add_node("FastAPI", "framework", properties={"language": "Python"})
graph.add_edge("Python", "FastAPI", "enables")

# Record and query decisions directly on the graph
decision_id = graph.record_decision(
    category="technology_choice",
    scenario="Web API framework selection",
    reasoning="FastAPI's async support and auto-docs match our requirements",
    outcome="selected_fastapi",
    confidence=0.92,
    entities=["Python", "FastAPI"],
)

similar = graph.find_precedents_by_scenario("web framework", limit=3)
stats   = graph.stats()
print("Nodes: {}, Edges: {}".format(stats["node_count"], stats["edge_count"]))

Constructor Options

Parameter	Type	Default	Description
`advanced_analytics`	`bool`	`True`	PageRank, betweenness centrality
`centrality_analysis`	`bool`	`True`	Full centrality suite
`community_detection`	`bool`	`True`	Louvain community clustering
`node_embeddings`	`bool`	`True`	Node2Vec embeddings for structural similarity

ContextGraph: Full Method Reference

Method	Returns	Description
`add_node(node_id, node_type, properties, valid_from, valid_until)`	`None`	Add a node; supports temporal validity windows
`add_edge(source_id, target_id, edge_type, weight, properties)`	`None`	Add a directed edge with optional weight
`add_nodes(nodes)`	`int`	Bulk-add from a list of dicts; returns count added
`add_edges(edges)`	`int`	Bulk-add edges; returns count added
`get_neighbors(node_id, hops)`	`List[Dict]`	BFS neighbors up to given depth
`get_neighbor_distances(node_id, hops)`	`List[Dict]`	Neighbors with confidence-decay scoring
`find_node(node_id)`	`Optional[Dict]`	Look up a single node by ID
`find_nodes(node_type, skip, limit)`	`List[Dict]`	Filter nodes by type with pagination
`find_active_nodes(node_type, at_time)`	`List[Dict]`	Nodes that are valid at a given timestamp
`find_edges(edge_type, skip, limit)`	`List[Dict]`	Filter edges by type with pagination
`record_decision(category, scenario, reasoning, outcome, confidence, entities, decision_maker)`	`str`	Add decision node with causal edges
`find_precedents_by_scenario(scenario, category, limit, use_semantic_search, as_of)`	`List[Dict]`	Semantically similar past scenarios
`query(query, skip, limit)`	`List[Dict]`	Full-text search over node content
`stats()`	`Dict`	Node/edge counts, type breakdowns, graph density
`density()`	`float`	Graph density score
`save_to_file(path, format="json")`	`None`	Persist graph as JSON or a Markdown directory
`load_from_file(path, format="json")`	`None`	Replace graph state from JSON or a Markdown directory
`build_from_conversations(conversations, link_entities)`	`Dict`	Build graph from conversation data
`link_graph(other_graph, source_node_id, target_node_id, link_type)`	`str`	Create cross-graph navigation link; returns `link_id`
`navigate_to(link_id)`	`Tuple`	Follow a cross-graph link to `(target_graph, target_node_id)`
`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

Distance Intelligence (v0.5.0)

ContextGraph exposes a full Distance Intelligence API for exploring semantic neighborhoods and blending proximity into retrieval.

Full Distance Intelligence reference (distance matrices, API endpoints, embedding cache, Explorer UI) is covered in the dedicated Distance Intelligence page. This section documents the context-layer API.

Neighbors with Distance Metadata

Pass include_distance_metadata=True to get_neighbors() to receive distance band, confidence decay, and path information alongside every neighbor:

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:

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,
)

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.

Link multiple independent ContextGraph instances so agents can traverse across problem spaces:

domain_graph    = ContextGraph()
decision_graph  = ContextGraph()

domain_graph.add_node("microservices", "architecture", properties={"style": "distributed"})
decision_graph.add_node("deploy_k8s",  "decision",     properties={"outcome": "approved"})

link_id = domain_graph.link_graph(
    other_graph=decision_graph,
    source_node_id="microservices",
    target_node_id="deploy_k8s",
    link_type="INFORMED_BY",
)

# Follow the link at traversal time
target_graph, entry_node = domain_graph.navigate_to(link_id)

# Cross-graph pathfinding
path = domain_graph.cross_graph_path(
    source_node_id="microservices",
    target_graph=decision_graph,
    target_node_id="deploy_k8s",
    max_hops=5,
)
print("Reachable: {}, hops: {}".format(path["reachable"], path["hop_count"]))

AgentMemory

For fine-grained control over memory storage and retrieval:

from semantica.context import AgentMemory
from semantica.vector_store import VectorStore

memory = AgentMemory(
    vector_store=VectorStore(backend="faiss", dimension=768),
    max_memory_size=10000,
    retention_policy="90_days",   # or "unlimited"
)

memory_id = memory.store(
    "Critical compliance rule: all trades must be pre-approved",
    metadata={"type": "compliance"},
)

results = memory.retrieve(
    query="trade approval requirements",
    max_results=5,
    min_score=0.0,
)

memory.delete_memory(memory_id)
memory.clear_memory(conversation_id="conv_001")

history = memory.get_conversation_history(conversation_id="conv_001", max_items=100)

Parameter	Type	Default	Description
`vector_store`	`VectorStore`	required	Embedding backend for semantic retrieval
`max_memory_size`	`int`	`10000`	Max items before LRU eviction
`retention_policy`	`str`	`"unlimited"`	`"N_days"` (e.g. `"30_days"`) or `"unlimited"`

Markdown Round Trips

AgentMemory can export human-editable Markdown and import the edited files back. Each file contains one memory item, with required metadata in YAML frontmatter and the memory content in the Markdown body:

---
id: mem_compliance_rule
created_at: '2026-07-22T09:00:00+00:00'
updated_at: '2026-07-22T10:30:00+00:00'
type: compliance
tags:
- trading
- approval
---

All trades must be pre-approved.

from pathlib import Path

# A single selected memory can be returned as Markdown text.
document = memory.export(format="markdown", type="compliance")

# Export a memory set as one stable Markdown file per item.
memory.export(format="markdown", destination="memory_export/")

# New IDs create memories; existing IDs are updated in place.
count = memory.import_data(Path("memory_export/"), format="markdown")

The required frontmatter fields are id, created_at, updated_at, and either type or kind. Optional metadata can be edited at the top level. Imports reject malformed or duplicate fields before changing memory, and re-importing unchanged files is idempotent. Memory-local entities and relationships are preserved as provenance but are not applied to ContextGraph by Markdown import. Use a dedicated export directory: matching files are overwritten, but unrelated or stale Markdown files are not deleted automatically. Export refuses to overwrite filesystem links and uses atomic file replacement; import also refuses symlinks, Windows directory junctions, and other Windows reparse points. Timestamp offsets are preserved in Markdown and normalized to UTC only for comparisons, so aware and local-naive records can be queried together safely. Vector-store writes are deferred until the in-memory import commits; adapter synchronization remains best-effort and logs failures.

ErasureCoordinator

ContextGraph.purge_node() is scoped to one graph: the node is removed and a tombstone is written, but the same content can still be live as an AgentMemory item and as an embedding in the vector store. ErasureCoordinator drives the cascade across every bound store and returns an ErasureReceipt recording what each one reported.

from semantica.context import AgentMemory, ContextGraph, ErasureCoordinator

coordinator = ErasureCoordinator(graph=graph, memory=memory)

receipt = coordinator.erase_entity(
    "customer-4471",
    reason="GDPR Art. 17 request #882",
)

if not receipt.complete:
    # These stores may still hold the entity; handle them out of band.
    print(receipt.incomplete_stores)

Check the receipt. The call returning is not proof the data is gone. FAISS, Milvus, and Weaviate expose no delete method, so erasure cannot be completed on those backends today; the receipt reports unsupported rather than a success it did not achieve.

Constructor Parameters

Parameter	Type	Default	Description
`graph`	`ContextGraph`	`None`	Anything exposing `purge_node()`
`memory`	`AgentMemory`	`None`	Anything exposing `find_by_entity()` and `batch_delete()`
`vector_store`	`VectorStore`	`memory.vector_store`	Store holding entity-keyed embeddings; pass `False` to disable the leg

At least one store is required; a store that is not supplied reports not_configured rather than being silently skipped.

Methods

Method	Returns	Description
`erase_entity(entity_id, reason, at, vector_ids)`	`ErasureReceipt`	Erase one entity from every bound store
`erase_entities(entity_ids, reason, at)`	`List[ErasureReceipt]`	One receipt per entity, in order; one failure does not stop the rest

Store Statuses

Status	Meaning
`erased`	Reached, data removed. On the vectors leg this means the store accepted the delete for the ids given; backends offer no portable existence check, so it is not a count of embeddings that were really there
`not_found`	Reached, held nothing for this entity
`not_configured`	No such store was bound: normal, not a failure
`unsupported`	The store cannot delete at all; retrying will not help
`failed`	The store was reached and the deletion did not succeed

ErasureReceipt

Member	Type	Description
`entity_id`	`str`	Entity the erasure was requested for
`reason`	`Optional[str]`	Recorded in the receipt and the graph tombstone
`erased_at`	`str`	ISO-8601; matches the tombstone’s `purged_at`
`stores`	`Dict[str, Dict]`	Per-store outcome keyed `vectors`, `memory`, `graph`
`complete`	`bool`	`False` when any store reports `unsupported` or `failed`
`incomplete_stores`	`List[str]`	Stores that may still hold the entity’s data
`to_dict()`	`Dict`	Serialized receipt, safe to persist as an audit record

receipt.to_dict()
# {
#   "entity_id": "customer-4471",
#   "reason": "GDPR Art. 17 request #882",
#   "erased_at": "2026-08-16T09:03:36.813220",
#   "complete": False,
#   "stores": {
#     "vectors": {"status": "unsupported", "backend": "faiss",
#                 "detail": "backend exposes no delete()/delete_vectors(); ..."},
#     "memory":  {"status": "erased", "items": 14},
#     "graph":   {"status": "erased", "nodes": 1, "edges": 3},
#   },
# }

Erasure runs outward-in: vectors, then memory, then the graph. The tombstone is the durable attestation that an erasure happened, so it is written last: a crash mid-cascade leaves the node present and the receipt incomplete, rather than a tombstone claiming more than actually happened. A store that raises is recorded as failed and the remaining stores are still erased. Erasing the same entity twice returns a receipt saying there was nothing left to do rather than raising.

PolicyEngine

PolicyEngine manages versioned policies stored in the knowledge graph. Policies are stored as nodes and can be linked to decisions:

from semantica.context import PolicyEngine
from semantica.context import ContextGraph
from semantica.context.decision_models import Policy, Decision
from datetime import datetime

graph  = ContextGraph()
policy = PolicyEngine(graph_store=graph)

# Create and store a policy
p = Policy(
    policy_id="policy_001",
    name="Confidence Threshold Policy",
    description="All decisions must have confidence >= 0.7",
    rules={"min_confidence": 0.7, "requires_reasoning": True},
    category="decision_quality",
    version="1.0",
    created_at=datetime.now(),
    updated_at=datetime.now(),
)
policy.add_policy(p)

# Check compliance of a specific decision
decision = Decision(
    decision_id="dec_001",
    category="loan_approval",
    scenario="First-time homebuyer",
    reasoning="Good credit score and stable employment",
    outcome="approved",
    confidence=0.94,
    timestamp=datetime.now(),
    decision_maker="loan_agent",
)
compliant = policy.check_compliance(decision, "policy_001")
print("Compliant:", compliant)

# Get applicable policies for a category
policies = policy.get_applicable_policies(category="decision_quality")
for p in policies:
    print("{} v{}".format(p.name, p.version))

EntityLinker

Maps entity text to URIs and creates typed links between entity IDs:

from semantica.context import EntityLinker

linker = EntityLinker(similarity_threshold=0.8)

# Assign a URI to an entity
uri = linker.assign_uri("apple_inc", "Apple Inc.", "ORGANIZATION")
print(uri)  # "https://semantica.dev/entity/apple_inc.#organization"

# Link entities from extracted text
entities = [
    {"id": "e1", "text": "Apple Inc.", "type": "ORGANIZATION"},
    {"id": "e2", "text": "Apple",      "type": "ORGANIZATION"},
]
linked = linker.link(text="Apple Inc. was founded by Steve Jobs.", entities=entities)
for e in linked:
    print("{} → {}  (confidence: {:.2f})".format(e.text, e.uri, e.confidence))

# Explicitly link two entity IDs (not a list: takes two IDs)
linker.link_entities(
    entity1_id="apple_inc",
    entity2_id="aapl",
    link_type="same_as",
    confidence=0.99,
)

# Build the full entity web
web = linker.build_entity_web()
print("Entities:", web["statistics"]["total_entities"])
print("Links:   ", web["statistics"]["total_links"])

EntityLinker.link_entities() links two entity IDs, not a list. Call link_entities(entity1_id, entity2_id, link_type) to create a typed edge between two known IDs. For linking entities extracted from text, use link(text, entities=[...]) instead.

LinkedEntity fields returned by link():

Field	Type	Description
`entity_id`	`str`	Entity identifier
`uri`	`str`	Generated URI (e.g. `"https://semantica.dev/entity/apple_inc."`)
`text`	`str`	Surface form text
`type`	`str`	Entity type
`linked_entities`	`List[EntityLink]`	Related entity links with `source_entity_id`, `target_entity_id`, `link_type`, `confidence`
`context`	`Dict`	Entity metadata
`confidence`	`float`	Overall confidence score

ContextRetriever

Hybrid retrieval combining vector similarity, graph traversal, and memory:

from semantica.context import ContextRetriever

retriever = ContextRetriever(
    memory_store=memory,
    knowledge_graph=context_graph,
    vector_store=vector_store,
    use_graph_expansion=True,
    max_expansion_hops=2,
    hybrid_alpha=0.5,
)

results = retriever.retrieve(
    query="What decisions were made about cloud infrastructure?",
    max_results=10,
    use_graph_expansion=True,
    min_relevance_score=0.3,
)

for r in results:
    print("[{}] score={:.3f}: {}".format(r.source, r.score, r.content[:80]))

Support-Aware Retrieval (`truth_filter`)

Cached vector or memory content can outlive the evidence that originally supported it. ContextRetriever.retrieve(..., truth_filter=...) accepts an opt-in TruthMaintenanceContextFilter that removes candidates whose declared dependencies are no longer supported, before ranking, while the stored records themselves are never modified. The example below runs standalone with no model, database, or network — only semantica itself:

from semantica.context import ContextRetriever, TruthMaintenanceContextFilter
from semantica.reasoning import FactSupport, Rule, TruthMaintenanceSession

# A tiny read-only store standing in for any real vector backend.
class StaticVectorStore:
    def __init__(self, rows):
        self.rows = [dict(row) for row in rows]

    def search(self, *, query, limit):
        return self.rows[:limit]

def annotated_row(identifier, fact, score, text):
    """One record whose content depends on one live fact."""
    return {
        "id": identifier,
        "score": score,
        "content": text,
        "metadata": {
            "truth_maintenance": {
                "schema_version": 1,
                "session_id": "employment-session-1",
                "required_facts": [fact],
                "required_support_ids": [],
            }
        },
    }

rules = [
    Rule(
        rule_id="employment",
        name="employment",
        conditions=["Employed(?x)"],
        conclusion="Eligible(?x)",
    ),
]
session = TruthMaintenanceSession(rules=rules)
session.apply(assertions=[FactSupport("v1", "Employed(Alice)")])

store = StaticVectorStore([
    annotated_row("cached", "Eligible(Alice)", 0.9, "Alice is eligible"),
])
retriever = ContextRetriever(vector_store=store, use_graph_expansion=False)
gate = TruthMaintenanceContextFilter(session, session_id="employment-session-1")

checked = retriever.retrieve("eligibility", truth_filter=gate)
assert [result.content for result in checked] == ["Alice is eligible"]
assert (
    checked[0].metadata["truth_maintenance_validation"]["version"]
    == session.version
)

# The supporting evidence is withdrawn. The vector record stays stored,
# but it no longer reaches the caller through the checked retrieval.
session.apply(retractions=["v1"])
checked = retriever.retrieve("eligibility", truth_filter=gate)
assert checked == []
assert store.rows[0]["content"] == "Alice is eligible"  # record untouched

Manual annotation semantics

The filter only validates what the application explicitly wrote into store metadata. It never infers dependencies, and it never annotates records on your behalf:

Each candidate must carry metadata["truth_maintenance"] with exactly four keys: schema_version (currently 1), session_id (must equal the filter’s session_id), required_facts, and required_support_ids. At least one fact or support ID must be declared.
A candidate is kept only if every entry of required_facts is in the snapshot’s facts and every entry of required_support_ids is in the snapshot’s active supports. Unannotated candidates and candidates with malformed annotations are removed whole — a record ID or node ID is never a trust signal.

Source-specific citations

Two records may assert the same conclusion but cite different sources. required_support_ids binds each record to the evidence it actually used, so withdrawing one support invalidates only the records citing it, while records backed by an alternative derivation or an explicit support are kept.

Graph bundles

For candidates retrieved from a ContextGraph, the root record and every member of related_entities / related_relationships must carry the same annotation shape on their own metadata. If the root or any attachment is unsupported, the entire candidate is excluded — a result never keeps its prose while silently dropping a stale attachment.

Behavior details

Filtering happens before ranking and before the legacy merge: a high-scoring stale record cannot crowd out valid lower-scoring records, and dependency metadata survives merging.
Survivors are returned as deep copies stamped with metadata["truth_maintenance_validation"] (session_id, version). Original candidates and stored records are never modified.
Each call validates against one immutable snapshot; if the session commits a new version while retrieval is in flight, retrieve raises ProcessingError instead of returning mixed-version results.
Vector, memory, and graph sources are filtered in the same pass: construct the retriever with memory_store= or knowledge_graph= and pass truth_filter= to retrieve, or to the search, vector_search, and graph_search delegates (they forward to retrieve). memory_search bypasses retrieve and does not apply the filter — use retrieve when memory results must be grounded.

Unsupported entry points

The filter is supported only by ContextRetriever.retrieve(mode="local", ...) (the default mode) and delegates that use that path. Passing truth_filter to retrieve with mode="global", "drift", or "hybrid" raises ValidationError before retrieval starts. Those modes and direct retrieve_global / retrieve_drift calls are outside the dependency-validation guarantee. truth_filter is intentionally not accepted on AgentContext.retrieve, AgentContext.query_with_reasoning, or ContextRetriever.query_with_reasoning — passing it there raises ValidationError immediately. Assemble the verified context yourself with ContextRetriever.retrieve(..., truth_filter=...) and feed it to the model. See Truth Maintenance for the session model behind the filter.

Data Structures

Decision

@dataclass
class Decision:
    decision_id:          str
    category:             str
    scenario:             str
    reasoning:            str
    outcome:              str
    confidence:           float               # 0.0 - 1.0
    timestamp:            datetime
    decision_maker:       str
    reasoning_embedding:  Optional[List[float]]  # generated embedding
    node2vec_embedding:   Optional[List[float]]  # structural embedding
    valid_from:           Optional[str]       # ISO datetime
    valid_until:          Optional[str]       # ISO datetime
    metadata:             Dict[str, Any]

Precedent

@dataclass
class Precedent:
    precedent_id:        str
    source_decision_id:  str
    similarity_score:    float               # 0-1 match score
    relationship_type:   str                 # "similar_scenario" | "same_policy" | "exception_precedent"
    metadata:            Dict[str, Any]

Policy

@dataclass
class Policy:
    policy_id:    str
    name:         str
    description:  str
    rules:        Dict[str, Any]    # rule definitions
    category:     str
    version:      str               # e.g. "1.0", "2.1"
    created_at:   datetime
    updated_at:   datetime
    metadata:     Dict[str, Any]

PolicyException

@dataclass
class PolicyException:
    exception_id:        str
    decision_id:         str
    policy_id:           str
    reason:              str
    approver:            str
    approval_timestamp:  datetime
    justification:       str
    metadata:            Dict[str, Any]

ApprovalChain

@dataclass
class ApprovalChain:
    approval_id:       str
    decision_id:       str
    approver:          str
    approval_method:   str          # "slack_dm" | "zoom_call" | "email" | "system"
    approval_context:  str
    timestamp:         datetime
    metadata:          Dict[str, Any]

LinkedEntity

@dataclass
class LinkedEntity:
    entity_id:       str
    uri:             str
    text:            str
    type:            str
    linked_entities: List[EntityLink]
    context:         Dict[str, Any]
    confidence:      float

@dataclass
class EntityLink:
    source_entity_id:  str
    target_entity_id:  str
    link_type:         str          # "same_as" | "related_to" | "part_of"
    confidence:        float
    source:            Optional[str]
    metadata:          Dict[str, Any]

Real-World Patterns

Healthcare: Treatment Decisions
Finance: Loan Decisions
Persist & Restore

from semantica.context import AgentContext, ContextGraph
from semantica.vector_store import VectorStore

health_agent = AgentContext(
    vector_store=VectorStore(backend="faiss", dimension=768),
    knowledge_graph=ContextGraph(),
    decision_tracking=True,
)

health_agent.store("Patient has hypertension, type 2 diabetes")
health_agent.store("Patient allergic to penicillin: verified 2024-01")

decision_id = health_agent.record_decision(
    category="treatment_plan",
    scenario="Hypertension with comorbid diabetes",
    reasoning="ACE inhibitors are renoprotective in diabetic patients",
    outcome="prescribed_lisinopril",
    confidence=0.91,
)

precedents = health_agent.find_precedents("hypertension diabetes", limit=5)
for p in precedents:
    print("Past: {}  (confidence: {:.2f})".format(p.outcome, p.confidence))

chain = health_agent.get_causal_chain(decision_id, direction="downstream")
print("Follow-up decisions triggered: {}".format(len(chain)))

from semantica.context import AgentContext, ContextGraph, PolicyEngine
from semantica.context.decision_models import Policy, Decision
from semantica.vector_store import VectorStore
from datetime import datetime

graph  = ContextGraph()
policy = PolicyEngine(graph_store=graph)

# Add compliance policy
p = Policy(
    policy_id="lending_policy",
    name="Lending Policy",
    description="Min confidence 0.8 for loan decisions",
    rules={"min_confidence": 0.8},
    category="loan_approval",
    version="1.0",
    created_at=datetime.now(),
    updated_at=datetime.now(),
)
policy.add_policy(p)

loan_agent = AgentContext(
    vector_store=VectorStore(backend="faiss", dimension=768),
    knowledge_graph=graph,
    decision_tracking=True,
)
loan_agent.store("Applicant: credit score 750, DTI 28%, stable employment 4yr")

# Check compliance before recording
d = Decision(
    decision_id="dec_loan_001",
    category="loan_approval",
    scenario="First-time homebuyer: 30yr fixed, 20% down",
    reasoning="Credit score above threshold, DTI within limits",
    outcome="approved_300k",
    confidence=0.94,
    timestamp=datetime.now(),
    decision_maker="loan_agent",
)
compliant = policy.check_compliance(d, "lending_policy")
if compliant:
    loan_agent.record_decision(
        category=d.category,
        scenario=d.scenario,
        reasoning=d.reasoning,
        outcome=d.outcome,
        confidence=d.confidence,
    )

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,
)

context.store("Important fact learned during session")
context.record_decision(
    category="ops", scenario="Scale up", reasoning="Load > 80%",
    outcome="scaled_to_10_replicas", confidence=0.97,
)

# Persist everything
context.save("agent_state/")

# Later: restore and continue
restored = AgentContext(
    vector_store=VectorStore(backend="faiss", dimension=768),
    knowledge_graph=ContextGraph(),
    decision_tracking=True,
)
restored.load("agent_state/")

results = restored.retrieve("load scaling decisions", max_results=3)

Vector Store: embedding storage backend for memory retrieval.
Knowledge Graph: graph algorithms and analytics used inside ContextGraph.
Reasoning: logical inference layered on top of context.
Provenance: W3C PROV-O lineage for every stored fact.
Context Module: memory and decision tracking · Intermediate
Advanced Context Engineering: production FAISS + Neo4j setup · Advanced

​Exported Classes

​What You Get

​Quick Start

​Usage Patterns

​AgentContext

​Constructor Parameters

​Memory Methods

​Conversation Methods

​Multi-Hop GraphRAG

​Decision Methods

​Checkpoint Methods

​ContextGraph

​Constructor Options

​ContextGraph: Full Method Reference

​Distance Intelligence (v0.5.0)

​Neighbors with Distance Metadata

​Proximity-Blended Retrieval

​Cross-Graph Navigation

​AgentMemory

​Markdown Round Trips

​ErasureCoordinator

​Constructor Parameters

​Methods

​Store Statuses

​ErasureReceipt

​PolicyEngine

​EntityLinker

​ContextRetriever

​Support-Aware Retrieval (truth_filter)

​Manual annotation semantics

​Source-specific citations

​Graph bundles

​Behavior details

​Unsupported entry points

​Data Structures

​Real-World Patterns

Exported Classes

What You Get

Quick Start

Usage Patterns

AgentContext

Constructor Parameters

Memory Methods

Conversation Methods

Multi-Hop GraphRAG

Decision Methods

Checkpoint Methods

ContextGraph

Constructor Options

ContextGraph: Full Method Reference

Distance Intelligence (v0.5.0)

Neighbors with Distance Metadata

Proximity-Blended Retrieval

Cross-Graph Navigation

AgentMemory

Markdown Round Trips

ErasureCoordinator

Constructor Parameters

Methods

Store Statuses

ErasureReceipt

PolicyEngine

EntityLinker

ContextRetriever

Support-Aware Retrieval (`truth_filter`)

Manual annotation semantics

Source-specific citations

Graph bundles

Behavior details

Unsupported entry points

Data Structures

Real-World Patterns