21a78fbbf0
## 배경 1차 Phase 2.2 eval에서 발견: 23개 쿼리가 순차 호출되지만 각 request의 background analyzer task는 모두 동시에 MLX에 요청 날림 → MLX single-inference 서버 queue 폭발 → 22개가 15초 timeout. cache 채워지지 않음. ## 수정 ### query_analyzer.py - LLM_CONCURRENCY = 1 상수 추가 - _LLM_SEMAPHORE: lazy init asyncio.Semaphore (event loop 바인딩) - analyze() 내부: semaphore → timeout(실제 LLM 호출만) 이중 래핑 semaphore 대기 시간이 timeout에 포함되지 않도록 주의 ### run_eval.py - --analyze true|false 파라미터 추가 (Phase 2.1+ 측정용) - call_search / evaluate 시그니처에 analyze 전달 ## 기대 효과 - prewarm/background/동기 호출 모두 1개씩 순차 MLX 호출 - 23개 대기 시 최악 230초 소요, 단 모두 성공해서 cache 채움 - MLX 서버 부하 안정 Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
443 lines
15 KiB
Python
443 lines
15 KiB
Python
"""Query analyzer — 자연어 쿼리 분석 (Phase 2.1, async-only 구조).
|
|
|
|
**핵심 철학** (memory `feedback_analyzer_async_only.md`):
|
|
> QueryAnalyzer는 "즉시 실행하는 기능"이 아니라 "미리 준비해두는 기능"이다.
|
|
|
|
Retrieval 경로에서 analyzer를 **동기 호출 금지**.
|
|
동기 호출 가능한 API는 prewarm 전용.
|
|
|
|
## Pipeline
|
|
|
|
```
|
|
query → retrieval (항상 즉시)
|
|
↘ trigger_background_analysis (fire-and-forget)
|
|
→ analyze() [5초+] → cache 저장
|
|
|
|
다음 호출 (동일 쿼리) → get_cached() 히트 → Phase 2 파이프라인 활성화
|
|
```
|
|
|
|
## 룰 (plan 영구)
|
|
- `LLM_TIMEOUT_MS = 5000` (background 이므로 여유 OK)
|
|
- `MAX_NORMALIZED_QUERIES = 3` (multilingual explosion 방지)
|
|
- weight 합 = 1.0 정규화 필수 (fusion 왜곡 방지)
|
|
- 실패/저신뢰(< 0.5) 결과는 캐시 금지 (잘못된 분석 고정 방지)
|
|
- `analyzer_confidence` default `float 0.0` 강제 (None 금지)
|
|
- analyze() 동기 호출 금지. retrieval 경로는 `get_cached()` + `trigger_background_analysis()` 만 사용.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import hashlib
|
|
import time
|
|
from typing import Any
|
|
|
|
from ai.client import AIClient, _load_prompt, parse_json_response
|
|
from core.config import settings
|
|
from core.utils import setup_logger
|
|
|
|
logger = setup_logger("query_analyzer")
|
|
|
|
# ─── 상수 (plan 영구 룰) ────────────────────────────────
|
|
PROMPT_VERSION = "v2" # prompts/query_analyze.txt 축소판
|
|
CACHE_TTL = 86400 # 24h
|
|
CACHE_MAXSIZE = 1000
|
|
LLM_TIMEOUT_MS = 15000 # async 구조 (background), 동기 경로 금지
|
|
# ↑ 실측: gemma-4-26b-a4b-it-8bit MLX, 축소 프롬프트(prompt_tok=802) 7~11초.
|
|
# generation이 dominant (max_tokens 무효, 자연 EOS ~289 tok 생성).
|
|
# background 실행이라 15초도 안전. 상향 필요 시 여기서만 조정.
|
|
LLM_CONCURRENCY = 1 # MLX는 single-inference, 동시 호출 시 queue 폭발.
|
|
# ↑ 실측: 23 concurrent 요청 → 22개 15초 timeout. semaphore로 순차 강제.
|
|
# prewarm/background/동기 호출 모두 이 semaphore 경유.
|
|
MIN_CACHE_CONFIDENCE = 0.5 # 이 미만은 캐시 금지
|
|
MAX_NORMALIZED_QUERIES = 3
|
|
|
|
|
|
def _model_version() -> str:
|
|
"""현재 primary 모델 ID를 캐시 키에 반영."""
|
|
if settings.ai and settings.ai.primary:
|
|
return settings.ai.primary.model
|
|
return "unknown-model"
|
|
|
|
|
|
# ─── in-memory LRU (FIFO 근사) ──────────────────────────
|
|
_CACHE: dict[str, dict[str, Any]] = {}
|
|
|
|
# background task 참조 유지 (premature GC 방지)
|
|
_PENDING: set[asyncio.Task[Any]] = set()
|
|
# 동일 쿼리 중복 실행 방지 (진행 중인 쿼리 집합)
|
|
_INFLIGHT: set[str] = set()
|
|
# MLX concurrency 제한 (single-inference → 1)
|
|
# 첫 호출 시 lazy init (event loop이 준비된 후)
|
|
_LLM_SEMAPHORE: asyncio.Semaphore | None = None
|
|
|
|
|
|
def _get_llm_semaphore() -> asyncio.Semaphore:
|
|
"""첫 호출 시 현재 event loop에 바인딩된 semaphore 생성."""
|
|
global _LLM_SEMAPHORE
|
|
if _LLM_SEMAPHORE is None:
|
|
_LLM_SEMAPHORE = asyncio.Semaphore(LLM_CONCURRENCY)
|
|
return _LLM_SEMAPHORE
|
|
|
|
|
|
def _cache_key(query: str) -> str:
|
|
raw = f"{query}|{PROMPT_VERSION}|{_model_version()}"
|
|
return hashlib.sha256(raw.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def get_cached(query: str) -> dict | None:
|
|
"""TTL 경과 entry는 자동 삭제. 없으면 None.
|
|
|
|
retrieval 경로에서 cache hit 판단용으로 호출. 호출 자체는 O(1).
|
|
"""
|
|
key = _cache_key(query)
|
|
entry = _CACHE.get(key)
|
|
if not entry:
|
|
return None
|
|
if time.time() - entry["ts"] > CACHE_TTL:
|
|
_CACHE.pop(key, None)
|
|
return None
|
|
return entry["value"]
|
|
|
|
|
|
def set_cached(query: str, value: dict) -> None:
|
|
"""저신뢰(< 0.5) / 빈 값은 캐시 금지. 상한 초과 시 FIFO eviction."""
|
|
if not value:
|
|
return
|
|
try:
|
|
conf = float(value.get("analyzer_confidence", 0.0) or 0.0)
|
|
except (TypeError, ValueError):
|
|
conf = 0.0
|
|
if conf < MIN_CACHE_CONFIDENCE:
|
|
return
|
|
key = _cache_key(query)
|
|
if key in _CACHE:
|
|
_CACHE[key] = {"value": value, "ts": time.time()}
|
|
return
|
|
if len(_CACHE) >= CACHE_MAXSIZE:
|
|
try:
|
|
oldest = next(iter(_CACHE))
|
|
_CACHE.pop(oldest, None)
|
|
except StopIteration:
|
|
pass
|
|
_CACHE[key] = {"value": value, "ts": time.time()}
|
|
|
|
|
|
def cache_stats() -> dict[str, int]:
|
|
"""debug/운영용 — 현재 캐시 크기 + inflight 수."""
|
|
return {
|
|
"size": len(_CACHE),
|
|
"maxsize": CACHE_MAXSIZE,
|
|
"inflight": len(_INFLIGHT),
|
|
"pending_tasks": len(_PENDING),
|
|
}
|
|
|
|
|
|
# ─── weight 정규화 (fusion 왜곡 방지) ───────────────────
|
|
def _normalize_weights(analysis: dict) -> dict:
|
|
"""normalized_queries를 MAX_NORMALIZED_QUERIES로 자르고 weight 합=1.0 정규화."""
|
|
queries = analysis.get("normalized_queries")
|
|
if not isinstance(queries, list) or not queries:
|
|
return analysis
|
|
|
|
sanitized: list[dict] = []
|
|
for q in queries[:MAX_NORMALIZED_QUERIES]:
|
|
if not isinstance(q, dict):
|
|
continue
|
|
lang = str(q.get("lang", "")).strip() or "ko"
|
|
text = str(q.get("text", "")).strip()
|
|
if not text:
|
|
continue
|
|
try:
|
|
w = float(q.get("weight", 1.0))
|
|
if w < 0:
|
|
w = 0.0
|
|
except (TypeError, ValueError):
|
|
w = 1.0
|
|
sanitized.append({"lang": lang, "text": text, "weight": w})
|
|
|
|
if not sanitized:
|
|
return analysis
|
|
|
|
total = sum(q["weight"] for q in sanitized)
|
|
if total <= 0:
|
|
equal = 1.0 / len(sanitized)
|
|
for q in sanitized:
|
|
q["weight"] = equal
|
|
else:
|
|
for q in sanitized:
|
|
q["weight"] /= total
|
|
|
|
analysis["normalized_queries"] = sanitized
|
|
return analysis
|
|
|
|
|
|
# ─── 프롬프트 로딩 (module 초기화 1회) ──────────────────
|
|
try:
|
|
ANALYZE_PROMPT = _load_prompt("query_analyze.txt")
|
|
except FileNotFoundError:
|
|
ANALYZE_PROMPT = ""
|
|
logger.warning("query_analyze.txt not found — analyzer will always return fallback")
|
|
|
|
|
|
# ─── 기본 fallback 응답 (None 금지) ─────────────────────
|
|
def _fallback(reason: str | None = None) -> dict:
|
|
"""LLM 실패/timeout/parse 실패 시 반환. analyzer_confidence는 반드시 float 0.0."""
|
|
result: dict[str, Any] = {
|
|
"intent": "semantic_search",
|
|
"query_type": "keyword",
|
|
"domain_hint": "mixed",
|
|
"language_scope": "limited",
|
|
"keywords": [],
|
|
"must_terms": [],
|
|
"optional_terms": [],
|
|
"hard_filters": {},
|
|
"soft_filters": {"domain": [], "document_type": []},
|
|
"normalized_queries": [],
|
|
"expanded_terms": [],
|
|
"synonyms": {},
|
|
"analyzer_confidence": 0.0,
|
|
}
|
|
if reason:
|
|
result["_fallback_reason"] = reason
|
|
return result
|
|
|
|
|
|
# ─── 메인 LLM 호출 (내부 사용) ──────────────────────────
|
|
async def analyze(query: str, ai_client: AIClient | None = None) -> dict:
|
|
"""쿼리 분석 결과 반환. 실패 시 analyzer_confidence=0.0 fallback.
|
|
|
|
**⚠️ 동기 검색 경로에서 직접 호출 금지**. 용도:
|
|
- `trigger_background_analysis` 내부 호출
|
|
- `prewarm_analyzer` startup 호출
|
|
- 디버깅/테스트
|
|
|
|
Args:
|
|
query: 사용자 쿼리 원문
|
|
ai_client: AIClient 인스턴스 (없으면 생성 후 자동 close)
|
|
|
|
Returns:
|
|
dict — 최소 `analyzer_confidence` 키는 항상 float로 존재.
|
|
"""
|
|
if not query or not query.strip():
|
|
return _fallback("empty_query")
|
|
|
|
if not ANALYZE_PROMPT:
|
|
return _fallback("prompt_not_loaded")
|
|
|
|
# cache hit 즉시 반환 (prewarm 재호출 방지)
|
|
cached = get_cached(query)
|
|
if cached is not None:
|
|
return cached
|
|
|
|
client_owned = False
|
|
if ai_client is None:
|
|
ai_client = AIClient()
|
|
client_owned = True
|
|
|
|
t_start = time.perf_counter()
|
|
semaphore = _get_llm_semaphore()
|
|
# ⚠️ 중요: semaphore 대기는 timeout 포함되면 안됨 (대기만 해도 timeout 발동)
|
|
# timeout은 실제 LLM 호출 구간에만 적용.
|
|
try:
|
|
async with semaphore:
|
|
async with asyncio.timeout(LLM_TIMEOUT_MS / 1000):
|
|
raw = await ai_client._call_chat(
|
|
ai_client.ai.primary,
|
|
ANALYZE_PROMPT.replace("{query}", query),
|
|
)
|
|
except asyncio.TimeoutError:
|
|
elapsed = (time.perf_counter() - t_start) * 1000
|
|
logger.warning(
|
|
"query_analyze timeout query=%r elapsed_ms=%.0f (LLM_TIMEOUT_MS=%d)",
|
|
query[:80],
|
|
elapsed,
|
|
LLM_TIMEOUT_MS,
|
|
)
|
|
return _fallback("timeout")
|
|
except Exception as exc:
|
|
elapsed = (time.perf_counter() - t_start) * 1000
|
|
logger.warning(
|
|
"query_analyze LLM error query=%r elapsed_ms=%.0f err=%r",
|
|
query[:80],
|
|
elapsed,
|
|
exc,
|
|
)
|
|
return _fallback(f"llm_error:{type(exc).__name__}")
|
|
finally:
|
|
if client_owned:
|
|
try:
|
|
await ai_client.close()
|
|
except Exception:
|
|
pass
|
|
|
|
elapsed_ms = (time.perf_counter() - t_start) * 1000
|
|
|
|
parsed = parse_json_response(raw)
|
|
if not isinstance(parsed, dict):
|
|
logger.warning(
|
|
"query_analyze parse failed query=%r elapsed_ms=%.0f raw=%r",
|
|
query[:80],
|
|
elapsed_ms,
|
|
(raw or "")[:200],
|
|
)
|
|
return _fallback("parse_failed")
|
|
|
|
try:
|
|
conf = float(parsed.get("analyzer_confidence", 0.0) or 0.0)
|
|
except (TypeError, ValueError):
|
|
conf = 0.0
|
|
parsed["analyzer_confidence"] = conf
|
|
parsed = _normalize_weights(parsed)
|
|
|
|
logger.info(
|
|
"query_analyze ok query=%r conf=%.2f intent=%s domain=%s elapsed_ms=%.0f",
|
|
query[:80],
|
|
conf,
|
|
parsed.get("intent"),
|
|
parsed.get("domain_hint"),
|
|
elapsed_ms,
|
|
)
|
|
|
|
set_cached(query, parsed)
|
|
return parsed
|
|
|
|
|
|
# ─── Background trigger (retrieval 경로에서 사용) ───────
|
|
async def _background_analyze(query: str) -> None:
|
|
"""Background task wrapper — inflight 집합 관리 + 예외 삼킴."""
|
|
try:
|
|
await analyze(query)
|
|
except Exception as exc:
|
|
logger.warning("background analyze crashed query=%r err=%r", query[:80], exc)
|
|
finally:
|
|
_INFLIGHT.discard(query)
|
|
|
|
|
|
def trigger_background_analysis(query: str) -> bool:
|
|
"""retrieval 경로에서 호출. cache miss 시 background task 생성.
|
|
|
|
- 동기 함수. 즉시 반환.
|
|
- 이미 inflight 또는 cache hit이면 아무 작업 X, False 반환.
|
|
- 새 task 생성 시 True 반환.
|
|
|
|
Returns:
|
|
bool — task 실제로 생성되었는지 여부
|
|
"""
|
|
if not query or not query.strip():
|
|
return False
|
|
if query in _INFLIGHT:
|
|
return False
|
|
if get_cached(query) is not None:
|
|
return False
|
|
|
|
try:
|
|
loop = asyncio.get_running_loop()
|
|
except RuntimeError:
|
|
logger.warning("trigger_background_analysis called outside event loop")
|
|
return False
|
|
|
|
_INFLIGHT.add(query)
|
|
task = loop.create_task(_background_analyze(query))
|
|
_PENDING.add(task)
|
|
task.add_done_callback(_PENDING.discard)
|
|
return True
|
|
|
|
|
|
# ─── Prewarm (app startup) ──────────────────────────────
|
|
# 운영에서 자주 발생하는 쿼리 샘플. 통계 기반으로 확장 예정.
|
|
DEFAULT_PREWARM_QUERIES: list[str] = [
|
|
# fixed 7 (Phase 2 평가셋 core)
|
|
"산업안전보건법 제6장",
|
|
"기계 사고 관련 법령",
|
|
"AI 산업 동향",
|
|
"Python async best practice",
|
|
"유해화학물질을 다루는 회사가 지켜야 할 안전 의무",
|
|
"recent AI safety news from Europe",
|
|
"이세상에 존재하지 않는 문서명",
|
|
# 법령 관련
|
|
"산업안전보건법 시행령",
|
|
"화학물질관리법",
|
|
"위험성평가 절차",
|
|
# 뉴스 관련
|
|
"EU AI Act",
|
|
"한국 AI 산업",
|
|
# 실무
|
|
"안전보건 교육 자료",
|
|
"사고 조사 보고서",
|
|
"MSDS 작성 방법",
|
|
]
|
|
|
|
|
|
async def prewarm_analyzer(
|
|
queries: list[str] | None = None,
|
|
delay_between: float = 0.2,
|
|
) -> dict[str, Any]:
|
|
"""app startup에서 호출. 대표 쿼리를 미리 분석해 cache에 적재.
|
|
|
|
Non-blocking으로 사용: `asyncio.create_task(prewarm_analyzer())`.
|
|
|
|
Args:
|
|
queries: 분석할 쿼리 리스트. None이면 DEFAULT_PREWARM_QUERIES 사용.
|
|
delay_between: 각 쿼리 간 대기 시간 (MLX 부하 완화).
|
|
|
|
Returns:
|
|
dict — {queries_total, success, failed, elapsed_ms, cache_size_after}
|
|
"""
|
|
if not ANALYZE_PROMPT:
|
|
logger.warning("prewarm skipped — prompt not loaded")
|
|
return {"status": "skipped", "reason": "prompt_not_loaded"}
|
|
|
|
targets = queries if queries is not None else DEFAULT_PREWARM_QUERIES
|
|
total = len(targets)
|
|
success = 0
|
|
failed = 0
|
|
t_start = time.perf_counter()
|
|
|
|
logger.info("prewarm_analyzer start queries=%d timeout_ms=%d", total, LLM_TIMEOUT_MS)
|
|
|
|
client = AIClient()
|
|
try:
|
|
for i, q in enumerate(targets, 1):
|
|
if get_cached(q) is not None:
|
|
logger.info("prewarm skip (already cached) [%d/%d] %r", i, total, q[:40])
|
|
success += 1
|
|
continue
|
|
result = await analyze(q, ai_client=client)
|
|
conf = float(result.get("analyzer_confidence", 0.0) or 0.0)
|
|
if conf >= MIN_CACHE_CONFIDENCE:
|
|
success += 1
|
|
logger.info("prewarm ok [%d/%d] conf=%.2f q=%r", i, total, conf, q[:40])
|
|
else:
|
|
failed += 1
|
|
reason = result.get("_fallback_reason", "low_conf")
|
|
logger.warning(
|
|
"prewarm fail [%d/%d] reason=%s q=%r", i, total, reason, q[:40]
|
|
)
|
|
if delay_between > 0 and i < total:
|
|
await asyncio.sleep(delay_between)
|
|
finally:
|
|
try:
|
|
await client.close()
|
|
except Exception:
|
|
pass
|
|
|
|
elapsed_ms = (time.perf_counter() - t_start) * 1000
|
|
stats = cache_stats()
|
|
logger.info(
|
|
"prewarm_analyzer done total=%d success=%d failed=%d elapsed_ms=%.0f cache_size=%d",
|
|
total,
|
|
success,
|
|
failed,
|
|
elapsed_ms,
|
|
stats["size"],
|
|
)
|
|
return {
|
|
"status": "complete",
|
|
"queries_total": total,
|
|
"success": success,
|
|
"failed": failed,
|
|
"elapsed_ms": elapsed_ms,
|
|
"cache_size_after": stats["size"],
|
|
}
|