Merge pull request #474 from Hawksight-AI/kg

feat(kg): Native `KnowledgeGraph` Support in `KGVisualizer`

CHANGELOG.md

@@ -7,6 +7,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+- **Enhancement: Native `KnowledgeGraph` type support in `KGVisualizer`** (PR `kg` by @KaifAhmad1, closes #471): Added `semantica/kg/knowledge_graph.py` — a formal `KnowledgeGraph` dataclass (`entities`, `relationships`, `metadata`) that is now the canonical in-memory type produced and consumed by the Semantica KG pipeline. Exported from `semantica.kg`. `KGVisualizer` gains `_convert_knowledge_graph()` — an authoritative, non-mutating conversion path from `KnowledgeGraph` to the internal dict format — and `_normalize_graph()` now routes `isinstance(graph, KnowledgeGraph)` through it as an explicit fast-path before duck-typing. All five public entry points (`visualize_network`, `visualize_communities`, `visualize_centrality`, `visualize_entity_types`, `visualize_relationship_matrix`) accept `KnowledgeGraph` directly; no manual conversion required. All existing callers passing dicts or duck-typed objects are unaffected. 15 new tests in `TestFormalKnowledgeGraphType` (conversion shape, non-mutation, determinism, routing, all five entry points, import availability).
+
 - **Fix: `KGVisualizer` now accepts `KnowledgeGraph` objects in all `visualize_*` methods** (PR `visualization` by @KaifAhmad1, closes #458): All five public methods (`visualize_network`, `visualize_communities`, `visualize_centrality`, `visualize_entity_types`, `visualize_relationship_matrix`) previously called `graph.get("entities", [])`, silently producing no output when passed a non-dict object. Added `_normalize_graph()` which duck-types the input — dicts pass through unchanged; any object exposing `.entities` / `.relationships` attributes (e.g. the result of `GraphBuilder.build()`) is converted to the canonical dict form; anything else raises a clear `ProcessingError` naming the offending type. 21 tests added in `tests/visualization/test_kg_visualizer_normalize_graph.py`.
 
 - **Security: 12 vulnerability fixes across CRITICAL → LOW severity** (PR `security-enhancement` by @KaifAhmad1):

semantica/kg/__init__.py

@@ -126,12 +126,14 @@ from .temporal_query import (
     TemporalPatternDetector,
     TemporalVersionManager,
 )
+from .knowledge_graph import KnowledgeGraph
 from .temporal_model import BiTemporalFact, TemporalBound
 from .temporal_normalizer import TemporalNormalizer
 from .temporal_query_rewriter import TemporalQueryRewriter, TemporalQueryResult
 
 __all__ = [
     # Core Classes
+    "KnowledgeGraph",
     "GraphBuilder",
     "GraphBuilderWithProvenance",
     "EntityResolver",

semantica/kg/knowledge_graph.py

@@ -0,0 +1,46 @@
+"""
+KnowledgeGraph dataclass — canonical in-memory representation.
+
+This is the formal type produced by the Semantica KG pipeline and consumed
+by visualizers, exporters, and other downstream components.  It is a thin,
+immutable-friendly wrapper around three plain collections so that isinstance
+checks, type hints, and IDEs can surface the type rather than relying on
+bare dicts.
+
+Keeping this in its own file avoids circular imports between the kg and
+visualization sub-packages.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any, Dict, List
+
+
+@dataclass
+class KnowledgeGraph:
+    """
+    Canonical in-memory knowledge graph.
+
+    Attributes:
+        entities:      List of entity dicts with at minimum ``id`` and ``type`` keys.
+        relationships: List of relationship dicts with at minimum ``source``,
+                       ``target``, and ``type`` keys.
+        metadata:      Arbitrary graph-level metadata (e.g. build timestamps,
+                       entity-resolution flags).
+    """
+
+    entities: List[Dict[str, Any]] = field(default_factory=list)
+    relationships: List[Dict[str, Any]] = field(default_factory=list)
+    metadata: Dict[str, Any] = field(default_factory=dict)
+
+    # ------------------------------------------------------------------
+    # Convenience helpers
+    # ------------------------------------------------------------------
+
+    def __len__(self) -> int:
+        """Return the number of entities (mirrors the most common 'size' query)."""
+        return len(self.entities)
+
+    def __bool__(self) -> bool:
+        return bool(self.entities or self.relationships)

semantica/visualization/kg_visualizer.py

@@ -53,6 +53,14 @@ except (ImportError, OSError):
 
 from ..utils.exceptions import ProcessingError
 from ..utils.logging import get_logger
+
+# Optional import — keeps the visualizer usable even if the kg sub-package
+# is not installed, and avoids circular-import risk at module level.
+try:
+    from ..kg.knowledge_graph import KnowledgeGraph as _KnowledgeGraph
+except Exception:  # pragma: no cover
+    _KnowledgeGraph = None  # type: ignore[assignment,misc]
+
 from ..utils.progress_tracker import get_progress_tracker
 from .utils.color_schemes import ColorPalette, ColorScheme
 from .utils.export_formats import (
@@ -118,18 +126,45 @@ class KGVisualizer:
                 "Install with: pip install plotly"
             )
 
-    def _normalize_graph(self, graph: Any) -> Dict[str, Any]:
+    def _convert_knowledge_graph(self, kg: Any) -> Dict[str, Any]:
         """
-        Normalize graph input to the expected dict format.
+        Convert a KnowledgeGraph instance to the internal dict format.
 
-        Accepts either:
-        - A dict with "entities" and "relationships" keys (canonical format)
-        - Any object that exposes .entities and .relationships attributes
-          (e.g. a KnowledgeGraph dataclass returned by GraphBuilder.build())
+        Non-mutating. Preserves node types, labels, properties, edge types,
+        weights, and direction.
+
+        Args:
+            kg: A ``KnowledgeGraph`` instance.
 
         Returns:
             Dict with "entities", "relationships", and "metadata" keys.
         """
+        entities = getattr(kg, "entities", None) or []
+        relationships = getattr(kg, "relationships", None) or []
+        metadata = getattr(kg, "metadata", None) or {}
+        return {
+            "entities": list(entities),
+            "relationships": list(relationships),
+            "metadata": dict(metadata),
+        }
+
+    def _normalize_graph(self, graph: Any) -> Dict[str, Any]:
+        """
+        Normalize graph input to the expected dict format.
+
+        Accepts:
+        - A ``KnowledgeGraph`` instance (routed through ``_convert_knowledge_graph``)
+        - A dict with "entities" and "relationships" keys (canonical format)
+        - Any object that exposes .entities and .relationships attributes
+          (duck-typed, e.g. custom dataclasses)
+
+        Returns:
+            Dict with "entities", "relationships", and "metadata" keys.
+        """
+        # Explicit fast-path for the formal KnowledgeGraph type
+        if _KnowledgeGraph is not None and isinstance(graph, _KnowledgeGraph):
+            return self._convert_knowledge_graph(graph)
+
         if isinstance(graph, dict):
             return graph
 

tests/visualization/test_kg_visualizer_normalize_graph.py

@@ -283,5 +283,166 @@ class TestAllVisualizeMethodsAcceptKGObject(unittest.TestCase):
         self.viz._normalize_graph.assert_called_once_with(self.kg)
 
 
+# ---------------------------------------------------------------------------
+# Issue #471 — formal KnowledgeGraph type support
+# ---------------------------------------------------------------------------
+
+class TestFormalKnowledgeGraphType(unittest.TestCase):
+    """
+    Regression tests for issue #471.
+
+    The formal ``semantica.kg.KnowledgeGraph`` dataclass must be accepted by
+    every public visualize_* method without requiring any manual conversion.
+    """
+
+    @classmethod
+    def setUpClass(cls):
+        try:
+            from semantica.kg.knowledge_graph import KnowledgeGraph
+            cls.KnowledgeGraph = KnowledgeGraph
+        except ImportError:
+            cls.KnowledgeGraph = None
+
+    def _make_kg(self):
+        if self.KnowledgeGraph is None:
+            self.skipTest("semantica.kg.KnowledgeGraph not available")
+        return self.KnowledgeGraph(
+            entities=ENTITIES,
+            relationships=RELATIONSHIPS,
+            metadata={"version": "test"},
+        )
+
+    def test_convert_knowledge_graph_entities(self):
+        kg = self._make_kg()
+        viz = _make_viz()
+        result = viz._convert_knowledge_graph(kg)
+        self.assertEqual(result["entities"], ENTITIES)
+
+    def test_convert_knowledge_graph_relationships(self):
+        kg = self._make_kg()
+        viz = _make_viz()
+        result = viz._convert_knowledge_graph(kg)
+        self.assertEqual(result["relationships"], RELATIONSHIPS)
+
+    def test_convert_knowledge_graph_metadata(self):
+        kg = self._make_kg()
+        viz = _make_viz()
+        result = viz._convert_knowledge_graph(kg)
+        self.assertEqual(result["metadata"], {"version": "test"})
+
+    def test_convert_knowledge_graph_does_not_mutate(self):
+        kg = self._make_kg()
+        original_entities = list(kg.entities)
+        original_relationships = list(kg.relationships)
+        viz = _make_viz()
+        viz._convert_knowledge_graph(kg)
+        self.assertEqual(kg.entities, original_entities)
+        self.assertEqual(kg.relationships, original_relationships)
+
+    def test_convert_knowledge_graph_is_deterministic(self):
+        kg = self._make_kg()
+        viz = _make_viz()
+        self.assertEqual(viz._convert_knowledge_graph(kg), viz._convert_knowledge_graph(kg))
+
+    def test_normalize_graph_routes_kg_type(self):
+        kg = self._make_kg()
+        viz = _make_viz()
+        viz._convert_knowledge_graph = MagicMock(return_value=GRAPH_DICT)
+        viz._normalize_graph(kg)
+        viz._convert_knowledge_graph.assert_called_once_with(kg)
+
+    def test_normalize_graph_returns_dict_for_kg_type(self):
+        kg = self._make_kg()
+        viz = _make_viz()
+        result = viz._normalize_graph(kg)
+        self.assertIsInstance(result, dict)
+        self.assertIn("entities", result)
+        self.assertIn("relationships", result)
+
+    def _run_visualize_network(self, graph_arg):
+        mock_fig = MagicMock()
+        mock_go = sys.modules["plotly.graph_objects"]
+        mock_go.Figure.return_value = mock_fig
+        mock_go.Scatter.return_value = MagicMock()
+        mock_go.Layout.return_value = MagicMock()
+        viz = _make_viz()
+        fake_pos = {"e1": (0.0, 0.0), "e2": (1.0, 1.0)}
+        viz.force_layout = MagicMock()
+        viz.force_layout.compute_layout.return_value = fake_pos
+        viz.hierarchical_layout = MagicMock()
+        viz.circular_layout = MagicMock()
+        with (
+            patch(
+                "semantica.visualization.kg_visualizer.ColorPalette.get_entity_type_colors",
+                return_value={"Person": "#ff0000"},
+            ),
+            patch(
+                "semantica.visualization.kg_visualizer.ColorPalette.get_colors",
+                return_value=["#ff0000"],
+            ),
+        ):
+            return viz.visualize_network(graph_arg, output="interactive")
+
+    def test_visualize_network_accepts_knowledge_graph(self):
+        self.assertIsNotNone(self._run_visualize_network(self._make_kg()))
+
+    def test_visualize_communities_accepts_knowledge_graph(self):
+        kg = self._make_kg()
+        viz = _make_viz()
+        viz._normalize_graph = MagicMock(return_value=GRAPH_DICT)
+        viz._visualize_network_plotly = MagicMock(return_value=MagicMock())
+        communities = {"node_assignments": {"e1": 0, "e2": 1}, "num_communities": 2}
+        with patch(
+            "semantica.visualization.kg_visualizer.ColorPalette.get_community_colors",
+            return_value=["#ff0000", "#00ff00"],
+        ):
+            viz.visualize_communities(kg, communities=communities)
+        viz._normalize_graph.assert_called_once_with(kg)
+
+    def test_visualize_centrality_accepts_knowledge_graph(self):
+        kg = self._make_kg()
+        viz = _make_viz()
+        viz._normalize_graph = MagicMock(return_value=GRAPH_DICT)
+        viz._visualize_network_plotly = MagicMock(return_value=MagicMock())
+        viz.visualize_centrality(kg, centrality={"centrality": {}})
+        viz._normalize_graph.assert_called_once_with(kg)
+
+    def test_visualize_entity_types_accepts_knowledge_graph(self):
+        kg = self._make_kg()
+        viz = _make_viz()
+        viz._normalize_graph = MagicMock(return_value=GRAPH_DICT)
+        sys.modules["plotly.express"].bar.return_value = MagicMock()
+        viz.visualize_entity_types(kg)
+        viz._normalize_graph.assert_called_once_with(kg)
+
+    def test_visualize_relationship_matrix_accepts_knowledge_graph(self):
+        kg = self._make_kg()
+        viz = _make_viz()
+        viz._normalize_graph = MagicMock(return_value=GRAPH_DICT)
+        sys.modules["plotly.graph_objects"].Figure.return_value = MagicMock()
+        sys.modules["plotly.graph_objects"].Heatmap.return_value = MagicMock()
+        viz.visualize_relationship_matrix(kg)
+        viz._normalize_graph.assert_called_once_with(kg)
+
+    def test_knowledge_graph_importable_from_kg_module(self):
+        if self.KnowledgeGraph is None:
+            self.skipTest("semantica.kg.KnowledgeGraph not available")
+        try:
+            import semantica.kg as _kg_module
+            _ = _kg_module.KnowledgeGraph
+        except (ImportError, AttributeError) as exc:
+            self.fail(f"KnowledgeGraph not exported from semantica.kg: {exc}")
+
+    def test_knowledge_graph_empty_defaults(self):
+        kg = self.KnowledgeGraph()
+        self.assertEqual(kg.entities, [])
+        self.assertEqual(kg.relationships, [])
+        self.assertFalse(bool(kg))
+
+    def test_knowledge_graph_len(self):
+        kg = self._make_kg()
+        self.assertEqual(len(kg), len(ENTITIES))
+
+
 if __name__ == "__main__":
     unittest.main()