From 4cd3ef9aa8be442024efb3d0210ee3578a33ca7d Mon Sep 17 00:00:00 2001 From: KaifAhmad1 Date: Sat, 14 Feb 2026 17:13:38 +0530 Subject: [PATCH 1/4] context_fixes --- README.md | 31 ++ semantica/context/agent_context.py | 249 ++++++--- semantica/context/causal_analyzer.py | 9 +- semantica/context/context_graph.py | 105 +++- semantica/context/context_retriever.py | 27 +- semantica/context/policy_engine.py | 531 +++++++++++++------ semantica/kg/centrality_calculator.py | 5 + semantica/kg/community_detector.py | 6 + semantica/kg/link_predictor.py | 5 +- semantica/kg/node_embeddings.py | 26 +- tests/context/test_agent_context_smoke.py | 66 +++ tests/context/test_policy_engine_fallback.py | 62 +++ 12 files changed, 845 insertions(+), 277 deletions(-) create mode 100644 tests/context/test_agent_context_smoke.py create mode 100644 tests/context/test_policy_engine_fallback.py 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) From b90ffcca9aec13d8331b7092ffe30d837fa712e6 Mon Sep 17 00:00:00 2001 From: KaifAhmad1 Date: Sat, 14 Feb 2026 18:02:06 +0530 Subject: [PATCH 2/4] context_compliance_fixes --- PR_CONTEXT.md | 89 ++++++++++++++++++++ README.md | 2 - semantica/context/agent_context.py | 70 +++++++++++---- semantica/context/context_graph.py | 7 +- tests/context/test_policy_engine_fallback.py | 28 +++--- 5 files changed, 167 insertions(+), 29 deletions(-) create mode 100644 PR_CONTEXT.md diff --git a/PR_CONTEXT.md b/PR_CONTEXT.md new file mode 100644 index 00000000..7edb829c --- /dev/null +++ b/PR_CONTEXT.md @@ -0,0 +1,89 @@ +# Context: PolicyEngine fixes, new context tests, cleanup — all tests passing + +## Summary +This PR improves Context Graph decision tracking reliability by fixing PolicyEngine behavior on non‑Cypher backends, resolving a compile issue, and adding focused tests. It also removes an example script per request. All tests pass. + +- Source branch: `context` +- Base branch: do not push directly to `main`; open PR for review +- Scope: Context graphs decision tracking and policy evaluation + +## Changes + +### Policy engine +- Fix: Correct indentation for Cypher branches to prevent compile errors. +- Fix: `get_policy` (non‑Cypher path) now selects the latest policy by semantic version first, falling back to `updated_at` as a secondary key. + +### Tests +- Added `tests/context/test_policy_engine_fallback.py` for ContextGraph (no Cypher) versioning, compliance, and `APPLIED_POLICY` edges. +- Added `tests/context/test_agent_context_smoke.py` for AgentContext + ContextGraph decision tracking (record decisions, causal chain, policy storage). + +### Cleanup +- Removed `examples/context_graphs_decision_tracking.py`. +- Deleted transient test artifacts from earlier local runs (not tracked). + +## Files Touched (high level) +- `semantica/context/policy_engine.py` +- `tests/context/test_policy_engine_fallback.py` (new) +- `tests/context/test_agent_context_smoke.py` (new) +- `examples/context_graphs_decision_tracking.py` (removed) + +## Rationale +- PolicyEngine compile error: Cypher code path indentation was causing a compile failure; fixed so it executes only on Cypher‑capable backends. +- Latest version retrieval: On the in‑memory ContextGraph backend, latest policy selection relied on `updated_at` strings and could return the wrong version. Updated to prefer semantic version order with robust fallback. + +## Test Coverage + +### New tests +- PolicyEngine fallback on ContextGraph: + - Adds/gets/updates policies and verifies the latest version is returned post‑update. + - Checks compliance and presence of `APPLIED_POLICY` edges. +- AgentContext + ContextGraph smoke: + - Records decisions, links causal relationship, validates upstream chain retrieval. + - Stores and retrieves policies via PolicyEngine obtained from AgentContext. + +### Existing tests (already in repo) cover +- ContextGraph decision nodes, precedents, and edge cases. +- Banking/Healthcare end‑to‑end flows. +- Context retriever hybrid/precedents and AgentContext integration. + +## Results +- Context-only tests: + - `python -m pytest -q tests/context` → passed +- Full test suite: + - `python -m pytest -q tests` → passed +- Example subset (smoke): + - `tests/context/test_agent_context_smoke.py` → passed (deprecation warnings are expected) + - `tests/context/test_policy_engine_fallback.py` → passed after fixes + +## How to Test Locally + +```bash +# Optional: fresh venv +python -m venv .venv +.\.venv\Scripts\Activate.ps1 + +# Install dev deps (includes pytest) +pip install -e ".[dev]" + +# Run context suite +python -m pytest -q tests\context + +# Run full suite +python -m pytest -q tests +``` + +## Backwards Compatibility & Risk +- No breaking API changes; fixes are internal behavior corrections for non‑Cypher PolicyEngine operations. +- Risk is low; covered by new tests and full suite. + +## Follow‑ups (optional) +- Add stricter warnings policy if desired (e.g., `filterwarnings` in `pyproject.toml`/`pytest.ini`). +- Expand negative tests (e.g., cycles in causal traversal, invalid rule sets). + +## Checklist +- [x] No direct pushes to `main`. +- [x] New tests added and passing locally. +- [x] No secrets or credentials added. +- [x] Example file removed per request. +- [x] Clear rationale and instructions provided. + diff --git a/README.md b/README.md index 66792a8f..6fce0ca0 100644 --- a/README.md +++ b/README.md @@ -763,8 +763,6 @@ precedents = context.find_precedents( ) ``` -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 0fdde432..cfead678 100644 --- a/semantica/context/agent_context.py +++ b/semantica/context/agent_context.py @@ -238,7 +238,9 @@ class AgentContext: 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.logger.warning( + f"Failed to initialize enhanced decision tracking ({type(e).__name__})" + ) self._decision_recorder = DecisionRecorder(knowledge_graph) self._decision_query = DecisionQuery(knowledge_graph) self._causal_analyzer = CausalChainAnalyzer(knowledge_graph) @@ -254,7 +256,9 @@ class AgentContext: use_graph_features=enable_kg_algorithms ) except Exception as e: - self.logger.warning(f"Failed to initialize decision pipeline: {e}") + self.logger.warning( + f"Failed to initialize decision pipeline ({type(e).__name__})" + ) @property def memory(self) -> AgentMemory: @@ -756,7 +760,7 @@ class AgentContext: "edge_count": graph.get("statistics", {}).get("edge_count", 0), } except Exception as e: - self.logger.warning(f"Failed to build graph from documents: {e}") + self.logger.warning(f"Failed to build graph from documents ({type(e).__name__})") return {"node_count": 0, "edge_count": 0} def _context_to_dict( @@ -1457,7 +1461,7 @@ class AgentContext: if memory_id: imported += 1 except Exception as e: - self.logger.warning(f"Failed to import memory: {e}") + self.logger.warning(f"Failed to import memory ({type(e).__name__})") return imported @@ -1590,13 +1594,30 @@ class AgentContext: self.knowledge_graph.add_node_attribute( decision.decision_id, {"cross_system_context": cross_system_context} ) + about_edge_failures: List[Dict[str, str]] = [] for entity_id in entities: try: self.knowledge_graph.add_edge(decision.decision_id, entity_id, edge_type="ABOUT") - except Exception: - continue + except Exception as e: + about_edge_failures.append( + {"entity_id": str(entity_id), "error_type": type(e).__name__} + ) + + if about_edge_failures: + failure_types = sorted( + {f.get("error_type", "") for f in about_edge_failures if f.get("error_type")} + ) + self.logger.warning( + f"record_decision ABOUT edge creation failures: {len(about_edge_failures)} " + f"({', '.join(failure_types) if failure_types else 'unknown'})" + ) + if hasattr(self.knowledge_graph, "add_node_attribute"): + self.knowledge_graph.add_node_attribute( + decision.decision_id, {"about_edge_failures": about_edge_failures} + ) vector_id = None + vector_store_error_type: Optional[str] = None if hasattr(self.vector_store, "store_decision"): try: vector_id = self.vector_store.store_decision( @@ -1610,8 +1631,17 @@ class AgentContext: decision_maker=decision.decision_maker, timestamp=decision.timestamp.isoformat() ) - except Exception: + except Exception as e: vector_id = None + vector_store_error_type = type(e).__name__ + self.logger.warning( + f"record_decision vector store write failed ({vector_store_error_type})" + ) + if hasattr(self.knowledge_graph, "add_node_attribute"): + self.knowledge_graph.add_node_attribute( + decision.decision_id, + {"vector_store_error_type": vector_store_error_type}, + ) if vector_id and hasattr(self.knowledge_graph, "add_node_attribute"): self.knowledge_graph.add_node_attribute(decision.decision_id, {"vector_id": vector_id}) @@ -1661,6 +1691,16 @@ class AgentContext: results: List[Decision] = [] + def _safe_parse_timestamp(value: Any) -> datetime: + if isinstance(value, datetime): + return value + if not value: + return datetime.now() + try: + return datetime.fromisoformat(str(value)) + except Exception: + return datetime.now() + 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( @@ -1683,7 +1723,7 @@ class AgentContext: 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(), + timestamp=_safe_parse_timestamp(data.get("timestamp")), decision_maker=data.get("decision_maker", "ai_agent"), reasoning_embedding=data.get("reasoning_embedding"), node2vec_embedding=data.get("node2vec_embedding"), @@ -1711,7 +1751,7 @@ class AgentContext: 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(), + timestamp=_safe_parse_timestamp(data.get("timestamp")), decision_maker=data.get("decision_maker", "ai_agent"), reasoning_embedding=data.get("reasoning_embedding"), node2vec_embedding=data.get("node2vec_embedding"), @@ -1966,7 +2006,7 @@ class AgentContext: "message": "Basic analysis only - KG features not available" } except Exception as e: - self.logger.error(f"Failed to analyze context graph: {e}") + self.logger.error(f"Failed to analyze context graph ({type(e).__name__})") return {"error": str(e)} def find_similar_entities( @@ -1993,7 +2033,7 @@ class AgentContext: # Fallback to basic content similarity return [] except Exception as e: - self.logger.error(f"Failed to find similar entities: {e}") + self.logger.error(f"Failed to find similar entities ({type(e).__name__})") return [] def get_entity_centrality(self, entity_id: str) -> Dict[str, float]: @@ -2015,7 +2055,7 @@ class AgentContext: else: return {"error": "Centrality analysis not available"} except Exception as e: - self.logger.error(f"Failed to get entity centrality: {e}") + self.logger.error(f"Failed to get entity centrality ({type(e).__name__})") return {"error": str(e)} def find_precedents_advanced( @@ -2055,7 +2095,7 @@ class AgentContext: # Fallback to basic method return self.find_precedents(scenario, category, limit) except Exception as e: - self.logger.error(f"Failed to find advanced precedents: {e}") + self.logger.error(f"Failed to find advanced precedents ({type(e).__name__})") return [] def analyze_decision_influence(self, decision_id: str, max_depth: int = 3) -> Dict[str, Any]: @@ -2084,7 +2124,7 @@ class AgentContext: "message": "Basic analysis only - KG features not available" } except Exception as e: - self.logger.error(f"Failed to analyze decision influence: {e}") + self.logger.error(f"Failed to analyze decision influence ({type(e).__name__})") return {"error": str(e)} def predict_decision_relationships(self, decision_id: str, top_k: int = 5) -> List[Dict]: @@ -2107,7 +2147,7 @@ class AgentContext: else: return [] except Exception as e: - self.logger.error(f"Failed to predict decision relationships: {e}") + self.logger.error(f"Failed to predict decision relationships ({type(e).__name__})") return [] def get_context_insights(self) -> Dict[str, Any]: diff --git a/semantica/context/context_graph.py b/semantica/context/context_graph.py index 6e201a80..abf35213 100644 --- a/semantica/context/context_graph.py +++ b/semantica/context/context_graph.py @@ -509,11 +509,14 @@ class ContextGraph: """Find a node by ID.""" node = self.nodes.get(node_id) if node: + merged_metadata = {} + merged_metadata.update(getattr(node, "metadata", {}) or {}) + merged_metadata.update(getattr(node, "properties", {}) or {}) return { "id": node.node_id, "type": node.node_type, "content": node.content, - "metadata": node.metadata, + "metadata": merged_metadata, } return None @@ -530,7 +533,7 @@ class ContextGraph: "id": n.node_id, "type": n.node_type, "content": n.content, - "metadata": n.metadata, + "metadata": {**(getattr(n, "metadata", {}) or {}), **(getattr(n, "properties", {}) or {})}, } for n in nodes ] diff --git a/tests/context/test_policy_engine_fallback.py b/tests/context/test_policy_engine_fallback.py index bc26f5bf..6c163e09 100644 --- a/tests/context/test_policy_engine_fallback.py +++ b/tests/context/test_policy_engine_fallback.py @@ -1,4 +1,3 @@ -import pytest from datetime import datetime from semantica.context import ContextGraph, PolicyEngine @@ -36,12 +35,16 @@ def _make_decision(decision_id, conf=0.9, outcome="approved"): def test_policy_engine_add_get_update_with_context_graph(): graph = ContextGraph() engine = PolicyEngine(graph) - p = _make_policy() - engine.add_policy(p) + policy = _make_policy() + engine.add_policy(policy) 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") + 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 @@ -51,12 +54,17 @@ def test_policy_engine_add_get_update_with_context_graph(): 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") + policy = _make_policy() + engine.add_policy(policy) + decision = _make_decision("d1", conf=0.9, outcome="approved") + graph.add_decision(decision) + ok = engine.check_compliance(decision, "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) + assert any( + e.get("source") == "d1" + and isinstance(e.get("target"), str) + and e.get("target").startswith("p1:") + for e in edges + ) From 47c0058dce1c880c9541aed685797963dae9f83e Mon Sep 17 00:00:00 2001 From: Mohd Kaif <98801504+KaifAhmad1@users.noreply.github.com> Date: Sat, 14 Feb 2026 18:04:37 +0530 Subject: [PATCH 3/4] Delete PR_CONTEXT.md --- PR_CONTEXT.md | 89 --------------------------------------------------- 1 file changed, 89 deletions(-) delete mode 100644 PR_CONTEXT.md diff --git a/PR_CONTEXT.md b/PR_CONTEXT.md deleted file mode 100644 index 7edb829c..00000000 --- a/PR_CONTEXT.md +++ /dev/null @@ -1,89 +0,0 @@ -# Context: PolicyEngine fixes, new context tests, cleanup — all tests passing - -## Summary -This PR improves Context Graph decision tracking reliability by fixing PolicyEngine behavior on non‑Cypher backends, resolving a compile issue, and adding focused tests. It also removes an example script per request. All tests pass. - -- Source branch: `context` -- Base branch: do not push directly to `main`; open PR for review -- Scope: Context graphs decision tracking and policy evaluation - -## Changes - -### Policy engine -- Fix: Correct indentation for Cypher branches to prevent compile errors. -- Fix: `get_policy` (non‑Cypher path) now selects the latest policy by semantic version first, falling back to `updated_at` as a secondary key. - -### Tests -- Added `tests/context/test_policy_engine_fallback.py` for ContextGraph (no Cypher) versioning, compliance, and `APPLIED_POLICY` edges. -- Added `tests/context/test_agent_context_smoke.py` for AgentContext + ContextGraph decision tracking (record decisions, causal chain, policy storage). - -### Cleanup -- Removed `examples/context_graphs_decision_tracking.py`. -- Deleted transient test artifacts from earlier local runs (not tracked). - -## Files Touched (high level) -- `semantica/context/policy_engine.py` -- `tests/context/test_policy_engine_fallback.py` (new) -- `tests/context/test_agent_context_smoke.py` (new) -- `examples/context_graphs_decision_tracking.py` (removed) - -## Rationale -- PolicyEngine compile error: Cypher code path indentation was causing a compile failure; fixed so it executes only on Cypher‑capable backends. -- Latest version retrieval: On the in‑memory ContextGraph backend, latest policy selection relied on `updated_at` strings and could return the wrong version. Updated to prefer semantic version order with robust fallback. - -## Test Coverage - -### New tests -- PolicyEngine fallback on ContextGraph: - - Adds/gets/updates policies and verifies the latest version is returned post‑update. - - Checks compliance and presence of `APPLIED_POLICY` edges. -- AgentContext + ContextGraph smoke: - - Records decisions, links causal relationship, validates upstream chain retrieval. - - Stores and retrieves policies via PolicyEngine obtained from AgentContext. - -### Existing tests (already in repo) cover -- ContextGraph decision nodes, precedents, and edge cases. -- Banking/Healthcare end‑to‑end flows. -- Context retriever hybrid/precedents and AgentContext integration. - -## Results -- Context-only tests: - - `python -m pytest -q tests/context` → passed -- Full test suite: - - `python -m pytest -q tests` → passed -- Example subset (smoke): - - `tests/context/test_agent_context_smoke.py` → passed (deprecation warnings are expected) - - `tests/context/test_policy_engine_fallback.py` → passed after fixes - -## How to Test Locally - -```bash -# Optional: fresh venv -python -m venv .venv -.\.venv\Scripts\Activate.ps1 - -# Install dev deps (includes pytest) -pip install -e ".[dev]" - -# Run context suite -python -m pytest -q tests\context - -# Run full suite -python -m pytest -q tests -``` - -## Backwards Compatibility & Risk -- No breaking API changes; fixes are internal behavior corrections for non‑Cypher PolicyEngine operations. -- Risk is low; covered by new tests and full suite. - -## Follow‑ups (optional) -- Add stricter warnings policy if desired (e.g., `filterwarnings` in `pyproject.toml`/`pytest.ini`). -- Expand negative tests (e.g., cycles in causal traversal, invalid rule sets). - -## Checklist -- [x] No direct pushes to `main`. -- [x] New tests added and passing locally. -- [x] No secrets or credentials added. -- [x] Example file removed per request. -- [x] Clear rationale and instructions provided. - From 4e31296c1eb6673b6efc5b3d467ec17bbc149f07 Mon Sep 17 00:00:00 2001 From: KaifAhmad1 Date: Sun, 15 Feb 2026 12:52:19 +0530 Subject: [PATCH 4/4] Fix Context Graphs decision tracking and add comprehensive tests - Fix empty/None decision ID handling in ContextGraph.add_decision() - Fix None metadata handling to prevent TypeError - Fix causal chain depth logic and node exclusion - Fix nonexistent node handling in add_causal_relationship() - Add missing properties field in to_dict serialization - Add missing from_dict method for graph deserialization - Fix precedent search direction in find_precedents() - Fix UUID generation logic in all decision models - Add comprehensive test suite with 9 tests covering all features - Test coverage: decision tracking, graph analytics, use cases, performance - All 71 context tests now passing (100% success rate) Resolves critical bugs in Context Graphs feature (#290) implementation --- semantica/context/context_graph.py | 111 +++- semantica/context/decision_models.py | 12 +- tests/context/test_context_graphs_examples.py | 521 ++++++++++++++++++ 3 files changed, 609 insertions(+), 35 deletions(-) create mode 100644 tests/context/test_context_graphs_examples.py diff --git a/semantica/context/context_graph.py b/semantica/context/context_graph.py index abf35213..a6c455aa 100644 --- a/semantica/context/context_graph.py +++ b/semantica/context/context_graph.py @@ -77,6 +77,7 @@ from collections import defaultdict, deque from dataclasses import dataclass, field from datetime import datetime from typing import Any, Dict, List, Optional, Set, Tuple, Union +import uuid from ..utils.logging import get_logger from ..utils.progress_tracker import get_progress_tracker @@ -841,6 +842,7 @@ class ContextGraph: "id": n.node_id, "type": n.node_type, "content": n.content, + "properties": n.properties, "metadata": n.metadata, } for n in self.nodes.values() @@ -860,6 +862,34 @@ class ContextGraph: }, } + def from_dict(self, graph_dict: Dict[str, Any]) -> None: + """Load graph from dictionary format.""" + # Clear existing graph + self.nodes.clear() + self.edges.clear() + + # Add nodes + for node_data in graph_dict.get("nodes", []): + node = ContextNode( + node_id=node_data["id"], + node_type=node_data["type"], + content=node_data.get("content", ""), + properties=node_data.get("properties", {}), + metadata=node_data.get("metadata", {}) + ) + self._add_internal_node(node) + + # Add edges + for edge_data in graph_dict.get("edges", []): + edge = ContextEdge( + source_id=edge_data["source"], + target_id=edge_data["target"], + edge_type=edge_data["type"], + weight=edge_data.get("weight", 1.0), + metadata=edge_data.get("metadata", {}) + ) + self._add_internal_edge(edge) + # Decision Support Methods def add_decision(self, decision: "Decision") -> None: """ @@ -870,8 +900,14 @@ class ContextGraph: """ from .decision_models import Decision + # Handle empty decision ID by generating UUID only if None + node_id = decision.decision_id if decision.decision_id is not None else str(uuid.uuid4()) + + # Handle None metadata + metadata = decision.metadata or {} + node = ContextNode( - node_id=decision.decision_id, + node_id=node_id, node_type="Decision", content=decision.scenario, properties={ @@ -883,7 +919,7 @@ class ContextGraph: "decision_maker": decision.decision_maker, "reasoning_embedding": decision.reasoning_embedding, "node2vec_embedding": decision.node2vec_embedding, - **decision.metadata + **metadata } ) self._add_internal_node(node) @@ -906,6 +942,15 @@ class ContextGraph: if relationship_type not in valid_types: raise ValueError(f"Relationship type must be one of: {valid_types}") + # Check if decisions exist - if not, skip adding relationship + if source_decision_id not in self.nodes or target_decision_id not in self.nodes: + return + + # Check if nodes are decision nodes - if not, skip adding relationship + if (self.nodes[source_decision_id].node_type != "Decision" or + self.nodes[target_decision_id].node_type != "Decision"): + return + edge = ContextEdge( source_id=source_decision_id, target_id=target_decision_id, @@ -949,41 +994,49 @@ class ContextGraph: visited.add(current_id) - # Get decision node - if current_id in self.nodes: - node = self.nodes[current_id] - if node.node_type == "Decision": - decision_data = node.properties - decision = Decision( - decision_id=current_id, - category=decision_data.get("category", ""), - scenario=node.content, - reasoning=decision_data.get("reasoning", ""), - outcome=decision_data.get("outcome", ""), - confidence=decision_data.get("confidence", 0.0), - timestamp=datetime.fromisoformat(decision_data.get("timestamp", datetime.now().isoformat())), - decision_maker=decision_data.get("decision_maker", ""), - reasoning_embedding=decision_data.get("reasoning_embedding"), - node2vec_embedding=decision_data.get("node2vec_embedding"), - metadata={k: v for k, v in decision_data.items() if k not in [ - "category", "reasoning", "outcome", "confidence", - "timestamp", "decision_maker", "reasoning_embedding", "node2vec_embedding" - ]} - ) - decision.metadata["causal_distance"] = depth - decisions.append(decision) + # Skip the starting decision - only add connected decisions + if current_id != decision_id: + # Get decision node + if current_id in self.nodes: + node = self.nodes[current_id] + if node.node_type == "Decision": + decision_data = node.properties + decision = Decision( + decision_id=current_id, + category=decision_data.get("category", ""), + scenario=node.content, + reasoning=decision_data.get("reasoning", ""), + outcome=decision_data.get("outcome", ""), + confidence=decision_data.get("confidence", 0.0), + timestamp=datetime.fromisoformat(decision_data.get("timestamp", datetime.now().isoformat())), + decision_maker=decision_data.get("decision_maker", ""), + reasoning_embedding=decision_data.get("reasoning_embedding"), + node2vec_embedding=decision_data.get("node2vec_embedding"), + metadata={k: v for k, v in decision_data.items() if k not in [ + "category", "reasoning", "outcome", "confidence", + "timestamp", "decision_maker", "reasoning_embedding", "node2vec_embedding" + ]} + ) + decision.metadata["causal_distance"] = depth + decisions.append(decision) # Find connected decisions for edge in self.edges: if direction == "upstream": if edge.target_id == current_id and edge.edge_type in ["CAUSED", "INFLUENCED", "PRECEDENT_FOR"]: - if edge.source_id not in visited: + if edge.source_id not in visited and depth < max_depth: queue.append((edge.source_id, depth + 1)) else: # downstream if edge.source_id == current_id and edge.edge_type in ["CAUSED", "INFLUENCED", "PRECEDENT_FOR"]: - if edge.target_id not in visited: + if edge.target_id not in visited and depth < max_depth: queue.append((edge.target_id, depth + 1)) + # Sort by depth for upstream (most distant first) and downstream (closest first) + if direction == "upstream": + decisions.sort(key=lambda d: d.metadata.get("causal_distance", 0), reverse=True) + else: + decisions.sort(key=lambda d: d.metadata.get("causal_distance", 0)) + return decisions def find_precedents(self, decision_id: str, limit: int = 10) -> List["Decision"]: @@ -1000,8 +1053,8 @@ class ContextGraph: # Find decisions connected via PRECEDENT_FOR relationships precedent_ids = [] for edge in self.edges: - if edge.source_id == decision_id and edge.edge_type == "PRECEDENT_FOR": - precedent_ids.append(edge.target_id) + if edge.target_id == decision_id and edge.edge_type == "PRECEDENT_FOR": + precedent_ids.append(edge.source_id) # Convert to Decision objects decisions = [] diff --git a/semantica/context/decision_models.py b/semantica/context/decision_models.py index 59fe6069..ca2de217 100644 --- a/semantica/context/decision_models.py +++ b/semantica/context/decision_models.py @@ -101,7 +101,7 @@ class Decision: def __post_init__(self): """Validate decision data.""" - if not self.decision_id: + if self.decision_id is None: self.decision_id = str(uuid.uuid4()) if not 0 <= self.confidence <= 1: raise ValueError("Confidence must be between 0 and 1") @@ -143,7 +143,7 @@ class DecisionContext: def __post_init__(self): """Validate context data.""" - if not self.context_id: + if self.context_id is None: self.context_id = str(uuid.uuid4()) def to_dict(self) -> Dict[str, Any]: @@ -179,7 +179,7 @@ class Policy: def __post_init__(self): """Validate policy data.""" - if not self.policy_id: + if self.policy_id is None: self.policy_id = str(uuid.uuid4()) def to_dict(self) -> Dict[str, Any]: @@ -220,7 +220,7 @@ class PolicyException: def __post_init__(self): """Validate exception data.""" - if not self.exception_id: + if self.exception_id is None: self.exception_id = str(uuid.uuid4()) def to_dict(self) -> Dict[str, Any]: @@ -256,7 +256,7 @@ class Precedent: def __post_init__(self): """Validate precedent data.""" - if not self.precedent_id: + if self.precedent_id is None: self.precedent_id = str(uuid.uuid4()) if not 0 <= self.similarity_score <= 1: raise ValueError("Similarity score must be between 0 and 1") @@ -294,7 +294,7 @@ class ApprovalChain: def __post_init__(self): """Validate approval data.""" - if not self.approval_id: + if self.approval_id is None: self.approval_id = str(uuid.uuid4()) valid_methods = ["slack_dm", "zoom_call", "email", "system"] if self.approval_method not in valid_methods: diff --git a/tests/context/test_context_graphs_examples.py b/tests/context/test_context_graphs_examples.py new file mode 100644 index 00000000..37fc4e15 --- /dev/null +++ b/tests/context/test_context_graphs_examples.py @@ -0,0 +1,521 @@ +#!/usr/bin/env python3 +""" +Comprehensive test suite for Context Graphs feature examples from issue #290. +This tests all the example use cases provided in the feature description. +""" + +import pytest +import sys +import os +from datetime import datetime +from unittest.mock import Mock, patch + +# Add the semantica package to Python path for testing +sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..')) + +from semantica.context import AgentContext +from semantica.context.context_graph import ContextGraph +from semantica.context.decision_models import Decision, Policy, PolicyException +from semantica.vector_store import VectorStore +from semantica.embeddings import EmbeddingGenerator + + +class TestContextGraphsExamples: + """Test suite for Context Graphs feature examples.""" + + @pytest.fixture + def mock_vector_store(self): + """Create a mock vector store for testing.""" + store = Mock(spec=VectorStore) + store.store = Mock(return_value="test_memory_id") + store.retrieve = Mock(return_value=[]) + store.embed = Mock(return_value=[0.1] * 384) # Mock embedding + return store + + @pytest.fixture + def mock_knowledge_graph(self): + """Create a mock knowledge graph for testing.""" + kg = Mock(spec=ContextGraph) + kg.execute_query = Mock(return_value=[]) + kg.build_from_conversations = Mock(return_value={"statistics": {"node_count": 0, "edge_count": 0}}) + return kg + + def test_context_graph_direct_functionality(self): + """Test ContextGraph directly with decision support features.""" + print("Testing ContextGraph Direct Functionality...") + + # Create context graph with advanced features + graph = ContextGraph( + enable_advanced_analytics=True, + enable_centrality_analysis=True, + enable_community_detection=True, + enable_node_embeddings=True + ) + + # Add a decision + decision = Decision( + decision_id="test_decision_001", + category="test", + scenario="Test scenario for credit approval", + reasoning="Good credit history and stable income", + outcome="approved", + confidence=0.95, + timestamp=datetime.now(), + decision_maker="ai_agent" + ) + + graph.add_decision(decision) + assert len(graph.nodes) == 1 + print("+ Added decision to context graph") + + # Add another decision and causal relationship + decision2 = Decision( + decision_id="test_decision_002", + category="test", + scenario="Related credit decision", + reasoning="Based on previous approval", + outcome="approved", + confidence=0.90, + timestamp=datetime.now(), + decision_maker="ai_agent" + ) + + graph.add_decision(decision2) + graph.add_causal_relationship("test_decision_001", "test_decision_002", "CAUSED") + assert len(graph.nodes) == 2 + assert len(graph.edges) == 1 + print("+ Added causal relationship") + + # Test causal chain + chain = graph.get_causal_chain("test_decision_002", direction="upstream") + assert len(chain) == 1 + assert chain[0].decision_id == "test_decision_001" + print("+ Found causal chain with decisions") + + # Test precedent search + precedents = graph.find_precedents("test_decision_002") + assert len(precedents) == 0 # No precedent relationships added + + # Add precedent relationship and test again + graph.add_causal_relationship("test_decision_001", "test_decision_002", "PRECEDENT_FOR") + precedents = graph.find_precedents("test_decision_002") + assert len(precedents) == 1 + assert precedents[0].decision_id == "test_decision_001" + print("+ Found precedents") + + # Test serialization + graph_dict = graph.to_dict() + assert len(graph_dict['nodes']) == 2 + assert len(graph_dict['edges']) == 2 + assert 'properties' in graph_dict['nodes'][0] + print("+ Serialized graph correctly") + + # Test deserialization + new_graph = ContextGraph() + new_graph.from_dict(graph_dict) + assert len(new_graph.nodes) == 2 + assert len(new_graph.edges) == 2 + print("+ Deserialized graph correctly") + + print("✓ ContextGraph direct functionality test passed") + + def test_financial_services_example(self, mock_vector_store, mock_knowledge_graph): + """Test the financial services example from the feature description.""" + print("Testing Financial Services Example...") + + # Initialize context with decision tracking + context = AgentContext( + vector_store=mock_vector_store, + knowledge_graph=mock_knowledge_graph, + enable_decision_tracking=True, + enable_advanced_analytics=True, + enable_kg_algorithms=True, + enable_vector_store_features=True + ) + + # Credit decision with precedent search + decision_id = context.record_decision( + category="credit_approval", + scenario="High-risk credit limit increase", + reasoning="Past fraud flag with velocity check failure", + outcome="rejected", + confidence=0.788, + entities=["customer:jessica_norris"] + ) + assert decision_id is not None + print("+ Recorded decision") + + # Find similar precedents + precedents = context.find_precedents( + scenario="High-risk customer credit increase", + category="credit_approval", + limit=5 + ) + assert isinstance(precedents, list) + print("+ Found precedents") + + # Analyze causal chain + causal_chain = context.get_causal_chain(decision_id, max_depth=5) + assert isinstance(causal_chain, list) + print("+ Analyzed causal chain") + + print("✓ Financial services example test passed") + + def test_healthcare_example(self, mock_vector_store, mock_knowledge_graph): + """Test the healthcare example from the feature description.""" + print("Testing Healthcare Example...") + + # Initialize context with decision tracking + context = AgentContext( + vector_store=mock_vector_store, + knowledge_graph=mock_knowledge_graph, + enable_decision_tracking=True, + enable_advanced_analytics=True, + enable_kg_algorithms=True, + enable_vector_store_features=True + ) + + # Treatment decision with policy compliance + decision_id = context.record_decision( + category="treatment_plan", + scenario="Diabetic patient with comorbidities", + reasoning="Standard protocol contraindicated due to renal function", + outcome="modified_treatment", + confidence=0.92 + ) + assert decision_id is not None + print("+ Recorded decision") + + # Check policy engine availability + policy_engine = context.get_policy_engine() + if policy_engine: + # Create a test policy + policy = Policy( + policy_id="diabetes_protocol_v2", + name="Diabetes Treatment Protocol v2", + description="Standard treatment protocol for diabetes patients", + rules={"contraindications": ["renal_impairment"], "max_dosage": 100}, + category="treatment", + version="v2", + created_at=datetime.now(), + updated_at=datetime.now() + ) + + # Test policy operations + assert policy.policy_id == "diabetes_protocol_v2" + assert policy.category == "treatment" + print("+ Policy operations working") + + print("✓ Healthcare example test passed") + + def test_legal_example(self, mock_vector_store, mock_knowledge_graph): + """Test the legal example from the feature description.""" + print("Testing Legal Example...") + + # Initialize context with decision tracking + context = AgentContext( + vector_store=mock_vector_store, + knowledge_graph=mock_knowledge_graph, + enable_decision_tracking=True, + enable_advanced_analytics=True, + enable_kg_algorithms=True, + enable_vector_store_features=True + ) + + # Legal decision with precedent analysis + decision_id = context.record_decision( + category="contract_review", + scenario="Non-standard liability clause", + reasoning="Precedent cases show similar clauses upheld", + outcome="approved_with_modifications", + confidence=0.85 + ) + assert decision_id is not None + print("+ Recorded decision") + + # Find legal precedents + precedents = context.find_precedents( + scenario="Liability limitation clauses", + category="contract_review", + limit=10 + ) + assert isinstance(precedents, list) + print("+ Found legal precedents") + + print("✓ Legal example test passed") + + def test_decision_models_functionality(self): + """Test decision models functionality.""" + print("Testing Decision Models...") + + # Test Decision model + decision = Decision( + decision_id="test_decision", + category="test_category", + scenario="Test scenario", + reasoning="Test reasoning", + outcome="approved", + confidence=0.95, + timestamp=datetime.now(), + decision_maker="test_agent" + ) + + assert decision.decision_id == "test_decision" + assert decision.category == "test_category" + assert 0 <= decision.confidence <= 1 + + # Test serialization + decision_dict = decision.to_dict() + assert decision_dict["decision_id"] == "test_decision" + assert "timestamp" in decision_dict + + # Test deserialization + restored_decision = Decision.from_dict(decision_dict) + assert restored_decision.decision_id == decision.decision_id + assert restored_decision.category == decision.category + print("+ Decision model serialization working") + + # Test Policy model + policy = Policy( + policy_id="test_policy", + name="Test Policy", + description="Test policy description", + rules={"max_amount": 1000}, + category="test", + version="1.0", + created_at=datetime.now(), + updated_at=datetime.now() + ) + + assert policy.policy_id == "test_policy" + assert policy.rules["max_amount"] == 1000 + + # Test PolicyException model + exception = PolicyException( + exception_id="test_exception", + decision_id="test_decision", + policy_id="test_policy", + reason="Test exception", + approver="test_approver", + approval_timestamp=datetime.now(), + justification="Test justification" + ) + + assert exception.exception_id == "test_exception" + assert exception.decision_id == "test_decision" + print("+ Policy models working") + + print("✓ Decision models functionality test passed") + + def test_context_graph_edge_cases(self): + """Test ContextGraph edge cases and error handling.""" + print("Testing ContextGraph Edge Cases...") + + graph = ContextGraph() + + # Test empty decision ID handling + decision_empty_id = Decision( + decision_id="", # Empty ID + category="test", + scenario="test scenario", + reasoning="test reasoning", + outcome="test outcome", + confidence=0.8, + timestamp=datetime.now(), + decision_maker="test_agent" + ) + + graph.add_decision(decision_empty_id) + assert "" in graph.nodes # Empty string should be preserved as key + print("+ Empty decision ID handling working") + + # Test None decision ID handling + decision_none_id = Decision( + decision_id=None, # None ID + category="test", + scenario="test scenario 2", + reasoning="test reasoning 2", + outcome="test outcome 2", + confidence=0.8, + timestamp=datetime.now(), + decision_maker="test_agent" + ) + + graph.add_decision(decision_none_id) + assert len(graph.nodes) == 2 # Should have generated UUID + print("+ None decision ID handling working") + + # Test causal relationship with nonexistent nodes (should not raise error) + graph.add_causal_relationship("nonexistent1", "nonexistent2", "CAUSED") + assert len(graph.edges) == 0 # Should not add relationship + print("+ Nonexistent node handling working") + + # Test invalid relationship type + with pytest.raises(ValueError): + graph.add_causal_relationship("test", "test2", "INVALID_TYPE") + print("+ Invalid relationship type validation working") + + # Test causal chain with nonexistent decision + chain = graph.get_causal_chain("nonexistent", direction="upstream") + assert len(chain) == 0 + print("+ Nonexistent decision handling working") + + print("✓ ContextGraph edge cases test passed") + + def test_advanced_features_integration(self, mock_vector_store, mock_knowledge_graph): + """Test advanced features integration.""" + print("Testing Advanced Features Integration...") + + # Test with all features enabled + context = AgentContext( + vector_store=mock_vector_store, + knowledge_graph=mock_knowledge_graph, + enable_decision_tracking=True, + enable_advanced_analytics=True, + enable_kg_algorithms=True, + enable_vector_store_features=True, + use_graph_expansion=True, + max_expansion_hops=3, + hybrid_alpha=0.7 + ) + + # Verify configuration + assert context.config["enable_decision_tracking"] is True + assert context.config["enable_advanced_analytics"] is True + assert context.config["enable_kg_algorithms"] is True + assert context.config["enable_vector_store_features"] is True + assert context.config["use_graph_expansion"] is True + assert context.config["max_expansion_hops"] == 3 + assert context.config["hybrid_alpha"] == 0.7 + print("+ Configuration validation working") + + # Test decision tracking with advanced features + decision_id = context.record_decision( + category="advanced_test", + scenario="Advanced feature test scenario", + reasoning="Testing advanced analytics integration", + outcome="processed", + confidence=0.88, + entities=["entity1", "entity2"] + ) + + assert decision_id is not None + print("+ Advanced decision recording working") + + # Test context insights + insights = context.get_context_insights() + assert isinstance(insights, dict) + print("+ Context insights working") + + print("✓ Advanced features integration test passed") + + +class TestContextGraphsPerformance: + """Performance tests for Context Graphs feature.""" + + def test_large_decision_network(self): + """Test handling of large decision networks.""" + print("Testing Large Decision Network...") + + graph = ContextGraph() + + # Create a network of 100 decisions + decisions = [] + for i in range(100): + decision = Decision( + decision_id=f"decision_{i:03d}", + category="performance_test", + scenario=f"Performance test scenario {i}", + reasoning=f"Performance test reasoning {i}", + outcome="processed", + confidence=0.8 + (i % 20) * 0.01, # Varying confidence + timestamp=datetime.now(), + decision_maker="performance_agent" + ) + decisions.append(decision) + graph.add_decision(decision) + + assert len(graph.nodes) == 100 + print("+ Created 100 decisions") + + # Add causal relationships to create a network + for i in range(99): + # Create a mix of relationship types + relationship_type = ["CAUSED", "INFLUENCED", "PRECEDENT_FOR"][i % 3] + graph.add_causal_relationship(f"decision_{i:03d}", f"decision_{i+1:03d}", relationship_type) + + assert len(graph.edges) == 99 + print("+ Created 99 causal relationships") + + # Test causal chain performance + chain = graph.get_causal_chain("decision_099", direction="upstream", max_depth=50) + assert len(chain) > 0 + print("+ Causal chain analysis working") + + # Test precedent search performance + precedents = graph.find_precedents("decision_050", limit=20) + assert isinstance(precedents, list) + print("+ Precedent search working") + + # Test serialization performance + graph_dict = graph.to_dict() + assert len(graph_dict['nodes']) == 100 + assert len(graph_dict['edges']) == 99 + print("+ Large graph serialization working") + + print("✓ Large decision network test passed") + + def test_concurrent_operations(self): + """Test concurrent decision operations.""" + print("Testing Concurrent Operations...") + + import threading + import time + + graph = ContextGraph() + results = [] + errors = [] + + def add_decisions(start_id, count): + """Add decisions in a separate thread.""" + try: + for i in range(count): + decision = Decision( + decision_id=f"concurrent_decision_{start_id + i:03d}", + category="concurrent_test", + scenario=f"Concurrent test {start_id + i}", + reasoning="Concurrent reasoning", + outcome="processed", + confidence=0.8, + timestamp=datetime.now(), + decision_maker="concurrent_agent" + ) + graph.add_decision(decision) + time.sleep(0.001) # Small delay to simulate real work + results.append(f"Thread {start_id} completed") + except Exception as e: + errors.append(f"Thread {start_id} error: {e}") + + # Create multiple threads + threads = [] + for i in range(5): + thread = threading.Thread(target=add_decisions, args=(i * 20, 20)) + threads.append(thread) + thread.start() + + # Wait for all threads to complete + for thread in threads: + thread.join() + + # Verify results + assert len(errors) == 0, f"Errors occurred: {errors}" + assert len(results) == 5 + assert len(graph.nodes) == 100 # 5 threads * 20 decisions each + print("+ Concurrent operations completed successfully") + + print("✓ Concurrent operations test passed") + + +if __name__ == "__main__": + # Run tests when script is executed directly + pytest.main([__file__, "-v"])