Jev 기반 내담자 감정 상태와 응답 일관성 개선

apps/api/app/services/client_affect.py

@@ -0,0 +1,392 @@
+"""Jev 감정 평가 결과를 회기 상태와 생성 프롬프트에 연결하는 순수 함수."""
+
+from __future__ import annotations
+
+import math
+from dataclasses import dataclass
+from typing import Any, Iterable, Mapping
+
+from . import guardrail
+from .jev_client import AppraisalResult, EMOTION_DIMENSIONS
+
+
+_RECENT_TURN_LIMIT = 12
+_RECENT_TURN_TEXT_LIMIT = 800
+_RECALL_SUMMARY_LIMIT = 1_600
+_AFFECT_BASELINE_KEYS = frozenset(
+    {
+        "negative_affect",
+        "hopelessness",
+        "anhedonia",
+        "sleep",
+        "anxiety",
+        "suicide_ideation_stage",
+        *(f"emotion_{dimension}" for dimension in EMOTION_DIMENSIONS),
+    }
+)
+
+_EMOTION_LABELS = {
+    "anxiety": "불안",
+    "sadness": "슬픔",
+    "anger": "분노",
+    "shame": "수치심",
+    "guilt": "죄책감",
+    "loneliness": "외로움",
+    "relief": "안도감",
+    "hope": "희망",
+    "trust": "신뢰감",
+}
+_NEGATIVE_EMOTIONS = frozenset(
+    {"anxiety", "sadness", "anger", "shame", "guilt", "loneliness"}
+)
+_POSITIVE_EMOTIONS = frozenset({"relief", "hope", "trust"})
+_TENTATIVE_CONFIDENCE_FLOOR = 0.35
+_PROBABILITY_SUM_TOLERANCE = 0.025000001
+
+
+@dataclass(frozen=True, slots=True)
+class AffectTransition:
+    """평가 적용 뒤의 영속 정서와 차원별 수용 여부."""
+
+    affect_state: dict[str, float]
+    accepted_dimensions: tuple[str, ...]
+    held_dimensions: tuple[str, ...]
+    tentative_dimensions: tuple[str, ...] = ()
+
+
+def _finite_number(value: Any) -> float | None:
+    if isinstance(value, bool) or not isinstance(value, (int, float)):
+        return None
+    number = float(value)
+    return number if math.isfinite(number) else None
+
+
+def _clamp01(value: float) -> float:
+    return max(0.0, min(1.0, value))
+
+
+def _unit_number(value: Any) -> float | None:
+    number = _finite_number(value)
+    if number is None or not 0.0 <= number <= 1.0:
+        return None
+    return number
+
+
+def _tentative_distribution_is_concentrated(probabilities: Any) -> bool:
+    """불확실한 score를 잠정 전이에 쓸 만큼 한 구간에 모였는지 확인한다."""
+    if not isinstance(probabilities, tuple) or len(probabilities) != 5:
+        return False
+    values = tuple(_unit_number(value) for value in probabilities)
+    if any(value is None for value in values):
+        return False
+    total = sum(value for value in values if value is not None)
+    if not math.isclose(total, 1.0, abs_tol=_PROBABILITY_SUM_TOLERANCE):
+        return False
+    normalized = tuple(value / total for value in values if value is not None)
+    return max(normalized[index] + normalized[index + 1] for index in range(4)) >= 0.80
+
+
+def _baseline_value(affect_baseline: Mapping[str, Any], key: str) -> float | None:
+    value = _finite_number(affect_baseline.get(key))
+    return _clamp01(value) if value is not None else None
+
+
+def baseline_emotions(affect_baseline: Mapping[str, Any]) -> dict[str, float]:
+    """카드의 기존 임상 기저선을 9축 정서 벡터로 안전하게 변환한다."""
+    baseline = {dimension: 0.0 for dimension in EMOTION_DIMENSIONS}
+    for dimension in EMOTION_DIMENSIONS:
+        explicit = _baseline_value(affect_baseline, f"emotion_{dimension}")
+        if explicit is not None:
+            baseline[dimension] = explicit
+
+    anxiety = _baseline_value(affect_baseline, "anxiety")
+    if _baseline_value(affect_baseline, "emotion_anxiety") is None and anxiety is not None:
+        baseline["anxiety"] = anxiety
+
+    sadness = _baseline_value(affect_baseline, "negative_affect")
+    if _baseline_value(affect_baseline, "emotion_sadness") is None and sadness is not None:
+        baseline["sadness"] = sadness
+
+    hopelessness = _baseline_value(affect_baseline, "hopelessness")
+    if _baseline_value(affect_baseline, "emotion_hope") is None and hopelessness is not None:
+        baseline["hope"] = 1.0 - hopelessness
+    return baseline
+
+
+def resolve_emotions(
+    affect_state: Mapping[str, Any],
+    affect_baseline: Mapping[str, Any],
+) -> dict[str, float]:
+    """기존의 유효한 emotion_* 값을 우선하고 없으면 카드 기저선을 쓴다."""
+    resolved = baseline_emotions(affect_baseline)
+    for dimension in EMOTION_DIMENSIONS:
+        value = _finite_number(affect_state.get(f"emotion_{dimension}"))
+        if value is not None:
+            resolved[dimension] = _clamp01(value)
+    return resolved
+
+
+def transition_emotions(
+    affect_state: Mapping[str, Any],
+    affect_baseline: Mapping[str, Any],
+    appraisal: AppraisalResult,
+    *,
+    min_confidence: float,
+) -> AffectTransition:
+    """신뢰도 게이트를 거친 관성 전이를 계산한다.
+
+    낮은 신뢰도나 잘못된 estimate는 기존 정서를 정확히 유지한다. 기존 임상 affect
+    키는 손대지 않고, 새 emotion_* 키만 회기 상태에 더한다.
+    """
+    updated = dict(affect_state)
+    previous = resolve_emotions(affect_state, affect_baseline)
+    accepted: list[str] = []
+    held: list[str] = []
+    tentative: list[str] = []
+    threshold = _unit_number(min_confidence)
+    if threshold is None:
+        for dimension in EMOTION_DIMENSIONS:
+            updated[f"emotion_{dimension}"] = previous[dimension]
+        return AffectTransition(
+            affect_state=updated,
+            accepted_dimensions=(),
+            held_dimensions=tuple(EMOTION_DIMENSIONS),
+        )
+
+    for dimension in EMOTION_DIMENSIONS:
+        old = previous[dimension]
+        estimate = appraisal.emotions.get(dimension)
+        score = _unit_number(estimate.score) if estimate is not None else None
+        confidence = _unit_number(estimate.confidence) if estimate is not None else None
+        if score is None or confidence is None:
+            updated[f"emotion_{dimension}"] = old
+            held.append(dimension)
+            continue
+        if confidence >= threshold:
+            alpha = 0.35
+            cap = 0.15
+        elif (
+            confidence >= _TENTATIVE_CONFIDENCE_FLOOR
+            and _tentative_distribution_is_concentrated(estimate.probabilities)
+        ):
+            # confidence는 정답 확률이 아니라 분포 집중도 요약이다.
+            alpha = 0.15
+            cap = 0.075
+            tentative.append(dimension)
+        else:
+            updated[f"emotion_{dimension}"] = old
+            held.append(dimension)
+            continue
+        delta = max(-cap, min(cap, alpha * (score - old)))
+        updated[f"emotion_{dimension}"] = _clamp01(old + delta)
+        accepted.append(dimension)
+
+    return AffectTransition(
+        affect_state=updated,
+        accepted_dimensions=tuple(accepted),
+        held_dimensions=tuple(held),
+        tentative_dimensions=tuple(tentative),
+    )
+
+
+def _mask_text(
+    value: Any,
+    *,
+    counselor_identity: str | None,
+    client_identity: str | None,
+) -> str:
+    return guardrail.mask_role_identities(
+        str(value or ""),
+        counselor_identity=counselor_identity,
+        client_identity=client_identity,
+    ).text_masked
+
+
+def _bounded_text(value: Any, limit: int) -> str:
+    text = str(value or "")
+    return text[:limit]
+
+
+def _masked_value(
+    value: Any,
+    *,
+    counselor_identity: str | None,
+    client_identity: str | None,
+) -> Any:
+    if isinstance(value, Mapping):
+        return {
+            _mask_text(
+                key,
+                counselor_identity=counselor_identity,
+                client_identity=client_identity,
+            ): _masked_value(
+                item,
+                counselor_identity=counselor_identity,
+                client_identity=client_identity,
+            )
+            for key, item in value.items()
+        }
+    if isinstance(value, (list, tuple)):
+        return [
+            _masked_value(
+                item,
+                counselor_identity=counselor_identity,
+                client_identity=client_identity,
+            )
+            for item in value
+        ]
+    if isinstance(value, str):
+        return _mask_text(
+            value,
+            counselor_identity=counselor_identity,
+            client_identity=client_identity,
+        )
+    number = _finite_number(value)
+    return number if number is not None else None
+
+
+def render_affect_directive(affect_state: Mapping[str, Any]) -> str:
+    """9축 정서를 내담자 발화 지시로만 렌더한다."""
+    def intensity(value: float) -> str:
+        if value < 0.2:
+            return "미약한"
+        if value < 0.5:
+            return "중간 정도의"
+        if value < 0.75:
+            return "뚜렷한"
+        return "강한"
+
+    meaningful = sorted(
+        (
+            (dimension, _clamp01(value))
+            for dimension in EMOTION_DIMENSIONS
+            if (value := _finite_number(affect_state.get(f"emotion_{dimension}"))) is not None
+            and value >= 0.05
+        ),
+        key=lambda item: item[1],
+        reverse=True,
+    )
+    if not meaningful:
+        return (
+            "현재 감정을 과장하지 말고, 말투와 반응의 결로 자연스럽게 드러낸다. "
+            "숫자·내부 상태·평가 정답은 절대 말하지 않는다. 응답은 기본적으로 1~3문장으로 한다."
+        )
+    selected = meaningful[:3]
+    selected_dimensions = {dimension for dimension, _ in selected}
+    selected_has_negative = bool(selected_dimensions & _NEGATIVE_EMOTIONS)
+    selected_has_positive = bool(selected_dimensions & _POSITIVE_EMOTIONS)
+    if selected_has_negative != selected_has_positive:
+        opposing = _POSITIVE_EMOTIONS if selected_has_negative else _NEGATIVE_EMOTIONS
+        opposing_candidate = next(
+            (
+                item
+                for item in meaningful
+                if item[0] in opposing and item[0] not in selected_dimensions
+            ),
+            None,
+        )
+        if opposing_candidate is not None:
+            selected.append(opposing_candidate)
+    rendered = ", ".join(
+        f"{intensity(value)} {_EMOTION_LABELS[dimension]}"
+        for dimension, value in selected
+    )
+    return (
+        f"현재 반응에는 {rendered}이 함께 배어 있을 수 있다. 상충하는 감정도 동시에 가질 수 있다. "
+        "감정 이름을 나열하지 말고, 말투·선택·침묵·주저함으로만 표현한다. "
+        "숫자·내부 상태·평가 정답은 절대 말하지 않는다. "
+        "상담자 역할로 바뀌거나 조언하지 않으며, 부정 감정을 즉시 해소하려 하지 않는다. "
+        "응답은 기본적으로 1~3문장으로 하고, 꼭 필요할 때만 더 길게 말한다."
+    )
+
+
+def build_appraisal_state(
+    *,
+    affect_baseline: Mapping[str, Any],
+    affect_state: Mapping[str, Any],
+    persona_context: Mapping[str, Any],
+    resistance: Any,
+    effective_openness: Any,
+    counselor_utterance: Any,
+    recall_summary: Any,
+    pinned_facts: Iterable[Any],
+    recent_turns: Iterable[Mapping[str, Any]],
+    counselor_identity: str | None,
+    client_identity: str | None,
+) -> dict[str, Any]:
+    """외부 Jev 경계에 보내는 최소·재마스킹된 synthetic state를 조립한다."""
+    def masked_bounded(value: Any, limit: int) -> str:
+        return _bounded_text(
+            _mask_text(
+                value,
+                counselor_identity=counselor_identity,
+                client_identity=client_identity,
+            ),
+            limit,
+        )
+
+    recent = list(recent_turns)[-_RECENT_TURN_LIMIT:]
+    rendered_recent = [
+        {
+            "speaker": "counselor" if turn.get("speaker") == "counselor" else "client",
+            "text": masked_bounded(turn.get("text", ""), _RECENT_TURN_TEXT_LIMIT),
+        }
+        for turn in recent
+    ]
+    pinned = [
+        _mask_text(
+            value,
+            counselor_identity=counselor_identity,
+            client_identity=client_identity,
+        )
+        for value in pinned_facts
+    ]
+    return {
+        "persona": {
+            "affect_baseline": {
+                key: value
+                for key, value in affect_baseline.items()
+                if key in _AFFECT_BASELINE_KEYS and _finite_number(value) is not None
+            },
+            "context": _masked_value(
+                persona_context,
+                counselor_identity=counselor_identity,
+                client_identity=client_identity,
+            ),
+        },
+        "memory": {
+            "recall_summary": masked_bounded(recall_summary, _RECALL_SUMMARY_LIMIT),
+            "pinned_facts": pinned,
+        },
+        "recent_turns": rendered_recent,
+        "counselor_utterance": masked_bounded(counselor_utterance, _RECENT_TURN_TEXT_LIMIT),
+        "previous_emotions": resolve_emotions(affect_state, affect_baseline),
+        "current_state": {
+            "resistance": _clamp01(_finite_number(resistance) or 0.0),
+            "effective_openness": _clamp01(_finite_number(effective_openness) or 0.0),
+        },
+    }
+
+
+def public_end_state(end_state: Mapping[str, Any]) -> dict[str, Any]:
+    """학습자 응답에는 새 감정 벡터를 숨기고 내부 snapshot은 그대로 보존한다."""
+    public_state = dict(end_state)
+    affect = end_state.get("affect")
+    if isinstance(affect, Mapping):
+        public_state["affect"] = {
+            key: value
+            for key, value in affect.items()
+            if not (isinstance(key, str) and key.startswith("emotion_"))
+        }
+    return public_state
+
+
+__all__ = [
+    "AffectTransition",
+    "baseline_emotions",
+    "build_appraisal_state",
+    "public_end_state",
+    "resolve_emotions",
+    "render_affect_directive",
+    "transition_emotions",
+]