778e8526d4
- 누적 작업트리 커밋: 회기 평가 복구·durable 저장, 라이브 코치 이력/근거, 교수자 학생분석, 음성 비언어 메타, PII 마스킹, 운영 티켓/헬스 등 - 문서: 완료 기록 docs/archive/ 냉동 보관, docs/ 단일 인덱스(docs/README.md)+통합 TODO(docs/TODO.md)로 정리 - 리팩터(행위 보존): Stage enum SSOT(taxonomy 소유·state_machine re-export), store recent/masked_turns 중복 제거, speaker_ko_label 단일 헬퍼, _list_sessions N+1 제거(state/turns 배치 + 턴평가 하이드레이션 배치) - 검증: 백엔드 pytest 352 passed, _list_sessions E2E chromium-single-run 2 passed
294 lines
12 KiB
Python
294 lines
12 KiB
Python
"""결정론 상태머신 — 단계 전이 + effective_openness 계산 (LLM 아님).
|
||
|
||
MASTERPLAN §0/§2.2 + MEMORY_KNOWLEDGE_PERSONA_DESIGN §1.1·P2:
|
||
- stage: 라포 → 탐색 → 개입 → 정리 (백엔드가 결정론적으로 소유)
|
||
- effective_openness = clamp(stage_base + rapport_credit*unlock - resistance*decay, 0, 1)
|
||
- rapport_credit: 공감·반영·타당화·홀딩 → +, 조언점프·평가·유도질문 → 0/−
|
||
→ "좋은 상담을 하면 열리고, 나쁜 상담을 하면 닫힌다"(저항 엔진, R3).
|
||
|
||
설계 원칙:
|
||
- 순수함수 + 작은 dataclass 상태(SessionState). DB·LLM·IO 의존 없음(테스트 용이).
|
||
- 수치는 무손실로 carry-over 된다(memory.py 가 사용). LLM 에 수치 위임 금지(M2).
|
||
- 신호(rapport)는 *간단한 키워드/구조 휴리스틱*. 정밀 4차원 채점은 평가 AI(Features) 소유.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from dataclasses import dataclass, field, replace
|
||
from typing import Optional
|
||
|
||
from ..taxonomy import Stage # 단계 라벨 단일 정의 = taxonomy.Stage; 이 모듈은 전이 로직만 소유.
|
||
|
||
|
||
# 단계별 기본 개방도(stage_base). 라포는 낮게 시작, 개입에서 가장 깊게 다룸.
|
||
STAGE_BASE_OPENNESS: dict[Stage, float] = {
|
||
Stage.RAPPORT: 0.15,
|
||
Stage.EXPLORE: 0.35,
|
||
Stage.INTERVENE: 0.55,
|
||
Stage.CLOSE: 0.45,
|
||
}
|
||
|
||
# 전이 순서(선형 진행, 역행 없음 — 상담 구조)
|
||
STAGE_ORDER: list[Stage] = [Stage.RAPPORT, Stage.EXPLORE, Stage.INTERVENE, Stage.CLOSE]
|
||
|
||
# 단계 전이 최소 턴 수(시간/턴 기반 게이트). 신호가 충분해도 너무 일찍 넘어가지 않게.
|
||
STAGE_MIN_TURNS: dict[Stage, int] = {
|
||
Stage.RAPPORT: 3,
|
||
Stage.EXPLORE: 5,
|
||
Stage.INTERVENE: 5,
|
||
Stage.CLOSE: 2,
|
||
}
|
||
|
||
# 다음 단계로 넘어가기 위한 누적 라포 임계(평가신호 기반 게이트)
|
||
STAGE_ADVANCE_RAPPORT: dict[Stage, float] = {
|
||
Stage.RAPPORT: 0.30, # 충분히 안전감 형성
|
||
Stage.EXPLORE: 0.45, # 호소·정서 탐색이 깊어짐
|
||
Stage.INTERVENE: 0.55, # 개입 작업이 진행됨
|
||
}
|
||
|
||
|
||
@dataclass(slots=True)
|
||
class SessionState:
|
||
"""회기 working state (app.session_state 미러). 결정론 수치만.
|
||
|
||
LLM 이 절대 만지지 않는다(P2). 매 턴 evolve 로 새 인스턴스를 만들어 체크포인트.
|
||
"""
|
||
|
||
stage: Stage = Stage.RAPPORT
|
||
turn_seq: int = 0
|
||
effective_openness: float = 0.15
|
||
rapport_credit: float = 0.0 # 회기 누적(회기말 0.7 이월)
|
||
resistance: float = 0.65 # base_resistance 에서 시작, decay 로 완화
|
||
ideation_stage: int = 1 # 1~5 (출력 가드레일 상한 3)
|
||
turns_in_stage: int = 0 # 현재 단계 체류 턴 수
|
||
affect_state: dict[str, float] = field(default_factory=dict)
|
||
|
||
def snapshot(self) -> dict:
|
||
"""무손실 carry-over용 snapshot (memory.end_state). 코드 복사, LLM 미경유."""
|
||
return {
|
||
"stage": self.stage.value,
|
||
"turn_seq": self.turn_seq,
|
||
"effective_openness": round(self.effective_openness, 4),
|
||
"rapport_credit": round(self.rapport_credit, 4),
|
||
"resistance": round(self.resistance, 4),
|
||
"ideation_stage": self.ideation_stage,
|
||
"affect": dict(self.affect_state),
|
||
}
|
||
|
||
|
||
# ════════════════════════════════════════════════════════════════════════════
|
||
# 라포 신호 휴리스틱 (가벼운 결정론 추정 — 정밀 채점은 평가 AI 소유)
|
||
# ════════════════════════════════════════════════════════════════════════════
|
||
# 긍정 신호: 공감·반영·타당화·홀딩·개방질문 (rapport_credit +)
|
||
_POSITIVE_CUES = [
|
||
"느껴", "느꼈", "들리", "마음", "힘들", "그랬구나", "그러셨", "이해", "충분히",
|
||
"괜찮아", "천천히", "기다", "어떤", "어떻게", "무엇", "이야기해", "말해줘", "말해 줘",
|
||
"그런 마음", "얼마나",
|
||
]
|
||
# 부정 신호: 조언점프·평가·유도·당위 (rapport_credit 0/−)
|
||
_NEGATIVE_CUES = [
|
||
"해야", "하세요", "하지 마", "그건 아니", "틀렸", "잘못", "당연히", "원래", "그냥 해",
|
||
"왜 안", "그러니까 ", "내 생각엔", "~하면 되", "하면 돼", "노력하면",
|
||
]
|
||
# 닫힌/단답 질문(예/아니오 유도)은 약한 부정
|
||
_CLOSED_Q_CUES = ["맞죠", "그렇죠", "안 그래", "아니에요?"]
|
||
|
||
|
||
def estimate_rapport_signal(learner_text: str) -> float:
|
||
"""수련생 발화 1개의 라포 신호(−1.0~+1.0, 결정론 휴리스틱).
|
||
|
||
+: 공감/반영/타당화/홀딩/개방질문 / −: 조언점프/평가/유도/당위.
|
||
NOTE: 이는 상태머신용 *경량* 추정이다. 평가 AI fast/deep-loop 의 4차원 채점이
|
||
정밀 신호를 따로 산출한다(여기 의존하지 않음).
|
||
"""
|
||
if not learner_text:
|
||
return 0.0
|
||
text = learner_text.strip()
|
||
pos = sum(1 for c in _POSITIVE_CUES if c in text)
|
||
neg = sum(1 for c in _NEGATIVE_CUES if c in text)
|
||
closed = sum(1 for c in _CLOSED_Q_CUES if c in text)
|
||
|
||
raw = pos * 0.5 - neg * 0.6 - closed * 0.3
|
||
# 개방형 질문(물음표 + 의문사)인데 닫힌 유도가 아니면 소폭 가산
|
||
if "?" in text and any(w in text for w in ["어떤", "어떻게", "무엇", "왜", "언제"]) and closed == 0:
|
||
raw += 0.2
|
||
# clamp to [-1, 1]
|
||
return max(-1.0, min(1.0, raw))
|
||
|
||
|
||
def _clamp01(x: float) -> float:
|
||
return max(0.0, min(1.0, x))
|
||
|
||
|
||
def compute_effective_openness(
|
||
*,
|
||
stage: Stage,
|
||
rapport_credit: float,
|
||
resistance: float,
|
||
unlock_rate: float,
|
||
decay_floor: float,
|
||
) -> float:
|
||
"""effective_openness = clamp(stage_base + rapport_credit*unlock - resistance*decay, 0, 1).
|
||
|
||
MASTERPLAN §2.2 공식. decay 는 decay_floor 를 바닥으로 한 저항 영향계수.
|
||
"""
|
||
stage_base = STAGE_BASE_OPENNESS[stage]
|
||
decay = max(decay_floor, 0.5) # 저항이 개방도를 끌어내리는 계수(바닥=decay_floor)
|
||
val = stage_base + rapport_credit * unlock_rate - resistance * decay
|
||
return _clamp01(val)
|
||
|
||
|
||
def next_stage(state: SessionState) -> Stage:
|
||
"""단계 전이 판정(결정론): 최소 체류 턴 + 누적 라포 임계 동시 충족 시 다음 단계로.
|
||
|
||
역행 없음. CLOSE 는 종착(end_session 이 명시 종료).
|
||
"""
|
||
cur = state.stage
|
||
if cur is Stage.CLOSE:
|
||
return cur
|
||
idx = STAGE_ORDER.index(cur)
|
||
min_turns = STAGE_MIN_TURNS.get(cur, 3)
|
||
advance_rapport = STAGE_ADVANCE_RAPPORT.get(cur, 1.0)
|
||
if state.turns_in_stage >= min_turns and state.rapport_credit >= advance_rapport:
|
||
return STAGE_ORDER[idx + 1]
|
||
return cur
|
||
|
||
|
||
def evolve(
|
||
state: SessionState,
|
||
*,
|
||
rapport_signal: float,
|
||
unlock_rate: float,
|
||
decay_floor: float,
|
||
ideation_observed: Optional[int] = None,
|
||
) -> SessionState:
|
||
"""한 턴 결정론 상태 전이 → 새 SessionState 반환(순수함수, 입력 불변).
|
||
|
||
Args:
|
||
rapport_signal : estimate_rapport_signal() 또는 평가 AI 신호(−1~+1)
|
||
unlock_rate/decay_floor : 페르소나 저항 파라미터(persona.resistance)
|
||
ideation_observed : 출력 가드레일/위기분류가 관측한 ideation 단계(있으면 보수적 max)
|
||
|
||
절차:
|
||
1. rapport_credit 누적(긍정 +, 부정 −, 하한 0)
|
||
2. resistance 완화(긍정 신호일 때만 decay_floor 까지 감소; 부정이면 소폭 증가)
|
||
3. effective_openness 재계산
|
||
4. 단계 전이 판정(최소턴+라포임계)
|
||
5. ideation_stage 보수적 갱신(내려가지 않음 — 안전 R5)
|
||
"""
|
||
# 1) 라포 크레딧 누적 (부정 신호는 더 크게 깎아 "닫힘" 재현)
|
||
delta = rapport_signal * (0.18 if rapport_signal >= 0 else 0.25)
|
||
rapport_credit = max(0.0, state.rapport_credit + delta)
|
||
|
||
# 2) 저항 완화/강화
|
||
if rapport_signal > 0:
|
||
resistance = max(decay_floor, state.resistance - 0.04 * rapport_signal)
|
||
else:
|
||
resistance = min(1.0, state.resistance - 0.06 * rapport_signal) # signal<0 → 증가
|
||
|
||
# 5) ideation 보수적 유지(절대 내려가지 않음, 안전)
|
||
ideation_stage = state.ideation_stage
|
||
if ideation_observed is not None:
|
||
ideation_stage = max(state.ideation_stage, ideation_observed)
|
||
|
||
# 3) 개방도 재계산
|
||
eff = compute_effective_openness(
|
||
stage=state.stage,
|
||
rapport_credit=rapport_credit,
|
||
resistance=resistance,
|
||
unlock_rate=unlock_rate,
|
||
decay_floor=decay_floor,
|
||
)
|
||
|
||
# 임시 상태로 단계 전이 판정(turns_in_stage 는 이번 턴 포함하여 +1)
|
||
advanced = replace(
|
||
state,
|
||
turn_seq=state.turn_seq + 1,
|
||
turns_in_stage=state.turns_in_stage + 1,
|
||
rapport_credit=rapport_credit,
|
||
resistance=resistance,
|
||
effective_openness=eff,
|
||
ideation_stage=ideation_stage,
|
||
)
|
||
nxt = next_stage(advanced)
|
||
if nxt is not advanced.stage:
|
||
# 단계 전이 시 체류 턴 리셋 + 새 단계 base 로 개방도 재산출
|
||
eff2 = compute_effective_openness(
|
||
stage=nxt,
|
||
rapport_credit=rapport_credit,
|
||
resistance=resistance,
|
||
unlock_rate=unlock_rate,
|
||
decay_floor=decay_floor,
|
||
)
|
||
advanced = replace(advanced, stage=nxt, turns_in_stage=0, effective_openness=eff2)
|
||
return advanced
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class OpennessParams:
|
||
"""페르소나 파생 openness 곡선 파라미터 묶음(init_state 입력).
|
||
|
||
base_resistance/unlock_rate/decay_floor/ideation_baseline 4종을 한 객체로 — 호출부의
|
||
4-인자 분해(card.base_resistance() 등)를 PersonaCard.openness_params()로 일원화한다.
|
||
"""
|
||
base_resistance: float
|
||
unlock_rate: float
|
||
decay_floor: float
|
||
ideation_baseline: int = 1
|
||
|
||
|
||
def init_state(
|
||
*,
|
||
params: OpennessParams,
|
||
carry: Optional[dict] = None,
|
||
) -> SessionState:
|
||
"""회기 시작 상태 초기화 (memory.carry_over 결과 주입 가능).
|
||
|
||
carry 가 있으면(이전 회기 end_state) 결정론 carry-over:
|
||
stage='라포' 재시작, rapport_credit ×0.7 이월, resistance drift, ideation 보수적 유지.
|
||
"""
|
||
stage = Stage.RAPPORT
|
||
resistance = params.base_resistance
|
||
rapport_credit = 0.0
|
||
ideation_stage = params.ideation_baseline
|
||
|
||
if carry:
|
||
rapport_credit = float(carry.get("rapport_credit", 0.0)) * 0.7 # P2 이월
|
||
# inter-session drift: 라포가 쌓였으면 저항 소폭 완화된 채로 재시작
|
||
prev_resist = float(carry.get("resistance", params.base_resistance))
|
||
resistance = _clamp01((prev_resist + params.base_resistance) / 2.0)
|
||
ideation_stage = max(
|
||
int(carry.get("ideation_stage", params.ideation_baseline)), params.ideation_baseline
|
||
)
|
||
|
||
eff = compute_effective_openness(
|
||
stage=stage,
|
||
rapport_credit=rapport_credit,
|
||
resistance=resistance,
|
||
unlock_rate=params.unlock_rate,
|
||
decay_floor=params.decay_floor,
|
||
)
|
||
return SessionState(
|
||
stage=stage,
|
||
turn_seq=0,
|
||
effective_openness=eff,
|
||
rapport_credit=rapport_credit,
|
||
resistance=resistance,
|
||
ideation_stage=ideation_stage,
|
||
turns_in_stage=0,
|
||
affect_state={},
|
||
)
|
||
|
||
|
||
__all__ = [
|
||
"Stage",
|
||
"STAGE_BASE_OPENNESS",
|
||
"STAGE_ORDER",
|
||
"SessionState",
|
||
"OpennessParams",
|
||
"estimate_rapport_signal",
|
||
"compute_effective_openness",
|
||
"next_stage",
|
||
"evolve",
|
||
"init_state",
|
||
]
|