"""BackendRegistry — 모델 어댑터 관리 + 헬스체크 루프.""" from __future__ import annotations import asyncio import logging import time from services.model_adapter import ModelAdapter logger = logging.getLogger(__name__) CLASSIFIER_PROMPT = """\ 너는 AI 라우터다. 사용��� 메시지를 분석하여 JSON으로 응답하라. 반드시 아래 4가�� action 중 하나를 선택하라: 1. "direct" — 인사, 잡담, 간단한 질문, 자기소개 요청 등. 네가 직접 답변한다. 2. "route" — 복잡한 질문, 분석, 설명, 코딩 등. 추론 모델에게 넘긴다. prompt 필드에 추론 모델용 프롬프트를 작성하라. 3. "clarify" — 질문이 모호하거나 정보가 부족할 때. 사용자에게 추가 질문���다. 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._status_cache: dict[str, dict | None] = {} self._status_cached_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) # /status 같이 캐시 (지원하는 백엔드만) status_data = await adapter.status_check() self._status_cache[role] = status_data self._status_cached_at[role] = time.time() 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의 부하 상태 판단. /status API 우선, 없으면 hybrid (health baseline + 조건부 ping).""" health_latency = self._latency.get(role, 0) baseline = self._health_baseline.get(role, 50) adapter = self.classifier if role == "classifier" else self.reasoner # === 1순위: /status API (GPU 무접촉) === # force_measure 시 fresh 호출, 아니면 캐시 사용 status_data = None if force_measure and adapter: status_data = await adapter.status_check() if status_data: self._status_cache[role] = status_data self._status_cached_at[role] = time.time() else: status_data = self._status_cache.get(role) if status_data is not None: active_jobs = status_data.get("active_jobs", 0) # 부드러운 임계치 if active_jobs >= 4: load = "매우 바쁨" elif active_jobs >= 2: load = "바쁨" elif active_jobs == 1: load = "보통" else: load = "여유" return { "load": load, "source": "status_api", "active_jobs": active_jobs, "total_requests": status_data.get("total_requests", 0), "health_ms": health_latency, "health_baseline_ms": round(baseline, 1), "inference_ms": None, "measured": False, } # === 2순위: hybrid fallback (status 미지원 백엔드) === 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 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 보조 판단 (자기 자신 제외 — active > 1) 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 > 1 and load == "보통": load = "바쁨" elif active > 1 and load == "여유": load = "보통" except Exception: pass return { "load": load, "source": "hybrid", "active_jobs": None, "total_requests": None, "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()