Remove kg_qa module and exports; update docs and notebooks to remove KG QA references and add temporary notices; adjust README Quality Assurance examples; add roadmap entry for KG QA in Q1; refine wording per request

This commit is contained in:
KaifAhmad1
2025-12-06 17:34:57 +05:30
parent 01791562f1
commit dd39c544fc
42 changed files with 232 additions and 4934 deletions
+7 -10
View File
@@ -387,23 +387,19 @@ result = ExecutionEngine().execute_pipeline(pipeline, parallel=True)
### Production-Ready Quality Assurance
> **Enterprise-Grade QA** • Conflict Detection • Deduplication • Quality Scoring
> **Enterprise-Grade QA** • Conflict Detection • Deduplication
```python
from semantica.kg_qa import QualityAssessor
from semantica.deduplication import DuplicateDetector
from semantica.conflicts import ConflictDetector
assessor = QualityAssessor()
report = assessor.assess(kg, check_completeness=True, check_consistency=True)
conflicts = ConflictDetector().detect_conflicts(kg)
duplicates = DuplicateDetector().find_duplicates(entities=kg.entities, similarity_threshold=0.85)
detector = DuplicateDetector()
duplicates = detector.find_duplicates(entities=kg.entities, similarity_threshold=0.85)
print(f"Quality Score: {report.overall_score}/100, Duplicates: {len(duplicates)}")
print(f"Conflicts: {len(conflicts)} | Duplicates: {len(duplicates)}")
```
[**Cookbook: Conflict Detection**](https://github.com/Hawksight-AI/semantica/tree/main/cookbook/introduction/Conflict_Detection.ipynb) • [**Deduplication**](https://github.com/Hawksight-AI/semantica/tree/main/cookbook/introduction/Deduplication.ipynb) • [**Graph Quality**](https://github.com/Hawksight-AI/semantica/tree/main/cookbook/introduction/Graph_Quality.ipynb)
[**Cookbook: Conflict Detection**](https://github.com/Hawksight-AI/semantica/tree/main/cookbook/introduction/Conflict_Detection.ipynb) • [**Deduplication**](https://github.com/Hawksight-AI/semantica/tree/main/cookbook/introduction/Deduplication.ipynb)
## 🚀 Quick Start
@@ -466,9 +462,10 @@ print(f"Answer: {result.answer} | Nodes: {kg.node_count}, Edges: {kg.edge_count}
- [x] Core framework (v1.0)
- [x] GraphRAG engine
- [x] 6-stage ontology pipeline
- [x] Quality assurance features
- [] Quality assurance features
- [ ] Enhanced multi-language support
- [ ] Real-time streaming improvements
- [ ] `semantica.kg_qa` (KG Quality Assurance) module
### Q2 2026
- [ ] Multi-modal processing
@@ -1,5 +1,12 @@
{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"**Notice**: The `semantica.kg_qa` module is temporarily unavailable and will be reintroduced in a future release. Any quality assessment examples in this notebook are disabled."
]
},
{
"cell_type": "markdown",
"metadata": {},
@@ -40,7 +47,7 @@
")\n",
"from semantica.kg import GraphBuilder, GraphAnalyzer\n",
"from semantica.embeddings import EmbeddingGenerator\n",
"from semantica.kg_qa import KGQualityAssessor\n",
"import numpy as np\n"
]
},
@@ -1,5 +1,12 @@
{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"**Notice**: The `semantica.kg_qa` module is temporarily unavailable and will be reintroduced in a future release. Any quality assessment examples in this notebook are disabled."
]
},
{
"cell_type": "markdown",
"metadata": {},
@@ -35,7 +42,7 @@
"outputs": [],
"source": [
"from semantica.kg import GraphBuilder\n",
"from semantica.kg_qa import ConsistencyChecker\n",
"from datetime import datetime\n",
"import json\n"
]
+12 -5
View File
@@ -1,5 +1,12 @@
{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"**Notice**: The `semantica.kg_qa` module is temporarily unavailable and will be reintroduced in a future release. The quality assessment examples in this notebook are disabled."
]
},
{
"cell_type": "markdown",
"metadata": {},
@@ -45,11 +52,11 @@
"metadata": {},
"outputs": [],
"source": [
"from semantica.kg_qa import KGQualityAssessor\n",
"from semantica.kg import GraphBuilder\n",
"\n",
"builder = GraphBuilder()\n",
"assessor = KGQualityAssessor()\n",
"\n",
"entities = [\n",
" {\"id\": \"e1\", \"type\": \"Organization\", \"name\": \"Apple Inc.\", \"properties\": {}}\n",
@@ -81,7 +88,7 @@
"metadata": {},
"outputs": [],
"source": [
"from semantica.kg_qa import ConsistencyChecker\n",
"\n",
"consistency_checker = ConsistencyChecker()\n",
"\n",
@@ -107,7 +114,7 @@
"metadata": {},
"outputs": [],
"source": [
"from semantica.kg_qa import CompletenessValidator\n",
"\n",
"completeness_validator = CompletenessValidator()\n",
"\n",
@@ -133,7 +140,7 @@
"metadata": {},
"outputs": [],
"source": [
"from semantica.kg_qa import QualityMetrics\n",
"\n",
"quality_metrics = QualityMetrics()\n",
"\n",
@@ -1402,7 +1402,7 @@
"outputs": [],
"source": [
"from semantica.reasoning import InferenceEngine, RuleManager\n",
"from semantica.kg_qa import KGQualityAssessor\n",
"\n",
"# Advanced Feature 1: Reasoning with Inference Engine\n",
"print(f\"Advanced Feature: Logical Reasoning\")\n",
@@ -1,5 +1,12 @@
{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"**Notice**: The `semantica.kg_qa` module is temporarily unavailable and will be reintroduced in a future release. Any quality assessment examples in this notebook are disabled."
]
},
{
"cell_type": "markdown",
"metadata": {},
@@ -62,7 +69,7 @@
"from semantica.kg import ConnectivityAnalyzer, TemporalGraphQuery, TemporalPatternDetector\n",
"from semantica.ontology import OntologyGenerator, ClassInferrer, PropertyGenerator, OntologyValidator\n",
"from semantica.reasoning import InferenceEngine, RuleManager, ExplanationGenerator\n",
"from semantica.kg_qa import KGQualityAssessor\n",
"from semantica.conflicts import ConflictDetector\n",
"from semantica.export import JSONExporter, RDFExporter, OWLExporter, ReportGenerator\n",
"from semantica.visualization import KGVisualizer, OntologyVisualizer, AnalyticsVisualizer\n",
@@ -1,5 +1,12 @@
{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"**Notice**: The `semantica.kg_qa` module is temporarily unavailable and will be reintroduced in a future release. Any quality assessment examples in this notebook are disabled."
]
},
{
"cell_type": "markdown",
"metadata": {},
@@ -61,7 +68,7 @@
"from semantica.kg import GraphBuilder, TemporalGraphQuery, TemporalPatternDetector, GraphAnalyzer\n",
"from semantica.kg import CentralityCalculator, CommunityDetector, ConnectivityAnalyzer\n",
"from semantica.reasoning import InferenceEngine, RuleManager, ExplanationGenerator\n",
"from semantica.kg_qa import KGQualityAssessor\n",
"from semantica.conflicts import ConflictDetector\n",
"from semantica.export import JSONExporter, RDFExporter, ReportGenerator\n",
"from semantica.visualization import KGVisualizer, TemporalVisualizer, AnalyticsVisualizer\n",
@@ -1,5 +1,12 @@
{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"**Notice**: The `semantica.kg_qa` module is temporarily unavailable and will be reintroduced in a future release. Any quality assessment examples in this notebook are disabled."
]
},
{
"cell_type": "markdown",
"metadata": {},
@@ -60,7 +67,7 @@
"from semantica.kg import GraphBuilder, TemporalPatternDetector, TemporalGraphQuery, GraphAnalyzer\n",
"from semantica.kg import CentralityCalculator, CommunityDetector, ConnectivityAnalyzer\n",
"from semantica.reasoning import InferenceEngine, RuleManager, ExplanationGenerator\n",
"from semantica.kg_qa import KGQualityAssessor, AutomatedFixer\n",
"from semantica.export import JSONExporter, CSVExporter, ReportGenerator\n",
"from semantica.visualization import KGVisualizer, TemporalVisualizer, AnalyticsVisualizer\n",
"import tempfile\n",
@@ -483,4 +490,4 @@
},
"nbformat": 4,
"nbformat_minor": 2
}
}
@@ -1,5 +1,12 @@
{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"**Notice**: The `semantica.kg_qa` module is temporarily unavailable and will be reintroduced in a future release. Any quality assessment examples in this notebook are disabled."
]
},
{
"cell_type": "markdown",
"metadata": {},
@@ -58,7 +65,7 @@
"from semantica.semantic_extract import NERExtractor, RelationExtractor, EventDetector, TripleExtractor\n",
"from semantica.kg import GraphBuilder, GraphAnalyzer, ConnectivityAnalyzer, CentralityCalculator\n",
"from semantica.reasoning import InferenceEngine, RuleManager, ExplanationGenerator\n",
"from semantica.kg_qa import KGQualityAssessor\n",
"from semantica.conflicts import ConflictDetector\n",
"from semantica.kg import ProvenanceTracker\n",
"from semantica.export import JSONExporter, RDFExporter, ReportGenerator\n",
@@ -463,4 +470,4 @@
},
"nbformat": 4,
"nbformat_minor": 2
}
}
@@ -1,5 +1,12 @@
{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"**Notice**: The `semantica.kg_qa` module is temporarily unavailable and will be reintroduced in a future release. Any quality assessment examples in this notebook are disabled."
]
},
{
"cell_type": "markdown",
"metadata": {},
@@ -58,7 +65,7 @@
"from semantica.semantic_extract import NERExtractor, RelationExtractor, EventDetector\n",
"from semantica.kg import GraphBuilder, TemporalGraphQuery, TemporalPatternDetector, GraphAnalyzer, ConnectivityAnalyzer\n",
"from semantica.reasoning import InferenceEngine, ExplanationGenerator\n",
"from semantica.kg_qa import KGQualityAssessor\n",
"from semantica.kg import ProvenanceTracker\n",
"from semantica.conflicts import ConflictDetector\n",
"from semantica.export import RDFExporter, ReportGenerator\n",
@@ -1,5 +1,12 @@
{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"**Notice**: The `semantica.kg_qa` module is temporarily unavailable and will be reintroduced in a future release. Any quality assessment examples in this notebook are disabled."
]
},
{
"cell_type": "markdown",
"metadata": {},
@@ -60,7 +67,7 @@
"from semantica.kg import GraphBuilder, TemporalGraphQuery, TemporalPatternDetector, GraphAnalyzer\n",
"from semantica.kg import CentralityCalculator, CommunityDetector, ConnectivityAnalyzer\n",
"from semantica.reasoning import InferenceEngine, RuleManager, ExplanationGenerator\n",
"from semantica.kg_qa import KGQualityAssessor\n",
"from semantica.conflicts import ConflictDetector\n",
"from semantica.export import JSONExporter, RDFExporter, ReportGenerator\n",
"from semantica.visualization import KGVisualizer, TemporalVisualizer, AnalyticsVisualizer\n",
@@ -1,5 +1,12 @@
{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"**Notice**: The `semantica.kg_qa` module is temporarily unavailable and will be reintroduced in a future release. Any quality assessment examples in this notebook are disabled."
]
},
{
"cell_type": "markdown",
"metadata": {},
@@ -60,7 +67,7 @@
"from semantica.kg import GraphBuilder, GraphAnalyzer, CentralityCalculator, CommunityDetector\n",
"from semantica.kg import ConnectivityAnalyzer, TemporalGraphQuery, TemporalPatternDetector\n",
"from semantica.reasoning import InferenceEngine, RuleManager, ExplanationGenerator\n",
"from semantica.kg_qa import KGQualityAssessor\n",
"from semantica.conflicts import ConflictDetector\n",
"from semantica.export import JSONExporter, CSVExporter, RDFExporter, ReportGenerator\n",
"from semantica.visualization import KGVisualizer, AnalyticsVisualizer, TemporalVisualizer\n",
@@ -425,4 +432,4 @@
},
"nbformat": 4,
"nbformat_minor": 2
}
}
@@ -1,5 +1,12 @@
{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"**Notice**: The `semantica.kg_qa` module is temporarily unavailable and will be reintroduced in a future release. Any quality assessment examples in this notebook are disabled."
]
},
{
"cell_type": "markdown",
"metadata": {},
@@ -59,7 +66,7 @@
"from semantica.semantic_extract import NERExtractor, RelationExtractor, EventDetector\n",
"from semantica.kg import GraphBuilder, TemporalPatternDetector, GraphAnalyzer\n",
"from semantica.reasoning import InferenceEngine, RuleManager, ExplanationGenerator\n",
"from semantica.kg_qa import KGQualityAssessor, AutomatedFixer\n",
"from semantica.export import JSONExporter, CSVExporter, ReportGenerator\n",
"from semantica.visualization import KGVisualizer, TemporalVisualizer, AnalyticsVisualizer\n",
"import tempfile\n",
@@ -1,5 +1,12 @@
{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"**Notice**: The `semantica.kg_qa` module is temporarily unavailable and will be reintroduced in a future release. Any quality assessment examples in this notebook are disabled."
]
},
{
"cell_type": "markdown",
"metadata": {},
@@ -60,7 +67,7 @@
"from semantica.kg import GraphBuilder, GraphAnalyzer, ConnectivityAnalyzer\n",
"from semantica.ontology import OntologyGenerator, ClassInferrer, PropertyGenerator, OntologyValidator\n",
"from semantica.reasoning import InferenceEngine, RuleManager, ExplanationGenerator\n",
"from semantica.kg_qa import KGQualityAssessor, ValidationEngine\n",
"from semantica.conflicts import ConflictDetector\n",
"from semantica.export import JSONExporter, CSVExporter, RDFExporter, OWLExporter, ReportGenerator\n",
"from semantica.visualization import KGVisualizer, OntologyVisualizer, AnalyticsVisualizer\n",
@@ -1,5 +1,12 @@
{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"**Notice**: The `semantica.kg_qa` module is temporarily unavailable and will be reintroduced in a future release. Any quality assessment examples in this notebook are disabled."
]
},
{
"cell_type": "markdown",
"metadata": {},
@@ -60,7 +67,7 @@
"from semantica.kg import GraphBuilder, GraphValidator, EntityResolver, GraphAnalyzer\n",
"from semantica.triple_store import TripleStore, TripleManager, QueryEngine\n",
"from semantica.reasoning import InferenceEngine, RuleManager, ExplanationGenerator\n",
"from semantica.kg_qa import KGQualityAssessor, ValidationEngine\n",
"from semantica.export import JSONExporter, RDFExporter, OWLExporter, ReportGenerator\n",
"from semantica.visualization import KGVisualizer, OntologyVisualizer, TemporalVisualizer\n",
"import tempfile\n",
@@ -1,5 +1,12 @@
{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"**Notice**: The `semantica.kg_qa` module is temporarily unavailable and will be reintroduced in a future release. Any quality assessment examples in this notebook are disabled."
]
},
{
"cell_type": "markdown",
"metadata": {},
@@ -62,7 +69,7 @@
"from semantica.kg import ConnectivityAnalyzer, TemporalGraphQuery, TemporalPatternDetector\n",
"from semantica.ontology import OntologyGenerator, ClassInferrer, PropertyGenerator, OntologyValidator\n",
"from semantica.reasoning import InferenceEngine, RuleManager, ExplanationGenerator\n",
"from semantica.kg_qa import KGQualityAssessor\n",
"from semantica.conflicts import ConflictDetector\n",
"from semantica.export import JSONExporter, RDFExporter, OWLExporter, ReportGenerator\n",
"from semantica.visualization import KGVisualizer, OntologyVisualizer, AnalyticsVisualizer\n",
@@ -414,4 +421,4 @@
},
"nbformat": 4,
"nbformat_minor": 2
}
}
@@ -1,5 +1,12 @@
{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"**Notice**: The `semantica.kg_qa` module is temporarily unavailable and will be reintroduced in a future release. Any quality assessment examples in this notebook are disabled."
]
},
{
"cell_type": "markdown",
"metadata": {},
@@ -62,7 +69,7 @@
"from semantica.kg import ConnectivityAnalyzer, TemporalGraphQuery\n",
"from semantica.ontology import OntologyGenerator, ClassInferrer, PropertyGenerator, OntologyValidator\n",
"from semantica.reasoning import InferenceEngine, RuleManager, ExplanationGenerator\n",
"from semantica.kg_qa import KGQualityAssessor\n",
"from semantica.conflicts import ConflictDetector\n",
"from semantica.export import JSONExporter, RDFExporter, OWLExporter, ReportGenerator\n",
"from semantica.visualization import KGVisualizer, OntologyVisualizer, AnalyticsVisualizer\n",
@@ -424,4 +431,4 @@
},
"nbformat": 4,
"nbformat_minor": 2
}
}
@@ -1,5 +1,12 @@
{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"**Notice**: The `semantica.kg_qa` module is temporarily unavailable and will be reintroduced in a future release. Any quality assessment examples in this notebook are disabled."
]
},
{
"cell_type": "markdown",
"metadata": {},
@@ -62,7 +69,7 @@
"from semantica.kg import GraphBuilder, GraphValidator, EntityResolver, GraphAnalyzer\n",
"from semantica.triple_store import TripleStore, TripleManager, QueryEngine\n",
"from semantica.reasoning import InferenceEngine, RuleManager, ExplanationGenerator\n",
"from semantica.kg_qa import KGQualityAssessor, ValidationEngine\n",
"from semantica.export import JSONExporter, RDFExporter, OWLExporter, ReportGenerator\n",
"from semantica.visualization import KGVisualizer, OntologyVisualizer, TemporalVisualizer\n",
"import json\n",
@@ -1,5 +1,12 @@
{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"**Notice**: The `semantica.kg_qa` module is temporarily unavailable and will be reintroduced in a future release. Any quality assessment examples in this notebook are disabled."
]
},
{
"cell_type": "markdown",
"metadata": {},
@@ -60,7 +67,7 @@
"from semantica.kg import GraphBuilder, TemporalGraphQuery, TemporalPatternDetector, GraphAnalyzer\n",
"from semantica.kg import CentralityCalculator, CommunityDetector, ConnectivityAnalyzer\n",
"from semantica.reasoning import InferenceEngine, RuleManager, ExplanationGenerator\n",
"from semantica.kg_qa import KGQualityAssessor\n",
"from semantica.export import JSONExporter, CSVExporter, RDFExporter, ReportGenerator\n",
"from semantica.visualization import KGVisualizer, TemporalVisualizer, AnalyticsVisualizer\n",
"import tempfile\n",
@@ -439,4 +446,4 @@
},
"nbformat": 4,
"nbformat_minor": 2
}
}
@@ -1,5 +1,12 @@
{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"**Notice**: The `semantica.kg_qa` module is temporarily unavailable and will be reintroduced in a future release. Any quality assessment examples in this notebook are disabled."
]
},
{
"cell_type": "markdown",
"metadata": {},
@@ -62,7 +69,7 @@
"from semantica.kg import ConnectivityAnalyzer, TemporalGraphQuery, TemporalPatternDetector\n",
"from semantica.ontology import OntologyGenerator, ClassInferrer, PropertyGenerator, OntologyValidator\n",
"from semantica.reasoning import InferenceEngine, RuleManager, ExplanationGenerator\n",
"from semantica.kg_qa import KGQualityAssessor\n",
"from semantica.conflicts import ConflictDetector\n",
"from semantica.export import JSONExporter, CSVExporter, RDFExporter, OWLExporter, ReportGenerator\n",
"from semantica.visualization import KGVisualizer, OntologyVisualizer, AnalyticsVisualizer\n",
@@ -1,5 +1,12 @@
{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"**Notice**: The `semantica.kg_qa` module is temporarily unavailable and will be reintroduced in a future release. Any quality assessment examples in this notebook are disabled."
]
},
{
"cell_type": "markdown",
"metadata": {},
@@ -60,7 +67,7 @@
"from semantica.kg import GraphBuilder, TemporalGraphQuery, TemporalPatternDetector, GraphAnalyzer\n",
"from semantica.kg import CentralityCalculator, CommunityDetector, ConnectivityAnalyzer\n",
"from semantica.reasoning import InferenceEngine, RuleManager, ExplanationGenerator\n",
"from semantica.kg_qa import KGQualityAssessor, AutomatedFixer\n",
"from semantica.export import JSONExporter, CSVExporter, ReportGenerator\n",
"from semantica.visualization import KGVisualizer, TemporalVisualizer, AnalyticsVisualizer\n",
"import tempfile\n",
@@ -467,4 +474,4 @@
},
"nbformat": 4,
"nbformat_minor": 2
}
}
@@ -1,5 +1,12 @@
{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"**Notice**: The `semantica.kg_qa` module is temporarily unavailable and will be reintroduced in a future release. Any quality assessment examples in this notebook are disabled."
]
},
{
"cell_type": "markdown",
"metadata": {},
@@ -60,7 +67,7 @@
"from semantica.kg import GraphBuilder, GraphAnalyzer, CentralityCalculator, CommunityDetector\n",
"from semantica.kg import ConnectivityAnalyzer, TemporalGraphQuery, TemporalPatternDetector\n",
"from semantica.reasoning import InferenceEngine, RuleManager, ExplanationGenerator\n",
"from semantica.kg_qa import KGQualityAssessor\n",
"from semantica.conflicts import ConflictDetector\n",
"from semantica.export import JSONExporter, CSVExporter, RDFExporter, ReportGenerator\n",
"from semantica.visualization import KGVisualizer, AnalyticsVisualizer, TemporalVisualizer\n",
@@ -1,5 +1,12 @@
{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"**Notice**: The `semantica.kg_qa` module is temporarily unavailable and will be reintroduced in a future release. Any quality assessment examples in this notebook are disabled."
]
},
{
"cell_type": "markdown",
"metadata": {},
@@ -60,7 +67,7 @@
"from semantica.kg import GraphBuilder, TemporalGraphQuery, TemporalPatternDetector, GraphAnalyzer\n",
"from semantica.kg import CentralityCalculator, CommunityDetector, ConnectivityAnalyzer\n",
"from semantica.reasoning import InferenceEngine, RuleManager, ExplanationGenerator\n",
"from semantica.kg_qa import KGQualityAssessor\n",
"from semantica.export import JSONExporter, CSVExporter, RDFExporter, ReportGenerator\n",
"from semantica.visualization import KGVisualizer, TemporalVisualizer, AnalyticsVisualizer\n",
"import tempfile\n",
@@ -364,4 +371,4 @@
},
"nbformat": 4,
"nbformat_minor": 2
}
}
@@ -1,5 +1,12 @@
{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"**Notice**: The `semantica.kg_qa` module is temporarily unavailable and will be reintroduced in a future release. Any quality assessment examples in this notebook are disabled."
]
},
{
"cell_type": "markdown",
"metadata": {},
@@ -62,7 +69,7 @@
"from semantica.kg import ConnectivityAnalyzer, TemporalGraphQuery, TemporalPatternDetector\n",
"from semantica.embeddings import EmbeddingGenerator, TextEmbedder\n",
"from semantica.reasoning import InferenceEngine, RuleManager, ExplanationGenerator\n",
"from semantica.kg_qa import KGQualityAssessor\n",
"from semantica.export import JSONExporter, CSVExporter, RDFExporter, ReportGenerator\n",
"from semantica.visualization import KGVisualizer, TemporalVisualizer, AnalyticsVisualizer\n",
"import tempfile\n",
@@ -1,5 +1,12 @@
{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"**Notice**: The `semantica.kg_qa` module is temporarily unavailable and will be reintroduced in a future release. Any quality assessment examples in this notebook are disabled."
]
},
{
"cell_type": "markdown",
"metadata": {},
@@ -60,7 +67,7 @@
"from semantica.kg import GraphBuilder, TemporalGraphQuery, TemporalPatternDetector, GraphAnalyzer\n",
"from semantica.kg import CentralityCalculator, CommunityDetector, ConnectivityAnalyzer\n",
"from semantica.reasoning import InferenceEngine, RuleManager, ExplanationGenerator\n",
"from semantica.kg_qa import KGQualityAssessor, AutomatedFixer\n",
"from semantica.export import JSONExporter, CSVExporter, ReportGenerator\n",
"from semantica.visualization import KGVisualizer, TemporalVisualizer, AnalyticsVisualizer\n",
"import tempfile\n",
@@ -406,4 +413,4 @@
},
"nbformat": 4,
"nbformat_minor": 2
}
}
@@ -1,5 +1,12 @@
{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"**Notice**: The `semantica.kg_qa` module is temporarily unavailable and will be reintroduced in a future release. Any quality assessment examples in this notebook are disabled."
]
},
{
"cell_type": "markdown",
"metadata": {},
@@ -1,5 +1,12 @@
{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"**Notice**: The `semantica.kg_qa` module is temporarily unavailable and will be reintroduced in a future release. Any quality assessment examples in this notebook are disabled."
]
},
{
"cell_type": "markdown",
"metadata": {},
@@ -60,7 +67,7 @@
"from semantica.kg import GraphBuilder, TemporalGraphQuery, TemporalPatternDetector, GraphAnalyzer\n",
"from semantica.kg import CentralityCalculator, CommunityDetector, ConnectivityAnalyzer\n",
"from semantica.reasoning import InferenceEngine, RuleManager, ExplanationGenerator\n",
"from semantica.kg_qa import KGQualityAssessor\n",
"from semantica.export import JSONExporter, CSVExporter, RDFExporter, ReportGenerator\n",
"from semantica.visualization import KGVisualizer, TemporalVisualizer, AnalyticsVisualizer\n",
"import tempfile\n",
@@ -418,4 +425,4 @@
},
"nbformat": 4,
"nbformat_minor": 2
}
}
+8 -53
View File
@@ -820,42 +820,7 @@ result = execution_engine.execute_pipeline(custom_pipeline)
### 6. Quality Assurance Examples
#### Knowledge Graph Quality Assessment
```python
from semantica.kg_qa import KGQualityAssessor, ValidationEngine
# Initialize quality assessor
quality_assessor = KGQualityAssessor(
config={
"consistency": {"enable": True},
"completeness": {"enable": True},
"validation": {"strict": True}
}
)
# Assess knowledge graph quality
quality_report = quality_assessor.assess_quality(graph)
print(f"Overall Quality Score: {quality_report.overall_score:.2f}")
print(f"Consistency Score: {quality_report.consistency_score:.2f}")
print(f"Completeness Score: {quality_report.completeness_score:.2f}")
# Get issues
for issue in quality_report.issues:
print(f"Issue: {issue.type}")
print(f"Severity: {issue.severity}")
print(f"Description: {issue.description}")
print()
# Validate graph
validation_engine = ValidationEngine()
validation_result = validation_engine.validate(graph)
if validation_result.valid:
print("Graph is valid!")
else:
print(f"Validation errors: {validation_result.errors}")
```
Note: The `semantica.kg_qa` module is temporarily unavailable and will be reintroduced in a future release.
### 7. Export Examples
@@ -902,7 +867,6 @@ from semantica.ingest import FileIngestor
from semantica.semantic_extract import NERExtractor, RelationExtractor
from semantica.embeddings import EmbeddingGenerator
from semantica.kg import GraphBuilder
from semantica.kg_qa import KGQualityAssessor
from semantica.export import JSONExporter
# No explicit initialization needed - framework auto-initializes on first use
@@ -948,10 +912,8 @@ graph = graph_builder.build({
"relationships": all_relationships
})
# Step 5: Assess quality
quality_assessor = KGQualityAssessor()
quality_report = quality_assessor.assess_quality(graph)
print(f"Knowledge Graph Quality: {quality_report.overall_score:.2f}")
# Step 5: (Optional) Quality assessment is temporarily unavailable
# The `semantica.kg_qa` module will be reintroduced in a future release.
# Step 6: Export results
json_exporter = JSONExporter()
@@ -1142,10 +1104,7 @@ from semantica.visualization import QualityVisualizer
quality_viz = QualityVisualizer()
# Quality dashboard
from semantica.kg_qa import KGQualityAssessor
quality_assessor = KGQualityAssessor()
quality_report = quality_assessor.generate_quality_report(graph)
quality_report = {"overall_score": 0.85, "issues": [], "consistency": {}, "completeness": {}}
quality_viz.visualize_dashboard(quality_report, output="html", file_path="quality_dashboard.html")
# Quality score distribution
@@ -1156,17 +1115,13 @@ quality_viz.visualize_score_distribution(quality_scores,
# Quality issues
quality_viz.visualize_issues(quality_report, output="html", file_path="quality_issues.html")
# Completeness metrics
from semantica.kg_qa import CompletenessMetrics
completeness_metrics = CompletenessMetrics()
completeness_data = completeness_metrics.calculate_entity_completeness(entities, schema)
# Completeness metrics (provide your precomputed data)
completeness_data = {"score": 0.82, "by_type": {"Person": 0.9, "Company": 0.75}}
quality_viz.visualize_completeness_metrics(completeness_data,
output="html", file_path="completeness.html")
# Consistency heatmap
from semantica.kg_qa import ConsistencyMetrics
consistency_metrics = ConsistencyMetrics()
consistency_data = consistency_metrics.calculate_logical_consistency(graph)
# Consistency heatmap (provide your precomputed data)
consistency_data = {"score": 0.88, "violations": []}
quality_viz.visualize_consistency_heatmap(consistency_data,
output="html", file_path="consistency_heatmap.html")
```
+1 -9
View File
@@ -73,7 +73,6 @@ graph TB
### Quality Assurance
- **`semantica.deduplication`** - Entity deduplication
- **`semantica.conflicts`** - Conflict detection and resolution
- **`semantica.kg_qa`** - Quality assessment
---
@@ -118,14 +117,7 @@ class CustomExtractor(BaseExtractor):
### Custom Validators
```python
from semantica.kg_qa import BaseValidator
class CustomValidator(BaseValidator):
def validate(self, graph):
# Custom validation logic
pass
```
Validators can be implemented within domain-specific modules (e.g., graph or ontology) as needed.
---
+2 -23
View File
@@ -16,7 +16,7 @@ Semantica's modules are organized into six logical layers:
| **Input Layer** | [Ingest](#ingest-module), [Parse](#parse-module), [Split](#split-module), [Normalize](#normalize-module) | Data ingestion, parsing, chunking, and cleaning |
| **Core Processing** | [Semantic Extract](#semantic-extract-module), [Knowledge Graph](#knowledge-graph-kg-module), [Ontology](#ontology-module), [Reasoning](#reasoning-module) | Entity extraction, graph construction, inference |
| **Storage** | [Embeddings](#embeddings-module), [Vector Store](#vector-store-module), [Graph Store](#graph-store-module), [Triple Store](#triple-store-module) | Vector and graph persistence |
| **Quality Assurance** | [Deduplication](#deduplication-module), [Conflicts](#conflicts-module), [KG QA](#kg-quality-assurance-module) | Data quality and consistency |
| **Quality Assurance** | [Deduplication](#deduplication-module), [Conflicts](#conflicts-module) | Data quality and consistency |
| **Context & Memory** | [Context](#context-module), [Seed](#seed-module) | Agent memory and foundation data |
| **Output & Orchestration** | [Export](#export-module), [Visualization](#visualization-module), [Pipeline](#pipeline-module) | Export, visualization, and workflow management |
@@ -765,27 +765,7 @@ conflicts = detector.detect_value_conflicts(entities, "name")
- `AutoMerger` — Automatic merging of duplicates
- `AutoResolver` — Automatic conflict resolution
**Quality Metrics:**
| Metric | Calculation |
| :--- | :--- |
| **Overall Score** | `(0.6 × completeness) + (0.4 × consistency)` |
| **Entity Quality** | Required field presence (ID, type) |
| **Relationship Quality** | Required field presence (source, target, type) |
**Quick Example:**
```python
from semantica.kg_qa import assess_quality, generate_quality_report, KGQualityAssessor
# Using convenience functions
score = assess_quality(knowledge_graph)
report = generate_quality_report(knowledge_graph, schema)
# Using classes directly
assessor = KGQualityAssessor()
score = assessor.assess_overall_quality(knowledge_graph)
```
Note: The KG quality assessment module has been temporarily removed and will be reintroduced in a future release.
---
@@ -1138,7 +1118,6 @@ new_facts = inference_engine.forward_chain(kg, rule_manager)
| **Triple Store** | `semantica.triple_store` | `TripleManager` | RDF storage |
| **Deduplication** | `semantica.deduplication` | `DuplicateDetector` | Duplicate removal |
| **Conflicts** | `semantica.conflicts` | `ConflictDetector` | Conflict resolution |
| **KG QA** | `semantica.kg_qa` | `KGQualityAssessor` | Quality assurance |
| **Context** | `semantica.context` | `AgentMemory` | Agent context |
| **Seed** | `semantica.seed` | `SeedDataManager` | Foundation data |
| **Export** | `semantica.export` | `JSONExporter` | Data export |
-216
View File
@@ -1,216 +0,0 @@
# KG QA
> **Knowledge Graph Quality Assurance system for validation, metrics, and automated repair.**
---
## 🎯 Overview
<div class="grid cards" markdown>
- :material-check-decagram:{ .lg .middle } **Quality Metrics**
---
Calculate Completeness, Consistency, and Accuracy scores
- :material-shield-check:{ .lg .middle } **Validation Engine**
---
Validate against schema constraints and custom rules
- :material-wrench:{ .lg .middle } **Automated Fixes**
---
Auto-repair duplicates, missing fields, and inconsistencies
- :material-file-document-edit:{ .lg .middle } **Reporting**
---
Generate detailed quality reports (JSON, HTML, YAML)
- :material-relation-many-to-many:{ .lg .middle } **Consistency**
---
Check logical, temporal, and hierarchical consistency
- :material-lightbulb:{ .lg .middle } **Suggestions**
---
Get actionable improvement suggestions
</div>
!!! tip "When to Use"
- **Pre-Deployment**: Validate graph quality before production use
- **Monitoring**: Continuous quality monitoring of live graphs
- **Debugging**: Identify and fix issues in problematic graphs
---
## ⚙️ Algorithms Used
### Quality Metrics
- **Weighted Averaging**: `Score = w1*Completeness + w2*Consistency`
- **Normalization**: Min-max scaling of scores to `0.0 - 1.0`
- **Completeness Ratio**: `PresentProperties / RequiredProperties`
### Consistency Checking
- **Logical Consistency**: Contradiction detection (e.g., A > B and B > A)
- **Temporal Consistency**: Time range validation (Start < End)
- **Hierarchical Consistency**: Cycle detection in taxonomy (DFS)
- **Domain/Range**: Type compatibility checking for relationships
### Automated Fixes
- **Duplicate Merging**: Using Deduplication module strategies
- **Conflict Resolution**: Using Conflicts module strategies
- **Default Injection**: Filling missing required fields with defaults
- **Inference**: Inferring missing types or links based on topology
---
## Main Classes
### KGQualityAssessor
Coordinator for overall quality assessment.
**Methods:**
| Method | Description |
|--------|-------------|
| `assess_quality(kg)` | Calculate all metrics |
| `generate_report(kg)` | Create full report |
**Example:**
```python
from semantica.kg_qa import KGQualityAssessor
assessor = KGQualityAssessor()
score = assessor.assess_overall_quality(kg)
print(f"Graph Quality Score: {score}")
```
### ConsistencyChecker
Validates graph consistency.
**Methods:**
| Method | Description | Algorithm |
|--------|-------------|-----------|
| `check_logical(kg)` | Logical rules | Rule Engine |
| `check_temporal(kg)` | Time validity | Range Check |
| `check_hierarchical(kg)` | Cycles/Tree | DFS |
### CompletenessValidator
Checks for missing data.
**Methods:**
| Method | Description |
|--------|-------------|
| `validate_entities(kg)` | Check entity fields |
| `validate_schema(kg)` | Check schema compliance |
### AutomatedFixer
Applies automatic repairs.
**Methods:**
| Method | Description |
|--------|-------------|
| `fix_issues(kg, issues)` | Fix reported issues |
| `merge_duplicates(kg)` | Fix duplicates |
| `resolve_conflicts(kg)` | Fix conflicts |
---
## Convenience Functions
```python
from semantica.kg_qa import assess_quality, generate_quality_report, fix_issues
# 1. Assess
score = assess_quality(kg)
# 2. Report
report = generate_quality_report(kg, schema=my_schema)
# 3. Fix
fixed_kg = fix_issues(kg, report.issues)
```
---
## Configuration
### Environment Variables
```bash
export KG_QA_MIN_SCORE=0.7
export KG_QA_STRICT_MODE=true
```
### YAML Configuration
```yaml
kg_qa:
thresholds:
overall: 0.7
completeness: 0.8
consistency: 0.9
weights:
completeness: 0.6
consistency: 0.4
auto_fix:
enabled: true
strategies:
duplicates: merge
missing_fields: default
```
---
## Integration Examples
### CI/CD Pipeline
```python
from semantica.kg_qa import assess_quality
def validate_graph_deployment(kg):
score = assess_quality(kg)
if score < 0.8:
raise ValueError(f"Quality score {score} too low for deployment!")
print("Graph passed quality checks.")
```
---
## Best Practices
1. **Define Schema**: QA is most effective when validated against a strict schema (Ontology).
2. **Run Regularly**: Graph quality degrades over time; run QA jobs periodically.
3. **Review Fixes**: Automated fixes are powerful but verify them for critical data.
4. **Handle Warnings**: Don't ignore warnings; they often indicate creeping data quality issues.
---
## See Also
- [Ontology Module](ontology.md) - Defining schemas for validation
- [Deduplication Module](deduplication.md) - Used for fixing duplicates
- [Conflicts Module](conflicts.md) - Used for resolving inconsistencies
-42
View File
@@ -22,24 +22,6 @@ from typing import Any, Dict, List, Optional, Union
# Core imports
from .core import Config, ConfigManager, LifecycleManager, PluginRegistry, Semantica
# KG Quality Assurance
from .kg_qa import (
AutomatedFixer,
AutoMerger,
AutoResolver,
CompletenessMetrics,
CompletenessValidator,
ConsistencyChecker,
ConsistencyMetrics,
ConstraintValidator,
ImprovementSuggestions,
IssueTracker,
KGQualityAssessor,
QualityMetrics,
QualityReporter,
RuleValidator,
ValidationEngine,
)
# Pipeline imports
from .pipeline import (
@@ -142,13 +124,6 @@ class _SemanticaModules:
self._visualization = _ModuleProxy("visualization")
return self._visualization
@property
def kg_qa(self):
"""Access KG quality assurance module."""
if self._kg_qa is None:
self._kg_qa = _ModuleProxy("kg_qa")
return self._kg_qa
@property
def pipeline(self):
"""Access pipeline module."""
@@ -289,22 +264,6 @@ __all__ = [
"ParallelismManager",
"ResourceScheduler",
"PipelineValidator",
# KG Quality Assurance
"KGQualityAssessor",
"ConsistencyChecker",
"CompletenessValidator",
"QualityMetrics",
"CompletenessMetrics",
"ConsistencyMetrics",
"ValidationEngine",
"RuleValidator",
"ConstraintValidator",
"QualityReporter",
"IssueTracker",
"ImprovementSuggestions",
"AutomatedFixer",
"AutoMerger",
"AutoResolver",
# Visualization
"KGVisualizer",
"OntologyVisualizer",
@@ -325,7 +284,6 @@ def __getattr__(name: str):
"embeddings",
"semantic_extract",
"visualization",
"kg_qa",
"pipeline",
"parse",
"normalize",
-180
View File
@@ -1,180 +0,0 @@
"""
Knowledge Graph Quality Assurance Module
This module provides comprehensive quality assurance capabilities for the
Semantica framework, enabling production-ready knowledge graph quality
assessment, validation, and automated fixes.
Algorithms Used:
Quality Metrics Calculation:
- Weighted Averaging: Overall quality score aggregation using weighted average formula: overall = (0.6 * completeness) + (0.4 * consistency)
- Entity Quality Scoring: Required field presence checking (ID/URI, type), binary scoring (0.5 per field), average calculation across entities: sum(scores) / len(scores)
- Relationship Quality Scoring: Required field presence checking (source/subject, target/object, type/predicate), weighted scoring (0.33 per field), average calculation across relationships
- Score Normalization: Min-max normalization with clamping to 0.0-1.0 range: min(1.0, max(0.0, score))
- Consistency Score Calculation: Logical inconsistency detection (placeholder for reasoner-based checking)
Completeness Metrics:
- Entity Completeness Calculation: Schema-based required property validation, ratio calculation present_props / required_props, average completeness across entities
- Relationship Completeness Calculation: Required field validation (source, target, type), completeness ratio (has_source + has_target + has_type) / 3.0, average across relationships
- Property Completeness Calculation: Schema-based property validation per entity type, completeness ratio calculation, average across entity types
- Schema Constraint Matching: Entity type to constraint mapping, required property extraction from schema constraints
Consistency Metrics:
- Logical Consistency Checking: Contradiction detection, conflicting relationship identification, inconsistent property value detection (placeholder for reasoner integration)
- Temporal Consistency Checking: Temporal contradiction detection, invalid time range validation, conflicting temporal relationship identification
- Hierarchical Consistency Checking: Circular inheritance detection (DFS-based cycle detection), invalid parent-child relationship validation, hierarchical structure validation
Validation Engine:
- Rule-Based Validation: Custom rule function execution, rule result parsing (error/warning extraction from dict), exception handling and error collection
- Constraint-Based Validation: Entity constraint validation (required properties), relationship constraint validation (domain and range), constraint matching algorithms
- Domain and Range Validation: Relationship type to domain/range mapping, entity type compatibility checking
- Validation Result Aggregation: Error and warning collection, validity determination (valid = len(errors) == 0)
Quality Reporting:
- Issue Identification: Threshold-based issue detection (overall < 0.7, completeness < 0.8), issue type classification (quality, completeness, consistency), severity assignment (low, medium, high)
- Recommendation Generation: Issue-based recommendation generation, score-based recommendation generation, actionable suggestion creation
- Report Serialization: JSON serialization (ISO timestamp formatting, nested structure), YAML serialization (with PyYAML fallback), HTML report generation (planned)
- Issue Tracking: Dictionary-based issue storage (ID as key), severity-based filtering, issue resolution tracking
Automated Fixes:
- Duplicate Detection: Entity duplicate identification (using deduplication module), relationship duplicate identification (same source, target, type matching)
- Duplicate Merging: Property aggregation strategies, relationship reference updating, entity consolidation
- Conflict Resolution: Conflicting property value detection, resolution strategy selection (highest confidence, most recent, source-based), conflict merging
- Missing Property Completion: Schema-based required property identification, default value assignment, value inference from context (planned)
- Inconsistency Resolution: Logical inconsistency detection, resolution strategy application, graph update
Quality Assessment Coordination:
- Metric Aggregation: Multi-metric collection (overall, completeness, consistency), score combination, report generation coordination
- Component Integration: Quality metrics integration, validation engine integration, reporting integration, automated fixing integration
Key Features:
- Quality metrics calculation (overall, completeness, consistency)
- Consistency checking (logical, temporal, hierarchical)
- Completeness validation (entity, relationship, property)
- Automated fixes (duplicates, inconsistencies, missing properties)
- Quality reporting with issue tracking
- Validation engine with rules and constraints
- Method registry for extensibility
- Configuration management with environment variables and config files
Main Classes:
- KGQualityAssessor: Overall quality assessment coordinator
- ConsistencyChecker: Consistency validation engine
- CompletenessValidator: Completeness validation engine
- QualityMetrics: Quality metrics calculator
- CompletenessMetrics: Completeness metrics calculator
- ConsistencyMetrics: Consistency metrics calculator
- ValidationEngine: Rule and constraint validation
- RuleValidator: Rule-based validation
- ConstraintValidator: Constraint-based validation
- QualityReporter: Quality report generation
- IssueTracker: Issue tracking and management
- ImprovementSuggestions: Improvement suggestions generator
- AutomatedFixer: Automated issue fixing
- AutoMerger: Automatic merging of duplicates and conflicts
- AutoResolver: Automatic conflict and inconsistency resolution
- MethodRegistry: Registry for custom QA methods
- KGQAConfig: Configuration manager for KG QA module
Convenience Functions:
- assess_quality: Quality assessment wrapper
- generate_quality_report: Quality report generation wrapper
- identify_quality_issues: Quality issue identification wrapper
- check_consistency: Consistency checking wrapper
- validate_completeness: Completeness validation wrapper
- calculate_quality_metrics: Quality metrics calculation wrapper
- validate_graph: Graph validation wrapper
- export_report: Report export wrapper
- fix_issues: Automated fixing wrapper
- get_qa_method: Get QA method by name
- list_available_methods: List registered methods
Example Usage:
>>> from semantica.kg_qa import assess_quality, generate_quality_report, KGQualityAssessor
>>> # Using convenience functions
>>> score = assess_quality(knowledge_graph, method="default")
>>> report = generate_quality_report(knowledge_graph, schema, method="default")
>>> # Using classes directly
>>> from semantica.kg_qa import KGQualityAssessor
>>> assessor = KGQualityAssessor()
>>> score = assessor.assess_overall_quality(knowledge_graph)
>>> report = assessor.generate_quality_report(knowledge_graph, schema)
Author: Semantica Contributors
License: MIT
"""
from typing import Any, Dict, List, Optional, Union
from .automated_fixes import AutomatedFixer, AutoMerger, AutoResolver, FixResult
from .config import KGQAConfig, kg_qa_config
from .kg_quality_assessor import (
CompletenessValidator,
ConsistencyChecker,
KGQualityAssessor,
)
from .methods import (
assess_quality,
calculate_quality_metrics,
check_consistency,
export_report,
fix_issues,
generate_quality_report,
get_qa_method,
identify_quality_issues,
list_available_methods,
validate_completeness,
validate_graph,
)
from .quality_metrics import CompletenessMetrics, ConsistencyMetrics, QualityMetrics
from .registry import MethodRegistry, method_registry
from .reporting import (
ImprovementSuggestions,
IssueTracker,
QualityReport,
QualityReporter,
)
from .validation_engine import ConstraintValidator, RuleValidator, ValidationEngine
__all__ = [
# Main classes
"KGQualityAssessor",
"ConsistencyChecker",
"CompletenessValidator",
# Quality metrics
"QualityMetrics",
"CompletenessMetrics",
"ConsistencyMetrics",
# Validation
"ValidationEngine",
"RuleValidator",
"ConstraintValidator",
# Reporting
"QualityReporter",
"IssueTracker",
"ImprovementSuggestions",
"QualityReport",
# Automated fixes
"AutomatedFixer",
"AutoMerger",
"AutoResolver",
"FixResult",
# Registry and Methods
"MethodRegistry",
"method_registry",
"assess_quality",
"generate_quality_report",
"identify_quality_issues",
"check_consistency",
"validate_completeness",
"calculate_quality_metrics",
"validate_graph",
"export_report",
"fix_issues",
"get_qa_method",
"list_available_methods",
# Configuration
"KGQAConfig",
"kg_qa_config",
]
-377
View File
@@ -1,377 +0,0 @@
"""
Automated Fixes Module
This module provides automated fixing capabilities for the Semantica framework,
enabling automatic resolution of common knowledge graph quality issues.
Key Features:
- Duplicate entity and relationship fixing
- Inconsistency resolution
- Missing property completion
- Conflicting property merging
- Conflict and disagreement resolution
Main Classes:
- AutomatedFixer: Main automated fixing engine
- AutoMerger: Automatic merging of duplicates and conflicts
- AutoResolver: Automatic conflict and inconsistency resolution
Example Usage:
>>> from semantica.kg_qa import AutomatedFixer
>>> fixer = AutomatedFixer()
>>> result = fixer.fix_duplicates(knowledge_graph)
>>> result = fixer.fix_missing_properties(knowledge_graph, schema)
Author: Semantica Contributors
License: MIT
"""
from dataclasses import dataclass
from typing import Any, Dict, List, Optional
from ..utils.logging import get_logger
from ..utils.progress_tracker import get_progress_tracker
from .quality_metrics import QualityMetrics
@dataclass
class FixResult:
"""
Fix result dataclass.
This dataclass represents the result of an automated fix operation,
containing success status, number of fixes applied, errors encountered,
and additional metadata.
Attributes:
success: Whether the fix operation was successful
fixed_count: Number of issues fixed
errors: List of error messages encountered during fixing
metadata: Additional metadata about the fix operation
"""
success: bool
fixed_count: int
errors: List[str]
metadata: Dict[str, Any]
class AutomatedFixer:
"""
Automated fixing engine.
This class provides automated fixing capabilities for common knowledge
graph quality issues, including duplicates, inconsistencies, and
missing properties.
Features:
- Duplicate entity fixing
- Inconsistency resolution
- Missing property completion
- Integration with quality metrics
Example Usage:
>>> fixer = AutomatedFixer()
>>> result = fixer.fix_duplicates(knowledge_graph)
>>> if result.success:
... print(f"Fixed {result.fixed_count} issues")
"""
def __init__(self, **kwargs):
"""
Initialize automated fixer.
Sets up the fixer with configuration and quality metrics calculator.
Args:
**kwargs: Configuration options (currently unused)
"""
self.logger = get_logger("automated_fixer")
self.config = kwargs
self.quality_metrics = QualityMetrics()
# Initialize progress tracker
self.progress_tracker = get_progress_tracker()
self.logger.debug("Automated fixer initialized")
def fix_duplicates(self, knowledge_graph: Any) -> FixResult:
"""
Fix duplicate entities.
This method identifies and fixes duplicate entities in the knowledge
graph. In practice, this would use the deduplication module to detect
and merge duplicates.
Args:
knowledge_graph: Knowledge graph instance (object with entities
and relationships, or dict with "entities" and
"relationships" keys)
Returns:
FixResult: Fix result containing:
- success: Whether fixing was successful
- fixed_count: Number of duplicates fixed
- errors: List of error messages
- metadata: Additional fix metadata
"""
# Track duplicate fixing
tracking_id = self.progress_tracker.start_tracking(
file=None,
module="kg_qa",
submodule="AutomatedFixer",
message="Fixing duplicate entities",
)
try:
self.logger.info("Fixing duplicate entities")
self.progress_tracker.update_tracking(
tracking_id, message="Detecting duplicates..."
)
# In practice, this would use deduplication module
# For now, return placeholder
result = FixResult(success=True, fixed_count=0, errors=[], metadata={})
self.progress_tracker.stop_tracking(
tracking_id,
status="completed",
message=f"Fixed {result.fixed_count} duplicate(s)",
)
return result
except Exception as e:
self.progress_tracker.stop_tracking(
tracking_id, status="failed", message=str(e)
)
raise
def fix_inconsistencies(self, knowledge_graph: Any) -> FixResult:
"""
Fix inconsistencies.
This method identifies and fixes logical inconsistencies in the
knowledge graph, such as conflicting property values or contradictory
relationships.
Args:
knowledge_graph: Knowledge graph instance
Returns:
FixResult: Fix result with success status and fix count
"""
self.logger.info("Fixing inconsistencies")
# In practice, this would resolve logical inconsistencies
return FixResult(success=True, fixed_count=0, errors=[], metadata={})
def fix_missing_properties(
self, knowledge_graph: Any, schema: Dict[str, Any]
) -> FixResult:
"""
Fix missing required properties.
This method identifies entities with missing required properties
(as defined in the schema) and attempts to fix them by adding
default values or inferring values from context.
Args:
knowledge_graph: Knowledge graph instance
schema: Schema definition containing required property constraints
Returns:
FixResult: Fix result with number of properties added
"""
self.logger.info("Fixing missing properties")
fixed_count = 0
errors = []
# In practice, this would:
# 1. Find entities with missing required properties
# 2. Add default values or infer values
# 3. Update the knowledge graph
return FixResult(
success=len(errors) == 0,
fixed_count=fixed_count,
errors=errors,
metadata={},
)
class AutoMerger:
"""
Automatic merging engine.
This class provides automatic merging capabilities for duplicate entities,
relationships, and conflicting properties in knowledge graphs.
Features:
- Duplicate entity merging
- Duplicate relationship merging
- Conflicting property resolution
Example Usage:
>>> merger = AutoMerger()
>>> result = merger.merge_duplicate_entities(knowledge_graph)
>>> result = merger.merge_conflicting_properties(knowledge_graph)
"""
def __init__(self, **kwargs):
"""
Initialize auto merger.
Sets up the merger with configuration options.
Args:
**kwargs: Configuration options (currently unused)
"""
self.logger = get_logger("auto_merger")
self.config = kwargs
self.logger.debug("Auto merger initialized")
def merge_duplicate_entities(self, knowledge_graph: Any) -> FixResult:
"""
Merge duplicate entities.
This method identifies duplicate entities and merges them into
single entities, combining properties and updating relationships.
In practice, this would use the deduplication module.
Args:
knowledge_graph: Knowledge graph instance
Returns:
FixResult: Merge result with number of entities merged
"""
self.logger.info("Merging duplicate entities")
# In practice, this would:
# 1. Identify duplicate entities
# 2. Merge properties
# 3. Update relationships
# 4. Remove duplicates
return FixResult(success=True, fixed_count=0, errors=[], metadata={})
def merge_duplicate_relationships(self, knowledge_graph: Any) -> FixResult:
"""
Merge duplicate relationships.
This method identifies duplicate relationships (same source, target,
and type) and merges them, combining properties and metadata.
Args:
knowledge_graph: Knowledge graph instance
Returns:
FixResult: Merge result with number of relationships merged
"""
self.logger.info("Merging duplicate relationships")
return FixResult(success=True, fixed_count=0, errors=[], metadata={})
def merge_conflicting_properties(self, knowledge_graph: Any) -> FixResult:
"""
Merge conflicting properties.
This method identifies entities with conflicting property values
(same property with different values) and resolves conflicts using
configurable strategies (e.g., highest confidence, most recent).
Args:
knowledge_graph: Knowledge graph instance
Returns:
FixResult: Merge result with number of conflicts resolved
"""
self.logger.info("Merging conflicting properties")
return FixResult(success=True, fixed_count=0, errors=[], metadata={})
class AutoResolver:
"""
Automatic resolution engine.
This class provides automatic resolution capabilities for conflicts,
disagreements, and inconsistencies in knowledge graphs.
Features:
- Conflict resolution
- Disagreement resolution
- Inconsistency resolution
Example Usage:
>>> resolver = AutoResolver()
>>> result = resolver.resolve_conflicts(knowledge_graph)
>>> result = resolver.resolve_inconsistencies(knowledge_graph)
"""
def __init__(self, **kwargs):
"""
Initialize auto resolver.
Sets up the resolver with configuration options.
Args:
**kwargs: Configuration options (currently unused)
"""
self.logger = get_logger("auto_resolver")
self.config = kwargs
self.logger.debug("Auto resolver initialized")
def resolve_conflicts(self, knowledge_graph: Any) -> FixResult:
"""
Resolve conflicts.
This method identifies and resolves conflicts in the knowledge graph,
such as conflicting property values or contradictory relationships.
In practice, this would use the conflict resolution module.
Args:
knowledge_graph: Knowledge graph instance
Returns:
FixResult: Resolution result with number of conflicts resolved
"""
self.logger.info("Resolving conflicts")
return FixResult(success=True, fixed_count=0, errors=[], metadata={})
def resolve_disagreements(self, knowledge_graph: Any) -> FixResult:
"""
Resolve disagreements.
This method identifies and resolves disagreements between different
sources or versions of the same information in the knowledge graph.
Args:
knowledge_graph: Knowledge graph instance
Returns:
FixResult: Resolution result with number of disagreements resolved
"""
self.logger.info("Resolving disagreements")
return FixResult(success=True, fixed_count=0, errors=[], metadata={})
def resolve_inconsistencies(self, knowledge_graph: Any) -> FixResult:
"""
Resolve inconsistencies.
This method identifies and resolves logical inconsistencies in the
knowledge graph, such as circular dependencies or contradictory
hierarchical relationships.
Args:
knowledge_graph: Knowledge graph instance
Returns:
FixResult: Resolution result with number of inconsistencies resolved
"""
self.logger.info("Resolving inconsistencies")
return FixResult(success=True, fixed_count=0, errors=[], metadata={})
-167
View File
@@ -1,167 +0,0 @@
"""
Configuration Management Module for KG QA
This module provides centralized configuration management for knowledge graph quality
assurance operations, supporting multiple configuration sources including environment
variables, config files, and programmatic configuration.
Supported Configuration Sources:
- Environment variables: KG_QA_QUALITY_THRESHOLD, KG_QA_CONSISTENCY_THRESHOLD, etc.
- Config files: YAML, JSON, TOML formats
- Programmatic: Python API for setting QA configurations
Algorithms Used:
- Environment Variable Parsing: OS-level environment variable access
- YAML Parsing: YAML parser for configuration file loading
- JSON Parsing: JSON parser for configuration file loading
- TOML Parsing: TOML parser for configuration file loading
- Fallback Chain: Priority-based configuration resolution
- Dictionary Merging: Deep merge algorithms for configuration updates
Key Features:
- Environment variable support for QA parameters
- Config file support (YAML, JSON, TOML formats)
- Programmatic configuration via Python API
- Method-specific configuration management
- Automatic fallback chain (config file -> environment -> defaults)
- Global config instance for easy access
Main Classes:
- KGQAConfig: Main configuration manager class for kg_qa module
Example Usage:
>>> from semantica.kg_qa.config import kg_qa_config
>>> threshold = kg_qa_config.get("quality_threshold", default=0.7)
>>> kg_qa_config.set("quality_threshold", 0.8)
>>> method_config = kg_qa_config.get_method_config("assess")
"""
import os
from pathlib import Path
from typing import Any, Dict, Optional
from ..utils.logging import get_logger
class KGQAConfig:
"""Configuration manager for KG QA module - supports .env files, environment variables, and programmatic config."""
def __init__(self, config_file: Optional[str] = None):
self.logger = get_logger("kg_qa_config")
self._configs: Dict[str, Any] = {}
self._method_configs: Dict[str, Dict] = {}
self._load_config_file(config_file)
self._load_env_vars()
def _load_config_file(self, config_file: Optional[str]):
if config_file and Path(config_file).exists():
try:
if config_file.endswith(".yaml") or config_file.endswith(".yml"):
import yaml
with open(config_file, "r") as f:
data = yaml.safe_load(f) or {}
self._configs.update(data.get("kg_qa", {}))
self._method_configs.update(data.get("kg_qa_methods", {}))
elif config_file.endswith(".json"):
import json
with open(config_file, "r") as f:
data = json.load(f) or {}
self._configs.update(data.get("kg_qa", {}))
self._method_configs.update(data.get("kg_qa_methods", {}))
elif config_file.endswith(".toml"):
import toml
with open(config_file, "r") as f:
data = toml.load(f) or {}
if "kg_qa" in data:
self._configs.update(data["kg_qa"])
if "kg_qa_methods" in data:
self._method_configs.update(data["kg_qa_methods"])
self.logger.info(f"Loaded KG QA config from {config_file}")
except Exception as e:
self.logger.warning(f"Failed to load config file {config_file}: {e}")
def _load_env_vars(self):
env_mappings = {
"KG_QA_QUALITY_THRESHOLD": ("quality_threshold", float),
"KG_QA_CONSISTENCY_THRESHOLD": ("consistency_threshold", float),
"KG_QA_COMPLETENESS_THRESHOLD": ("completeness_threshold", float),
"KG_QA_ENABLE_AUTO_FIX": ("enable_auto_fix", bool),
"KG_QA_REPORT_FORMAT": ("report_format", str),
}
for env_key, (config_key, type_func) in env_mappings.items():
value = os.getenv(env_key)
if value:
try:
if type_func == bool:
self._configs[config_key] = value.lower() in (
"true",
"1",
"yes",
"on",
)
else:
self._configs[config_key] = type_func(value)
except (ValueError, TypeError):
self.logger.warning(f"Failed to parse {env_key}={value}")
env_prefix = "KG_QA_"
for key, value in os.environ.items():
if key.startswith(env_prefix) and key not in env_mappings:
config_key = key[len(env_prefix) :].lower()
if value.lower() in ("true", "false"):
self._configs[config_key] = value.lower() == "true"
elif value.isdigit():
self._configs[config_key] = int(value)
else:
try:
self._configs[config_key] = float(value)
except ValueError:
self._configs[config_key] = value
def set(self, key: str, value: Any):
"""Set configuration value programmatically."""
self._configs[key] = value
def get(self, key: str, default: Any = None) -> Any:
"""Get configuration value with fallback chain: config -> env -> default."""
if key in self._configs:
return self._configs[key]
env_key = f"KG_QA_{key.upper()}"
value = os.getenv(env_key)
if value:
try:
if isinstance(default, int):
return int(value)
elif isinstance(default, float):
return float(value)
elif isinstance(default, bool):
return value.lower() in ("true", "1", "yes", "on")
return value
except (ValueError, TypeError):
pass
return default
def set_method_config(self, method: str, **config):
"""Set method-specific configuration."""
self._method_configs[method] = config
def get_method_config(self, method: str) -> Dict:
"""Get method-specific configuration."""
return self._method_configs.get(method, {})
def get_all(self) -> Dict[str, Any]:
"""Get all configuration."""
return {
"config": self._configs.copy(),
"method_configs": self._method_configs.copy(),
}
# Global config instance
kg_qa_config = KGQAConfig()
File diff suppressed because it is too large Load Diff
-421
View File
@@ -1,421 +0,0 @@
"""
KG Quality Assessor Module
This module provides the main quality assessment coordination for the Semantica
framework, integrating all quality assurance components to provide comprehensive
quality assessment and reporting.
Key Features:
- Overall quality assessment
- Quality report generation
- Quality issue identification
- Consistency checking
- Completeness validation
Main Classes:
- KGQualityAssessor: Main quality assessment coordinator
- ConsistencyChecker: Consistency validation engine
- CompletenessValidator: Completeness validation engine
Example Usage:
>>> from semantica.kg_qa import KGQualityAssessor
>>> assessor = KGQualityAssessor()
>>> score = assessor.assess_overall_quality(knowledge_graph)
>>> report = assessor.generate_quality_report(knowledge_graph, schema)
>>> issues = assessor.identify_quality_issues(knowledge_graph, schema)
Author: Semantica Contributors
License: MIT
"""
from typing import Any, Dict, List, Optional
from ..utils.logging import get_logger
from ..utils.progress_tracker import get_progress_tracker
from .quality_metrics import CompletenessMetrics, ConsistencyMetrics, QualityMetrics
from .reporting import QualityReport, QualityReporter
from .validation_engine import ValidationEngine
class KGQualityAssessor:
"""
Knowledge Graph Quality Assessor.
This class serves as the main coordinator for knowledge graph quality
assessment, integrating quality metrics, validation, and reporting
components to provide comprehensive quality analysis.
Features:
- Overall quality score calculation
- Comprehensive quality report generation
- Quality issue identification
- Integration with all QA components
Example Usage:
>>> assessor = KGQualityAssessor()
>>> score = assessor.assess_overall_quality(knowledge_graph)
>>> report = assessor.generate_quality_report(knowledge_graph, schema)
>>> issues = assessor.identify_quality_issues(knowledge_graph, schema)
"""
def __init__(self, **kwargs):
"""
Initialize KG quality assessor.
Sets up the assessor with all quality assurance components including
quality metrics, completeness metrics, consistency metrics, validation
engine, and quality reporter.
Args:
**kwargs: Configuration options passed to all components
"""
self.logger = get_logger("kg_quality_assessor")
self.config = kwargs
# Initialize components
self.quality_metrics = QualityMetrics(**kwargs)
self.completeness_metrics = CompletenessMetrics(**kwargs)
self.consistency_metrics = ConsistencyMetrics(**kwargs)
self.validation_engine = ValidationEngine(**kwargs)
self.quality_reporter = QualityReporter(**kwargs)
# Initialize progress tracker
self.progress_tracker = get_progress_tracker()
self.logger.debug("KG quality assessor initialized")
def assess_overall_quality(self, knowledge_graph: Any) -> float:
"""
Assess overall quality of knowledge graph.
This method calculates an overall quality score for the knowledge
graph by aggregating various quality metrics (completeness, consistency,
etc.) into a single score.
Args:
knowledge_graph: Knowledge graph instance (object with entities
and relationships, or dict with "entities" and
"relationships" keys)
Returns:
float: Overall quality score between 0.0 and 1.0 (higher is better)
"""
# Track quality assessment
tracking_id = self.progress_tracker.start_tracking(
file=None,
module="kg_qa",
submodule="KGQualityAssessor",
message="Assessing overall quality",
)
try:
self.logger.info("Assessing overall quality")
self.progress_tracker.update_tracking(
tracking_id, message="Calculating quality metrics..."
)
# Calculate metrics
overall_score = self.quality_metrics.calculate_overall_score(
knowledge_graph
)
self.progress_tracker.stop_tracking(
tracking_id,
status="completed",
message=f"Overall quality score: {overall_score:.2f}",
)
return overall_score
except Exception as e:
self.progress_tracker.stop_tracking(
tracking_id, status="failed", message=str(e)
)
raise
def generate_quality_report(
self, knowledge_graph: Any, schema: Optional[Dict[str, Any]] = None
) -> QualityReport:
"""
Generate comprehensive quality report.
This method generates a comprehensive quality report including overall
quality score, completeness score, consistency score, identified issues,
and improvement recommendations.
Args:
knowledge_graph: Knowledge graph instance
schema: Optional schema definition for validation (if provided,
enables completeness checking against schema constraints)
Returns:
QualityReport: Comprehensive quality report containing:
- timestamp: Report generation timestamp
- overall_score: Overall quality score
- completeness_score: Completeness score
- consistency_score: Consistency score
- issues: List of identified quality issues
- recommendations: List of improvement recommendations
- metadata: Additional report metadata
"""
self.logger.info("Generating quality report")
# Calculate metrics
overall_score = self.quality_metrics.calculate_overall_score(knowledge_graph)
# Get entities and relationships (simplified - in practice would query graph)
entities = getattr(knowledge_graph, "entities", [])
relationships = getattr(knowledge_graph, "relationships", [])
completeness_score = 0.0
if schema and entities:
completeness_score = (
self.completeness_metrics.calculate_entity_completeness(
entities, schema
)
)
consistency_score = self.consistency_metrics.calculate_logical_consistency(
knowledge_graph
)
quality_metrics = {
"overall": overall_score,
"completeness": completeness_score,
"consistency": consistency_score,
}
# Generate report
report = self.quality_reporter.generate_report(knowledge_graph, quality_metrics)
return report
def identify_quality_issues(
self, knowledge_graph: Any, schema: Optional[Dict[str, Any]] = None
) -> List[Dict[str, Any]]:
"""
Identify quality issues in knowledge graph.
This method identifies and returns all quality issues found in the
knowledge graph, including completeness issues, consistency issues,
and other quality problems.
Args:
knowledge_graph: Knowledge graph instance
schema: Optional schema for validation
Returns:
list: List of quality issue dictionaries, each containing:
- id: Issue identifier
- type: Issue type (e.g., "completeness", "consistency")
- severity: Issue severity ("low", "medium", "high")
- description: Issue description
- entity_id: Related entity ID (if applicable)
- relationship_id: Related relationship ID (if applicable)
"""
self.logger.info("Identifying quality issues")
# Generate report to get issues
report = self.generate_quality_report(knowledge_graph, schema)
# Convert issues to dictionaries
issues = [
{
"id": issue.id,
"type": issue.type,
"severity": issue.severity,
"description": issue.description,
"entity_id": issue.entity_id,
"relationship_id": issue.relationship_id,
}
for issue in report.issues
]
return issues
class ConsistencyChecker:
"""
Consistency checking engine.
This class provides consistency checking capabilities for knowledge graphs,
validating logical, temporal, and hierarchical consistency.
Features:
- Logical consistency checking
- Temporal consistency checking
- Hierarchical consistency checking
Example Usage:
>>> checker = ConsistencyChecker()
>>> is_logical = checker.check_logical_consistency(knowledge_graph)
>>> is_temporal = checker.check_temporal_consistency(knowledge_graph)
"""
def __init__(self, **kwargs):
"""
Initialize consistency checker.
Sets up the checker with consistency metrics calculator.
Args:
**kwargs: Configuration options passed to ConsistencyMetrics
"""
self.logger = get_logger("consistency_checker")
self.consistency_metrics = ConsistencyMetrics(**kwargs)
self.config = kwargs
self.logger.debug("Consistency checker initialized")
def check_logical_consistency(self, knowledge_graph: Any) -> bool:
"""
Check logical consistency.
This method checks for logical inconsistencies in the knowledge graph,
such as contradictory relationships or conflicting property values.
Returns True if the consistency score is above the threshold (0.8).
Args:
knowledge_graph: Knowledge graph instance
Returns:
bool: True if logically consistent (score >= 0.8), False otherwise
"""
score = self.consistency_metrics.calculate_logical_consistency(knowledge_graph)
return score >= 0.8
def check_temporal_consistency(self, knowledge_graph: Any) -> bool:
"""
Check temporal consistency.
This method checks for temporal inconsistencies in the knowledge graph,
such as relationships with invalid time ranges or temporal contradictions.
Returns True if the consistency score is above the threshold (0.8).
Args:
knowledge_graph: Knowledge graph instance
Returns:
bool: True if temporally consistent (score >= 0.8), False otherwise
"""
score = self.consistency_metrics.calculate_temporal_consistency(knowledge_graph)
return score >= 0.8
def check_hierarchical_consistency(self, knowledge_graph: Any) -> bool:
"""
Check hierarchical consistency.
This method checks for hierarchical inconsistencies in the knowledge
graph, such as circular inheritance or invalid parent-child relationships.
Returns True if the consistency score is above the threshold (0.8).
Args:
knowledge_graph: Knowledge graph instance
Returns:
bool: True if hierarchically consistent (score >= 0.8), False otherwise
"""
score = self.consistency_metrics.calculate_hierarchical_consistency(
knowledge_graph
)
return score >= 0.8
class CompletenessValidator:
"""
Completeness validation engine.
This class provides completeness validation capabilities for knowledge graphs,
checking whether entities, relationships, and properties meet schema
requirements.
Features:
- Entity completeness validation
- Relationship completeness validation
- Property completeness validation
Example Usage:
>>> validator = CompletenessValidator()
>>> is_complete = validator.validate_entity_completeness(entities, schema)
>>> is_rel_complete = validator.validate_relationship_completeness(relationships, schema)
"""
def __init__(self, **kwargs):
"""
Initialize completeness validator.
Sets up the validator with completeness metrics calculator.
Args:
**kwargs: Configuration options passed to CompletenessMetrics
"""
self.logger = get_logger("completeness_validator")
self.completeness_metrics = CompletenessMetrics(**kwargs)
self.config = kwargs
self.logger.debug("Completeness validator initialized")
def validate_entity_completeness(
self, entities: List[Dict[str, Any]], schema: Dict[str, Any]
) -> bool:
"""
Validate entity completeness.
This method validates whether entities have all required properties
as defined in the schema. Returns True if the completeness score
is above the threshold (0.8).
Args:
entities: List of entity dictionaries
schema: Schema definition containing required property constraints
Returns:
bool: True if entities are complete (score >= 0.8), False otherwise
"""
score = self.completeness_metrics.calculate_entity_completeness(
entities, schema
)
return score >= 0.8
def validate_relationship_completeness(
self, relationships: List[Dict[str, Any]], schema: Dict[str, Any]
) -> bool:
"""
Validate relationship completeness.
This method validates whether relationships have all required properties
as defined in the schema. Returns True if the completeness score
is above the threshold (0.8).
Args:
relationships: List of relationship dictionaries
schema: Schema definition containing relationship constraints
Returns:
bool: True if relationships are complete (score >= 0.8), False otherwise
"""
score = self.completeness_metrics.calculate_relationship_completeness(
relationships, schema
)
return score >= 0.8
def validate_property_completeness(
self, properties: Dict[str, Any], schema: Dict[str, Any]
) -> bool:
"""
Validate property completeness.
This method validates whether properties meet schema requirements
for completeness. Returns True if the completeness score is above
the threshold (0.8).
Args:
properties: Properties dictionary (mapping entity types to property dicts)
schema: Schema definition containing property constraints
Returns:
bool: True if properties are complete (score >= 0.8), False otherwise
"""
score = self.completeness_metrics.calculate_property_completeness(
properties, schema
)
return score >= 0.8
-747
View File
@@ -1,747 +0,0 @@
"""
Knowledge Graph Quality Assurance Methods Module
This module provides all KG QA methods as simple, reusable functions for
quality assessment, validation, reporting, and automated fixing. It supports
multiple approaches and integrates with the method registry for extensibility.
Supported Methods:
Quality Assessment:
- "default": Default quality assessment using KGQualityAssessor
- "comprehensive": Comprehensive assessment with all metrics
- "quick": Quick assessment with basic metrics
Quality Reporting:
- "default": Default report generation
- "detailed": Detailed report with all issues
- "summary": Summary report only
Consistency Checking:
- "logical": Logical consistency checking
- "temporal": Temporal consistency checking
- "hierarchical": Hierarchical consistency checking
- "all": All consistency checks
Completeness Validation:
- "entity": Entity completeness validation
- "relationship": Relationship completeness validation
- "property": Property completeness validation
- "all": All completeness checks
Quality Metrics:
- "overall": Overall quality score
- "entity": Entity quality score
- "relationship": Relationship quality score
- "completeness": Completeness metrics
- "consistency": Consistency metrics
Validation:
- "default": Default validation with stored rules
- "custom": Custom rule validation
- "constraints": Constraint-based validation
Automated Fixes:
- "duplicates": Fix duplicate entities
- "inconsistencies": Fix inconsistencies
- "missing_properties": Fix missing properties
- "all": Apply all fixes
Algorithms Used:
Quality Metrics Calculation:
- Weighted Averaging: Overall quality score aggregation using weighted average formula: overall = (0.6 * completeness) + (0.4 * consistency)
- Entity Quality Scoring: Required field presence checking (ID/URI, type), binary scoring (0.5 per field), average calculation across entities
- Relationship Quality Scoring: Required field presence checking (source/subject, target/object, type/predicate), weighted scoring (0.33 per field), average calculation across relationships
- Score Normalization: Min-max normalization with clamping to 0.0-1.0 range
Completeness Metrics:
- Entity Completeness Calculation: Schema-based required property validation, ratio calculation present_props / required_props, average completeness across entities
- Relationship Completeness Calculation: Required field validation (source, target, type), completeness ratio calculation, average across relationships
- Property Completeness Calculation: Schema-based property validation per entity type, completeness ratio calculation, average across entity types
Consistency Metrics:
- Logical Consistency Checking: Contradiction detection, conflicting relationship identification, inconsistent property value detection
- Temporal Consistency Checking: Temporal contradiction detection, invalid time range validation, conflicting temporal relationship identification
- Hierarchical Consistency Checking: Circular inheritance detection (DFS-based cycle detection), invalid parent-child relationship validation
Validation Engine:
- Rule-Based Validation: Custom rule function execution, rule result parsing (error/warning extraction), exception handling and error collection
- Constraint-Based Validation: Entity constraint validation (required properties), relationship constraint validation (domain and range), constraint matching algorithms
Quality Reporting:
- Issue Identification: Threshold-based issue detection (overall < 0.7, completeness < 0.8), issue type classification, severity assignment
- Recommendation Generation: Issue-based recommendation generation, score-based recommendation generation, actionable suggestion creation
- Report Serialization: JSON serialization (ISO timestamp formatting), YAML serialization (with PyYAML fallback), HTML report generation
Automated Fixes:
- Duplicate Detection: Entity duplicate identification (using deduplication module), relationship duplicate identification
- Duplicate Merging: Property aggregation strategies, relationship reference updating, entity consolidation
- Conflict Resolution: Conflicting property value detection, resolution strategy selection, conflict merging
- Missing Property Completion: Schema-based required property identification, default value assignment, value inference
Key Features:
- Multiple QA operation methods
- Quality assessment with method dispatch
- Method dispatchers with registry support
- Custom method registration capability
- Consistent interface across all methods
Main Functions:
- assess_quality: Quality assessment wrapper
- generate_quality_report: Quality report generation wrapper
- identify_quality_issues: Quality issue identification wrapper
- check_consistency: Consistency checking wrapper
- validate_completeness: Completeness validation wrapper
- calculate_quality_metrics: Quality metrics calculation wrapper
- validate_graph: Graph validation wrapper
- export_report: Report export wrapper
- fix_issues: Automated fixing wrapper
- get_qa_method: Get QA method by name
- list_available_methods: List registered methods
Example Usage:
>>> from semantica.kg_qa.methods import assess_quality, generate_quality_report
>>> score = assess_quality(knowledge_graph, method="default")
>>> report = generate_quality_report(knowledge_graph, schema, method="default")
"""
from typing import Any, Callable, Dict, List, Optional, Union
from ..utils.exceptions import ConfigurationError, ProcessingError
from ..utils.logging import get_logger
from .automated_fixes import AutomatedFixer, FixResult
from .config import kg_qa_config
from .kg_quality_assessor import (
CompletenessValidator,
ConsistencyChecker,
KGQualityAssessor,
)
from .quality_metrics import CompletenessMetrics, ConsistencyMetrics, QualityMetrics
from .registry import method_registry
from .reporting import QualityReport, QualityReporter
from .validation_engine import ValidationEngine
logger = get_logger("kg_qa_methods")
def assess_quality(knowledge_graph: Any, method: str = "default", **kwargs) -> float:
"""
Assess overall quality of knowledge graph (convenience function).
This is a user-friendly wrapper that assesses knowledge graph quality
using the specified method.
Args:
knowledge_graph: Knowledge graph instance (object with entities
and relationships, or dict with "entities" and
"relationships" keys)
method: Assessment method (default: "default")
- "default": Use KGQualityAssessor with default settings
- "comprehensive": Comprehensive assessment with all metrics
- "quick": Quick assessment with basic metrics
**kwargs: Additional options passed to KGQualityAssessor
Returns:
float: Overall quality score between 0.0 and 1.0 (higher is better)
Examples:
>>> from semantica.kg_qa.methods import assess_quality
>>> score = assess_quality(knowledge_graph, method="default")
>>> quick_score = assess_quality(knowledge_graph, method="quick")
"""
custom_method = method_registry.get("assess", method)
if custom_method:
try:
return custom_method(knowledge_graph, **kwargs)
except Exception as e:
logger.warning(
f"Custom method {method} failed: {e}, falling back to default"
)
try:
config = kg_qa_config.get_method_config("assess")
config.update(kwargs)
assessor = KGQualityAssessor(**config)
return assessor.assess_overall_quality(knowledge_graph)
except Exception as e:
logger.error(f"Failed to assess quality: {e}")
raise
def generate_quality_report(
knowledge_graph: Any,
schema: Optional[Dict[str, Any]] = None,
method: str = "default",
**kwargs,
) -> QualityReport:
"""
Generate comprehensive quality report (convenience function).
This is a user-friendly wrapper that generates a quality report using
the specified method.
Args:
knowledge_graph: Knowledge graph instance
schema: Optional schema definition for validation
method: Report generation method (default: "default")
- "default": Use KGQualityAssessor with default settings
- "detailed": Detailed report with all issues
- "summary": Summary report only
**kwargs: Additional options passed to KGQualityAssessor
Returns:
QualityReport: Comprehensive quality report containing:
- timestamp: Report generation timestamp
- overall_score: Overall quality score
- completeness_score: Completeness score
- consistency_score: Consistency score
- issues: List of identified quality issues
- recommendations: List of improvement recommendations
- metadata: Additional report metadata
Examples:
>>> from semantica.kg_qa.methods import generate_quality_report
>>> report = generate_quality_report(knowledge_graph, schema, method="default")
"""
custom_method = method_registry.get("report", method)
if custom_method:
try:
return custom_method(knowledge_graph, schema, **kwargs)
except Exception as e:
logger.warning(
f"Custom method {method} failed: {e}, falling back to default"
)
try:
config = kg_qa_config.get_method_config("report")
config.update(kwargs)
assessor = KGQualityAssessor(**config)
return assessor.generate_quality_report(knowledge_graph, schema)
except Exception as e:
logger.error(f"Failed to generate quality report: {e}")
raise
def identify_quality_issues(
knowledge_graph: Any,
schema: Optional[Dict[str, Any]] = None,
method: str = "default",
**kwargs,
) -> List[Dict[str, Any]]:
"""
Identify quality issues in knowledge graph (convenience function).
This is a user-friendly wrapper that identifies quality issues using
the specified method.
Args:
knowledge_graph: Knowledge graph instance
schema: Optional schema for validation
method: Issue identification method (default: "default")
**kwargs: Additional options passed to KGQualityAssessor
Returns:
list: List of quality issue dictionaries, each containing:
- id: Issue identifier
- type: Issue type (e.g., "completeness", "consistency")
- severity: Issue severity ("low", "medium", "high")
- description: Issue description
- entity_id: Related entity ID (if applicable)
- relationship_id: Related relationship ID (if applicable)
Examples:
>>> from semantica.kg_qa.methods import identify_quality_issues
>>> issues = identify_quality_issues(knowledge_graph, schema, method="default")
"""
custom_method = method_registry.get("assess", method)
if custom_method:
try:
return custom_method(knowledge_graph, schema, **kwargs)
except Exception as e:
logger.warning(
f"Custom method {method} failed: {e}, falling back to default"
)
try:
config = kg_qa_config.get_method_config("assess")
config.update(kwargs)
assessor = KGQualityAssessor(**config)
return assessor.identify_quality_issues(knowledge_graph, schema)
except Exception as e:
logger.error(f"Failed to identify quality issues: {e}")
raise
def check_consistency(
knowledge_graph: Any,
consistency_type: str = "logical",
method: str = "default",
**kwargs,
) -> Union[bool, Dict[str, bool]]:
"""
Check consistency of knowledge graph (convenience function).
This is a user-friendly wrapper that checks consistency using
the specified method.
Args:
knowledge_graph: Knowledge graph instance
consistency_type: Type of consistency to check (default: "logical")
- "logical": Logical consistency checking
- "temporal": Temporal consistency checking
- "hierarchical": Hierarchical consistency checking
- "all": All consistency checks (returns dict)
method: Consistency checking method (default: "default")
**kwargs: Additional options passed to ConsistencyChecker
Returns:
bool or dict: Consistency check result(s)
- If consistency_type is "all", returns dict with keys:
"logical", "temporal", "hierarchical"
- Otherwise returns bool (True if consistent)
Examples:
>>> from semantica.kg_qa.methods import check_consistency
>>> is_consistent = check_consistency(knowledge_graph, consistency_type="logical")
>>> all_checks = check_consistency(knowledge_graph, consistency_type="all")
"""
custom_method = method_registry.get("consistency", method)
if custom_method:
try:
return custom_method(knowledge_graph, consistency_type, **kwargs)
except Exception as e:
logger.warning(
f"Custom method {method} failed: {e}, falling back to default"
)
try:
config = kg_qa_config.get_method_config("consistency")
config.update(kwargs)
checker = ConsistencyChecker(**config)
if consistency_type == "all":
return {
"logical": checker.check_logical_consistency(knowledge_graph),
"temporal": checker.check_temporal_consistency(knowledge_graph),
"hierarchical": checker.check_hierarchical_consistency(knowledge_graph),
}
elif consistency_type == "logical":
return checker.check_logical_consistency(knowledge_graph)
elif consistency_type == "temporal":
return checker.check_temporal_consistency(knowledge_graph)
elif consistency_type == "hierarchical":
return checker.check_hierarchical_consistency(knowledge_graph)
else:
raise ValueError(f"Unknown consistency type: {consistency_type}")
except Exception as e:
logger.error(f"Failed to check consistency: {e}")
raise
def validate_completeness(
entities: Optional[List[Dict[str, Any]]] = None,
relationships: Optional[List[Dict[str, Any]]] = None,
properties: Optional[Dict[str, Any]] = None,
schema: Dict[str, Any] = None,
completeness_type: str = "entity",
method: str = "default",
**kwargs,
) -> Union[bool, Dict[str, bool]]:
"""
Validate completeness of knowledge graph (convenience function).
This is a user-friendly wrapper that validates completeness using
the specified method.
Args:
entities: Optional list of entity dictionaries
relationships: Optional list of relationship dictionaries
properties: Optional properties dictionary
schema: Schema definition containing constraints
completeness_type: Type of completeness to validate (default: "entity")
- "entity": Entity completeness validation
- "relationship": Relationship completeness validation
- "property": Property completeness validation
- "all": All completeness checks (returns dict)
method: Completeness validation method (default: "default")
**kwargs: Additional options passed to CompletenessValidator
Returns:
bool or dict: Completeness validation result(s)
- If completeness_type is "all", returns dict with keys:
"entity", "relationship", "property"
- Otherwise returns bool (True if complete)
Examples:
>>> from semantica.kg_qa.methods import validate_completeness
>>> is_complete = validate_completeness(entities, schema, completeness_type="entity")
>>> all_checks = validate_completeness(entities, relationships, properties, schema, completeness_type="all")
"""
custom_method = method_registry.get("completeness", method)
if custom_method:
try:
return custom_method(
entities, relationships, properties, schema, completeness_type, **kwargs
)
except Exception as e:
logger.warning(
f"Custom method {method} failed: {e}, falling back to default"
)
try:
config = kg_qa_config.get_method_config("completeness")
config.update(kwargs)
validator = CompletenessValidator(**config)
if completeness_type == "all":
results = {}
if entities and schema:
results["entity"] = validator.validate_entity_completeness(
entities, schema
)
if relationships and schema:
results["relationship"] = validator.validate_relationship_completeness(
relationships, schema
)
if properties and schema:
results["property"] = validator.validate_property_completeness(
properties, schema
)
return results
elif completeness_type == "entity":
if not entities or not schema:
raise ValueError(
"entities and schema are required for entity completeness validation"
)
return validator.validate_entity_completeness(entities, schema)
elif completeness_type == "relationship":
if not relationships or not schema:
raise ValueError(
"relationships and schema are required for relationship completeness validation"
)
return validator.validate_relationship_completeness(relationships, schema)
elif completeness_type == "property":
if not properties or not schema:
raise ValueError(
"properties and schema are required for property completeness validation"
)
return validator.validate_property_completeness(properties, schema)
else:
raise ValueError(f"Unknown completeness type: {completeness_type}")
except Exception as e:
logger.error(f"Failed to validate completeness: {e}")
raise
def calculate_quality_metrics(
knowledge_graph: Any,
metrics_type: str = "overall",
method: str = "default",
**kwargs,
) -> Union[float, Dict[str, float]]:
"""
Calculate quality metrics for knowledge graph (convenience function).
This is a user-friendly wrapper that calculates quality metrics using
the specified method.
Args:
knowledge_graph: Knowledge graph instance
metrics_type: Type of metrics to calculate (default: "overall")
- "overall": Overall quality score
- "entity": Entity quality score
- "relationship": Relationship quality score
- "completeness": Completeness metrics
- "consistency": Consistency metrics
- "all": All metrics (returns dict)
method: Metrics calculation method (default: "default")
**kwargs: Additional options passed to QualityMetrics
Returns:
float or dict: Quality metric(s)
- If metrics_type is "all", returns dict with all metrics
- Otherwise returns float score
Examples:
>>> from semantica.kg_qa.methods import calculate_quality_metrics
>>> score = calculate_quality_metrics(knowledge_graph, metrics_type="overall")
>>> all_metrics = calculate_quality_metrics(knowledge_graph, metrics_type="all")
"""
custom_method = method_registry.get("metrics", method)
if custom_method:
try:
return custom_method(knowledge_graph, metrics_type, **kwargs)
except Exception as e:
logger.warning(
f"Custom method {method} failed: {e}, falling back to default"
)
try:
config = kg_qa_config.get_method_config("metrics")
config.update(kwargs)
quality_metrics = QualityMetrics(**config)
entities = getattr(
knowledge_graph,
"entities",
knowledge_graph.get("entities", [])
if isinstance(knowledge_graph, dict)
else [],
)
relationships = getattr(
knowledge_graph,
"relationships",
knowledge_graph.get("relationships", [])
if isinstance(knowledge_graph, dict)
else [],
)
if metrics_type == "all":
return {
"overall": quality_metrics.calculate_overall_score(knowledge_graph),
"entity": quality_metrics.calculate_entity_quality(entities)
if entities
else 0.0,
"relationship": quality_metrics.calculate_relationship_quality(
relationships
)
if relationships
else 0.0,
}
elif metrics_type == "overall":
return quality_metrics.calculate_overall_score(knowledge_graph)
elif metrics_type == "entity":
if not entities:
raise ValueError(
"Knowledge graph must have entities for entity quality calculation"
)
return quality_metrics.calculate_entity_quality(entities)
elif metrics_type == "relationship":
if not relationships:
raise ValueError(
"Knowledge graph must have relationships for relationship quality calculation"
)
return quality_metrics.calculate_relationship_quality(relationships)
else:
raise ValueError(f"Unknown metrics type: {metrics_type}")
except Exception as e:
logger.error(f"Failed to calculate quality metrics: {e}")
raise
def validate_graph(
knowledge_graph: Any,
rules: Optional[List[Callable]] = None,
method: str = "default",
**kwargs,
) -> Any:
"""
Validate knowledge graph (convenience function).
This is a user-friendly wrapper that validates a knowledge graph using
the specified method.
Args:
knowledge_graph: Knowledge graph instance to validate
rules: Optional list of validation rule functions
method: Validation method (default: "default")
- "default": Default validation with stored rules
- "custom": Custom rule validation
- "constraints": Constraint-based validation
**kwargs: Additional options passed to ValidationEngine
Returns:
ValidationResult: Validation result containing:
- valid: True if no errors, False otherwise
- errors: List of error messages
- warnings: List of warning messages
- metadata: Additional validation metadata
Examples:
>>> from semantica.kg_qa.methods import validate_graph
>>> result = validate_graph(knowledge_graph, method="default")
"""
custom_method = method_registry.get("validate", method)
if custom_method:
try:
return custom_method(knowledge_graph, rules, **kwargs)
except Exception as e:
logger.warning(
f"Custom method {method} failed: {e}, falling back to default"
)
try:
config = kg_qa_config.get_method_config("validate")
config.update(kwargs)
engine = ValidationEngine(**config)
return engine.validate(knowledge_graph, rules)
except Exception as e:
logger.error(f"Failed to validate graph: {e}")
raise
def export_report(
report: QualityReport, format: str = "json", method: str = "default", **kwargs
) -> str:
"""
Export quality report to specified format (convenience function).
This is a user-friendly wrapper that exports a quality report using
the specified method.
Args:
report: Quality report to export
format: Export format (default: "json")
- "json": JSON format
- "yaml": YAML format
- "html": HTML format (planned)
method: Export method (default: "default")
**kwargs: Additional options passed to QualityReporter
Returns:
str: Exported report as string in the specified format
Examples:
>>> from semantica.kg_qa.methods import export_report
>>> json_report = export_report(report, format="json")
>>> yaml_report = export_report(report, format="yaml")
"""
custom_method = method_registry.get("report", method)
if custom_method:
try:
return custom_method(report, format, **kwargs)
except Exception as e:
logger.warning(
f"Custom method {method} failed: {e}, falling back to default"
)
try:
config = kg_qa_config.get_method_config("report")
config.update(kwargs)
reporter = QualityReporter(**config)
return reporter.export_report(report, format=format)
except Exception as e:
logger.error(f"Failed to export report: {e}")
raise
def fix_issues(
knowledge_graph: Any,
fix_type: str = "duplicates",
schema: Optional[Dict[str, Any]] = None,
method: str = "default",
**kwargs,
) -> FixResult:
"""
Fix quality issues in knowledge graph (convenience function).
This is a user-friendly wrapper that fixes quality issues using
the specified method.
Args:
knowledge_graph: Knowledge graph instance
fix_type: Type of fix to apply (default: "duplicates")
- "duplicates": Fix duplicate entities
- "inconsistencies": Fix inconsistencies
- "missing_properties": Fix missing properties
- "all": Apply all fixes
schema: Optional schema definition (required for missing_properties)
method: Fixing method (default: "default")
**kwargs: Additional options passed to AutomatedFixer
Returns:
FixResult: Fix result containing:
- success: Whether fixing was successful
- fixed_count: Number of issues fixed
- errors: List of error messages
- metadata: Additional fix metadata
Examples:
>>> from semantica.kg_qa.methods import fix_issues
>>> result = fix_issues(knowledge_graph, fix_type="duplicates")
>>> result = fix_issues(knowledge_graph, fix_type="missing_properties", schema=schema)
"""
custom_method = method_registry.get("fix", method)
if custom_method:
try:
return custom_method(knowledge_graph, fix_type, schema, **kwargs)
except Exception as e:
logger.warning(
f"Custom method {method} failed: {e}, falling back to default"
)
try:
config = kg_qa_config.get_method_config("fix")
config.update(kwargs)
fixer = AutomatedFixer(**config)
if fix_type == "all":
# Apply all fixes sequentially
results = []
results.append(fixer.fix_duplicates(knowledge_graph))
results.append(fixer.fix_inconsistencies(knowledge_graph))
if schema:
results.append(fixer.fix_missing_properties(knowledge_graph, schema))
total_fixed = sum(r.fixed_count for r in results)
all_errors = []
for r in results:
all_errors.extend(r.errors)
return FixResult(
success=all(r.success for r in results),
fixed_count=total_fixed,
errors=all_errors,
metadata={"fixes_applied": [fix_type for r in results if r.success]},
)
elif fix_type == "duplicates":
return fixer.fix_duplicates(knowledge_graph)
elif fix_type == "inconsistencies":
return fixer.fix_inconsistencies(knowledge_graph)
elif fix_type == "missing_properties":
if not schema:
raise ValueError("schema is required for missing_properties fix")
return fixer.fix_missing_properties(knowledge_graph, schema)
else:
raise ValueError(f"Unknown fix type: {fix_type}")
except Exception as e:
logger.error(f"Failed to fix issues: {e}")
raise
def get_qa_method(task: str, name: str) -> Optional[Callable]:
"""Get QA method by task and name."""
return method_registry.get(task, name)
def list_available_methods(task: Optional[str] = None) -> Dict[str, List[str]]:
"""List all registered QA methods."""
return method_registry.list_all(task)
# Register default methods
method_registry.register("assess", "default", assess_quality)
method_registry.register("report", "default", generate_quality_report)
method_registry.register("consistency", "default", check_consistency)
method_registry.register("completeness", "default", validate_completeness)
method_registry.register("metrics", "default", calculate_quality_metrics)
method_registry.register("validate", "default", validate_graph)
method_registry.register("fix", "default", fix_issues)
-471
View File
@@ -1,471 +0,0 @@
"""
Quality Metrics Module
This module provides comprehensive quality metrics calculation for the Semantica
framework, enabling quantitative assessment of knowledge graph quality across
multiple dimensions.
Key Features:
- Overall quality score calculation
- Entity quality metrics
- Relationship quality metrics
- Completeness metrics (entity, relationship, property)
- Consistency metrics (logical, temporal, hierarchical)
Main Classes:
- QualityMetrics: Overall quality metrics calculator
- CompletenessMetrics: Completeness metrics calculator
- ConsistencyMetrics: Consistency metrics calculator
Example Usage:
>>> from semantica.kg_qa import QualityMetrics
>>> metrics = QualityMetrics()
>>> score = metrics.calculate_overall_score(knowledge_graph)
>>> entity_score = metrics.calculate_entity_quality(entities)
Author: Semantica Contributors
License: MIT
"""
from dataclasses import dataclass
from typing import Any, Dict, List, Optional
from ..utils.logging import get_logger
from ..utils.progress_tracker import get_progress_tracker
@dataclass
class QualityScore:
"""
Quality score dataclass.
This dataclass represents a comprehensive quality score for a knowledge graph,
containing scores for different quality dimensions and optional metadata.
Attributes:
overall: Overall quality score (0.0 to 1.0)
completeness: Completeness score (0.0 to 1.0)
consistency: Consistency score (0.0 to 1.0)
accuracy: Accuracy score (0.0 to 1.0)
metadata: Additional metadata dictionary (optional)
"""
overall: float
completeness: float
consistency: float
accuracy: float
metadata: Optional[Dict[str, Any]] = None
class QualityMetrics:
"""
Quality metrics calculator.
This class provides overall quality metrics calculation for knowledge graphs,
aggregating entity quality, relationship quality, and consistency into
comprehensive quality scores.
Features:
- Overall quality score calculation
- Entity quality assessment
- Relationship quality assessment
- Weighted aggregation of metrics
Example Usage:
>>> metrics = QualityMetrics()
>>> score = metrics.calculate_overall_score(knowledge_graph)
>>> entity_score = metrics.calculate_entity_quality(entities)
"""
def __init__(self, **kwargs):
"""
Initialize quality metrics calculator.
Sets up the calculator with configuration options.
Args:
**kwargs: Configuration options (currently unused)
"""
self.logger = get_logger("quality_metrics")
self.config = kwargs
# Initialize progress tracker
self.progress_tracker = get_progress_tracker()
self.logger.debug("Quality metrics calculator initialized")
def calculate_overall_score(self, knowledge_graph: Any) -> float:
"""
Calculate overall quality score.
This method calculates an overall quality score by aggregating entity
quality and consistency metrics using weighted averaging (60% completeness,
40% consistency).
Args:
knowledge_graph: Knowledge graph instance (object with entities and
relationships, or dict with "entities" and "relationships")
Returns:
float: Overall quality score between 0.0 and 1.0 (higher is better)
"""
# Track quality calculation
tracking_id = self.progress_tracker.start_tracking(
file=None,
module="kg_qa",
submodule="QualityMetrics",
message="Calculating quality metrics",
)
try:
self.progress_tracker.update_tracking(
tracking_id, message="Calculating entity quality..."
)
completeness = self.calculate_entity_quality(knowledge_graph)
self.progress_tracker.update_tracking(
tracking_id, message="Calculating consistency..."
)
consistency = self._calculate_consistency(knowledge_graph)
self.progress_tracker.update_tracking(
tracking_id, message="Aggregating scores..."
)
# Weighted average
overall = (0.6 * completeness) + (0.4 * consistency)
result = min(1.0, max(0.0, overall))
self.progress_tracker.stop_tracking(
tracking_id,
status="completed",
message=f"Overall quality score: {result:.2f}",
)
return result
except Exception as e:
self.progress_tracker.stop_tracking(
tracking_id, status="failed", message=str(e)
)
raise
def calculate_entity_quality(self, entities: List[Dict[str, Any]]) -> float:
"""
Calculate entity quality score.
This method calculates a quality score for entities based on the presence
of required fields (ID and type). Each entity is scored, and the average
is returned.
Args:
entities: List of entity dictionaries
Returns:
float: Entity quality score between 0.0 and 1.0 (average across all entities)
"""
if not entities:
return 0.0
# Calculate quality based on entity completeness
scores = []
for entity in entities:
# Check required fields
has_id = "id" in entity or "uri" in entity
has_type = "type" in entity
score = 0.0
if has_id:
score += 0.5
if has_type:
score += 0.5
scores.append(score)
return sum(scores) / len(scores) if scores else 0.0
def calculate_relationship_quality(
self, relationships: List[Dict[str, Any]]
) -> float:
"""
Calculate relationship quality score.
This method calculates a quality score for relationships based on the
presence of required fields (source/subject, target/object, type/predicate).
Each relationship is scored, and the average is returned.
Args:
relationships: List of relationship dictionaries
Returns:
float: Relationship quality score between 0.0 and 1.0 (average across all relationships)
"""
if not relationships:
return 0.0
scores = []
for rel in relationships:
# Check required fields
has_source = "source" in rel or "subject" in rel
has_target = "target" in rel or "object" in rel
has_type = "type" in rel or "predicate" in rel
score = 0.0
if has_source:
score += 0.33
if has_target:
score += 0.33
if has_type:
score += 0.34
scores.append(score)
return sum(scores) / len(scores) if scores else 0.0
def _calculate_consistency(self, knowledge_graph: Any) -> float:
"""
Calculate consistency score (simplified).
This is a placeholder method. In practice, this would check for logical
inconsistencies in the knowledge graph.
Args:
knowledge_graph: Knowledge graph instance
Returns:
float: Consistency score between 0.0 and 1.0 (placeholder: 0.8)
"""
# In practice, this would check for logical inconsistencies
return 0.8 # Placeholder
class CompletenessMetrics:
"""
Completeness metrics calculator.
This class provides completeness metrics calculation for knowledge graphs,
assessing whether entities, relationships, and properties meet schema
requirements for completeness.
Features:
- Entity completeness calculation
- Relationship completeness calculation
- Property completeness calculation
- Schema-based validation
Example Usage:
>>> metrics = CompletenessMetrics()
>>> score = metrics.calculate_entity_completeness(entities, schema)
>>> rel_score = metrics.calculate_relationship_completeness(relationships, schema)
"""
def __init__(self, **kwargs):
"""
Initialize completeness metrics calculator.
Sets up the calculator with configuration options.
Args:
**kwargs: Configuration options (currently unused)
"""
self.logger = get_logger("completeness_metrics")
self.config = kwargs
self.logger.debug("Completeness metrics calculator initialized")
def calculate_entity_completeness(
self, entities: List[Dict[str, Any]], schema: Dict[str, Any]
) -> float:
"""
Calculate entity completeness.
This method calculates completeness scores for entities by checking
whether they have all required properties as defined in the schema.
Returns the average completeness score across all entities.
Args:
entities: List of entity dictionaries
schema: Schema definition containing constraints with required_props
for each entity type
Returns:
float: Completeness score between 0.0 and 1.0 (average across entities)
"""
if not entities:
return 0.0
constraints = schema.get("constraints", {})
scores = []
for entity in entities:
entity_type = entity.get("type")
if not entity_type:
scores.append(0.0)
continue
constraint = constraints.get(entity_type, {})
required_props = constraint.get("required_props", [])
if not required_props:
scores.append(1.0)
continue
# Count how many required properties are present
present_props = sum(1 for prop in required_props if prop in entity)
completeness = (
present_props / len(required_props) if required_props else 1.0
)
scores.append(completeness)
return sum(scores) / len(scores) if scores else 0.0
def calculate_property_completeness(
self, properties: Dict[str, Any], schema: Dict[str, Any]
) -> float:
"""
Calculate property completeness.
This method calculates completeness scores for properties by checking
whether entity types have all required properties as defined in the schema.
Args:
properties: Properties dictionary (mapping entity types to property dictionaries)
schema: Schema definition containing constraints with required_props
Returns:
float: Completeness score between 0.0 and 1.0 (average across entity types)
"""
constraints = schema.get("constraints", {})
scores = []
for entity_type, constraint in constraints.items():
required_props = constraint.get("required_props", [])
if entity_type in properties:
entity_props = properties[entity_type]
present_props = sum(
1 for prop in required_props if prop in entity_props
)
completeness = (
present_props / len(required_props) if required_props else 1.0
)
scores.append(completeness)
return sum(scores) / len(scores) if scores else 1.0
def calculate_relationship_completeness(
self, relationships: List[Dict[str, Any]], schema: Dict[str, Any]
) -> float:
"""
Calculate relationship completeness.
This method calculates completeness scores for relationships by checking
whether they have all required fields (source/subject, target/object,
type/predicate). Returns the average completeness score.
Args:
relationships: List of relationship dictionaries
schema: Schema definition (currently unused, reserved for future
relationship-specific constraints)
Returns:
float: Completeness score between 0.0 and 1.0 (average across relationships)
"""
if not relationships:
return 0.0
# Check if relationships have required fields
scores = []
for rel in relationships:
has_source = "source" in rel or "subject" in rel
has_target = "target" in rel or "object" in rel
has_type = "type" in rel or "predicate" in rel
completeness = (has_source + has_target + has_type) / 3.0
scores.append(completeness)
return sum(scores) / len(scores) if scores else 0.0
class ConsistencyMetrics:
"""
Consistency metrics calculator.
This class provides consistency metrics calculation for knowledge graphs,
assessing logical, temporal, and hierarchical consistency.
Features:
- Logical consistency calculation
- Temporal consistency calculation
- Hierarchical consistency calculation
Example Usage:
>>> metrics = ConsistencyMetrics()
>>> logical_score = metrics.calculate_logical_consistency(knowledge_graph)
>>> temporal_score = metrics.calculate_temporal_consistency(knowledge_graph)
"""
def __init__(self, **kwargs):
"""
Initialize consistency metrics calculator.
Sets up the calculator with configuration options.
Args:
**kwargs: Configuration options (currently unused)
"""
self.logger = get_logger("consistency_metrics")
self.config = kwargs
self.logger.debug("Consistency metrics calculator initialized")
def calculate_logical_consistency(self, knowledge_graph: Any) -> float:
"""
Calculate logical consistency.
This method calculates a logical consistency score by checking for
logical contradictions, conflicting relationships, and inconsistent
property values. Currently returns a placeholder value.
Args:
knowledge_graph: Knowledge graph instance
Returns:
float: Logical consistency score between 0.0 and 1.0 (placeholder: 0.9)
"""
# In practice, this would use a reasoner
# For now, return a placeholder
return 0.9
def calculate_temporal_consistency(self, knowledge_graph: Any) -> float:
"""
Calculate temporal consistency.
This method calculates a temporal consistency score by checking for
temporal contradictions, invalid time ranges, and conflicting
temporal relationships. Currently returns a placeholder value.
Args:
knowledge_graph: Knowledge graph instance
Returns:
float: Temporal consistency score between 0.0 and 1.0 (placeholder: 0.85)
"""
# Check for temporal contradictions
return 0.85
def calculate_hierarchical_consistency(self, knowledge_graph: Any) -> float:
"""
Calculate hierarchical consistency.
This method calculates a hierarchical consistency score by checking for
hierarchical contradictions such as circular inheritance, invalid
parent-child relationships, and conflicting hierarchical structures.
Currently returns a placeholder value.
Args:
knowledge_graph: Knowledge graph instance
Returns:
float: Hierarchical consistency score between 0.0 and 1.0 (placeholder: 0.9)
"""
# Check for hierarchical contradictions (e.g., circular inheritance)
return 0.9
-131
View File
@@ -1,131 +0,0 @@
"""
Method Registry Module for KG QA
This module provides a method registry system for registering custom KG QA methods,
enabling extensibility and community contributions to the quality assurance toolkit.
Supported Registration Types:
- Method Registry: Register custom QA methods for:
* "assess": Quality assessment methods
* "report": Report generation methods
* "consistency": Consistency checking methods
* "completeness": Completeness validation methods
* "metrics": Quality metrics calculation methods
* "validate": Validation engine methods
* "fix": Automated fixing methods
Algorithms Used:
- Registry Pattern: Dictionary-based registration and lookup
- Dynamic Registration: Runtime function registration
- Type Checking: Type validation for registered components
- Lookup Algorithms: Hash-based O(1) lookup for methods
- Task-based Organization: Hierarchical organization by task type
Key Features:
- Method registry for custom QA methods
- Task-based method organization (assess, report, consistency, completeness, metrics, validate, fix)
- Dynamic registration and unregistration
- Easy discovery of available methods
- Support for community-contributed extensions
Main Classes:
- MethodRegistry: Registry for custom KG QA methods
Global Instances:
- method_registry: Global method registry instance
Example Usage:
>>> from semantica.kg_qa.registry import method_registry
>>> method_registry.register("assess", "custom_method", custom_assessment_function)
>>> available = method_registry.list_all("assess")
"""
from typing import Any, Callable, Dict, List, Optional
class MethodRegistry:
"""Registry for custom KG QA methods."""
_methods: Dict[str, Dict[str, Callable]] = {
"assess": {},
"report": {},
"consistency": {},
"completeness": {},
"metrics": {},
"validate": {},
"fix": {},
}
@classmethod
def register(cls, task: str, name: str, method_func: Callable):
"""
Register a custom QA method.
Args:
task: Task type ("assess", "report", "consistency", "completeness", "metrics", "validate", "fix")
name: Method name
method_func: Method function
"""
if task not in cls._methods:
cls._methods[task] = {}
cls._methods[task][name] = method_func
@classmethod
def get(cls, task: str, name: str) -> Optional[Callable]:
"""
Get method by task and name.
Args:
task: Task type ("assess", "report", "consistency", "completeness", "metrics", "validate", "fix")
name: Method name
Returns:
Method function or None
"""
return cls._methods.get(task, {}).get(name)
@classmethod
def list_all(cls, task: Optional[str] = None) -> Dict[str, List[str]]:
"""
List all registered methods.
Args:
task: Optional task type to filter by
Returns:
Dictionary mapping task types to method names
"""
if task:
return {task: list(cls._methods.get(task, {}).keys())}
return {t: list(m.keys()) for t, m in cls._methods.items()}
@classmethod
def unregister(cls, task: str, name: str):
"""
Unregister a method.
Args:
task: Task type ("assess", "report", "consistency", "completeness", "metrics", "validate", "fix")
name: Method name
"""
if task in cls._methods and name in cls._methods[task]:
del cls._methods[task][name]
@classmethod
def clear(cls, task: Optional[str] = None):
"""
Clear all registered methods for a task or all tasks.
Args:
task: Optional task type to clear (clears all if None)
"""
if task:
if task in cls._methods:
cls._methods[task].clear()
else:
for task_dict in cls._methods.values():
task_dict.clear()
# Global registry
method_registry = MethodRegistry()
-491
View File
@@ -1,491 +0,0 @@
"""
Quality Reporting Module
This module provides comprehensive quality reporting capabilities for the
Semantica framework, enabling generation of quality reports, issue tracking,
and improvement suggestions.
Key Features:
- Quality report generation
- Issue identification and tracking
- Improvement suggestions generation
- Report export (JSON, YAML, HTML)
- Issue management (add, get, list, resolve)
Main Classes:
- QualityReporter: Quality report generation engine
- IssueTracker: Issue tracking and management
- ImprovementSuggestions: Improvement suggestions generator
Example Usage:
>>> from semantica.kg_qa import QualityReporter
>>> reporter = QualityReporter()
>>> report = reporter.generate_report(knowledge_graph, quality_metrics)
>>> json_report = reporter.export_report(report, format="json")
Author: Semantica Contributors
License: MIT
"""
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any, Dict, List, Optional
from ..utils.logging import get_logger
from ..utils.progress_tracker import get_progress_tracker
@dataclass
class QualityIssue:
"""
Quality issue dataclass.
This dataclass represents a quality issue found in a knowledge graph,
containing issue identification, type, severity, and related entity/relationship
information.
Attributes:
id: Unique issue identifier
type: Issue type (e.g., "completeness", "consistency", "quality")
severity: Issue severity ("low", "medium", "high")
description: Human-readable issue description
entity_id: Related entity ID (optional)
relationship_id: Related relationship ID (optional)
metadata: Additional issue metadata dictionary
"""
id: str
type: str
severity: str
description: str
entity_id: Optional[str] = None
relationship_id: Optional[str] = None
metadata: Dict[str, Any] = field(default_factory=dict)
@dataclass
class QualityReport:
"""
Quality report dataclass.
This dataclass represents a comprehensive quality report for a knowledge graph,
containing quality scores, identified issues, recommendations, and metadata.
Attributes:
timestamp: Report generation timestamp
overall_score: Overall quality score (0.0 to 1.0)
completeness_score: Completeness score (0.0 to 1.0)
consistency_score: Consistency score (0.0 to 1.0)
issues: List of identified quality issues
recommendations: List of improvement recommendations
metadata: Additional report metadata dictionary
"""
timestamp: datetime
overall_score: float
completeness_score: float
consistency_score: float
issues: List[QualityIssue] = field(default_factory=list)
recommendations: List[str] = field(default_factory=list)
metadata: Dict[str, Any] = field(default_factory=dict)
class QualityReporter:
"""
Quality report generation engine.
This class provides quality report generation capabilities, including issue
identification, recommendation generation, and report export in various formats.
Features:
- Quality report generation
- Issue identification
- Recommendation generation
- Report export (JSON, YAML, HTML)
Example Usage:
>>> reporter = QualityReporter()
>>> report = reporter.generate_report(knowledge_graph, quality_metrics)
>>> json_report = reporter.export_report(report, format="json")
"""
def __init__(self, **kwargs):
"""
Initialize quality reporter.
Sets up the reporter with configuration options.
Args:
**kwargs: Configuration options (currently unused)
"""
self.logger = get_logger("quality_reporter")
self.config = kwargs
# Initialize progress tracker
self.progress_tracker = get_progress_tracker()
self.logger.debug("Quality reporter initialized")
def generate_report(
self, knowledge_graph: Any, quality_metrics: Dict[str, float]
) -> QualityReport:
"""
Generate quality report.
This method generates a comprehensive quality report by identifying
issues based on quality metrics and generating recommendations.
Args:
knowledge_graph: Knowledge graph instance
quality_metrics: Quality metrics dictionary containing:
- overall: Overall quality score
- completeness: Completeness score
- consistency: Consistency score
Returns:
QualityReport: Comprehensive quality report with scores, issues,
and recommendations
"""
# Track report generation
tracking_id = self.progress_tracker.start_tracking(
file=None,
module="kg_qa",
submodule="QualityReporter",
message="Generating quality report",
)
try:
self.progress_tracker.update_tracking(
tracking_id, message="Identifying issues..."
)
issues = self._identify_issues(knowledge_graph, quality_metrics)
self.progress_tracker.update_tracking(
tracking_id, message="Generating recommendations..."
)
recommendations = self._generate_recommendations(issues)
report = QualityReport(
timestamp=datetime.now(),
overall_score=quality_metrics.get("overall", 0.0),
completeness_score=quality_metrics.get("completeness", 0.0),
consistency_score=quality_metrics.get("consistency", 0.0),
issues=issues,
recommendations=recommendations,
)
self.progress_tracker.stop_tracking(
tracking_id,
status="completed",
message=f"Generated quality report with {len(issues)} issues",
)
return report
except Exception as e:
self.progress_tracker.stop_tracking(
tracking_id, status="failed", message=str(e)
)
raise
def export_report(self, report: QualityReport, format: str = "json") -> str:
"""
Export report to specified format.
This method exports a quality report to the specified format (JSON, YAML,
or HTML). For unsupported formats, returns string representation.
Args:
report: Quality report to export
format: Export format ("json", "yaml", or "html", default: "json")
Returns:
str: Exported report as string in the specified format
Note:
YAML export requires the `pyyaml` library. If not available, falls
back to string representation.
"""
if format == "json":
import json
return json.dumps(
{
"timestamp": report.timestamp.isoformat(),
"overall_score": report.overall_score,
"completeness_score": report.completeness_score,
"consistency_score": report.consistency_score,
"issues": [
{
"id": issue.id,
"type": issue.type,
"severity": issue.severity,
"description": issue.description,
}
for issue in report.issues
],
"recommendations": report.recommendations,
},
indent=2,
)
elif format == "yaml":
try:
import yaml
return yaml.dump(
{
"timestamp": report.timestamp.isoformat(),
"overall_score": report.overall_score,
"issues": [
{
"id": issue.id,
"type": issue.type,
"description": issue.description,
}
for issue in report.issues
],
}
)
except ImportError:
self.logger.warning(
"PyYAML not available, falling back to string representation"
)
return str(report)
else:
return str(report)
def _identify_issues(
self, knowledge_graph: Any, metrics: Dict[str, float]
) -> List[QualityIssue]:
"""
Identify quality issues.
This method identifies quality issues based on quality metrics,
checking for low scores and generating appropriate issue objects.
Args:
knowledge_graph: Knowledge graph instance
metrics: Quality metrics dictionary
Returns:
list: List of identified quality issues
"""
issues = []
# Check for low scores
if metrics.get("overall", 1.0) < 0.7:
issues.append(
QualityIssue(
id="low_overall_score",
type="quality",
severity="high",
description="Overall quality score is below threshold",
)
)
if metrics.get("completeness", 1.0) < 0.8:
issues.append(
QualityIssue(
id="low_completeness",
type="completeness",
severity="medium",
description="Completeness score is below threshold",
)
)
return issues
def _generate_recommendations(self, issues: List[QualityIssue]) -> List[str]:
"""
Generate improvement recommendations.
This method generates improvement recommendations based on identified
quality issues, providing actionable suggestions for improving
knowledge graph quality.
Args:
issues: List of quality issues
Returns:
list: List of improvement recommendation strings
"""
recommendations = []
for issue in issues:
if issue.type == "completeness":
recommendations.append("Add missing required properties to entities")
elif issue.type == "consistency":
recommendations.append(
"Resolve consistency violations in the knowledge graph"
)
return recommendations
class IssueTracker:
"""
Issue tracking and management engine.
This class provides issue tracking capabilities, enabling storage, retrieval,
filtering, and resolution of quality issues.
Features:
- Issue storage and retrieval
- Issue filtering by severity
- Issue resolution tracking
Example Usage:
>>> tracker = IssueTracker()
>>> tracker.add_issue(issue)
>>> issues = tracker.list_issues(severity="high")
>>> tracker.resolve_issue(issue_id)
"""
def __init__(self, **kwargs):
"""
Initialize issue tracker.
Sets up the tracker with configuration and initializes issue storage.
Args:
**kwargs: Configuration options (currently unused)
"""
self.logger = get_logger("issue_tracker")
self.config = kwargs
self.issues: Dict[str, QualityIssue] = {}
# Initialize progress tracker
self.progress_tracker = get_progress_tracker()
self.logger.debug("Issue tracker initialized")
def add_issue(self, issue: QualityIssue) -> None:
"""
Add an issue to the tracker.
This method adds a quality issue to the tracker's issue dictionary,
using the issue ID as the key.
Args:
issue: Quality issue to add
"""
self.issues[issue.id] = issue
def get_issue(self, issue_id: str) -> Optional[QualityIssue]:
"""
Get issue by ID.
This method retrieves a quality issue from the tracker by its ID.
Args:
issue_id: Issue identifier
Returns:
QualityIssue: The issue if found, None otherwise
"""
return self.issues.get(issue_id)
def list_issues(self, severity: Optional[str] = None) -> List[QualityIssue]:
"""
List issues, optionally filtered by severity.
This method returns all tracked issues, optionally filtered by severity
level ("low", "medium", "high").
Args:
severity: Optional severity filter ("low", "medium", "high")
Returns:
list: List of quality issues (filtered by severity if provided)
"""
issues = list(self.issues.values())
if severity:
issues = [i for i in issues if i.severity == severity]
return issues
def resolve_issue(self, issue_id: str) -> bool:
"""
Mark issue as resolved.
This method removes an issue from the tracker, effectively marking
it as resolved.
Args:
issue_id: Issue identifier to resolve
Returns:
bool: True if issue was found and resolved, False otherwise
"""
if issue_id in self.issues:
del self.issues[issue_id]
return True
return False
class ImprovementSuggestions:
"""
Improvement suggestions generator.
This class provides improvement suggestions generation capabilities,
analyzing quality reports and generating actionable recommendations
for improving knowledge graph quality.
Features:
- Issue-based suggestions
- Score-based suggestions
- Actionable recommendations
Example Usage:
>>> generator = ImprovementSuggestions()
>>> suggestions = generator.generate_suggestions(quality_report)
"""
def __init__(self, **kwargs):
"""
Initialize improvement suggestions generator.
Sets up the generator with configuration options.
Args:
**kwargs: Configuration options (currently unused)
"""
self.logger = get_logger("improvement_suggestions")
self.config = kwargs
# Initialize progress tracker
self.progress_tracker = get_progress_tracker()
self.logger.debug("Improvement suggestions generator initialized")
def generate_suggestions(self, quality_report: QualityReport) -> List[str]:
"""
Generate improvement suggestions.
This method generates improvement suggestions based on the quality report,
analyzing issues and scores to provide actionable recommendations.
Args:
quality_report: Quality report containing scores and issues
Returns:
list: List of improvement suggestion strings
"""
suggestions = []
# Based on issues
for issue in quality_report.issues:
if issue.type == "completeness":
suggestions.append(f"Improve completeness for {issue.description}")
elif issue.type == "consistency":
suggestions.append(f"Resolve consistency issue: {issue.description}")
# Based on scores
if quality_report.overall_score < 0.7:
suggestions.append("Overall quality needs improvement")
if quality_report.completeness_score < 0.8:
suggestions.append("Add missing required properties")
return suggestions
-344
View File
@@ -1,344 +0,0 @@
"""
Validation Engine Module
This module provides comprehensive validation capabilities for the Semantica
framework, enabling rule-based and constraint-based validation of knowledge graphs.
Key Features:
- Rule-based validation
- Constraint-based validation
- Custom validation rules
- Validation result reporting
Main Classes:
- ValidationEngine: Main validation engine
- RuleValidator: Rule-based validation
- ConstraintValidator: Constraint-based validation
Example Usage:
>>> from semantica.kg_qa import ValidationEngine
>>> engine = ValidationEngine()
>>> result = engine.validate(knowledge_graph, rules=[rule1, rule2])
>>> engine.add_rule(custom_rule)
Author: Semantica Contributors
License: MIT
"""
from dataclasses import dataclass, field
from typing import Any, Callable, Dict, List, Optional
from ..utils.exceptions import ValidationError
from ..utils.logging import get_logger
from ..utils.progress_tracker import get_progress_tracker
@dataclass
class ValidationResult:
"""
Validation result dataclass.
This dataclass represents the result of a validation operation, containing
validation status, errors, warnings, and optional metadata.
Attributes:
valid: Whether the validation passed (True if no errors)
errors: List of error messages (critical validation failures)
warnings: List of warning messages (non-critical issues)
metadata: Additional validation metadata dictionary
"""
valid: bool
errors: List[str] = field(default_factory=list)
warnings: List[str] = field(default_factory=list)
metadata: Dict[str, Any] = field(default_factory=dict)
class ValidationEngine:
"""
Validation engine.
This class provides rule-based validation capabilities for knowledge graphs,
enabling custom validation rules and constraint checking.
Features:
- Custom validation rules
- Rule management (add, remove)
- Validation result reporting
- Error and warning collection
Example Usage:
>>> engine = ValidationEngine()
>>> engine.add_rule(custom_validation_rule)
>>> result = engine.validate(knowledge_graph)
>>> if not result.valid:
... print(f"Errors: {result.errors}")
"""
def __init__(self, **kwargs):
"""
Initialize validation engine.
Sets up the engine with configuration and initializes rule storage.
Args:
**kwargs: Configuration options (currently unused)
"""
self.logger = get_logger("validation_engine")
self.config = kwargs
self.rules: List[Callable] = []
# Initialize progress tracker
self.progress_tracker = get_progress_tracker()
self.logger.debug("Validation engine initialized")
def validate(
self, knowledge_graph: Any, rules: Optional[List[Callable]] = None
) -> ValidationResult:
"""
Validate knowledge graph.
This method validates a knowledge graph against a list of validation
rules. Rules can be provided as arguments or use the engine's stored
rules. Each rule should return a dict with "error" and/or "warning"
keys, or raise an exception.
Args:
knowledge_graph: Knowledge graph instance to validate
rules: Optional list of validation rule functions (if None, uses
stored rules). Each rule should accept the knowledge graph
as argument and return a dict or raise an exception.
Returns:
ValidationResult: Validation result containing:
- valid: True if no errors, False otherwise
- errors: List of error messages
- warnings: List of warning messages
- metadata: Additional validation metadata
"""
# Track validation
tracking_id = self.progress_tracker.start_tracking(
file=None,
module="kg_qa",
submodule="ValidationEngine",
message="Validating graph",
)
try:
rules_to_use = rules or self.rules
errors = []
warnings = []
self.progress_tracker.update_tracking(
tracking_id, message=f"Validating with {len(rules_to_use)} rule(s)..."
)
for rule in rules_to_use:
try:
result = rule(knowledge_graph)
if isinstance(result, dict):
if result.get("error"):
errors.append(result["error"])
if result.get("warning"):
warnings.append(result["warning"])
except Exception as e:
self.logger.error(f"Validation rule error: {e}")
errors.append(f"Validation rule failed: {e}")
result = ValidationResult(
valid=len(errors) == 0, errors=errors, warnings=warnings
)
self.progress_tracker.stop_tracking(
tracking_id,
status="completed",
message=f"Validation complete: {len(errors)} errors, {len(warnings)} warnings",
)
return result
except Exception as e:
self.progress_tracker.stop_tracking(
tracking_id, status="failed", message=str(e)
)
raise
def add_rule(self, rule: Callable) -> None:
"""
Add validation rule.
This method adds a validation rule function to the engine's rule list.
The rule will be used in subsequent validate() calls.
Args:
rule: Validation rule function (should accept knowledge graph and
return dict with "error"/"warning" keys or raise exception)
"""
self.rules.append(rule)
def remove_rule(self, rule: Callable) -> None:
"""
Remove validation rule.
This method removes a validation rule function from the engine's rule list.
Args:
rule: Validation rule function to remove
"""
if rule in self.rules:
self.rules.remove(rule)
class RuleValidator:
"""
Rule-based validation engine.
This class provides rule-based validation capabilities, enabling validation
against specific rule strings or identifiers.
Features:
- Single rule validation
- Multiple rule validation
- Rule parsing and execution (planned)
Example Usage:
>>> validator = RuleValidator()
>>> result = validator.validate_rule(knowledge_graph, "rule_name")
>>> results = validator.validate_all_rules(knowledge_graph, ["rule1", "rule2"])
"""
def __init__(self, **kwargs):
"""
Initialize rule validator.
Sets up the validator with configuration options.
Args:
**kwargs: Configuration options (currently unused)
"""
self.logger = get_logger("rule_validator")
self.config = kwargs
self.logger.debug("Rule validator initialized")
def validate_rule(self, knowledge_graph: Any, rule: str) -> ValidationResult:
"""
Validate against a specific rule.
This method validates a knowledge graph against a specific rule string
or identifier. Currently returns a placeholder result. In practice,
this would parse and execute the rule.
Args:
knowledge_graph: Knowledge graph instance
rule: Rule string or identifier
Returns:
ValidationResult: Validation result (currently placeholder)
"""
# In practice, this would parse and execute the rule
# For now, return a placeholder
return ValidationResult(valid=True)
def validate_all_rules(
self, knowledge_graph: Any, rules: List[str]
) -> Dict[str, ValidationResult]:
"""
Validate against multiple rules.
This method validates a knowledge graph against multiple rules and
returns a dictionary mapping each rule name to its validation result.
Args:
knowledge_graph: Knowledge graph instance
rules: List of rule strings or identifiers
Returns:
dict: Dictionary mapping rule names to ValidationResult objects
"""
results = {}
for rule in rules:
results[rule] = self.validate_rule(knowledge_graph, rule)
return results
class ConstraintValidator:
"""
Constraint-based validation engine.
This class provides constraint-based validation capabilities, enabling
validation against schema constraints such as required properties, domain
and range constraints for relationships.
Features:
- Entity constraint validation
- Relationship constraint validation
- Domain and range validation
Example Usage:
>>> validator = ConstraintValidator()
>>> result = validator.validate_constraints(knowledge_graph, constraints)
"""
def __init__(self, **kwargs):
"""
Initialize constraint validator.
Sets up the validator with configuration options.
Args:
**kwargs: Configuration options (currently unused)
"""
self.logger = get_logger("constraint_validator")
self.config = kwargs
self.logger.debug("Constraint validator initialized")
def validate_constraints(
self, knowledge_graph: Any, constraints: Dict[str, Any]
) -> ValidationResult:
"""
Validate against constraints.
This method validates a knowledge graph against schema constraints,
checking entity constraints (required properties) and relationship
constraints (domain and range).
Args:
knowledge_graph: Knowledge graph instance
constraints: Constraints dictionary containing:
- entities: Dictionary mapping entity types to constraint dicts
with "required_props" list
- relationships: Dictionary mapping relationship types to
constraint dicts with "domain" and "range"
Returns:
ValidationResult: Validation result with errors and warnings
"""
errors = []
warnings = []
# Validate entity constraints
entity_constraints = constraints.get("entities", {})
for entity_type, constraint in entity_constraints.items():
required_props = constraint.get("required_props", [])
# Check if entities of this type have required properties
# This is simplified - in practice would query the graph
if required_props:
warnings.append(
f"Entity type {entity_type} requires properties: {required_props}"
)
# Validate relationship constraints
rel_constraints = constraints.get("relationships", {})
for rel_type, constraint in rel_constraints.items():
domain = constraint.get("domain")
range_val = constraint.get("range")
if domain and range_val:
# Check domain and range constraints
pass # Would validate in practice
return ValidationResult(
valid=len(errors) == 0, errors=errors, warnings=warnings
)