"""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", ]