aleksey
6fd2f0e689
LLM classify via LiteLLM gigachat-classifier when heuristic confidence is low. CLASSIFIER_MODE=heuristic|hybrid|llm. Metrics classifier_source.
127 lines
4.5 KiB
Python
127 lines
4.5 KiB
Python
"""Optional LLM tier classifier via LiteLLM (GigaChat-2-Lite)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
from rules_loader import load_orchestration
|
|
|
|
log = logging.getLogger("classifier_llm")
|
|
|
|
TEXT_TIERS = frozenset(
|
|
{"SIMPLE", "MEDIUM_OPS", "MEDIUM_CODE", "COMPLEX", "REASONING"},
|
|
)
|
|
|
|
CLASSIFY_SYSTEM = """You classify coding-assistant requests into exactly one tier.
|
|
Reply with ONLY valid JSON, no markdown:
|
|
{"tier":"SIMPLE|MEDIUM_OPS|MEDIUM_CODE|COMPLEX|REASONING","confidence":0.0-1.0}
|
|
|
|
SIMPLE — greetings, definitions, short questions
|
|
MEDIUM_OPS — bash, docker, devops, infrastructure
|
|
MEDIUM_CODE — write/refactor code, functions, bugs
|
|
COMPLEX — architecture, migrations, system design
|
|
REASONING — step-by-step proof, deep analysis"""
|
|
|
|
|
|
class LlmClassifier:
|
|
def __init__(self) -> None:
|
|
orch = load_orchestration()
|
|
clf = orch.get("classifier", {})
|
|
self.mode = os.environ.get("CLASSIFIER_MODE", clf.get("mode", "hybrid")).lower()
|
|
self.model = os.environ.get(
|
|
"CLASSIFIER_LLM_MODEL",
|
|
clf.get("llm_model", "gigachat-classifier"),
|
|
)
|
|
self.low_conf = float(clf.get("low_confidence_threshold", 0.6))
|
|
self.litellm_url = os.environ.get("LITELLM_INTERNAL_URL", "http://litellm:4000").rstrip("/")
|
|
self.litellm_key = os.environ.get("LITELLM_MASTER_KEY", "")
|
|
self.timeout = float(clf.get("timeout_sec", 15))
|
|
self.max_tokens = int(clf.get("max_tokens", 64))
|
|
self.enabled = self.mode in ("hybrid", "llm") and bool(self.litellm_key)
|
|
|
|
def should_use_llm(self, tier_value: str, confidence: float, has_image: bool) -> bool:
|
|
if not self.enabled or has_image:
|
|
return False
|
|
if tier_value.startswith("VISION"):
|
|
return False
|
|
if self.mode == "llm":
|
|
return True
|
|
if self.mode == "hybrid":
|
|
return confidence < self.low_conf
|
|
return False
|
|
|
|
@staticmethod
|
|
def _parse_json(content: str) -> dict[str, Any] | None:
|
|
text = content.strip()
|
|
fence = re.search(r"```(?:json)?\s*([\s\S]*?)```", text)
|
|
if fence:
|
|
text = fence.group(1).strip()
|
|
try:
|
|
data = json.loads(text)
|
|
return data if isinstance(data, dict) else None
|
|
except json.JSONDecodeError:
|
|
match = re.search(r"\{[\s\S]*\}", text)
|
|
if not match:
|
|
return None
|
|
try:
|
|
data = json.loads(match.group(0))
|
|
return data if isinstance(data, dict) else None
|
|
except json.JSONDecodeError:
|
|
return None
|
|
|
|
async def classify(self, text: str) -> tuple[str, float] | None:
|
|
snippet = text.strip()[:4000]
|
|
if not snippet:
|
|
return None
|
|
|
|
payload = {
|
|
"model": self.model,
|
|
"max_tokens": self.max_tokens,
|
|
"temperature": 0,
|
|
"messages": [
|
|
{"role": "system", "content": CLASSIFY_SYSTEM},
|
|
{"role": "user", "content": snippet},
|
|
],
|
|
}
|
|
headers = {
|
|
"Authorization": f"Bearer {self.litellm_key}",
|
|
"Content-Type": "application/json",
|
|
}
|
|
|
|
try:
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
resp = await client.post(
|
|
f"{self.litellm_url}/v1/chat/completions",
|
|
headers=headers,
|
|
json=payload,
|
|
)
|
|
if resp.status_code >= 400:
|
|
log.warning("LLM classify HTTP %s: %s", resp.status_code, resp.text[:200])
|
|
return None
|
|
data = resp.json()
|
|
content = (
|
|
data.get("choices", [{}])[0]
|
|
.get("message", {})
|
|
.get("content", "")
|
|
)
|
|
parsed = self._parse_json(content)
|
|
if not parsed:
|
|
log.warning("LLM classify: invalid JSON in response: %s", content[:120])
|
|
return None
|
|
tier_raw = str(parsed.get("tier", "")).upper().replace("-", "_")
|
|
if tier_raw not in TEXT_TIERS:
|
|
log.warning("LLM classify: unknown tier %s", tier_raw)
|
|
return None
|
|
confidence = float(parsed.get("confidence", 0.75))
|
|
confidence = max(0.0, min(1.0, confidence))
|
|
return tier_raw, confidence
|
|
except Exception as exc: # noqa: BLE001
|
|
log.warning("LLM classify failed: %s", exc)
|
|
return None
|