File size: 3,565 Bytes
921d377 | 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 | """
Rule evaluator — picks the winning rule for a given viewer turn.
Inputs: a list of ``Rule``s + a ``PersonalizationProfile`` + the
runtime state (mood, affinity, metrics). Output: a ``RouterHint``
describing what the rule wants the router to do.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any, Dict, List, Optional
from .profile import PersonalizationProfile
from .rules import Rule
if TYPE_CHECKING:
from ..interaction.state import RuntimeState
@dataclass(frozen=True)
class RouterHint:
"""What the personalization layer is asking the router to do."""
route_to_node: Optional[str] = None
prefer_tone: Optional[str] = None
bump_affinity: float = 0.0
matched_rule_id: Optional[str] = None
_NO_HINT = RouterHint()
def _condition_matches(
rule: Rule, profile: PersonalizationProfile, state: "RuntimeState",
) -> bool:
c = rule.condition
if c.get("role") and profile.role != c.get("role"):
return False
if c.get("level") and profile.level != c.get("level"):
return False
if c.get("language") and profile.language != c.get("language"):
return False
if c.get("country") and profile.country.upper() != str(c.get("country")).upper():
return False
if c.get("has_tag") and c.get("has_tag") not in profile.tags:
return False
if c.get("mood") and state.character_mood != c.get("mood"):
return False
if c.get("min_affinity") is not None:
try:
if state.affinity_score < float(c.get("min_affinity")):
return False
except (TypeError, ValueError):
return False
if c.get("max_affinity") is not None:
try:
if state.affinity_score > float(c.get("max_affinity")):
return False
except (TypeError, ValueError):
return False
metric = c.get("metric")
if isinstance(metric, dict):
scheme = str(metric.get("scheme") or "")
key = str(metric.get("key") or "")
val = state.progress.get(scheme, {}).get(key)
if val is None:
return False
if "min" in metric:
try:
if val < float(metric["min"]):
return False
except (TypeError, ValueError):
return False
if "max" in metric:
try:
if val > float(metric["max"]):
return False
except (TypeError, ValueError):
return False
return True
def evaluate(
rules: List[Rule], profile: PersonalizationProfile, state: "RuntimeState",
) -> RouterHint:
"""Pick the best-matching rule. Lower priority wins ties.
``rules`` should already be filtered to ``enabled=True``. The
evaluator does NOT read from storage — caller assembles the
list.
"""
if not rules:
return _NO_HINT
# Sort by priority ASC (lower = higher priority), then by id for stability.
applicable: List[Rule] = []
for r in rules:
if not r.enabled:
continue
if _condition_matches(r, profile, state):
applicable.append(r)
if not applicable:
return _NO_HINT
applicable.sort(key=lambda r: (r.priority, r.id))
winner = applicable[0]
a = winner.action
return RouterHint(
route_to_node=a.get("route_to_node"),
prefer_tone=a.get("prefer_tone"),
bump_affinity=float(a.get("bump_affinity") or 0.0),
matched_rule_id=winner.id,
)
|