392 lines
13 KiB
Python
392 lines
13 KiB
Python
"""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",
|
|
]
|