From 8177d887538560d137edc705c7d60abcb9e23faa Mon Sep 17 00:00:00 2001 From: "Guofang.Tang" <136770748@qq.com> Date: Sun, 16 Aug 2026 14:23:24 +0800 Subject: [PATCH] fix(kg): preserve isolated nodes in graph analytics (#1011) * fix(kg): preserve isolated nodes in graph analytics * fix(kg): support node fallbacks and community payloads --------- --- semantica/kg/_graph_view.py | 206 ++++++++++++++++++++++++++ semantica/kg/centrality_calculator.py | 72 +-------- semantica/kg/community_detector.py | 127 ++++++---------- semantica/kg/connectivity_analyzer.py | 49 +----- tests/kg/test_analytics_node_scope.py | 112 ++++++++++++++ 5 files changed, 374 insertions(+), 192 deletions(-) create mode 100644 semantica/kg/_graph_view.py create mode 100644 tests/kg/test_analytics_node_scope.py diff --git a/semantica/kg/_graph_view.py b/semantica/kg/_graph_view.py new file mode 100644 index 00000000..f467125b --- /dev/null +++ b/semantica/kg/_graph_view.py @@ -0,0 +1,206 @@ +"""Internal graph view helpers shared by KG analytics modules.""" + +from dataclasses import dataclass +from typing import Any, Dict, Iterable, List, Optional, Set, Tuple + + +@dataclass +class GraphView: + """Normalized node and edge view used by graph analytics.""" + + nodes: List[Any] + edges: List[Tuple[Any, Any]] + + +def build_graph_view(graph: Any) -> GraphView: + """Build a graph view without dropping explicitly declared nodes. + + Graph analytics accepts graph dictionaries, ContextGraph-like objects, and + NetworkX graphs. Nodes declared without an incident edge remain in the + returned view so callers can choose how to handle isolated nodes. + """ + nodes: List[Any] = [] + edges: List[Tuple[Any, Any]] = [] + seen_nodes: Set[Any] = set() + seen_edges: Set[Tuple[Any, Any]] = set() + + def add_node(value: Any) -> Optional[Any]: + node_id = _node_id(value) + if node_id is None or node_id == "": + return None + if node_id not in seen_nodes: + seen_nodes.add(node_id) + nodes.append(node_id) + return node_id + + for node in _extract_nodes(graph): + add_node(node) + + for raw_edge in _extract_edges(graph): + edge = _edge_endpoints(raw_edge) + if edge is None: + continue + source, target = edge + source = add_node(source) + target = add_node(target) + if source is None or target is None: + continue + if (source, target) not in seen_edges: + seen_edges.add((source, target)) + edges.append((source, target)) + + return GraphView(nodes=nodes, edges=edges) + + +def build_adjacency(graph: Any, directed: bool = False) -> Dict[Any, List[Any]]: + """Build an adjacency list while preserving isolated graph nodes.""" + view = build_graph_view(graph) + adjacency: Dict[Any, List[Any]] = {node: [] for node in view.nodes} + + for source, target in view.edges: + if target not in adjacency[source]: + adjacency[source].append(target) + if not directed and source not in adjacency[target]: + adjacency[target].append(source) + + return adjacency + + +def _extract_nodes(graph: Any) -> Iterable[Any]: + if isinstance(graph, dict): + raw_nodes: List[Any] = [] + for key in ("entities", "nodes"): + values = graph.get(key, []) + if isinstance(values, dict): + raw_nodes.extend(values.keys()) + elif values: + raw_nodes.extend(values) + return raw_nodes + + raw_nodes = getattr(graph, "nodes", None) + if callable(raw_nodes): + return raw_nodes() + if isinstance(raw_nodes, dict): + return raw_nodes.keys() + if raw_nodes is not None: + return raw_nodes + + get_nodes = getattr(graph, "get_nodes", None) + if callable(get_nodes): + return get_nodes() + return [] + + +def _extract_edges(graph: Any) -> Iterable[Any]: + if isinstance(graph, dict): + raw_edges: List[Any] = [] + for key in ("relationships", "edges"): + values = graph.get(key, []) + if values: + raw_edges.extend(values) + return raw_edges + + raw_edges: List[Any] = [] + relationships = getattr(graph, "relationships", None) + if relationships is not None: + raw_edges.extend(relationships) + edges = getattr(graph, "edges", None) + if callable(edges): + raw_edges.extend(edges()) + elif edges is not None: + raw_edges.extend(edges) + if raw_edges: + return raw_edges + + get_relationships = getattr(graph, "get_relationships", None) + if callable(get_relationships): + return get_relationships() + return [] + + +def _edge_endpoints(edge: Any) -> Optional[Tuple[Any, Any]]: + if isinstance(edge, (tuple, list)) and len(edge) >= 2: + return edge[0], edge[1] + + if isinstance(edge, dict): + source = _first_value( + edge, + "source", + "source_id", + "subject", + "start", + "start_id", + "from", + "src", + "START_ID", + ":START_ID", + ) + target = _first_value( + edge, + "target", + "target_id", + "object", + "end", + "end_id", + "to", + "dst", + "END_ID", + ":END_ID", + ) + else: + source = _first_attribute( + edge, + "source_id", + "source", + "subject", + "start", + "start_id", + "from_id", + ) + target = _first_attribute( + edge, + "target_id", + "target", + "object", + "end", + "end_id", + "to_id", + ) + + if source is None or target is None: + return None + return source, target + + +def _node_id(value: Any) -> Any: + if isinstance(value, dict): + value = _first_value( + value, "id", "node_id", "entity_id", "key", "name", "text" + ) + elif not isinstance(value, (str, int, float, bool, bytes, tuple)): + value = _first_attribute( + value, "node_id", "id", "entity_id", "key", "name", "text" + ) + + if value is None: + return None + try: + hash(value) + except TypeError: + return str(value) + return value + + +def _first_value(mapping: Dict[str, Any], *keys: str) -> Any: + for key in keys: + if key in mapping and mapping[key] not in (None, ""): + return mapping[key] + return None + + +def _first_attribute(value: Any, *names: str) -> Any: + for name in names: + attribute = getattr(value, name, None) + if attribute not in (None, ""): + return attribute + return None diff --git a/semantica/kg/centrality_calculator.py b/semantica/kg/centrality_calculator.py index 9fe9a956..ac59db6e 100644 --- a/semantica/kg/centrality_calculator.py +++ b/semantica/kg/centrality_calculator.py @@ -43,7 +43,7 @@ Author: Semantica Contributors License: MIT """ -from collections import defaultdict, deque +from collections import deque from typing import Any, Dict, List, Optional import numpy as np @@ -51,6 +51,7 @@ from scipy import sparse from ..utils.logging import get_logger from ..utils.progress_tracker import get_progress_tracker +from ._graph_view import build_adjacency, build_graph_view class CentralityCalculator: @@ -518,76 +519,15 @@ class CentralityCalculator: def _build_adjacency(self, graph) -> Dict[str, List[str]]: """Build adjacency list from graph.""" - adjacency = defaultdict(list) - - # Extract relationships - relationships = [] - if hasattr(graph, "relationships"): - relationships = graph.relationships - elif hasattr(graph, "get_relationships"): - relationships = graph.get_relationships() - elif isinstance(graph, dict): - relationships = graph.get("relationships", graph.get("edges", [])) - elif hasattr(graph, "edges") and not callable(graph.edges): - # ContextGraph-style: edges is a list of dataclass objects with source_id/target_id - for edge in (graph.edges or []): - if isinstance(edge, dict): - src = edge.get("source") or edge.get("source_id") - tgt = edge.get("target") or edge.get("target_id") - else: - src = getattr(edge, "source_id", None) or getattr(edge, "source", None) - tgt = getattr(edge, "target_id", None) or getattr(edge, "target", None) - if src and tgt: - src, tgt = str(src), str(tgt) - if tgt not in adjacency[src]: - adjacency[src].append(tgt) - if src not in adjacency[tgt]: - adjacency[tgt].append(src) - return dict(adjacency) - - # Build adjacency - for rel in relationships: - # Handle tuple/list edges (e.g., from NetworkX) - if isinstance(rel, (tuple, list)) and len(rel) >= 2: - source, target = str(rel[0]), str(rel[1]) - if source and target: - if target not in adjacency[source]: - adjacency[source].append(target) - if source not in adjacency[target]: - adjacency[target].append(source) - continue - source = rel.get("source") or rel.get("subject") - target = rel.get("target") or rel.get("object") - - # Extract IDs if objects are passed - if source and not isinstance(source, (str, int, float)): - if isinstance(source, dict): - source = source.get("id") or source.get("entity_id") or source.get("text") or str(source) - else: - source = getattr(source, "id", getattr(source, "text", str(source))) - - if target and not isinstance(target, (str, int, float)): - if isinstance(target, dict): - target = target.get("id") or target.get("entity_id") or target.get("text") or str(target) - else: - target = getattr(target, "id", getattr(target, "text", str(target))) - - if source and target: - if target not in adjacency[source]: - adjacency[source].append(target) - if source not in adjacency[target]: - adjacency[target].append(source) - - return dict(adjacency) + return build_adjacency(graph) def _to_networkx(self, graph): """Convert graph to NetworkX format.""" - adjacency = self._build_adjacency(graph) + view = build_graph_view(graph) nx_graph = self.nx.Graph() - for source, targets in adjacency.items(): - for target in targets: - nx_graph.add_edge(source, target) + nx_graph.add_nodes_from(view.nodes) + nx_graph.add_edges_from(view.edges) return nx_graph diff --git a/semantica/kg/community_detector.py b/semantica/kg/community_detector.py index 8aaaa236..01fa6064 100644 --- a/semantica/kg/community_detector.py +++ b/semantica/kg/community_detector.py @@ -49,6 +49,16 @@ from typing import Any, Dict, List, Optional from ..utils.logging import get_logger from ..utils.progress_tracker import get_progress_tracker +from ._graph_view import build_adjacency, build_graph_view + + +def _is_hashable(value: Any) -> bool: + """Return whether a community identifier can be used in a set.""" + try: + hash(value) + except TypeError: + return False + return True class CommunityDetector: @@ -157,17 +167,18 @@ class CommunityDetector: nx_graph = self._to_networkx(graph) - # Check if graph is empty or has no edges + # An empty graph has no communities. A graph with nodes but + # no edges still has singleton communities. num_nodes = nx_graph.number_of_nodes() num_edges = nx_graph.number_of_edges() self.logger.debug(f"Graph stats: nodes={num_nodes}, edges={num_edges}") - if num_nodes == 0 or num_edges == 0: - self.logger.warning("Graph is empty or has no edges, returning 0 communities") + if num_nodes == 0: + self.logger.warning("Graph is empty, returning 0 communities") self.progress_tracker.stop_tracking( tracking_id, status="completed", - message="Detected 0 communities (empty graph/no edges)", + message="Detected 0 communities (empty graph)", ) return { "communities": [], @@ -350,17 +361,7 @@ class CommunityDetector: adjacency = self._build_adjacency(graph) - # Extract community structure - if isinstance(communities, dict): - node_communities = communities - elif isinstance(communities, dict) and "node_assignments" in communities: - node_communities = communities["node_assignments"] - else: - # Convert list of communities to node assignments - node_communities = {} - for i, community in enumerate(communities): - for node in community: - node_communities[node] = i + node_communities = self._to_node_assignments(communities) # Calculate metrics num_communities = len(set(node_communities.values())) @@ -408,16 +409,7 @@ class CommunityDetector: metrics = self.calculate_community_metrics(graph, communities) - # Extract node assignments - if isinstance(communities, dict) and "node_assignments" in communities: - node_communities = communities["node_assignments"] - elif isinstance(communities, dict): - node_communities = communities - else: - node_communities = {} - for i, community in enumerate(communities): - for node in community: - node_communities[node] = i + node_communities = self._to_node_assignments(communities) # Analyze connectivity between communities adjacency = self._build_adjacency(graph) @@ -440,6 +432,32 @@ class CommunityDetector: "edge_ratio": intra_community_edges / (inter_community_edges + 1), } + @staticmethod + def _to_node_assignments(communities: Any) -> Dict[Any, Any]: + """Normalize community results to a node-to-community mapping.""" + if isinstance(communities, dict): + assignments = communities.get("node_assignments") + if isinstance(assignments, dict): + return assignments + + detected_communities = communities.get("communities") + if isinstance(detected_communities, (list, tuple)): + communities = detected_communities + elif "communities" in communities: + raise ValueError("Community results must contain a list of communities") + elif not all(_is_hashable(value) for value in communities.values()): + raise ValueError( + "Community assignments must map nodes to hashable community IDs" + ) + else: + return communities + + node_assignments: Dict[Any, Any] = {} + for community_id, community in enumerate(communities or []): + for node in community: + node_assignments[node] = community_id + return node_assignments + def detect_communities( self, graph: Any, algorithm: str = "louvain", method: str = None, **options ) -> Dict[str, Any]: @@ -478,57 +496,7 @@ class CommunityDetector: def _build_adjacency(self, graph) -> Dict[str, List[str]]: """Build adjacency list from graph.""" - from collections import defaultdict - - adjacency = defaultdict(list) - - # Extract relationships - relationships = [] - raw_edges = [] # flat (u, v) tuples - if hasattr(graph, "relationships"): - relationships = graph.relationships - elif hasattr(graph, "get_relationships"): - relationships = graph.get_relationships() - elif isinstance(graph, dict): - relationships = graph.get("relationships", []) - # Also handle 'edges' key (list of tuples or dicts) - for edge in graph.get("edges", []): - if isinstance(edge, (list, tuple)) and len(edge) >= 2: - raw_edges.append((str(edge[0]), str(edge[1]))) - elif isinstance(edge, dict): - relationships.append(edge) - - # Add raw (u, v) edges - for u, v in raw_edges: - if u and v: - adjacency[u].append(v) - adjacency[v].append(u) - - # Build adjacency - for rel in relationships: - source = rel.get("source") or rel.get("subject") - target = rel.get("target") or rel.get("object") - - # Extract IDs if objects are passed - if source and not isinstance(source, (str, int, float)): - if isinstance(source, dict): - source = source.get("id") or source.get("entity_id") or source.get("text") or str(source) - else: - source = getattr(source, "id", getattr(source, "text", str(source))) - - if target and not isinstance(target, (str, int, float)): - if isinstance(target, dict): - target = target.get("id") or target.get("entity_id") or target.get("text") or str(target) - else: - target = getattr(target, "id", getattr(target, "text", str(target))) - - if source and target: - if target not in adjacency[source]: - adjacency[source].append(target) - if source not in adjacency[target]: - adjacency[target].append(source) - - return dict(adjacency) + return build_adjacency(graph) def _to_networkx(self, graph): """Convert graph to NetworkX format.""" @@ -536,12 +504,11 @@ class CommunityDetector: if hasattr(graph, 'nodes') and hasattr(graph, 'edges') and hasattr(graph, 'number_of_nodes'): return graph - adjacency = self._build_adjacency(graph) + view = build_graph_view(graph) nx_graph = self.nx.Graph() - for source, targets in adjacency.items(): - for target in targets: - nx_graph.add_edge(source, target) + nx_graph.add_nodes_from(view.nodes) + nx_graph.add_edges_from(view.edges) return nx_graph diff --git a/semantica/kg/connectivity_analyzer.py b/semantica/kg/connectivity_analyzer.py index 00d2ad22..463c9ed5 100644 --- a/semantica/kg/connectivity_analyzer.py +++ b/semantica/kg/connectivity_analyzer.py @@ -48,11 +48,12 @@ Author: Semantica Contributors License: MIT """ -from collections import defaultdict, deque +from collections import deque from typing import Any, Dict, List, Optional, Set, Tuple from ..utils.logging import get_logger from ..utils.progress_tracker import get_progress_tracker +from ._graph_view import build_adjacency class ConnectivityAnalyzer: @@ -385,51 +386,7 @@ class ConnectivityAnalyzer: def _build_adjacency(self, graph) -> Dict[str, List[str]]: """Build adjacency list from graph.""" - adjacency = defaultdict(list) - - # Extract relationships - relationships = [] - if hasattr(graph, "relationships"): - relationships = graph.relationships - elif hasattr(graph, "get_relationships"): - relationships = graph.get_relationships() - elif isinstance(graph, dict): - relationships = graph.get("relationships", graph.get("edges", [])) - - # Build adjacency - for rel in relationships: - # Handle tuple/list edges (e.g., from NetworkX) - if isinstance(rel, (tuple, list)) and len(rel) >= 2: - source, target = str(rel[0]), str(rel[1]) - if source and target: - if target not in adjacency[source]: - adjacency[source].append(target) - if source not in adjacency[target]: - adjacency[target].append(source) - continue - source = rel.get("source") or rel.get("subject") - target = rel.get("target") or rel.get("object") - - # Extract IDs if objects are passed - if source and not isinstance(source, (str, int, float)): - if isinstance(source, dict): - source = source.get("id") or source.get("entity_id") or source.get("text") or str(source) - else: - source = getattr(source, "id", getattr(source, "text", str(source))) - - if target and not isinstance(target, (str, int, float)): - if isinstance(target, dict): - target = target.get("id") or target.get("entity_id") or target.get("text") or str(target) - else: - target = getattr(target, "id", getattr(target, "text", str(target))) - - if source and target: - if target not in adjacency[source]: - adjacency[source].append(target) - if source not in adjacency[target]: - adjacency[target].append(source) - - return dict(adjacency) + return build_adjacency(graph) def _bfs_shortest_path( self, adjacency: Dict[str, List[str]], source: str, target: str diff --git a/tests/kg/test_analytics_node_scope.py b/tests/kg/test_analytics_node_scope.py new file mode 100644 index 00000000..fda5d4f8 --- /dev/null +++ b/tests/kg/test_analytics_node_scope.py @@ -0,0 +1,112 @@ +"""Regression tests for KG analytics node scope handling.""" + +import networkx as nx + +from semantica.kg.centrality_calculator import CentralityCalculator +from semantica.kg.community_detector import CommunityDetector +from semantica.kg.connectivity_analyzer import ConnectivityAnalyzer + + +def _graph_with_isolated_node(): + return { + "entities": [{"id": "A"}, {"id": "B"}, {"id": "C"}], + "relationships": [{"source": "A", "target": "B"}], + } + + +def test_centrality_keeps_declared_isolated_nodes(): + result = CentralityCalculator().calculate_degree_centrality( + _graph_with_isolated_node() + ) + + assert result["total_nodes"] == 3 + assert result["centrality"]["C"] == 0.0 + + +def test_connectivity_reports_declared_isolated_nodes(): + result = ConnectivityAnalyzer().analyze_connectivity( + _graph_with_isolated_node() + ) + + assert result["num_nodes"] == 3 + assert result["num_components"] == 2 + assert ["C"] in result["components"] + assert result["is_connected"] is False + + +def test_community_detection_keeps_declared_isolated_nodes(): + detector = CommunityDetector() + result = detector.detect_communities(_graph_with_isolated_node()) + + assert set(result["node_assignments"]) == {"A", "B", "C"} + metrics = detector.calculate_community_metrics( + _graph_with_isolated_node(), result + ) + assert metrics["num_communities"] == 2 + structure = detector.analyze_community_structure( + _graph_with_isolated_node(), result + ) + assert structure["num_communities"] == 2 + + +def test_community_detection_returns_singletons_for_edgeless_graph(): + graph = {"entities": [{"id": "A"}, {"id": "B"}], "relationships": []} + + result = CommunityDetector().detect_communities(graph) + + assert {frozenset(community) for community in result["communities"]} == { + frozenset({"A"}), + frozenset({"B"}), + } + + +def test_networkx_graph_keeps_isolated_nodes_for_analytics(): + graph = nx.Graph() + graph.add_nodes_from(["A", "B", "C"]) + graph.add_edge("A", "B") + + centrality = CentralityCalculator().calculate_degree_centrality(graph) + connectivity = ConnectivityAnalyzer().analyze_connectivity(graph) + + assert centrality["total_nodes"] == 3 + assert centrality["centrality"]["C"] == 0.0 + assert connectivity["num_nodes"] == 3 + assert connectivity["num_components"] == 2 + + +def test_nodes_edges_payload_keeps_declared_isolated_nodes(): + graph = { + "nodes": [{"id": "A"}, {"id": "B"}, {"id": "C"}], + "edges": [("A", "B")], + } + + result = CentralityCalculator().calculate_degree_centrality(graph) + + assert result["total_nodes"] == 3 + assert result["centrality"]["C"] == 0.0 + + +def test_name_and_text_nodes_are_kept_when_ids_are_missing(): + graph = { + "entities": [{"name": "Alice"}, {"text": "Bob"}], + "relationships": [], + } + + result = CentralityCalculator().calculate_degree_centrality(graph) + + assert result["total_nodes"] == 2 + assert set(result["centrality"]) == {"Alice", "Bob"} + + +def test_community_metrics_accepts_communities_payload(): + detector = CommunityDetector() + graph = { + "entities": [{"id": "A"}, {"id": "B"}, {"id": "C"}], + "relationships": [{"source": "A", "target": "B"}], + } + result = {"communities": [["A", "B"], ["C"]]} + + metrics = detector.calculate_community_metrics(graph, result) + + assert metrics["num_communities"] == 2 + assert metrics["community_sizes"] == {0: 2, 1: 1}