29c406d89f
상담자 발화 판정(A)·감정(B)·표현(C) 20문항 질문 세트, 감쇠 없는 이번 턴 반응과 비대칭 기분 전이, 개방도 게이트로 생성 지시를 만들고 ccd.coping_strategy 전달 누락을 고친다. trace v2와 고정 문구 속마음 요약(migration 24, AI 경로 차단 RLS)을 같은 트랜잭션에 저장하고 피드백 정책이 켜진 경우에만 done·TurnResponse·음성 reply·리뷰로 노출한다. 회기 화면 속마음 보기 토글, 리뷰 접힘 블록, 관리자 감정 관측 v2 표시를 추가한다.
288 lines
9.9 KiB
Python
288 lines
9.9 KiB
Python
"""관리자 감정 관측용 Jev 감정 전이 trace 계약 (v1·v2)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import math
|
|
from typing import Literal, Union
|
|
|
|
from pydantic import BaseModel, ConfigDict, Field, model_validator
|
|
|
|
|
|
CLIENT_AFFECT_DIMENSIONS = (
|
|
"anxiety",
|
|
"sadness",
|
|
"anger",
|
|
"shame",
|
|
"guilt",
|
|
"loneliness",
|
|
"relief",
|
|
"hope",
|
|
"trust",
|
|
)
|
|
|
|
ClientAffectDecision = Literal["accepted", "tentative", "held"]
|
|
|
|
|
|
class ClientAffectPolicyV1(BaseModel):
|
|
model_config = ConfigDict(extra="forbid", frozen=True, protected_namespaces=())
|
|
|
|
version: Literal["jev-affect-v1"]
|
|
min_confidence: float = Field(ge=0.0, le=1.0)
|
|
accepted_alpha: float = Field(ge=0.0, le=1.0)
|
|
accepted_cap: float = Field(ge=0.0, le=1.0)
|
|
tentative_alpha: float = Field(ge=0.0, le=1.0)
|
|
tentative_cap: float = Field(ge=0.0, le=1.0)
|
|
tentative_confidence_floor: float = Field(ge=0.0, le=1.0)
|
|
adjacent_probability_threshold: float = Field(ge=0.0, le=1.0)
|
|
|
|
|
|
class ClientAffectContextV1(BaseModel):
|
|
model_config = ConfigDict(extra="forbid", frozen=True, protected_namespaces=())
|
|
|
|
stage: str
|
|
resistance: float = Field(ge=0.0, le=1.0)
|
|
effective_openness: float = Field(ge=0.0, le=1.0)
|
|
rapport_credit: float = Field(ge=0.0)
|
|
|
|
|
|
class ClientAffectDimensionTraceV1(BaseModel):
|
|
model_config = ConfigDict(extra="forbid", frozen=True, protected_namespaces=())
|
|
|
|
key: str
|
|
before: float = Field(ge=0.0, le=1.0)
|
|
target: float | None = Field(default=None, ge=0.0, le=1.0)
|
|
after: float = Field(ge=0.0, le=1.0)
|
|
confidence: float | None = Field(default=None, ge=0.0, le=1.0)
|
|
probabilities: tuple[float, float, float, float, float] | None = None
|
|
decision: ClientAffectDecision
|
|
|
|
@model_validator(mode="after")
|
|
def require_probability_distribution(self) -> "ClientAffectDimensionTraceV1":
|
|
if self.probabilities is None:
|
|
return self
|
|
if any(
|
|
not math.isfinite(value) or value < 0.0 or value > 1.0
|
|
for value in self.probabilities
|
|
):
|
|
raise ValueError("probabilities must be finite values within 0..1")
|
|
return self
|
|
|
|
|
|
class ClientAffectTraceV1(BaseModel):
|
|
model_config = ConfigDict(extra="forbid", frozen=True, protected_namespaces=())
|
|
|
|
schema_version: Literal[1]
|
|
provider: str
|
|
model: str
|
|
latency_ms: int = Field(ge=0)
|
|
input_tokens: int = Field(ge=0)
|
|
output_tokens: int = Field(ge=0)
|
|
cost_usd: float | None = Field(default=None, ge=0.0)
|
|
turn_seq: int = Field(ge=1)
|
|
policy: ClientAffectPolicyV1
|
|
context: ClientAffectContextV1
|
|
dimensions: tuple[ClientAffectDimensionTraceV1, ...] = Field(min_length=9, max_length=9)
|
|
|
|
@model_validator(mode="after")
|
|
def require_fixed_dimension_order(self) -> "ClientAffectTraceV1":
|
|
if tuple(dimension.key for dimension in self.dimensions) != CLIENT_AFFECT_DIMENSIONS:
|
|
raise ValueError("dimensions must use the fixed client affect order")
|
|
return self
|
|
|
|
|
|
AppraisalKind = Literal["noul", "choice"]
|
|
|
|
|
|
class ClientAffectPolicyV2(BaseModel):
|
|
"""jev-affect-v2 비대칭 기분 전이 계수(공학적 기본값, §6.3)."""
|
|
|
|
model_config = ConfigDict(extra="forbid", frozen=True, protected_namespaces=())
|
|
|
|
version: Literal["jev-affect-v2"]
|
|
min_confidence: float = Field(ge=0.0, le=1.0)
|
|
tentative_confidence_floor: float = Field(ge=0.0, le=1.0)
|
|
adjacent_probability_threshold: float = Field(ge=0.0, le=1.0)
|
|
worsening_accepted_alpha: float = Field(ge=0.0, le=1.0)
|
|
worsening_accepted_cap: float = Field(ge=0.0, le=1.0)
|
|
worsening_tentative_alpha: float = Field(ge=0.0, le=1.0)
|
|
worsening_tentative_cap: float = Field(ge=0.0, le=1.0)
|
|
recovery_accepted_alpha: float = Field(ge=0.0, le=1.0)
|
|
recovery_accepted_cap: float = Field(ge=0.0, le=1.0)
|
|
recovery_tentative_alpha: float = Field(ge=0.0, le=1.0)
|
|
recovery_tentative_cap: float = Field(ge=0.0, le=1.0)
|
|
|
|
|
|
class AppraisalQuestionTraceV1(BaseModel):
|
|
"""A층 질문 하나의 판정 원자료(§8.1). 보내지 않은 질문은 항목 자체가 없다."""
|
|
|
|
model_config = ConfigDict(extra="forbid", frozen=True)
|
|
|
|
key: str
|
|
kind: AppraisalKind
|
|
probability: float | None = Field(default=None, ge=0.0, le=1.0)
|
|
choice: str | None = None
|
|
probabilities: dict[str, float] | None = None
|
|
confidence: float | None = Field(default=None, ge=0.0, le=1.0)
|
|
decision: str
|
|
|
|
@model_validator(mode="after")
|
|
def require_kind_matched_fields(self) -> "AppraisalQuestionTraceV1":
|
|
if self.kind == "noul":
|
|
if self.probability is None or self.choice is not None or self.probabilities is not None:
|
|
raise ValueError("noul appraisal trace must carry only probability")
|
|
else:
|
|
if self.choice is None or self.probabilities is None or self.probability is not None:
|
|
raise ValueError("choice appraisal trace must carry choice and probabilities")
|
|
if any(
|
|
not math.isfinite(value) or value < 0.0 or value > 1.0
|
|
for value in self.probabilities.values()
|
|
):
|
|
raise ValueError("probabilities must be finite values within 0..1")
|
|
return self
|
|
|
|
|
|
class ReactionDimensionTraceV1(BaseModel):
|
|
"""이번 턴 반응(§6.2) 9축 고정 순서 trace."""
|
|
|
|
model_config = ConfigDict(extra="forbid", frozen=True)
|
|
|
|
key: str
|
|
value: float | None = Field(default=None, ge=0.0, le=1.0)
|
|
included: bool
|
|
|
|
|
|
class ExpressionChoiceTraceV1(BaseModel):
|
|
"""c_behavior·c_display choice 판정 원자료."""
|
|
|
|
model_config = ConfigDict(extra="forbid", frozen=True)
|
|
|
|
choice: str
|
|
probabilities: dict[str, float]
|
|
confidence: float | None = Field(default=None, ge=0.0, le=1.0)
|
|
decision: str
|
|
|
|
@model_validator(mode="after")
|
|
def require_probability_distribution(self) -> "ExpressionChoiceTraceV1":
|
|
if any(
|
|
not math.isfinite(value) or value < 0.0 or value > 1.0
|
|
for value in self.probabilities.values()
|
|
):
|
|
raise ValueError("probabilities must be finite values within 0..1")
|
|
return self
|
|
|
|
|
|
class ExpressionDiscloseTraceV1(BaseModel):
|
|
"""c_disclose_ready noul 판정 원자료."""
|
|
|
|
model_config = ConfigDict(extra="forbid", frozen=True)
|
|
|
|
probability: float = Field(ge=0.0, le=1.0)
|
|
confidence: float | None = Field(default=None, ge=0.0, le=1.0)
|
|
decision: str
|
|
|
|
|
|
class ExpressionTraceV1(BaseModel):
|
|
"""표현 계획(§6.4) trace — 개방도 게이트 전/후 값을 함께 남긴다."""
|
|
|
|
model_config = ConfigDict(extra="forbid", frozen=True)
|
|
|
|
behavior: ExpressionChoiceTraceV1
|
|
gated_behavior: str
|
|
gate_reason: Literal["openness_closed", "openness_guarded"] | None = None
|
|
stance: Literal["engage", "cautious", "pull_back", "push_back"] | None = None
|
|
display: ExpressionChoiceTraceV1
|
|
disclose_ready: ExpressionDiscloseTraceV1
|
|
hidden_gap: bool
|
|
|
|
|
|
class ClientAffectTraceV2(BaseModel):
|
|
model_config = ConfigDict(extra="forbid", frozen=True, protected_namespaces=())
|
|
|
|
schema_version: Literal[2]
|
|
provider: str
|
|
model: str
|
|
latency_ms: int = Field(ge=0)
|
|
input_tokens: int = Field(ge=0)
|
|
output_tokens: int = Field(ge=0)
|
|
cost_usd: float | None = Field(default=None, ge=0.0)
|
|
turn_seq: int = Field(ge=1)
|
|
policy: ClientAffectPolicyV2
|
|
context: ClientAffectContextV1
|
|
dimensions: tuple[ClientAffectDimensionTraceV1, ...] = Field(min_length=9, max_length=9)
|
|
appraisal: tuple[AppraisalQuestionTraceV1, ...]
|
|
reaction: tuple[ReactionDimensionTraceV1, ...] = Field(min_length=9, max_length=9)
|
|
expression: ExpressionTraceV1
|
|
sore_spot_count: int = Field(ge=0)
|
|
|
|
@model_validator(mode="after")
|
|
def require_fixed_dimension_order(self) -> "ClientAffectTraceV2":
|
|
if tuple(dimension.key for dimension in self.dimensions) != CLIENT_AFFECT_DIMENSIONS:
|
|
raise ValueError("dimensions must use the fixed client affect order")
|
|
if tuple(dimension.key for dimension in self.reaction) != CLIENT_AFFECT_DIMENSIONS:
|
|
raise ValueError("reaction must use the fixed client affect order")
|
|
return self
|
|
|
|
|
|
# OpenAPI discriminator mapping은 키를 문자열로 만들어 생성 타입이 schema_version을 "1"/"2"로
|
|
# 선언한다(실제 JSON은 정수). 판별은 admin_affect._parse_trace가 하므로 일반 Union으로 둔다.
|
|
ClientAffectTrace = Union[ClientAffectTraceV1, ClientAffectTraceV2]
|
|
|
|
|
|
class ClientInnerFeelingV1(BaseModel):
|
|
"""속마음 요약(§8.2)의 감정 한 항목."""
|
|
|
|
model_config = ConfigDict(extra="forbid", frozen=True)
|
|
|
|
label: str
|
|
intensity: str
|
|
|
|
|
|
class ClientInnerStanceV1(BaseModel):
|
|
model_config = ConfigDict(extra="forbid", frozen=True)
|
|
|
|
code: Literal["engage", "cautious", "pull_back", "push_back"]
|
|
label: str
|
|
|
|
|
|
class ClientInnerDisplayV1(BaseModel):
|
|
model_config = ConfigDict(extra="forbid", frozen=True)
|
|
|
|
code: Literal["as_felt", "softened", "covered_by_agreement", "masked"]
|
|
label: str
|
|
|
|
|
|
class ClientInnerReactionV1(BaseModel):
|
|
"""학습자·교수자용 속마음 요약(§8.2). 고정 문구 표에서만 만든다."""
|
|
|
|
model_config = ConfigDict(extra="forbid", frozen=True)
|
|
|
|
schema_version: Literal[1]
|
|
turn_seq: int = Field(ge=1)
|
|
experienced: tuple[str, ...] = Field(max_length=3)
|
|
feelings: tuple[ClientInnerFeelingV1, ...] = Field(max_length=4)
|
|
stance: ClientInnerStanceV1 | None = None
|
|
display: ClientInnerDisplayV1 | None = None
|
|
hidden_gap: bool
|
|
|
|
|
|
__all__ = [
|
|
"AppraisalKind",
|
|
"AppraisalQuestionTraceV1",
|
|
"CLIENT_AFFECT_DIMENSIONS",
|
|
"ClientAffectContextV1",
|
|
"ClientAffectDecision",
|
|
"ClientAffectDimensionTraceV1",
|
|
"ClientAffectPolicyV1",
|
|
"ClientAffectPolicyV2",
|
|
"ClientAffectTrace",
|
|
"ClientAffectTraceV1",
|
|
"ClientAffectTraceV2",
|
|
"ClientInnerDisplayV1",
|
|
"ClientInnerFeelingV1",
|
|
"ClientInnerReactionV1",
|
|
"ClientInnerStanceV1",
|
|
"ExpressionChoiceTraceV1",
|
|
"ExpressionDiscloseTraceV1",
|
|
"ExpressionTraceV1",
|
|
"ReactionDimensionTraceV1",
|
|
]
|