Spaces:
Running
Running
File size: 3,024 Bytes
f0347b4 | 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 | """Step 3 of the living loop: deterministic signal -> delta mapping (F2).
Pure, documented, no model calls. Takes the appraisal dict produced by
engine/appraise.py and returns a list of (field, delta, reason) tuples to be
passed to engine/spec_bridge.mutate(). The spec engine (clamp + governance +
audit) has the final say - this module only PROPOSES deltas.
Mapping table (each rule capped to keep per-turn movement small and stable,
per MASTER_CHECKLIST F2 "Estabilidad: limitar magnitud de deltas por turno"):
| signal | field(s) | delta formula | max |
|-------------------------------|-----------------------------------------|--------------------------------------|------|
| sentiment in [-1, 1] | mood.tone, affect.valence | sentiment * SENTIMENT_SCALE | 0.05 |
| engagement in [0, 1] | traits.extraversion | (engagement - 0.5) * EXTRA_SCALE | 0.04 |
| engagement in [0, 1] | traits.openness | (engagement - 0.5) * OPEN_SCALE | 0.03 |
| correction & target != "none" | mapped trait, see TARGET_FIELD | direction * CORRECTION_SCALE | 0.08 |
"""
from __future__ import annotations
SENTIMENT_SCALE = 0.05
EXTRA_SCALE = 0.04
OPEN_SCALE = 0.03
CORRECTION_SCALE = 0.08
TARGET_FIELD = {
"tone": "mood.tone",
"openness": "traits.openness",
"extraversion": "traits.extraversion",
"agreeableness": "traits.agreeableness",
}
_MIN_DELTA = 1e-3
def signals_to_deltas(signals: dict) -> list[tuple[str, float, str]]:
"""Turn an appraisal dict into a list of (field, delta, reason).
Deltas with magnitude below _MIN_DELTA are dropped so a neutral turn
produces no audit-log noise.
"""
deltas: list[tuple[str, float, str]] = []
reason_suffix = str(signals.get("reason", "")).strip()
sentiment = float(signals.get("sentiment", 0.0))
if abs(sentiment) >= _MIN_DELTA:
reason = f"sentiment={sentiment:+.2f}" + (f" ({reason_suffix})" if reason_suffix else "")
deltas.append(("mood.tone", sentiment * SENTIMENT_SCALE, reason))
deltas.append(("affect.valence", sentiment * SENTIMENT_SCALE, reason))
engagement = float(signals.get("engagement", 0.0))
centered = engagement - 0.5
if abs(centered) >= _MIN_DELTA:
reason = f"engagement={engagement:.2f}" + (f" ({reason_suffix})" if reason_suffix else "")
deltas.append(("traits.extraversion", centered * EXTRA_SCALE, reason))
deltas.append(("traits.openness", centered * OPEN_SCALE, reason))
if signals.get("correction") and signals.get("direction", 0) != 0:
field = TARGET_FIELD.get(signals.get("target", "none"))
if field is not None:
direction = signals["direction"]
reason = f"user correction: {reason_suffix or 'requested change'}"
deltas.append((field, direction * CORRECTION_SCALE, reason))
return deltas
|