hyungi/hyungi_document_server
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test(search): Phase 0.2 평가셋 + 평가 스크립트

Browse Source 
22개 쿼리(6개 카테고리)와 Recall/MRR/NDCG@10 + latency p50/p95
측정 스크립트 추가. wiggly-weaving-puppy 플랜 Phase 0.2 산출물.

- queries.yaml: 정확키워드/한국어자연어/crosslingual/뉴스/실패 케이스
  실제 코퍼스(2026-04-07, 753 docs) 기반 정답 doc_id 매핑
- run_eval.py: 단일 평가 + A/B 비교 모드, CSV 저장

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Hyungi Ahn
2026-04-07 08:19:38 +09:00
parent f523752971
commit 8490cfed10
2 changed files with 730 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

257
tests/search_eval/queries.yaml Normal file
View File

@@ -0,0 +1,257 @@
# Document Server 검색 평가셋 v0.1
#
# Phase 0.2 산출물 — 정량 지표(Recall@10, MRR@10, NDCG@10) 측정용.
# 각 쿼리는 실제 코퍼스(2026-04-07 시점, 753 documents)에서 추출한
# 정답 doc_id와 함께 정의된다.
#
# 메타데이터:
# - intent : semantic_search | fact_lookup | filter_browse (Phase 2 QueryAnalyzer 분기 기준)
# - domain_hint : document | news | mixed (Phase 1 도메인 분기 기준)
# - category : 쿼리 유형 (eval 결과 그루핑용)
# - relevant_ids : 정답 doc_id 리스트 (eval에서 recall/ndcg 계산)
# - top3_ids : 반드시 top-3 안에 들어와야 하는 강한 정답 (선택)
# - notes : 의도/배경 (사람용)
#
# 주의:
# - 정답은 "현재 코퍼스에 실제 존재하는 문서"만 기재. 코퍼스가 바뀌면 갱신 필요.
# - relevant_ids가 빈 리스트인 쿼리는 "결과 없어야 정상" 또는 "low confidence가 정상"인 케이스.
version: "0.1"
created_at: "2026-04-07"
corpus_size: 753
notes: |
Phase 0.2 초기 평가셋. 22개 쿼리, 6개 카테고리.
Phase 1 reranker 통합 후 NDCG@10 비교 baseline으로 사용.
search_failure_logs(Phase 0.3)에서 자동 수집된 쿼리로 점진 확장 예정.
queries:
# ─────────────────────────────────────────────────────────
# 1. 정확 키워드 검색 (fact_lookup, document)
# ─────────────────────────────────────────────────────────
- id: kw_001
query: "산업안전보건법 제6장"
category: exact_keyword
intent: fact_lookup
domain_hint: document
relevant_ids: [3856, 3868, 3879]
top3_ids: [3856]
notes: |
"유해·위험 기계 등에 대한 조치"가 들어있는 6장.
Act(3856), Decree(3868), Rule(3879) 모두 정답이지만 본법(Act)이 최우선.
- id: kw_002
query: "중대재해 처벌 등에 관한 법률 제2장 중대산업재해"
category: exact_keyword
intent: fact_lookup
domain_hint: document
relevant_ids: [3917, 3921]
top3_ids: [3917]
notes: 본법 제2장(3917)이 정답. 시행령 동일 장(3921)도 허용.
- id: kw_003
query: "화학물질관리법 유해화학물질 영업자"
category: exact_keyword
intent: fact_lookup
domain_hint: document
relevant_ids: [3981]
top3_ids: [3981]
notes: 화학물질관리법 제4장 = 유해화학물질 영업자.
- id: kw_004
query: "근로기준법 안전과 보건"
category: exact_keyword
intent: fact_lookup
domain_hint: document
relevant_ids: [4041]
top3_ids: [4041]
notes: 근로기준법 제6장 = 안전과 보건.
- id: kw_005
query: "산업안전보건기준에 관한 규칙 보호구"
category: exact_keyword
intent: fact_lookup
domain_hint: document
relevant_ids: [3888]
top3_ids: [3888]
notes: 산업안전보건기준 규칙 제4장 = 보호구.
# ─────────────────────────────────────────────────────────
# 2. 한국어 자연어 질의 (semantic_search, document)
# ─────────────────────────────────────────────────────────
- id: nl_001
query: "기계로 인한 산업재해 관련 법령"
category: natural_language_ko
intent: semantic_search
domain_hint: document
relevant_ids: [3856, 3868, 3879, 3854]
top3_ids: [3856]
notes: |
플랜의 대표 예시 쿼리. 기계 안전 = 산안법 6장(3856).
4장 유해·위험 방지 조치(3854)도 의미상 관련.
- id: nl_002
query: "사업주가 도급을 줄 때 산업재해를 예방하기 위해 해야 할 일"
category: natural_language_ko
intent: semantic_search
domain_hint: document
relevant_ids: [3855, 3867, 3878]
top3_ids: [3855]
notes: 산안법 제5장 도급 시 산업재해 예방.
- id: nl_003
query: "유해화학물질을 다루는 회사가 지켜야 할 안전 의무"
category: natural_language_ko
intent: semantic_search
domain_hint: document
relevant_ids: [3980, 3981, 3982]
notes: 화관법 제3-5장(유해화학물질 관리/영업자/사고 대응).
- id: nl_004
query: "중대재해가 발생했을 때 경영책임자가 처벌받는 기준"
category: natural_language_ko
intent: semantic_search
domain_hint: document
relevant_ids: [3916, 3917, 3920, 3921]
top3_ids: [3917]
notes: 중대재해처벌법 본법+시행령 제1-2장.
- id: nl_005
query: "안전보건교육은 누가 받아야 하고 어떤 내용을 다루는가"
category: natural_language_ko
intent: semantic_search
domain_hint: document
relevant_ids: [3853, 3865]
top3_ids: [3853]
notes: 산안법 제3장 안전보건교육 + 시행령 제3장.
# ─────────────────────────────────────────────────────────
# 3. 한국어 → 영어 crosslingual (semantic_search, document)
# ─────────────────────────────────────────────────────────
- id: cl_001
query: "기계 안전 가드 설계 원리"
category: crosslingual_ko_en
intent: semantic_search
domain_hint: document
relevant_ids: [3770, 3856]
top3_ids: [3770]
notes: |
Industrial Safety and Health Management(7-ED) Ch15 Machine Guarding(3770)이
한국어 쿼리로 검색되어야 함. 한국 산안법 6장(3856)도 관련.
- id: cl_002
query: "산업 안전 입문서"
category: crosslingual_ko_en
intent: semantic_search
domain_hint: document
relevant_ids: [3755, 3775, 3776, 3777]
notes: |
Safety and Health for Engineers / Industrial Safety and Health Management
영문 교재 입문 챕터들이 한국어 쿼리로 검색되어야 함.
- id: cl_003
query: "전기 안전 위험"
category: crosslingual_ko_en
intent: semantic_search
domain_hint: document
relevant_ids: [3772, 3790]
notes: |
Electrical Hazards(3772), Electrical Safety(3790) 영문 챕터.
한국어 안전기준 규칙 중 전기 관련 장도 있을 수 있음(보수적으로 영문만 정답).
# ─────────────────────────────────────────────────────────
# 4. 뉴스 / 다국어 (semantic_search, news)
# ─────────────────────────────────────────────────────────
- id: news_001
query: "이란과 미국의 군사 충돌"
category: news_ko
intent: semantic_search
domain_hint: news
relevant_ids: [4303, 4304, 4307, 4316, 4322, 4323, 4327, 4335]
notes: |
경향신문의 이란-미국 전쟁 보도. recall 위주 평가.
diversity 제약 적용 후에도 최소 5건은 top-10에 들어와야 함.
- id: news_002
query: "호르무즈 해협 봉쇄"
category: news_ko
intent: semantic_search
domain_hint: news
relevant_ids: [4316, 4320, 4322, 4327]
top3_ids: [4316]
notes: 호르무즈 해협 직접 언급 기사.
- id: news_003
query: "Trump Iran ultimatum"
category: news_en
intent: semantic_search
domain_hint: news
relevant_ids: [4258, 4260, 4262]
notes: Der Spiegel 영어판 Iran 관련 기사.
- id: news_004
query: "guerre en Iran"
category: news_fr
intent: semantic_search
domain_hint: news
relevant_ids: [4199, 4202, 4210, 4361, 4363, 4507, 4519, 4521]
notes: Le Monde 불어 Iran 전쟁 보도.
- id: news_005
query: "이란 미국 전쟁 글로벌 반응"
category: news_crosslingual
intent: semantic_search
domain_hint: news
relevant_ids: [4202, 4258, 4262, 4536, 4303, 4304, 4316]
notes: |
한국어 쿼리로 한/영/불/독 뉴스가 골고루 검색되어야 함.
Phase 1 domain-aware retrieval + multilingual embedding 효과 측정용.
diversity 제약(국가당 max 2)이 동작하면 최소 4개국 이상 노출.
# ─────────────────────────────────────────────────────────
# 5. 기타 도메인 (semantic_search, document)
# ─────────────────────────────────────────────────────────
- id: misc_001
query: "강체의 평면 운동학"
category: other_domain
intent: fact_lookup
domain_hint: document
relevant_ids: [4063, 4065]
top3_ids: [4063]
notes: 공업역학 동역학 ch16, ch18.
- id: misc_002
query: "질점의 운동역학"
category: other_domain
intent: semantic_search
domain_hint: document
relevant_ids: [4060, 4061, 4062]
notes: 공업역학 동역학 ch13~15 (질점 운동역학).
# ─────────────────────────────────────────────────────────
# 6. 실패 / 애매 케이스 (low confidence 기대)
# ─────────────────────────────────────────────────────────
- id: fail_001
query: "Rust async runtime tokio scheduler 내부 구조"
category: failure_expected
intent: semantic_search
domain_hint: document
relevant_ids: []
notes: |
코퍼스에 Rust/프로그래밍 문서 없음. Phase 0.3 search_failure_logs로 자동 수집되어야 함.
Phase 1+에서 confidence 점수 < 0.5로 분류되는지 확인.
- id: fail_002
query: "양자컴퓨터 큐비트 디코히어런스"
category: failure_expected
intent: semantic_search
domain_hint: document
relevant_ids: []
notes: 코퍼스에 양자물리 문서 없음.
- id: fail_003
query: "재즈 보컬리스트 빌리 홀리데이"
category: failure_expected
intent: semantic_search
domain_hint: news
relevant_ids: []
notes: 코퍼스에 음악/재즈 문서 없음.

473
tests/search_eval/run_eval.py Normal file
View File

@@ -0,0 +1,473 @@
#!/usr/bin/env python3
"""Document Server 검색 평가 스크립트 (Phase 0.2)
queries.yaml을 읽어 /api/search 엔드포인트에 호출하고
Recall@10, MRR@10, NDCG@10, Top3 hit-rate, Latency p50/p95를 계산한다.
A/B 비교 모드: --baseline-url, --candidate-url 를 각각 지정하면
두 엔드포인트에 동일 쿼리셋을 던지고 결과를 비교한다.
사용 예:
# 단일 평가
export DOCSRV_TOKEN="eyJ..."
python tests/search_eval/run_eval.py \
--base-url https://docs.hyungi.net \
--output reports/baseline_2026-04-07.csv
# A/B 비교 (같은 토큰)
python tests/search_eval/run_eval.py \
--baseline-url https://docs.hyungi.net \
--candidate-url http://localhost:8000 \
--output reports/phase1_vs_baseline.csv
토큰은 env DOCSRV_TOKEN 또는 --token 플래그로 전달.
"""
from __future__ import annotations
import argparse
import asyncio
import csv
import math
import os
import statistics
import sys
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any
import httpx
import yaml
# ─────────────────────────────────────────────────────────
# 데이터 구조
# ─────────────────────────────────────────────────────────
@dataclass
class Query:
id: str
query: str
category: str
intent: str
domain_hint: str
relevant_ids: list[int]
top3_ids: list[int] = field(default_factory=list)
notes: str = ""
@dataclass
class QueryResult:
query: Query
label: str # "baseline" or "candidate"
returned_ids: list[int]
latency_ms: float
recall_at_10: float
mrr_at_10: float
ndcg_at_10: float
top3_hit: bool
error: str | None = None
# ─────────────────────────────────────────────────────────
# 평가 지표
# ─────────────────────────────────────────────────────────
def recall_at_k(returned: list[int], relevant: list[int], k: int = 10) -> float:
"""top-k 안에 들어간 정답 비율. 정답 0개면 1.0(빈 케이스는 별도 fail metric)."""
if not relevant:
return 1.0 if not returned else 0.0 # 비어야 정상인 케이스: 결과 있으면 fail
top_k = set(returned[:k])
hits = sum(1 for doc_id in relevant if doc_id in top_k)
return hits / len(relevant)
def mrr_at_k(returned: list[int], relevant: list[int], k: int = 10) -> float:
"""top-k 안 첫 정답의 reciprocal rank. 정답 없으면 0."""
if not relevant:
return 0.0
relevant_set = set(relevant)
for rank, doc_id in enumerate(returned[:k], start=1):
if doc_id in relevant_set:
return 1.0 / rank
return 0.0
def ndcg_at_k(returned: list[int], relevant: list[int], k: int = 10) -> float:
"""binary relevance 기반 NDCG@k. top3_ids 같은 가중치는 v0.1에선 무시."""
if not relevant:
return 0.0
relevant_set = set(relevant)
dcg = 0.0
for rank, doc_id in enumerate(returned[:k], start=1):
if doc_id in relevant_set:
# binary gain = 1, DCG = 1 / log2(rank+1)
dcg += 1.0 / math.log2(rank + 1)
# ideal DCG: 정답을 1..min(len(relevant), k) 위치에 모두 채운 경우
ideal_hits = min(len(relevant), k)
idcg = sum(1.0 / math.log2(r + 1) for r in range(1, ideal_hits + 1))
return dcg / idcg if idcg > 0 else 0.0
def top3_hit(returned: list[int], top3_ids: list[int]) -> bool:
"""top3_ids가 비어있으면 True (체크 안함). 있으면 그 중 하나라도 top-3에 들어와야 함."""
if not top3_ids:
return True
top3 = set(returned[:3])
return any(doc_id in top3 for doc_id in top3_ids)
# ─────────────────────────────────────────────────────────
# API 호출
# ─────────────────────────────────────────────────────────
async def call_search(
client: httpx.AsyncClient,
base_url: str,
token: str,
query: str,
mode: str = "hybrid",
limit: int = 20,
) -> tuple[list[int], float]:
"""검색 API 호출 → (doc_ids, latency_ms)."""
url = f"{base_url.rstrip('/')}/api/search/"
headers = {"Authorization": f"Bearer {token}"}
params = {"q": query, "mode": mode, "limit": limit}
import time
start = time.perf_counter()
response = await client.get(url, headers=headers, params=params, timeout=30.0)
latency_ms = (time.perf_counter() - start) * 1000
response.raise_for_status()
data = response.json()
returned_ids = [r["id"] for r in data.get("results", [])]
return returned_ids, latency_ms
# ─────────────────────────────────────────────────────────
# 평가 실행
# ─────────────────────────────────────────────────────────
async def evaluate(
queries: list[Query],
base_url: str,
token: str,
label: str,
mode: str = "hybrid",
) -> list[QueryResult]:
"""전체 쿼리셋 평가."""
results: list[QueryResult] = []
async with httpx.AsyncClient() as client:
for q in queries:
try:
returned_ids, latency_ms = await call_search(
client, base_url, token, q.query, mode=mode
)
results.append(
QueryResult(
query=q,
label=label,
returned_ids=returned_ids,
latency_ms=latency_ms,
recall_at_10=recall_at_k(returned_ids, q.relevant_ids, 10),
mrr_at_10=mrr_at_k(returned_ids, q.relevant_ids, 10),
ndcg_at_10=ndcg_at_k(returned_ids, q.relevant_ids, 10),
top3_hit=top3_hit(returned_ids, q.top3_ids),
)
)
except Exception as exc:
results.append(
QueryResult(
query=q,
label=label,
returned_ids=[],
latency_ms=0.0,
recall_at_10=0.0,
mrr_at_10=0.0,
ndcg_at_10=0.0,
top3_hit=False,
error=str(exc),
)
)
return results
# ─────────────────────────────────────────────────────────
# 결과 집계 / 출력
# ─────────────────────────────────────────────────────────
def percentile(values: list[float], p: float) -> float:
if not values:
return 0.0
s = sorted(values)
k = (len(s) - 1) * p
f = int(k)
c = min(f + 1, len(s) - 1)
if f == c:
return s[f]
return s[f] + (s[c] - s[f]) * (k - f)
def print_summary(label: str, results: list[QueryResult]) -> dict[str, float]:
"""전체 + 카테고리별 요약 출력. 집계 dict 반환."""
n = len(results)
if n == 0:
return {}
# 실패 케이스(relevant_ids=[])는 평균 recall/mrr/ndcg에서 제외
scored = [r for r in results if r.query.relevant_ids]
failure_cases = [r for r in results if not r.query.relevant_ids]
avg_recall = statistics.mean([r.recall_at_10 for r in scored]) if scored else 0.0
avg_mrr = statistics.mean([r.mrr_at_10 for r in scored]) if scored else 0.0
avg_ndcg = statistics.mean([r.ndcg_at_10 for r in scored]) if scored else 0.0
top3_rate = sum(1 for r in scored if r.top3_hit) / len(scored) if scored else 0.0
latencies = [r.latency_ms for r in results if r.latency_ms > 0]
p50 = percentile(latencies, 0.50)
p95 = percentile(latencies, 0.95)
# 실패 케이스: 결과 0건이어야 정상
failure_correct = sum(1 for r in failure_cases if not r.returned_ids)
failure_precision = (
failure_correct / len(failure_cases) if failure_cases else 0.0
)
print(f"\n=== {label} (n={n}, scored={len(scored)}) ===")
print(f" Recall@10 : {avg_recall:.3f}")
print(f" MRR@10 : {avg_mrr:.3f}")
print(f" NDCG@10 : {avg_ndcg:.3f}")
print(f" Top-3 hit : {top3_rate:.3f}")
print(f" Latency p50: {p50:.0f} ms")
print(f" Latency p95: {p95:.0f} ms")
if failure_cases:
print(
f" Failure-case precision: {failure_correct}/{len(failure_cases)}"
f" ({failure_precision:.2f}) — empty result expected"
)
# 카테고리별
by_cat: dict[str, list[QueryResult]] = {}
for r in scored:
by_cat.setdefault(r.query.category, []).append(r)
print(" by category:")
for cat, items in sorted(by_cat.items()):
cat_recall = statistics.mean([r.recall_at_10 for r in items])
cat_ndcg = statistics.mean([r.ndcg_at_10 for r in items])
print(
f" {cat:<22} n={len(items):>2} recall={cat_recall:.2f} ndcg={cat_ndcg:.2f}"
)
# 에러 케이스
errors = [r for r in results if r.error]
if errors:
print(f" ERRORS ({len(errors)}):")
for r in errors:
print(f" [{r.query.id}] {r.error}")
return {
"n": n,
"recall_at_10": avg_recall,
"mrr_at_10": avg_mrr,
"ndcg_at_10": avg_ndcg,
"top3_hit_rate": top3_rate,
"latency_p50": p50,
"latency_p95": p95,
"failure_precision": failure_precision,
}
def write_csv(results: list[QueryResult], output_path: Path) -> None:
output_path.parent.mkdir(parents=True, exist_ok=True)
with output_path.open("w", newline="", encoding="utf-8") as f:
writer = csv.writer(f)
writer.writerow(
[
"label",
"id",
"category",
"intent",
"domain_hint",
"query",
"relevant_ids",
"returned_ids_top10",
"latency_ms",
"recall_at_10",
"mrr_at_10",
"ndcg_at_10",
"top3_hit",
"error",
]
)
for r in results:
writer.writerow(
[
r.label,
r.query.id,
r.query.category,
r.query.intent,
r.query.domain_hint,
r.query.query,
";".join(map(str, r.query.relevant_ids)),
";".join(map(str, r.returned_ids[:10])),
f"{r.latency_ms:.1f}",
f"{r.recall_at_10:.3f}",
f"{r.mrr_at_10:.3f}",
f"{r.ndcg_at_10:.3f}",
"1" if r.top3_hit else "0",
r.error or "",
]
)
print(f"\nCSV written: {output_path}")
# ─────────────────────────────────────────────────────────
# 로딩
# ─────────────────────────────────────────────────────────
def load_queries(yaml_path: Path) -> list[Query]:
with yaml_path.open(encoding="utf-8") as f:
data = yaml.safe_load(f)
queries: list[Query] = []
for q in data["queries"]:
queries.append(
Query(
id=q["id"],
query=q["query"],
category=q["category"],
intent=q["intent"],
domain_hint=q["domain_hint"],
relevant_ids=q.get("relevant_ids", []) or [],
top3_ids=q.get("top3_ids", []) or [],
notes=q.get("notes", "") or "",
)
)
return queries
# ─────────────────────────────────────────────────────────
# CLI
# ─────────────────────────────────────────────────────────
def main() -> int:
parser = argparse.ArgumentParser(description="Document Server 검색 평가")
parser.add_argument(
"--queries",
type=Path,
default=Path(__file__).parent / "queries.yaml",
help="평가셋 YAML 경로",
)
parser.add_argument(
"--base-url",
type=str,
default=None,
help="단일 평가용 URL (예: https://docs.hyungi.net)",
)
parser.add_argument(
"--baseline-url",
type=str,
default=None,
help="A/B 비교용 baseline URL",
)
parser.add_argument(
"--candidate-url",
type=str,
default=None,
help="A/B 비교용 candidate URL",
)
parser.add_argument(
"--mode",
type=str,
default="hybrid",
choices=["fts", "trgm", "vector", "hybrid"],
help="검색 mode 파라미터",
)
parser.add_argument(
"--token",
type=str,
default=os.environ.get("DOCSRV_TOKEN"),
help="Bearer 토큰 (env DOCSRV_TOKEN)",
)
parser.add_argument(
"--output",
type=Path,
default=None,
help="CSV 출력 경로 (지정하면 raw 결과 저장)",
)
args = parser.parse_args()
if not args.token:
print("ERROR: --token 또는 env DOCSRV_TOKEN 필요", file=sys.stderr)
return 2
if not args.base_url and not (args.baseline_url and args.candidate_url):
print(
"ERROR: --base-url 또는 (--baseline-url + --candidate-url) 둘 중 하나 필요",
file=sys.stderr,
)
return 2
queries = load_queries(args.queries)
print(f"Loaded {len(queries)} queries from {args.queries}")
print(f"Mode: {args.mode}")
all_results: list[QueryResult] = []
if args.base_url:
print(f"\n>>> evaluating: {args.base_url}")
results = asyncio.run(
evaluate(queries, args.base_url, args.token, "single", mode=args.mode)
)
print_summary("single", results)
all_results.extend(results)
else:
print(f"\n>>> baseline: {args.baseline_url}")
baseline_results = asyncio.run(
evaluate(queries, args.baseline_url, args.token, "baseline", mode=args.mode)
)
baseline_summary = print_summary("baseline", baseline_results)
print(f"\n>>> candidate: {args.candidate_url}")
candidate_results = asyncio.run(
evaluate(
queries, args.candidate_url, args.token, "candidate", mode=args.mode
)
)
candidate_summary = print_summary("candidate", candidate_results)
# 델타
print("\n=== Δ (candidate - baseline) ===")
for k in (
"recall_at_10",
"mrr_at_10",
"ndcg_at_10",
"top3_hit_rate",
"latency_p50",
"latency_p95",
):
delta = candidate_summary[k] - baseline_summary[k]
sign = "+" if delta >= 0 else ""
print(f" {k:<16}: {sign}{delta:.3f}")
all_results.extend(baseline_results)
all_results.extend(candidate_results)
if args.output:
write_csv(all_results, args.output)
return 0
if __name__ == "__main__":
sys.exit(main())