diff --git a/apps/api/app/services/evaluator.py b/apps/api/app/services/evaluator.py index c62cb30..6c105f5 100644 --- a/apps/api/app/services/evaluator.py +++ b/apps/api/app/services/evaluator.py @@ -25,6 +25,7 @@ MASTERPLAN §2.3 (평가 AI 2-tier 루프): from __future__ import annotations import json +import os from typing import TYPE_CHECKING, Any, Optional from pydantic import BaseModel, Field @@ -300,6 +301,62 @@ def _client_state_candidates_block() -> str: return f"[내담자 상태 후보 — code:라벨]\n- {items}" +# ─ few-shot 골든셋 예시(data/golden) — 발화→기법 라벨 일관성 보정 ──────────── +# 윤찬 6결정 '재귀학습 few-shot부터' + taxonomy '0615 합성변형' 설계의 실제 활용. +# 컨테이너/배포에선 GOLDEN_DIR env 로 마운트 경로 지정. 없으면 graceful(빈 블록). +_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]] = [] + 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_target 에서 이론 모드 힌트(이론부합 평가용). 없으면 None.""" tt = getattr(ctx.persona, "theory_target", None) @@ -325,10 +382,11 @@ def build_fast_messages(ctx: "TurnContext", client_reply: str) -> list[EngineMes ) 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" @@ -338,7 +396,7 @@ def build_fast_messages(ctx: "TurnContext", client_reply: str) -> list[EngineMes "없으면 null. 이 항목은 가장 중요하다 — 무리한 생성 금지, 진짜 이탈만.\n" "추가로 rapport_signal(−1~+1): 이 발화가 라포에 끼친 방향(공감·반영=+, 조언점프·평가=−)." ), - ] + ] if p ) user = (