feat: 분석 API를 1차(계획)→2차(분석) 흐름으로 맞춤

app/agents/intent/__init__.py

@@ -2,12 +2,16 @@
 
 from .service import (
     FixedExpiryRenewalIntentAgent,
+    HybridHfIntentAgent,
     IntentClassifier,
+    IntentResult,
     build_intent_agent,
 )
 
 __all__ = [
     "FixedExpiryRenewalIntentAgent",
+    "HybridHfIntentAgent",
     "IntentClassifier",
+    "IntentResult",
     "build_intent_agent",
-]
+]

app/agents/intent/guardrail.py

@@ -0,0 +1,76 @@
+# BERT 예측을 A.X로 넘길지 결정하는 라우팅 규칙
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+
+
+@dataclass
+# 가드레일 라우팅 결과
+class RoutingResult:
+
+    should_route: bool
+    reason: str
+    category: str
+
+
+@dataclass
+# HR Intent BERT→A.X 라우팅 가드레일
+class HRRoutingGuardrail:
+
+    margin_threshold: float = 0.76
+    max_trained_labels: int = 3
+    label_prob_threshold: float = 0.55
+    status_kw: list[str] = field(
+        default_factory=lambda: ["없음", "완료", "이상없", "특이사항", "특이문의"]
+    )
+    action_kw: list[str] = field(default_factory=lambda: ["배치", "라인", "지시"])
+    doc_kw: list[str] = field(
+        default_factory=lambda: ["신청서", "서류", "챙겨", "접수", "명단확인"]
+    )
+
+    @staticmethod
+    # 공백 제거 정규화
+    def _normalize(text: str) -> str:
+        return text.replace(" ", "")
+
+    # A.X 호출 여부·사유 판정
+    def should_route_to_ax(
+        self, hr_input: str, probs: dict[str, float], margin: float
+    ) -> RoutingResult:
+        clean_input = self._normalize(hr_input)
+        activated_count = sum(1 for p in probs.values() if p >= self.label_prob_threshold)
+        if activated_count >= self.max_trained_labels:
+            return RoutingResult(
+                should_route=True,
+                reason=f"활성 label {activated_count}개 (학습 최댓값 {self.max_trained_labels}개 이상)",
+                category="OOD_Label_Count",
+            )
+        if any(kw in clean_input for kw in self.status_kw):
+            return RoutingResult(
+                should_route=True, reason="완료/상태 보고 키워드 감지", category="Rule_Status"
+            )
+        action_matches = sum(1 for kw in self.action_kw if kw in clean_input)
+        if action_matches >= 2:
+            return RoutingResult(
+                should_route=True,
+                reason=f"배치/라인/지시 키워드 {action_matches}개 감지",
+                category="Rule_Action",
+            )
+        if "급여계좌" in clean_input or ("급여" in clean_input and "확인" in clean_input):
+            return RoutingResult(
+                should_route=True, reason="급여계좌 관련 경계 키워드 감지", category="Rule_Salary"
+            )
+        if any(kw in clean_input for kw in self.doc_kw):
+            return RoutingResult(
+                should_route=True, reason="서류 확보 키워드 감지", category="Rule_Document"
+            )
+        if margin < self.margin_threshold:
+            return RoutingResult(
+                should_route=True,
+                reason=f"margin {margin:.3f} < {self.margin_threshold}",
+                category="Low_Margin",
+            )
+        return RoutingResult(
+            should_route=False, reason="BERT 신뢰 구간 통과", category="Pass_BERT"
+        )

app/agents/intent/hybrid.py

@@ -0,0 +1,100 @@
+# BERT + 가드레일 + (선택) A.X 하이브리드 Intent 추론
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass, field
+
+from .guardrail import HRRoutingGuardrail
+from .models_hf import AxIntentModel, BertIntentModel
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+# 하이브리드 분류 한 건의 정규화 결과
+class HybridIntentPrediction:
+
+    intents: list[str]
+    scores: dict[str, float] = field(default_factory=dict)
+    evidence: dict[str, str | None] = field(default_factory=dict)
+    selected_model: str = "BERT"
+    degraded: bool = False
+
+
+# BERT 우선·필요 시 A.X 보조 파이프라인
+class HybridIntentPipeline:
+
+    # 설정값으로 BERT·가드레일·선택적 A.X 구성
+    def __init__(
+        self,
+        *,
+        bert_model_dir: str,
+        device: str = "cpu",
+        label_prob_threshold: float = 0.55,
+        margin_threshold: float = 0.76,
+        max_trained_labels: int = 3,
+        hf_token: str | None = None,
+        enable_ax: bool = False,
+        ax_base_model_name: str = "skt/A.X-4.0-Light",
+        ax_adapter_path: str = "fowoco/ax-intent-qlora",
+        ax_max_new_tokens: int = 96,
+    ) -> None:
+        self.bert = BertIntentModel(
+            model_dir=bert_model_dir,
+            device=device,
+            label_prob_threshold=label_prob_threshold,
+            hf_token=hf_token,
+        )
+        self.guardrail = HRRoutingGuardrail(
+            margin_threshold=margin_threshold,
+            max_trained_labels=max_trained_labels,
+            label_prob_threshold=label_prob_threshold,
+        )
+        self.ax: AxIntentModel | None = None
+        if enable_ax:
+            try:
+                self.ax = AxIntentModel(
+                    base_model_name=ax_base_model_name,
+                    adapter_path=ax_adapter_path,
+                    device=self.bert.device,
+                    max_new_tokens=ax_max_new_tokens,
+                    hf_token=hf_token,
+                )
+            except Exception:
+                logger.exception("A.X load failed — BERT-only degraded mode")
+
+    # instruction → 정규화 Intent 예측
+    def predict(self, instruction: str) -> HybridIntentPrediction:
+        probs, margin, bert_intents = self.bert.predict(instruction)
+        route = self.guardrail.should_route_to_ax(instruction, probs, margin)
+        if route.should_route and self.ax is not None:
+            try:
+                ax_items = self.ax.predict(instruction)
+                intents = [str(i.get("intent")) for i in ax_items if i.get("intent")]
+                evidence = {
+                    str(i.get("intent")): i.get("evidence")
+                    for i in ax_items
+                    if i.get("intent")
+                }
+                return HybridIntentPrediction(
+                    intents=intents or bert_intents,
+                    scores=probs,
+                    evidence=evidence,
+                    selected_model="AX",
+                    degraded=False,
+                )
+            except Exception:
+                logger.exception("A.X inference failed — BERT fallback")
+                return HybridIntentPrediction(
+                    intents=bert_intents,
+                    scores=probs,
+                    selected_model="BERT_FALLBACK",
+                    degraded=True,
+                )
+        return HybridIntentPrediction(
+            intents=bert_intents,
+            scores=probs,
+            selected_model="BERT",
+            degraded=False,
+        )

app/agents/intent/models_hf.py

@@ -0,0 +1,132 @@
+# klue/roberta-base Intent 분류 모델 로드·추론
+
+from __future__ import annotations
+
+from typing import Any
+
+
+# BERT multilabel Intent 분류기
+class BertIntentModel:
+
+    # HF Hub/로컬 경로에서 분류기 로드
+    def __init__(
+        self,
+        model_dir: str,
+        device: str,
+        label_prob_threshold: float = 0.55,
+        hf_token: str | None = None,
+    ) -> None:
+        import torch
+        from transformers import AutoModelForSequenceClassification, AutoTokenizer
+
+        self._torch = torch
+        self.device = (
+            "cuda"
+            if (device == "auto" and torch.cuda.is_available())
+            else (device if device != "auto" else "cpu")
+        )
+        self.tokenizer = AutoTokenizer.from_pretrained(model_dir, token=hf_token)
+        self.model = AutoModelForSequenceClassification.from_pretrained(
+            model_dir, token=hf_token
+        )
+        self.model.to(self.device).eval()
+        self.id2label = self.model.config.id2label
+        self.label_prob_threshold = label_prob_threshold
+
+    # 확률 dict·margin·활성 intent 목록
+    def predict(self, text: str) -> tuple[dict[str, float], float, list[str]]:
+        torch = self._torch
+        with torch.no_grad():
+            enc = self.tokenizer(
+                text, truncation=True, max_length=64, return_tensors="pt"
+            ).to(self.device)
+            logits = self.model(**enc).logits
+            probs_array = torch.sigmoid(logits)[0].cpu().numpy()
+
+        probs_dict = {self.id2label[i]: float(p) for i, p in enumerate(probs_array)}
+        activated = [p for p in probs_array if p >= self.label_prob_threshold]
+        not_activated = [p for p in probs_array if p < self.label_prob_threshold]
+        if not activated:
+            margin = float(max(probs_array)) - self.label_prob_threshold
+        else:
+            margin = float(min(activated)) - (
+                float(max(not_activated)) if not_activated else 0.0
+            )
+        picked = [
+            self.id2label[i]
+            for i, p in enumerate(probs_array)
+            if p >= self.label_prob_threshold
+        ]
+        if not picked:
+            picked = [self.id2label[int(probs_array.argmax())]]
+        return probs_dict, margin, picked
+
+
+# A.X-4.0-Light QLoRA Intent 보조 모델 (GPU·bitsandbytes 필요)
+class AxIntentModel:
+
+    _SYSTEM_PROMPT = (
+        "당신은 HR 업무 요청 문장(hr_input)을 분석하여 의도(Intent)를 분류하는 전문 AI 에이전트입니다.\n"
+        "Intent + evidence 추출까지가 책임입니다.\n"
+        '출력은 JSON만: {"intents": [{"intent": "INTENT_CODE", "evidence": "...|null"}]}'
+    )
+
+    # 4bit 베이스 + Peft 어댑터 로드
+    def __init__(
+        self,
+        base_model_name: str,
+        adapter_path: str,
+        device: str,
+        max_new_tokens: int = 96,
+        hf_token: str | None = None,
+    ) -> None:
+        import torch
+        from peft import PeftModel
+        from transformers import AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig
+
+        self._torch = torch
+        self.max_new_tokens = max_new_tokens
+        bnb_config = BitsAndBytesConfig(
+            load_in_4bit=True,
+            bnb_4bit_quant_type="nf4",
+            bnb_4bit_compute_dtype=torch.float16,
+        )
+        base_model = AutoModelForCausalLM.from_pretrained(
+            base_model_name,
+            quantization_config=bnb_config,
+            torch_dtype=torch.float16,
+            device_map={"": 0} if device != "cpu" else "cpu",
+            token=hf_token,
+        )
+        self.tokenizer = AutoTokenizer.from_pretrained(base_model_name, token=hf_token)
+        self.model = PeftModel.from_pretrained(base_model, adapter_path, token=hf_token)
+        self.model.eval()
+
+    # Intent 목록 [{"intent","evidence"}] — 실패 시 예외
+    def predict(self, hr_input: str) -> list[dict[str, Any]]:
+        import json
+        import re
+
+        messages = [
+            {"role": "system", "content": self._SYSTEM_PROMPT},
+            {"role": "user", "content": hr_input},
+        ]
+        inputs = self.tokenizer.apply_chat_template(
+            messages,
+            add_generation_prompt=True,
+            tokenize=True,
+            return_dict=True,
+            return_tensors="pt",
+        ).to(self.model.device)
+        with self._torch.no_grad():
+            output = self.model.generate(
+                **inputs, max_new_tokens=self.max_new_tokens, do_sample=False
+            )
+        raw = self.tokenizer.decode(
+            output[0][inputs["input_ids"].shape[-1] :], skip_special_tokens=True
+        )
+        match = re.search(r"\{.*\}", raw, re.DOTALL)
+        if not match:
+            raise ValueError(f"A.X output could not be parsed as JSON: {raw!r}")
+        parsed = json.loads(match.group(0))
+        return list(parsed.get("intents") or [])