diff --git a/README.md b/README.md index dee65814..66792a8f 100644 --- a/README.md +++ b/README.md @@ -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. diff --git a/semantica/context/agent_context.py b/semantica/context/agent_context.py index 6a2ecaaa..0fdde432 100644 --- a/semantica/context/agent_context.py +++ b/semantica/context/agent_context.py @@ -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: """ diff --git a/semantica/context/causal_analyzer.py b/semantica/context/causal_analyzer.py index ab53699c..9ad64215 100644 --- a/semantica/context/causal_analyzer.py +++ b/semantica/context/causal_analyzer.py @@ -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'") diff --git a/semantica/context/context_graph.py b/semantica/context/context_graph.py index 4084391c..6e201a80 100644 --- a/semantica/context/context_graph.py +++ b/semantica/context/context_graph.py @@ -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 diff --git a/semantica/context/context_retriever.py b/semantica/context/context_retriever.py index c1131f7d..b72833c9 100644 --- a/semantica/context/context_retriever.py +++ b/semantica/context/context_retriever.py @@ -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 diff --git a/semantica/context/policy_engine.py b/semantica/context/policy_engine.py index cb7eb669..9cefaad6 100644 --- a/semantica/context/policy_engine.py +++ b/semantica/context/policy_engine.py @@ -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 diff --git a/semantica/kg/centrality_calculator.py b/semantica/kg/centrality_calculator.py index 95ee72f7..1fda57a7 100644 --- a/semantica/kg/centrality_calculator.py +++ b/semantica/kg/centrality_calculator.py @@ -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 = [] diff --git a/semantica/kg/community_detector.py b/semantica/kg/community_detector.py index cd25eb88..2077e4e1 100644 --- a/semantica/kg/community_detector.py +++ b/semantica/kg/community_detector.py @@ -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'): diff --git a/semantica/kg/link_predictor.py b/semantica/kg/link_predictor.py index 2144fa59..e811957b 100644 --- a/semantica/kg/link_predictor.py +++ b/semantica/kg/link_predictor.py @@ -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 [] diff --git a/semantica/kg/node_embeddings.py b/semantica/kg/node_embeddings.py index ca746f51..e4018279 100644 --- a/semantica/kg/node_embeddings.py +++ b/semantica/kg/node_embeddings.py @@ -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) diff --git a/tests/context/test_agent_context_smoke.py b/tests/context/test_agent_context_smoke.py new file mode 100644 index 00000000..d019ef20 --- /dev/null +++ b/tests/context/test_agent_context_smoke.py @@ -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 diff --git a/tests/context/test_policy_engine_fallback.py b/tests/context/test_policy_engine_fallback.py new file mode 100644 index 00000000..bc26f5bf --- /dev/null +++ b/tests/context/test_policy_engine_fallback.py @@ -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)