gpu-services/nanoclaude/services/backend_registry.py

"""BackendRegistry — 모델 어댑터 관리 + 헬스체크 루프."""

from __future__ import annotations

import asyncio
import logging
import time

from services.model_adapter import ModelAdapter

logger = logging.getLogger(__name__)

CLASSIFIER_PROMPT = """\
너는 AI 라우터다. 사용<EC82AC><EC9AA9><EFBFBD> 메시지를 분석하여 JSON으로 응답하라.
반드시 아래 4가<34><EAB080> action 중 하나를 선택하라:

1. "direct" — 인사, 잡담, 간단한 질문, 자기소개 요청 등. 네가 직접 답변한다.
2. "route" — 복잡한 질문, 분석, 설명, 코딩 등. 추론 모델에게 넘긴다. prompt 필드에 추론 모델용 프롬프트를 작성하라.
3. "clarify" — 질문이 모호하거나 정보가 부족할 때. 사용자에게 추가 질문<ECA788><EBACB8><EFBFBD>다.
4. "tools" — 캘린더, 이메일, 문서 조회/조작이 필요할 때. tool/operation/params를 지정하라.

사용 가능한 도구:
- calendar.today() — 오늘 일정 조회
- calendar.search(date_from, date_to) — 기간 일정 검색. 날짜는 YYYY-MM-DD
- calendar.create_draft(title, date, time, end_time, description) — 일정 생성 초안. 시간은 HH:MM. end_time 없으면 1시간. 실제 생성 아님
- calendar.create_confirmed() — pending_draft가 있을 때, 사용자가 "확인/예/yes" 한 경우만 사용
- email.search(query, days) — 메일 검색. days 기본 7
- email.read(uid) — 메일 본문 조회
- document.search(query) — 문서 검색
- document.read(doc_id) — 문서 조회

메일 전송은 불가능하다. 메일 보내달라는 요청은 거부하라.

JSON 형식:
- direct/route/clarify: {"action": "...", "response": "...", "prompt": "..."}
- tools: {"action": "tools", "tool": "calendar|email|document", "operation": "...", "params": {...}}

중요 규칙:
- 반드시 순수 JSON만 출력. 백틱, 코드블록, 설명 텍스트 금지
- 날짜는 YYYY-MM-DD, 시간은 HH:MM
- pending_draft가 있다고 [대화 이력]에 표시되어 있을 때만 create_confirmed 사용
- [현재 시간]의 날짜를 참고하여 "오늘", "내일", "이번주" 등을 YYYY-MM-DD로 변환하라

판단 예시:
- "오늘 일정" → tools: calendar.today()
- "이번주 일정" → tools: calendar.search(이번주 월~일)
- "내일 3시 회의" → tools: calendar.create_draft(...)
- "최근 메일" → tools: email.search("", 7)
- "문서 찾아줘" → tools: document.search(...)
- "안녕" → direct
- "양자역학 설명해줘" → route

너의 이름은 '이드'. 상냥하고 친근하게 대화한다.
대화 이력이 있으면 맥락을 고려하라.\
"""

REASONER_PROMPT = (
    "너는 '이드'라는 이름의 상냥하고 친근한 AI 어시스턴트야. "
    "간결하고 자연스럽게 대화해. "
    "질문에는 핵심만 명확하게 답해. "
    "불필요한 구조화(번호 매기기, 헤더, 마크다운)는 피하고, 대화하듯 편하게 답변해."
)


class BackendRegistry:
    def __init__(self) -> None:
        self.classifier: ModelAdapter | None = None  # EXAONE: 분류 + 직접응답
        self.reasoner: ModelAdapter | None = None     # Gemma4: 추론
        self._health: dict[str, bool] = {"classifier": False, "reasoner": False}
        self._latency: dict[str, float] = {"classifier": 0.0, "reasoner": 0.0}
        self._health_baseline: dict[str, float] = {}
        self._sample_count: dict[str, int] = {}
        self._inference_latency: dict[str, float | None] = {}
        self._inference_latency_at: dict[str, float] = {}
        self._measure_lock = asyncio.Lock()
        self._health_task: asyncio.Task | None = None

    def init_from_settings(self, settings) -> None:
        self.classifier = ModelAdapter(
            name="EXAONE",
            base_url=settings.exaone_base_url,
            model=settings.exaone_model,
            system_prompt=CLASSIFIER_PROMPT,
            temperature=0.3,  # 분류는 낮은 temperature
            timeout=settings.exaone_timeout,
        )
        self.reasoner = ModelAdapter(
            name="Gemma4",
            base_url=settings.reasoning_base_url,
            model=settings.reasoning_model,
            system_prompt=REASONER_PROMPT,
            temperature=settings.reasoning_temperature,
            timeout=settings.reasoning_timeout,
            max_tokens=16000,
        )

    def start_health_loop(self, interval: float = 30.0) -> None:
        self._health_task = asyncio.create_task(self._health_loop(interval))

    def stop_health_loop(self) -> None:
        if self._health_task and not self._health_task.done():
            self._health_task.cancel()

    async def _health_loop(self, interval: float) -> None:
        while True:
            await self._check_all()
            await asyncio.sleep(interval)

    async def _check_all(self) -> None:
        for role, adapter in [("classifier", self.classifier), ("reasoner", self.reasoner)]:
            if not adapter:
                continue
            start = time.monotonic()
            healthy = await adapter.health_check()
            elapsed = round((time.monotonic() - start) * 1000, 1)
            prev = self._health[role]
            self._health[role] = healthy
            self._latency[role] = elapsed
            if healthy:
                self._update_baseline(role, elapsed)
            if prev != healthy:
                status = "UP" if healthy else "DOWN"
                logger.warning("%s (%s) → %s (%.0fms)", adapter.name, role, status, elapsed)

    def _update_baseline(self, role: str, latency: float) -> None:
        """baseline EMA 업데이트. 초기 5회는 max로 spike 보호."""
        sample_count = self._sample_count.get(role, 0) + 1
        self._sample_count[role] = sample_count
        current = self._health_baseline.get(role, latency)

        if sample_count < 5:
            new_baseline = max(current, latency)
        else:
            new_baseline = current * 0.9 + latency * 0.1

        # 절대 최솟값 (10ms 이하면 의미 없음)
        self._health_baseline[role] = max(new_baseline, 10)

    async def _measure_inference(self, role: str) -> float | None:
        """inference latency 측정 (lock으로 동시 1개 제한)."""
        adapter = self.classifier if role == "classifier" else self.reasoner
        if not adapter:
            return None
        async with self._measure_lock:
            latency = await adapter.measure_inference_latency()
            if latency >= 0:
                self._inference_latency[role] = latency
                self._inference_latency_at[role] = time.time()
                return latency
            return None

    async def get_load_status(self, role: str, force_measure: bool = False) -> dict:
        """role의 부하 상태 판단. inference latency는 캐시 또는 조건부 측정."""
        health_latency = self._latency.get(role, 0)
        baseline = self._health_baseline.get(role, 50)
        cached_inference = self._inference_latency.get(role)
        cached_at = self._inference_latency_at.get(role, 0)

        now = time.time()
        cache_valid = (now - cached_at) < 30
        cooldown_active = (now - cached_at) < 10

        # 조건부 측정 (상대값 + 쿨다운)
        should_measure = False
        if force_measure and not cooldown_active:
            should_measure = True
        elif not cache_valid and not cooldown_active:
            if health_latency > baseline * 3:
                should_measure = True

        if should_measure:
            inference_latency = await self._measure_inference(role)
        else:
            inference_latency = cached_inference if cache_valid else None

        # 부하 판단 (inference 우선, health는 fallback)
        if inference_latency is not None and inference_latency > 8000:
            load = "매우 바쁨"
        elif inference_latency is not None and inference_latency > 4000:
            load = "바쁨"
        elif health_latency > baseline * 5:
            load = "바쁨"
        elif health_latency > baseline * 2:
            load = "보통"
        else:
            load = "여유"

        # local queue로 보조 판단 (한 단계 상향)
        try:
            from services import job_queue as jq_module
            if jq_module.job_queue:
                active = jq_module.job_queue.stats.get("active", 0)
                if active > 0 and load == "보통":
                    load = "바쁨"
                elif active > 0 and load == "여유":
                    load = "보통"
        except Exception:
            pass

        return {
            "load": load,
            "health_ms": health_latency,
            "health_baseline_ms": round(baseline, 1),
            "inference_ms": inference_latency,
            "measured": inference_latency is not None,
        }

    def is_healthy(self, role: str) -> bool:
        return self._health.get(role, False)

    def health_summary(self) -> dict:
        result = {}
        for role, adapter in [("classifier", self.classifier), ("reasoner", self.reasoner)]:
            if adapter:
                result[role] = {
                    "name": adapter.name,
                    "model": adapter.model,
                    "healthy": self._health[role],
                    "latency_ms": self._latency[role],
                }
        return result


backend_registry = BackendRegistry()