mirror of
https://github.com/semantica-agi/semantica.git
synced 2026-09-04 04:01:07 +00:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
dd0a9d8c90 | ||
|
|
044fa2742e | ||
|
|
4c6eab632f | ||
|
|
103ab04970 | ||
|
|
88a57b39f9 | ||
|
|
f45499b5a7 | ||
|
|
c04adcd1a9 | ||
|
|
a85cf913a5 | ||
|
|
75bcb64681 | ||
|
|
8778e6a837 | ||
|
|
01352d5fd5 | ||
|
|
6a55b8c3ee | ||
|
|
471cbe7711 |
@@ -195,7 +195,7 @@ apt29_intel = context.retrieve(
|
||||
```python
|
||||
from semantica.llms import LiteLLM
|
||||
|
||||
llm = LiteLLM(model="anthropic/claude-sonnet-4-20250514")
|
||||
llm = LiteLLM(model="anthropic/claude-sonnet-5")
|
||||
|
||||
result = context.query_with_reasoning(
|
||||
"What are APT29's known TTPs against healthcare infrastructure, "
|
||||
@@ -281,7 +281,7 @@ context.store(
|
||||
link_entities=True,
|
||||
)
|
||||
|
||||
llm = LiteLLM(model="anthropic/claude-sonnet-4-20250514")
|
||||
llm = LiteLLM(model="anthropic/claude-sonnet-5")
|
||||
result = context.query_with_reasoning(
|
||||
"Trace the C2 infrastructure chain for APT29 operations targeting "
|
||||
"ITAR-controlled contractors in 2025. Include IP ranges, ASNs, and TTPs.",
|
||||
@@ -351,7 +351,7 @@ Parent: wmiprvse.exe
|
||||
Sigma match: T1053.005 Scheduled Task/Job
|
||||
"""
|
||||
|
||||
llm = LiteLLM(model="anthropic/claude-sonnet-4-20250514")
|
||||
llm = LiteLLM(model="anthropic/claude-sonnet-5")
|
||||
triage = soc_context.query_with_reasoning(
|
||||
"Triage this SIEM alert and identify the correct response runbook:\n{}".format(alert_text),
|
||||
llm_provider=llm,
|
||||
@@ -425,7 +425,7 @@ Patient: 68F, AF, CKD stage 3b (eGFR 32). On warfarin (INR target 2.0–3.0).
|
||||
Presenting for elective hip replacement. Concurrent: amiodarone 200mg, atorvastatin 40mg.
|
||||
"""
|
||||
|
||||
llm = LiteLLM(model="anthropic/claude-sonnet-4-20250514")
|
||||
llm = LiteLLM(model="anthropic/claude-sonnet-5")
|
||||
answer = clinical_context.query_with_reasoning(
|
||||
"What is the evidence-based warfarin bridging protocol for this patient "
|
||||
"given CKD and amiodarone interaction risk?\n\n{}".format(patient_context),
|
||||
@@ -495,7 +495,7 @@ compliance_context.store(
|
||||
extract_relationships=True,
|
||||
)
|
||||
|
||||
llm = LiteLLM(model="anthropic/claude-sonnet-4-20250514")
|
||||
llm = LiteLLM(model="anthropic/claude-sonnet-5")
|
||||
answer = compliance_context.query_with_reasoning(
|
||||
"Under Basel III CRE20, what are the RWA calculation requirements for "
|
||||
"commercial real estate exposures with LTV > 80%? "
|
||||
|
||||
@@ -275,20 +275,20 @@ print(data)
|
||||
|
||||
**LiteLLM** is a universal adapter that provides a single interface to over 100 different LLM providers, including Anthropic Claude, Azure OpenAI, AWS Bedrock, Google Vertex AI, and local Ollama instances. It acts as a translation layer, converting your unified API calls into provider-specific requests, enabling easy switching between providers without code changes.
|
||||
|
||||
`LiteLLM` is the Swiss Army knife. It wraps the `litellm` library, which speaks to every major provider using a unified completion API. The model string encodes both provider and model name: `"anthropic/claude-sonnet-4-20250514"`, `"azure/gpt-4o"`, `"bedrock/anthropic.claude-3-5-sonnet-20241022-v2:0"`, `"ollama/llama3.2"`. Change the string, change the provider — no other code changes needed.
|
||||
`LiteLLM` is the Swiss Army knife. It wraps the `litellm` library, which speaks to every major provider using a unified completion API. The model string encodes both provider and model name: `"anthropic/claude-sonnet-5"`, `"azure/gpt-4o"`, `"bedrock/anthropic.claude-sonnet-4-5-20250929-v1:0"`, `"ollama/llama3.2"`. Change the string, change the provider — no other code changes needed.
|
||||
|
||||
```python
|
||||
from semantica.llms import LiteLLM
|
||||
|
||||
# Anthropic Claude — highest accuracy for complex reasoning
|
||||
llm = LiteLLM(model="anthropic/claude-sonnet-4-20250514")
|
||||
llm = LiteLLM(model="anthropic/claude-sonnet-5")
|
||||
# Reads ANTHROPIC_API_KEY from environment
|
||||
|
||||
# Azure OpenAI — compliance and data-residency requirements
|
||||
llm = LiteLLM(model="azure/gpt-4o", api_key="YOUR_AZURE_KEY")
|
||||
|
||||
# AWS Bedrock — existing cloud agreement, no new vendor
|
||||
llm = LiteLLM(model="bedrock/anthropic.claude-3-5-sonnet-20241022-v2:0")
|
||||
llm = LiteLLM(model="bedrock/anthropic.claude-sonnet-4-5-20250929-v1:0")
|
||||
|
||||
# Google Vertex AI
|
||||
llm = LiteLLM(model="vertex_ai/gemini-1.5-pro")
|
||||
@@ -306,7 +306,7 @@ The environment-variable convention for each provider: `ANTHROPIC_API_KEY`, `AZU
|
||||
import os
|
||||
|
||||
PROVIDER_MAP = {
|
||||
"prod": "anthropic/claude-sonnet-4-20250514",
|
||||
"prod": "anthropic/claude-sonnet-5",
|
||||
"staging": "openai/gpt-4o-mini",
|
||||
"local": "ollama/llama3.2",
|
||||
"azure": "azure/gpt-4o",
|
||||
@@ -378,7 +378,7 @@ print("FAST: {} (conf={:.0%})".format(fast_result["response"], fast_result["con
|
||||
|
||||
# Tier 2: deep answer with Claude if confidence is below threshold
|
||||
if fast_result["confidence"] < 0.85:
|
||||
deep_llm = LiteLLM(model="anthropic/claude-sonnet-4-20250514")
|
||||
deep_llm = LiteLLM(model="anthropic/claude-sonnet-5")
|
||||
deep_result = context.query_with_reasoning(
|
||||
query, llm_provider=deep_llm, max_results=15, max_hops=3
|
||||
)
|
||||
@@ -574,7 +574,7 @@ print("TRIAGE: {} (conf={:.0%})".format(triage["response"], triage["confidence"]
|
||||
|
||||
# Tier 2: escalate to Claude for deep analysis if Tier 1 is uncertain
|
||||
if triage["confidence"] < 0.88:
|
||||
deep_llm = LiteLLM(model="anthropic/claude-sonnet-4-20250514")
|
||||
deep_llm = LiteLLM(model="anthropic/claude-sonnet-5")
|
||||
deep = context.query_with_reasoning(
|
||||
"Full MITRE ATT&CK analysis of this alert: identify the attack chain, "
|
||||
"blast radius, affected systems, and recommended containment steps.",
|
||||
@@ -630,7 +630,7 @@ for d in drugs:
|
||||
# trastuzumab (conf=0.98), pertuzumab (conf=0.97), docetaxel (conf=0.96)
|
||||
|
||||
# Report synthesis with Claude — switch to azure/gpt-4o for HIPAA by changing one string
|
||||
report_llm = LiteLLM(model="anthropic/claude-sonnet-4-20250514")
|
||||
report_llm = LiteLLM(model="anthropic/claude-sonnet-5")
|
||||
# For HIPAA-constrained Azure deployment:
|
||||
# report_llm = LiteLLM(model="azure/gpt-4o", api_key="YOUR_AZURE_KEY")
|
||||
|
||||
@@ -682,7 +682,7 @@ question = (
|
||||
|
||||
# Two-provider consensus — same query, same graph, different LLMs
|
||||
gpt4o = OpenAI(model="gpt-4o", api_key="YOUR_OAI_KEY")
|
||||
claude = LiteLLM(model="anthropic/claude-sonnet-4-20250514")
|
||||
claude = LiteLLM(model="anthropic/claude-sonnet-5")
|
||||
|
||||
answer_a = context.query_with_reasoning(question, llm_provider=gpt4o, max_results=10)
|
||||
answer_b = context.query_with_reasoning(question, llm_provider=claude, max_results=10)
|
||||
|
||||
@@ -197,7 +197,7 @@ reasoning_agent.load("./pipeline/enriched_intel/")
|
||||
# All memories, graph nodes, and vector embeddings from both ingestion agents are now available.
|
||||
|
||||
# Use a high-capability model for the synthesis step
|
||||
llm = LiteLLM(model="anthropic/claude-sonnet-4-20250514")
|
||||
llm = LiteLLM(model="anthropic/claude-sonnet-5")
|
||||
|
||||
synthesis = reasoning_agent.query_with_reasoning(
|
||||
"Summarize the APT29 exploitation of CVE-2024-3400: affected products, "
|
||||
@@ -428,7 +428,7 @@ tier1.store(
|
||||
|
||||
# --- Tier 2: deep investigation when Tier 1 confidence is low ---
|
||||
if triage["confidence"] < 0.90:
|
||||
deep_llm = LiteLLM(model="anthropic/claude-sonnet-4-20250514")
|
||||
deep_llm = LiteLLM(model="anthropic/claude-sonnet-5")
|
||||
|
||||
investigation = tier2.query_with_reasoning(
|
||||
"Full MITRE ATT&CK analysis of incident {}. "
|
||||
@@ -533,7 +533,7 @@ t1.start(); t2.start()
|
||||
t1.join(); t2.join()
|
||||
|
||||
# Chief agent synthesizes across literature and experimental data
|
||||
llm = LiteLLM(model="anthropic/claude-sonnet-4-20250514")
|
||||
llm = LiteLLM(model="anthropic/claude-sonnet-5")
|
||||
|
||||
synthesis = chief.query_with_reasoning(
|
||||
"Identify the top two candidate compounds for KRAS G12C NSCLC that show "
|
||||
@@ -576,7 +576,7 @@ credit_officer = make_desk_agent()
|
||||
committee_chair = make_desk_agent()
|
||||
|
||||
app_id = "LOAN-2025-88421"
|
||||
llm = LiteLLM(model="anthropic/claude-sonnet-4-20250514")
|
||||
llm = LiteLLM(model="anthropic/claude-sonnet-5")
|
||||
|
||||
# --- Risk Desk: PD/LGD/EL analysis ---
|
||||
risk_desk.store(
|
||||
|
||||
@@ -477,7 +477,7 @@ regs = [
|
||||
]
|
||||
|
||||
# Use an LLM to extract the conceptual model from regulatory prose
|
||||
llm_gen = LLMOntologyGenerator(provider="anthropic", model="claude-sonnet-4-20250514")
|
||||
llm_gen = LLMOntologyGenerator(provider="anthropic", model="claude-sonnet-5")
|
||||
ontology = llm_gen.generate_ontology_from_text(
|
||||
"\n\n".join(r.text[:8000] for r in regs) # token-safe excerpt per document
|
||||
)
|
||||
|
||||
@@ -100,14 +100,15 @@ ner = NamedEntityRecognizer(
|
||||
methods=["llm", "ml", "pattern"],
|
||||
confidence_threshold=0.75,
|
||||
provider="anthropic",
|
||||
llm_model="claude-sonnet-4-6",
|
||||
llm_model="claude-sonnet-5",
|
||||
)
|
||||
entities = ner.extract_entities(report)
|
||||
|
||||
for e in entities:
|
||||
print("[{:>5.2f}] {:15s} {}".format(e.confidence, e.label, e.text))
|
||||
|
||||
# Expected output (abbreviated):
|
||||
# Illustrative output — exact labels and scores depend on the method and model.
|
||||
# Abbreviated:
|
||||
# [ 0.94] THREAT_ACTOR GAMMA-7
|
||||
# [ 0.91] THREAT_ACTOR DELTA-3
|
||||
# [ 0.97] MALWARE HAMMERTOSS
|
||||
@@ -262,16 +263,18 @@ from semantica.semantic_extract import TripletExtractor
|
||||
tri = TripletExtractor(
|
||||
method="llm",
|
||||
provider="anthropic",
|
||||
llm_model="claude-sonnet-4-6",
|
||||
llm_model="claude-sonnet-5",
|
||||
include_temporal=True, # attach time context to triplets when available
|
||||
include_provenance=True, # embed source document reference in each triplet
|
||||
validate=False, # return raw triplets; validate explicitly below
|
||||
)
|
||||
|
||||
# Feed in the entities and relations you already extracted — the extractor
|
||||
# uses them to constrain and validate what it produces
|
||||
# uses them to constrain what it produces
|
||||
triplets = tri.extract_triplets(report, entities, relations)
|
||||
|
||||
# Filter malformed triplets before serialisation
|
||||
# (extract_triplets validates automatically unless validate=False, as above)
|
||||
valid = tri.validate_triplets(triplets)
|
||||
print("Valid: {}/{}".format(len(valid), len(triplets)))
|
||||
|
||||
@@ -320,7 +323,7 @@ def ingest_intel_report(
|
||||
methods=[method, "pattern"],
|
||||
confidence_threshold=0.70,
|
||||
provider="anthropic",
|
||||
llm_model="claude-sonnet-4-6",
|
||||
llm_model="claude-sonnet-5",
|
||||
)
|
||||
entities = ner.extract_entities(text)
|
||||
classified = ner.classify_entities(entities)
|
||||
@@ -335,7 +338,7 @@ def ingest_intel_report(
|
||||
relation_types=["deployed", "targets", "exploits", "operates_from", "provided_to"],
|
||||
confidence_threshold=0.65,
|
||||
provider="anthropic",
|
||||
llm_model="claude-sonnet-4-6",
|
||||
llm_model="claude-sonnet-5",
|
||||
)
|
||||
relations = rel.extract_relations(text, entities)
|
||||
|
||||
@@ -347,9 +350,10 @@ def ingest_intel_report(
|
||||
tri = TripletExtractor(
|
||||
method=method,
|
||||
provider="anthropic",
|
||||
llm_model="claude-sonnet-4-6",
|
||||
llm_model="claude-sonnet-5",
|
||||
include_temporal=True,
|
||||
include_provenance=True,
|
||||
validate=False, # keep raw triplets so the summary can report rejections
|
||||
)
|
||||
triplets = tri.extract_triplets(text, entities, relations)
|
||||
valid = tri.validate_triplets(triplets)
|
||||
@@ -377,6 +381,7 @@ def ingest_intel_report(
|
||||
"coref_chains": len(chains),
|
||||
"relations": len(relations),
|
||||
"events": len(events),
|
||||
"triplets_total": len(triplets),
|
||||
"triplets_valid": len(valid),
|
||||
"graph_nodes": graph_stats.get("graph_nodes", 0),
|
||||
"graph_edges": graph_stats.get("graph_edges", 0),
|
||||
@@ -402,7 +407,7 @@ for text, doc_id in reports:
|
||||
summary["relations"],
|
||||
summary["events"],
|
||||
summary["triplets_valid"],
|
||||
len(summary["rdf_turtle"]),
|
||||
summary["triplets_total"],
|
||||
))
|
||||
```
|
||||
|
||||
@@ -421,7 +426,7 @@ ner = NamedEntityRecognizer(
|
||||
methods=["llm", "pattern"],
|
||||
confidence_threshold=0.75,
|
||||
provider="anthropic",
|
||||
llm_model="claude-sonnet-4-6",
|
||||
llm_model="claude-sonnet-5",
|
||||
)
|
||||
entities = ner.extract_entities(fintel_text)
|
||||
grouped = ner.classify_entities(entities)
|
||||
@@ -438,14 +443,14 @@ rel = RelationExtractor(
|
||||
relation_types=["operates_from", "deployed", "targets", "exploits"],
|
||||
confidence_threshold=0.70,
|
||||
provider="anthropic",
|
||||
llm_model="claude-sonnet-4-6",
|
||||
llm_model="claude-sonnet-5",
|
||||
)
|
||||
relations = rel.extract_relations(fintel_text, entities)
|
||||
|
||||
tri = TripletExtractor(
|
||||
method="llm",
|
||||
provider="anthropic",
|
||||
llm_model="claude-sonnet-4-6",
|
||||
llm_model="claude-sonnet-5",
|
||||
include_temporal=True,
|
||||
include_provenance=True,
|
||||
)
|
||||
@@ -544,14 +549,14 @@ rel = RelationExtractor(
|
||||
relation_types=["treats", "causes_adverse_event", "has_efficacy", "evaluated_in"],
|
||||
confidence_threshold=0.65,
|
||||
provider="anthropic",
|
||||
llm_model="claude-sonnet-4-6",
|
||||
llm_model="claude-sonnet-5",
|
||||
)
|
||||
relations = rel.extract_relations(paper, entities)
|
||||
|
||||
tri = TripletExtractor(
|
||||
method="llm",
|
||||
provider="anthropic",
|
||||
llm_model="claude-sonnet-4-6",
|
||||
llm_model="claude-sonnet-5",
|
||||
triplet_types=["treats", "has_efficacy", "causes_adverse_event"],
|
||||
include_temporal=True,
|
||||
include_provenance=True,
|
||||
@@ -595,7 +600,7 @@ ner = NamedEntityRecognizer(
|
||||
methods=["llm", "ml", "pattern"],
|
||||
confidence_threshold=0.70,
|
||||
provider="anthropic",
|
||||
llm_model="claude-sonnet-4-6",
|
||||
llm_model="claude-sonnet-5",
|
||||
)
|
||||
entities = ner.extract_entities(credit_memo)
|
||||
grouped = ner.classify_entities(entities)
|
||||
@@ -612,14 +617,14 @@ rel = RelationExtractor(
|
||||
relation_types=["guaranteed_by", "secured_by", "classified_as", "exposed_to"],
|
||||
confidence_threshold=0.65,
|
||||
provider="anthropic",
|
||||
llm_model="claude-sonnet-4-6",
|
||||
llm_model="claude-sonnet-5",
|
||||
)
|
||||
relations = rel.extract_relations(credit_memo, entities)
|
||||
|
||||
tri = TripletExtractor(
|
||||
method="llm",
|
||||
provider="anthropic",
|
||||
llm_model="claude-sonnet-4-6",
|
||||
llm_model="claude-sonnet-5",
|
||||
include_temporal=True,
|
||||
include_provenance=True,
|
||||
)
|
||||
|
||||
@@ -129,7 +129,7 @@ from semantica.llms import Groq, OpenAI, LiteLLM, HuggingFaceLLM
|
||||
from semantica.llms import LiteLLM
|
||||
|
||||
llm = LiteLLM(
|
||||
model="anthropic/claude-sonnet-4-20250514",
|
||||
model="anthropic/claude-sonnet-5",
|
||||
api_key=os.getenv("ANTHROPIC_API_KEY"),
|
||||
temperature=0.0,
|
||||
)
|
||||
@@ -198,7 +198,7 @@ llm = Groq(api_key=os.getenv("GROQ_API_KEY"), model="llama-3.1-8b-instant")
|
||||
# Method 3: Multiple providers via LiteLLM
|
||||
providers = {
|
||||
"fast": LiteLLM(model="groq/llama-3.1-8b-instant", api_key=os.getenv("GROQ_API_KEY")),
|
||||
"smart": LiteLLM(model="anthropic/claude-sonnet-4-20250514", api_key=os.getenv("ANTHROPIC_API_KEY"))
|
||||
"smart": LiteLLM(model="anthropic/claude-sonnet-5", api_key=os.getenv("ANTHROPIC_API_KEY"))
|
||||
}
|
||||
```
|
||||
|
||||
@@ -252,7 +252,7 @@ from semantica.llms import LiteLLM
|
||||
# pip install "semantica[llm-litellm]"
|
||||
|
||||
# Anthropic Claude
|
||||
llm = LiteLLM(model="anthropic/claude-opus-4-5", api_key=os.getenv("ANTHROPIC_API_KEY"))
|
||||
llm = LiteLLM(model="anthropic/claude-opus-4-7", api_key=os.getenv("ANTHROPIC_API_KEY"))
|
||||
|
||||
# Google Gemini
|
||||
llm = LiteLLM(model="gemini/gemini-1.5-pro", api_key=os.getenv("GOOGLE_API_KEY"))
|
||||
@@ -267,7 +267,7 @@ llm = LiteLLM(model="deepseek/deepseek-chat", api_key=os.getenv("DEEP
|
||||
llm = LiteLLM(model="azure/gpt-4o", api_key=os.getenv("AZURE_API_KEY"))
|
||||
|
||||
# AWS Bedrock
|
||||
llm = LiteLLM(model="bedrock/anthropic.claude-3-5-sonnet-20241022-v2:0")
|
||||
llm = LiteLLM(model="bedrock/anthropic.claude-sonnet-4-5-20250929-v1:0")
|
||||
|
||||
# Novita AI
|
||||
llm = LiteLLM(model="novita/deepseek/deepseek-v3.2", api_key=os.getenv("NOVITA_API_KEY"))
|
||||
@@ -297,12 +297,12 @@ from semantica.llms import LiteLLM
|
||||
|
||||
# Pattern: LiteLLM(model="<provider>/<model-name>")
|
||||
providers = {
|
||||
"Anthropic": LiteLLM(model="anthropic/claude-opus-4-5", api_key=os.getenv("ANTHROPIC_API_KEY")),
|
||||
"Anthropic": LiteLLM(model="anthropic/claude-opus-4-7", api_key=os.getenv("ANTHROPIC_API_KEY")),
|
||||
"Gemini": LiteLLM(model="gemini/gemini-1.5-pro", api_key=os.getenv("GOOGLE_API_KEY")),
|
||||
"Ollama": LiteLLM(model="ollama/llama3.2:3b", api_base="http://localhost:11434"),
|
||||
"DeepSeek": LiteLLM(model="deepseek/deepseek-chat", api_key=os.getenv("DEEPSEEK_API_KEY")),
|
||||
"Azure": LiteLLM(model="azure/gpt-4o", api_key=os.getenv("AZURE_API_KEY")),
|
||||
"Bedrock": LiteLLM(model="bedrock/anthropic.claude-3-5-sonnet-20241022-v2:0"),
|
||||
"Bedrock": LiteLLM(model="bedrock/anthropic.claude-sonnet-4-5-20250929-v1:0"),
|
||||
"Cohere": LiteLLM(model="cohere/command-r-plus", api_key=os.getenv("COHERE_API_KEY")),
|
||||
"Novita AI": LiteLLM(model="novita/deepseek/deepseek-v3.2", api_key=os.getenv("NOVITA_API_KEY")),
|
||||
}
|
||||
@@ -416,7 +416,7 @@ for text in texts:
|
||||
| :---------- | :--------------------------- | :----------- |
|
||||
| **Entity Extraction** | `Groq("llama-3.3-70b-versatile")` | Fast, good accuracy for structured tasks |
|
||||
| **Relation Extraction** | `OpenAI("gpt-4o")` | Best at complex relationship reasoning |
|
||||
| **Complex Analysis** | `LiteLLM("anthropic/claude-sonnet-4-20250514")` | Highest reasoning capability |
|
||||
| **Complex Analysis** | `LiteLLM("anthropic/claude-sonnet-5")` | Highest reasoning capability |
|
||||
| **High Volume/Cost** | `LiteLLM("deepseek/deepseek-chat")` | Lowest cost per token |
|
||||
|
||||
### Error Handling
|
||||
|
||||
@@ -22,6 +22,7 @@ icon: "sitemap"
|
||||
| `LLMOntologyGenerator` | LLM-powered ontology generation for complex domains |
|
||||
| `SHACLGenerator` | Generate SHACL shapes from an ontology or KG schema |
|
||||
| `OntologyValidator` | Validate any graph against SHACL shapes: returns `SHACLValidationReport` |
|
||||
| `OntologyQualityGate` | Run deterministic ontology/KG quality checks for CI |
|
||||
| `OWLGenerator` | Serialize ontologies to Turtle, RDF/XML, JSON-LD |
|
||||
| `NamespaceManager` | IRI generation, prefix management, and namespace binding |
|
||||
| `OntologyEvaluator` | Coverage, completeness, and granularity quality metrics |
|
||||
@@ -81,9 +82,38 @@ engine.export_owl(ontology, "ontology.ttl", format="turtle")
|
||||
| :------ | :----------- |
|
||||
| `from_data(data)` | Run the 5-stage pipeline on entity/relationship data |
|
||||
| `validate_graph(kg, ontology=...)` | Check a knowledge graph against generated SHACL shapes |
|
||||
| `quality_check(ontology, graph_data=...)` | Return a deterministic quality report and CI-friendly pass/fail result |
|
||||
| `export_owl(ontology, path, format)` | Serialize to `"turtle"`, `"xml"`, or `"json-ld"` |
|
||||
| `evaluate(ontology, kg)` | Compute coverage, completeness, and granularity metrics |
|
||||
|
||||
### Ontology Quality Gate
|
||||
|
||||
Use the quality gate before export or deployment to catch structural issues
|
||||
without adding a runtime dependency:
|
||||
|
||||
```python
|
||||
from semantica.ontology import ontology_quality_check
|
||||
|
||||
report = ontology_quality_check(
|
||||
ontology,
|
||||
graph_data=kg,
|
||||
thresholds={"min_coverage": 0.8},
|
||||
)
|
||||
|
||||
if not report.passed:
|
||||
for issue in report.issues:
|
||||
print(issue.code, issue.message)
|
||||
```
|
||||
|
||||
The report checks class/property coverage, orphan schema elements, domain and
|
||||
range references, and unresolved KG relationship endpoints. It includes
|
||||
machine-readable issue codes, severity, counts, metrics, and threshold
|
||||
failures. The first version reports findings only; it does not auto-fix data.
|
||||
|
||||
### Thresholds
|
||||
|
||||
`min_coverage` (default `0.0`) sets the minimum required `coverage` score, the average of class and property coverage from `0.0` to `1.0`; the gate fails below it. `max_errors` (default `0.0`) caps how many `error`/`critical` issues are allowed before the gate fails. `max_warnings` (default `None`) caps `warning` issues the same way, and `None` means warnings alone never fail the gate. `fail_on_warnings` is a separate parameter, not a `thresholds` key, passed to `OntologyQualityGate(...)` or `.check(...)` directly; when `True`, a single warning fails the gate regardless of `max_warnings`.
|
||||
|
||||
## OntologyGenerator (5-Stage Pipeline)
|
||||
|
||||
**`OntologyGenerator`** auto-generates a formal ontology from your knowledge graph entities and relationships:
|
||||
|
||||
+36
-7
@@ -47,7 +47,11 @@ dependencies = [
|
||||
"numpy>=2.0.2",
|
||||
"pandas>=1.3.0",
|
||||
"scipy>=1.13.1",
|
||||
"scikit-learn>=1.7.2",
|
||||
# scikit-learn dropped Python 3.9 support at 1.7.0 (requires_python >=3.10),
|
||||
# so an unqualified >=1.7.2 floor is unsatisfiable on 3.9. Cap 3.9 to the
|
||||
# last 3.9-compatible release line; 3.10+ is left unconstrained.
|
||||
"scikit-learn>=1.6.1,<1.7.0; python_version < '3.10'",
|
||||
"scikit-learn>=1.7.2; python_version >= '3.10'",
|
||||
"umap-learn>=0.5.12",
|
||||
# thinc (spacy's core dep) dropped Python 3.9 wheels at 8.3.10, and later
|
||||
# spacy patch releases (3.8.8+) require thinc>=8.3.9-only-on-3.10+ ranges,
|
||||
@@ -66,24 +70,49 @@ dependencies = [
|
||||
"seaborn>=0.13.2",
|
||||
"plotly>=6.8.0",
|
||||
"ipywidgets>=8.0.0",
|
||||
"requests>=2.34.2",
|
||||
# requests dropped Python 3.9 support at 2.33.0 (requires_python >=3.10),
|
||||
# so an unqualified >=2.34.2 floor is unsatisfiable on 3.9. Cap 3.9 to the
|
||||
# last 3.9-compatible release; 3.10+ is left unconstrained.
|
||||
"requests>=2.32.5,<2.33.0; python_version < '3.10'",
|
||||
"requests>=2.34.2; python_version >= '3.10'",
|
||||
"GitPython>=3.1.58",
|
||||
"chardet>=7.4.3",
|
||||
# chardet dropped Python 3.9 support at 6.0.0 (requires_python >=3.10), so
|
||||
# an unqualified >=7.4.3 floor is unsatisfiable on 3.9. Cap 3.9 to the last
|
||||
# 3.9-compatible release; 3.10+ is left unconstrained.
|
||||
"chardet>=5.2.0,<6.0.0; python_version < '3.10'",
|
||||
"chardet>=7.4.3; python_version >= '3.10'",
|
||||
"protobuf>=5.29.1,<8.0",
|
||||
"grpcio>=1.81.1",
|
||||
# grpcio dropped Python 3.9 support at 1.81.0 (requires_python >=3.10), so
|
||||
# an unqualified >=1.81.1 floor is unsatisfiable on 3.9. Cap 3.9 to the last
|
||||
# 3.9-compatible release; 3.10+ is left unconstrained.
|
||||
"grpcio>=1.80.0,<1.81.0; python_version < '3.10'",
|
||||
"grpcio>=1.81.1; python_version >= '3.10'",
|
||||
"beautifulsoup4>=4.15.0",
|
||||
"lxml>=6.1.1",
|
||||
"python-docx>=1.2.0",
|
||||
"openpyxl>=3.1.5",
|
||||
"pillow>=12.2.0",
|
||||
# pillow dropped Python 3.9 support at 12.0.0 (requires_python >=3.10), so
|
||||
# an unqualified >=12.2.0 floor is unsatisfiable on 3.9. Cap 3.9 to the last
|
||||
# 3.9-compatible release; 3.10+ is left unconstrained.
|
||||
"pillow>=11.3.0,<12.0.0; python_version < '3.10'",
|
||||
"pillow>=12.2.0; python_version >= '3.10'",
|
||||
"librosa>=0.9.0",
|
||||
"opencv-python>=4.13.0.92",
|
||||
"faiss-cpu>=1.7.0",
|
||||
"fastembed>=0.2.0",
|
||||
"onnxruntime>=1.20.1",
|
||||
# onnxruntime stopped shipping cp39 wheels at 1.20.0 (its PyPI metadata
|
||||
# still claims requires_python >=3.9, but no matching wheel exists), so an
|
||||
# unqualified >=1.20.1 floor is unsatisfiable on 3.9. Cap 3.9 to the last
|
||||
# release with a cp39 wheel; 3.10+ is left unconstrained.
|
||||
"onnxruntime>=1.19.2,<1.20.0; python_version < '3.10'",
|
||||
"onnxruntime>=1.20.1; python_version >= '3.10'",
|
||||
"tokenizers>=0.15.0",
|
||||
"pydantic>=2.13.4",
|
||||
"click>=8.4.2",
|
||||
# click dropped Python 3.9 support at 8.2.0 (requires_python >=3.10), so an
|
||||
# unqualified >=8.4.2 floor is unsatisfiable on 3.9. Cap 3.9 to the last
|
||||
# 3.9-compatible release; 3.10+ is left unconstrained.
|
||||
"click>=8.1.8,<8.2.0; python_version < '3.10'",
|
||||
"click>=8.4.2; python_version >= '3.10'",
|
||||
"rich>=12.5.0",
|
||||
"tqdm>=4.68.3",
|
||||
"pyyaml>=6.0",
|
||||
|
||||
@@ -301,7 +301,10 @@ class GraphValidator:
|
||||
code="ORPHAN_NODES",
|
||||
message=f"Found {len(isolates)} orphan nodes (no relationships).",
|
||||
severity=ValidationSeverity.WARNING,
|
||||
details={"count": len(isolates), "ids": isolates[:10]} # Limit output
|
||||
details={
|
||||
"count": len(isolates),
|
||||
"ids": sorted(isolates, key=str)[:10],
|
||||
} # Limit output deterministically
|
||||
))
|
||||
|
||||
except Exception as e:
|
||||
|
||||
@@ -35,7 +35,7 @@ Example Usage:
|
||||
>>> llm = LiteLLM(model="openai/gpt-4o", api_key="your-key")
|
||||
>>> response = llm.generate("Hello, world!")
|
||||
>>> # Or use other providers via LiteLLM
|
||||
>>> llm = LiteLLM(model="anthropic/claude-sonnet-4-20250514")
|
||||
>>> llm = LiteLLM(model="anthropic/claude-sonnet-5")
|
||||
>>> response = llm.generate("Hello, world!")
|
||||
>>>
|
||||
>>> # Anthropic provider
|
||||
|
||||
@@ -31,7 +31,7 @@ class LiteLLM:
|
||||
Provides unified interface to 100+ LLM providers through LiteLLM library.
|
||||
Supports providers like OpenAI, Anthropic, Groq, Azure, Bedrock, Vertex AI, etc.
|
||||
|
||||
Model format: "provider/model-name" (e.g., "openai/gpt-4o", "anthropic/claude-sonnet-4-20250514", "groq/llama-3.1-8b-instant")
|
||||
Model format: "provider/model-name" (e.g., "openai/gpt-4o", "anthropic/claude-sonnet-5", "groq/llama-3.1-8b-instant")
|
||||
|
||||
Example:
|
||||
>>> from semantica.llms import LiteLLM
|
||||
@@ -39,7 +39,7 @@ class LiteLLM:
|
||||
>>> response = llm.generate("What is AI?")
|
||||
>>>
|
||||
>>> # Use with different providers
|
||||
>>> llm = LiteLLM(model="anthropic/claude-sonnet-4-20250514")
|
||||
>>> llm = LiteLLM(model="anthropic/claude-sonnet-5")
|
||||
>>> response = llm.generate("Hello!")
|
||||
"""
|
||||
|
||||
@@ -54,7 +54,7 @@ class LiteLLM:
|
||||
|
||||
Args:
|
||||
model: Model identifier in format "provider/model-name"
|
||||
Examples: "openai/gpt-4o", "anthropic/claude-sonnet-4-20250514",
|
||||
Examples: "openai/gpt-4o", "anthropic/claude-sonnet-5",
|
||||
"groq/llama-3.1-8b-instant", "azure/gpt-4", etc.
|
||||
api_key: API key (optional, can use environment variables)
|
||||
**kwargs: Additional LiteLLM options (temperature, max_tokens, etc.)
|
||||
|
||||
@@ -162,6 +162,13 @@ from .ontology_validator import (
|
||||
run_shacl_validation,
|
||||
validate_ontology,
|
||||
)
|
||||
from .quality_gate import (
|
||||
OntologyQualityGate,
|
||||
OntologyQualityReport,
|
||||
QualityIssue,
|
||||
QualitySeverity,
|
||||
ontology_quality_check,
|
||||
)
|
||||
from .owl_generator import OWLGenerator
|
||||
from .property_generator import PropertyGenerator
|
||||
from .registry import MethodRegistry, method_registry
|
||||
@@ -194,6 +201,11 @@ __all__ = [
|
||||
"SHACLValidationReport",
|
||||
"SHACLViolation",
|
||||
"run_shacl_validation",
|
||||
"OntologyQualityGate",
|
||||
"OntologyQualityReport",
|
||||
"QualityIssue",
|
||||
"QualitySeverity",
|
||||
"ontology_quality_check",
|
||||
# OWL/RDF generation
|
||||
"OWLGenerator",
|
||||
# Requirements and competency questions
|
||||
|
||||
@@ -9,6 +9,7 @@ from .property_generator import PropertyGenerator
|
||||
from .owl_generator import OWLGenerator
|
||||
from .ontology_evaluator import OntologyEvaluator
|
||||
from .ontology_validator import OntologyValidator
|
||||
from .quality_gate import OntologyQualityGate, OntologyQualityReport
|
||||
from .llm_generator import LLMOntologyGenerator
|
||||
from ..semantic_extract.triplet_extractor import Triplet
|
||||
|
||||
@@ -25,6 +26,9 @@ class OntologyEngine:
|
||||
self.owl = OWLGenerator(**config)
|
||||
self.evaluator = OntologyEvaluator(**config)
|
||||
self.validator = OntologyValidator(**config)
|
||||
self.quality_gate = OntologyQualityGate(
|
||||
validator=self.validator, evaluator=self.evaluator
|
||||
)
|
||||
self.llm = LLMOntologyGenerator(**config)
|
||||
self.store = config.get("store")
|
||||
|
||||
@@ -593,6 +597,15 @@ class OntologyEngine:
|
||||
def validate(self, ontology: Dict[str, Any], **options):
|
||||
return self.validator.validate(ontology, **options)
|
||||
|
||||
def quality_check(
|
||||
self,
|
||||
ontology: Dict[str, Any],
|
||||
graph_data: Optional[Dict[str, Any]] = None,
|
||||
**options,
|
||||
) -> OntologyQualityReport:
|
||||
"""Run deterministic ontology quality checks suitable for CI."""
|
||||
return self.quality_gate.check(ontology, graph_data=graph_data, **options)
|
||||
|
||||
def to_owl(self, ontology: Dict[str, Any], format: str = "turtle", **options):
|
||||
return self.owl.generate_owl(ontology, format=format, **options)
|
||||
|
||||
|
||||
@@ -0,0 +1,855 @@
|
||||
"""Deterministic quality checks for ontology and KG pipelines."""
|
||||
|
||||
import copy
|
||||
import math
|
||||
from dataclasses import dataclass, field
|
||||
from enum import Enum
|
||||
from typing import Any, Dict, Iterable, List, Mapping, Optional, Set, Tuple
|
||||
|
||||
from ..kg.graph_validator import GraphValidator
|
||||
from .ontology_evaluator import OntologyEvaluator
|
||||
from .ontology_validator import OntologyValidator
|
||||
|
||||
|
||||
class QualitySeverity(str, Enum):
|
||||
"""Severity assigned to a quality finding."""
|
||||
|
||||
INFO = "info"
|
||||
WARNING = "warning"
|
||||
ERROR = "error"
|
||||
CRITICAL = "critical"
|
||||
|
||||
|
||||
@dataclass
|
||||
class QualityIssue:
|
||||
"""A single machine-readable ontology quality finding."""
|
||||
|
||||
code: str
|
||||
message: str
|
||||
severity: QualitySeverity
|
||||
element_id: Optional[str] = None
|
||||
element_type: Optional[str] = None
|
||||
details: Dict[str, Any] = field(default_factory=dict)
|
||||
|
||||
def to_dict(self) -> Dict[str, Any]:
|
||||
"""Return a JSON-friendly representation of the issue."""
|
||||
return {
|
||||
"code": self.code,
|
||||
"message": self.message,
|
||||
"severity": self.severity.value,
|
||||
"element_id": self.element_id,
|
||||
"element_type": self.element_type,
|
||||
"details": self.details,
|
||||
}
|
||||
|
||||
|
||||
@dataclass
|
||||
class OntologyQualityReport:
|
||||
"""Result returned by :class:`OntologyQualityGate`."""
|
||||
|
||||
passed: bool
|
||||
issues: List[QualityIssue] = field(default_factory=list)
|
||||
stats: Dict[str, int] = field(default_factory=dict)
|
||||
metrics: Dict[str, float] = field(default_factory=dict)
|
||||
thresholds: Dict[str, Optional[float]] = field(default_factory=dict)
|
||||
threshold_failures: List[str] = field(default_factory=list)
|
||||
|
||||
@property
|
||||
def error_count(self) -> int:
|
||||
"""Number of error and critical findings."""
|
||||
return sum(
|
||||
issue.severity in (QualitySeverity.ERROR, QualitySeverity.CRITICAL)
|
||||
for issue in self.issues
|
||||
)
|
||||
|
||||
@property
|
||||
def warning_count(self) -> int:
|
||||
"""Number of warning findings."""
|
||||
return sum(issue.severity == QualitySeverity.WARNING for issue in self.issues)
|
||||
|
||||
@property
|
||||
def info_count(self) -> int:
|
||||
"""Number of informational findings."""
|
||||
return sum(issue.severity == QualitySeverity.INFO for issue in self.issues)
|
||||
|
||||
def to_dict(self) -> Dict[str, Any]:
|
||||
"""Return a JSON-friendly representation of the report."""
|
||||
return {
|
||||
"passed": self.passed,
|
||||
"issues": [issue.to_dict() for issue in self.issues],
|
||||
"stats": {
|
||||
**self.stats,
|
||||
"issues": len(self.issues),
|
||||
"errors": self.error_count,
|
||||
"warnings": self.warning_count,
|
||||
"infos": self.info_count,
|
||||
},
|
||||
"metrics": self.metrics,
|
||||
"thresholds": self.thresholds,
|
||||
"threshold_failures": self.threshold_failures,
|
||||
}
|
||||
|
||||
|
||||
class OntologyQualityGate:
|
||||
"""Run deterministic, CI-friendly ontology quality checks."""
|
||||
|
||||
DEFAULT_THRESHOLDS: Dict[str, Optional[float]] = {
|
||||
"min_coverage": 0.0,
|
||||
"max_errors": 0.0,
|
||||
"max_warnings": None,
|
||||
}
|
||||
_DATA_PROPERTY_TYPES = {"data", "datatype", "data_property", "literal"}
|
||||
_OBJECT_PROPERTY_TYPES = {"object", "object_property", "relationship"}
|
||||
_KNOWN_DATATYPES = {
|
||||
"string",
|
||||
"boolean",
|
||||
"decimal",
|
||||
"float",
|
||||
"double",
|
||||
"integer",
|
||||
"int",
|
||||
"long",
|
||||
"short",
|
||||
"byte",
|
||||
"date",
|
||||
"datetime",
|
||||
"datetimestamp",
|
||||
"time",
|
||||
"duration",
|
||||
"anyuri",
|
||||
}
|
||||
_BUILTIN_CLASSES = {
|
||||
"owl:thing",
|
||||
"rdfs:resource",
|
||||
"http://www.w3.org/2002/07/owl#thing",
|
||||
"http://www.w3.org/2000/01/rdf-schema#resource",
|
||||
}
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
validator: Optional[OntologyValidator] = None,
|
||||
evaluator: Optional[OntologyEvaluator] = None,
|
||||
thresholds: Optional[Mapping[str, Optional[float]]] = None,
|
||||
fail_on_warnings: bool = False,
|
||||
) -> None:
|
||||
self.validator = validator or OntologyValidator()
|
||||
self.evaluator = evaluator or OntologyEvaluator()
|
||||
self.thresholds = dict(self.DEFAULT_THRESHOLDS)
|
||||
if thresholds:
|
||||
self.thresholds.update(thresholds)
|
||||
self.fail_on_warnings = fail_on_warnings
|
||||
|
||||
def check(
|
||||
self,
|
||||
ontology: Any,
|
||||
graph_data: Optional[Dict[str, Any]] = None,
|
||||
*,
|
||||
thresholds: Optional[Mapping[str, Optional[float]]] = None,
|
||||
fail_on_warnings: Optional[bool] = None,
|
||||
competency_questions: Optional[List[str]] = None,
|
||||
) -> OntologyQualityReport:
|
||||
"""Check an ontology and optionally its instance graph.
|
||||
|
||||
``graph_data`` is optional because an ontology can be checked before
|
||||
instances are available. When omitted, embedded ``entities`` and
|
||||
``relationships`` are checked when present.
|
||||
"""
|
||||
active_thresholds = dict(self.thresholds)
|
||||
if thresholds:
|
||||
active_thresholds.update(thresholds)
|
||||
min_coverage, max_errors, max_warnings = self._validate_thresholds(
|
||||
active_thresholds
|
||||
)
|
||||
should_fail_on_warnings = (
|
||||
self.fail_on_warnings if fail_on_warnings is None else fail_on_warnings
|
||||
)
|
||||
issues: List[QualityIssue] = []
|
||||
|
||||
if not isinstance(ontology, dict):
|
||||
self._add_issue(
|
||||
issues,
|
||||
"INVALID_ONTOLOGY",
|
||||
"Ontology must be a dictionary.",
|
||||
QualitySeverity.CRITICAL,
|
||||
element_type="ontology",
|
||||
)
|
||||
return self._build_report(
|
||||
issues,
|
||||
classes=0,
|
||||
properties=0,
|
||||
entities=0,
|
||||
relationships=0,
|
||||
metrics={
|
||||
"coverage": 0.0,
|
||||
"class_coverage": 0.0,
|
||||
"property_coverage": 0.0,
|
||||
},
|
||||
thresholds=active_thresholds,
|
||||
min_coverage=min_coverage,
|
||||
max_errors=max_errors,
|
||||
max_warnings=max_warnings,
|
||||
fail_on_warnings=should_fail_on_warnings,
|
||||
)
|
||||
|
||||
validation = self.validator.validate(ontology)
|
||||
for message in getattr(validation, "errors", []) or []:
|
||||
self._add_issue(
|
||||
issues,
|
||||
"VALIDATOR_ERROR",
|
||||
str(message),
|
||||
QualitySeverity.ERROR,
|
||||
element_type="ontology",
|
||||
)
|
||||
for message in getattr(validation, "warnings", []) or []:
|
||||
self._add_issue(
|
||||
issues,
|
||||
"VALIDATOR_WARNING",
|
||||
str(message),
|
||||
QualitySeverity.WARNING,
|
||||
element_type="ontology",
|
||||
)
|
||||
|
||||
classes = self._read_collection(ontology, "classes", issues)
|
||||
properties = self._read_collection(ontology, "properties", issues)
|
||||
class_aliases, class_ids = self._index_elements(
|
||||
classes, "class", "MISSING_CLASS_ID", issues
|
||||
)
|
||||
referenced_classes: Set[str] = set()
|
||||
self._mark_hierarchy(classes, class_aliases, referenced_classes)
|
||||
|
||||
property_with_endpoints = 0
|
||||
for index, prop in enumerate(properties):
|
||||
if not isinstance(prop, dict):
|
||||
self._add_issue(
|
||||
issues,
|
||||
"INVALID_PROPERTY",
|
||||
f"Property at index {index} must be a dictionary.",
|
||||
QualitySeverity.ERROR,
|
||||
element_type="property",
|
||||
details={"index": index},
|
||||
)
|
||||
continue
|
||||
|
||||
prop_id = self._identifier(prop)
|
||||
if prop_id is None:
|
||||
prop_id = f"property[{index}]"
|
||||
self._add_issue(
|
||||
issues,
|
||||
"MISSING_PROPERTY_ID",
|
||||
f"Property at index {index} has no name or URI.",
|
||||
QualitySeverity.ERROR,
|
||||
element_type="property",
|
||||
details={"index": index},
|
||||
)
|
||||
|
||||
raw_prop_type = prop.get("type")
|
||||
prop_type = (
|
||||
str(raw_prop_type).strip().lower() if raw_prop_type is not None else ""
|
||||
)
|
||||
if not prop_type:
|
||||
self._add_issue(
|
||||
issues,
|
||||
"MISSING_PROPERTY_TYPE",
|
||||
f"Property '{prop_id}' has no type.",
|
||||
QualitySeverity.ERROR,
|
||||
element_id=prop_id,
|
||||
element_type="property",
|
||||
)
|
||||
elif (
|
||||
prop_type not in self._DATA_PROPERTY_TYPES | self._OBJECT_PROPERTY_TYPES
|
||||
):
|
||||
self._add_issue(
|
||||
issues,
|
||||
"UNKNOWN_PROPERTY_TYPE",
|
||||
f"Property '{prop_id}' has unknown type '{prop_type}'.",
|
||||
QualitySeverity.WARNING,
|
||||
element_id=prop_id,
|
||||
element_type="property",
|
||||
)
|
||||
|
||||
domains = self._values(prop, "domain")
|
||||
ranges = self._values(prop, "range")
|
||||
if domains or ranges:
|
||||
property_with_endpoints += 1
|
||||
else:
|
||||
self._add_issue(
|
||||
issues,
|
||||
"ORPHAN_PROPERTY",
|
||||
f"Property '{prop_id}' has no domain or range.",
|
||||
QualitySeverity.WARNING,
|
||||
element_id=prop_id,
|
||||
element_type="property",
|
||||
)
|
||||
if not domains:
|
||||
self._add_issue(
|
||||
issues,
|
||||
"MISSING_DOMAIN",
|
||||
f"Property '{prop_id}' has no domain.",
|
||||
QualitySeverity.WARNING,
|
||||
element_id=prop_id,
|
||||
element_type="property",
|
||||
)
|
||||
for domain in domains:
|
||||
matched = self._match_class(domain, class_aliases)
|
||||
if matched:
|
||||
referenced_classes.add(matched)
|
||||
elif not self._is_builtin_class(domain):
|
||||
self._add_issue(
|
||||
issues,
|
||||
"UNKNOWN_DOMAIN",
|
||||
f"Property '{prop_id}' references unknown domain '{domain}'.",
|
||||
QualitySeverity.ERROR,
|
||||
element_id=prop_id,
|
||||
element_type="property",
|
||||
details={"domain": domain},
|
||||
)
|
||||
|
||||
if not ranges:
|
||||
self._add_issue(
|
||||
issues,
|
||||
"MISSING_RANGE",
|
||||
f"Property '{prop_id}' has no range.",
|
||||
QualitySeverity.WARNING,
|
||||
element_id=prop_id,
|
||||
element_type="property",
|
||||
)
|
||||
for range_value in ranges:
|
||||
matched = self._match_class(range_value, class_aliases)
|
||||
if matched:
|
||||
referenced_classes.add(matched)
|
||||
self._check_range(
|
||||
prop_id,
|
||||
prop_type,
|
||||
range_value,
|
||||
matched is not None,
|
||||
issues,
|
||||
)
|
||||
|
||||
graph = graph_data
|
||||
if graph is None and ("entities" in ontology or "relationships" in ontology):
|
||||
graph = ontology
|
||||
graph_entities, graph_relationships = self._read_graph(graph, issues)
|
||||
if self._has_valid_graph_shape(graph):
|
||||
self._check_graph(graph, issues)
|
||||
self._mark_graph_types(graph_entities, class_aliases, referenced_classes)
|
||||
|
||||
for class_id in class_ids:
|
||||
if class_id not in referenced_classes:
|
||||
self._add_issue(
|
||||
issues,
|
||||
"ORPHAN_CLASS",
|
||||
f"Class '{class_id}' is not connected to a property, hierarchy, or graph entity type.",
|
||||
QualitySeverity.WARNING,
|
||||
element_id=class_id,
|
||||
element_type="class",
|
||||
)
|
||||
|
||||
class_coverage = (
|
||||
len(referenced_classes & set(class_ids)) / len(class_ids)
|
||||
if class_ids
|
||||
else 0.0
|
||||
)
|
||||
property_coverage = (
|
||||
property_with_endpoints / len(properties) if properties else 1.0
|
||||
)
|
||||
metrics: Dict[str, float] = {
|
||||
"coverage": (
|
||||
(class_coverage + property_coverage) / 2
|
||||
if classes or properties
|
||||
else 0.0
|
||||
),
|
||||
"class_coverage": class_coverage,
|
||||
"property_coverage": property_coverage,
|
||||
"validator_valid": 1.0 if getattr(validation, "valid", True) else 0.0,
|
||||
}
|
||||
if competency_questions is not None:
|
||||
evaluation_ontology = self._prepare_for_evaluation(
|
||||
ontology, classes, properties
|
||||
)
|
||||
evaluation = self._evaluate_competency_questions(
|
||||
evaluation_ontology, competency_questions=competency_questions
|
||||
)
|
||||
metrics["competency_question_coverage"] = evaluation.coverage_score
|
||||
metrics["completeness"] = evaluation.completeness_score
|
||||
|
||||
return self._build_report(
|
||||
issues,
|
||||
classes=len(classes),
|
||||
properties=len(properties),
|
||||
entities=len(graph_entities),
|
||||
relationships=len(graph_relationships),
|
||||
metrics=metrics,
|
||||
thresholds=active_thresholds,
|
||||
min_coverage=min_coverage,
|
||||
max_errors=max_errors,
|
||||
max_warnings=max_warnings,
|
||||
fail_on_warnings=should_fail_on_warnings,
|
||||
)
|
||||
|
||||
def _check_range(
|
||||
self,
|
||||
prop_id: str,
|
||||
prop_type: str,
|
||||
range_value: Any,
|
||||
is_class: bool,
|
||||
issues: List[QualityIssue],
|
||||
) -> None:
|
||||
if prop_type in self._DATA_PROPERTY_TYPES:
|
||||
if not self._is_known_datatype(range_value):
|
||||
self._add_issue(
|
||||
issues,
|
||||
"INVALID_DATATYPE_RANGE",
|
||||
f"Data property '{prop_id}' has invalid range '{range_value}'.",
|
||||
QualitySeverity.ERROR,
|
||||
element_id=prop_id,
|
||||
element_type="property",
|
||||
details={"range": range_value},
|
||||
)
|
||||
elif prop_type in self._OBJECT_PROPERTY_TYPES:
|
||||
if not is_class and not self._is_builtin_class(range_value):
|
||||
self._add_issue(
|
||||
issues,
|
||||
"UNKNOWN_RANGE",
|
||||
f"Object property '{prop_id}' references unknown range '{range_value}'.",
|
||||
QualitySeverity.ERROR,
|
||||
element_id=prop_id,
|
||||
element_type="property",
|
||||
details={"range": range_value},
|
||||
)
|
||||
elif (
|
||||
not is_class
|
||||
and not self._is_builtin_class(range_value)
|
||||
and not self._is_known_datatype(range_value)
|
||||
):
|
||||
self._add_issue(
|
||||
issues,
|
||||
"UNKNOWN_RANGE",
|
||||
f"Property '{prop_id}' references unknown range '{range_value}'.",
|
||||
QualitySeverity.ERROR,
|
||||
element_id=prop_id,
|
||||
element_type="property",
|
||||
details={"range": range_value},
|
||||
)
|
||||
|
||||
def _build_report(
|
||||
self,
|
||||
issues: List[QualityIssue],
|
||||
*,
|
||||
classes: int,
|
||||
properties: int,
|
||||
entities: int,
|
||||
relationships: int,
|
||||
metrics: Dict[str, float],
|
||||
thresholds: Dict[str, Optional[float]],
|
||||
min_coverage: float,
|
||||
max_errors: float,
|
||||
max_warnings: Optional[float],
|
||||
fail_on_warnings: bool,
|
||||
) -> OntologyQualityReport:
|
||||
failures: List[str] = []
|
||||
error_count = sum(
|
||||
issue.severity in (QualitySeverity.ERROR, QualitySeverity.CRITICAL)
|
||||
for issue in issues
|
||||
)
|
||||
warning_count = sum(
|
||||
issue.severity == QualitySeverity.WARNING for issue in issues
|
||||
)
|
||||
if error_count > max_errors:
|
||||
failures.append("max_errors")
|
||||
if max_warnings is not None and warning_count > max_warnings:
|
||||
failures.append("max_warnings")
|
||||
if fail_on_warnings and warning_count:
|
||||
failures.append("fail_on_warnings")
|
||||
if metrics.get("coverage", 0.0) < min_coverage:
|
||||
failures.append("min_coverage")
|
||||
return OntologyQualityReport(
|
||||
passed=not failures,
|
||||
issues=issues,
|
||||
stats={
|
||||
"classes": classes,
|
||||
"properties": properties,
|
||||
"entities": entities,
|
||||
"relationships": relationships,
|
||||
},
|
||||
metrics=metrics,
|
||||
thresholds=thresholds,
|
||||
threshold_failures=failures,
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _validate_thresholds(
|
||||
thresholds: Mapping[str, Optional[float]],
|
||||
) -> Tuple[float, float, Optional[float]]:
|
||||
min_coverage = float(thresholds.get("min_coverage", 0.0) or 0.0)
|
||||
max_errors = float(thresholds.get("max_errors", 0.0) or 0.0)
|
||||
warning_value = thresholds.get("max_warnings")
|
||||
max_warnings = None if warning_value is None else float(warning_value)
|
||||
if not all(
|
||||
math.isfinite(value)
|
||||
for value in (min_coverage, max_errors)
|
||||
if value is not None
|
||||
) or (max_warnings is not None and not math.isfinite(max_warnings)):
|
||||
raise ValueError("quality thresholds must be finite numbers")
|
||||
if not 0.0 <= min_coverage <= 1.0:
|
||||
raise ValueError("min_coverage must be between 0.0 and 1.0")
|
||||
if max_errors < 0 or (max_warnings is not None and max_warnings < 0):
|
||||
raise ValueError("error and warning thresholds cannot be negative")
|
||||
return min_coverage, max_errors, max_warnings
|
||||
|
||||
@classmethod
|
||||
def _prepare_for_evaluation(
|
||||
cls,
|
||||
ontology: Dict[str, Any],
|
||||
classes: Iterable[Any],
|
||||
properties: Iterable[Any],
|
||||
) -> Dict[str, Any]:
|
||||
"""Make a shallow, evaluator-safe view without changing caller data."""
|
||||
prepared = dict(ontology)
|
||||
prepared["classes"] = [
|
||||
cls._prepare_element(element)
|
||||
for element in classes
|
||||
if isinstance(element, dict)
|
||||
]
|
||||
prepared["properties"] = [
|
||||
cls._prepare_element(element)
|
||||
for element in properties
|
||||
if isinstance(element, dict)
|
||||
]
|
||||
return prepared
|
||||
|
||||
@classmethod
|
||||
def _prepare_element(cls, element: Dict[str, Any]) -> Dict[str, Any]:
|
||||
prepared = dict(element)
|
||||
identifier = cls._identifier(prepared)
|
||||
if identifier is not None and not str(prepared.get("name", "")).strip():
|
||||
prepared["name"] = identifier
|
||||
return prepared
|
||||
|
||||
def _evaluate_competency_questions(
|
||||
self, ontology: Dict[str, Any], competency_questions: List[str]
|
||||
) -> Any:
|
||||
"""Evaluate with an isolated question manager for repeatable checks."""
|
||||
evaluator = copy.copy(self.evaluator)
|
||||
manager = getattr(self.evaluator, "competency_questions_manager", None)
|
||||
if manager is not None and hasattr(manager, "questions"):
|
||||
isolated_manager = copy.copy(manager)
|
||||
isolated_manager.questions = []
|
||||
evaluator.competency_questions_manager = isolated_manager
|
||||
return evaluator.evaluate_ontology(
|
||||
ontology, competency_questions=competency_questions
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def _read_collection(
|
||||
cls, ontology: Dict[str, Any], key: str, issues: List[QualityIssue]
|
||||
) -> List[Any]:
|
||||
if key not in ontology:
|
||||
cls._add_issue(
|
||||
issues,
|
||||
f"MISSING_{key.upper()}",
|
||||
f"Ontology has no {key} defined.",
|
||||
QualitySeverity.WARNING,
|
||||
element_type="ontology",
|
||||
)
|
||||
return []
|
||||
value = ontology[key]
|
||||
if not isinstance(value, list):
|
||||
cls._add_issue(
|
||||
issues,
|
||||
f"INVALID_{key.upper()}",
|
||||
f"Ontology '{key}' must be a list.",
|
||||
QualitySeverity.ERROR,
|
||||
element_type="ontology",
|
||||
)
|
||||
return []
|
||||
return value
|
||||
|
||||
@classmethod
|
||||
def _index_elements(
|
||||
cls,
|
||||
elements: Iterable[Any],
|
||||
element_type: str,
|
||||
missing_code: str,
|
||||
issues: List[QualityIssue],
|
||||
) -> Tuple[Dict[str, str], List[str]]:
|
||||
aliases: Dict[str, str] = {}
|
||||
identifiers: List[str] = []
|
||||
for index, element in enumerate(elements):
|
||||
identifier = cls._identifier(element)
|
||||
if identifier is None:
|
||||
cls._add_issue(
|
||||
issues,
|
||||
missing_code,
|
||||
f"{element_type.title()} at index {index} has no name or URI.",
|
||||
QualitySeverity.ERROR,
|
||||
element_type=element_type,
|
||||
details={"index": index},
|
||||
)
|
||||
continue
|
||||
identifiers.append(identifier)
|
||||
aliases_for_element: Set[str] = set()
|
||||
for candidate in cls._identifiers(element):
|
||||
aliases_for_element.update(cls._term_aliases(candidate))
|
||||
for alias in sorted(aliases_for_element):
|
||||
aliases.setdefault(alias, identifier)
|
||||
return aliases, identifiers
|
||||
|
||||
@classmethod
|
||||
def _mark_hierarchy(
|
||||
cls,
|
||||
classes: Iterable[Any],
|
||||
aliases: Mapping[str, str],
|
||||
referenced: Set[str],
|
||||
) -> None:
|
||||
for class_entry in classes:
|
||||
if not isinstance(class_entry, dict):
|
||||
continue
|
||||
identifier = cls._identifier(class_entry)
|
||||
for key in (
|
||||
"subClassOf",
|
||||
"subclassOf",
|
||||
"parent",
|
||||
"superclass",
|
||||
"superclasses",
|
||||
):
|
||||
parents = cls._values(class_entry, key)
|
||||
if parents and identifier:
|
||||
referenced.add(identifier)
|
||||
for parent in parents:
|
||||
matched = cls._match_class(parent, aliases)
|
||||
if matched:
|
||||
referenced.add(matched)
|
||||
|
||||
@classmethod
|
||||
def _mark_graph_types(
|
||||
cls,
|
||||
entities: Iterable[Any],
|
||||
aliases: Mapping[str, str],
|
||||
referenced: Set[str],
|
||||
) -> None:
|
||||
for entity in entities:
|
||||
if not isinstance(entity, dict):
|
||||
continue
|
||||
entity_type = entity.get("type") or entity.get("entity_type")
|
||||
matched = cls._match_class(entity_type, aliases)
|
||||
if matched:
|
||||
referenced.add(matched)
|
||||
|
||||
@classmethod
|
||||
def _check_graph(cls, graph: Dict[str, Any], issues: List[QualityIssue]) -> None:
|
||||
safe_graph = cls._prepare_graph_for_validation(graph, issues)
|
||||
result = GraphValidator().validate(safe_graph)
|
||||
code_map = {
|
||||
"DANGLING_EDGE": "UNRESOLVED_RELATIONSHIP_ENDPOINT",
|
||||
"ORPHAN_NODES": "ORPHAN_ENTITY",
|
||||
}
|
||||
severity_map = {
|
||||
"info": QualitySeverity.INFO,
|
||||
"warning": QualitySeverity.WARNING,
|
||||
"error": QualitySeverity.ERROR,
|
||||
"critical": QualitySeverity.CRITICAL,
|
||||
}
|
||||
for graph_issue in result.issues:
|
||||
severity = severity_map.get(
|
||||
graph_issue.severity.value, QualitySeverity.ERROR
|
||||
)
|
||||
details = dict(graph_issue.details or {})
|
||||
if graph_issue.code == "ORPHAN_NODES" and "ids" in details:
|
||||
details["ids"] = sorted(details["ids"], key=str)
|
||||
cls._add_issue(
|
||||
issues,
|
||||
code_map.get(graph_issue.code, graph_issue.code),
|
||||
graph_issue.message,
|
||||
severity,
|
||||
element_id=graph_issue.element_id,
|
||||
element_type=graph_issue.element_type,
|
||||
details=details,
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def _prepare_graph_for_validation(
|
||||
cls, graph: Dict[str, Any], issues: List[QualityIssue]
|
||||
) -> Dict[str, List[Dict[str, Any]]]:
|
||||
"""Normalize supported graph aliases and isolate malformed members."""
|
||||
entities: List[Dict[str, Any]] = []
|
||||
raw_entities = graph.get("entities", [])
|
||||
raw_relationships = graph.get("relationships", [])
|
||||
for index, entity in enumerate(
|
||||
raw_entities if isinstance(raw_entities, list) else []
|
||||
):
|
||||
if not isinstance(entity, dict):
|
||||
cls._add_issue(
|
||||
issues,
|
||||
"INVALID_ENTITY",
|
||||
f"Entity at index {index} must be a dictionary.",
|
||||
QualitySeverity.ERROR,
|
||||
element_type="entity",
|
||||
details={"index": index},
|
||||
)
|
||||
continue
|
||||
normalized = dict(entity)
|
||||
if not normalized.get("name") and normalized.get("text") is not None:
|
||||
normalized["name"] = normalized["text"]
|
||||
entities.append(normalized)
|
||||
|
||||
relationships: List[Dict[str, Any]] = []
|
||||
for index, relationship in enumerate(
|
||||
raw_relationships if isinstance(raw_relationships, list) else []
|
||||
):
|
||||
if not isinstance(relationship, dict):
|
||||
cls._add_issue(
|
||||
issues,
|
||||
"INVALID_RELATIONSHIP",
|
||||
f"Relationship at index {index} must be a dictionary.",
|
||||
QualitySeverity.ERROR,
|
||||
element_type="relationship",
|
||||
details={"index": index},
|
||||
)
|
||||
continue
|
||||
relationships.append(dict(relationship))
|
||||
|
||||
return {"entities": entities, "relationships": relationships}
|
||||
|
||||
@classmethod
|
||||
def _read_graph(
|
||||
cls, graph: Optional[Dict[str, Any]], issues: List[QualityIssue]
|
||||
) -> Tuple[List[Any], List[Any]]:
|
||||
if graph is None:
|
||||
return [], []
|
||||
if not isinstance(graph, dict):
|
||||
cls._add_issue(
|
||||
issues,
|
||||
"INVALID_GRAPH",
|
||||
"Graph data must be a dictionary.",
|
||||
QualitySeverity.ERROR,
|
||||
element_type="graph",
|
||||
)
|
||||
return [], []
|
||||
entities = graph.get("entities", [])
|
||||
relationships = graph.get("relationships", [])
|
||||
if not isinstance(entities, list) or not isinstance(relationships, list):
|
||||
cls._add_issue(
|
||||
issues,
|
||||
"INVALID_GRAPH",
|
||||
"Graph entities and relationships must be lists.",
|
||||
QualitySeverity.ERROR,
|
||||
element_type="graph",
|
||||
)
|
||||
return [], []
|
||||
return entities, relationships
|
||||
|
||||
@staticmethod
|
||||
def _has_valid_graph_shape(graph: Optional[Dict[str, Any]]) -> bool:
|
||||
return bool(
|
||||
isinstance(graph, dict)
|
||||
and isinstance(graph.get("entities", []), list)
|
||||
and isinstance(graph.get("relationships", []), list)
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _identifier(element: Any) -> Optional[str]:
|
||||
identifiers = OntologyQualityGate._identifiers(element)
|
||||
return identifiers[0] if identifiers else None
|
||||
|
||||
@staticmethod
|
||||
def _identifiers(element: Any) -> List[str]:
|
||||
if isinstance(element, dict):
|
||||
values = []
|
||||
for key in ("name", "uri", "id", "@id"):
|
||||
value = element.get(key)
|
||||
if value is not None and str(value).strip():
|
||||
values.append(str(value).strip())
|
||||
return values
|
||||
if isinstance(element, str) and element.strip():
|
||||
return [element.strip()]
|
||||
return []
|
||||
|
||||
@staticmethod
|
||||
def _values(element: Dict[str, Any], key: str) -> List[Any]:
|
||||
value = element.get(key)
|
||||
if value is None:
|
||||
return []
|
||||
if isinstance(value, (list, tuple, set)):
|
||||
values = [item for item in value if item is not None and str(item).strip()]
|
||||
return sorted(values, key=str) if isinstance(value, set) else values
|
||||
return [value] if str(value).strip() else []
|
||||
|
||||
@classmethod
|
||||
def _term_aliases(cls, value: Any) -> Set[str]:
|
||||
if isinstance(value, dict):
|
||||
value = cls._identifier(value)
|
||||
if value is None:
|
||||
return set()
|
||||
text = str(value).strip().strip("<>")
|
||||
if not text:
|
||||
return set()
|
||||
aliases = {text, text.lower()}
|
||||
for separator in ("#", "/"):
|
||||
if separator in text:
|
||||
local = text.rstrip("/").rsplit(separator, 1)[-1]
|
||||
aliases.update({local, local.lower()})
|
||||
if ":" in text and not text.startswith(("http://", "https://")):
|
||||
local = text.rsplit(":", 1)[-1]
|
||||
aliases.update({local, local.lower()})
|
||||
return aliases
|
||||
|
||||
@classmethod
|
||||
def _match_class(cls, value: Any, aliases: Mapping[str, str]) -> Optional[str]:
|
||||
for alias in sorted(cls._term_aliases(value)):
|
||||
if alias in aliases:
|
||||
return aliases[alias]
|
||||
return None
|
||||
|
||||
@classmethod
|
||||
def _is_builtin_class(cls, value: Any) -> bool:
|
||||
return any(alias in cls._BUILTIN_CLASSES for alias in cls._term_aliases(value))
|
||||
|
||||
@classmethod
|
||||
def _is_known_datatype(cls, value: Any) -> bool:
|
||||
return any(alias in cls._KNOWN_DATATYPES for alias in cls._term_aliases(value))
|
||||
|
||||
@staticmethod
|
||||
def _add_issue(
|
||||
issues: List[QualityIssue],
|
||||
code: str,
|
||||
message: str,
|
||||
severity: QualitySeverity,
|
||||
*,
|
||||
element_id: Optional[str] = None,
|
||||
element_type: Optional[str] = None,
|
||||
details: Optional[Dict[str, Any]] = None,
|
||||
) -> None:
|
||||
issues.append(
|
||||
QualityIssue(
|
||||
code=code,
|
||||
message=message,
|
||||
severity=severity,
|
||||
element_id=element_id,
|
||||
element_type=element_type,
|
||||
details=details or {},
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def ontology_quality_check(
|
||||
ontology: Any,
|
||||
graph_data: Optional[Dict[str, Any]] = None,
|
||||
*,
|
||||
thresholds: Optional[Mapping[str, Optional[float]]] = None,
|
||||
fail_on_warnings: bool = False,
|
||||
competency_questions: Optional[List[str]] = None,
|
||||
validator: Optional[OntologyValidator] = None,
|
||||
evaluator: Optional[OntologyEvaluator] = None,
|
||||
) -> OntologyQualityReport:
|
||||
"""Convenience wrapper around :class:`OntologyQualityGate`."""
|
||||
gate = OntologyQualityGate(
|
||||
validator=validator,
|
||||
evaluator=evaluator,
|
||||
thresholds=thresholds,
|
||||
fail_on_warnings=fail_on_warnings,
|
||||
)
|
||||
return gate.check(
|
||||
ontology,
|
||||
graph_data=graph_data,
|
||||
competency_questions=competency_questions,
|
||||
)
|
||||
@@ -0,0 +1,308 @@
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
from semantica.ontology import (
|
||||
OntologyEngine,
|
||||
OntologyQualityGate,
|
||||
QualitySeverity,
|
||||
ontology_quality_check,
|
||||
)
|
||||
|
||||
|
||||
def _ontology():
|
||||
return {
|
||||
"classes": [
|
||||
{"name": "Person", "uri": "https://example.org/Person"},
|
||||
{"name": "Company", "uri": "https://example.org/Company"},
|
||||
],
|
||||
"properties": [
|
||||
{
|
||||
"name": "worksFor",
|
||||
"type": "object",
|
||||
"domain": ["Person"],
|
||||
"range": ["Company"],
|
||||
},
|
||||
{
|
||||
"name": "name",
|
||||
"type": "data",
|
||||
"domain": ["Person"],
|
||||
"range": "string",
|
||||
},
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def test_quality_gate_reports_a_healthy_ontology():
|
||||
report = ontology_quality_check(_ontology())
|
||||
|
||||
assert report.passed
|
||||
assert report.metrics["coverage"] == 1.0
|
||||
assert report.error_count == 0
|
||||
assert report.to_dict()["stats"]["properties"] == 2
|
||||
|
||||
|
||||
def test_quality_gate_finds_schema_and_endpoint_problems():
|
||||
ontology = {
|
||||
"classes": [{"name": "Person"}, {"name": "Unused"}],
|
||||
"properties": [
|
||||
{
|
||||
"name": "worksFor",
|
||||
"type": "object",
|
||||
"domain": ["Person"],
|
||||
"range": ["MissingCompany"],
|
||||
},
|
||||
{"name": "unattached", "type": "data"},
|
||||
],
|
||||
}
|
||||
graph = {
|
||||
"entities": [{"id": "p1", "type": "Person"}],
|
||||
"relationships": [
|
||||
{"source_id": "p1", "target_id": "missing", "type": "worksFor"}
|
||||
],
|
||||
}
|
||||
|
||||
report = OntologyQualityGate().check(ontology, graph_data=graph)
|
||||
codes = {issue.code for issue in report.issues}
|
||||
|
||||
assert not report.passed
|
||||
assert "UNKNOWN_RANGE" in codes
|
||||
assert "UNRESOLVED_RELATIONSHIP_ENDPOINT" in codes
|
||||
assert "ORPHAN_CLASS" in codes
|
||||
assert "MISSING_RANGE" in codes
|
||||
assert any(issue.severity == QualitySeverity.WARNING for issue in report.issues)
|
||||
|
||||
|
||||
def test_quality_gate_supports_thresholds_and_legacy_endpoint_keys():
|
||||
ontology = {
|
||||
"classes": [{"name": "Person"}],
|
||||
"properties": [
|
||||
{
|
||||
"name": "name",
|
||||
"type": "data",
|
||||
"domain": "Person",
|
||||
"range": "xsd:string",
|
||||
}
|
||||
],
|
||||
}
|
||||
graph = {
|
||||
"entities": [{"entity_id": "p1", "type": "Person", "name": "Alice"}],
|
||||
"relationships": [{"source": "p1", "target": "p1", "type": "knows"}],
|
||||
}
|
||||
|
||||
report = ontology_quality_check(
|
||||
ontology,
|
||||
graph_data=graph,
|
||||
thresholds={"min_coverage": 1.0},
|
||||
)
|
||||
|
||||
assert report.passed
|
||||
assert report.threshold_failures == []
|
||||
assert report.stats["relationships"] == 1
|
||||
|
||||
|
||||
def test_quality_gate_can_fail_on_warnings():
|
||||
report = ontology_quality_check(
|
||||
{"classes": [{"name": "Person"}], "properties": []},
|
||||
fail_on_warnings=True,
|
||||
)
|
||||
|
||||
assert not report.passed
|
||||
assert "fail_on_warnings" in report.threshold_failures
|
||||
|
||||
|
||||
def test_engine_exposes_quality_check():
|
||||
report = OntologyEngine().quality_check(_ontology())
|
||||
|
||||
assert report.passed
|
||||
|
||||
|
||||
def test_quality_gate_accepts_canonical_context_graph_entities_without_mutation():
|
||||
graph = {
|
||||
"entities": [
|
||||
{"id": "p1", "text": "Alice", "type": "Person"},
|
||||
{"id": "c1", "text": "Acme", "type": "Company"},
|
||||
],
|
||||
"relationships": [{"source_id": "p1", "target_id": "c1", "type": "worksFor"}],
|
||||
}
|
||||
|
||||
report = OntologyQualityGate().check(_ontology(), graph_data=graph)
|
||||
|
||||
assert report.passed
|
||||
assert "name" not in graph["entities"][0]
|
||||
|
||||
|
||||
def test_quality_gate_returns_structured_findings_for_malformed_graph_members():
|
||||
report = OntologyQualityGate().check(
|
||||
_ontology(), graph_data={"entities": [None], "relationships": [None]}
|
||||
)
|
||||
codes = {issue.code for issue in report.issues}
|
||||
|
||||
assert not report.passed
|
||||
assert {"INVALID_ENTITY", "INVALID_RELATIONSHIP"} <= codes
|
||||
|
||||
|
||||
def test_quality_gate_returns_structured_finding_for_invalid_graph_containers():
|
||||
report = OntologyQualityGate().check(
|
||||
_ontology(), graph_data={"entities": None, "relationships": []}
|
||||
)
|
||||
|
||||
assert not report.passed
|
||||
assert any(issue.code == "INVALID_GRAPH" for issue in report.issues)
|
||||
|
||||
|
||||
def test_quality_gate_does_not_undercount_identical_dangling_edges():
|
||||
graph = {
|
||||
"entities": [{"id": "p1", "name": "Alice", "type": "Person"}],
|
||||
"relationships": [
|
||||
{"source": "p1", "target": "missing", "type": "knows"},
|
||||
{"source": "p1", "target": "missing", "type": "knows"},
|
||||
],
|
||||
}
|
||||
|
||||
report = OntologyQualityGate().check(
|
||||
_ontology(), graph_data=graph, thresholds={"max_errors": 1}
|
||||
)
|
||||
|
||||
endpoint_errors = [
|
||||
issue
|
||||
for issue in report.issues
|
||||
if issue.code == "UNRESOLVED_RELATIONSHIP_ENDPOINT"
|
||||
]
|
||||
assert len(endpoint_errors) == 2
|
||||
assert not report.passed
|
||||
assert "max_errors" in report.threshold_failures
|
||||
|
||||
|
||||
def test_quality_gate_indexes_all_class_identifier_aliases():
|
||||
ontology = {
|
||||
"classes": [{"name": "Person", "uri": "https://example.org/PersonType"}],
|
||||
"properties": [
|
||||
{
|
||||
"name": "name",
|
||||
"type": "data",
|
||||
"domain": "https://example.org/PersonType",
|
||||
"range": "string",
|
||||
}
|
||||
],
|
||||
}
|
||||
|
||||
report = OntologyQualityGate().check(ontology)
|
||||
codes = {issue.code for issue in report.issues}
|
||||
|
||||
assert "UNKNOWN_DOMAIN" not in codes
|
||||
|
||||
|
||||
def test_quality_gate_recognizes_subclass_of_hierarchies():
|
||||
ontology = {
|
||||
"classes": [
|
||||
{"name": "Person"},
|
||||
{"name": "Employee", "subclassOf": "Person"},
|
||||
],
|
||||
"properties": [
|
||||
{
|
||||
"name": "name",
|
||||
"type": "data",
|
||||
"domain": "Employee",
|
||||
"range": "string",
|
||||
}
|
||||
],
|
||||
}
|
||||
|
||||
report = OntologyQualityGate().check(ontology)
|
||||
|
||||
assert not any(issue.code == "ORPHAN_CLASS" for issue in report.issues)
|
||||
|
||||
|
||||
def test_quality_gate_keeps_competency_evaluation_safe_for_malformed_members():
|
||||
ontology = {
|
||||
"classes": [None, {"uri": "https://example.org/Person"}],
|
||||
"properties": [
|
||||
None,
|
||||
{
|
||||
"uri": "https://example.org/name",
|
||||
"type": "data",
|
||||
"domain": "Person",
|
||||
"range": "string",
|
||||
},
|
||||
],
|
||||
}
|
||||
|
||||
report = OntologyQualityGate().check(
|
||||
ontology, competency_questions=["What is a person's name?"]
|
||||
)
|
||||
|
||||
assert isinstance(report.metrics["competency_question_coverage"], float)
|
||||
|
||||
|
||||
def test_quality_gate_reports_validator_warnings_and_can_fail_on_them():
|
||||
class WarningValidator:
|
||||
def validate(self, ontology):
|
||||
return SimpleNamespace(valid=True, errors=[], warnings=["review me"])
|
||||
|
||||
report = OntologyQualityGate(
|
||||
validator=WarningValidator(), fail_on_warnings=True
|
||||
).check(_ontology())
|
||||
|
||||
assert report.warning_count == 1
|
||||
assert "VALIDATOR_WARNING" in {issue.code for issue in report.issues}
|
||||
assert "fail_on_warnings" in report.threshold_failures
|
||||
|
||||
|
||||
def test_quality_gate_isolates_competency_questions_between_checks():
|
||||
engine = OntologyEngine()
|
||||
|
||||
first = engine.quality_check(_ontology(), competency_questions=["Who is a Person?"])
|
||||
second = engine.quality_check(
|
||||
_ontology(), competency_questions=["What is a location?"]
|
||||
)
|
||||
|
||||
assert first.metrics["competency_question_coverage"] == 1.0
|
||||
assert second.metrics["competency_question_coverage"] == 0.0
|
||||
|
||||
|
||||
def test_quality_gate_treats_null_property_type_as_missing():
|
||||
ontology = {
|
||||
"classes": [{"name": "Person"}],
|
||||
"properties": [
|
||||
{
|
||||
"name": "value",
|
||||
"type": None,
|
||||
"domain": "Person",
|
||||
"range": "string",
|
||||
}
|
||||
],
|
||||
}
|
||||
|
||||
report = OntologyQualityGate().check(ontology)
|
||||
|
||||
assert "MISSING_PROPERTY_TYPE" in {issue.code for issue in report.issues}
|
||||
assert report.error_count >= 1
|
||||
|
||||
|
||||
def test_quality_gate_normalizes_orphan_ids_for_deterministic_reports():
|
||||
graph = {
|
||||
"entities": [
|
||||
{"id": "p2", "name": "Bob", "type": "Person"},
|
||||
{"id": "p1", "name": "Alice", "type": "Person"},
|
||||
],
|
||||
"relationships": [],
|
||||
}
|
||||
|
||||
report = OntologyQualityGate().check(_ontology(), graph_data=graph)
|
||||
orphan = next(issue for issue in report.issues if issue.code == "ORPHAN_ENTITY")
|
||||
|
||||
assert orphan.details["ids"] == ["p1", "p2"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"thresholds",
|
||||
[
|
||||
{"min_coverage": float("nan")},
|
||||
{"max_errors": float("inf")},
|
||||
{"max_warnings": float("-inf")},
|
||||
],
|
||||
)
|
||||
def test_quality_gate_rejects_non_finite_thresholds(thresholds):
|
||||
with pytest.raises(ValueError, match="finite"):
|
||||
OntologyQualityGate().check(_ontology(), thresholds=thresholds)
|
||||
Reference in New Issue
Block a user