Jev 내담자 평가·표현 v2와 속마음 공개
상담자 발화 판정(A)·감정(B)·표현(C) 20문항 질문 세트, 감쇠 없는 이번 턴 반응과 비대칭 기분 전이, 개방도 게이트로 생성 지시를 만들고 ccd.coping_strategy 전달 누락을 고친다. trace v2와 고정 문구 속마음 요약(migration 24, AI 경로 차단 RLS)을 같은 트랜잭션에 저장하고 피드백 정책이 켜진 경우에만 done·TurnResponse·음성 reply·리뷰로 노출한다. 회기 화면 속마음 보기 토글, 리뷰 접힘 블록, 관리자 감정 관측 v2 표시를 추가한다.
This commit is contained in:
Yun Chan
parent
36cb847e38
commit
29c406d89f
39 changed files with 5726 additions and 343 deletions
|
|
@ -1,4 +1,4 @@
|
|||
"""Jev 감정 평가 결과를 회기 상태와 생성 프롬프트에 연결하는 순수 함수."""
|
||||
"""Jev 감정·판정 평가 결과를 회기 상태와 생성 프롬프트에 연결하는 순수 함수 (v1·v2)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
|
|
@ -7,13 +7,30 @@ 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, EMOTION_DIMENSIONS
|
||||
from .jev_client import (
|
||||
AppraisalResult,
|
||||
ChoiceJudgment,
|
||||
EMOTION_DIMENSIONS,
|
||||
NoulJudgment,
|
||||
SORE_SPOT_QUESTION_ID,
|
||||
)
|
||||
|
||||
|
||||
_RECENT_TURN_LIMIT = 12
|
||||
|
|
@ -55,6 +72,101 @@ _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:
|
||||
|
|
@ -66,6 +178,19 @@ class AffectTransition:
|
|||
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
|
||||
|
|
@ -148,7 +273,7 @@ def transition_emotions(
|
|||
*,
|
||||
min_confidence: float,
|
||||
) -> AffectTransition:
|
||||
"""신뢰도 게이트를 거친 관성 전이를 계산한다.
|
||||
"""v1 관성 전이(legacy·carry 경로 전용). 동작은 바꾸지 않는다.
|
||||
|
||||
낮은 신뢰도나 잘못된 estimate는 기존 정서를 정확히 유지한다. 기존 임상 affect
|
||||
키는 손대지 않고, 새 emotion_* 키만 회기 상태에 더한다.
|
||||
|
|
@ -204,6 +329,65 @@ def transition_emotions(
|
|||
)
|
||||
|
||||
|
||||
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
|
||||
|
|
@ -233,7 +417,7 @@ def build_client_affect_trace(
|
|||
rapport_credit: float,
|
||||
min_confidence: float,
|
||||
) -> ClientAffectTraceV1:
|
||||
"""전이와 같은 입력으로 관리자 전용 trace를 고정 순서로 만든다."""
|
||||
"""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)
|
||||
|
|
@ -292,6 +476,167 @@ def build_client_affect_trace(
|
|||
)
|
||||
|
||||
|
||||
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,
|
||||
*,
|
||||
|
|
@ -348,17 +693,18 @@ def _masked_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 "강한"
|
||||
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))
|
||||
|
|
@ -391,7 +737,7 @@ def render_affect_directive(affect_state: Mapping[str, Any]) -> str:
|
|||
if opposing_candidate is not None:
|
||||
selected.append(opposing_candidate)
|
||||
rendered = ", ".join(
|
||||
f"{intensity(value)} {_EMOTION_LABELS[dimension]}"
|
||||
f"{_intensity_word(value)} {_EMOTION_LABELS[dimension]}"
|
||||
for dimension, value in selected
|
||||
)
|
||||
return (
|
||||
|
|
@ -403,11 +749,101 @@ def render_affect_directive(affect_state: Mapping[str, Any]) -> str:
|
|||
)
|
||||
|
||||
|
||||
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(
|
||||
*,
|
||||
affect_baseline: Mapping[str, Any],
|
||||
client_profile: Mapping[str, Any],
|
||||
affect_state: Mapping[str, Any],
|
||||
persona_context: Mapping[str, Any],
|
||||
affect_baseline: Mapping[str, Any],
|
||||
stage: Any,
|
||||
resistance: Any,
|
||||
effective_openness: Any,
|
||||
counselor_utterance: Any,
|
||||
|
|
@ -417,7 +853,10 @@ def build_appraisal_state(
|
|||
counselor_identity: str | None,
|
||||
client_identity: str | None,
|
||||
) -> dict[str, Any]:
|
||||
"""외부 Jev 경계에 보내는 최소·재마스킹된 synthetic state를 조립한다."""
|
||||
"""§4 state v2 — 외부 Jev 경계에 보내는 최소·재마스킹된 synthetic state를 조립한다.
|
||||
|
||||
숫자 정서 벡터는 보내지 않는다(previous_feelings는 단어 구간으로만).
|
||||
"""
|
||||
def masked_bounded(value: Any, limit: int) -> str:
|
||||
return _bounded_text(
|
||||
_mask_text(
|
||||
|
|
@ -428,6 +867,13 @@ def build_appraisal_state(
|
|||
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 = [
|
||||
{
|
||||
|
|
@ -444,33 +890,281 @@ def build_appraisal_state(
|
|||
)
|
||||
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 {
|
||||
"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),
|
||||
"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)
|
||||
|
|
@ -486,11 +1180,21 @@ def public_end_state(end_state: Mapping[str, Any]) -> dict[str, Any]:
|
|||
|
||||
__all__ = [
|
||||
"AffectTransition",
|
||||
"ExpressionPlan",
|
||||
"baseline_emotions",
|
||||
"build_client_affect_trace",
|
||||
"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",
|
||||
"resolve_emotions",
|
||||
"render_affect_directive",
|
||||
"render_affect_directive_v2",
|
||||
"resolve_emotions",
|
||||
"transition_emotions",
|
||||
"transition_mood",
|
||||
]
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue