945 lines
39 KiB
Python
945 lines
39 KiB
Python
"""평가 AI(슈퍼바이저/교수 AI) — 학습자 발화 2-loop 평가. 한신대 AI 3종 중 ③.
|
||
|
||
MASTERPLAN §2.3 (평가 AI 2-tier 루프):
|
||
- fast-loop : 턴 직후 가벼운 4차원 태깅(기법/내담자상태 읽기/적절성 신호/의도이탈 여부).
|
||
콜드스타트 지연·비용 통제. tier='feedback' Opus 라우팅(게이트웨이가 최종 결정).
|
||
- deep-loop : 단계전환/회기말 정밀(기법 분포·잘한 순간·개선점 최대3·슈퍼바이저 rationale/critique).
|
||
전체 회기 + 골든라벨 후보 enum 으로 0-5 채점 근거 + 대안 발화.
|
||
|
||
설계 원칙 (소유권 분리):
|
||
- 평가 AI 는 *전부 봐도 된다*(CCD/정답/상태수치 포함). 비노출은 client AI 쪽 책임이고,
|
||
학습자에겐 RBAC×AIView 로 차단(deps.AIView.EVALUATOR). 이 모듈은 평가 신호만 산출한다.
|
||
- 정답 라벨 enum 은 taxonomy.py 가 단일 원천(SoT). 프롬프트는 TECHNIQUE_KO/CLIENT_STATE_KO 를
|
||
후보로 제시하고 *근거(rationale)* 를 요구한다. LLM 출력은 enum 으로 안전 파싱(미지값은 버림).
|
||
- intent_deviation('의도와 다른 부분')은 1급 시민 → SupervisorComment(critique) 형식과 정합:
|
||
{dimension, expected, actual, severity}.
|
||
- engine_client.generate(structured_schema=...) 로 LLM 평가. 엔진 장애·파싱
|
||
실패는 *비치명적* — orchestrator 의 eval_hook 가 None 으로 흡수(상담 루프를 막지 않음).
|
||
- 결과는 pydantic 모델로 반환. orchestrator 가 주입형으로 부르는 async 함수
|
||
evaluate_turn(...) / evaluate_session(...) 을 export.
|
||
|
||
⚠️ taxonomy / engine_client / orchestrator / state_machine / persona 는 *읽기 전용* 의존이다.
|
||
이 모듈만 평가 로직을 소유한다(라우트는 routes/eval.py).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import time
|
||
from collections import OrderedDict
|
||
from typing import TYPE_CHECKING, Any, Optional
|
||
|
||
from pydantic import BaseModel, Field
|
||
|
||
from ..config import settings
|
||
from ..contracts.engine_gateway import (
|
||
ENGINE_GATEWAY_DEFAULT_MODEL_SENTINEL,
|
||
structured_payload_from_response,
|
||
)
|
||
from ..engine_client import (
|
||
EngineClient,
|
||
EngineError,
|
||
EngineMessage,
|
||
GenerateRequest,
|
||
)
|
||
from ..taxonomy import (
|
||
CLIENT_STATE_KO,
|
||
TECHNIQUE_CATEGORY,
|
||
TECHNIQUE_KO,
|
||
ClientState,
|
||
Technique,
|
||
TechniqueCategory,
|
||
speaker_ko_label,
|
||
)
|
||
from . import guardrail
|
||
from .evaluation_contract import APPROPRIATENESS_VALUES, Appropriateness
|
||
from .llm_audit import LlmAuditHook, generate_with_audit
|
||
|
||
if TYPE_CHECKING: # 런타임 import 회피(순환·소유권 경계). 타입 힌트 전용.
|
||
from .orchestrator import TurnContext
|
||
|
||
|
||
# ════════════════════════════════════════════════════════════════════════════
|
||
# 0. enum 역인덱스 (LLM 한글/코드 출력 → taxonomy enum 안전 파싱)
|
||
# ════════════════════════════════════════════════════════════════════════════
|
||
# LLM 은 후보로 한글 라벨을 받지만, 코드값(value)을 돌려줄 수도 있어 둘 다 받는다.
|
||
_TECHNIQUE_BY_KO: dict[str, Technique] = {ko: t for t, ko in TECHNIQUE_KO.items()}
|
||
_TECHNIQUE_BY_CODE: dict[str, Technique] = {t.value: t for t in Technique}
|
||
_CLIENT_STATE_BY_KO: dict[str, ClientState] = {
|
||
ko: s for s, ko in CLIENT_STATE_KO.items()
|
||
}
|
||
_CLIENT_STATE_BY_CODE: dict[str, ClientState] = {s.value: s for s in ClientState}
|
||
|
||
# 적절성 신호 — fast-loop 의 경량 판단(상태머신 라포 추정과 별개 차원).
|
||
# 의도이탈 심각도 (taxonomy.SupervisorComment.severity 와 동일 어휘).
|
||
_SEVERITY = ("minor", "moderate", "major")
|
||
|
||
_EVALUATOR_CACHE_VERSION = "evaluator-semantic-cache-v1"
|
||
_EVALUATOR_CACHE: "OrderedDict[str, tuple[float, dict[str, Any]]]" = OrderedDict()
|
||
_EVALUATOR_CACHE_STATS = {
|
||
"hits": 0,
|
||
"misses": 0,
|
||
"stores": 0,
|
||
"evictions": 0,
|
||
}
|
||
|
||
|
||
def _configured_model(value: str | None) -> str | None:
|
||
model = (value or "").strip()
|
||
return model or None
|
||
|
||
|
||
def clear_evaluator_semantic_cache() -> None:
|
||
"""Clear in-process evaluator cache and counters. Test/support hook only."""
|
||
_EVALUATOR_CACHE.clear()
|
||
for key in _EVALUATOR_CACHE_STATS:
|
||
_EVALUATOR_CACHE_STATS[key] = 0
|
||
|
||
|
||
def evaluator_semantic_cache_stats() -> dict[str, int]:
|
||
"""Return in-process evaluator cache counters without exposing keys."""
|
||
stats = dict(_EVALUATOR_CACHE_STATS)
|
||
stats["entries"] = len(_EVALUATOR_CACHE)
|
||
return stats
|
||
|
||
|
||
def _semantic_cache_enabled() -> bool:
|
||
return (
|
||
bool(settings.evaluator_semantic_cache_enabled)
|
||
and settings.evaluator_semantic_cache_ttl_seconds > 0
|
||
and settings.evaluator_semantic_cache_max_entries > 0
|
||
)
|
||
|
||
|
||
def _canonical_json(value: Any) -> str:
|
||
return json.dumps(
|
||
value,
|
||
ensure_ascii=False,
|
||
sort_keys=True,
|
||
separators=(",", ":"),
|
||
)
|
||
|
||
|
||
def _evaluator_cache_key(req: GenerateRequest) -> str:
|
||
payload = {
|
||
"version": _EVALUATOR_CACHE_VERSION,
|
||
"ai_role": req.ai_role,
|
||
"messages": [m.model_dump() for m in req.messages],
|
||
"model": req.model or ENGINE_GATEWAY_DEFAULT_MODEL_SENTINEL,
|
||
"max_tokens": req.max_tokens,
|
||
"temperature": req.temperature,
|
||
"structured_schema": req.structured_schema,
|
||
"session_id": req.session_id,
|
||
"metadata": req.metadata,
|
||
}
|
||
return hashlib.sha256(_canonical_json(payload).encode("utf-8")).hexdigest()
|
||
|
||
|
||
def _evaluator_cache_get(cache_key: str) -> Optional[dict[str, Any]]:
|
||
if not _semantic_cache_enabled():
|
||
return None
|
||
now = time.monotonic()
|
||
entry = _EVALUATOR_CACHE.get(cache_key)
|
||
if entry is None:
|
||
_EVALUATOR_CACHE_STATS["misses"] += 1
|
||
return None
|
||
expires_at, value = entry
|
||
if expires_at <= now:
|
||
_EVALUATOR_CACHE.pop(cache_key, None)
|
||
_EVALUATOR_CACHE_STATS["evictions"] += 1
|
||
_EVALUATOR_CACHE_STATS["misses"] += 1
|
||
return None
|
||
_EVALUATOR_CACHE.move_to_end(cache_key)
|
||
_EVALUATOR_CACHE_STATS["hits"] += 1
|
||
return json.loads(_canonical_json(value))
|
||
|
||
|
||
def _evaluator_cache_put(cache_key: str, value: dict[str, Any]) -> None:
|
||
if not _semantic_cache_enabled():
|
||
return
|
||
now = time.monotonic()
|
||
ttl = float(settings.evaluator_semantic_cache_ttl_seconds)
|
||
_EVALUATOR_CACHE[cache_key] = (now + ttl, json.loads(_canonical_json(value)))
|
||
_EVALUATOR_CACHE.move_to_end(cache_key)
|
||
_EVALUATOR_CACHE_STATS["stores"] += 1
|
||
max_entries = int(settings.evaluator_semantic_cache_max_entries)
|
||
while len(_EVALUATOR_CACHE) > max_entries:
|
||
_EVALUATOR_CACHE.popitem(last=False)
|
||
_EVALUATOR_CACHE_STATS["evictions"] += 1
|
||
|
||
|
||
def _parse_technique(raw: str) -> Optional[Technique]:
|
||
s = (raw or "").strip()
|
||
return _TECHNIQUE_BY_KO.get(s) or _TECHNIQUE_BY_CODE.get(s)
|
||
|
||
|
||
def _parse_client_state(raw: str) -> Optional[ClientState]:
|
||
s = (raw or "").strip()
|
||
return _CLIENT_STATE_BY_KO.get(s) or _CLIENT_STATE_BY_CODE.get(s)
|
||
|
||
|
||
def _coerce_str_list(val: Any) -> list[str]:
|
||
if val is None:
|
||
return []
|
||
if isinstance(val, str):
|
||
return [val]
|
||
if isinstance(val, (list, tuple)):
|
||
return [str(x) for x in val if x is not None]
|
||
return []
|
||
|
||
|
||
# ════════════════════════════════════════════════════════════════════════════
|
||
# 1. 결과 모델 (pydantic) — orchestrator/route 반환 + intent_deviation 1급 시민
|
||
# ════════════════════════════════════════════════════════════════════════════
|
||
class IntentDeviation(BaseModel):
|
||
"""'의도와 다른 부분'(윤찬 1급 시민). taxonomy.SupervisorComment(critique) intent_deviation 정합.
|
||
|
||
{dimension, expected, actual, severity} 구조화. dimension 은 평가 차원
|
||
(예: 'reflection', 'self_disclosure', 'pacing', 'risk_assessment').
|
||
"""
|
||
|
||
dimension: str = Field(..., description="관련 평가 차원(기법/페이싱/위험사정 등)")
|
||
expected: str = Field(..., description="권장된 반응/의도")
|
||
actual: str = Field(..., description="실제 나타난 반응")
|
||
severity: str = Field("minor", description="minor | moderate | major")
|
||
|
||
|
||
class TechniqueTag(BaseModel):
|
||
"""fast-loop 기법 태그 1건 — taxonomy.Technique 코드 + 한글 + 군집 + 근거."""
|
||
|
||
code: str # taxonomy.Technique.value
|
||
label_ko: str # TECHNIQUE_KO
|
||
category: str # TechniqueCategory.value (분포 집계축)
|
||
rationale: Optional[str] = None # 왜 이 기법으로 봤는지(근거 요구)
|
||
|
||
|
||
class ClientStateRead(BaseModel):
|
||
"""내담자 상태 '읽기' — 학습자 발화 직후 내담자 응답에서 관측된 상태(읽기 채점 근거)."""
|
||
|
||
code: str # taxonomy.ClientState.value
|
||
label_ko: str # CLIENT_STATE_KO
|
||
rationale: Optional[str] = None
|
||
|
||
|
||
class TurnEvaluation(BaseModel):
|
||
"""fast-loop 턴 평가 결과(턴 직후 경량 4차원).
|
||
|
||
4차원:
|
||
① technique[] : 학습자(상담자) 발화에 부착된 기법 라벨(복수)
|
||
② client_state_read[] : 내담자 응답에서 읽은 상태(복수)
|
||
③ appropriateness : 적절성 신호 pos|warn|neutral (경량)
|
||
④ intent_deviation : '의도와 다른 부분' 있으면 구조화(없으면 None)
|
||
"""
|
||
|
||
loop: str = "fast"
|
||
turn_seq: int
|
||
stage: str
|
||
techniques: list[TechniqueTag] = Field(default_factory=list)
|
||
client_state_read: list[ClientStateRead] = Field(default_factory=list)
|
||
appropriateness: Appropriateness = "neutral"
|
||
appropriateness_note: Optional[str] = None
|
||
intent_deviation: Optional[IntentDeviation] = None # 있을 때만(1급 시민)
|
||
rapport_signal: Optional[float] = (
|
||
None # 평가 AI 가 본 라포 신호(−1~+1, 상태머신 주입 가능)
|
||
)
|
||
theory_mode: Optional[str] = None
|
||
error: Optional[str] = None # 평가 실패 시 사유(비치명적; None 이면 정상)
|
||
|
||
def to_hook_dict(self) -> dict[str, Any]:
|
||
"""orchestrator.EvalHook 가 기대하는 평가 dict(turns.evaluation 적재용)."""
|
||
return self.model_dump(exclude_none=True)
|
||
|
||
|
||
class TechniqueDistribution(BaseModel):
|
||
"""deep-loop 기법 분포 — 군집별 카운트 + 과다/과소 진단."""
|
||
|
||
by_category: dict[str, int] = Field(default_factory=dict) # category.value -> count
|
||
by_technique: dict[str, int] = Field(
|
||
default_factory=dict
|
||
) # technique.value -> count
|
||
total: int = 0
|
||
overused: list[str] = Field(default_factory=list) # 과다 사용 군집(category.value)
|
||
underused: list[str] = Field(default_factory=list) # 과소/미사용 군집
|
||
|
||
|
||
class TurnValencePoint(BaseModel):
|
||
"""deep-loop 턴별 내담자 정서가 — 축어록 seq(1-based) 지목 + v(−1~+1)."""
|
||
|
||
seq: int # 마스킹 축어록의 1-based 턴 번호
|
||
v: float # 정서가(−1 매우 부정 ~ +1 매우 긍정)
|
||
|
||
|
||
class SessionEvaluation(BaseModel):
|
||
"""deep-loop 회기말/단계전환 정밀 평가 결과.
|
||
|
||
기법분포 + 잘한 순간 + 개선점(최대3) + 슈퍼바이저 rationale/critique + 의도이탈 집계.
|
||
"""
|
||
|
||
loop: str = "deep"
|
||
session_id: str
|
||
stage: str # 평가 시점 단계(전환 트리거면 from-stage)
|
||
scope: str = "session_end" # 'session_end' | 'stage_transition'
|
||
turns_evaluated: int = 0
|
||
distribution: TechniqueDistribution = Field(default_factory=TechniqueDistribution)
|
||
strengths: list[str] = Field(default_factory=list) # 잘한 순간(근거 포함 문장)
|
||
improvements: list[str] = Field(default_factory=list) # 개선점(최대 3)
|
||
intent_deviations: list[IntentDeviation] = Field(default_factory=list)
|
||
supervisor_rationale: Optional[str] = None # CommentKind.RATIONALE 종합
|
||
supervisor_critique: Optional[str] = None # CommentKind.CRITIQUE 종합
|
||
alternative_utterances: list[str] = Field(default_factory=list) # 대안 발화 제시
|
||
turn_valence: list[TurnValencePoint] = Field(
|
||
default_factory=list
|
||
) # 내담자 발화별 정서가(리뷰 감정 밸런스 차트 원천)
|
||
theory_mode: Optional[str] = None
|
||
error: Optional[str] = None
|
||
|
||
def to_dict(self) -> dict[str, Any]:
|
||
return self.model_dump(exclude_none=True)
|
||
|
||
|
||
# ════════════════════════════════════════════════════════════════════════════
|
||
# 2. structured_schema (게이트웨이 Structured Outputs 강제 — CCD/형식 안전)
|
||
# ════════════════════════════════════════════════════════════════════════════
|
||
def _technique_enum_values() -> list[str]:
|
||
return [t.value for t in Technique]
|
||
|
||
|
||
def _client_state_enum_values() -> list[str]:
|
||
return [s.value for s in ClientState]
|
||
|
||
|
||
def _fast_schema() -> dict[str, Any]:
|
||
"""fast-loop 구조화 출력 스키마. enum 후보를 코드값으로 강제(파싱 안정)."""
|
||
return {
|
||
"type": "object",
|
||
"additionalProperties": False,
|
||
"properties": {
|
||
"techniques": {
|
||
"type": "array",
|
||
"items": {
|
||
"type": "object",
|
||
"additionalProperties": False,
|
||
"properties": {
|
||
"code": {"type": "string", "enum": _technique_enum_values()},
|
||
"rationale": {"type": "string"},
|
||
},
|
||
"required": ["code"],
|
||
},
|
||
},
|
||
"client_state_read": {
|
||
"type": "array",
|
||
"items": {
|
||
"type": "object",
|
||
"additionalProperties": False,
|
||
"properties": {
|
||
"code": {"type": "string", "enum": _client_state_enum_values()},
|
||
"rationale": {"type": "string"},
|
||
},
|
||
"required": ["code"],
|
||
},
|
||
},
|
||
"appropriateness": {"type": "string", "enum": list(APPROPRIATENESS_VALUES)},
|
||
"appropriateness_note": {"type": "string"},
|
||
"rapport_signal": {"type": "number", "minimum": -1, "maximum": 1},
|
||
"intent_deviation": {
|
||
"type": ["object", "null"],
|
||
"additionalProperties": False,
|
||
"properties": {
|
||
"dimension": {"type": "string"},
|
||
"expected": {"type": "string"},
|
||
"actual": {"type": "string"},
|
||
"severity": {"type": "string", "enum": list(_SEVERITY)},
|
||
},
|
||
"required": ["dimension", "expected", "actual", "severity"],
|
||
},
|
||
},
|
||
"required": ["techniques", "client_state_read", "appropriateness"],
|
||
}
|
||
|
||
|
||
def _deep_schema() -> dict[str, Any]:
|
||
"""deep-loop 구조화 출력 스키마. 분포는 코드로 재집계하므로 LLM 엔 정성 평가만 요구."""
|
||
return {
|
||
"type": "object",
|
||
"additionalProperties": False,
|
||
"properties": {
|
||
"strengths": {"type": "array", "items": {"type": "string"}},
|
||
"improvements": {
|
||
"type": "array",
|
||
"items": {"type": "string"},
|
||
"maxItems": 3,
|
||
},
|
||
"intent_deviations": {
|
||
"type": "array",
|
||
"items": {
|
||
"type": "object",
|
||
"additionalProperties": False,
|
||
"properties": {
|
||
"dimension": {"type": "string"},
|
||
"expected": {"type": "string"},
|
||
"actual": {"type": "string"},
|
||
"severity": {"type": "string", "enum": list(_SEVERITY)},
|
||
},
|
||
"required": ["dimension", "expected", "actual", "severity"],
|
||
},
|
||
},
|
||
"supervisor_rationale": {"type": "string"},
|
||
"supervisor_critique": {"type": "string"},
|
||
"alternative_utterances": {"type": "array", "items": {"type": "string"}},
|
||
# 선택 필드 — 내담자 발화별 정서가(리뷰 감정 밸런스 차트 원천).
|
||
"turn_valence": {
|
||
"type": "array",
|
||
"items": {
|
||
"type": "object",
|
||
"additionalProperties": False,
|
||
"properties": {
|
||
"seq": {"type": "integer", "minimum": 1},
|
||
"v": {"type": "number", "minimum": -1, "maximum": 1},
|
||
},
|
||
"required": ["seq", "v"],
|
||
},
|
||
},
|
||
},
|
||
"required": ["strengths", "improvements"],
|
||
}
|
||
|
||
|
||
# ════════════════════════════════════════════════════════════════════════════
|
||
# 3. 프롬프트 빌더 — 정답 라벨 enum 후보 제시 + 근거 요구
|
||
# ════════════════════════════════════════════════════════════════════════════
|
||
_EVAL_ROLE = (
|
||
"당신은 상담 수련생을 지도하는 슈퍼바이저(평가 AI)다. 학습자(상담자) 발화를 임상적으로 평가한다.\n"
|
||
"당신은 내담자의 내부 설정(핵심신념·진단 차원·정답 라벨·상태 수치)을 *전부 볼 수 있다*. "
|
||
"이 정보는 평가 근거로만 쓰고, 평가 결과 자체는 학습자에게 직접 노출되지 않는다(시스템이 차단).\n"
|
||
"반드시 제시된 후보 라벨(code) 중에서만 고르고, 각 판단에 한국어 근거(rationale)를 붙인다. "
|
||
"추측·과잉 라벨링을 피하고, 근거가 약하면 부착하지 않는다."
|
||
)
|
||
|
||
|
||
def _technique_candidates_block() -> str:
|
||
"""기법 후보(군집별 그룹핑) — code: 한글 형식으로 제시."""
|
||
lines: list[str] = ["[기법 후보 — code: 라벨(군집)]"]
|
||
# 군집 순서대로 보기 좋게 그룹핑(평가자가 분포를 의식하도록).
|
||
by_cat: dict[TechniqueCategory, list[Technique]] = {}
|
||
for t, cat in TECHNIQUE_CATEGORY.items():
|
||
by_cat.setdefault(cat, []).append(t)
|
||
for cat, techs in by_cat.items():
|
||
items = ", ".join(f"{t.value}:{TECHNIQUE_KO[t]}" for t in techs)
|
||
lines.append(f"- {cat.value}: {items}")
|
||
return "\n".join(lines)
|
||
|
||
|
||
def _client_state_candidates_block() -> str:
|
||
items = ", ".join(f"{s.value}:{ko}" for s, ko in CLIENT_STATE_KO.items())
|
||
return f"[내담자 상태 후보 — code:라벨]\n- {items}"
|
||
|
||
|
||
# ─ few-shot 골든셋 예시(data/golden) — 명시적으로 켠 환경에서만 로딩 ─────────
|
||
# 골든셋은 학습/평가 보정 자료이지 운영 런타임의 기본 데이터가 아니다.
|
||
_GOLDEN_FEWSHOT_ENABLED = os.environ.get(
|
||
"EVALUATOR_GOLDEN_FEWSHOT_ENABLED", ""
|
||
).lower() in {
|
||
"1",
|
||
"true",
|
||
"yes",
|
||
"on",
|
||
}
|
||
_GOLDEN_DIR = os.environ.get("GOLDEN_DIR") or os.path.join(
|
||
os.path.dirname(__file__), "..", "..", "..", "..", "data", "golden"
|
||
)
|
||
|
||
|
||
def _load_fewshot_examples(max_n: int = 6) -> list[dict[str, Any]]:
|
||
"""골든셋에서 기법 다양성을 커버하는 상담자 발화 few-shot 예시(없으면 빈 리스트)."""
|
||
out: list[dict[str, Any]] = []
|
||
if not _GOLDEN_FEWSHOT_ENABLED:
|
||
return out
|
||
seen: set[str] = set()
|
||
try:
|
||
files = sorted(f for f in os.listdir(_GOLDEN_DIR) if f.endswith(".jsonl"))
|
||
except OSError:
|
||
return out
|
||
for fn in files:
|
||
try:
|
||
with open(os.path.join(_GOLDEN_DIR, fn), encoding="utf-8") as fh:
|
||
for line in fh:
|
||
line = line.strip()
|
||
if not line:
|
||
continue
|
||
d = json.loads(line)
|
||
if d.get("speaker") != "counselor" or not d.get("techniques"):
|
||
continue
|
||
if set(d["techniques"]) - seen: # 새 기법 커버 예시 우선
|
||
out.append(d)
|
||
seen.update(d["techniques"])
|
||
if len(out) >= max_n:
|
||
return out
|
||
except (OSError, json.JSONDecodeError):
|
||
continue
|
||
return out
|
||
|
||
|
||
def _fewshot_block() -> str:
|
||
"""발화→기법 few-shot 예시 블록(평가 라벨 일관성 보정). 골든셋 없으면 빈 문자열."""
|
||
ex = _load_fewshot_examples()
|
||
if not ex:
|
||
return ""
|
||
lines = ["[few-shot 예시 — 발화 → 기법 code (골든셋 참고, 라벨 일관성 보정용)]"]
|
||
for e in ex:
|
||
techs = ",".join(e.get("techniques", []))
|
||
text = (e.get("text") or "").replace("\n", " ")[:70]
|
||
rat = next(
|
||
(
|
||
c.get("text", "")
|
||
for c in e.get("comments", [])
|
||
if c.get("kind") == "rationale"
|
||
),
|
||
"",
|
||
)
|
||
line = f'- "{text}…" → {techs}'
|
||
if rat:
|
||
line += f" (근거: {rat[:48]}…)"
|
||
lines.append(line)
|
||
return "\n".join(lines)
|
||
|
||
|
||
def _theory_mode(ctx: "TurnContext") -> Optional[str]:
|
||
"""이론 모드(이론부합 평가용): 학습자 선택(회기 theory_mode) 우선, 없으면 페르소나 theory_target."""
|
||
sess_theory = getattr(ctx, "theory_mode", None)
|
||
if sess_theory:
|
||
return str(sess_theory)
|
||
tt = getattr(ctx.persona, "theory_target", None)
|
||
if isinstance(tt, (list, tuple)) and tt:
|
||
return ", ".join(str(x) for x in tt)
|
||
return None
|
||
|
||
|
||
def build_fast_messages(ctx: "TurnContext", client_reply: str) -> list[EngineMessage]:
|
||
"""fast-loop 평가 프롬프트(L0 역할 + 후보 라벨 + 이번 턴 맥락)."""
|
||
st = ctx.state_after or ctx.state_before
|
||
theory = _theory_mode(ctx)
|
||
client_reply_masked = guardrail.mask_role_identities(
|
||
client_reply,
|
||
counselor_identity=ctx.counselor_identity,
|
||
client_identity=ctx.client_identity,
|
||
synthetic_generated=True,
|
||
).text_masked
|
||
recent = (
|
||
"\n".join(
|
||
f"{speaker_ko_label(t.get('speaker'))}: {t.get('text', '')}"
|
||
for t in (ctx.memory.recent_turns or [])[-4:]
|
||
)
|
||
or "(직전 맥락 없음)"
|
||
)
|
||
|
||
crisis_note = ""
|
||
if ctx.crisis is not None and getattr(ctx.crisis, "escalate", False):
|
||
crisis_note = (
|
||
"\n[안전] 학습자 발화에서 실제 위기 신호가 감지됨 — 위험사정(risk_assessment) "
|
||
"적절성과 안전 페이싱을 특히 살펴라."
|
||
)
|
||
|
||
system = "\n\n".join(
|
||
p
|
||
for p in [
|
||
_EVAL_ROLE,
|
||
_technique_candidates_block(),
|
||
_client_state_candidates_block(),
|
||
_fewshot_block(), # 골든셋 few-shot(없으면 빈 문자열 → 필터됨)
|
||
(
|
||
"[평가 4차원]\n"
|
||
"① technique: 이번 *상담자(학습자)* 발화에 부착되는 기법(복수 가능, 후보 code 만).\n"
|
||
"② client_state_read: 이어진 *내담자* 응답에서 읽히는 상태(복수, 후보 code 만).\n"
|
||
"③ appropriateness: 이번 상담자 반응의 적절성 — pos(적절)/warn(주의)/neutral(중립).\n"
|
||
"④ intent_deviation: '의도와 다른 부분'이 있으면 {dimension, expected, actual, severity}로. "
|
||
"없으면 null. 이 항목은 가장 중요하다 — 무리한 생성 금지, 진짜 이탈만.\n"
|
||
"추가로 rapport_signal(−1~+1): 이 발화가 라포에 끼친 방향(공감·반영=+, 조언점프·평가=−)."
|
||
),
|
||
]
|
||
if p
|
||
)
|
||
|
||
user = (
|
||
f"[단계] {st.stage.value} [effective_openness] {st.effective_openness:.2f} "
|
||
f"[ideation_stage] {st.ideation_stage}"
|
||
+ (f" [이론모드] {theory}" if theory else "")
|
||
+ crisis_note
|
||
+ f"\n\n[직전 맥락]\n{recent}\n\n"
|
||
f"[평가 대상 — 상담자(학습자) 발화]\n{ctx.learner_text_masked}\n\n"
|
||
f"[이어진 내담자 응답]\n{client_reply_masked}\n\n"
|
||
"위 4차원으로 구조화 평가하라. 후보 code 외 라벨 금지, 각 판단에 rationale 첨부."
|
||
)
|
||
return [
|
||
EngineMessage(role="system", content=system, cache=True),
|
||
EngineMessage(role="user", content=user, cache=False),
|
||
]
|
||
|
||
|
||
def build_deep_messages(
|
||
*,
|
||
stage: str,
|
||
scope: str,
|
||
theory_mode: Optional[str],
|
||
masked_turns: list[dict[str, str]],
|
||
distribution: "TechniqueDistribution",
|
||
) -> list[EngineMessage]:
|
||
"""deep-loop 평가 프롬프트(전체 회기 + 코드 집계 분포 + 골든라벨 후보)."""
|
||
transcript = (
|
||
"\n".join(
|
||
f"{t.get('seq', '')}{speaker_ko_label(t.get('speaker'))}: {t.get('text', '')}"
|
||
for t in masked_turns
|
||
)
|
||
or "(축어록 없음)"
|
||
)
|
||
dist_lines = (
|
||
", ".join(f"{k}:{v}" for k, v in distribution.by_category.items()) or "(없음)"
|
||
)
|
||
over = ", ".join(distribution.overused) or "(없음)"
|
||
under = ", ".join(distribution.underused) or "(없음)"
|
||
|
||
system = "\n\n".join(
|
||
[
|
||
_EVAL_ROLE,
|
||
_technique_candidates_block(),
|
||
(
|
||
"[deep-loop 지시]\n"
|
||
"전체 회기를 보고 정밀 평가한다. 다음을 산출하라:\n"
|
||
"- strengths: 학습자가 잘한 구체적 순간(근거 포함, 발화 인용 가능).\n"
|
||
"- improvements: 개선점 최대 3개(우선순위 순, 실행가능한 코칭).\n"
|
||
"- intent_deviations: '의도와 다른 부분' 전부 {dimension, expected, actual, severity}로.\n"
|
||
"- supervisor_rationale: 회기 전반에서 적절했던 개입의 근거(rationale) 종합.\n"
|
||
"- supervisor_critique: 과도/부족/평가적 시각 등 주의점(critique) 종합.\n"
|
||
"- alternative_utterances: 핵심 장면에 더 나은 대안 상담자 발화 1~3개.\n"
|
||
"- turn_valence: *내담자* 발화 각각의 정서가를 {seq, v}로 산출. seq 는 축어록의 "
|
||
"1-based 턴 번호, v 는 −1(매우 부정)~+1(매우 긍정). 근거 없는 극단값을 피하고 "
|
||
"불확실하면 0 근처로."
|
||
),
|
||
]
|
||
)
|
||
|
||
user = (
|
||
f"[평가 시점] {scope} [단계] {stage}"
|
||
+ (f" [이론모드] {theory_mode}" if theory_mode else "")
|
||
+ f"\n[기법 군집 분포(코드 집계)] {dist_lines}\n"
|
||
f"[과다 군집] {over} [과소/미사용 군집] {under}\n\n"
|
||
f"[마스킹 축어록]\n{transcript}\n\n"
|
||
"위를 근거로 deep-loop 평가를 구조화 산출하라. improvements 는 최대 3개."
|
||
)
|
||
return [
|
||
EngineMessage(role="system", content=system, cache=True),
|
||
EngineMessage(role="user", content=user, cache=False),
|
||
]
|
||
|
||
|
||
def _parse_intent_deviation(d: Any) -> Optional[IntentDeviation]:
|
||
if not isinstance(d, dict):
|
||
return None
|
||
dim = str(d.get("dimension") or "").strip()
|
||
exp = str(d.get("expected") or "").strip()
|
||
act = str(d.get("actual") or "").strip()
|
||
if not (dim and exp and act):
|
||
return None
|
||
sev = str(d.get("severity") or "minor").strip()
|
||
if sev not in _SEVERITY:
|
||
sev = "minor"
|
||
return IntentDeviation(dimension=dim, expected=exp, actual=act, severity=sev)
|
||
|
||
|
||
def _parse_fast(
|
||
payload: dict[str, Any], *, turn_seq: int, stage: str, theory: Optional[str]
|
||
) -> TurnEvaluation:
|
||
techniques: list[TechniqueTag] = []
|
||
for item in payload.get("techniques") or []:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
t = _parse_technique(str(item.get("code", "")))
|
||
if t is None:
|
||
continue
|
||
techniques.append(
|
||
TechniqueTag(
|
||
code=t.value,
|
||
label_ko=TECHNIQUE_KO[t],
|
||
category=TECHNIQUE_CATEGORY[t].value,
|
||
rationale=(str(item.get("rationale")).strip() or None)
|
||
if item.get("rationale")
|
||
else None,
|
||
)
|
||
)
|
||
|
||
states: list[ClientStateRead] = []
|
||
for item in payload.get("client_state_read") or []:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
s = _parse_client_state(str(item.get("code", "")))
|
||
if s is None:
|
||
continue
|
||
states.append(
|
||
ClientStateRead(
|
||
code=s.value,
|
||
label_ko=CLIENT_STATE_KO[s],
|
||
rationale=(str(item.get("rationale")).strip() or None)
|
||
if item.get("rationale")
|
||
else None,
|
||
)
|
||
)
|
||
|
||
appro = str(payload.get("appropriateness") or "neutral").strip()
|
||
if appro not in APPROPRIATENESS_VALUES:
|
||
appro = "neutral"
|
||
|
||
rapport = payload.get("rapport_signal")
|
||
rapport_val: Optional[float] = None
|
||
if isinstance(rapport, (int, float)):
|
||
rapport_val = max(-1.0, min(1.0, float(rapport)))
|
||
|
||
return TurnEvaluation(
|
||
loop="fast",
|
||
turn_seq=turn_seq,
|
||
stage=stage,
|
||
techniques=techniques,
|
||
client_state_read=states,
|
||
appropriateness=appro,
|
||
appropriateness_note=(str(payload.get("appropriateness_note")).strip() or None)
|
||
if payload.get("appropriateness_note")
|
||
else None,
|
||
intent_deviation=_parse_intent_deviation(payload.get("intent_deviation")),
|
||
rapport_signal=rapport_val,
|
||
theory_mode=theory,
|
||
)
|
||
|
||
|
||
# ════════════════════════════════════════════════════════════════════════════
|
||
# 5. 분포 집계 (코드 결정론 — LLM 미경유, 무손실)
|
||
# ════════════════════════════════════════════════════════════════════════════
|
||
def aggregate_distribution(technique_codes: list[str]) -> TechniqueDistribution:
|
||
"""부착된 기법 코드 리스트 → 군집/기법 분포 + 과다·과소 진단(결정론).
|
||
|
||
과다/과소는 회기 전체 5개 군집 균형 기준의 *경량 휴리스틱*이다(정밀 채점은 deep LLM).
|
||
"""
|
||
by_tech: dict[str, int] = {}
|
||
by_cat: dict[str, int] = {}
|
||
for code in technique_codes:
|
||
t = _TECHNIQUE_BY_CODE.get(code)
|
||
if t is None:
|
||
continue
|
||
by_tech[t.value] = by_tech.get(t.value, 0) + 1
|
||
cat = TECHNIQUE_CATEGORY[t].value
|
||
by_cat[cat] = by_cat.get(cat, 0) + 1
|
||
|
||
total = sum(by_cat.values())
|
||
all_cats = [c.value for c in TechniqueCategory]
|
||
overused: list[str] = []
|
||
underused: list[str] = []
|
||
if total > 0:
|
||
# 균등 기대치 = total / 군집수. 1.6배↑=과다, 미사용=과소.
|
||
expected = total / len(all_cats)
|
||
for c in all_cats:
|
||
cnt = by_cat.get(c, 0)
|
||
if cnt == 0:
|
||
underused.append(c)
|
||
elif cnt >= max(2, expected * 1.6):
|
||
overused.append(c)
|
||
|
||
return TechniqueDistribution(
|
||
by_category=by_cat,
|
||
by_technique=by_tech,
|
||
total=total,
|
||
overused=overused,
|
||
underused=underused,
|
||
)
|
||
|
||
|
||
# ════════════════════════════════════════════════════════════════════════════
|
||
# 6. 공개 평가 함수 (orchestrator 주입형) — async
|
||
# ════════════════════════════════════════════════════════════════════════════
|
||
async def evaluate_turn(
|
||
ctx: "TurnContext",
|
||
client_reply: str,
|
||
*,
|
||
engine: EngineClient,
|
||
audit_hook: Optional["LlmAuditHook"] = None,
|
||
) -> TurnEvaluation:
|
||
"""fast-loop 턴 평가 — 턴 직후 경량 4차원 태깅(비치명적).
|
||
|
||
engine 장애·파싱 실패 시 빈 평가(error 사유 기록)를 반환한다. 절대 raise 하지 않는다
|
||
(orchestrator.run_turn_generate 의 eval_hook 가 None 으로 흡수하지만, 여기서 1차 흡수).
|
||
"""
|
||
st = ctx.state_after or ctx.state_before
|
||
theory = _theory_mode(ctx)
|
||
base = TurnEvaluation(
|
||
loop="fast", turn_seq=st.turn_seq, stage=st.stage.value, theory_mode=theory
|
||
)
|
||
|
||
try:
|
||
req = GenerateRequest(
|
||
ai_role="evaluator",
|
||
messages=build_fast_messages(ctx, client_reply),
|
||
structured_schema=_fast_schema(),
|
||
model=_configured_model(settings.evaluator_fast_model),
|
||
max_tokens=900,
|
||
temperature=0.2, # 평가는 보수적·재현적으로
|
||
session_id=ctx.session_id,
|
||
metadata={"loop": "fast", "stage": st.stage.value, "turn_seq": st.turn_seq},
|
||
)
|
||
cache_key = _evaluator_cache_key(req)
|
||
cached = _evaluator_cache_get(cache_key)
|
||
if cached is not None:
|
||
return TurnEvaluation.model_validate(cached)
|
||
resp = await generate_with_audit(engine, req, audit_hook)
|
||
except EngineError:
|
||
base.error = "engine_error"
|
||
return base
|
||
except Exception: # 방어 — 어떤 예외도 상담 루프를 막지 않게
|
||
base.error = "eval_error"
|
||
return base
|
||
|
||
payload = structured_payload_from_response(resp)
|
||
if payload is None:
|
||
base.error = "no_structured_output"
|
||
return base
|
||
try:
|
||
result = _parse_fast(
|
||
payload, turn_seq=st.turn_seq, stage=st.stage.value, theory=theory
|
||
)
|
||
_evaluator_cache_put(cache_key, result.model_dump())
|
||
return result
|
||
except Exception: # 파싱 방어
|
||
base.error = "parse_error"
|
||
return base
|
||
|
||
|
||
async def evaluate_session(
|
||
*,
|
||
session_id: str,
|
||
stage: str,
|
||
masked_turns: list[dict[str, Any]],
|
||
engine: EngineClient,
|
||
technique_codes: Optional[list[str]] = None,
|
||
theory_mode: Optional[str] = None,
|
||
scope: str = "session_end",
|
||
audit_hook: Optional["LlmAuditHook"] = None,
|
||
timeout: float | None = None,
|
||
) -> SessionEvaluation:
|
||
"""deep-loop 정밀 평가 — 단계전환/회기말. 전체 축어록 + 코드 집계 분포 + LLM 정성 평가.
|
||
|
||
technique_codes: fast-loop 들에서 누적된 부착 기법 코드(없으면 빈 분포).
|
||
엔진/파싱 실패는 비치명적(error 기록 + 분포는 코드로 채움).
|
||
"""
|
||
distribution = aggregate_distribution(technique_codes or [])
|
||
counselor_turns = sum(1 for t in masked_turns if t.get("speaker") == "counselor")
|
||
base = SessionEvaluation(
|
||
loop="deep",
|
||
session_id=session_id,
|
||
stage=stage,
|
||
scope=scope,
|
||
turns_evaluated=counselor_turns,
|
||
distribution=distribution,
|
||
theory_mode=theory_mode,
|
||
)
|
||
|
||
try:
|
||
req = GenerateRequest(
|
||
ai_role="evaluator",
|
||
messages=build_deep_messages(
|
||
stage=stage,
|
||
scope=scope,
|
||
theory_mode=theory_mode,
|
||
masked_turns=[{k: v for k, v in t.items()} for t in masked_turns],
|
||
distribution=distribution,
|
||
),
|
||
structured_schema=_deep_schema(),
|
||
model=_configured_model(settings.evaluator_deep_model),
|
||
max_tokens=2048,
|
||
temperature=0.3,
|
||
session_id=session_id,
|
||
metadata={"loop": "deep", "scope": scope, "stage": stage},
|
||
)
|
||
cache_key = _evaluator_cache_key(req)
|
||
cached = _evaluator_cache_get(cache_key)
|
||
if cached is not None:
|
||
return SessionEvaluation.model_validate(cached)
|
||
resp = await generate_with_audit(engine, req, audit_hook, timeout=timeout)
|
||
except EngineError as e:
|
||
base.error = f"engine_error: {e}"
|
||
return base
|
||
except Exception as e:
|
||
base.error = f"eval_error: {e}"
|
||
return base
|
||
|
||
payload = structured_payload_from_response(resp)
|
||
if payload is None:
|
||
base.error = "no_structured_output"
|
||
return base
|
||
|
||
base.strengths = _coerce_str_list(payload.get("strengths"))
|
||
base.improvements = _coerce_str_list(payload.get("improvements"))[:3] # 최대 3
|
||
base.alternative_utterances = _coerce_str_list(
|
||
payload.get("alternative_utterances")
|
||
)
|
||
rationale = payload.get("supervisor_rationale")
|
||
critique = payload.get("supervisor_critique")
|
||
base.supervisor_rationale = str(rationale).strip() if rationale else None
|
||
base.supervisor_critique = str(critique).strip() if critique else None
|
||
for d in payload.get("intent_deviations") or []:
|
||
dev = _parse_intent_deviation(d)
|
||
if dev is not None:
|
||
base.intent_deviations.append(dev)
|
||
for item in payload.get("turn_valence") or []:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
seq = item.get("seq")
|
||
v = item.get("v")
|
||
if not isinstance(seq, int) or isinstance(seq, bool) or seq < 1:
|
||
continue
|
||
if not isinstance(v, (int, float)) or isinstance(v, bool):
|
||
continue
|
||
base.turn_valence.append(
|
||
TurnValencePoint(seq=seq, v=max(-1.0, min(1.0, float(v))))
|
||
)
|
||
_evaluator_cache_put(cache_key, base.model_dump())
|
||
return base
|
||
|
||
|
||
# ════════════════════════════════════════════════════════════════════════════
|
||
# 7. orchestrator EvalHook 어댑터 — 주입형 클로저(엔진 바인딩)
|
||
# ════════════════════════════════════════════════════════════════════════════
|
||
def make_eval_hook(
|
||
engine: EngineClient,
|
||
*,
|
||
audit_hook: Optional["LlmAuditHook"] = None,
|
||
):
|
||
"""orchestrator.EvalHook(Callable[[TurnContext, str], Awaitable[Optional[dict]]]) 호환 클로저.
|
||
|
||
sessions 라우트가 run_turn_generate(ctx, engine, eval_hook=make_eval_hook(engine_client)) 로
|
||
주입한다. 평가 실패는 None 으로(상담 루프 비차단).
|
||
"""
|
||
|
||
async def _hook(ctx: "TurnContext", client_reply: str) -> Optional[dict[str, Any]]:
|
||
ev = await evaluate_turn(
|
||
ctx, client_reply, engine=engine, audit_hook=audit_hook
|
||
)
|
||
d = ev.to_hook_dict()
|
||
return d if d else None
|
||
|
||
return _hook
|
||
|
||
|
||
__all__ = [
|
||
"IntentDeviation",
|
||
"TechniqueTag",
|
||
"ClientStateRead",
|
||
"TurnEvaluation",
|
||
"TechniqueDistribution",
|
||
"TurnValencePoint",
|
||
"SessionEvaluation",
|
||
"aggregate_distribution",
|
||
"evaluate_turn",
|
||
"evaluate_session",
|
||
"make_eval_hook",
|
||
"build_fast_messages",
|
||
"build_deep_messages",
|
||
"clear_evaluator_semantic_cache",
|
||
"evaluator_semantic_cache_stats",
|
||
]
|