# S-BERT 임베딩 → UMAP 차원축소 → HDBSCAN 밀도 군집화로 키워드를 의미 클러스터로 묶는 모듈 import warnings from functools import lru_cache import hdbscan import pandas as pd import umap from sentence_transformers import SentenceTransformer warnings.filterwarnings("ignore") # UMAP/HDBSCAN의 비치명 경고 억제 # 다국어 경량 모델(한국어 포함, ~120MB). 품질 우선 시 jhgan/ko-sroberta-multitask로 교체 가능 MODEL_NAME = "paraphrase-multilingual-MiniLM-L12-v2" MIN_KEYWORDS = 20 # 이보다 적으면 군집이 무의미 RELEVANCE_THRESHOLD = 0.35 # 키워드별 시드 유사도 하한 (계산기·키보드 등 무관어 제거) CLUSTER_RELEVANCE_THRESHOLD = 0.45 # 클러스터 평균 시드 관련성 하한 (백팩·샌들 등 패션 잡화 군집 제거) @lru_cache(maxsize=1) def _model() -> SentenceTransformer: # 모델은 한 번만 로드해 재사용 (콜드 스타트 비용 절감) return SentenceTransformer(MODEL_NAME) def filter_by_relevance(df: pd.DataFrame, seed: str, threshold: float = RELEVANCE_THRESHOLD): # 시드와 유사도가 낮은 무관 키워드 제거. (필터된 df, 정규화 임베딩) 반환 model = _model() seed_emb = model.encode([seed], normalize_embeddings=True)[0] kw_emb = model.encode(df["keyword"].tolist(), normalize_embeddings=True) sims = kw_emb @ seed_emb # 정규화돼 있어 내적 = 코사인 유사도 df = df.copy() df["relevance"] = sims mask = sims >= threshold return df[mask].reset_index(drop=True), kw_emb[mask] def cluster_keywords(df: pd.DataFrame, embeddings=None, min_cluster_size: int = 10) -> list[dict]: # 키워드 df를 의미 군집으로 묶어 검색량순 클러스터 요약 반환 (노이즈 -1 제외) # embeddings: filter_by_relevance가 만든 임베딩을 넘기면 재계산을 생략한다 if len(df) < MIN_KEYWORDS: return [] df = df.copy() df["total_volume"] = df["search_volume_pc"].fillna(0) + df["search_volume_mobile"].fillna(0) # 1) 정규화 임베딩 (없으면 새로 계산) emb = embeddings if embeddings is not None else _model().encode( df["keyword"].tolist(), normalize_embeddings=True, show_progress_bar=False ) # 2) UMAP으로 5차원 축소 (고차원 메가 클러스터 방지, 코사인 기반) n_neighbors = min(10, len(df) - 1) reduced = umap.UMAP( n_neighbors=n_neighbors, n_components=5, min_dist=0.0, metric="cosine", random_state=42, ).fit_transform(emb) # 3) HDBSCAN leaf 방식으로 세분화된 동질 군집 추출 df["cluster_id"] = hdbscan.HDBSCAN( min_cluster_size=min_cluster_size, cluster_selection_method="leaf" ).fit_predict(reduced) clusters = [] for cid, g in df[df["cluster_id"] != -1].groupby("cluster_id"): # 클러스터 전체가 시드와 느슨한 패션 잡화면 군집째 제외 if "relevance" in g.columns and g["relevance"].mean() < CLUSTER_RELEVANCE_THRESHOLD: continue g = g.sort_values("total_volume", ascending=False) clusters.append({ "cluster_id": int(cid), "cluster_label": g.iloc[0]["keyword"], # 대표 키워드(최고 검색량) "keyword_count": int(len(g)), "total_search_volume": int(g["total_volume"].sum()), "top_keywords": g["keyword"].head(4).tolist(), # 주요 포함 키워드 }) clusters.sort(key=lambda c: c["total_search_volume"], reverse=True) return clusters