yunchan/vignette
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: P1 풀빌드 — React 프론트 7화면 + 백엔드 상담루프·평가·음성·RAG

Browse source 
web (Vite+React19+TS, Cloudflare Pages 배포):
- 디자인토큰(세이지틸/테라코타 SSOT), 앱셸, 공통 UI 프리미티브
- 7화면: 로그인/학습자홈/상담세션/회기리뷰/교수자/관리자/설정
- ClientAvatar: SVG 반구상 흉상 4상태 + RMS 립싱크 + 6파라미터 정서
- 회기리뷰는 외부 레퍼런스 디자인을 Vignette 토큰으로 리스킨

api (FastAPI):
- 게이트웨이 /v1/generate·/v1/stream 어댑터(상주풀/EngineSession 보존)
- services: 페르소나 L0~L6 빌더 / 결정론 상태머신 / 가드레일 /
  턴 오케스트레이터 / 회기간 메모리 / 평가AI / 음성 / RAG
- store: DB off 폴백(in-memory), sessions 실구현

검증:
- web: node22 tsc+vite build 통과(node23 segfault 회피), Pages 배포 200
- api: app.main import 통과
- 핫픽스: Topbar initials undefined-safe (undefined.trim 크래시)
- E2E: 서연(P1) 상담 1턴 — 좋은/나쁜 상담에 차등 반응 실증
This commit is contained in:
Yun Chan 2026-06-25 23:37:22 +09:00
parent 859ab26314
commit 24b1b7a6e1
84 changed files with 19645 additions and 107 deletions
Download patch file Download diff file Expand all files Collapse all files

701
apps/api/app/services/evaluator.py Normal file
View file

@ -0,0 +1,701 @@
"""평가 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(tier='feedback', 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 json
from typing import TYPE_CHECKING, Any, Optional
from pydantic import BaseModel, Field
from ..engine_client import (
EngineClient,
EngineError,
EngineMessage,
GenerateRequest,
GenerateResponse,
)
from ..taxonomy import (
CLIENT_STATE_KO,
TECHNIQUE_CATEGORY,
TECHNIQUE_KO,
ClientState,
CommentKind,
Technique,
TechniqueCategory,
)
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 의 경량 판단(상태머신 라포 추정과 별개 차원).
_APPROPRIATENESS = ("pos", "warn", "neutral")
# 의도이탈 심각도 (taxonomy.SupervisorComment.severity 와 동일 어휘).
_SEVERITY = ("minor", "moderate", "major")
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: str = "neutral" # pos | warn | 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 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) # 대안 발화 제시
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)},
"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"}},
},
"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}"
def _theory_mode(ctx: "TurnContext") -> Optional[str]:
"""페르소나 theory_target 에서 이론 모드 힌트(이론부합 평가용). 없으면 None."""
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)
recent = "\n".join(
f"{('상담자' if t.get('speaker') == 'counselor' else '내담자')}: {t.get('text', '')}"
for t in (ctx.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(
[
_EVAL_ROLE,
_technique_candidates_block(),
_client_state_candidates_block(),
(
"[평가 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): 이 발화가 라포에 끼친 방향(공감·반영=+, 조언점프·평가=−)."
),
]
)
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}\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', '')}{('상담자' if t.get('speaker') == 'counselor' else '내담자')}: {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개."
),
]
)
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),
]
# ════════════════════════════════════════════════════════════════════════════
# 4. 응답 파싱 — structured 우선, 없으면 text(JSON) 폴백, 실패는 빈 결과
# ════════════════════════════════════════════════════════════════════════════
def _structured_payload(resp: GenerateResponse) -> Optional[dict[str, Any]]:
"""게이트웨이 structured 우선, 없으면 text 에서 JSON 추출(코드펜스/잡텍스트 관용)."""
if resp.structured is not None and isinstance(resp.structured, dict):
return resp.structured
raw = (resp.text or "").strip()
if not raw:
return None
# ```json ... ``` 펜스 제거
if raw.startswith("```"):
raw = raw.split("```", 2)[1] if raw.count("```") >= 2 else raw.strip("`")
if raw.lstrip().lower().startswith("json"):
raw = raw.lstrip()[4:]
try:
obj = json.loads(raw)
return obj if isinstance(obj, dict) else None
except (json.JSONDecodeError, ValueError):
# 본문 안에 묻힌 첫 객체만 시도
start, end = raw.find("{"), raw.rfind("}")
if 0 <= start < end:
try:
obj = json.loads(raw[start : end + 1])
return obj if isinstance(obj, dict) else None
except (json.JSONDecodeError, ValueError):
return None
return None
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:
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,
) -> 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",
tier="feedback",
messages=build_fast_messages(ctx, client_reply),
structured_schema=_fast_schema(),
max_tokens=900,
temperature=0.2, # 평가는 보수적·재현적으로
session_id=ctx.session_id,
metadata={"loop": "fast", "stage": st.stage.value, "turn_seq": st.turn_seq},
)
resp = await engine.generate(req)
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(resp)
if payload is None:
base.error = "no_structured_output"
return base
try:
return _parse_fast(payload, turn_seq=st.turn_seq, stage=st.stage.value, theory=theory)
except Exception as e: # 파싱 방어
base.error = f"parse_error: {e}"
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",
) -> 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",
tier="feedback",
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(),
max_tokens=2048,
temperature=0.3,
session_id=session_id,
metadata={"loop": "deep", "scope": scope, "stage": stage},
)
resp = await engine.generate(req)
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(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)
return base
# ════════════════════════════════════════════════════════════════════════════
# 7. orchestrator EvalHook 어댑터 — 주입형 클로저(엔진 바인딩)
# ════════════════════════════════════════════════════════════════════════════
def make_eval_hook(engine: EngineClient):
"""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)
d = ev.to_hook_dict()
return d if d else None
return _hook
__all__ = [
"IntentDeviation",
"TechniqueTag",
"ClientStateRead",
"TurnEvaluation",
"TechniqueDistribution",
"SessionEvaluation",
"aggregate_distribution",
"evaluate_turn",
"evaluate_session",
"make_eval_hook",
"build_fast_messages",
"build_deep_messages",
]