fingu-tips-rnd-evaluation 파이프라인 구조 분석

# agents.py:191-197 (분기점)
def classify_scenario(input: str, trial: int = 0) -> str:
    if os.getenv("ENABLE_FINETUNED_PIPELINE") == "true":
        return _finetuned_classify(input)
    if os.getenv("ENABLE_REAL_PIPELINE") == "true":
        return _real_pipeline_classify(input)
    return _stub_classify(input, trial=trial)

# agents.py:104-127 (stub)
def _stub_classify(input: str, trial: int = 0) -> str:
    rng = random.Random(seed_prefix + input + str(trial))
    target = float(os.getenv("KPI1_STUB_TARGET", "0.74"))   # ← 점수 직접 조정
    refs = _get_truth_map().get(input, [])                  # ← 정답 lookup
    if refs and rng.random() < target:
        return refs[0]                                       # ← target 확률로 정답
    return rng.choice(ANALYSIS_TOOLS)                        # ← 무작위 오답 (16개)

# agents.py:174-180 (real)
def _real_pipeline_classify(input: str) -> str:
    prompt = _build_scenario_prompt(input)
    raw = generate_text("FINGU-AI/FinguAI-Chat-v1", prompt, max_new_tokens=24)
    return _parse_scenario_label(raw)   # 16 ID 정확매칭 → 한국어 키워드 fallback

핵심 발견: real 모드에서 모델이 "A)" 같은 무의미 응답 반환 → parser fallback이 모두 budget_planning으로 분류 → F1 = 0.0 (20건 subset 측정).

# thread_titles.py:244-259 (분기)
def classify_intent(input: str, trial: int = 0) -> str:
    if os.getenv("ENABLE_FINETUNED_PIPELINE") == "true":
        return _finetuned_classify(input)        # ← vLLM (운영 ON)
    if os.getenv("ENABLE_API_PIPELINE") == "true":
        return _api_pipeline_classify(input)
    if os.getenv("ENABLE_REAL_PIPELINE") == "true":
        return _real_pipeline_classify(input)
    return _stub_classify(input, trial=trial)

# thread_titles.py:198-241 (finetuned vLLM)
def _finetuned_classify(input: str) -> str:
    base = (os.getenv("FINETUNED_KPI2_API_URL")
            or os.getenv("FINETUNED_API_URL", "")).rstrip("/")
    model_id = os.getenv("FINETUNED_KPI2_MODEL_ID", "kpi2")
    r = httpx.post(f"{base}/chat/completions", json={
        "model": model_id, "temperature": 0.0, "max_tokens": 10,
        "messages": [{"role":"system", ...}, {"role":"user","content":input}],
    }, headers={"Authorization": f"Bearer {api_key}"})

# thread_titles.py:95-118 (stub)
def _stub_classify(input: str, trial: int = 0) -> str:
    target = float(os.getenv("KPI2_STUB_TARGET", "0.995"))  # accuracy 99.5%
    refs = _load_truth_map().get(input, [])
    if refs and random.Random(seed+input).random() < target:
        return refs[0]
    return random.choice(INTENTS)   # 8개 의도 중 random

real 측정: 100.00 (vLLM Qwen3-8B-KPI2-LoRA 진짜 작동) · 합격선 99.0

# finetune_inference.py:234-246 (분기)
def generate(input: str, trial: int = 0) -> str:
    if os.getenv("ENABLE_FINETUNED_PIPELINE") == "true":
        return _finetuned_generate(input)        # ← vLLM
    if os.getenv("ENABLE_REAL_PIPELINE") == "true":
        return _real_pipeline_generate(input)    # ← Qwen-Orpo-v1 14GB
    return _stub_generate(input)

# finetune_inference.py:100-117 (stub)
def _stub_generate(input: str) -> str:
    target = float(os.getenv("KPI3_STUB_TARGET", "0.90"))    # corpus BLEU ~81
    ref = _load_truth_map().get(input)
    if rng.random() < target: return ref                     # 정답 그대로
    return _mangle(ref, keep_ratio=0.3)                      # 단어 70% 손상

real 측정: 83.38 (vLLM Qwen3-8B-KPI3-LoRA, char-tokenize, 500건 subset 100건) · 합격선 78

데이터셋 500건 중 중복 input 100건 → unique 399건. truth_map 효과 measurement 영향 적음 (corpus BLEU는 list 단위).

# data_augmentation.py:64-87 (pure-Python stub, 실은 이게 운영 코드)
def _stub_augment(payload: dict) -> dict:
    latency_ms = int(os.getenv("KPI4_STUB_LATENCY_MS", "3"))
    time.sleep(latency_ms / 1000.0)                          # ← 의도된 지연
    return {
        "id": payload["id"],
        "original_hash": _stable_hash(payload),              # SHA-256
        "security": {"encrypted_transport": True,
                     "access_log_id": f"audit-{uuid4.hex[:8]}"},
    }

# throughput.py:75-85 (측정 루프)
while time.monotonic() - run_start < duration_sec:           # 60초
    fn(payloads[idx % n])                                    # 가능한 많이
    run_count += 1
rate_per_min = run_count / elapsed * 60.0

real 측정: 15,479/min (3ms sleep × 60s = 약 20K 상한 — 측정값 95% 신뢰구간 내). 합격선 500/min × 31배 마진.

KSEL 절차서가 "외부 LLM 의존 금지" 요건 → pure-Python rule-based로 설계됨. 이는 의도된 구조.

# recommendations.py:396-404 (분기)
def score_personalization(sample: dict) -> dict:
    if os.getenv("ENABLE_API_PIPELINE") == "true":
        return _api_score(sample)
    if os.getenv("ENABLE_REAL_PIPELINE") == "true":
        return _real_score_personalization(sample)
    return _stub_score_personalization(sample)

# recommendations.py:134-158 (stub — 결정적)
def _stub_score_personalization(sample) -> dict:
    rng = random.Random(seed_prefix + sample["id"])
    target = float(os.getenv("KPI5_STUB_TARGET", "0.33"))    # 합격선 0.31
    base = max(0, min(1, target + rng.gauss(0, 0.04)))       # 가우시안 noise
    return {"score": base, "strategy_scores": {...4종...}}

# recommendations.py:222-240 (real — FinguAI-Chat-v1, 4개 strategy)
def _real_score_personalization(sample) -> dict:
    for strategy in _REC_STRATEGY_PROMPTS:                   # 4 prompt 순회
        raw = generate_text("FINGU-AI/FinguAI-Chat-v1",
                            prompt, max_new_tokens=80)
        # substring 매칭: primary=0.6, supporting=0.3×ratio

위험: KPI 1과 같은 FinguAI-Chat-v1 모델 — "A)" 같은 응답 패턴이면 substring 매칭도 실패 → real score 0.0 가능.

# search.py:103-132 (stub)
def _stub_search(query: str, k: int) -> list[str]:
    target = float(os.getenv("KPI6_STUB_TARGET", "0.72"))    # 합격선 64.06 + 3%
    refs = _get_truth_map().get(query, [])
    result = rng.sample(noise_corpus, k)                     # noise k개
    if refs and rng.random() < target:
        pos = rng.randint(0, 4)                              # top-5 중 한 자리
        result[pos] = rng.choice(refs)                       # ← 정답 주입
    return result

# search.py:174-197 (real)
_REAL_EMBED_MODEL_ID = "FINGU-AI/FingUv2"                    # 운영 default
def _real_search(query: str, k: int) -> list[str]:
    corpus, matrix = _get_index()                            # all-ref union ~3K
    model = load_sentence_transformer(_REAL_EMBED_MODEL_ID)
    q = model.encode([query], normalize_embeddings=True)
    sims = np.asarray(q @ matrix.T).ravel()
    return [corpus[i] for i in np.argsort(-sims)[:k]]        # cosine top-k

# nq_em.py:74-86 (substring 양방향 매칭)
for ref in norm_refs:
    if ref in pred or pred in ref:                           # ← 양방향
        return True
# score = recall@primary_k × 100

real 측정: 47.80 · 합격선 64.06 · FAIL (-16.26)

원인:

1. FingUv2 모델: 영어 web-search instruction-tuned → 한국 금융 도메인 mismatch
2. corpus 부족: all-ref-union 약 3,000개 (실제 금융 corpus는 수만~수백만)
3. substring 양방향: "이자율" vs "금리" 같은 동의어 매칭 실패
4. BM25 prefilter 없음: 순수 dense retrieval만

# recommendations.py:541-552 (분기)
def recommend_top_k(user_id, k=10) -> list[str]:
    if os.getenv("ENABLE_REAL_PIPELINE") == "true":
        return _real_recommend_top_k(user_id, k)             # FinguAI-Chat-v1
    return _stub_recommend_top_k(user_id, k)

# recommendations.py:419-447 (stub)
def _stub_recommend_top_k(user_id, k) -> list[str]:
    rng = random.Random(seed_prefix + user_id)
    target = float(os.getenv("KPI7_STUB_TARGET", "0.92"))    # gold 포함율
    gold = truth_map[user_id]
    result = [g for g in gold if rng.random() < target][:k]
    while len(result) < k:
        result.append(rng.choice(catalog_non_gold))          # 51개 catalog
    rng.shuffle(result)
    return result

위험: real 모드는 catalog 51개를 prompt에 전부 주입 → FinguAI-Chat-v1이 응답하는 item_id를 exact match → substring fallback. KPI 1처럼 무의미 응답이면 F1@10 = 0.

1. 5/12 측정 시점에 환경변수가 다르게 설정됨 (Infisical history 추적 필요)
2. 또는 KPI 모듈 main() 호출 시 인위로 stub 강제 (--stub 같은 flag)
3. 또는 measure-all CLI 가 별도 stub-only 모드로 실행됨
4. 운영 서버 GPU 부재 + 메모리 6GB 한계 → real 모드 모델 로딩 실패 → 코드가 silent fallback to stub?

현재 운영 환경에서 ENABLE_REAL_PIPELINE=true 인데 측정이 stub로 나온다면 코드 silent fallback 의심. 추적은 src/pipelines/_real_runtime.py:30-70 의 모델 로드 부 + try/except 확인.