1편에서는 뉴스 기사를 전처리하고 BERTopic으로 클러스터링하는 파이프라인을 만들었습니다. 이번 편에서는 그 결과가 실제로 쓸 만한지 어떻게 수치로 검증했는지 다룹니다.
배경
클러스터링 코드를 돌려서 결과가 나왔다고 끝이 아닙니다. noise_ratio가 얼마인지, 토픽이 몇 개 잡혔는지, 전처리에서 HTML이 제대로 제거됐는지를 매번 눈으로 확인할 수는 없습니다. 그래서 이 프로젝트는 평가 로직을 프로덕션 코드와 분리된 evaluations/ 디렉터리에 별도로 구성했습니다. 이유는 세 가지입니다. 평가 코드에 버그가 생겨도 앱 서비스는 영향을 받지 않아야 하고, RAGAS 같은 무거운 평가 라이브러리를 앱 컨테이너에 넣고 싶지 않으며, CI/CD와 별개로 언제든 독립 실행할 수 있어야 하기 때문입니다.
문제: "잘 됐다"를 어떻게 증명할 것인가
클러스터링과 검색 품질을 검증하려면 두 가지가 필요했습니다.
정량 지표: noise_ratio, Hit Rate, MRR, NDCG 같은 계산 가능한 숫자
정답 데이터: 어떤 질의에 어떤 문서가 관련 있는지 미리 라벨링한 Golden Dataset
이 둘이 없으면 "리랭커를 추가했더니 좋아진 것 같다"는 말만 반복하게 됩니다. KPI는 클러스터링(L1), 검색(L2), 생성(L3) 세 계층으로 나눠 설계했습니다.
계층
목적
대표 지표
목표값
L1 클러스터링
전처리·군집화 품질 확인
noise_ratio, avg_topic_size
noise_ratio < 0.30
L2 검색
임베딩·리랭커 조합 정확도 확인
Hit Rate@K, MRR@K, NDCG@K
Hit@5 > 0.7
L3 생성
LLM 답변 품질 확인 (RAGAS)
faithfulness, answer_relevancy
faithfulness > 0.8
해결: KPI 계산 모듈 + Golden Dataset 구축
1단계. 평가 코드 디렉터리 분리
evaluations/
├── datasets/
│ └── golden_dataset.json # 수동 라벨링한 정답 데이터
├── metrics/
│ ├── clustering_metrics.py # L1: 순수 계산 함수, 외부 IO 없음
│ └── retrieval_metrics.py # L2: Hit@K, MRR, NDCG 계산
└── runners/
└── run_clustering_eval.py # CLI 실행 스크립트
metrics/ 폴더의 코드는 철저히 순수 함수로만 작성해서 pytest로 바로 단위 테스트가 가능하게 했습니다. 외부 시스템 연결은 runners/에서만 담당합니다.
2단계. Golden Dataset 만들기
질의(query)마다 정답 문서 ID를 사람이 직접 라벨링합니다. 최소 50개 질의, 질의당 관련 문서 1~5개, 난이도 3단계(easy/medium/hard)로 구성했습니다.
[
{
"id": "q001",
"query": "삼성 반도체 최신 뉴스",
"relevant_doc_ids": ["doc_abc123", "doc_def456"],
"expected_answer": "삼성전자는 2026년 1분기...",
"topic_keywords": ["삼성", "반도체", "HBM"],
"difficulty": "easy",
"created_at": "2026-05-13"
}
]
3단계. L1 클러스터링 KPI 계산
클러스터링 결과에서 노이즈 비율, 평균 토픽 크기, HTML 잔류율을 계산하는 순수 함수입니다.
# evaluations/metrics/clustering_metrics.py
"""
L1 클러스터링 품질 KPI 계산 모듈
- 순수 함수로만 구성: 외부 IO 없음 -> pytest 단위 테스트 가능
- 입력: cluster_service.cluster()의 반환값
"""
import re
from dataclasses import dataclass
_RE_HTML_RESIDUAL = re.compile(r"<[^>]+>")
@dataclass
class ClusteringKPI:
"""L1 KPI 집계 결과를 담는 데이터 클래스"""
exp_id: str
total_articles: int
total_topics: int
noise_count: int
noise_ratio: float
avg_topic_size: float
html_residual_rate: float
empty_text_rate: float
embed_elapsed_sec: float
cluster_elapsed_sec: float
total_elapsed_sec: float
cluster_with: bool
def compute_clustering_kpi(
cluster_result: dict,
exp_id: str,
cluster_with: bool = True,
) -> ClusteringKPI:
"""클러스터링 결과 dict로부터 L1 KPI를 계산한다.
Args:
cluster_result: NewsClusterService.cluster()의 반환값
exp_id: 실험 식별자 (예: "EXP-08")
cluster_with: 클러스터링 적용 여부
Returns:
ClusteringKPI 인스턴스
"""
articles: list[dict] = cluster_result.get("articles_with_topic", [])
total: int = cluster_result.get("total_articles", 0)
noise_count: int = cluster_result.get("noise_count", 0)
topics: dict = cluster_result.get("topics", {})
metadata: dict = cluster_result.get("metadata", {})
noise_ratio = noise_count / total if total > 0 else 0.0
valid_count = total - noise_count
avg_topic_size = valid_count / len(topics) if topics else 0.0
html_residual_count = sum(
1 for a in articles
if _RE_HTML_RESIDUAL.search(a.get("title", "") + a.get("description", ""))
)
html_residual_rate = html_residual_count / total if total > 0 else 0.0
empty_count = sum(
1 for a in articles
if not (a.get("title", "").strip() or a.get("description", "").strip())
)
empty_text_rate = empty_count / total if total > 0 else 0.0
return ClusteringKPI(
exp_id=exp_id,
total_articles=total,
total_topics=len(topics),
noise_count=noise_count,
noise_ratio=noise_ratio,
avg_topic_size=round(avg_topic_size, 2),
html_residual_rate=round(html_residual_rate, 4),
empty_text_rate=round(empty_text_rate, 4),
embed_elapsed_sec=metadata.get("embed_elapsed_sec", 0.0),
cluster_elapsed_sec=metadata.get("cluster_elapsed_sec", 0.0),
total_elapsed_sec=metadata.get("total_elapsed_sec", 0.0),
cluster_with=cluster_with,
)
4단계. L2 검색 KPI 계산
Golden Dataset의 정답 ID와 검색 결과 ID를 비교해 Hit Rate, MRR, NDCG를 계산합니다.
# evaluations/metrics/retrieval_metrics.py
"""
L2 검색 품질 KPI 계산 모듈 (순수 함수)
- Hit Rate@K, MRR@K, NDCG@K, Precision@K, Recall@K
- 외부 의존성 없음. 표준 라이브러리만 사용
"""
import math
from dataclasses import dataclass
@dataclass
class RetrievalMetrics:
"""L2 검색 품질 KPI 집계 결과"""
exp_id: str
k: int
hit_rate: float
mrr: float
ndcg: float
precision: float
recall: float
cluster_filter_hit_rate_gain: float = 0.0
def hit_rate_at_k(retrieved_ids: list[str], relevant_ids: set[str], k: int) -> float:
"""K개 결과 안에 정답 문서가 하나라도 있으면 1점"""
return 1.0 if any(doc_id in relevant_ids for doc_id in retrieved_ids[:k]) else 0.0
def mrr_at_k(retrieved_ids: list[str], relevant_ids: set[str], k: int) -> float:
"""첫 번째 정답 문서 순위의 역수 (못 찾으면 0)"""
for rank, doc_id in enumerate(retrieved_ids[:k], start=1):
if doc_id in relevant_ids:
return 1.0 / rank
return 0.0
def ndcg_at_k(retrieved_ids: list[str], relevant_ids: set[str], k: int) -> float:
"""순위를 반영한 정규화 누적 이득 지표"""
dcg = sum(
1.0 / math.log2(rank + 1)
for rank, doc_id in enumerate(retrieved_ids[:k], start=1)
if doc_id in relevant_ids
)
ideal_hits = min(len(relevant_ids), k)
idcg = sum(1.0 / math.log2(rank + 1) for rank in range(1, ideal_hits + 1))
return dcg / idcg if idcg > 0 else 0.0
def compute_retrieval_metrics(
queries: list[dict],
k: int,
exp_id: str,
) -> RetrievalMetrics:
"""전체 질의에 대한 평균 검색 품질 KPI를 계산한다.
Args:
queries: [{"retrieved_ids": [...], "relevant_ids": [...]}, ...]
k: 평가 기준 상위 K
exp_id: 실험 식별자
Returns:
RetrievalMetrics 인스턴스 (전체 질의 평균)
"""
if not queries:
return RetrievalMetrics(exp_id=exp_id, k=k, hit_rate=0.0,
mrr=0.0, ndcg=0.0, precision=0.0, recall=0.0)
n = len(queries)
hit_rates, mrrs, ndcgs = [], [], []
precisions, recalls = [], []
for q in queries:
retrieved = q["retrieved_ids"]
relevant = set(q["relevant_ids"])
hit_rates.append(hit_rate_at_k(retrieved, relevant, k))
mrrs.append(mrr_at_k(retrieved, relevant, k))
ndcgs.append(ndcg_at_k(retrieved, relevant, k))
hits = sum(1 for doc_id in retrieved[:k] if doc_id in relevant)
precisions.append(hits / k if k > 0 else 0.0)
recalls.append(hits / len(relevant) if relevant else 0.0)
return RetrievalMetrics(
exp_id=exp_id,
k=k,
hit_rate=round(sum(hit_rates) / n, 4),
mrr=round(sum(mrrs) / n, 4),
ndcg=round(sum(ndcgs) / n, 4),
precision=round(sum(precisions) / n, 4),
recall=round(sum(recalls) / n, 4),
)
5단계 실행
# L1 클러스터링 품질 평가 실행
uv run python -m evaluations.runners.run_clustering_eval --exp --query AI --display 100
# 결과 확인
cat evaluations/results/clustering.json
결과
이 KPI 모듈은 noise_ratio 0.18, avg_topic_size 7.2 같은 수치로 바꿀 수 있습니다. L2 검색 KPI는 Golden Dataset과 결합해 이후 임베딩·리랭커 실험 비교의 공통으로 쓰였습니다.
