KyeongminandClaude Opus 4.8
b836e79ee1
- src: phase_z2 composition/mapper/pipeline/placement_planner/retry, ai_fallback(prompts/schema/validate), mdx_text_atoms 신규 - Front: PipelineTracePanel 신규, FramePanel/SlideCanvas/Home/designAgentApi 등 갱신 + 테스트 4종 추가 - templates/phase_z2: catalog(component_expansion_registry, node_slot_mapping 신규), frames, families, slide_base 갱신 - tests/matching: phase2~26 매칭 실험 스크립트·리포트·온톨로지 전체 (미커밋 진행분) - tests: b4_v4 evidence, task5~28.5 시리즈, regression(imp95 baseline) 등 신규 테스트 대량 추가 - docs/reference: MDX 구조 인벤토리, MDX→Frame 구조 계약 문서 - scripts: mdx 계약/parity/coverage/viewport 체크, gitea comment, run sync 유틸 - .gitignore: tmp*.json, chromedriver, .orchestrator, *.pkl, Front_test* 등 임시/스냅샷 제외 미완성 작업의 보존용 스냅샷 커밋 (2026-07-02) Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
1200 lines
46 KiB
Python
1200 lines
46 KiB
Python
"""8개 매칭 방법 구현"""
|
||
import math
|
||
import re
|
||
from collections import Counter
|
||
from common import tokenize_simple, clean_text, char_ngrams
|
||
|
||
|
||
# ═══════════════════════════════════════
|
||
# 1. TF-IDF + cosine
|
||
# ═══════════════════════════════════════
|
||
def _build_tfidf(docs):
|
||
"""docs: {id: text} → {id: {word: tfidf}}"""
|
||
tokenized = {k: tokenize_simple(v) for k, v in docs.items()}
|
||
df = Counter()
|
||
for toks in tokenized.values():
|
||
for w in set(toks):
|
||
df[w] += 1
|
||
N = len(tokenized)
|
||
vecs = {}
|
||
for k, toks in tokenized.items():
|
||
tf = Counter(toks)
|
||
total = len(toks) if toks else 1
|
||
vec = {}
|
||
for w, c in tf.items():
|
||
if df[w] < 2:
|
||
continue
|
||
vec[w] = (c / total) * math.log(N / df[w])
|
||
vecs[k] = vec
|
||
return vecs, df, N
|
||
|
||
|
||
def _cosine(v1, v2):
|
||
common = set(v1) & set(v2)
|
||
if not common:
|
||
return 0.0
|
||
dot = sum(v1[w] * v2[w] for w in common)
|
||
n1 = math.sqrt(sum(x**2 for x in v1.values()))
|
||
n2 = math.sqrt(sum(x**2 for x in v2.values()))
|
||
return dot / (n1 * n2) if n1 and n2 else 0.0
|
||
|
||
|
||
def method_tfidf(mdx_text, figma_texts):
|
||
all_docs = {"__query__": mdx_text, **figma_texts}
|
||
vecs, _, _ = _build_tfidf(all_docs)
|
||
q = vecs["__query__"]
|
||
scores = [(fid, _cosine(q, vecs[fid])) for fid in figma_texts]
|
||
return sorted(scores, key=lambda x: -x[1])
|
||
|
||
|
||
# ═══════════════════════════════════════
|
||
# 2-new. IDF-only (공통 희귀 단어의 IDF 합)
|
||
# ═══════════════════════════════════════
|
||
def method_idf_only(mdx_text, figma_texts):
|
||
"""TF 없이 IDF만 사용. 공통 단어의 IDF 합으로 매칭."""
|
||
all_docs = {"__query__": tokenize_simple(mdx_text)}
|
||
for fid, text in figma_texts.items():
|
||
all_docs[fid] = tokenize_simple(text)
|
||
df = Counter()
|
||
for toks in all_docs.values():
|
||
for w in set(toks):
|
||
df[w] += 1
|
||
N = len(all_docs)
|
||
idf = {w: math.log(N / c) for w, c in df.items()}
|
||
|
||
query_set = set(all_docs["__query__"])
|
||
scores = []
|
||
for fid in figma_texts:
|
||
doc_set = set(all_docs[fid])
|
||
common = query_set & doc_set
|
||
score = sum(idf.get(w, 0) for w in common)
|
||
scores.append((fid, score))
|
||
return sorted(scores, key=lambda x: -x[1])
|
||
|
||
|
||
# ═══════════════════════════════════════
|
||
# 2. BM25 (bm25s)
|
||
# ═══════════════════════════════════════
|
||
def method_bm25(mdx_text, figma_texts):
|
||
import bm25s
|
||
frame_ids = list(figma_texts.keys())
|
||
corpus = [tokenize_simple(figma_texts[fid]) for fid in frame_ids]
|
||
retriever = bm25s.BM25()
|
||
retriever.index(corpus)
|
||
query_tokens = tokenize_simple(mdx_text)
|
||
scores = retriever.get_scores(query_tokens)
|
||
paired = list(zip(frame_ids, scores.tolist()))
|
||
return sorted(paired, key=lambda x: -x[1])
|
||
|
||
|
||
# ═══════════════════════════════════════
|
||
# 3. Char 2~3 gram Jaccard
|
||
# ═══════════════════════════════════════
|
||
def method_char_ngram(mdx_text, figma_texts):
|
||
# 2gram + 3gram 합집합으로 판단
|
||
q2 = char_ngrams(mdx_text, 2)
|
||
q3 = char_ngrams(mdx_text, 3)
|
||
scores = []
|
||
for fid, text in figma_texts.items():
|
||
f2 = char_ngrams(text, 2)
|
||
f3 = char_ngrams(text, 3)
|
||
j2 = len(q2 & f2) / len(q2 | f2) if (q2 | f2) else 0
|
||
j3 = len(q3 & f3) / len(q3 | f3) if (q3 | f3) else 0
|
||
# 2gram은 덜, 3gram은 더 (3gram이 더 의미있음)
|
||
combined = 0.4 * j2 + 0.6 * j3
|
||
scores.append((fid, combined))
|
||
return sorted(scores, key=lambda x: -x[1])
|
||
|
||
|
||
# ═══════════════════════════════════════
|
||
# 4. Kiwipiepy 형태소 + BM25
|
||
# ═══════════════════════════════════════
|
||
_kiwi = None
|
||
|
||
|
||
def _kiwi_tokenize(text):
|
||
global _kiwi
|
||
if _kiwi is None:
|
||
from kiwipiepy import Kiwi
|
||
_kiwi = Kiwi()
|
||
text = clean_text(text)
|
||
tokens = _kiwi.tokenize(text)
|
||
# 명사/동사/형용사 어근만 추출 (NNG, NNP, VV, VA, SL 영어)
|
||
keep = {"NNG", "NNP", "VV", "VA", "VX", "SL", "SH", "SN"}
|
||
result = []
|
||
for t in tokens:
|
||
if t.tag in keep and len(t.form) >= 1:
|
||
result.append(t.form)
|
||
return result
|
||
|
||
|
||
def method_kiwi_bm25(mdx_text, figma_texts):
|
||
import bm25s
|
||
frame_ids = list(figma_texts.keys())
|
||
corpus = [_kiwi_tokenize(figma_texts[fid]) for fid in frame_ids]
|
||
retriever = bm25s.BM25()
|
||
retriever.index(corpus)
|
||
query_tokens = _kiwi_tokenize(mdx_text)
|
||
scores = retriever.get_scores(query_tokens)
|
||
paired = list(zip(frame_ids, scores.tolist()))
|
||
return sorted(paired, key=lambda x: -x[1])
|
||
|
||
|
||
# ═══════════════════════════════════════
|
||
# 5. 구조 메타데이터 매칭 (파싱만)
|
||
# ═══════════════════════════════════════
|
||
def _parse_mdx_structure(mdx_text):
|
||
"""MDX 텍스트에서 구조 추출."""
|
||
# bullet 개수, 표 존재, 이미지, details, 계층 깊이
|
||
lines = mdx_text.split("\n")
|
||
bullet_count = sum(1 for ln in lines if re.match(r"^\s*[\*\-]\s", ln))
|
||
table_rows = sum(1 for ln in lines if re.match(r"^\s*\|.*\|.*\|", ln))
|
||
has_details = "<details>" in mdx_text
|
||
images = len(re.findall(r"!\[.*?\]\(.*?\)", mdx_text))
|
||
# bullet depth
|
||
max_indent = 0
|
||
for ln in lines:
|
||
m = re.match(r"^(\s+)[\*\-]", ln)
|
||
if m:
|
||
max_indent = max(max_indent, len(m.group(1)) // 2)
|
||
return {
|
||
"bullets": bullet_count,
|
||
"table_rows": table_rows,
|
||
"has_details": has_details,
|
||
"images": images,
|
||
"depth": max_indent + 1,
|
||
}
|
||
|
||
|
||
def _parse_figma_structure(figma_text):
|
||
"""texts.md에서 구조 추출 — ## 개수, 리스트 여부 등"""
|
||
lines = figma_text.split("\n")
|
||
h2_count = sum(1 for ln in lines if ln.startswith("## "))
|
||
h3_count = sum(1 for ln in lines if ln.startswith("### "))
|
||
# "행1", "행2" 같은 표시나 "열1" 같은 표시
|
||
row_col_mentions = len(re.findall(r"[열행]\d", figma_text))
|
||
return {
|
||
"h2_count": h2_count,
|
||
"h3_count": h3_count,
|
||
"row_col": row_col_mentions,
|
||
}
|
||
|
||
|
||
def _structural_score(mdx_s, fig_s):
|
||
"""단순 규칙: 행/열 수와 비슷한가, 계층 깊이 비슷한가"""
|
||
score = 0.0
|
||
# MDX bullet 수와 Figma 행/열 수 비교
|
||
expected_cols = fig_s["row_col"] if fig_s["row_col"] > 0 else fig_s["h2_count"]
|
||
if expected_cols > 0:
|
||
diff = abs(mdx_s["bullets"] - expected_cols)
|
||
score += max(0, 1 - diff / 10) * 0.5
|
||
# 표가 있으면 Figma도 행/열 구조여야
|
||
if mdx_s["table_rows"] > 3 and fig_s["row_col"] > 3:
|
||
score += 0.3
|
||
# depth 비교
|
||
depth_sim = 1 - abs(mdx_s["depth"] - fig_s["h3_count"] - 1) / 5
|
||
score += max(0, depth_sim) * 0.2
|
||
return score
|
||
|
||
|
||
def method_structural(mdx_text, figma_texts):
|
||
mdx_s = _parse_mdx_structure(mdx_text)
|
||
scores = []
|
||
for fid, text in figma_texts.items():
|
||
fig_s = _parse_figma_structure(text)
|
||
scores.append((fid, _structural_score(mdx_s, fig_s)))
|
||
return sorted(scores, key=lambda x: -x[1])
|
||
|
||
|
||
# ═══════════════════════════════════════
|
||
# 6. AI 추출 메타데이터 매칭
|
||
# (Claude가 사전에 추출한 concept set을 YAML로 저장 → Jaccard)
|
||
# ═══════════════════════════════════════
|
||
def method_ai_metadata(mdx_text, figma_texts, metadata_db=None):
|
||
"""metadata_db: {key: {concepts: [...], topic: str, layout: str}}
|
||
key는 frame_id 또는 MDX section_id"""
|
||
if metadata_db is None:
|
||
return [] # 메타데이터 없으면 빈 결과
|
||
# query section id를 어떻게 찾을지... 이 함수는 별도 로직 필요
|
||
# run_method 시그니처를 건드리지 않고 구현하려면
|
||
# mdx_text를 메타데이터 DB에서 역매칭해서 찾아야 함
|
||
# 대신 메타데이터 DB에 mdx→concepts 매핑이 있다고 가정하고,
|
||
# 외부에서 직접 호출하는 별도 함수를 제공
|
||
return []
|
||
|
||
|
||
def method_ai_metadata_matcher(section_id, frame_metadata, mdx_metadata):
|
||
"""section_id 기준으로 매칭."""
|
||
if section_id not in mdx_metadata:
|
||
return []
|
||
mdx_concepts = set(mdx_metadata[section_id].get("concepts", []))
|
||
mdx_layout = mdx_metadata[section_id].get("layout_hint", "")
|
||
|
||
scores = []
|
||
for fid, fmeta in frame_metadata.items():
|
||
f_concepts = set(fmeta.get("concepts", []))
|
||
f_layout = fmeta.get("layout", "")
|
||
|
||
# concept Jaccard
|
||
if mdx_concepts or f_concepts:
|
||
jac = len(mdx_concepts & f_concepts) / len(mdx_concepts | f_concepts)
|
||
else:
|
||
jac = 0.0
|
||
|
||
# layout 호환성 (exact match or null)
|
||
layout_match = 1.0 if (mdx_layout and f_layout and mdx_layout == f_layout) else 0.3
|
||
|
||
score = 0.7 * jac + 0.3 * layout_match
|
||
scores.append((fid, score))
|
||
return sorted(scores, key=lambda x: -x[1])
|
||
|
||
|
||
# ═══════════════════════════════════════
|
||
# 7. 가중 평균 (BM25 × α + Char n-gram × (1-α))
|
||
# ═══════════════════════════════════════
|
||
def _normalize(scores):
|
||
"""[(id, score), ...] → min-max normalize to [0, 1]"""
|
||
vals = [s for _, s in scores]
|
||
mn, mx = min(vals), max(vals)
|
||
if mx == mn:
|
||
return {fid: 0.0 for fid, _ in scores}
|
||
return {fid: (s - mn) / (mx - mn) for fid, s in scores}
|
||
|
||
|
||
def method_weighted(mdx_text, figma_texts, alpha=0.5):
|
||
bm25_scores = method_bm25(mdx_text, figma_texts)
|
||
cng_scores = method_char_ngram(mdx_text, figma_texts)
|
||
bm25_norm = _normalize(bm25_scores)
|
||
cng_norm = _normalize(cng_scores)
|
||
combined = []
|
||
for fid in figma_texts:
|
||
s = alpha * bm25_norm.get(fid, 0) + (1 - alpha) * cng_norm.get(fid, 0)
|
||
combined.append((fid, s))
|
||
return sorted(combined, key=lambda x: -x[1])
|
||
|
||
|
||
# ═══════════════════════════════════════
|
||
# 8. Hard filter + BM25
|
||
# 구조 점수가 threshold 이상인 프레임만 남기고 BM25로 재정렬
|
||
# ═══════════════════════════════════════
|
||
def method_hard_filter(mdx_text, figma_texts, threshold=0.2):
|
||
# 구조 스코어로 필터
|
||
struct_scores = method_structural(mdx_text, figma_texts)
|
||
passed = {fid for fid, s in struct_scores if s >= threshold}
|
||
if len(passed) < 3:
|
||
# 통과가 적으면 top-10 강제
|
||
passed = {fid for fid, _ in struct_scores[:10]}
|
||
filtered = {fid: figma_texts[fid] for fid in passed}
|
||
bm25_scores = method_bm25(mdx_text, filtered)
|
||
# 통과 못한 프레임은 뒤에 (점수 0)
|
||
result_ids = set(fid for fid, _ in bm25_scores)
|
||
tail = [(fid, 0.0) for fid in figma_texts if fid not in result_ids]
|
||
return bm25_scores + tail
|
||
|
||
|
||
# ═══════════════════════════════════════
|
||
# 새 하이브리드: Kiwi+BM25 + Char n-gram 가중평균 (AI-Meta 제외)
|
||
# ═══════════════════════════════════════
|
||
def method_hybrid_kiwi_char(mdx_text, figma_texts, alpha=0.5):
|
||
kiwi_scores = method_kiwi_bm25(mdx_text, figma_texts)
|
||
cng_scores = method_char_ngram(mdx_text, figma_texts)
|
||
k_norm = _normalize(kiwi_scores)
|
||
c_norm = _normalize(cng_scores)
|
||
combined = []
|
||
for fid in figma_texts:
|
||
s = alpha * k_norm.get(fid, 0) + (1 - alpha) * c_norm.get(fid, 0)
|
||
combined.append((fid, s))
|
||
return sorted(combined, key=lambda x: -x[1])
|
||
|
||
|
||
# ═══════════════════════════════════════
|
||
# 가중 토큰화: 목차/라벨에 가중치 부여 (CASE 2용)
|
||
# ═══════════════════════════════════════
|
||
def weighted_tokenize(text):
|
||
"""극단 가중: 목차 ×20, 볼드체 ×30, 블릿 ×0.1"""
|
||
lines = text.split("\n")
|
||
all_tokens = []
|
||
for line in lines:
|
||
stripped = line.strip()
|
||
if not stripped:
|
||
continue
|
||
|
||
# 대/중/소목차 ×20
|
||
if stripped.startswith("# ") and not stripped.startswith("## "):
|
||
title = stripped[2:].strip()
|
||
toks = tokenize_simple(title)
|
||
all_tokens.extend(toks * 20)
|
||
continue
|
||
if stripped.startswith("## "):
|
||
title = stripped[3:].strip()
|
||
toks = tokenize_simple(title)
|
||
all_tokens.extend(toks * 20)
|
||
continue
|
||
if stripped.startswith("### "):
|
||
title = stripped[4:].strip()
|
||
toks = tokenize_simple(title)
|
||
all_tokens.extend(toks * 20)
|
||
continue
|
||
# **볼드체** 라벨 ×30 (극단)
|
||
bold_labels = re.findall(r"\*\*([^*]+)\*\*", stripped)
|
||
for lbl in bold_labels:
|
||
toks = tokenize_simple(lbl)
|
||
all_tokens.extend(toks * 30)
|
||
# 블릿·일반 ×0.1 (10토큰 중 1개만)
|
||
toks = tokenize_simple(stripped)
|
||
all_tokens.extend(toks[::10])
|
||
|
||
# 빈 결과 방어: 원본 토큰 최소 1개 보장
|
||
if not all_tokens:
|
||
all_tokens = tokenize_simple(text)
|
||
if not all_tokens:
|
||
all_tokens = ["__empty__"]
|
||
return all_tokens
|
||
|
||
|
||
# 가중 토큰화 적용 방법들
|
||
def _weighted_expand_text(text):
|
||
"""극단 가중: 목차 ×20, 볼드체 ×30, 블릿 ×0.1"""
|
||
parts = []
|
||
for line in text.split("\n"):
|
||
stripped = line.strip()
|
||
if not stripped:
|
||
continue
|
||
if stripped.startswith("# ") and not stripped.startswith("## "):
|
||
parts.extend([stripped[2:].strip()] * 20)
|
||
elif stripped.startswith("## ") or stripped.startswith("### "):
|
||
parts.extend([stripped.lstrip("#").strip()] * 20)
|
||
else:
|
||
bolds = re.findall(r"\*\*([^*]+)\*\*", stripped)
|
||
for b in bolds:
|
||
parts.extend([b] * 30)
|
||
words = stripped.split()
|
||
parts.append(" ".join(words[::10]))
|
||
return " ".join(parts)
|
||
|
||
|
||
def method_tfidf_weighted(mdx_text, figma_texts):
|
||
"""양쪽 가중: MDX와 Figma 모두 제목 가중 토큰화"""
|
||
all_docs = {"__query__": mdx_text, **figma_texts}
|
||
tokenized = {}
|
||
for k, v in all_docs.items():
|
||
tokenized[k] = weighted_tokenize(v) # 양쪽 모두 가중
|
||
df = Counter()
|
||
for toks in tokenized.values():
|
||
for w in set(toks):
|
||
df[w] += 1
|
||
N = len(tokenized)
|
||
vecs = {}
|
||
for k, toks in tokenized.items():
|
||
tf = Counter(toks)
|
||
total = len(toks) if toks else 1
|
||
vec = {}
|
||
for w, c in tf.items():
|
||
if df[w] < 2:
|
||
continue
|
||
vec[w] = (c / total) * math.log(N / df[w])
|
||
vecs[k] = vec
|
||
q = vecs["__query__"]
|
||
scores = [(fid, _cosine(q, vecs[fid])) for fid in figma_texts]
|
||
return sorted(scores, key=lambda x: -x[1])
|
||
|
||
|
||
def method_bm25_weighted(mdx_text, figma_texts):
|
||
"""양쪽 가중: MDX와 Figma 모두 제목 가중 토큰화"""
|
||
import bm25s
|
||
frame_ids = list(figma_texts.keys())
|
||
corpus = [weighted_tokenize(figma_texts[fid]) for fid in frame_ids] # Figma도 가중
|
||
retriever = bm25s.BM25()
|
||
retriever.index(corpus)
|
||
query_tokens = weighted_tokenize(mdx_text)
|
||
scores = retriever.get_scores(query_tokens)
|
||
paired = list(zip(frame_ids, scores.tolist()))
|
||
return sorted(paired, key=lambda x: -x[1])
|
||
|
||
|
||
def method_kiwi_bm25_weighted(mdx_text, figma_texts):
|
||
"""양쪽 가중: MDX와 Figma 모두 제목 확장 텍스트 → Kiwi 형태소"""
|
||
import bm25s
|
||
frame_ids = list(figma_texts.keys())
|
||
# Figma 쪽도 확장 텍스트 → Kiwi. 빈 결과 방어
|
||
corpus = []
|
||
for fid in frame_ids:
|
||
toks = _kiwi_tokenize(_weighted_expand_text(figma_texts[fid]))
|
||
if not toks:
|
||
toks = ["__empty__"]
|
||
corpus.append(toks)
|
||
retriever = bm25s.BM25()
|
||
retriever.index(corpus)
|
||
query_tokens = _kiwi_tokenize(_weighted_expand_text(mdx_text))
|
||
if not query_tokens:
|
||
query_tokens = ["__empty__"]
|
||
scores = retriever.get_scores(query_tokens)
|
||
paired = list(zip(frame_ids, scores.tolist()))
|
||
return sorted(paired, key=lambda x: -x[1])
|
||
|
||
|
||
def method_char_ngram_weighted(mdx_text, figma_texts):
|
||
"""양쪽 가중: MDX와 Figma 모두 확장 텍스트에서 ngram 추출"""
|
||
mdx_expanded = _weighted_expand_text(mdx_text)
|
||
q2 = char_ngrams(mdx_expanded, 2)
|
||
q3 = char_ngrams(mdx_expanded, 3)
|
||
scores = []
|
||
for fid, text in figma_texts.items():
|
||
fig_expanded = _weighted_expand_text(text) # Figma도 확장
|
||
f2 = char_ngrams(fig_expanded, 2)
|
||
f3 = char_ngrams(fig_expanded, 3)
|
||
j2 = len(q2 & f2) / len(q2 | f2) if (q2 | f2) else 0
|
||
j3 = len(q3 & f3) / len(q3 | f3) if (q3 | f3) else 0
|
||
combined = 0.4 * j2 + 0.6 * j3
|
||
scores.append((fid, combined))
|
||
return sorted(scores, key=lambda x: -x[1])
|
||
|
||
|
||
def method_hybrid_kiwi_char_weighted(mdx_text, figma_texts, alpha=0.5):
|
||
kiwi_scores = method_kiwi_bm25_weighted(mdx_text, figma_texts)
|
||
cng_scores = method_char_ngram_weighted(mdx_text, figma_texts)
|
||
k_norm = _normalize(kiwi_scores)
|
||
c_norm = _normalize(cng_scores)
|
||
combined = []
|
||
for fid in figma_texts:
|
||
s = alpha * k_norm.get(fid, 0) + (1 - alpha) * c_norm.get(fid, 0)
|
||
combined.append((fid, s))
|
||
return sorted(combined, key=lambda x: -x[1])
|
||
|
||
|
||
# ═══════════════════════════════════════
|
||
# Semantic 매칭 (3개) — ko-sroberta 기반 3가지 전략
|
||
# ═══════════════════════════════════════
|
||
_sbert_model = None
|
||
|
||
|
||
def _get_sbert():
|
||
global _sbert_model
|
||
if _sbert_model is None:
|
||
from sentence_transformers import SentenceTransformer
|
||
_sbert_model = SentenceTransformer('jhgan/ko-sroberta-multitask')
|
||
# 기본 128 → 512로 확장 (긴 MDX/Figma 텍스트 반영)
|
||
_sbert_model.max_seq_length = 512
|
||
return _sbert_model
|
||
|
||
|
||
def _np_cosine(v1, v2):
|
||
import numpy as np
|
||
n1 = np.linalg.norm(v1)
|
||
n2 = np.linalg.norm(v2)
|
||
if n1 == 0 or n2 == 0:
|
||
return 0.0
|
||
return float(np.dot(v1, v2) / (n1 * n2))
|
||
|
||
|
||
# 3-1. ko-sroberta 문장 전체 임베딩
|
||
def method_sbert_sentence(mdx_text, figma_texts):
|
||
"""문장 전체를 한 벡터로. 가장 기본."""
|
||
model = _get_sbert()
|
||
q_vec = model.encode(mdx_text, show_progress_bar=False)
|
||
scores = []
|
||
for fid, text in figma_texts.items():
|
||
d_vec = model.encode(text, show_progress_bar=False)
|
||
scores.append((fid, _np_cosine(q_vec, d_vec)))
|
||
return sorted(scores, key=lambda x: -x[1])
|
||
|
||
|
||
# 3-2. SBERT 축약 (제목 + 볼드 라벨만 뽑아서 짧게)
|
||
def _extract_summary(text):
|
||
"""제목(#/##/###) + 볼드(**xx**) + 최상위 블릿 라벨만 뽑기"""
|
||
parts = []
|
||
for line in text.split("\n"):
|
||
s = line.strip()
|
||
if not s:
|
||
continue
|
||
if s.startswith("# ") or s.startswith("## ") or s.startswith("### "):
|
||
parts.append(s.lstrip("#").strip())
|
||
continue
|
||
# 볼드 라벨
|
||
bolds = re.findall(r"\*\*([^*]+)\*\*", s)
|
||
parts.extend(bolds)
|
||
summary = " ".join(parts)
|
||
# 최대 300자 제한
|
||
return summary[:300] if summary else text[:300]
|
||
|
||
|
||
def method_sbert_summary(mdx_text, figma_texts):
|
||
"""MDX와 Figma 모두 제목/라벨로 축약 후 SBERT"""
|
||
model = _get_sbert()
|
||
q_sum = _extract_summary(mdx_text)
|
||
q_vec = model.encode(q_sum, show_progress_bar=False)
|
||
scores = []
|
||
for fid, text in figma_texts.items():
|
||
d_sum = _extract_summary(text)
|
||
d_vec = model.encode(d_sum, show_progress_bar=False)
|
||
scores.append((fid, _np_cosine(q_vec, d_vec)))
|
||
return sorted(scores, key=lambda x: -x[1])
|
||
|
||
|
||
# 3-3. SBERT 청크 기반 (블릿/섹션별 쪼개서 최대 유사도)
|
||
def _chunk_text(text):
|
||
"""블릿/섹션 단위로 쪼갬. 각 chunk는 의미 단위."""
|
||
chunks = []
|
||
current = []
|
||
for line in text.split("\n"):
|
||
s = line.strip()
|
||
if not s:
|
||
continue
|
||
is_new_chunk = (
|
||
s.startswith("# ") or s.startswith("## ") or s.startswith("### ")
|
||
or s.startswith("* ") or s.startswith("- ")
|
||
)
|
||
if is_new_chunk and current:
|
||
chunks.append(" ".join(current))
|
||
current = [s]
|
||
else:
|
||
current.append(s)
|
||
if current:
|
||
chunks.append(" ".join(current))
|
||
return chunks if chunks else [text]
|
||
|
||
|
||
def method_sbert_chunk(mdx_text, figma_texts):
|
||
"""청크별 임베딩 → 각 MDX chunk에 대해 최대 Figma chunk 유사도 → 평균"""
|
||
import numpy as np
|
||
model = _get_sbert()
|
||
q_chunks = _chunk_text(mdx_text)
|
||
q_vecs = model.encode(q_chunks, show_progress_bar=False)
|
||
scores = []
|
||
for fid, text in figma_texts.items():
|
||
d_chunks = _chunk_text(text)
|
||
if not d_chunks:
|
||
scores.append((fid, 0.0))
|
||
continue
|
||
d_vecs = model.encode(d_chunks, show_progress_bar=False)
|
||
# 각 MDX chunk의 최대 유사도 → 평균
|
||
max_sims = []
|
||
for qv in q_vecs:
|
||
sims = [_np_cosine(qv, dv) for dv in d_vecs]
|
||
max_sims.append(max(sims) if sims else 0)
|
||
score = sum(max_sims) / len(max_sims) if max_sims else 0
|
||
scores.append((fid, score))
|
||
return sorted(scores, key=lambda x: -x[1])
|
||
|
||
|
||
# ═══════════════════════════════════════
|
||
# Semantic 확장 — 다른 family (KoSimCSE / E5 / Cross-encoder / Word-avg)
|
||
# ═══════════════════════════════════════
|
||
|
||
# ── KoSimCSE: 대조학습 기반 sentence embedding (SBERT와 다른 family)
|
||
_kosimcse_model = None
|
||
|
||
|
||
def _get_kosimcse():
|
||
global _kosimcse_model
|
||
if _kosimcse_model is None:
|
||
from sentence_transformers import SentenceTransformer
|
||
_kosimcse_model = SentenceTransformer('BM-K/KoSimCSE-roberta-multitask')
|
||
_kosimcse_model.max_seq_length = 512
|
||
return _kosimcse_model
|
||
|
||
|
||
def method_kosimcse(mdx_text, figma_texts):
|
||
"""대조학습(SimCSE) 기반 한국어 sentence embedding — 원본"""
|
||
model = _get_kosimcse()
|
||
q_vec = model.encode(mdx_text, show_progress_bar=False)
|
||
scores = []
|
||
for fid, text in figma_texts.items():
|
||
d_vec = model.encode(text, show_progress_bar=False)
|
||
scores.append((fid, _np_cosine(q_vec, d_vec)))
|
||
return sorted(scores, key=lambda x: -x[1])
|
||
|
||
|
||
def method_kosimcse_summary(mdx_text, figma_texts):
|
||
"""KoSimCSE + 축약 (제목/볼드 라벨만)"""
|
||
model = _get_kosimcse()
|
||
q_sum = _extract_summary(mdx_text)
|
||
q_vec = model.encode(q_sum, show_progress_bar=False)
|
||
scores = []
|
||
for fid, text in figma_texts.items():
|
||
d_sum = _extract_summary(text)
|
||
d_vec = model.encode(d_sum, show_progress_bar=False)
|
||
scores.append((fid, _np_cosine(q_vec, d_vec)))
|
||
return sorted(scores, key=lambda x: -x[1])
|
||
|
||
|
||
def method_kosimcse_chunk(mdx_text, figma_texts):
|
||
"""KoSimCSE + 청크 (블릿/섹션별, 각 MDX chunk → max Figma chunk → 평균)"""
|
||
model = _get_kosimcse()
|
||
q_chunks = _chunk_text(mdx_text)
|
||
q_vecs = model.encode(q_chunks, show_progress_bar=False)
|
||
scores = []
|
||
for fid, text in figma_texts.items():
|
||
d_chunks = _chunk_text(text)
|
||
if not d_chunks:
|
||
scores.append((fid, 0.0))
|
||
continue
|
||
d_vecs = model.encode(d_chunks, show_progress_bar=False)
|
||
max_sims = []
|
||
for qv in q_vecs:
|
||
sims = [_np_cosine(qv, dv) for dv in d_vecs]
|
||
max_sims.append(max(sims) if sims else 0)
|
||
score = sum(max_sims) / len(max_sims) if max_sims else 0
|
||
scores.append((fid, score))
|
||
return sorted(scores, key=lambda x: -x[1])
|
||
|
||
|
||
# ── E5: 다국어 retrieval, query/passage 비대칭 처리
|
||
_e5_model = None
|
||
|
||
|
||
def _get_e5():
|
||
global _e5_model
|
||
if _e5_model is None:
|
||
from sentence_transformers import SentenceTransformer
|
||
_e5_model = SentenceTransformer('intfloat/multilingual-e5-base')
|
||
_e5_model.max_seq_length = 512
|
||
return _e5_model
|
||
|
||
|
||
def method_e5(mdx_text, figma_texts):
|
||
"""Multilingual E5 — query/passage 비대칭 prefix (원본)"""
|
||
model = _get_e5()
|
||
q_vec = model.encode("query: " + mdx_text, show_progress_bar=False, normalize_embeddings=True)
|
||
scores = []
|
||
for fid, text in figma_texts.items():
|
||
d_vec = model.encode("passage: " + text, show_progress_bar=False, normalize_embeddings=True)
|
||
scores.append((fid, _np_cosine(q_vec, d_vec)))
|
||
return sorted(scores, key=lambda x: -x[1])
|
||
|
||
|
||
def method_e5_summary(mdx_text, figma_texts):
|
||
"""E5 + 축약"""
|
||
model = _get_e5()
|
||
q_sum = _extract_summary(mdx_text)
|
||
q_vec = model.encode("query: " + q_sum, show_progress_bar=False, normalize_embeddings=True)
|
||
scores = []
|
||
for fid, text in figma_texts.items():
|
||
d_sum = _extract_summary(text)
|
||
d_vec = model.encode("passage: " + d_sum, show_progress_bar=False, normalize_embeddings=True)
|
||
scores.append((fid, _np_cosine(q_vec, d_vec)))
|
||
return sorted(scores, key=lambda x: -x[1])
|
||
|
||
|
||
def method_e5_chunk(mdx_text, figma_texts):
|
||
"""E5 + 청크 (query chunk × passage chunk max → 평균)"""
|
||
model = _get_e5()
|
||
q_chunks = _chunk_text(mdx_text)
|
||
q_vecs = model.encode(["query: " + c for c in q_chunks],
|
||
show_progress_bar=False, normalize_embeddings=True)
|
||
scores = []
|
||
for fid, text in figma_texts.items():
|
||
d_chunks = _chunk_text(text)
|
||
if not d_chunks:
|
||
scores.append((fid, 0.0))
|
||
continue
|
||
d_vecs = model.encode(["passage: " + c for c in d_chunks],
|
||
show_progress_bar=False, normalize_embeddings=True)
|
||
max_sims = []
|
||
for qv in q_vecs:
|
||
sims = [_np_cosine(qv, dv) for dv in d_vecs]
|
||
max_sims.append(max(sims) if sims else 0)
|
||
score = sum(max_sims) / len(max_sims) if max_sims else 0
|
||
scores.append((fid, score))
|
||
return sorted(scores, key=lambda x: -x[1])
|
||
|
||
|
||
# ── Cross-encoder (reranker): 임베딩 X, 쌍 직접 채점
|
||
_cross_encoder_model = None
|
||
|
||
|
||
def _get_cross_encoder():
|
||
global _cross_encoder_model
|
||
if _cross_encoder_model is None:
|
||
from sentence_transformers import CrossEncoder
|
||
_cross_encoder_model = CrossEncoder('Dongjin-kr/ko-reranker', max_length=512)
|
||
return _cross_encoder_model
|
||
|
||
|
||
def method_cross_encoder(mdx_text, figma_texts):
|
||
"""Cross-encoder: 질의-문서 쌍을 직접 채점 (원본)"""
|
||
import math as _m
|
||
model = _get_cross_encoder()
|
||
pairs = [[mdx_text, text] for _, text in figma_texts.items()]
|
||
fids = list(figma_texts.keys())
|
||
raw_scores = model.predict(pairs, show_progress_bar=False)
|
||
scores = [(fids[i], 1 / (1 + _m.exp(-float(raw_scores[i])))) for i in range(len(fids))]
|
||
return sorted(scores, key=lambda x: -x[1])
|
||
|
||
|
||
def method_cross_encoder_summary(mdx_text, figma_texts):
|
||
"""Cross-encoder + 축약"""
|
||
import math as _m
|
||
model = _get_cross_encoder()
|
||
q_sum = _extract_summary(mdx_text)
|
||
pairs = [[q_sum, _extract_summary(text)] for _, text in figma_texts.items()]
|
||
fids = list(figma_texts.keys())
|
||
raw_scores = model.predict(pairs, show_progress_bar=False)
|
||
scores = [(fids[i], 1 / (1 + _m.exp(-float(raw_scores[i])))) for i in range(len(fids))]
|
||
return sorted(scores, key=lambda x: -x[1])
|
||
|
||
|
||
def method_cross_encoder_chunk(mdx_text, figma_texts):
|
||
"""Cross-encoder + 청크 (MDX chunk × Figma chunk 모든 쌍 채점 → 각 MDX chunk의 max → 평균)"""
|
||
import math as _m
|
||
model = _get_cross_encoder()
|
||
q_chunks = _chunk_text(mdx_text)
|
||
scores = []
|
||
for fid, text in figma_texts.items():
|
||
d_chunks = _chunk_text(text)
|
||
if not d_chunks or not q_chunks:
|
||
scores.append((fid, 0.0))
|
||
continue
|
||
# 모든 (q_chunk, d_chunk) 쌍을 만들어서 한 번에 채점 (배치)
|
||
pairs = []
|
||
for qc in q_chunks:
|
||
for dc in d_chunks:
|
||
pairs.append([qc, dc])
|
||
raw = model.predict(pairs, show_progress_bar=False)
|
||
# MDX chunk별 최대 → 평균
|
||
max_sims = []
|
||
idx = 0
|
||
for _ in q_chunks:
|
||
chunk_scores = raw[idx:idx + len(d_chunks)]
|
||
max_sims.append(float(max(chunk_scores)))
|
||
idx += len(d_chunks)
|
||
avg_logit = sum(max_sims) / len(max_sims)
|
||
scores.append((fid, 1 / (1 + _m.exp(-avg_logit))))
|
||
return sorted(scores, key=lambda x: -x[1])
|
||
|
||
|
||
# ── Word-avg (FastText 계열): 단어 수준 임베딩 평균
|
||
# 추가 모델 다운로드 없이, 이미 받은 ko-sroberta의 word_embedding 레이어를 그대로 사용
|
||
# (transformer contextualization 없이 static word vector만 평균)
|
||
def _wordavg_fn():
|
||
"""word embedding 평균을 내는 함수 factory (재사용)"""
|
||
import torch
|
||
model = _get_sbert()
|
||
transformer = model[0].auto_model
|
||
tokenizer = model.tokenizer
|
||
word_embeddings = transformer.embeddings.word_embeddings
|
||
device = next(transformer.parameters()).device
|
||
|
||
def _avg(text):
|
||
enc = tokenizer(text, return_tensors='pt', truncation=True, max_length=512)
|
||
enc = {k: v.to(device) for k, v in enc.items()}
|
||
with torch.no_grad():
|
||
embs = word_embeddings(enc['input_ids'])
|
||
mask = enc['attention_mask'].unsqueeze(-1).float()
|
||
summed = (embs * mask).sum(dim=1)
|
||
denom = mask.sum(dim=1).clamp(min=1)
|
||
return (summed / denom).squeeze().cpu().numpy()
|
||
|
||
return _avg
|
||
|
||
|
||
def method_wordavg(mdx_text, figma_texts):
|
||
"""단어 임베딩 평균 — FastText 계열 baseline (원본)"""
|
||
avg = _wordavg_fn()
|
||
q_vec = avg(mdx_text)
|
||
scores = []
|
||
for fid, text in figma_texts.items():
|
||
d_vec = avg(text)
|
||
scores.append((fid, _np_cosine(q_vec, d_vec)))
|
||
return sorted(scores, key=lambda x: -x[1])
|
||
|
||
|
||
def method_wordavg_summary(mdx_text, figma_texts):
|
||
"""Word-avg + 축약"""
|
||
avg = _wordavg_fn()
|
||
q_vec = avg(_extract_summary(mdx_text))
|
||
scores = []
|
||
for fid, text in figma_texts.items():
|
||
d_vec = avg(_extract_summary(text))
|
||
scores.append((fid, _np_cosine(q_vec, d_vec)))
|
||
return sorted(scores, key=lambda x: -x[1])
|
||
|
||
|
||
def method_wordavg_chunk(mdx_text, figma_texts):
|
||
"""Word-avg + 청크 (MDX chunk → max Figma chunk → 평균)"""
|
||
avg = _wordavg_fn()
|
||
q_chunks = _chunk_text(mdx_text)
|
||
q_vecs = [avg(c) for c in q_chunks]
|
||
scores = []
|
||
for fid, text in figma_texts.items():
|
||
d_chunks = _chunk_text(text)
|
||
if not d_chunks:
|
||
scores.append((fid, 0.0))
|
||
continue
|
||
d_vecs = [avg(c) for c in d_chunks]
|
||
max_sims = []
|
||
for qv in q_vecs:
|
||
sims = [_np_cosine(qv, dv) for dv in d_vecs]
|
||
max_sims.append(max(sims) if sims else 0)
|
||
score = sum(max_sims) / len(max_sims) if max_sims else 0
|
||
scores.append((fid, score))
|
||
return sorted(scores, key=lambda x: -x[1])
|
||
|
||
|
||
# ═══════════════════════════════════════
|
||
# AI-Meta (Phase 14) — 사전 AI 추출 개념(tag) 집합 Jaccard + IDF 가중
|
||
# ═══════════════════════════════════════
|
||
_metadata_db = None
|
||
|
||
|
||
def _load_metadata_db():
|
||
global _metadata_db
|
||
if _metadata_db is None:
|
||
import yaml
|
||
from pathlib import Path
|
||
p = Path(__file__).parent / "metadata_db.yaml"
|
||
with open(p, encoding="utf-8") as f:
|
||
_metadata_db = yaml.safe_load(f)
|
||
return _metadata_db
|
||
|
||
|
||
def _mdx_unit_to_meta_key(unit_id):
|
||
"""MDX 유닛 ID → metadata_db 키 매핑
|
||
세부 단위(intro-details, image, details, table, .1/.2 등)는 부모 섹션의 concepts 재사용"""
|
||
# 직접 매칭
|
||
meta = _load_metadata_db()
|
||
if unit_id in meta.get("mdx_sections", {}):
|
||
return unit_id
|
||
# 부모 섹션으로 폴백
|
||
# MDX01-2-details → MDX01-2
|
||
# MDX02-2.2-table → MDX02-2 (가장 가까운 상위)
|
||
# MDX03-2.1-table → MDX03-2
|
||
base = unit_id.split("-")[0] + "-" + unit_id.split("-")[1].split(".")[0] if "-" in unit_id else unit_id
|
||
base = re.sub(r"-details$|-image$|-table$|-intro$", "", unit_id)
|
||
if "." in base:
|
||
base = base.rsplit(".", 1)[0] # MDX02-2.2 → MDX02-2
|
||
# details/image/table 같은 suffix 떼기
|
||
base = re.sub(r"-(details|image|table|intro)$", "", base)
|
||
if base in meta.get("mdx_sections", {}):
|
||
return base
|
||
# intro-details → intro
|
||
if unit_id.endswith("-intro-details"):
|
||
b = unit_id.replace("-intro-details", "-intro")
|
||
if b in meta.get("mdx_sections", {}):
|
||
return b
|
||
return None
|
||
|
||
|
||
def method_ai_meta(mdx_text_or_uid, figma_texts):
|
||
"""AI-Meta: 사전 추출된 개념 집합 매칭 (IDF 가중 Jaccard)
|
||
사용: figma_texts의 키가 frame_id 문자열이어야 함.
|
||
MDX는 text가 아닌 uid를 받아야 해서 래퍼 필요 — 여기선 text 무시하고
|
||
외부에서 uid를 인자로 넘기도록 설계. phase14에서 method_ai_meta_for_uid 사용.
|
||
"""
|
||
# 이 함수는 실제론 쓰이지 않음 — uid 기반으로 phase14에서 직접 처리
|
||
raise NotImplementedError("method_ai_meta_for_uid 사용")
|
||
|
||
|
||
def _detect_mdx_layout(text):
|
||
"""MDX 본문에서 코드로 구조 힌트 감지"""
|
||
import re as _re
|
||
lines = text.split("\n")
|
||
# 최상위 블릿 개수 (들여쓰기 없는 `-` 또는 `*`)
|
||
top_bullets = [ln for ln in lines if _re.match(r"^\s{0,2}[-*]\s+\*\*", ln)]
|
||
n = len(top_bullets)
|
||
has_table = bool(_re.search(r"\n\|[^\n]+\|[^\n]+\|", text))
|
||
if has_table:
|
||
return "table"
|
||
if n == 3:
|
||
return "3col-parallel"
|
||
if n == 2:
|
||
return "compare-2col"
|
||
if n >= 4:
|
||
return f"{n}col-parallel"
|
||
return "single-column"
|
||
|
||
|
||
def _extract_mdx_concepts_by_code(text, vocabulary):
|
||
"""MDX에서 Kiwi 형태소로 토큰 추출 후 vocabulary(=Figma AI 개념집합)에 포함된 것만 유지
|
||
= AI 호출 없이 코드로 MDX → 개념 매핑"""
|
||
kiwi = _get_kiwi()
|
||
tokens = _extract_content_tokens(text, kiwi)
|
||
# vocabulary 교집합 (표기 일치 기준)
|
||
return {w for w in tokens if w in vocabulary}
|
||
|
||
|
||
def method_semi_ai(mdx_text, figma_texts, use_layout_bonus=True):
|
||
"""Semi-AI: Figma는 AI pre-tag (metadata_db) + MDX는 코드 추출.
|
||
- Figma 개념 집합은 사전 AI가 뽑아둔 것 사용
|
||
- MDX 개념은 Kiwi 형태소 중 Figma vocabulary에 있는 것만 선택
|
||
- IDF 가중 Jaccard + layout 일치 보너스
|
||
"""
|
||
import math as _m
|
||
import collections as _col
|
||
meta = _load_metadata_db()
|
||
figma_frames = meta.get("figma_frames", {})
|
||
|
||
# Figma 전체 vocabulary + IDF
|
||
vocab = set()
|
||
df = _col.Counter()
|
||
N = 0
|
||
for fid, v in figma_frames.items():
|
||
if isinstance(v, dict):
|
||
for c in v.get("concepts", []):
|
||
vocab.add(c)
|
||
df[c] += 1
|
||
N += 1
|
||
N = max(N, 1)
|
||
idf = {c: _m.log(N / cnt) if cnt > 0 else 0 for c, cnt in df.items()}
|
||
|
||
# MDX 코드 추출
|
||
mdx_concepts = _extract_mdx_concepts_by_code(mdx_text, vocab)
|
||
mdx_layout = _detect_mdx_layout(mdx_text)
|
||
|
||
scores = []
|
||
for fid in figma_texts:
|
||
v = figma_frames.get(str(fid), {})
|
||
if not isinstance(v, dict):
|
||
scores.append((fid, 0.0))
|
||
continue
|
||
fig_concepts = set(v.get("concepts", []))
|
||
fig_layout = v.get("layout_hint", "")
|
||
if not fig_concepts:
|
||
scores.append((fid, 0.0))
|
||
continue
|
||
inter = mdx_concepts & fig_concepts
|
||
union = mdx_concepts | fig_concepts
|
||
num = sum(idf.get(c, 0.5) for c in inter)
|
||
den = sum(idf.get(c, 0.5) for c in union)
|
||
s = num / den if den > 0 else 0
|
||
# 레이아웃 일치 보너스 (×1.3)
|
||
if use_layout_bonus and mdx_layout and mdx_layout == fig_layout:
|
||
s *= 1.3
|
||
scores.append((fid, s))
|
||
return sorted(scores, key=lambda x: -x[1])
|
||
|
||
|
||
def method_ai_meta_for_uid(unit_id, figma_texts):
|
||
"""unit_id 기반으로 metadata_db 찾아서 Figma와 개념 매칭"""
|
||
import math
|
||
meta = _load_metadata_db()
|
||
mdx_sections = meta.get("mdx_sections", {})
|
||
figma_meta = meta.get("figma_frames", {}) or meta.get("frames", {})
|
||
# yaml 구조에 따라 위치 다를 수 있음 — 상위 key 확인
|
||
if not figma_meta:
|
||
# 직접 키 검색
|
||
figma_meta = {k: v for k, v in meta.items()
|
||
if isinstance(k, str) and k.startswith("1171")}
|
||
|
||
mdx_key = _mdx_unit_to_meta_key(unit_id)
|
||
if mdx_key is None:
|
||
# fallback: 빈 결과
|
||
return [(fid, 0.0) for fid in figma_texts]
|
||
mdx_concepts = set(mdx_sections[mdx_key].get("concepts", []))
|
||
if not mdx_concepts:
|
||
return [(fid, 0.0) for fid in figma_texts]
|
||
|
||
# IDF: Figma 전체에서 각 개념이 얼마나 드문가
|
||
import collections
|
||
df = collections.Counter()
|
||
N = 0
|
||
for fid, v in figma_meta.items():
|
||
if not isinstance(v, dict):
|
||
continue
|
||
for c in v.get("concepts", []):
|
||
df[c] += 1
|
||
N += 1
|
||
N = max(N, 1)
|
||
idf = {c: math.log(N / cnt) if cnt > 0 else 0 for c, cnt in df.items()}
|
||
|
||
scores = []
|
||
for fid in figma_texts:
|
||
v = figma_meta.get(str(fid), {})
|
||
if not isinstance(v, dict):
|
||
scores.append((fid, 0.0))
|
||
continue
|
||
fig_concepts = set(v.get("concepts", []))
|
||
if not fig_concepts:
|
||
scores.append((fid, 0.0))
|
||
continue
|
||
intersection = mdx_concepts & fig_concepts
|
||
# IDF 가중 Jaccard: sum(idf) of common / sum(idf) of union
|
||
union = mdx_concepts | fig_concepts
|
||
num = sum(idf.get(c, 0.5) for c in intersection)
|
||
den = sum(idf.get(c, 0.5) for c in union)
|
||
s = num / den if den > 0 else 0
|
||
scores.append((fid, s))
|
||
return sorted(scores, key=lambda x: -x[1])
|
||
|
||
|
||
# ═══════════════════════════════════════
|
||
# Distinctive-Kiwi (Phase 8) — 형태소 분석 + corpus 공통어 제거 + IDF 매칭
|
||
# ═══════════════════════════════════════
|
||
_kiwi_instance = None
|
||
|
||
|
||
def _get_kiwi():
|
||
global _kiwi_instance
|
||
if _kiwi_instance is None:
|
||
from kiwipiepy import Kiwi
|
||
_kiwi_instance = Kiwi()
|
||
return _kiwi_instance
|
||
|
||
|
||
# Kiwi tag 중 의미 보유 품사만: 명사/동사/형용사/외국어/숫자
|
||
_MEANINGFUL_TAGS = ("NNG", "NNP", "NNB", "NR", "VV", "VA", "VX", "SL", "SH", "SN")
|
||
|
||
|
||
def _extract_content_tokens(text, kiwi=None):
|
||
"""명사/동사/형용사 등 내용어만 추출"""
|
||
if kiwi is None:
|
||
kiwi = _get_kiwi()
|
||
tokens = kiwi.tokenize(text)
|
||
return [t.form for t in tokens
|
||
if any(t.tag.startswith(tag) for tag in _MEANINGFUL_TAGS)
|
||
and len(t.form) >= 2] # 1글자 토큰 제거
|
||
|
||
|
||
def method_distinctive_kiwi(mdx_text, figma_texts,
|
||
general_threshold=0.5, return_debug=False):
|
||
"""Corpus 공통어 제거 + 각 Figma의 고유 키워드만 IDF 가중 매칭
|
||
general_threshold: 이 비율 이상의 Figma 프레임에 나타나면 공통어로 간주 (기본 50%)
|
||
"""
|
||
import math
|
||
from collections import Counter
|
||
kiwi = _get_kiwi()
|
||
|
||
# Figma 각각 토큰화
|
||
figma_tokens = {fid: _extract_content_tokens(text, kiwi)
|
||
for fid, text in figma_texts.items()}
|
||
|
||
# Corpus DF 계산 (몇 개 프레임에 나오는지)
|
||
df = Counter()
|
||
for toks in figma_tokens.values():
|
||
for w in set(toks):
|
||
df[w] += 1
|
||
N = len(figma_tokens) if figma_tokens else 1
|
||
general_words = {w for w, c in df.items() if c / N > general_threshold}
|
||
|
||
# 각 Figma의 고유 키워드 (공통어 제외)
|
||
figma_distinctive = {
|
||
fid: [w for w in toks if w not in general_words]
|
||
for fid, toks in figma_tokens.items()
|
||
}
|
||
|
||
# MDX도 같은 전처리
|
||
mdx_tokens = _extract_content_tokens(mdx_text, kiwi)
|
||
mdx_distinctive = [w for w in mdx_tokens if w not in general_words]
|
||
mdx_counter = Counter(mdx_distinctive)
|
||
|
||
# 스코어링: MDX ∩ Figma 교집합 × IDF
|
||
scores = []
|
||
for fid, toks in figma_distinctive.items():
|
||
figma_counter = Counter(toks)
|
||
s = 0.0
|
||
for w, m_cnt in mdx_counter.items():
|
||
if w in figma_counter:
|
||
idf = math.log(N / df[w]) if df[w] > 0 else 0
|
||
s += m_cnt * figma_counter[w] * idf
|
||
scores.append((fid, s))
|
||
|
||
# 정규화 [0,1]
|
||
max_s = max(s for _, s in scores) if scores else 1.0
|
||
if max_s > 0:
|
||
scores = [(fid, s / max_s) for fid, s in scores]
|
||
ranked = sorted(scores, key=lambda x: -x[1])
|
||
|
||
if return_debug:
|
||
return ranked, {
|
||
"general_words": sorted(general_words),
|
||
"mdx_distinctive": mdx_distinctive,
|
||
"figma_distinctive": figma_distinctive,
|
||
"df": dict(df),
|
||
}
|
||
return ranked
|
||
|
||
|
||
# Hybrid: Distinctive-Kiwi로 top-k → Cross rerank
|
||
def method_hybrid_distinctive_cross(mdx_text, figma_texts, top_k=5):
|
||
base = method_distinctive_kiwi(mdx_text, figma_texts)
|
||
top_k_ids = [fid for fid, _ in base[:top_k]]
|
||
candidates = {fid: figma_texts[fid] for fid in top_k_ids}
|
||
reranked = method_cross_encoder(mdx_text, candidates)
|
||
picked = set(fid for fid, _ in reranked)
|
||
tail = [(fid, 0.0) for fid in figma_texts if fid not in picked]
|
||
return reranked + tail
|
||
|
||
|
||
# ═══════════════════════════════════════
|
||
# Hybrid Pipeline (Phase 7) — Retriever → Reranker → 구조 감점
|
||
# ═══════════════════════════════════════
|
||
def method_hybrid_bm25_cross(mdx_text, figma_texts, top_k=5):
|
||
"""Kiwi+BM25로 top-k 후보 추출 → Cross-encoder로 rerank"""
|
||
bm25_scores = method_kiwi_bm25(mdx_text, figma_texts)
|
||
top_k_ids = [fid for fid, _ in bm25_scores[:top_k]]
|
||
candidates = {fid: figma_texts[fid] for fid in top_k_ids}
|
||
reranked = method_cross_encoder(mdx_text, candidates)
|
||
picked = set(fid for fid, _ in reranked)
|
||
tail = [(fid, 0.0) for fid in figma_texts if fid not in picked]
|
||
return reranked + tail
|
||
|
||
|
||
def method_hybrid_bm25_sbert_chunk(mdx_text, figma_texts, top_k=5):
|
||
"""Kiwi+BM25로 top-k 후보 추출 → 청킹+SBERT로 rerank"""
|
||
bm25_scores = method_kiwi_bm25(mdx_text, figma_texts)
|
||
top_k_ids = [fid for fid, _ in bm25_scores[:top_k]]
|
||
candidates = {fid: figma_texts[fid] for fid in top_k_ids}
|
||
reranked = method_sbert_chunk(mdx_text, candidates)
|
||
picked = set(fid for fid, _ in reranked)
|
||
tail = [(fid, 0.0) for fid in figma_texts if fid not in picked]
|
||
return reranked + tail
|
||
|
||
|
||
def _apply_struct_penalty(ranked_scores, mdx_text, figma_texts, alpha=0.3):
|
||
"""rerank 결과에 구조 유사도를 가중 평균으로 반영
|
||
alpha=0.3: 최종 = 0.7×rerank + 0.3×struct_norm"""
|
||
struct_scores = method_structural(mdx_text, figma_texts)
|
||
struct_map = dict(struct_scores)
|
||
s_max = max(struct_map.values()) if struct_map else 1.0
|
||
s_max = s_max if s_max > 0 else 1.0
|
||
# rerank 점수 정규화
|
||
r_max = max(s for _, s in ranked_scores) if ranked_scores else 1.0
|
||
r_max = r_max if r_max > 0 else 1.0
|
||
combined = []
|
||
for fid, r_score in ranked_scores:
|
||
r_norm = r_score / r_max
|
||
s_norm = struct_map.get(fid, 0) / s_max
|
||
final = (1 - alpha) * r_norm + alpha * s_norm
|
||
combined.append((fid, final))
|
||
return sorted(combined, key=lambda x: -x[1])
|
||
|
||
|
||
def method_hybrid_bm25_cross_struct(mdx_text, figma_texts, top_k=5, alpha=0.3):
|
||
"""BM25+Cross rerank + 구조 감점"""
|
||
bm25_scores = method_kiwi_bm25(mdx_text, figma_texts)
|
||
top_k_ids = [fid for fid, _ in bm25_scores[:top_k]]
|
||
candidates = {fid: figma_texts[fid] for fid in top_k_ids}
|
||
reranked = method_cross_encoder(mdx_text, candidates)
|
||
final = _apply_struct_penalty(reranked, mdx_text, candidates, alpha=alpha)
|
||
picked = set(fid for fid, _ in final)
|
||
tail = [(fid, 0.0) for fid in figma_texts if fid not in picked]
|
||
return final + tail
|
||
|
||
|
||
def method_hybrid_bm25_sbert_chunk_struct(mdx_text, figma_texts, top_k=5, alpha=0.3):
|
||
"""BM25+SBERT-청크 rerank + 구조 감점"""
|
||
bm25_scores = method_kiwi_bm25(mdx_text, figma_texts)
|
||
top_k_ids = [fid for fid, _ in bm25_scores[:top_k]]
|
||
candidates = {fid: figma_texts[fid] for fid in top_k_ids}
|
||
reranked = method_sbert_chunk(mdx_text, candidates)
|
||
final = _apply_struct_penalty(reranked, mdx_text, candidates, alpha=alpha)
|
||
picked = set(fid for fid, _ in final)
|
||
tail = [(fid, 0.0) for fid in figma_texts if fid not in picked]
|
||
return final + tail
|
||
|
||
|
||
# ═══════════════════════════════════════
|
||
# 8-reversed. BM25 filter → 구조로 rerank (역방향 하이브리드)
|
||
# ═══════════════════════════════════════
|
||
def method_bm25_then_struct(mdx_text, figma_texts, top_k=10):
|
||
# Step 1: BM25로 top-K 후보 선정
|
||
bm25_scores = method_bm25(mdx_text, figma_texts)
|
||
candidate_ids = [fid for fid, _ in bm25_scores[:top_k]]
|
||
candidates = {fid: figma_texts[fid] for fid in candidate_ids}
|
||
# Step 2: 구조로 rerank
|
||
struct_scores = method_structural(mdx_text, candidates)
|
||
# 나머지는 뒤에
|
||
result_ids = set(fid for fid, _ in struct_scores)
|
||
tail = [(fid, 0.0) for fid in figma_texts if fid not in result_ids]
|
||
return struct_scores + tail
|