File size: 5,823 Bytes
4554903 | 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 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 | """BDI drift classification from coherence score changes."""
from dataclasses import dataclass
from typing import List
from .coherence import CoherenceScore
@dataclass(frozen=True)
class DriftClassification:
category: str # healthy_adaptation | stale_beliefs | intention_drift | values_tension
confidence: float
evidence: tuple # frozen needs immutable
recommendation: str
_VALID_CATEGORIES = {
"healthy_adaptation",
"stale_beliefs",
"intention_drift",
"values_tension",
}
class BDIDriftClassifier:
"""Classifies drift from coherence score deltas."""
def __init__(self) -> None:
pass
def classify(
self,
coherence_before: CoherenceScore,
coherence_after: CoherenceScore,
time_delta_days: float,
) -> DriftClassification:
overall_delta = coherence_after.overall - coherence_before.overall
bd_delta = (
coherence_after.belief_desire_alignment
- coherence_before.belief_desire_alignment
)
di_delta = (
coherence_after.desire_intention_alignment
- coherence_before.desire_intention_alignment
)
bi_delta = (
coherence_after.belief_intention_alignment
- coherence_before.belief_intention_alignment
)
new_issues = set(coherence_after.issues) - set(coherence_before.issues)
evidence_items: List[str] = []
# Healthy adaptation: overall improved or stable, no new issues
if overall_delta >= 0 and not new_issues:
evidence_items.append(f"Overall coherence delta: +{overall_delta:.4f}")
return DriftClassification(
category="healthy_adaptation",
confidence=min(1.0, 0.6 + overall_delta),
evidence=tuple(evidence_items),
recommendation="Continue current trajectory. BDI coherence is stable or improving.",
)
# Stale beliefs: belief-related scores dropped + large time delta
if (bd_delta < -0.05 or bi_delta < -0.05) and time_delta_days > 60:
evidence_items.append(f"Belief-desire delta: {bd_delta:.4f}")
evidence_items.append(f"Belief-intention delta: {bi_delta:.4f}")
evidence_items.append(f"Time since last check: {time_delta_days:.0f} days")
return DriftClassification(
category="stale_beliefs",
confidence=min(1.0, 0.5 + abs(bd_delta) + abs(bi_delta)),
evidence=tuple(evidence_items),
recommendation="Schedule a belief review session. Several beliefs may be outdated.",
)
# Intention drift: intention alignment dropped
if di_delta < -0.05 or bi_delta < -0.05:
evidence_items.append(f"Desire-intention delta: {di_delta:.4f}")
evidence_items.append(f"Belief-intention delta: {bi_delta:.4f}")
return DriftClassification(
category="intention_drift",
confidence=min(1.0, 0.5 + abs(di_delta) + abs(bi_delta)),
evidence=tuple(evidence_items),
recommendation="Review active intentions. Some may no longer serve current desires or beliefs.",
)
# Values tension: desire alignment dropped
if bd_delta < -0.05:
evidence_items.append(f"Belief-desire delta: {bd_delta:.4f}")
return DriftClassification(
category="values_tension",
confidence=min(1.0, 0.5 + abs(bd_delta)),
evidence=tuple(evidence_items),
recommendation="Facilitate a values alignment session. Desires may have shifted away from core beliefs.",
)
# Default fallback
evidence_items.append(f"Overall delta: {overall_delta:.4f}")
if new_issues:
evidence_items.append(f"New issues: {len(new_issues)}")
return DriftClassification(
category="values_tension",
confidence=0.4,
evidence=tuple(evidence_items),
recommendation="Review BDI coherence. Minor tensions detected across layers.",
)
def classify_from_events(self, events: List[dict]) -> DriftClassification:
"""Classify drift from a list of drift event dicts."""
if not events:
return DriftClassification(
category="healthy_adaptation",
confidence=0.5,
evidence=("No events provided.",),
recommendation="No drift events to classify.",
)
category_counts: dict = {}
evidence_items: List[str] = []
for event in events:
cat = event.get("category", "values_tension")
category_counts[cat] = category_counts.get(cat, 0) + 1
desc = event.get("description", "")
if desc:
evidence_items.append(desc)
# Pick the most frequent category
dominant = max(category_counts, key=lambda k: category_counts[k])
if dominant not in _VALID_CATEGORIES:
dominant = "values_tension"
total = sum(category_counts.values())
confidence = category_counts[dominant] / total if total else 0.5
recommendations = {
"healthy_adaptation": "Continue current trajectory.",
"stale_beliefs": "Schedule a belief review session.",
"intention_drift": "Review active intentions for relevance.",
"values_tension": "Facilitate a values alignment session.",
}
return DriftClassification(
category=dominant,
confidence=round(confidence, 4),
evidence=tuple(evidence_items[:5]),
recommendation=recommendations.get(dominant, "Review BDI coherence."),
)
|