29c406d89f
상담자 발화 판정(A)·감정(B)·표현(C) 20문항 질문 세트, 감쇠 없는 이번 턴 반응과 비대칭 기분 전이, 개방도 게이트로 생성 지시를 만들고 ccd.coping_strategy 전달 누락을 고친다. trace v2와 고정 문구 속마음 요약(migration 24, AI 경로 차단 RLS)을 같은 트랜잭션에 저장하고 피드백 정책이 켜진 경우에만 done·TurnResponse·음성 reply·리뷰로 노출한다. 회기 화면 속마음 보기 토글, 리뷰 접힘 블록, 관리자 감정 관측 v2 표시를 추가한다.
1200 lines
43 KiB
Python
1200 lines
43 KiB
Python
"""Jev 감정·판정 평가 결과를 회기 상태와 생성 프롬프트에 연결하는 순수 함수 (v1·v2)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import math
|
|
from dataclasses import dataclass
|
|
from typing import Any, Iterable, Mapping
|
|
|
|
from ..contracts.client_affect import (
|
|
AppraisalQuestionTraceV1,
|
|
ClientAffectContextV1,
|
|
ClientAffectDimensionTraceV1,
|
|
ClientAffectPolicyV1,
|
|
ClientAffectPolicyV2,
|
|
ClientAffectTraceV1,
|
|
ClientAffectTraceV2,
|
|
ClientInnerDisplayV1,
|
|
ClientInnerFeelingV1,
|
|
ClientInnerReactionV1,
|
|
ClientInnerStanceV1,
|
|
ExpressionChoiceTraceV1,
|
|
ExpressionDiscloseTraceV1,
|
|
ExpressionTraceV1,
|
|
ReactionDimensionTraceV1,
|
|
)
|
|
from . import guardrail
|
|
from .jev_client import (
|
|
AppraisalResult,
|
|
ChoiceJudgment,
|
|
EMOTION_DIMENSIONS,
|
|
NoulJudgment,
|
|
SORE_SPOT_QUESTION_ID,
|
|
)
|
|
|
|
|
|
_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"})
|
|
_AFFECT_POLICY_VERSION = "jev-affect-v1"
|
|
_ACCEPTED_ALPHA = 0.35
|
|
_ACCEPTED_CAP = 0.15
|
|
_TENTATIVE_ALPHA = 0.15
|
|
_TENTATIVE_CAP = 0.075
|
|
_TENTATIVE_CONFIDENCE_FLOOR = 0.35
|
|
_ADJACENT_PROBABILITY_THRESHOLD = 0.8
|
|
_PROBABILITY_SUM_TOLERANCE = 0.025000001
|
|
|
|
# v2 비대칭 기분 전이 계수(§6.3, 공학적 기본값)
|
|
_AFFECT_POLICY_VERSION_V2 = "jev-affect-v2"
|
|
_WORSENING_ACCEPTED_ALPHA = 0.35
|
|
_WORSENING_ACCEPTED_CAP = 0.15
|
|
_WORSENING_TENTATIVE_ALPHA = 0.15
|
|
_WORSENING_TENTATIVE_CAP = 0.075
|
|
_RECOVERY_ACCEPTED_ALPHA = 0.20
|
|
_RECOVERY_ACCEPTED_CAP = 0.08
|
|
_RECOVERY_TENTATIVE_ALPHA = 0.08
|
|
_RECOVERY_TENTATIVE_CAP = 0.04
|
|
|
|
_BIG5_TRAIT_NAMES: dict[str, str] = {
|
|
"O": "openness",
|
|
"C": "conscientiousness",
|
|
"E": "extraversion",
|
|
"A": "agreeableness",
|
|
"N": "neuroticism",
|
|
}
|
|
|
|
_A_LAYER_NOUL_IDS = (
|
|
"a_understood",
|
|
"a_judged",
|
|
"a_autonomy",
|
|
"a_directionless",
|
|
"a_fact_conflict",
|
|
)
|
|
_A_LAYER_CHOICE_IDS = ("a_coping", "a_move", SORE_SPOT_QUESTION_ID)
|
|
|
|
# 7.1 경험 문구 우선순위 — (질문 id, 기대 판정, 문구) 순서대로 참인 것만 고른다.
|
|
_EXPERIENCE_PRIORITY: tuple[tuple[str, str, str], ...] = (
|
|
("a_fact_conflict", "true", "자신의 사정과 다른 전제를 들었다고 느꼈다"),
|
|
("a_sore_spot", "not_none", "건드리고 싶지 않은 부분이 건드려졌다고 느꼈다"),
|
|
("a_judged", "true", "평가받거나 탓을 듣는 것처럼 느꼈다"),
|
|
("a_autonomy", "true", "무엇을 할지 정해 주는 것 같아 압박을 느꼈다"),
|
|
("a_coping", "overwhelming", "제안받은 것이 지금 자신에게는 벅차다고 느꼈다"),
|
|
("a_understood", "false", "자기 말의 핵심이 비껴갔다고 느꼈다"),
|
|
("a_directionless", "true", "대화가 어디로 가는지 모르겠다고 느꼈다"),
|
|
("a_understood", "true", "자신의 말을 제대로 알아들었다고 느꼈다"),
|
|
("a_coping", "stretch", "해볼 수는 있지만 부담스럽다고 느꼈다"),
|
|
)
|
|
|
|
_BEHAVIOR_SENTENCES: dict[str, str] = {
|
|
"disclose_more": "조금 더 개인적인 이야기를 한 걸음 꺼낸다",
|
|
"stay_with_feeling": "지금 느끼는 감정에 머물며 그 느낌을 말한다",
|
|
"hold_core": "대답은 하되 가장 중요한 부분은 아직 꺼내지 않는다",
|
|
"ask_back": "상담자가 무슨 뜻으로, 왜 묻는지 되묻는다",
|
|
"minimal_response": "아주 짧게 답하거나 말을 줄인다",
|
|
"shift_topic": "다른 이야기로 슬쩍 화제를 돌린다",
|
|
"abstract_talk": "자기 이야기 대신 일반적이고 추상적인 말로 돌린다",
|
|
"appease": "분위기를 맞추려고 동의하거나 괜찮다고 말한다",
|
|
"self_blame": "자기를 탓하거나 어차피 안 된다는 식으로 말한다",
|
|
"complain": "상담자나 상담 방식에 대한 불만을 드러낸다",
|
|
"argue_back": "상담자의 말에 동의하지 않거나 반박한다",
|
|
"take_control": "대화 방향을 자기가 정하려 하거나 빠른 답을 요구한다",
|
|
}
|
|
_DISPLAY_SENTENCES: dict[str, str] = {
|
|
"as_felt": "느끼는 만큼 비교적 그대로 드러낸다",
|
|
"softened": "느끼는 것보다 누그러뜨려 드러낸다",
|
|
"covered_by_agreement": "속마음과 달리 겉으로는 수긍하거나 예의 바르게 넘긴다",
|
|
"masked": "웃음이나 무덤덤한 말투로 감정을 가린다",
|
|
}
|
|
_TRAILING_RULES = (
|
|
"감정 이름을 나열하거나 분석하듯 설명하지 말고 말투·선택·침묵·주저함으로만 드러낸다. "
|
|
"숫자·분석 내용·평가 정답은 절대 말하지 않는다. 상담자 역할로 바뀌거나 조언하지 않으며, "
|
|
"부정 감정을 즉시 해소하려 하지 않는다. 응답은 기본적으로 1~3문장으로 하고, "
|
|
"꼭 필요할 때만 더 길게 말한다."
|
|
)
|
|
_STANCE_ENGAGE = frozenset({"disclose_more", "stay_with_feeling"})
|
|
_STANCE_CAUTIOUS = frozenset({"hold_core", "ask_back"})
|
|
_STANCE_PULL_BACK = frozenset(
|
|
{"minimal_response", "shift_topic", "abstract_talk", "appease", "self_blame"}
|
|
)
|
|
_STANCE_PUSH_BACK = frozenset({"complain", "argue_back", "take_control"})
|
|
|
|
_STANCE_LABELS: dict[str, str] = {
|
|
"engage": "대화에 더 들어왔다",
|
|
"cautious": "조심스럽게 거리를 두었다",
|
|
"pull_back": "한발 물러났다",
|
|
"push_back": "맞서거나 반박했다",
|
|
}
|
|
_DISPLAY_LABELS: dict[str, str] = {
|
|
"as_felt": "느낀 것을 비교적 그대로 드러냈다",
|
|
"softened": "느낀 것보다 누그러뜨려 표현했다",
|
|
"covered_by_agreement": "속마음과 달리 겉으로는 수긍하는 말로 덮었다",
|
|
"masked": "웃음이나 무덤덤한 말투로 감정을 가렸다",
|
|
}
|
|
|
|
_GATE_MAP_CLOSED: dict[str, str] = {
|
|
"disclose_more": "minimal_response",
|
|
"stay_with_feeling": "minimal_response",
|
|
"hold_core": "minimal_response",
|
|
"ask_back": "minimal_response",
|
|
}
|
|
_GATE_MAP_GUARDED: dict[str, str] = {"disclose_more": "hold_core"}
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class AffectTransition:
|
|
"""평가 적용 뒤의 영속 정서와 차원별 수용 여부."""
|
|
|
|
affect_state: dict[str, float]
|
|
accepted_dimensions: tuple[str, ...]
|
|
held_dimensions: tuple[str, ...]
|
|
tentative_dimensions: tuple[str, ...] = ()
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ExpressionPlan:
|
|
"""표현 계획(§6.4) — 개방도 게이트 적용 전/후 행동과 태도·드러내는 방식."""
|
|
|
|
behavior: str | None
|
|
gated_behavior: str | None
|
|
gate_reason: str | None
|
|
stance: str | None
|
|
display: str | None
|
|
disclose_ready: bool | str | None
|
|
hidden_gap: bool
|
|
|
|
|
|
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))
|
|
>= _ADJACENT_PROBABILITY_THRESHOLD
|
|
)
|
|
|
|
|
|
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:
|
|
"""v1 관성 전이(legacy·carry 경로 전용). 동작은 바꾸지 않는다.
|
|
|
|
낮은 신뢰도나 잘못된 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 = _ACCEPTED_ALPHA
|
|
cap = _ACCEPTED_CAP
|
|
elif (
|
|
confidence >= _TENTATIVE_CONFIDENCE_FLOOR
|
|
and _tentative_distribution_is_concentrated(estimate.probabilities)
|
|
):
|
|
# confidence는 정답 확률이 아니라 분포 집중도 요약이다.
|
|
alpha = _TENTATIVE_ALPHA
|
|
cap = _TENTATIVE_CAP
|
|
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 transition_mood(
|
|
affect_state: Mapping[str, Any],
|
|
affect_baseline: Mapping[str, Any],
|
|
appraisal: AppraisalResult,
|
|
*,
|
|
min_confidence: float,
|
|
) -> AffectTransition:
|
|
"""v2 비대칭 기분 전이(§6.3). 악화/회복 방향에 따라 다른 계수를 쓴다."""
|
|
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
|
|
is_worsening = score > old if dimension in _NEGATIVE_EMOTIONS else score < old
|
|
if confidence >= threshold:
|
|
alpha = _WORSENING_ACCEPTED_ALPHA if is_worsening else _RECOVERY_ACCEPTED_ALPHA
|
|
cap = _WORSENING_ACCEPTED_CAP if is_worsening else _RECOVERY_ACCEPTED_CAP
|
|
elif (
|
|
confidence >= _TENTATIVE_CONFIDENCE_FLOOR
|
|
and _tentative_distribution_is_concentrated(estimate.probabilities)
|
|
):
|
|
alpha = _WORSENING_TENTATIVE_ALPHA if is_worsening else _RECOVERY_TENTATIVE_ALPHA
|
|
cap = _WORSENING_TENTATIVE_CAP if is_worsening else _RECOVERY_TENTATIVE_CAP
|
|
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 _trace_probabilities(value: Any) -> tuple[float, float, float, float, float] | None:
|
|
if not isinstance(value, tuple) or len(value) != 5:
|
|
return None
|
|
normalized = tuple(_unit_number(item) for item in value)
|
|
if any(item is None for item in normalized):
|
|
return None
|
|
return (
|
|
normalized[0],
|
|
normalized[1],
|
|
normalized[2],
|
|
normalized[3],
|
|
normalized[4],
|
|
)
|
|
|
|
|
|
def build_client_affect_trace(
|
|
*,
|
|
affect_state_before: Mapping[str, Any],
|
|
affect_baseline: Mapping[str, Any],
|
|
affect_state_after: Mapping[str, Any],
|
|
appraisal: AppraisalResult,
|
|
transition: AffectTransition,
|
|
turn_seq: int,
|
|
stage: str,
|
|
resistance: float,
|
|
effective_openness: float,
|
|
rapport_credit: float,
|
|
min_confidence: float,
|
|
) -> ClientAffectTraceV1:
|
|
"""v1 관리자 전용 trace(legacy·carry 경로 전용). 동작은 바꾸지 않는다."""
|
|
before = resolve_emotions(affect_state_before, affect_baseline)
|
|
after = resolve_emotions(affect_state_after, affect_baseline)
|
|
tentative = set(transition.tentative_dimensions)
|
|
accepted = set(transition.accepted_dimensions)
|
|
dimensions: list[ClientAffectDimensionTraceV1] = []
|
|
for key in EMOTION_DIMENSIONS:
|
|
estimate = appraisal.emotions.get(key)
|
|
target = _unit_number(estimate.score) if estimate is not None else None
|
|
confidence = _unit_number(estimate.confidence) if estimate is not None else None
|
|
probabilities = (
|
|
_trace_probabilities(estimate.probabilities) if estimate is not None else None
|
|
)
|
|
if key in tentative:
|
|
decision = "tentative"
|
|
elif key in accepted:
|
|
decision = "accepted"
|
|
else:
|
|
decision = "held"
|
|
dimensions.append(
|
|
ClientAffectDimensionTraceV1(
|
|
key=key,
|
|
before=before[key],
|
|
target=target,
|
|
after=after[key],
|
|
confidence=confidence,
|
|
probabilities=probabilities,
|
|
decision=decision,
|
|
)
|
|
)
|
|
return ClientAffectTraceV1(
|
|
schema_version=1,
|
|
provider=appraisal.provider,
|
|
model=appraisal.model,
|
|
latency_ms=appraisal.latency_ms,
|
|
input_tokens=appraisal.input_tokens,
|
|
output_tokens=appraisal.output_tokens,
|
|
cost_usd=appraisal.cost_usd,
|
|
turn_seq=turn_seq,
|
|
policy=ClientAffectPolicyV1(
|
|
version=_AFFECT_POLICY_VERSION,
|
|
min_confidence=min_confidence,
|
|
accepted_alpha=_ACCEPTED_ALPHA,
|
|
accepted_cap=_ACCEPTED_CAP,
|
|
tentative_alpha=_TENTATIVE_ALPHA,
|
|
tentative_cap=_TENTATIVE_CAP,
|
|
tentative_confidence_floor=_TENTATIVE_CONFIDENCE_FLOOR,
|
|
adjacent_probability_threshold=_ADJACENT_PROBABILITY_THRESHOLD,
|
|
),
|
|
context=ClientAffectContextV1(
|
|
stage=stage,
|
|
resistance=resistance,
|
|
effective_openness=effective_openness,
|
|
rapport_credit=rapport_credit,
|
|
),
|
|
dimensions=tuple(dimensions),
|
|
)
|
|
|
|
|
|
def _appraisal_trace_entries(appraisal: AppraisalResult) -> tuple[AppraisalQuestionTraceV1, ...]:
|
|
"""A층 질문 중 실제로 보낸 것만 판정 trace로 남긴다(§8.1)."""
|
|
entries: list[AppraisalQuestionTraceV1] = []
|
|
for question_id in _A_LAYER_NOUL_IDS:
|
|
judgment = appraisal.noul_judgments.get(question_id)
|
|
if judgment is None:
|
|
continue
|
|
decision = interpret_noul(judgment)
|
|
entries.append(
|
|
AppraisalQuestionTraceV1(
|
|
key=question_id,
|
|
kind="noul",
|
|
probability=judgment.probability,
|
|
confidence=judgment.confidence,
|
|
decision=_noul_decision_label(decision),
|
|
)
|
|
)
|
|
for question_id in _A_LAYER_CHOICE_IDS:
|
|
judgment = appraisal.choice_judgments.get(question_id)
|
|
if judgment is None:
|
|
continue
|
|
entries.append(
|
|
AppraisalQuestionTraceV1(
|
|
key=question_id,
|
|
kind="choice",
|
|
choice=judgment.choice,
|
|
probabilities=dict(judgment.probabilities),
|
|
confidence=judgment.confidence,
|
|
decision=interpret_choice(judgment) or "uncertain",
|
|
)
|
|
)
|
|
return tuple(entries)
|
|
|
|
|
|
def _reaction_trace_entries(
|
|
reaction: Mapping[str, float]
|
|
) -> tuple[ReactionDimensionTraceV1, ...]:
|
|
return tuple(
|
|
ReactionDimensionTraceV1(
|
|
key=dimension,
|
|
value=reaction.get(dimension),
|
|
included=dimension in reaction,
|
|
)
|
|
for dimension in EMOTION_DIMENSIONS
|
|
)
|
|
|
|
|
|
def _expression_choice_trace(judgment: ChoiceJudgment, decision: str | None) -> ExpressionChoiceTraceV1:
|
|
return ExpressionChoiceTraceV1(
|
|
choice=judgment.choice,
|
|
probabilities=dict(judgment.probabilities),
|
|
confidence=judgment.confidence,
|
|
decision=decision or "uncertain",
|
|
)
|
|
|
|
|
|
def _expression_trace(
|
|
appraisal: AppraisalResult, expression: ExpressionPlan
|
|
) -> ExpressionTraceV1:
|
|
behavior_judgment = appraisal.choice_judgments["c_behavior"]
|
|
display_judgment = appraisal.choice_judgments["c_display"]
|
|
disclose_judgment = appraisal.noul_judgments["c_disclose_ready"]
|
|
disclose_decision = interpret_noul(disclose_judgment)
|
|
return ExpressionTraceV1(
|
|
behavior=_expression_choice_trace(behavior_judgment, expression.behavior),
|
|
gated_behavior=expression.gated_behavior or "uncertain",
|
|
gate_reason=expression.gate_reason,
|
|
stance=expression.stance,
|
|
display=_expression_choice_trace(display_judgment, expression.display),
|
|
disclose_ready=ExpressionDiscloseTraceV1(
|
|
probability=disclose_judgment.probability,
|
|
confidence=disclose_judgment.confidence,
|
|
decision=_noul_decision_label(disclose_decision),
|
|
),
|
|
hidden_gap=expression.hidden_gap,
|
|
)
|
|
|
|
|
|
def build_client_affect_trace_v2(
|
|
*,
|
|
affect_state_before: Mapping[str, Any],
|
|
affect_baseline: Mapping[str, Any],
|
|
affect_state_after: Mapping[str, Any],
|
|
appraisal: AppraisalResult,
|
|
transition: AffectTransition,
|
|
expression: ExpressionPlan,
|
|
turn_seq: int,
|
|
stage: str,
|
|
resistance: float,
|
|
effective_openness: float,
|
|
rapport_credit: float,
|
|
min_confidence: float,
|
|
) -> ClientAffectTraceV2:
|
|
"""v2 관리자 전용 trace(§8.1) — 판정·반응·표현 계획 원자료를 함께 남긴다."""
|
|
before = resolve_emotions(affect_state_before, affect_baseline)
|
|
after = resolve_emotions(affect_state_after, affect_baseline)
|
|
tentative = set(transition.tentative_dimensions)
|
|
accepted = set(transition.accepted_dimensions)
|
|
dimensions: list[ClientAffectDimensionTraceV1] = []
|
|
for key in EMOTION_DIMENSIONS:
|
|
estimate = appraisal.emotions.get(key)
|
|
target = _unit_number(estimate.score) if estimate is not None else None
|
|
confidence = _unit_number(estimate.confidence) if estimate is not None else None
|
|
probabilities = (
|
|
_trace_probabilities(estimate.probabilities) if estimate is not None else None
|
|
)
|
|
if key in tentative:
|
|
decision = "tentative"
|
|
elif key in accepted:
|
|
decision = "accepted"
|
|
else:
|
|
decision = "held"
|
|
dimensions.append(
|
|
ClientAffectDimensionTraceV1(
|
|
key=key,
|
|
before=before[key],
|
|
target=target,
|
|
after=after[key],
|
|
confidence=confidence,
|
|
probabilities=probabilities,
|
|
decision=decision,
|
|
)
|
|
)
|
|
reaction = build_reaction(appraisal)
|
|
return ClientAffectTraceV2(
|
|
schema_version=2,
|
|
provider=appraisal.provider,
|
|
model=appraisal.model,
|
|
latency_ms=appraisal.latency_ms,
|
|
input_tokens=appraisal.input_tokens,
|
|
output_tokens=appraisal.output_tokens,
|
|
cost_usd=appraisal.cost_usd,
|
|
turn_seq=turn_seq,
|
|
policy=ClientAffectPolicyV2(
|
|
version=_AFFECT_POLICY_VERSION_V2,
|
|
min_confidence=min_confidence,
|
|
tentative_confidence_floor=_TENTATIVE_CONFIDENCE_FLOOR,
|
|
adjacent_probability_threshold=_ADJACENT_PROBABILITY_THRESHOLD,
|
|
worsening_accepted_alpha=_WORSENING_ACCEPTED_ALPHA,
|
|
worsening_accepted_cap=_WORSENING_ACCEPTED_CAP,
|
|
worsening_tentative_alpha=_WORSENING_TENTATIVE_ALPHA,
|
|
worsening_tentative_cap=_WORSENING_TENTATIVE_CAP,
|
|
recovery_accepted_alpha=_RECOVERY_ACCEPTED_ALPHA,
|
|
recovery_accepted_cap=_RECOVERY_ACCEPTED_CAP,
|
|
recovery_tentative_alpha=_RECOVERY_TENTATIVE_ALPHA,
|
|
recovery_tentative_cap=_RECOVERY_TENTATIVE_CAP,
|
|
),
|
|
context=ClientAffectContextV1(
|
|
stage=stage,
|
|
resistance=resistance,
|
|
effective_openness=effective_openness,
|
|
rapport_credit=rapport_credit,
|
|
),
|
|
dimensions=tuple(dimensions),
|
|
appraisal=_appraisal_trace_entries(appraisal),
|
|
reaction=_reaction_trace_entries(reaction),
|
|
expression=_expression_trace(appraisal, expression),
|
|
sore_spot_count=appraisal.sore_spot_count,
|
|
)
|
|
|
|
|
|
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 _intensity_word(value: float) -> str:
|
|
if value < 0.2:
|
|
return "미약한"
|
|
if value < 0.5:
|
|
return "중간 정도의"
|
|
if value < 0.75:
|
|
return "뚜렷한"
|
|
return "강한"
|
|
|
|
|
|
def render_affect_directive(affect_state: Mapping[str, Any]) -> str:
|
|
"""v1 9축 정서 발화 지시(legacy·carry 경로 전용). 동작은 바꾸지 않는다."""
|
|
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_word(value)} {_EMOTION_LABELS[dimension]}"
|
|
for dimension, value in selected
|
|
)
|
|
return (
|
|
f"현재 반응에는 {rendered}이 함께 배어 있을 수 있다. 상충하는 감정도 동시에 가질 수 있다. "
|
|
"감정 이름을 나열하지 말고, 말투·선택·침묵·주저함으로만 표현한다. "
|
|
"숫자·내부 상태·평가 정답은 절대 말하지 않는다. "
|
|
"상담자 역할로 바뀌거나 조언하지 않으며, 부정 감정을 즉시 해소하려 하지 않는다. "
|
|
"응답은 기본적으로 1~3문장으로 하고, 꼭 필요할 때만 더 길게 말한다."
|
|
)
|
|
|
|
|
|
def _top_n(
|
|
values: Mapping[str, float], count: int, *, min_value: float = 0.2
|
|
) -> list[tuple[str, float]]:
|
|
return sorted(
|
|
((dimension, value) for dimension, value in values.items() if value >= min_value),
|
|
key=lambda item: item[1],
|
|
reverse=True,
|
|
)[:count]
|
|
|
|
|
|
def _select_top_emotions(
|
|
values: Mapping[str, float], *, min_value: float = 0.2, max_count: int = 3
|
|
) -> list[tuple[str, float]]:
|
|
"""반응값 상위 max_count개 + 계열이 한쪽으로 치우치면 반대 계열 1개를 더한다."""
|
|
candidates = sorted(
|
|
((dimension, value) for dimension, value in values.items() if value >= min_value),
|
|
key=lambda item: item[1],
|
|
reverse=True,
|
|
)
|
|
if not candidates:
|
|
return []
|
|
selected = candidates[:max_count]
|
|
selected_dimensions = {dimension for dimension, _ in selected}
|
|
has_negative = bool(selected_dimensions & _NEGATIVE_EMOTIONS)
|
|
has_positive = bool(selected_dimensions & _POSITIVE_EMOTIONS)
|
|
if has_negative != has_positive:
|
|
opposing = _POSITIVE_EMOTIONS if has_negative else _NEGATIVE_EMOTIONS
|
|
candidate = next(
|
|
(
|
|
item
|
|
for item in candidates
|
|
if item[0] in opposing and item[0] not in selected_dimensions
|
|
),
|
|
None,
|
|
)
|
|
if candidate is not None:
|
|
selected.append(candidate)
|
|
return selected
|
|
|
|
|
|
def _temperament_labels(big5: Mapping[str, Any]) -> list[str]:
|
|
"""big5 0.67 이상은 high, 0.33 이하는 low로만 넣고 중간값은 생략한다(§4)."""
|
|
labels: list[str] = []
|
|
for key, name in _BIG5_TRAIT_NAMES.items():
|
|
value = _finite_number(big5.get(key))
|
|
if value is None:
|
|
continue
|
|
if value >= 0.67:
|
|
labels.append(f"high {name}")
|
|
elif value <= 0.33:
|
|
labels.append(f"low {name}")
|
|
return labels
|
|
|
|
|
|
def _openness_word(value: Any) -> str:
|
|
number = _clamp01(_finite_number(value) or 0.0)
|
|
if number < 0.2:
|
|
return "closed"
|
|
if number < 0.4:
|
|
return "guarded"
|
|
if number < 0.65:
|
|
return "partly_open"
|
|
if number < 0.85:
|
|
return "open"
|
|
return "deep"
|
|
|
|
|
|
def _resistance_word(value: Any) -> str:
|
|
number = _clamp01(_finite_number(value) or 0.0)
|
|
if number < 0.34:
|
|
return "low"
|
|
if number < 0.67:
|
|
return "moderate"
|
|
return "high"
|
|
|
|
|
|
def _mood_word(value: Any) -> str:
|
|
number = _clamp01(_finite_number(value) or 0.0)
|
|
if number < 0.1:
|
|
return "absent"
|
|
if number < 0.3:
|
|
return "slight"
|
|
if number < 0.55:
|
|
return "moderate"
|
|
if number < 0.8:
|
|
return "strong"
|
|
return "overwhelming"
|
|
|
|
|
|
def build_appraisal_state(
|
|
*,
|
|
client_profile: Mapping[str, Any],
|
|
affect_state: Mapping[str, Any],
|
|
affect_baseline: Mapping[str, Any],
|
|
stage: 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]:
|
|
"""§4 state v2 — 외부 Jev 경계에 보내는 최소·재마스킹된 synthetic state를 조립한다.
|
|
|
|
숫자 정서 벡터는 보내지 않는다(previous_feelings는 단어 구간으로만).
|
|
"""
|
|
def masked_bounded(value: Any, limit: int) -> str:
|
|
return _bounded_text(
|
|
_mask_text(
|
|
value,
|
|
counselor_identity=counselor_identity,
|
|
client_identity=client_identity,
|
|
),
|
|
limit,
|
|
)
|
|
|
|
def masked(value: Any) -> Any:
|
|
return _masked_value(
|
|
value,
|
|
counselor_identity=counselor_identity,
|
|
client_identity=client_identity,
|
|
)
|
|
|
|
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
|
|
]
|
|
|
|
profile_out: dict[str, Any] = {
|
|
"presenting": masked(client_profile.get("presenting", "")),
|
|
"history": masked(client_profile.get("history", "")),
|
|
}
|
|
core_belief = client_profile.get("core_belief")
|
|
if core_belief:
|
|
profile_out["core_belief"] = masked(core_belief)
|
|
automatic_thought = client_profile.get("automatic_thought")
|
|
if automatic_thought:
|
|
profile_out["automatic_thought"] = masked(automatic_thought)
|
|
coping_strategy = client_profile.get("coping_strategy")
|
|
if coping_strategy:
|
|
profile_out["coping_strategy"] = masked(coping_strategy)
|
|
temperament = _temperament_labels(client_profile.get("big5") or {})
|
|
if temperament:
|
|
profile_out["temperament"] = temperament
|
|
profile_out["sore_spots"] = [
|
|
masked(value)
|
|
for value in (client_profile.get("sore_spots") or [])
|
|
]
|
|
profile_out["forbidden"] = [
|
|
masked(value)
|
|
for value in (client_profile.get("forbidden") or [])
|
|
]
|
|
speech_style = client_profile.get("speech_style")
|
|
if speech_style:
|
|
profile_out["speech_style"] = masked(speech_style)
|
|
|
|
return {
|
|
"counselor_utterance": masked_bounded(counselor_utterance, _RECENT_TURN_TEXT_LIMIT),
|
|
"recent_turns": rendered_recent,
|
|
"client_profile": profile_out,
|
|
"pinned_facts": pinned,
|
|
"recall_summary": masked_bounded(recall_summary, _RECALL_SUMMARY_LIMIT),
|
|
"relationship": {
|
|
"stage": masked(stage) if stage else "",
|
|
"openness": _openness_word(effective_openness),
|
|
"resistance": _resistance_word(resistance),
|
|
},
|
|
"previous_feelings": {
|
|
dimension: _mood_word(value)
|
|
for dimension, value in resolve_emotions(affect_state, affect_baseline).items()
|
|
},
|
|
}
|
|
|
|
|
|
def interpret_noul(judgment: NoulJudgment | None) -> bool | str | None:
|
|
"""§6.1 noul 해석. 질문을 보내지 않았으면 None."""
|
|
if judgment is None:
|
|
return None
|
|
if judgment.probability >= 0.6:
|
|
return True
|
|
if judgment.probability <= 0.4:
|
|
return False
|
|
return "uncertain"
|
|
|
|
|
|
def _noul_decision_label(decision: bool | str | None) -> str:
|
|
if decision is True:
|
|
return "true"
|
|
if decision is False:
|
|
return "false"
|
|
return "uncertain"
|
|
|
|
|
|
def interpret_choice(judgment: ChoiceJudgment | None) -> str | None:
|
|
"""§6.1 choice 해석. 질문을 보내지 않았으면 None."""
|
|
if judgment is None:
|
|
return None
|
|
best_code, best_probability = max(
|
|
judgment.probabilities.items(), key=lambda item: item[1]
|
|
)
|
|
if best_probability >= 0.45:
|
|
return best_code
|
|
return "uncertain"
|
|
|
|
|
|
def build_reaction(appraisal: AppraisalResult) -> dict[str, float]:
|
|
"""§6.2 ① 이번 턴 반응. confidence>=0.35인 축만 감쇠 없이 포함한다."""
|
|
reaction: dict[str, float] = {}
|
|
for dimension in EMOTION_DIMENSIONS:
|
|
estimate = appraisal.emotions.get(dimension)
|
|
if estimate is None:
|
|
continue
|
|
confidence = _unit_number(estimate.confidence)
|
|
if confidence is None or confidence < 0.35:
|
|
continue
|
|
score = _unit_number(estimate.score)
|
|
if score is None:
|
|
continue
|
|
reaction[dimension] = score
|
|
return reaction
|
|
|
|
|
|
def _apply_openness_gate(behavior: str, effective_openness: float) -> tuple[str, str | None]:
|
|
"""§6.4 개방도 게이트. 바뀌었을 때만 gate_reason을 채운다."""
|
|
openness = _clamp01(_finite_number(effective_openness) or 0.0)
|
|
if openness < 0.2:
|
|
gated = _GATE_MAP_CLOSED.get(behavior, behavior)
|
|
return (gated, "openness_closed") if gated != behavior else (behavior, None)
|
|
if openness < 0.4:
|
|
gated = _GATE_MAP_GUARDED.get(behavior, behavior)
|
|
return (gated, "openness_guarded") if gated != behavior else (behavior, None)
|
|
return behavior, None
|
|
|
|
|
|
def _stance_for(behavior: str | None) -> str | None:
|
|
if behavior is None or behavior == "uncertain":
|
|
return None
|
|
if behavior in _STANCE_ENGAGE:
|
|
return "engage"
|
|
if behavior in _STANCE_CAUTIOUS:
|
|
return "cautious"
|
|
if behavior in _STANCE_PULL_BACK:
|
|
return "pull_back"
|
|
if behavior in _STANCE_PUSH_BACK:
|
|
return "push_back"
|
|
return None
|
|
|
|
|
|
def _hidden_gap(display: str | None, reaction: Mapping[str, float]) -> bool:
|
|
if display not in {"covered_by_agreement", "masked"}:
|
|
return False
|
|
return any(reaction.get(dimension, 0.0) >= 0.5 for dimension in _NEGATIVE_EMOTIONS)
|
|
|
|
|
|
def build_expression_plan(
|
|
appraisal: AppraisalResult, *, effective_openness: Any
|
|
) -> ExpressionPlan:
|
|
"""§6.4 ③ 표현 계획을 조립한다."""
|
|
behavior = interpret_choice(appraisal.choice_judgments.get("c_behavior"))
|
|
if behavior is not None and behavior != "uncertain":
|
|
gated_behavior, gate_reason = _apply_openness_gate(behavior, effective_openness)
|
|
else:
|
|
gated_behavior, gate_reason = behavior, None
|
|
display = interpret_choice(appraisal.choice_judgments.get("c_display"))
|
|
disclose_ready = interpret_noul(appraisal.noul_judgments.get("c_disclose_ready"))
|
|
stance = _stance_for(gated_behavior)
|
|
reaction = build_reaction(appraisal)
|
|
return ExpressionPlan(
|
|
behavior=behavior,
|
|
gated_behavior=gated_behavior,
|
|
gate_reason=gate_reason,
|
|
stance=stance,
|
|
display=display,
|
|
disclose_ready=disclose_ready,
|
|
hidden_gap=_hidden_gap(display, reaction),
|
|
)
|
|
|
|
|
|
def _appraisal_decisions(appraisal: AppraisalResult) -> dict[str, Any]:
|
|
decisions: dict[str, Any] = {}
|
|
for question_id in ("a_understood", "a_judged", "a_autonomy", "a_directionless", "a_fact_conflict"):
|
|
decisions[question_id] = interpret_noul(appraisal.noul_judgments.get(question_id))
|
|
for question_id in ("a_coping", "a_move"):
|
|
decisions[question_id] = interpret_choice(appraisal.choice_judgments.get(question_id))
|
|
decisions[SORE_SPOT_QUESTION_ID] = interpret_choice(
|
|
appraisal.choice_judgments.get(SORE_SPOT_QUESTION_ID)
|
|
)
|
|
return decisions
|
|
|
|
|
|
def experienced_phrases(appraisal: AppraisalResult, *, limit: int) -> list[str]:
|
|
"""§7.1 경험 문구를 우선순위대로 최대 limit개 고른다."""
|
|
decisions = _appraisal_decisions(appraisal)
|
|
phrases: list[str] = []
|
|
for question_id, expected, phrase in _EXPERIENCE_PRIORITY:
|
|
value = decisions.get(question_id)
|
|
if value is None:
|
|
continue
|
|
matched = (
|
|
(expected == "true" and value is True)
|
|
or (expected == "false" and value is False)
|
|
or (
|
|
expected == "not_none"
|
|
and isinstance(value, str)
|
|
and value not in ("none", "uncertain")
|
|
)
|
|
or (expected in ("overwhelming", "stretch") and value == expected)
|
|
)
|
|
if matched:
|
|
phrases.append(phrase)
|
|
if len(phrases) >= limit:
|
|
break
|
|
return phrases
|
|
|
|
|
|
def render_affect_directive_v2(
|
|
appraisal: AppraisalResult,
|
|
expression: ExpressionPlan,
|
|
*,
|
|
affect_state_after: Mapping[str, Any],
|
|
affect_baseline: Mapping[str, Any],
|
|
) -> str:
|
|
"""§7 생성 지시 v2. 판정이 uncertain이거나 해당 없으면 그 줄을 생략한다."""
|
|
bullets: list[str] = []
|
|
experienced = experienced_phrases(appraisal, limit=2)
|
|
if experienced:
|
|
bullets.append(
|
|
"이번 상담자 말을 내담자는 이렇게 받아들였다: " + ", ".join(experienced) + "."
|
|
)
|
|
reaction = build_reaction(appraisal)
|
|
top_reaction = _select_top_emotions(reaction)
|
|
if top_reaction:
|
|
bullets.append(
|
|
"지금 속에서 올라온 감정: "
|
|
+ ", ".join(
|
|
f"{_intensity_word(value)} {_EMOTION_LABELS[dimension]}"
|
|
for dimension, value in top_reaction
|
|
)
|
|
+ "."
|
|
)
|
|
mood_values = resolve_emotions(affect_state_after, affect_baseline)
|
|
top_mood = _top_n(mood_values, 2)
|
|
if top_mood and {dimension for dimension, _ in top_mood} != {
|
|
dimension for dimension, _ in top_reaction
|
|
}:
|
|
bullets.append(
|
|
"배경에 깔린 기분: "
|
|
+ ", ".join(
|
|
f"{_intensity_word(value)} {_EMOTION_LABELS[dimension]}"
|
|
for dimension, value in top_mood
|
|
)
|
|
+ "."
|
|
)
|
|
behavior = expression.gated_behavior
|
|
if behavior is not None and behavior != "uncertain":
|
|
bullets.append(f"다음 말의 방향: {_BEHAVIOR_SENTENCES[behavior]}.")
|
|
display = expression.display
|
|
if display is not None and display != "uncertain":
|
|
bullets.append(f"드러내는 방식: {_DISPLAY_SENTENCES[display]}.")
|
|
if not bullets:
|
|
# 표현 계획이 전부 uncertain이면 v1 문장 대신 v2 공통 규칙만 남긴다(v1 문장은
|
|
# render_affect_directive 전용이며 "내부 상태" 문구를 포함해 v2 누설 검사와 충돌한다).
|
|
return "정서 연기 지시:\n- " + _TRAILING_RULES
|
|
bullets.append(_TRAILING_RULES)
|
|
return "정서 연기 지시:\n" + "\n".join(f"- {bullet}" for bullet in bullets)
|
|
|
|
|
|
def build_inner_reaction(
|
|
appraisal: AppraisalResult,
|
|
expression: ExpressionPlan,
|
|
*,
|
|
turn_seq: int,
|
|
) -> ClientInnerReactionV1:
|
|
"""§8.2 속마음 요약. 고정 문구 표에서만 만든다."""
|
|
experienced = tuple(experienced_phrases(appraisal, limit=3))
|
|
reaction = build_reaction(appraisal)
|
|
top_reaction = _select_top_emotions(reaction)
|
|
feelings = tuple(
|
|
ClientInnerFeelingV1(label=_EMOTION_LABELS[dimension], intensity=_intensity_word(value))
|
|
for dimension, value in top_reaction
|
|
)
|
|
stance = (
|
|
ClientInnerStanceV1(code=expression.stance, label=_STANCE_LABELS[expression.stance])
|
|
if expression.stance is not None
|
|
else None
|
|
)
|
|
display = (
|
|
ClientInnerDisplayV1(code=expression.display, label=_DISPLAY_LABELS[expression.display])
|
|
if expression.display is not None and expression.display != "uncertain"
|
|
else None
|
|
)
|
|
return ClientInnerReactionV1(
|
|
schema_version=1,
|
|
turn_seq=turn_seq,
|
|
experienced=experienced,
|
|
feelings=feelings,
|
|
stance=stance,
|
|
display=display,
|
|
hidden_gap=expression.hidden_gap,
|
|
)
|
|
|
|
|
|
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",
|
|
"ExpressionPlan",
|
|
"baseline_emotions",
|
|
"build_appraisal_state",
|
|
"build_client_affect_trace",
|
|
"build_client_affect_trace_v2",
|
|
"build_expression_plan",
|
|
"build_inner_reaction",
|
|
"build_reaction",
|
|
"experienced_phrases",
|
|
"interpret_choice",
|
|
"interpret_noul",
|
|
"public_end_state",
|
|
"render_affect_directive",
|
|
"render_affect_directive_v2",
|
|
"resolve_emotions",
|
|
"transition_emotions",
|
|
"transition_mood",
|
|
]
|