semantica/docs/reference/context.md

title, description, icon

title	description	icon
Context Module	Agent context graphs, decision tracking, causal chains, precedent search, policy enforcement, and multi-hop GraphRAG.	brain

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

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

```python 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"]))
```

```python 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", ) ``` ```python # 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

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

```python
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"]))
```

<Check>
  Swap `backend="faiss"` to `backend="inmemory"` for zero-dependency local development.
</Check>

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

```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,  # 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.

```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),
    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"]))
```

<Tip>
  Increase `max_expansion_hops` for deeper traversal at the cost of latency. Start at 2 and tune upward.
</Tip>

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

```python
from semantica.context import AgentContext, ContextGraph, PolicyEngine
from semantica.vector_store import VectorStore

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

engine = PolicyEngine(knowledge_graph=context.knowledge_graph)

engine.add_policy(
    name="data_privacy",
    description="No PII stored without user consent flag",
    version="1.2",
    effective_date="2024-01-01",
    category="privacy",
    rules={"requires_consent": True, "max_retention_days": 90},
)

decision_data = {"action": "store_user_email", "user_consent": True}
result = engine.check_compliance(decision_data, policy_names=["data_privacy"])

if result["compliant"]:
    context.record_decision(
        category="data_storage",
        scenario="Store user profile",
        outcome="stored",
        confidence=1.0,
    )
else:
    print("Blocked by policy:", result["violations"])
```

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](#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](distance) 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.

Cross-Graph Navigation

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

Data Structures

@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]

@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]

@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]

@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]

@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]

@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

```python 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)))
```

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

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

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