kvl/scorer.py

"""KVS aggregation — combines five dimension scores into the Knowledge Value Score."""

from __future__ import annotations

WEIGHTS = {
    "novelty": 0.30,
    "retrieval": 0.20,
    "generation": 0.25,
    "attribution": 0.15,
    "demand": 0.10,
}

CLASSIFICATION = [
    (81, "Transformational Value"),
    (61, "High Value"),
    (41, "Moderate Value"),
    (21, "Incremental Value"),
    (0,  "Minimal Value"),
]

RECOMMENDATIONS = {
    "high_novelty":       "Prioritize indexing in AI retrieval systems — contains knowledge not in foundation models.",
    "low_novelty":        "Widely known content; consider whether curation effort is justified.",
    "high_retrieval":     "Well-structured for retrieval — integrate directly into RAG pipelines.",
    "low_retrieval":      "Improve chunking, structure, or metadata to boost retrievability.",
    "high_generation":    "Strongly improves AI-generated answers — valuable for advisory and QA systems.",
    "low_generation":     "Limited generation uplift — content may be too abstract or redundant.",
    "high_attribution":   "Answers are well-grounded in document — high trustworthiness for deployment.",
    "low_attribution":    "Grounding is weak — review document structure and specificity.",
    "high_demand":        "High user demand — wide deployment and open access recommended.",
    "low_demand":         "Narrow or specialized demand — consider targeted distribution.",
    "translate":          "Consider translation into additional languages to broaden impact.",
    "open_access":        "Recommend open access publication to maximize societal return.",
}


def classify(score: int) -> str:
    for threshold, label in CLASSIFICATION:
        if score >= threshold:
            return label
    return "Minimal Value"


def compute(scores: dict[str, int]) -> dict:
    """Compute weighted KVS from dimension scores."""
    kvs = sum(scores[dim] * weight for dim, weight in WEIGHTS.items())
    kvs = round(kvs)

    weighted_contributions = {
        dim: round(scores[dim] * weight, 1)
        for dim, weight in WEIGHTS.items()
    }

    recommendations = _recommend(scores)

    return {
        "kvs": kvs,
        "classification": classify(kvs),
        "dimension_scores": scores,
        "weighted_contributions": weighted_contributions,
        "recommendations": recommendations,
    }


def _recommend(scores: dict[str, int]) -> list[str]:
    recs = []
    if scores.get("novelty", 50) >= 65:
        recs.append(RECOMMENDATIONS["high_novelty"])
    else:
        recs.append(RECOMMENDATIONS["low_novelty"])

    if scores.get("retrieval", 50) < 50:
        recs.append(RECOMMENDATIONS["low_retrieval"])
    else:
        recs.append(RECOMMENDATIONS["high_retrieval"])

    if scores.get("generation", 50) >= 60:
        recs.append(RECOMMENDATIONS["high_generation"])
    else:
        recs.append(RECOMMENDATIONS["low_generation"])

    if scores.get("attribution", 50) < 55:
        recs.append(RECOMMENDATIONS["low_attribution"])
    else:
        recs.append(RECOMMENDATIONS["high_attribution"])

    if scores.get("demand", 50) >= 60:
        recs.append(RECOMMENDATIONS["high_demand"])
        recs.append(RECOMMENDATIONS["open_access"])
    else:
        recs.append(RECOMMENDATIONS["low_demand"])

    if scores.get("novelty", 50) >= 70:
        recs.append(RECOMMENDATIONS["translate"])

    return recs