사용자 프로파일링, 클릭 이력 기반 개인화, 임베딩 기반 사용자 벡터, 인기도 편향 문제, 프라이버시 고려사항과 실시간 개인화를 다룹니다.
같은 "최적화"라는 검색어를 입력하더라도, 백엔드 개발자는 데이터베이스 쿼리 최적화에 관심이 있고, 프론트엔드 개발자는 React 렌더링 최적화에 관심이 있을 수 있습니다. **검색 개인화(Search Personalization)**는 사용자의 특성, 이력, 맥락을 반영하여 개인에게 최적화된 검색 결과를 제공하는 기술입니다.
개인화의 기초는 사용자를 이해하는 것입니다. 사용자 프로파일은 명시적 정보와 암묵적 정보로 구성됩니다.
사용자가 직접 설정한 정보입니다. 선호 기술 스택, 관심 분야, 경력 수준 등을 설정 페이지에서 수집합니다.
사용자의 행동에서 추론한 정보입니다. 어떤 문서를 클릭했는지, 얼마나 오래 읽었는지, 어떤 태그의 글을 자주 방문하는지 등을 분석합니다.
from dataclasses import dataclass, field
from datetime import datetime
@dataclass
class UserProfile:
user_id: str
# 명시적 정보
preferred_languages: list[str] = field(default_factory=list)
skill_level: str = "intermediate" # beginner, intermediate, advanced
interests: list[str] = field(default_factory=list)
# 암묵적 정보 (행동 기반)
topic_affinity: dict[str, float] = field(default_factory=dict)
click_history: list[dict] = field(default_factory=list)
search_history: list[str] = field(default_factory=list)
def update_topic_affinity(self, topic: str, weight: float = 1.0, decay: float = 0.95):
"""토픽 관심도 업데이트 (지수 감쇠 적용)"""
# 기존 관심도 감쇠
for t in self.topic_affinity:
self.topic_affinity[t] *= decay
# 새 관심도 추가
current = self.topic_affinity.get(topic, 0.0)
self.topic_affinity[topic] = current + weight
def get_top_interests(self, n: int = 5) -> list[tuple[str, float]]:
"""상위 관심 토픽 반환"""
sorted_topics = sorted(
self.topic_affinity.items(),
key=lambda x: x[1],
reverse=True,
)
return sorted_topics[:n]가장 직관적인 개인화 방법은 사용자의 클릭 이력을 활용하는 것입니다. 과거에 클릭한 문서와 유사한 문서의 랭킹을 높이는 방식입니다.
import numpy as np
class ClickBasedPersonalizer:
"""클릭 이력 기반 검색 결과 개인화"""
def __init__(self, embedding_model, click_weight: float = 0.2):
self.model = embedding_model
self.click_weight = click_weight
def compute_user_vector(self, click_history: list[dict]) -> np.ndarray:
"""클릭한 문서들의 임베딩 가중 평균으로 사용자 벡터 생성"""
if not click_history:
return None
vectors = []
weights = []
for i, click in enumerate(click_history):
vec = self.model.encode(
f"passage: {click['title']} {click['content'][:200]}",
normalize_embeddings=True,
)
# 최신 클릭에 더 높은 가중치 (시간 감쇠)
recency_weight = 0.9 ** (len(click_history) - 1 - i)
dwell_weight = min(click.get("dwell_time", 30) / 60, 3.0)
vectors.append(vec)
weights.append(recency_weight * dwell_weight)
weights = np.array(weights)
weights = weights / weights.sum() # 정규화
user_vector = np.average(vectors, axis=0, weights=weights)
return user_vector / np.linalg.norm(user_vector)
def personalize_results(
self,
results: list[dict],
user_vector: np.ndarray,
) -> list[dict]:
"""검색 결과에 개인화 점수 반영"""
if user_vector is None:
return results
for result in results:
doc_vector = np.array(result["embedding"])
personal_score = float(np.dot(user_vector, doc_vector))
original_score = result["score"]
result["personalized_score"] = (
(1 - self.click_weight) * original_score
+ self.click_weight * personal_score
)
results.sort(key=lambda x: x["personalized_score"], reverse=True)
return results사용자 벡터를 생성할 때 **체류 시간(dwell time)**을 가중치로 활용하면 더 정확한 관심사를 반영할 수 있습니다. 짧게 보고 나간 문서(30초 이하)는 관심이 아닌 실수 클릭일 가능성이 높으므로 가중치를 낮추는 것이 좋습니다.
클릭 이력 외에도 다양한 사용자 행동을 임베딩 공간에서 통합하여 더 풍부한 사용자 벡터를 구성할 수 있습니다.
import numpy as np
class MultiSignalUserEncoder:
"""다중 신호 기반 사용자 벡터 인코더"""
SIGNAL_WEIGHTS = {
"clicked_docs": 0.4,
"searched_queries": 0.3,
"bookmarked_docs": 0.2,
"preferred_tags": 0.1,
}
def __init__(self, embedding_model):
self.model = embedding_model
def encode_signal(self, texts: list[str], signal_type: str) -> np.ndarray:
"""특정 신호의 텍스트를 벡터화하여 평균"""
if not texts:
return np.zeros(self.model.get_sentence_embedding_dimension())
prefix = "query: " if signal_type == "searched_queries" else "passage: "
embeddings = self.model.encode(
[f"{prefix}{t}" for t in texts],
normalize_embeddings=True,
)
return np.mean(embeddings, axis=0)
def build_user_vector(self, user_signals: dict[str, list[str]]) -> np.ndarray:
"""다중 신호를 결합한 사용자 벡터 생성"""
combined = np.zeros(self.model.get_sentence_embedding_dimension())
for signal_type, weight in self.SIGNAL_WEIGHTS.items():
texts = user_signals.get(signal_type, [])
signal_vec = self.encode_signal(texts, signal_type)
combined += weight * signal_vec
# 정규화
norm = np.linalg.norm(combined)
if norm > 0:
combined = combined / norm
return combined사용자의 현재 세션 맥락을 반영하는 것도 중요한 개인화 요소입니다. 동일한 사용자라도 지금 무엇을 하고 있느냐에 따라 검색 의도가 달라집니다.
from dataclasses import dataclass, field
from datetime import datetime
@dataclass
class SessionContext:
"""현재 세션의 컨텍스트"""
session_id: str
start_time: datetime
pages_visited: list[str] = field(default_factory=list)
queries_in_session: list[str] = field(default_factory=list)
current_topic: str = ""
def infer_current_topic(self) -> str:
"""세션 내 행동에서 현재 관심 토픽 추론"""
if not self.pages_visited:
return ""
# 최근 방문 페이지에서 토픽 추출 (간단한 빈도 기반)
topic_counts: dict[str, int] = {}
for page in self.pages_visited[-5:]: # 최근 5페이지
for topic in extract_topics(page):
topic_counts[topic] = topic_counts.get(topic, 0) + 1
if topic_counts:
return max(topic_counts, key=topic_counts.get)
return ""
def apply_session_boost(
results: list[dict],
session: SessionContext,
boost_factor: float = 0.1,
) -> list[dict]:
"""세션 컨텍스트에 기반한 결과 부스팅"""
current_topic = session.infer_current_topic()
if not current_topic:
return results
for result in results:
if current_topic in result.get("tags", []):
result["score"] *= (1 + boost_factor)
results.sort(key=lambda x: x["score"], reverse=True)
return results검색 개인화에서 가장 주의해야 할 문제는 **인기도 편향(Popularity Bias)**입니다. 이미 인기 있는 콘텐츠가 더 많이 노출되고, 더 많이 클릭되며, 그 결과 더 높은 순위를 받는 양의 피드백 루프(positive feedback loop)가 형성됩니다.
1. 탐색-활용 균형(Exploration-Exploitation)
검색 결과에 확률적으로 다양한 문서를 섞어 새로운 콘텐츠가 노출될 기회를 제공합니다.
import random
def add_exploration(
ranked_results: list[dict],
candidate_pool: list[dict],
exploration_rate: float = 0.1,
num_explore: int = 2,
) -> list[dict]:
"""탐색-활용 균형: 일부 슬롯에 랜덤 문서 배치"""
if random.random() > exploration_rate:
return ranked_results
# 기존 결과에 없는 문서에서 랜덤 선택
existing_ids = {r["id"] for r in ranked_results}
explore_candidates = [c for c in candidate_pool if c["id"] not in existing_ids]
if not explore_candidates:
return ranked_results
explore_docs = random.sample(
explore_candidates,
min(num_explore, len(explore_candidates)),
)
# 중간 순위에 삽입 (상위 3개 아래)
result = ranked_results[:3]
for doc in explore_docs:
doc["is_exploration"] = True
result.append(doc)
result.extend(ranked_results[3:])
return result[:len(ranked_results)]2. 역인기도 가중치(Inverse Popularity Weighting)
인기도에 반비례하는 가중치를 적용하여, 덜 인기 있는 문서에 부스트를 줍니다.
import math
def apply_inverse_popularity(
results: list[dict],
popularity_scores: dict[str, float],
alpha: float = 0.3,
) -> list[dict]:
"""역인기도 가중치 적용"""
max_popularity = max(popularity_scores.values()) if popularity_scores else 1.0
for result in results:
pop = popularity_scores.get(result["id"], 0.0)
normalized_pop = pop / max_popularity
# 인기도가 높을수록 penalty, 낮을수록 boost
diversity_boost = 1.0 - alpha * normalized_pop
result["adjusted_score"] = result["score"] * diversity_boost
results.sort(key=lambda x: x["adjusted_score"], reverse=True)
return results3. MMR(Maximal Marginal Relevance)
결과의 다양성을 보장하기 위해, 이미 선택된 문서와 유사한 문서의 순위를 낮춥니다.
인기도 편향 완화를 너무 강하게 적용하면 관련성이 떨어지는 문서가 상위에 올 수 있습니다. A/B 테스트를 통해 다양성과 관련성의 적절한 균형점을 찾는 것이 중요합니다.
검색 개인화를 위해 수집하는 사용자 데이터는 프라이버시 관련 법규와 윤리적 고려가 필요합니다.
이번 장에서는 검색 결과를 사용자 개개인에게 맞춤화하는 개인화 기법을 다루었습니다. 명시적/암묵적 사용자 프로파일링, 클릭 이력 기반의 사용자 벡터 생성, 세션 컨텍스트 반영 방법을 학습했습니다. 인기도 편향이라는 개인화의 핵심 문제와 탐색-활용 균형, 역인기도 가중치 등의 완화 전략도 확인했습니다. 마지막으로 프라이버시 보존의 중요성과 대응 방법을 살펴보았습니다.
다음 장에서는 검색 시스템을 지속적으로 개선하기 위한 피드백 루프를 다룹니다. 클릭 신호 수집, 온라인 학습, A/B 테스트 자동화, 검색 품질 모니터링, 차가운 시작 문제 해결까지 살펴보겠습니다.
이 글이 도움이 되셨나요?
클릭 신호 수집, 암묵적/명시적 피드백, 온라인 학습, A/B 테스트 자동화, 검색 품질 모니터링, 차가운 시작 문제를 다룹니다.
BM25와 시맨틱 검색의 결합 전략, RRF/선형 보간, 리랭킹 캐스케이드, 다단계 검색 파이프라인 설계와 성능-품질 트레이드오프를 다룹니다.
Elasticsearch, Cross-encoder 리랭킹, 개인화를 통합한 AI 검색 시스템의 전체 아키텍처 설계부터 구현, 벤치마킹, 운영 체크리스트까지 다룹니다.