클릭 신호 수집, 암묵적/명시적 피드백, 온라인 학습, A/B 테스트 자동화, 검색 품질 모니터링, 차가운 시작 문제를 다룹니다.
검색 시스템은 배포 후에도 끊임없이 개선해야 합니다. 사용자의 검색 패턴은 변화하고, 새로운 문서가 추가되며, 이전에 잘 작동하던 모델이 시간이 지나면서 성능이 저하될 수 있습니다. **피드백 루프(Feedback Loop)**는 이러한 변화에 대응하여 검색 품질을 지속적으로 개선하는 메커니즘입니다.
사용자의 자연스러운 행동에서 추출하는 신호입니다. 별도의 입력을 요구하지 않으므로 데이터 양이 풍부합니다.
from dataclasses import dataclass
from datetime import datetime
from enum import Enum
class FeedbackSignal(Enum):
CLICK = "click"
SKIP = "skip"
DWELL = "dwell"
BOUNCE = "bounce"
BOOKMARK = "bookmark"
SHARE = "share"
SCROLL_DEPTH = "scroll_depth"
@dataclass
class ImplicitFeedback:
query_id: str
query_text: str
doc_id: str
doc_position: int
signal: FeedbackSignal
value: float # 체류 시간(초), 스크롤 비율(0-1) 등
timestamp: datetime
session_id: str
user_id: str | None = None
class FeedbackCollector:
"""암묵적 피드백 수집기"""
# 신호별 관련성 가중치
SIGNAL_WEIGHTS = {
FeedbackSignal.CLICK: 1.0,
FeedbackSignal.DWELL: 0.0, # 체류 시간에 따라 동적 계산
FeedbackSignal.BOUNCE: -0.5,
FeedbackSignal.BOOKMARK: 2.0,
FeedbackSignal.SHARE: 2.5,
FeedbackSignal.SCROLL_DEPTH: 0.0, # 스크롤 비율에 따라 동적 계산
}
def compute_relevance_score(self, feedbacks: list[ImplicitFeedback]) -> float:
"""여러 신호를 종합한 관련성 점수 계산"""
score = 0.0
for fb in feedbacks:
if fb.signal == FeedbackSignal.DWELL:
# 30초 이상 체류 시 양의 신호, 10초 이하는 약한 음의 신호
if fb.value >= 60:
score += 1.5
elif fb.value >= 30:
score += 1.0
elif fb.value <= 10:
score -= 0.3
elif fb.signal == FeedbackSignal.SCROLL_DEPTH:
# 스크롤 비율에 비례
score += fb.value * 1.5
else:
score += self.SIGNAL_WEIGHTS.get(fb.signal, 0.0)
return score사용자가 직접 제공하는 피드백입니다. 정확도가 높지만 수집량이 적습니다.
@dataclass
class ExplicitFeedback:
query_id: str
doc_id: str
rating: int # 1-5 스케일
feedback_text: str | None = None
timestamp: datetime = None
user_id: str | None = None
class ExplicitFeedbackUI:
"""명시적 피드백 수집 UI 구성"""
FEEDBACK_OPTIONS = [
{"label": "매우 관련 있음", "value": 5},
{"label": "관련 있음", "value": 4},
{"label": "보통", "value": 3},
{"label": "관련 없음", "value": 2},
{"label": "전혀 관련 없음", "value": 1},
]암묵적 피드백은 양이 풍부하지만 노이즈가 많고, 명시적 피드백은 정확하지만 양이 적습니다. 실전에서는 암묵적 피드백을 주 데이터 소스로 사용하되, 명시적 피드백으로 보정하는 전략이 효과적입니다.
클릭 데이터를 분석할 때는 여러 편향을 고려해야 합니다.
상위 순위의 문서가 단순히 위치 때문에 더 많이 클릭되는 **위치 편향(Position Bias)**을 보정합니다.
import numpy as np
class PositionBiasCorrector:
"""위치 편향 보정"""
# 위치별 관찰 확률 (경험적 추정값)
POSITION_PROPENSITY = {
1: 1.0,
2: 0.75,
3: 0.55,
4: 0.40,
5: 0.30,
6: 0.22,
7: 0.17,
8: 0.13,
9: 0.10,
10: 0.08,
}
def correct_ctr(self, position: int, raw_ctr: float) -> float:
"""IPW(Inverse Propensity Weighting) 보정"""
propensity = self.POSITION_PROPENSITY.get(position, 0.05)
return raw_ctr / propensity
def analyze_clicks(self, click_logs: list[dict]) -> dict[str, float]:
"""위치 편향을 보정한 문서별 관련성 점수"""
doc_scores: dict[str, list[float]] = {}
for log in click_logs:
doc_id = log["doc_id"]
position = log["position"]
clicked = log["clicked"]
corrected = self.correct_ctr(position, 1.0 if clicked else 0.0)
if doc_id not in doc_scores:
doc_scores[doc_id] = []
doc_scores[doc_id].append(corrected)
return {
doc_id: np.mean(scores)
for doc_id, scores in doc_scores.items()
}사용자가 상위 결과를 건너뛰고 하위 결과를 클릭한 경우, 건너뛴 문서는 관련성이 낮다는 강한 신호입니다.
def extract_skip_signals(click_log: dict) -> list[tuple[str, str]]:
"""클릭 로그에서 건너뛰기 신호 추출
반환: (건너뛴 문서 ID, 클릭한 문서 ID) 쌍 리스트
"""
results = click_log["results"]
clicked_positions = [
i for i, r in enumerate(results) if r["clicked"]
]
skip_pairs = []
for click_pos in clicked_positions:
for skip_pos in range(click_pos):
if not results[skip_pos]["clicked"]:
skip_pairs.append(
(results[skip_pos]["doc_id"], results[click_pos]["doc_id"])
)
return skip_pairs # (less_relevant, more_relevant) 쌍검색 모델을 실시간 피드백으로 지속적으로 업데이트하는 방식입니다.
| 방식 | 주기 | 장점 | 단점 |
|---|---|---|---|
| 배치 업데이트 | 일/주 단위 | 안정적, 검증 용이 | 반영 지연 |
| 마이크로 배치 | 시간 단위 | 빠른 반영, 안정성 유지 | 인프라 복잡도 |
| 실시간 업데이트 | 즉시 | 즉각 반영 | 노이즈 취약, 검증 어려움 |
from datetime import datetime, timedelta
class OnlineLearningPipeline:
"""온라인 학습 파이프라인"""
def __init__(self, model, update_interval_hours: int = 6):
self.model = model
self.feedback_buffer: list[dict] = []
self.update_interval = timedelta(hours=update_interval_hours)
self.last_update = datetime.now()
self.min_samples = 100 # 최소 학습 샘플 수
def add_feedback(self, feedback: dict):
"""피드백 버퍼에 추가"""
self.feedback_buffer.append(feedback)
self._maybe_update()
def _maybe_update(self):
"""조건 충족 시 모델 업데이트"""
time_elapsed = datetime.now() - self.last_update
enough_data = len(self.feedback_buffer) >= self.min_samples
time_to_update = time_elapsed >= self.update_interval
if enough_data and time_to_update:
self._update_model()
def _update_model(self):
"""피드백 기반 모델 업데이트"""
training_data = self._prepare_training_data(self.feedback_buffer)
# 검증 데이터 분리 (최근 20%를 검증용으로)
split_idx = int(len(training_data) * 0.8)
train_set = training_data[:split_idx]
val_set = training_data[split_idx:]
# 업데이트 전 성능
pre_score = self._evaluate(val_set)
# 모델 미세 조정
self.model.fine_tune(train_set)
# 업데이트 후 성능
post_score = self._evaluate(val_set)
# 성능이 하락하면 롤백
if post_score < pre_score * 0.95:
self.model.rollback()
print(f"Rollback: {pre_score:.4f} -> {post_score:.4f}")
else:
print(f"Updated: {pre_score:.4f} -> {post_score:.4f}")
self.feedback_buffer.clear()
self.last_update = datetime.now()
def _prepare_training_data(self, feedbacks: list[dict]) -> list[dict]:
"""피드백을 학습 데이터로 변환"""
# 구현 생략 - 피드백 신호를 관련성 레이블로 변환
pass
def _evaluate(self, val_set: list[dict]) -> float:
"""검증 세트에서 NDCG 평가"""
# 구현 생략
pass온라인 학습에서 가장 중요한 것은 성능 하락 방지입니다. 노이즈가 많은 피드백으로 모델이 오히려 나빠질 수 있으므로, 반드시 업데이트 전후의 성능을 비교하고 하락 시 롤백하는 안전장치가 필요합니다.
검색 시스템의 변경사항을 프로덕션에 적용하기 전에 A/B 테스트로 검증합니다.
import random
import hashlib
from dataclasses import dataclass, field
from datetime import datetime
@dataclass
class Experiment:
name: str
control_config: dict
treatment_config: dict
traffic_split: float = 0.5 # treatment에 할당할 트래픽 비율
start_date: datetime = field(default_factory=datetime.now)
min_samples: int = 1000
metrics: dict = field(default_factory=dict)
class ABTestManager:
"""A/B 테스트 관리자"""
def __init__(self):
self.experiments: dict[str, Experiment] = {}
def create_experiment(self, experiment: Experiment):
self.experiments[experiment.name] = experiment
def assign_variant(self, experiment_name: str, user_id: str) -> str:
"""사용자를 실험군/대조군에 일관되게 할당"""
exp = self.experiments[experiment_name]
hash_input = f"{experiment_name}:{user_id}"
hash_value = int(hashlib.md5(hash_input.encode()).hexdigest(), 16)
bucket = (hash_value % 1000) / 1000.0
return "treatment" if bucket < exp.traffic_split else "control"
def record_metric(
self, experiment_name: str, variant: str, metric_name: str, value: float
):
"""메트릭 기록"""
exp = self.experiments[experiment_name]
key = f"{variant}:{metric_name}"
if key not in exp.metrics:
exp.metrics[key] = []
exp.metrics[key].append(value)
def get_results(self, experiment_name: str) -> dict:
"""실험 결과 요약"""
exp = self.experiments[experiment_name]
results = {}
metric_names = set()
for key in exp.metrics:
_, metric = key.split(":", 1)
metric_names.add(metric)
for metric in metric_names:
control_key = f"control:{metric}"
treatment_key = f"treatment:{metric}"
control_values = exp.metrics.get(control_key, [])
treatment_values = exp.metrics.get(treatment_key, [])
if control_values and treatment_values:
import numpy as np
control_mean = np.mean(control_values)
treatment_mean = np.mean(treatment_values)
lift = (treatment_mean - control_mean) / control_mean * 100
results[metric] = {
"control_mean": control_mean,
"treatment_mean": treatment_mean,
"lift_percent": lift,
"control_n": len(control_values),
"treatment_n": len(treatment_values),
}
return results프로덕션 검색 시스템의 품질을 지속적으로 모니터링하는 대시보드를 구성합니다.
from dataclasses import dataclass
from datetime import datetime, timedelta
@dataclass
class SearchMetrics:
"""검색 품질 모니터링 지표"""
timestamp: datetime
# 기본 지표
total_queries: int = 0
zero_result_rate: float = 0.0
avg_result_count: float = 0.0
# 사용자 행동 지표
avg_click_position: float = 0.0
click_through_rate: float = 0.0
abandonment_rate: float = 0.0
# 성능 지표
p50_latency_ms: float = 0.0
p95_latency_ms: float = 0.0
p99_latency_ms: float = 0.0
# 품질 지표
avg_session_success_rate: float = 0.0
class SearchMonitor:
"""검색 품질 모니터"""
ALERT_THRESHOLDS = {
"zero_result_rate": 0.15, # 15% 초과 시 알림
"abandonment_rate": 0.40, # 40% 초과 시 알림
"p95_latency_ms": 500, # 500ms 초과 시 알림
"click_through_rate_drop": 0.10, # 10% 이상 하락 시 알림
}
def check_alerts(self, current: SearchMetrics, previous: SearchMetrics) -> list[str]:
"""임계값 초과 알림 확인"""
alerts = []
if current.zero_result_rate > self.ALERT_THRESHOLDS["zero_result_rate"]:
alerts.append(
f"Zero result rate: {current.zero_result_rate:.1%} "
f"(threshold: {self.ALERT_THRESHOLDS['zero_result_rate']:.1%})"
)
if current.p95_latency_ms > self.ALERT_THRESHOLDS["p95_latency_ms"]:
alerts.append(
f"P95 latency: {current.p95_latency_ms:.0f}ms "
f"(threshold: {self.ALERT_THRESHOLDS['p95_latency_ms']}ms)"
)
if previous.click_through_rate > 0:
ctr_change = (
(current.click_through_rate - previous.click_through_rate)
/ previous.click_through_rate
)
if ctr_change < -self.ALERT_THRESHOLDS["click_through_rate_drop"]:
alerts.append(
f"CTR drop: {ctr_change:.1%} "
f"({previous.click_through_rate:.1%} -> {current.click_through_rate:.1%})"
)
return alerts검색 품질 모니터링에서 가장 민감하게 추적해야 할 지표는 **제로 결과율(Zero Result Rate)**과 **이탈률(Abandonment Rate)**입니다. 이 두 지표의 급격한 변화는 시스템 장애나 심각한 품질 저하를 나타낼 수 있습니다.
**차가운 시작(Cold Start)**은 새로운 사용자, 새로운 문서, 또는 새로운 시스템에 충분한 데이터가 없는 상황을 말합니다.
행동 이력이 없는 사용자에게는 개인화가 어렵습니다.
새로 추가된 문서는 클릭 이력이 없어 개인화나 인기도 기반 랭킹에서 불리합니다.
def new_document_boost(doc_age_hours: float, max_boost: float = 0.2) -> float:
"""새 문서 시간 기반 부스트 (24시간 이내 감쇠)"""
if doc_age_hours >= 24:
return 0.0
return max_boost * (1 - doc_age_hours / 24)
def content_based_initial_score(
new_doc_embedding,
popular_doc_embeddings: list,
popular_doc_scores: list[float],
) -> float:
"""유사 인기 문서 기반 초기 점수 추정"""
import numpy as np
similarities = [
float(np.dot(new_doc_embedding, pop_emb))
for pop_emb in popular_doc_embeddings
]
# 상위 5개 유사 문서의 점수 가중 평균
top_indices = np.argsort(similarities)[-5:]
weights = np.array([similarities[i] for i in top_indices])
scores = np.array([popular_doc_scores[i] for i in top_indices])
if weights.sum() > 0:
return float(np.average(scores, weights=weights))
return 0.0이번 장에서는 검색 시스템을 지속적으로 개선하기 위한 피드백 루프의 설계와 구현을 다루었습니다. 암묵적/명시적 피드백 수집, 위치 편향 보정, 온라인 학습의 안전한 모델 업데이트, A/B 테스트 자동화, 검색 품질 모니터링 대시보드, 차가운 시작 문제의 해결 전략을 학습했습니다.
다음 장은 시리즈의 마지막으로, 지금까지 배운 모든 기술을 통합하여 실전 AI 검색 시스템을 구축하는 프로젝트를 수행합니다. 전체 아키텍처 설계부터 인덱싱 파이프라인, 검색 API, 리랭킹 통합, 피드백 수집, 성능 벤치마킹, 운영 체크리스트까지 실전에 바로 적용할 수 있는 내용을 다루겠습니다.
이 글이 도움이 되셨나요?
Elasticsearch, Cross-encoder 리랭킹, 개인화를 통합한 AI 검색 시스템의 전체 아키텍처 설계부터 구현, 벤치마킹, 운영 체크리스트까지 다룹니다.
사용자 프로파일링, 클릭 이력 기반 개인화, 임베딩 기반 사용자 벡터, 인기도 편향 문제, 프라이버시 고려사항과 실시간 개인화를 다룹니다.
BM25와 시맨틱 검색의 결합 전략, RRF/선형 보간, 리랭킹 캐스케이드, 다단계 검색 파이프라인 설계와 성능-품질 트레이드오프를 다룹니다.