79 lines
2.7 KiB
Python
79 lines
2.7 KiB
Python
"""ECP classification adapter consuming src.classifier.InherenceClassifier."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any, Dict, Optional
|
|
|
|
import jsonschema
|
|
from referencing import Registry, Resource
|
|
|
|
from src.runtime.core.config import create_schema_registry, load_schema
|
|
from src.tools.classifier import InherenceClassifier
|
|
from src.tools.models import ECPSnapshot
|
|
|
|
CANONICAL_ECP_SCHEMA_PATH = (
|
|
Path(__file__).resolve().parent.parent.parent
|
|
/ "tools"
|
|
/ "adapters"
|
|
/ "ecp"
|
|
/ "schemas"
|
|
/ "ecp-profile.schema.json"
|
|
)
|
|
|
|
|
|
def load_ecp_schema_registry() -> Registry:
|
|
registry = create_schema_registry()
|
|
if CANONICAL_ECP_SCHEMA_PATH.exists():
|
|
schema_data = json.loads(CANONICAL_ECP_SCHEMA_PATH.read_text(encoding="utf-8"))
|
|
schema_id = schema_data.get(
|
|
"$id", "https://schemas.aftech.internal/ecp/v1/ecp-profile.schema.json"
|
|
)
|
|
resource = Resource.from_contents(schema_data)
|
|
registry = registry.with_resource(schema_id, resource)
|
|
return registry
|
|
|
|
|
|
def validate_ecp_snapshot(ecp_data: Dict[str, Any]) -> None:
|
|
schema = load_schema("ecp-snapshot.schema.json")
|
|
registry = load_ecp_schema_registry()
|
|
validator = jsonschema.Draft202012Validator(schema, registry=registry)
|
|
errors = list(validator.iter_errors(ecp_data))
|
|
if errors:
|
|
msg = "; ".join([f"{e.json_path}: {e.message}" for e in errors])
|
|
raise ValueError(f"ECP Snapshot schema validation failed: {msg}")
|
|
|
|
|
|
class ECPClassificationAdapter:
|
|
def __init__(self, classifier: Optional[InherenceClassifier] = None):
|
|
self.classifier = classifier or InherenceClassifier()
|
|
|
|
def classify(self, ecp_dict: Dict[str, Any], content_md: str) -> Dict[str, Any]:
|
|
"""Classifies content_md against ecp_dict using InherenceClassifier."""
|
|
# 1. Validate ECP schema
|
|
validate_ecp_snapshot(ecp_dict)
|
|
|
|
# 2. Build model object
|
|
ecp_snapshot = ECPSnapshot.from_dict(ecp_dict)
|
|
|
|
# 3. Invoke classifier
|
|
result = self.classifier.classify(ecp=ecp_snapshot, content_md=content_md)
|
|
|
|
# 4. Extract fields & validate evidences grounding
|
|
decision_val = (
|
|
result.decision.value if hasattr(result.decision, "value") else str(result.decision)
|
|
)
|
|
is_inherent = bool(result.is_inherent)
|
|
confidence = float(getattr(result, "confidence", 1.0))
|
|
rationale = str(getattr(result, "rationale", ""))
|
|
evidences = [str(e) for e in (getattr(result, "evidence", []) or [])]
|
|
|
|
return {
|
|
"category": decision_val,
|
|
"is_inherent": is_inherent,
|
|
"confidence": confidence,
|
|
"rationale": rationale,
|
|
"evidences": evidences,
|
|
}
|