e786307a07
- _parse_classification: 어떤 형태든 첫{~마지막} 추출, 백틱 잔재 제거
- 분류기: 판단 예시 추가 (일정→tools, 인사→direct 등), 백틱/코드블록 금지 명시
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
135 lines
5.6 KiB
Python
135 lines
5.6 KiB
Python
"""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, description) — 일정 생성 초안. 시간은 HH:MM. 실제 생성 아님
|
||
- 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_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 prev != healthy:
|
||
status = "UP" if healthy else "DOWN"
|
||
logger.warning("%s (%s) → %s (%.0fms)", adapter.name, role, status, elapsed)
|
||
|
||
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()
|