150 lines
5.3 KiB
Python
150 lines
5.3 KiB
Python
"""灵核情绪状态机：纯计算（无 IO）。
|
||
|
||
消费辰正 ``time.*`` 事件（time.reunion / time.user_care / time.sentiment），
|
||
派生 ``EmotionState``（mood + intensity）。intensity 向人格版本 ``emotion_baseline`` 指数衰减回归。
|
||
设计见 ``docs/domain/core-domains.md`` §11。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
from typing import Any
|
||
|
||
from pydantic import BaseModel
|
||
|
||
# v1 mood 词表（七种）。v2 路线图见设计 §11.6。
|
||
MOODS: tuple[str, ...] = (
|
||
"neutral",
|
||
"pleased",
|
||
"touched",
|
||
"wistful",
|
||
"concerned",
|
||
"festive",
|
||
"melancholy",
|
||
)
|
||
|
||
MOOD_LABELS_ZH: dict[str, str] = {
|
||
"neutral": "平静",
|
||
"pleased": "愉悦",
|
||
"touched": "感动",
|
||
"wistful": "怅然怀念",
|
||
"concerned": "关切",
|
||
"festive": "喜庆",
|
||
"melancholy": "惆怅",
|
||
}
|
||
|
||
# 衰减半衰期（对话回合）：intensity 向 baseline 走一半所需的回合数。
|
||
DECAY_HALFLIFE_TURNS: float = 6.0
|
||
# intensity 衰减至此阈值以下时，mood 回 baseline.mood。
|
||
NEUTRAL_THRESHOLD: float = 0.15
|
||
|
||
# time.sentiment 无 payload 强度字段，使用固定中值。
|
||
_SENTIMENT_DEFAULT_DELTA: float = 0.5
|
||
|
||
# time.sentiment trigger_type → mood（late_night → melancholy；其余 → wistful）。
|
||
_SENTIMENT_TRIGGER_MOOD: dict[str, str] = {
|
||
"late_night": "melancholy",
|
||
}
|
||
|
||
|
||
class EmotionState(BaseModel):
|
||
"""情绪状态：离散 mood（``MOODS`` 枚举）+ intensity（0-1 标量）。"""
|
||
|
||
mood: str = "neutral"
|
||
intensity: float = 0.2
|
||
|
||
|
||
def parse_baseline(value: dict[str, Any] | str | None) -> EmotionState:
|
||
"""从人格版本 ``emotion_baseline``（dict / JSON str / None）解析 baseline。
|
||
|
||
容错：非法 JSON / 非 dict / 未知 mood / 越界 intensity → 回退默认（neutral 0.2）。
|
||
"""
|
||
if value is None:
|
||
return EmotionState()
|
||
if isinstance(value, str):
|
||
try:
|
||
value = json.loads(value)
|
||
except (json.JSONDecodeError, TypeError):
|
||
return EmotionState()
|
||
if not isinstance(value, dict):
|
||
return EmotionState()
|
||
mood = value.get("mood", "neutral")
|
||
if mood not in MOODS:
|
||
mood = "neutral"
|
||
try:
|
||
intensity = float(value.get("intensity", 0.2))
|
||
except (TypeError, ValueError):
|
||
intensity = 0.2
|
||
intensity = max(0.0, min(1.0, intensity))
|
||
return EmotionState(mood=mood, intensity=intensity)
|
||
|
||
|
||
def map_signal_to_mood(event_type: str, payload: dict[str, Any]) -> str | None:
|
||
"""辰正 time.* 事件 → mood 映射。无映射的事件类型返回 None。"""
|
||
if event_type == "time.reunion":
|
||
return "touched"
|
||
if event_type == "time.user_care":
|
||
if payload.get("care_kind") == "festival_lonely":
|
||
return "melancholy"
|
||
return "concerned"
|
||
if event_type == "time.sentiment":
|
||
trigger = payload.get("trigger_type", "")
|
||
return _SENTIMENT_TRIGGER_MOOD.get(trigger, "wistful")
|
||
return None
|
||
|
||
|
||
def signal_intensity_delta(event_type: str, payload: dict[str, Any]) -> float:
|
||
"""从事件 payload 提取 intensity delta（复用辰正既有强度，不重算）。"""
|
||
if event_type == "time.reunion":
|
||
return float(payload.get("reunion_intensity", 0.5))
|
||
if event_type == "time.user_care":
|
||
return float(payload.get("care_suggest", 0.5))
|
||
if event_type == "time.sentiment":
|
||
return _SENTIMENT_DEFAULT_DELTA
|
||
return 0.0
|
||
|
||
|
||
def apply_signal(
|
||
state: EmotionState,
|
||
event_type: str,
|
||
payload: dict[str, Any],
|
||
baseline: EmotionState,
|
||
) -> EmotionState:
|
||
"""事件驱动状态转移。
|
||
|
||
- mood 变化时：intensity = ``baseline.intensity + delta``（新情绪从基线起跳）。
|
||
- mood 不变时：intensity = ``state.intensity + delta``（累积）。
|
||
- clamp 至 [0, 1]。
|
||
- 无映射的事件（``map_signal_to_mood`` 返回 None）原样返回 state。
|
||
"""
|
||
new_mood = map_signal_to_mood(event_type, payload)
|
||
if new_mood is None:
|
||
return state
|
||
delta = signal_intensity_delta(event_type, payload)
|
||
base = state.intensity if new_mood == state.mood else baseline.intensity
|
||
intensity = max(0.0, min(1.0, base + delta))
|
||
return EmotionState(mood=new_mood, intensity=round(intensity, 3))
|
||
|
||
|
||
def decay(state: EmotionState, baseline: EmotionState, elapsed_turns: float) -> EmotionState:
|
||
"""intensity 向 baseline 指数衰减；与 baseline 的差距收敛到阈值内时 mood 回 baseline.mood。
|
||
|
||
``elapsed_turns <= 0`` 时原样返回 state（无衰减）。
|
||
判断用相对差距（``decayed - baseline.intensity``）而非绝对阈值——否则当
|
||
baseline.intensity 本身 > 阈值时（如默认 neutral 0.2），mood 永远不会回归。
|
||
"""
|
||
if elapsed_turns <= 0:
|
||
return state
|
||
factor = 0.5 ** (elapsed_turns / DECAY_HALFLIFE_TURNS)
|
||
decayed = baseline.intensity + (state.intensity - baseline.intensity) * factor
|
||
if decayed - baseline.intensity <= NEUTRAL_THRESHOLD:
|
||
return EmotionState(mood=baseline.mood, intensity=baseline.intensity)
|
||
return EmotionState(mood=state.mood, intensity=round(decayed, 3))
|
||
|
||
|
||
def to_prompt_text(state: EmotionState) -> str:
|
||
"""注入对话 system prompt 的情绪摘要（中文 label + 强度百分比）。"""
|
||
label = MOOD_LABELS_ZH.get(state.mood, state.mood)
|
||
pct = round(state.intensity * 100)
|
||
return f"[当前情绪] {label}（强度 {pct}%）"
|