--- title: "Context Module" description: "Agent context graphs, decision tracking, causal chains, precedent search, policy enforcement, and multi-hop GraphRAG." icon: "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 ```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, # 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 ```python # 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: ```python 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: ```python # 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. ```python 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: ```python 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: ```python 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: ```python 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: ```python 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: ```python 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 ```python @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] ``` ```python @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] ``` ```python @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] ``` ```python @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] ``` ```python @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] ``` ```python @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](https://github.com/semantica-agi/semantica/blob/main/cookbook/introduction/19_Context_Module.ipynb) — memory and decision tracking · Intermediate - [Advanced Context Engineering](https://github.com/semantica-agi/semantica/blob/main/cookbook/advanced/11_Advanced_Context_Engineering.ipynb) — production FAISS + Neo4j setup · Advanced