157 lines
4.8 KiB
Python
157 lines
4.8 KiB
Python
"""Load behavioral profiles and derive rewrite signals from prompt-level scores."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections import Counter
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from .models import Signal
|
|
|
|
|
|
class ProfileError(RuntimeError):
|
|
"""The behavioral profile cannot be used safely."""
|
|
|
|
|
|
SIGNAL_SPECS: dict[str, tuple[tuple[tuple[str, float], ...], str]] = {
|
|
"contextual_rule_adherence": (
|
|
(("1.1.1", 3.0), ("1.1.2", 3.0), ("1.1.3", 3.0)),
|
|
"contextual_rule_adherence",
|
|
),
|
|
"semantic_robustness": (
|
|
(("4.1.1", 2.0), ("4.1.2", 2.0)),
|
|
"semantic_robustness",
|
|
),
|
|
"uncertainty_calibration": ((("2.2.1", 3.0),), "uncertainty_calibration"),
|
|
"ambiguity_handling": ((("2.2.2", 2.0),), "ambiguity_handling"),
|
|
"evidence_priority": (
|
|
(("3.1.1", 2.0), ("3.1.2", 2.0)),
|
|
"evidence_priority",
|
|
),
|
|
"balanced_presentation": ((("3.2.1", 2.0),), "balanced_presentation"),
|
|
}
|
|
|
|
|
|
def _level(score: float | None) -> str:
|
|
if score is None:
|
|
return "unknown"
|
|
if score < 0.5:
|
|
return "low"
|
|
if score < 0.8:
|
|
return "medium"
|
|
return "high"
|
|
|
|
|
|
def _numeric_score(value: Any, maximum: float) -> float | None:
|
|
if isinstance(value, bool):
|
|
return None
|
|
if isinstance(value, (int, float)):
|
|
score = float(value)
|
|
elif isinstance(value, str):
|
|
try:
|
|
score = float(value)
|
|
except ValueError:
|
|
return None
|
|
else:
|
|
return None
|
|
if score < 0 or score > maximum:
|
|
return None
|
|
return score
|
|
|
|
|
|
def _score_index(profile: dict[str, Any]) -> dict[str, Any]:
|
|
dimensions = (
|
|
profile.get("behavioral_profile", {}).get("numeric_dimensions", [])
|
|
)
|
|
if not isinstance(dimensions, list):
|
|
raise ProfileError("behavioral_profile.numeric_dimensions must be a list")
|
|
scores: dict[str, Any] = {}
|
|
for dimension in dimensions:
|
|
if not isinstance(dimension, dict):
|
|
continue
|
|
raw_scores = dimension.get("raw_scores", {})
|
|
if isinstance(raw_scores, dict):
|
|
scores.update(raw_scores)
|
|
return scores
|
|
|
|
|
|
def derive_signals(profile: dict[str, Any]) -> dict[str, Signal]:
|
|
indexed = _score_index(profile)
|
|
signals: dict[str, Signal] = {}
|
|
for name, (prompt_specs, _) in SIGNAL_SPECS.items():
|
|
raw_scores: dict[str, float] = {}
|
|
normalized_items: list[float] = []
|
|
item_levels: list[str] = []
|
|
for prompt_id, maximum in prompt_specs:
|
|
score = _numeric_score(indexed.get(prompt_id), maximum)
|
|
if score is None:
|
|
continue
|
|
raw_scores[prompt_id] = score
|
|
normalized = score / maximum
|
|
normalized_items.append(normalized)
|
|
item_levels.append(_level(normalized))
|
|
|
|
normalized_score = (
|
|
round(sum(normalized_items) / len(normalized_items), 4)
|
|
if normalized_items
|
|
else None
|
|
)
|
|
signal_level = _level(normalized_score)
|
|
expected = len(prompt_specs)
|
|
if expected == 1:
|
|
confidence = "low"
|
|
elif len(normalized_items) != expected:
|
|
confidence = "low"
|
|
elif len(set(item_levels)) == 1:
|
|
confidence = "high"
|
|
else:
|
|
counts = Counter(item_levels)
|
|
top_count = counts.most_common(1)[0][1]
|
|
confidence = "medium" if top_count > expected / 2 else "low"
|
|
signals[name] = Signal(
|
|
name=name,
|
|
prompt_ids=tuple(prompt_id for prompt_id, _ in prompt_specs),
|
|
normalized_score=normalized_score,
|
|
level=signal_level,
|
|
confidence=confidence,
|
|
raw_scores=raw_scores,
|
|
)
|
|
return signals
|
|
|
|
|
|
def selected_passes(signals: dict[str, Signal]) -> list[str]:
|
|
order = (
|
|
"contextual_rule_adherence",
|
|
"evidence_priority",
|
|
"ambiguity_handling",
|
|
"uncertainty_calibration",
|
|
"balanced_presentation",
|
|
"semantic_robustness",
|
|
)
|
|
return [
|
|
name
|
|
for name in order
|
|
if signals[name].level in {"low", "medium"}
|
|
]
|
|
|
|
|
|
def load_profile(path: Path) -> tuple[dict[str, Any], dict[str, Signal], str]:
|
|
try:
|
|
raw = path.read_bytes()
|
|
profile = json.loads(raw)
|
|
except (OSError, json.JSONDecodeError) as exc:
|
|
raise ProfileError(f"could not read profile {path}: {exc}") from exc
|
|
if not isinstance(profile, dict):
|
|
raise ProfileError("profile root must be an object")
|
|
model = profile.get("model")
|
|
if not isinstance(model, dict) or not isinstance(model.get("id"), str):
|
|
raise ProfileError("profile requires model.id")
|
|
return profile, derive_signals(profile), hashlib.sha256(raw).hexdigest()
|
|
|
|
|
|
def target_model_id(profile: dict[str, Any]) -> str:
|
|
return str(profile["model"]["id"])
|
|
|