mengzhx/semantica
1
0
Fork 0
mirror of https://github.com/semantica-agi/semantica.git synced 2026-08-29 04:26:20 +00:00
Code Issues Packages Projects Releases Wiki Activity

context_fixes

Browse Source
This commit is contained in:
KaifAhmad1
2026-02-14 17:13:38 +05:30
parent 2df5edf30a
commit 4cd3ef9aa8
12 changed files with 845 additions and 277 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+31
View File
@@ -734,6 +734,37 @@ reasoned_result = context.query_with_reasoning(
- **ContextRetriever**: Performs hybrid retrieval combining vector search, graph traversal, and memory for optimal context relevance
- **AgentContext**: High-level interface integrating Context Graph and Context Retriever for GraphRAG applications
#### Context Graphs: Decision Tracking
```python
from semantica.context import AgentContext, ContextGraph
from semantica.vector_store import VectorStore
context = AgentContext(
vector_store=VectorStore(backend="inmemory", dimension=128),
knowledge_graph=ContextGraph(),
enable_decision_tracking=True,
enable_kg_algorithms=False, # semantic-only precedent search
)
decision_id = context.record_decision(
category="credit_approval",
scenario="High-risk credit limit increase",
reasoning="Recent velocity-check failure and prior fraud flag",
outcome="rejected",
confidence=0.78,
entities=["customer:jessica_norris"],
)
precedents = context.find_precedents(
scenario="High-risk customer credit increase",
category="credit_approval",
limit=5,
)
```
Runnable script: `examples/context_graphs_decision_tracking.py`
**Core Notebooks:**
- [**Context Module Introduction**](https://github.com/Hawksight-AI/semantica/tree/main/cookbook/introduction/19_Context_Module.ipynb) - Basic memory and storage.
- [**Advanced Context Engineering**](https://github.com/Hawksight-AI/semantica/tree/main/cookbook/advanced/11_Advanced_Context_Engineering.ipynb) - Hybrid retrieval, graph builders, and custom memory policies.
semantica/context/agent_context.py
+173 -76
View File
@@ -209,62 +209,52 @@ class AgentContext:
if knowledge_graph and hasattr(knowledge_graph, "build_from_conversations"):
self._graph_builder = knowledge_graph
# Store config
self.config = {
self.config.update({
"retention_days": retention_days,
"max_memories": max_memories,
"use_graph_expansion": use_graph_expansion,
"max_expansion_hops": max_expansion_hops,
"hybrid_alpha": hybrid_alpha,
"enable_decision_tracking": enable_decision_tracking,
}
})
# Initialize decision tracking components if enabled
self._decision_backend = None
self._decision_recorder = None
self._decision_query = None
self._causal_analyzer = None
self._policy_engine = None
if enable_decision_tracking and knowledge_graph:
# Validate that knowledge_graph supports required GraphStore interface
if not hasattr(knowledge_graph, 'execute_query'):
self.logger.error(
"Decision tracking requires a GraphStore-compatible knowledge graph with execute_query() method. "
"Provided knowledge_graph type does not support Cypher queries. "
"Use GraphStore (Neo4j, FalkorDB) or disable decision tracking."
)
raise ValueError(
"Decision tracking requires a GraphStore-compatible knowledge graph. "
"The provided knowledge_graph does not have an execute_query() method. "
"For decision tracking, use a GraphStore backend (Neo4j, FalkorDB) "
"or set enable_decision_tracking=False."
)
# Initialize enhanced decision tracking components
try:
self._decision_recorder = DecisionRecorder(knowledge_graph)
# Enhanced DecisionQuery with KG and vector store integration
self._decision_query = DecisionQuery(
graph_store=knowledge_graph,
vector_store=vector_store if enable_vector_store_features else None,
enable_advanced_analytics=enable_advanced_analytics,
enable_centrality_analysis=enable_kg_algorithms,
enable_community_detection=enable_kg_algorithms,
enable_node_embeddings=enable_kg_algorithms
)
self._causal_analyzer = CausalChainAnalyzer(knowledge_graph)
if hasattr(knowledge_graph, "execute_query"):
self._decision_backend = "graph_store"
try:
self._decision_recorder = DecisionRecorder(knowledge_graph)
self._decision_query = DecisionQuery(
graph_store=knowledge_graph,
vector_store=vector_store if enable_vector_store_features else None,
enable_advanced_analytics=enable_advanced_analytics,
enable_centrality_analysis=enable_kg_algorithms,
enable_community_detection=enable_kg_algorithms,
enable_node_embeddings=enable_kg_algorithms
)
self._causal_analyzer = CausalChainAnalyzer(knowledge_graph)
self._policy_engine = PolicyEngine(knowledge_graph)
self.logger.info("Enhanced decision tracking components initialized successfully")
except Exception as e:
self.logger.warning(f"Failed to initialize enhanced decision tracking: {e}")
self._decision_recorder = DecisionRecorder(knowledge_graph)
self._decision_query = DecisionQuery(knowledge_graph)
self._causal_analyzer = CausalChainAnalyzer(knowledge_graph)
self._policy_engine = PolicyEngine(knowledge_graph)
else:
self._decision_backend = "context_graph"
self._policy_engine = PolicyEngine(knowledge_graph)
self.logger.info("Enhanced decision tracking components initialized successfully")
except Exception as e:
self.logger.warning(f"Failed to initialize enhanced decision tracking: {e}")
# Fallback to basic components
self._decision_recorder = DecisionRecorder(knowledge_graph)
self._decision_query = DecisionQuery(knowledge_graph)
self._causal_analyzer = CausalChainAnalyzer(knowledge_graph)
self._policy_engine = PolicyEngine(knowledge_graph)
if enable_vector_store_features and hasattr(self.vector_store, "initialize_decision_pipeline"):
try:
self.vector_store.initialize_decision_pipeline(
graph_store=knowledge_graph if enable_kg_algorithms else None,
use_graph_features=enable_kg_algorithms
)
except Exception as e:
self.logger.warning(f"Failed to initialize decision pipeline: {e}")
@property
def memory(self) -> AgentMemory:
@@ -1560,7 +1550,7 @@ class AgentContext:
Raises:
RuntimeError: If decision tracking is not enabled
"""
if not self._decision_recorder:
if not self._decision_backend:
raise RuntimeError("Decision tracking is not enabled")
from .decision_models import Decision
@@ -1579,18 +1569,54 @@ class AgentContext:
entities = entities or []
source_documents = [] # Could be enhanced to capture source docs
decision_id = self._decision_recorder.record_decision(
decision, entities, source_documents
)
# Capture cross-system context if provided
if cross_system_context:
self._decision_recorder.capture_cross_system_context(
decision_id, cross_system_context
if self._decision_backend == "graph_store":
decision_id = self._decision_recorder.record_decision(
decision, entities, source_documents
)
return decision_id
if cross_system_context:
self._decision_recorder.capture_cross_system_context(
decision_id, cross_system_context
)
return decision_id
if not hasattr(self.knowledge_graph, "add_decision"):
raise RuntimeError("Decision tracking backend does not support decisions")
self.knowledge_graph.add_decision(decision)
if cross_system_context and hasattr(self.knowledge_graph, "add_node_attribute"):
self.knowledge_graph.add_node_attribute(
decision.decision_id, {"cross_system_context": cross_system_context}
)
for entity_id in entities:
try:
self.knowledge_graph.add_edge(decision.decision_id, entity_id, edge_type="ABOUT")
except Exception:
continue
vector_id = None
if hasattr(self.vector_store, "store_decision"):
try:
vector_id = self.vector_store.store_decision(
scenario=scenario,
reasoning=reasoning,
outcome=outcome,
confidence=confidence,
entities=entities,
category=category,
decision_id=decision.decision_id,
decision_maker=decision.decision_maker,
timestamp=decision.timestamp.isoformat()
)
except Exception:
vector_id = None
if vector_id and hasattr(self.knowledge_graph, "add_node_attribute"):
self.knowledge_graph.add_node_attribute(decision.decision_id, {"vector_id": vector_id})
return decision.decision_id
def find_precedents(
self,
@@ -1618,24 +1644,85 @@ class AgentContext:
Raises:
RuntimeError: If decision tracking is not enabled
"""
if not self._decision_query:
if not self._decision_backend:
raise RuntimeError("Decision tracking is not enabled")
if use_hybrid_search:
try:
return self._decision_query.find_precedents_hybrid(
scenario, category, limit
)
except Exception:
# Fallback to basic search if hybrid fails
return self._decision_query._find_precedents_basic(scenario, category, limit)
else:
# Simple category-based search
if self._decision_backend == "graph_store":
if use_hybrid_search:
try:
return self._decision_query.find_precedents_hybrid(
scenario, category, limit
)
except Exception:
return self._decision_query._find_precedents_basic(scenario, category, limit)
if category:
return self._decision_query.find_by_category(category, limit)
else:
# Use basic search
return self._decision_query._find_precedents_basic(scenario, category, limit)
return self._decision_query._find_precedents_basic(scenario, category, limit)
results: List[Decision] = []
if use_hybrid_search and hasattr(self.vector_store, "search_decisions"):
filters = {"category": category} if category else None
vector_results = self.vector_store.search_decisions(
query=scenario,
filters=filters,
limit=limit,
use_hybrid_search=True
)
for r in vector_results:
meta = r.get("metadata") or {}
decision_id = meta.get("decision_id") or meta.get("id")
if decision_id and hasattr(self.knowledge_graph, "nodes") and decision_id in self.knowledge_graph.nodes:
node = self.knowledge_graph.nodes[decision_id]
if getattr(node, "node_type", None) == "Decision":
data = getattr(node, "properties", {}) or {}
decision = Decision(
decision_id=decision_id,
category=data.get("category", ""),
scenario=getattr(node, "content", ""),
reasoning=data.get("reasoning", ""),
outcome=data.get("outcome", ""),
confidence=float(data.get("confidence", 0.0) or 0.0),
timestamp=datetime.fromisoformat(data.get("timestamp")) if data.get("timestamp") else datetime.now(),
decision_maker=data.get("decision_maker", "ai_agent"),
reasoning_embedding=data.get("reasoning_embedding"),
node2vec_embedding=data.get("node2vec_embedding"),
metadata={k: v for k, v in data.items() if k not in [
"category", "reasoning", "outcome", "confidence",
"timestamp", "decision_maker", "reasoning_embedding", "node2vec_embedding"
]}
)
decision.metadata["score"] = r.get("score")
results.append(decision)
if results:
return results[:limit]
if hasattr(self.knowledge_graph, "find_nodes"):
for node in self.knowledge_graph.find_nodes(node_type="Decision"):
if category and node.get("metadata", {}).get("category") != category:
continue
data = node.get("metadata", {}) or {}
results.append(
Decision(
decision_id=node.get("id", ""),
category=data.get("category", ""),
scenario=node.get("content", ""),
reasoning=data.get("reasoning", ""),
outcome=data.get("outcome", ""),
confidence=float(data.get("confidence", 0.0) or 0.0),
timestamp=datetime.fromisoformat(data.get("timestamp")) if data.get("timestamp") else datetime.now(),
decision_maker=data.get("decision_maker", "ai_agent"),
reasoning_embedding=data.get("reasoning_embedding"),
node2vec_embedding=data.get("node2vec_embedding"),
metadata={k: v for k, v in data.items() if k not in [
"category", "reasoning", "outcome", "confidence",
"timestamp", "decision_maker", "reasoning_embedding", "node2vec_embedding"
]}
)
)
return results[:limit]
def get_causal_chain(
self,
@@ -1657,12 +1744,22 @@ class AgentContext:
Raises:
RuntimeError: If decision tracking is not enabled
"""
if not self._causal_analyzer:
if not self._decision_backend:
raise RuntimeError("Decision tracking is not enabled")
return self._causal_analyzer.get_causal_chain(
decision_id, direction, max_depth
)
if self._decision_backend == "graph_store":
return self._causal_analyzer.get_causal_chain(
decision_id, direction, max_depth
)
if hasattr(self.knowledge_graph, "get_causal_chain"):
return self.knowledge_graph.get_causal_chain(
decision_id=decision_id,
direction=direction,
max_depth=max_depth
)
raise RuntimeError("Decision tracking backend does not support causal chains")
def get_policy_engine(self) -> PolicyEngine:
"""
semantica/context/causal_analyzer.py
+8 -1
View File
@@ -77,7 +77,7 @@ class CausalChainAnalyzer:
using graph traversal.
"""
def __init__(self, graph_store: GraphStore):
def __init__(self, graph_store: Any):
"""
Initialize CausalChainAnalyzer.
@@ -105,6 +105,13 @@ class CausalChainAnalyzer:
List of decisions in causal chain
"""
try:
if hasattr(self.graph_store, "get_causal_chain") and not hasattr(self.graph_store, "execute_query"):
return self.graph_store.get_causal_chain(
decision_id=decision_id,
direction=direction,
max_depth=max_depth
)
if direction not in ["upstream", "downstream"]:
raise ValueError("Direction must be 'upstream' or 'downstream'")
semantica/context/context_graph.py
+85 -20
View File
@@ -260,7 +260,69 @@ class ContextGraph:
count += 1
return count
def get_neighbors(self, node_id: str, hops: int = 1) -> List[Dict[str, Any]]:
def __contains__(self, node_id: object) -> bool:
if not isinstance(node_id, str):
return False
return node_id in self.nodes
def has_node(self, node_id: str) -> bool:
return node_id in self.nodes
def neighbors(self, node_id: str) -> List[str]:
return self.get_neighbor_ids(node_id)
def get_neighbor_ids(
self,
node_id: str,
relationship_types: Optional[List[str]] = None,
) -> List[str]:
if node_id not in self.nodes:
return []
rel_filter = set(relationship_types) if relationship_types else None
neighbor_ids: List[str] = []
for edge in self._adjacency.get(node_id, []):
if rel_filter is None or edge.edge_type in rel_filter:
neighbor_ids.append(edge.target_id)
return neighbor_ids
def get_nodes_by_label(self, label: str) -> List[str]:
return list(self.node_type_index.get(label, set()))
def get_node_property(self, node_id: str, property_name: str) -> Any:
node = self.nodes.get(node_id)
if not node:
return None
return node.properties.get(property_name)
def get_node_attributes(self, node_id: str) -> Dict[str, Any]:
node = self.nodes.get(node_id)
if not node:
return {}
return node.properties.copy()
def add_node_attribute(self, node_id: str, attributes: Dict[str, Any]) -> None:
node = self.nodes.get(node_id)
if not node:
return
node.properties.update(attributes)
node.metadata.update(attributes)
def get_edge_data(self, source_id: str, target_id: str) -> Dict[str, Any]:
for edge in self._adjacency.get(source_id, []):
if edge.target_id == target_id:
data = edge.metadata.copy()
data["type"] = edge.edge_type
data["weight"] = edge.weight
return data
return {}
def get_neighbors(
self,
node_id: str,
hops: int = 1,
relationship_types: Optional[List[str]] = None,
) -> List[Dict[str, Any]]:
"""
Get neighbors of a node.
@@ -269,36 +331,39 @@ class ContextGraph:
if node_id not in self.nodes:
return []
neighbors = []
neighbors: List[Dict[str, Any]] = []
visited = {node_id}
queue = deque([(node_id, 0)]) # (current_id, current_hop)
queue = deque([(node_id, 0)])
rel_filter = set(relationship_types) if relationship_types else None
while queue:
current_id, current_hop = queue.popleft()
if current_hop >= hops:
continue
# Get outgoing edges
outgoing_edges = self._adjacency.get(current_id, [])
for edge in outgoing_edges:
if rel_filter is not None and edge.edge_type not in rel_filter:
continue
neighbor_id = edge.target_id
if neighbor_id not in visited:
visited.add(neighbor_id)
queue.append((neighbor_id, current_hop + 1))
if neighbor_id in visited:
continue
visited.add(neighbor_id)
queue.append((neighbor_id, current_hop + 1))
if neighbor_id in self.nodes:
node = self.nodes[neighbor_id]
neighbors.append(
{
"id": node.node_id,
"type": node.node_type,
"content": node.content,
"relationship": edge.edge_type,
"weight": edge.weight,
"hop": current_hop + 1,
}
)
node = self.nodes.get(neighbor_id)
if not node:
continue
neighbors.append(
{
"id": node.node_id,
"type": node.node_type,
"content": node.content,
"relationship": edge.edge_type,
"weight": edge.weight,
"hop": current_hop + 1,
}
)
return neighbors
semantica/context/context_retriever.py
+23 -4
View File
@@ -2011,8 +2011,18 @@ Answer:"""
try:
# Basic neighbor expansion
if hasattr(self.knowledge_graph, 'get_neighbors'):
neighbors = self.knowledge_graph.get_neighbors(entity_name)
for neighbor in neighbors[:5]: # Limit to prevent explosion
if hasattr(self.knowledge_graph, "neighbors"):
neighbor_ids = list(self.knowledge_graph.neighbors(entity_name))
elif hasattr(self.knowledge_graph, "get_neighbor_ids"):
neighbor_ids = self.knowledge_graph.get_neighbor_ids(entity_name)
else:
neighbor_details = self.knowledge_graph.get_neighbors(entity_name, hops=1)
neighbor_ids = [
n.get("id") for n in neighbor_details
if isinstance(n, dict) and n.get("id")
]
for neighbor in neighbor_ids[:5]: # Limit to prevent explosion
expanded_entities.append({
"name": neighbor,
"type": "related_entity",
@@ -2082,8 +2092,17 @@ Answer:"""
if hasattr(self.centrality_calculator, 'calculate_degree_centrality'):
# Simplified centrality calculation
if hasattr(self.knowledge_graph, 'get_neighbors'):
neighbors = self.knowledge_graph.get_neighbors(entity_name)
centrality_scores[entity_name] = len(neighbors)
if hasattr(self.knowledge_graph, "neighbors"):
neighbor_ids = list(self.knowledge_graph.neighbors(entity_name))
elif hasattr(self.knowledge_graph, "get_neighbor_ids"):
neighbor_ids = self.knowledge_graph.get_neighbor_ids(entity_name)
else:
neighbor_details = self.knowledge_graph.get_neighbors(entity_name, hops=1)
neighbor_ids = [
n.get("id") for n in neighbor_details
if isinstance(n, dict) and n.get("id")
]
centrality_scores[entity_name] = len(neighbor_ids)
else:
centrality_scores[entity_name] = 1
semantica/context/policy_engine.py
+363 -168
View File
@@ -84,7 +84,7 @@ class PolicyEngine:
records exceptions, and analyzes policy impact.
"""
def __init__(self, graph_store: GraphStore):
def __init__(self, graph_store: Any):
"""
Initialize PolicyEngine.
@@ -93,6 +93,7 @@ class PolicyEngine:
"""
self.graph_store = graph_store
self.logger = get_logger(__name__)
self._supports_cypher = hasattr(graph_store, "execute_query")
def add_policy(self, policy: Policy) -> str:
"""
@@ -105,35 +106,54 @@ class PolicyEngine:
Policy ID
"""
try:
# Store policy node
query = """
CREATE (p:Policy {
policy_id: $policy_id,
name: $name,
description: $description,
rules: $rules,
category: $category,
version: $version,
created_at: $created_at,
updated_at: $updated_at,
metadata: $metadata
})
"""
self.graph_store.execute_query(query, {
"policy_id": policy.policy_id,
"name": policy.name,
"description": policy.description,
"rules": policy.rules,
"category": policy.category,
"version": policy.version,
"created_at": policy.created_at,
"updated_at": policy.updated_at,
"metadata": policy.metadata
})
if self._supports_cypher:
query = """
CREATE (p:Policy {
policy_id: $policy_id,
name: $name,
description: $description,
rules: $rules,
category: $category,
version: $version,
created_at: $created_at,
updated_at: $updated_at,
metadata: $metadata
})
"""
self.graph_store.execute_query(query, {
"policy_id": policy.policy_id,
"name": policy.name,
"description": policy.description,
"rules": policy.rules,
"category": policy.category,
"version": policy.version,
"created_at": policy.created_at,
"updated_at": policy.updated_at,
"metadata": policy.metadata
})
self.logger.info(f"Added policy: {policy.policy_id} version {policy.version}")
return policy.policy_id
if not hasattr(self.graph_store, "add_node"):
raise RuntimeError("Graph backend does not support policy storage")
node_id = f"{policy.policy_id}:{policy.version}"
self.graph_store.add_node(
node_id=node_id,
node_type="Policy",
content=policy.name or policy.policy_id,
policy_id=policy.policy_id,
name=policy.name,
description=policy.description,
rules=policy.rules,
category=policy.category,
version=policy.version,
created_at=policy.created_at.isoformat() if hasattr(policy.created_at, "isoformat") else str(policy.created_at),
updated_at=policy.updated_at.isoformat() if hasattr(policy.updated_at, "isoformat") else str(policy.updated_at),
metadata=policy.metadata or {}
)
self.logger.info(f"Added policy: {policy.policy_id} version {policy.version}")
return policy.policy_id
except Exception as e:
self.logger.error(f"Failed to add policy: {e}")
raise
@@ -187,17 +207,26 @@ class PolicyEngine:
# Store new version
self.add_policy(updated_policy)
# Link versions
query = """
MATCH (old:Policy {policy_id: $policy_id, version: $old_version})
MATCH (new:Policy {policy_id: $policy_id, version: $new_version})
MERGE (old)-[:VERSION_OF]->(new)
"""
self.graph_store.execute_query(query, {
"policy_id": policy_id,
"old_version": current_policy.version,
"new_version": new_version
})
if self._supports_cypher:
query = """
MATCH (old:Policy {policy_id: $policy_id, version: $old_version})
MATCH (new:Policy {policy_id: $policy_id, version: $new_version})
MERGE (old)-[:VERSION_OF]->(new)
"""
self.graph_store.execute_query(query, {
"policy_id": policy_id,
"old_version": current_policy.version,
"new_version": new_version
})
else:
if hasattr(self.graph_store, "add_edge"):
self.graph_store.add_edge(
f"{policy_id}:{current_policy.version}",
f"{policy_id}:{new_version}",
edge_type="VERSION_OF",
changed_at=datetime.now().isoformat(),
change_reason=change_reason
)
self.logger.info(f"Updated policy {policy_id} to version {new_version}")
return new_version
@@ -222,26 +251,60 @@ class PolicyEngine:
List of applicable policies (latest versions)
"""
try:
if self._supports_cypher:
# Get latest policies for category
query = """
MATCH (p:Policy {category: $category})
WHERE NOT (p)-[:VERSION_OF]->(:Policy)
RETURN p
ORDER BY p.updated_at DESC
"""
results = self.graph_store.execute_query(query, {"category": category})
policies = []
for record in results:
policy_data = record.get("p", {})
policies.append(self._dict_to_policy(policy_data))
# Filter by entities if specified
if entities:
# This would require entity-specific policy relationships
# For now, return all category policies
pass
query = """
MATCH (p:Policy {category: $category})
WHERE NOT (p)-[:VERSION_OF]->(:Policy)
RETURN p
ORDER BY p.updated_at DESC
"""
results = self.graph_store.execute_query(query, {"category": category})
policies = []
for record in results:
policy_data = record.get("p", {})
policies.append(self._dict_to_policy(policy_data))
if entities:
pass
self.logger.info(f"Found {len(policies)} applicable policies for category {category}")
return policies
if not hasattr(self.graph_store, "find_nodes"):
return []
latest_by_policy_id: Dict[str, Dict[str, Any]] = {}
for node in self.graph_store.find_nodes(node_type="Policy"):
data = node.get("metadata", {}) or {}
if data.get("category") != category:
continue
pid = data.get("policy_id")
if not pid:
continue
updated_at = data.get("updated_at") or ""
prev = latest_by_policy_id.get(pid)
if not prev:
latest_by_policy_id[pid] = data
else:
if str(updated_at) > str(prev.get("updated_at") or ""):
latest_by_policy_id[pid] = data
policies: List[Policy] = []
for data in latest_by_policy_id.values():
policies.append(self._dict_to_policy({
"policy_id": data.get("policy_id"),
"name": data.get("name"),
"description": data.get("description"),
"rules": data.get("rules", {}),
"category": data.get("category"),
"version": data.get("version"),
"created_at": data.get("created_at"),
"updated_at": data.get("updated_at"),
"metadata": data.get("metadata", {})
}))
self.logger.info(f"Found {len(policies)} applicable policies for category {category}")
return policies
@@ -303,20 +366,33 @@ class PolicyEngine:
version: Policy version that was applied
"""
try:
query = """
MATCH (d:Decision {decision_id: $decision_id})
MATCH (p:Policy {policy_id: $policy_id, version: $version})
MERGE (d)-[:APPLIED_POLICY]->(p)
SET d.policy_applied_at = timestamp()
"""
self.graph_store.execute_query(query, {
"decision_id": decision_id,
"policy_id": policy_id,
"version": version
})
if self._supports_cypher:
query = """
MATCH (d:Decision {decision_id: $decision_id})
MATCH (p:Policy {policy_id: $policy_id, version: $version})
MERGE (d)-[:APPLIED_POLICY]->(p)
SET d.policy_applied_at = timestamp()
"""
self.graph_store.execute_query(query, {
"decision_id": decision_id,
"policy_id": policy_id,
"version": version
})
self.logger.info(f"Recorded policy application: {policy_id} v{version} to decision {decision_id}")
return
if not hasattr(self.graph_store, "add_edge"):
raise RuntimeError("Graph backend does not support relationships")
policy_node_id = f"{policy_id}:{version}"
self.graph_store.add_edge(
decision_id,
policy_node_id,
edge_type="APPLIED_POLICY",
applied_at=datetime.now().isoformat(),
policy_id=policy_id,
version=version
)
self.logger.info(f"Recorded policy application: {policy_id} v{version} to decision {decision_id}")
except Exception as e:
self.logger.error(f"Failed to record policy application: {e}")
raise
@@ -340,40 +416,61 @@ class PolicyEngine:
"""
try:
exception_id = str(uuid.uuid4())
query = """
CREATE (e:Exception {
exception_id: $exception_id,
decision_id: $decision_id,
policy_id: $policy_id,
reason: $reason,
created_at: datetime()
})
"""
self.graph_store.execute_query(query, {
"exception_id": exception_id,
"decision_id": decision_id,
"policy_id": policy_id,
"reason": reason
})
# Link to decision and policy
query = """
MATCH (d:Decision {decision_id: $decision_id})
MATCH (p:Policy {policy_id: $policy_id})
MATCH (e:Exception {exception_id: $exception_id})
MERGE (d)-[:GRANTED_EXCEPTION]->(e)
MERGE (e)-[:OVERRIDDEN_POLICY]->(p)
"""
self.graph_store.execute_query(query, {
"decision_id": decision_id,
"policy_id": policy_id,
"exception_id": exception_id
})
if self._supports_cypher:
query = """
CREATE (e:Exception {
exception_id: $exception_id,
decision_id: $decision_id,
policy_id: $policy_id,
reason: $reason,
created_at: datetime()
})
"""
self.graph_store.execute_query(query, {
"exception_id": exception_id,
"decision_id": decision_id,
"policy_id": policy_id,
"reason": reason
})
query = """
MATCH (d:Decision {decision_id: $decision_id})
MATCH (p:Policy {policy_id: $policy_id})
MATCH (e:Exception {exception_id: $exception_id})
MERGE (d)-[:GRANTED_EXCEPTION]->(e)
MERGE (e)-[:OVERRIDDEN_POLICY]->(p)
"""
self.graph_store.execute_query(query, {
"decision_id": decision_id,
"policy_id": policy_id,
"exception_id": exception_id
})
self.logger.info(f"Recorded policy exception: {exception_id}")
return exception_id
if not hasattr(self.graph_store, "add_node") or not hasattr(self.graph_store, "add_edge"):
raise RuntimeError("Graph backend does not support exceptions")
self.graph_store.add_node(
node_id=exception_id,
node_type="Exception",
content=reason,
exception_id=exception_id,
decision_id=decision_id,
policy_id=policy_id,
reason=reason,
created_at=datetime.now().isoformat()
)
self.graph_store.add_edge(decision_id, exception_id, edge_type="GRANTED_EXCEPTION")
policy = self.get_policy(policy_id)
if policy:
self.graph_store.add_edge(exception_id, f"{policy_id}:{policy.version}", edge_type="OVERRIDDEN_POLICY")
self.logger.info(f"Recorded policy exception: {exception_id}")
return exception_id
except Exception as e:
self.logger.error(f"Failed to record exception: {e}")
raise
@@ -389,23 +486,45 @@ class PolicyEngine:
List of policy versions
"""
try:
query = """
MATCH (p:Policy {policy_id: $policy_id})
OPTIONAL MATCH (p)-[:VERSION_OF*]->(future:Policy)
WITH collect(p) + collect(future) as all_versions
UNWIND all_versions as version
RETURN DISTINCT version
ORDER BY version.updated_at
"""
results = self.graph_store.execute_query(query, {"policy_id": policy_id})
if self._supports_cypher:
query = """
MATCH (p:Policy {policy_id: $policy_id})
OPTIONAL MATCH (p)-[:VERSION_OF*]->(future:Policy)
WITH collect(p) + collect(future) as all_versions
UNWIND all_versions as version
RETURN DISTINCT version
ORDER BY version.updated_at
"""
results = self.graph_store.execute_query(query, {"policy_id": policy_id})
policies = []
for record in results:
policy_data = record.get("version", {})
policies.append(self._dict_to_policy(policy_data))
policies = []
for record in results:
policy_data = record.get("version", {})
policies.append(self._dict_to_policy(policy_data))
self.logger.info(f"Found {len(policies)} versions for policy {policy_id}")
return policies
self.logger.info(f"Found {len(policies)} versions for policy {policy_id}")
return policies
if not hasattr(self.graph_store, "find_nodes"):
return []
versions: List[Policy] = []
for node in self.graph_store.find_nodes(node_type="Policy"):
data = node.get("metadata", {}) or {}
if data.get("policy_id") != policy_id:
continue
versions.append(self._dict_to_policy({
"policy_id": data.get("policy_id"),
"name": data.get("name"),
"description": data.get("description"),
"rules": data.get("rules", {}),
"category": data.get("category"),
"version": data.get("version"),
"created_at": data.get("created_at"),
"updated_at": data.get("updated_at"),
"metadata": data.get("metadata", {})
}))
versions.sort(key=lambda p: str(p.updated_at))
return versions
except Exception as e:
self.logger.error(f"Failed to get policy history: {e}")
@@ -429,23 +548,33 @@ class PolicyEngine:
List of affected decision IDs
"""
try:
query = """
MATCH (d:Decision)-[:APPLIED_POLICY]->(p:Policy {
policy_id: $policy_id,
version: $from_version
})
RETURN d.decision_id as decision_id
"""
results = self.graph_store.execute_query(query, {
"policy_id": policy_id,
"from_version": from_version
})
if self._supports_cypher:
query = """
MATCH (d:Decision)-[:APPLIED_POLICY]->(p:Policy {
policy_id: $policy_id,
version: $from_version
})
RETURN d.decision_id as decision_id
"""
results = self.graph_store.execute_query(query, {
"policy_id": policy_id,
"from_version": from_version
})
decision_ids = []
for record in results:
decision_ids.append(record.get("decision_id", ""))
decision_ids = []
for record in results:
decision_ids.append(record.get("decision_id", ""))
self.logger.info(f"Found {len(decision_ids)} decisions affected by policy change")
self.logger.info(f"Found {len(decision_ids)} decisions affected by policy change")
return decision_ids
if not hasattr(self.graph_store, "find_edges"):
return []
policy_node_id = f"{policy_id}:{from_version}"
decision_ids: List[str] = []
for edge in self.graph_store.find_edges(edge_type="APPLIED_POLICY"):
if edge.get("target") == policy_node_id:
decision_ids.append(edge.get("source"))
return decision_ids
except Exception as e:
@@ -471,16 +600,38 @@ class PolicyEngine:
current_policy = self.get_policy(policy_id)
if not current_policy:
raise ValueError(f"Policy {policy_id} not found")
# Get decisions that used this policy
query = """
MATCH (d:Decision)-[:APPLIED_POLICY]->(p:Policy {policy_id: $policy_id})
RETURN d.decision_id as decision_id, d.confidence as confidence,
d.outcome as outcome, d.category as category
"""
results = self.graph_store.execute_query(query, {"policy_id": policy_id})
# Analyze impact
results: List[Dict[str, Any]] = []
if self._supports_cypher:
query = """
MATCH (d:Decision)-[:APPLIED_POLICY]->(p:Policy {policy_id: $policy_id})
RETURN d.decision_id as decision_id, d.confidence as confidence,
d.outcome as outcome, d.category as category
"""
results = self.graph_store.execute_query(query, {"policy_id": policy_id})
else:
if hasattr(self.graph_store, "find_edges") and hasattr(self.graph_store, "nodes"):
for edge in self.graph_store.find_edges(edge_type="APPLIED_POLICY"):
target = edge.get("target")
if not target or not isinstance(target, str):
continue
policy_node = self.graph_store.nodes.get(target)
if not policy_node:
continue
props = getattr(policy_node, "properties", {}) or {}
if props.get("policy_id") != policy_id:
continue
decision_node = self.graph_store.nodes.get(edge.get("source"))
if not decision_node:
continue
dprops = getattr(decision_node, "properties", {}) or {}
results.append({
"decision_id": edge.get("source"),
"confidence": dprops.get("confidence", 0.0),
"outcome": dprops.get("outcome", ""),
"category": dprops.get("category", "")
})
impact_analysis = {
"total_decisions": len(results),
"affected_decisions": 0,
@@ -488,19 +639,16 @@ class PolicyEngine:
"risk_assessment": "low",
"recommendations": []
}
for record in results:
decision_data = {
"confidence": record.get("confidence", 0.0),
"outcome": record.get("outcome", ""),
"category": record.get("category", "")
}
# Check if decision would still comply with new rules
would_comply = self._check_compliance_with_rules(
decision_data, proposed_rules
)
if not would_comply:
impact_analysis["affected_decisions"] += 1
@@ -546,29 +694,76 @@ class PolicyEngine:
Policy object or None
"""
try:
if self._supports_cypher:
if version:
query = """
MATCH (p:Policy {policy_id: $policy_id, version: $version})
RETURN p
"""
params = {"policy_id": policy_id, "version": version}
else:
query = """
MATCH (p:Policy {policy_id: $policy_id})
WHERE NOT (p)-[:VERSION_OF]->(:Policy)
RETURN p
"""
params = {"policy_id": policy_id}
results = self.graph_store.execute_query(query, params)
if results:
policy_data = results[0].get("p", {})
return self._dict_to_policy(policy_data)
return None
if not hasattr(self.graph_store, "find_nodes"):
return None
candidates: List[Dict[str, Any]] = []
for node in self.graph_store.find_nodes(node_type="Policy"):
data = node.get("metadata", {}) or {}
if data.get("policy_id") != policy_id:
continue
if version and data.get("version") != version:
continue
candidates.append(data)
if not candidates:
return None
if version:
query = """
MATCH (p:Policy {policy_id: $policy_id, version: $version})
RETURN p
"""
params = {"policy_id": policy_id, "version": version}
data = candidates[0]
else:
# Get latest version
query = """
MATCH (p:Policy {policy_id: $policy_id})
WHERE NOT (p)-[:VERSION_OF]->(:Policy)
RETURN p
"""
params = {"policy_id": policy_id}
results = self.graph_store.execute_query(query, params)
if results:
policy_data = results[0].get("p", {})
return self._dict_to_policy(policy_data)
return None
# Prefer highest semantic version if available, fallback to updated_at
def _version_key(v: str) -> tuple:
try:
parts = [int(p) for p in str(v).split(".")]
# Normalize length for comparison
while len(parts) < 3:
parts.append(-1)
return tuple(parts[:3])
except Exception:
return (-1, -1, -1)
try:
data = max(
candidates,
key=lambda d: (_version_key(d.get("version")), str(d.get("updated_at") or "")),
)
except Exception:
data = max(candidates, key=lambda d: str(d.get("updated_at") or ""))
return self._dict_to_policy({
"policy_id": data.get("policy_id"),
"name": data.get("name"),
"description": data.get("description"),
"rules": data.get("rules", {}),
"category": data.get("category"),
"version": data.get("version"),
"created_at": data.get("created_at"),
"updated_at": data.get("updated_at"),
"metadata": data.get("metadata", {})
})
except Exception as e:
self.logger.error(f"Failed to get policy: {e}")
return None
semantica/kg/centrality_calculator.py
+5
View File
@@ -739,6 +739,11 @@ class CentralityCalculator:
neighbors = list(graph.neighbors(node))
elif hasattr(graph, 'get_neighbors'):
neighbors = graph.get_neighbors(node)
if neighbors and isinstance(neighbors[0], dict):
neighbors = [
n.get("id") for n in neighbors
if isinstance(n, dict) and n.get("id")
]
else:
neighbors = []
semantica/kg/community_detector.py
+6
View File
@@ -891,6 +891,12 @@ class CommunityDetector:
all_neighbors = graph.get_neighbors(node)
else:
all_neighbors = []
if all_neighbors and isinstance(all_neighbors[0], dict):
all_neighbors = [
n.get("id") for n in all_neighbors
if isinstance(n, dict) and n.get("id")
]
# Filter by relationship types if specified
if relationship_types is not None and hasattr(graph, 'get_edge_data'):
semantica/kg/link_predictor.py
+4 -1
View File
@@ -455,6 +455,9 @@ class LinkPredictor:
if hasattr(graph_store, 'neighbors'):
return list(graph_store.neighbors(node_id))
elif hasattr(graph_store, 'get_neighbors'):
return graph_store.get_neighbors(node_id)
neighbors = graph_store.get_neighbors(node_id)
if neighbors and isinstance(neighbors[0], dict):
return [n.get("id") for n in neighbors if isinstance(n, dict) and n.get("id")]
return neighbors
else:
return []
semantica/kg/node_embeddings.py
+19 -7
View File
@@ -350,13 +350,25 @@ class NodeEmbedder:
# Build adjacency
for node in nodes:
if hasattr(graph_store, 'get_neighbors'):
neighbors = graph_store.get_neighbors(node, relationship_types)
adjacency[node] = neighbors
else:
# Fallback for NetworkX
if hasattr(graph_store, 'neighbors'):
adjacency[node] = list(graph_store.neighbors(node))
if hasattr(graph_store, 'neighbors'):
adjacency[node] = list(graph_store.neighbors(node))
elif hasattr(graph_store, 'get_neighbor_ids'):
adjacency[node] = graph_store.get_neighbor_ids(node, relationship_types)
elif hasattr(graph_store, 'get_neighbors'):
try:
neighbor_details = graph_store.get_neighbors(node, hops=1, relationship_types=relationship_types)
adjacency[node] = [
n.get("id") for n in neighbor_details
if isinstance(n, dict) and n.get("id")
]
except TypeError:
neighbor_details = graph_store.get_neighbors(node)
adjacency[node] = [
n.get("id") for n in neighbor_details
if isinstance(n, dict) and n.get("id")
]
elif hasattr(graph_store, 'nodes') and hasattr(graph_store, 'edges'):
adjacency[node] = []
return dict(adjacency)
tests/context/test_agent_context_smoke.py
+66
View File
@@ -0,0 +1,66 @@
import pytest
from semantica.context import AgentContext, ContextGraph
from semantica.context.decision_models import Policy
from semantica.vector_store import VectorStore
from datetime import datetime
def test_agent_context_minimal_decisions_and_chain():
vs = VectorStore(backend="inmemory", dimension=64)
graph = ContextGraph()
ctx = AgentContext(
vector_store=vs,
knowledge_graph=graph,
enable_decision_tracking=True,
enable_kg_algorithms=False,
enable_vector_store_features=False,
)
d1 = ctx.record_decision(
category="credit_approval",
scenario="s1",
reasoning="r1",
outcome="rejected",
confidence=0.8,
entities=["e1"],
decision_maker="tester",
)
d2 = ctx.record_decision(
category="credit_approval",
scenario="s2",
reasoning="r2",
outcome="rejected",
confidence=0.85,
entities=["e1"],
decision_maker="tester",
)
graph.add_causal_relationship(d1, d2, "INFLUENCED")
chain = ctx.get_causal_chain(d2, direction="upstream", max_depth=5)
assert isinstance(chain, list)
assert len(chain) >= 1
def test_agent_context_policy_engine_with_graph_backend():
vs = VectorStore(backend="inmemory", dimension=64)
graph = ContextGraph()
ctx = AgentContext(
vector_store=vs,
knowledge_graph=graph,
enable_decision_tracking=True,
enable_kg_algorithms=False,
enable_vector_store_features=False,
)
pe = ctx.get_policy_engine()
pol = Policy(
policy_id="cp",
name="Credit Policy",
description="d",
rules={"min_confidence": 0.8, "allowed_outcomes": ["approved", "rejected"]},
category="credit_approval",
version="1.0.0",
created_at=datetime.now(),
updated_at=datetime.now(),
metadata={},
)
pe.add_policy(pol)
found = pe.get_policy("cp")
assert found is not None
tests/context/test_policy_engine_fallback.py
+62
View File
@@ -0,0 +1,62 @@
import pytest
from datetime import datetime
from semantica.context import ContextGraph, PolicyEngine
from semantica.context.decision_models import Policy, Decision
def _make_policy(pid="p1", version="1.0.0", min_conf=0.8):
return Policy(
policy_id=pid,
name="Test Policy",
description="desc",
rules={"min_confidence": min_conf, "allowed_outcomes": ["approved", "rejected"]},
category="test",
version=version,
created_at=datetime.now(),
updated_at=datetime.now(),
metadata={},
)
def _make_decision(decision_id, conf=0.9, outcome="approved"):
return Decision(
decision_id=decision_id,
category="test",
scenario="s",
reasoning="r",
outcome=outcome,
confidence=conf,
timestamp=datetime.now(),
decision_maker="tester",
metadata={},
)
def test_policy_engine_add_get_update_with_context_graph():
graph = ContextGraph()
engine = PolicyEngine(graph)
p = _make_policy()
engine.add_policy(p)
latest = engine.get_policy("p1")
assert latest is not None
assert latest.version == "1.0.0"
new_ver = engine.update_policy("p1", {"min_confidence": 0.85, "allowed_outcomes": ["approved", "rejected"]}, "raise min")
assert isinstance(new_ver, str)
latest2 = engine.get_policy("p1")
assert latest2 is not None
assert latest2.version != "1.0.0"
def test_policy_engine_compliance_and_application_edges():
graph = ContextGraph()
engine = PolicyEngine(graph)
p = _make_policy()
engine.add_policy(p)
d = _make_decision("d1", conf=0.9, outcome="approved")
graph.add_decision(d)
ok = engine.check_compliance(d, "p1")
assert ok is True
engine.record_policy_application("d1", "p1", "1.0.0")
edges = graph.find_edges(edge_type="APPLIED_POLICY")
assert any(e.get("source") == "d1" and isinstance(e.get("target"), str) and e.get("target").startswith("p1:") for e in edges)