Spaces:
Runtime error
Runtime error
File size: 4,374 Bytes
31e8a9f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 | """Fusion feature contract (PRD §10.1).
This module is the SINGLE source of truth for the 40-dimensional feature vector.
Notebook 05 (training), notebook 06 (calibration), and the production pipeline
all call :func:`build_feature_vector`, so train/serve skew is impossible by
construction. Changing the order or meaning of any feature requires retraining
the ensemble and re-fitting the calibrator.
Imputation of missing metadata happens upstream in Agent 5
(:class:`~app.ml.agents.quality.QualityAssessor`); the PRD's
``build_feature_vector(imputation=...)`` parameter was never actually used.
"""
from __future__ import annotations
from typing import Any, Final
import numpy as np
from app.ml.constants import EARBUD_ASPECTS, SENTIMENT_TO_NUMERIC
FEATURE_NAMES: Final[tuple[str, ...]] = (
# Base signals (3)
"star_rating",
"experience_score",
"domain_score",
# Consistency signals (4)
"star_text_gap",
"absolute_gap",
"star_inflated",
"consistency_score",
# Quality signals (4)
"quality_score",
"helpfulness_score",
"freshness_score",
"verified_purchase",
# Text quality signals (4)
"word_count_norm",
"vocab_richness",
"avg_sentence_length_norm",
"experience_model_uncertainty",
# Aspect detection (10) + aspect sentiment (10)
*(f"aspect_detected__{a}" for a in EARBUD_ASPECTS),
*(f"aspect_sentiment__{a}" for a in EARBUD_ASPECTS),
# Interactions (5)
"star_x_experience",
"abs_star_minus_domain",
"experience_x_quality",
"consistency_x_trust",
"domain_confidence_x_domain_score",
)
N_FEATURES: Final = len(FEATURE_NAMES)
#: Feature-name groups used by the ablation study (PRD §13.2).
FEATURE_GROUPS: Final[dict[str, tuple[str, ...]]] = {
"star": ("star_rating",),
"experience": ("experience_score", "experience_model_uncertainty", "star_x_experience"),
"domain": ("domain_score", "abs_star_minus_domain", "domain_confidence_x_domain_score"),
"consistency": ("star_text_gap", "absolute_gap", "star_inflated", "consistency_score"),
"quality": (
"quality_score",
"helpfulness_score",
"freshness_score",
"verified_purchase",
"word_count_norm",
"vocab_richness",
"avg_sentence_length_norm",
"experience_x_quality",
"consistency_x_trust",
),
"absa": tuple(n for n in FEATURE_NAMES if n.startswith("aspect_")),
}
def build_feature_vector(agent_outputs: dict[str, Any]) -> np.ndarray:
"""Assemble the fusion features from Agents 1–6.
Args:
agent_outputs: Mapping with keys ``preprocessor``, ``experience_extractor``,
``absa``, ``consistency``, ``quality``, ``domain_rating`` (each the
agent's ``process``/``predict`` output) and ``star_rating``.
Returns:
``float32`` array of shape ``(N_FEATURES,)``.
"""
pre = agent_outputs["preprocessor"]
exp = agent_outputs["experience_extractor"]
absa = agent_outputs["absa"]["aspects"]
cons = agent_outputs["consistency"]
qual = agent_outputs["quality"]
dom = agent_outputs["domain_rating"]
star = float(agent_outputs["star_rating"])
feats: list[float] = [
star,
exp["predicted_score"],
dom["domain_score"],
cons["star_text_gap"],
cons["absolute_gap"],
float(cons["star_inflated"]),
cons["consistency_score"],
qual["quality_score"],
qual["helpfulness_score"],
qual["freshness_score"],
float(qual["verified_purchase"]),
pre["word_count"] / 500,
pre["vocab_richness"],
pre["avg_sentence_length"] / 20,
exp["model_uncertainty"],
]
feats += [float(absa[a]["detected"]) for a in EARBUD_ASPECTS]
feats += [float(SENTIMENT_TO_NUMERIC[absa[a]["sentiment"]]) for a in EARBUD_ASPECTS]
feats += [
star * exp["predicted_score"],
abs(star - dom["domain_score"]),
exp["predicted_score"] * qual["quality_score"],
cons["consistency_score"] * qual["trust_weight"],
dom["domain_confidence"] * dom["domain_score"],
]
vector = np.asarray(feats, dtype=np.float32)
if vector.shape != (N_FEATURES,): # guards against silent contract drift
raise AssertionError(f"feature vector has {vector.shape[0]} dims, expected {N_FEATURES}")
return vector
|