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

ClassRole
AgentContextPrimary entry point: memory, retrieval, decisions, graph traversal, checkpoints
ContextGraphIn-memory knowledge graph with centrality, community detection, and decision tracking
AgentMemoryVector-backed persistent memory: store(text), retrieve(query, max_results)
EntityLinkerLink entity mentions to URIs; create typed edges between entity IDs
ContextRetrieverHybrid vector + graph retrieval with min-score and graph expansion options
DecisionRecorderRecord decisions with embeddings, causal chains, and metadata
PolicyEnginePolicy management: add_policy(), check_compliance(), get_applicable_policies()
CausalChainAnalyzerTrace how decisions influenced each other: get_causal_chain(decision_id)

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

1

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

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

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

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

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.

AgentContext

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

Constructor Parameters

ParameterTypeDefaultDescription
vector_storeVectorStorerequiredBackend for embedding-based memory retrieval
knowledge_graphContextGraphNoneEnables graph-backed relationships and GraphRAG
decision_trackingboolFalseActivates DecisionRecorder: requires knowledge_graph to also be set
retention_daysOptional[int]30Auto-expire memories older than N days; None = keep forever
max_memoriesint10000Hard cap before LRU eviction
graph_expansionboolTrueAuto-expands graph from stored memories
max_expansion_hopsint2Max hops for graph expansion during retrieval
hybrid_alphafloat0.5Balance between vector (0.0) and graph (1.0) retrieval
advanced_analyticsboolTrueEnables PageRank, centrality, and community analysis
kg_algorithmsboolTrueAdds 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 vector store between runs. Pass index_path="context.faiss" to VectorStore so the FAISS index survives process restarts.

Memory Methods

MethodReturnsDescription
store(content, metadata, conversation_id, user_id)str or DictStore 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)intDelete memories by ID, conversation, or age
update(memory_id, content, metadata)boolUpdate content or metadata of a stored memory
get_memory(memory_id)Optional[Dict]Fetch a specific memory by ID
stats()DictMemory counts, vector store status, graph stats
health()DictSystem health: all backends, status flags
save(path)NonePersist full context state (memory + graph) to disk
load(path)NoneRestore context state from disk
export(conversation_id, format)str | DictExport memories as JSON or dict
import_data(data, format)intImport 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

MethodReturnsDescription
record_decision(category, scenario, reasoning, outcome, confidence, entities, decision_maker, valid_from, valid_until)strRecord 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)DictFull explainability: causes, effects, relationship paths
get_policy_engine()PolicyEngineAccess 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

ParameterTypeDefaultDescription
advanced_analyticsboolTruePageRank, betweenness centrality
centrality_analysisboolTrueFull centrality suite
community_detectionboolTrueLouvain community clustering
node_embeddingsboolTrueNode2Vec embeddings for structural similarity

ContextGraph: Full Method Reference

MethodReturnsDescription
add_node(node_id, node_type, properties, valid_from, valid_until)NoneAdd a node; supports temporal validity windows
add_edge(source_id, target_id, edge_type, weight, properties)NoneAdd a directed edge with optional weight
add_nodes(nodes)intBulk-add from a list of dicts; returns count added
add_edges(edges)intBulk-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)strAdd 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()DictNode/edge counts, type breakdowns, graph density
density()floatGraph density score
save_to_file(path)NonePersist graph to JSON
load_from_file(path)NoneLoad graph from JSON
build_from_conversations(conversations, link_entities)DictBuild graph from conversation data
link_graph(other_graph, source_node_id, target_node_id, link_type)strCreate cross-graph navigation link; returns link_id
navigate_to(link_id)TupleFollow a cross-graph link to (target_graph, target_node_id)
cross_graph_path(source_node_id, target_graph, target_node_id, max_hops)DictShortest path across linked graphs
clear()NoneReset 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

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)
ParameterTypeDefaultDescription
vector_storeVectorStorerequiredEmbedding backend for semantic retrieval
max_memory_sizeint10000Max items before LRU eviction
retention_policystr"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 (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():
FieldTypeDescription
entity_idstrEntity identifier
uristrGenerated URI (e.g. "https://semantica.dev/entity/apple_inc.")
textstrSurface form text
typestrEntity type
linked_entitiesList[EntityLink]Related entity links with source_entity_id, target_entity_id, link_type, confidence
contextDictEntity metadata
confidencefloatOverall 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

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

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