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. It stores facts with provenance, records decisions as first-class objects with full causal chains, lets agents search their own history to stay consistent across runs, and answers complex queries by traversing the knowledge graph.

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)

What You Get

Unified interface for memory, decision tracking, graph-backed retrieval, conversation history, checkpoints, and persistence. Thread-safe in-memory knowledge graph with centrality analysis, community detection, temporal validity, cross-graph links, and decision management. Embedding-backed memory with retention policy and LRU eviction. Records decisions with causal chains, confidence scores, temporal validity windows, and cross-system context capture. Manages policy versions, checks compliance for recorded decisions, and tracks policy exceptions in the graph. Maps entity text to URIs and creates typed links between entity IDs — prevents "Apple", "Apple Inc.", and "AAPL" from becoming three separate nodes. Hybrid retrieval fusing vector similarity, graph traversal, and agent memory for richer context than pure vector search. Traces upstream causes and downstream effects of any decision through the knowledge graph.

Getting Started

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,   # requires knowledge_graph to be set
)

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

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

# Record a decision
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",
)

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

AgentContext

The main entry point. Wraps memory, graph, and decision tracking behind a single 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

`decision_tracking=True` has no effect unless `knowledge_graph` is also provided. Both must be set at construction time for decision tracking to be active.

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

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:

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

Checkpoint Methods

Useful for detecting what changed across reasoning runs:

# 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

The knowledge graph backing AgentContext. Can also be used standalone for relationship modelling.

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)	None	Persist graph to JSON
load_from_file(path)	None	Load graph from JSON
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

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 (Low-Level)

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"

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

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

Tips and Common Pitfalls

**`decision_tracking=True` silently does nothing without `knowledge_graph`.** Both must be set at construction. Passing only `decision_tracking=True` without a `knowledge_graph` instance leaves the decision backend uninitialised — `record_decision()` will raise `RuntimeError`. **Persist your vector store between runs.** Pass `index_path="context.faiss"` to `VectorStore` — without it the FAISS index lives only in memory and is lost on shutdown. **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." **`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. **Set `retention_days` to avoid memory bloat.** The default `AgentContext.retention_days=30` prunes automatically. Compliance-critical agents may need `retention_days=None` with explicit archival via `export()`. **Use `checkpoint()` + `diff_checkpoints()` to audit reasoning loops.** Take a snapshot before and after a reasoning pass to see exactly which decisions and relationships were added. **`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. Embedding storage backend for memory retrieval. Graph algorithms and analytics used inside ContextGraph. Logical inference layered on top of context. W3C PROV-O lineage for every stored fact.

Cookbooks

• Context Module — memory and decision tracking · Intermediate
• Advanced Context Engineering — production FAISS + Neo4j setup · Advanced