"""Pipeline Step 9 — V2 실패 원인 진단. 목적: V2(의미 기반 rerank) 가 TARGET 4 섹션에서 V1 1위를 한 번도 바꾸지 못한 이유를 **숫자로** 분리 진단. 가설 A~E 중 어느 것이 주범인지 가려낸다. 가설: A 입력 품질 (MDX summary) — title+첫문단+slot labels 가 부실한가? B Frame content 품질 — analysis.md "내용 설명" 이 프레임 간 구분을 못 주는가? C 모델 — ko-sroberta-multitask 가 BIM/DX 도메인에 약한가? (이 스크립트는 모델 교체 안 함. 다음 단계) D 입력 조합 — summary vs 전문 vs title 중 어느 게 나은가? E rerank 범위 — V1 Top-5 로 잘라서 leverage 가 안 보이는가? 사용자 지침 반영: 1. cosine 절대값 대신 rank / gap / percentile 을 주 지표로 2. V1 Top-5 내부 진단과 32 전체 진단 **분리** 3. frame content 길이/중복도 별도 점검 입력: - mdx_matching_result.yaml - structure_ontology / analysis.md (frame.content, phase_common.load_32_frames) - MDX 7 섹션 원문 출력: - V2_DIAGNOSIS.yaml (원시 수치) - V2_DIAGNOSIS.md (해석) """ from collections import Counter import datetime from pathlib import Path import sys import numpy as np import yaml HERE = Path(__file__).parent sys.path.insert(0, str(HERE)) from phase_common import load_32_frames, load_frame_index from detect_mdx import detect_mdx_analysis from embeddings import embed_texts, cosine as cos_sim from pipeline_01_extract_nodes import MDX_SECTIONS, MDX_DIR MD_PATH = HERE / 'V2_DIAGNOSIS.md' YAML_PATH = HERE / 'V2_DIAGNOSIS.yaml' # 입력 variant — ablation 대상 INPUT_VARIANTS = [ ('summary', '현재 V2 입력: title + 첫 일반 문단 + slot labels (detect_mdx.build_summary)'), ('full_text', 'MDX 섹션 전문'), ('title_only', '섹션 제목만'), ('summary_plus_tokens','summary + 정규화 토큰 목록 (normalized_text_tokens.yaml)'), ] TOP_K = 5 # V1 Top-K (현재 합의값) # ============================================================ # 유틸 # ============================================================ def extract_mdx_raw(sid): cfg = MDX_SECTIONS[sid] p = MDX_DIR / cfg['file'] lines = p.read_text(encoding='utf-8').split('\n') start_idx = None for i, ln in enumerate(lines): if ln.strip() == cfg['start'].strip(): start_idx = i break end_idx = len(lines) if cfg.get('end_prefix'): for i in range(start_idx + 1, len(lines)): if lines[i].strip().startswith(cfg['end_prefix']): end_idx = i break section = lines[start_idx:end_idx] title = section[0].lstrip('#').strip() return title, '\n'.join(section) def build_input(variant_id, title, raw_text, tokens): if variant_id == 'summary': analysis = detect_mdx_analysis(raw_text, title) return analysis['summary'] if variant_id == 'full_text': return raw_text if variant_id == 'title_only': return title if variant_id == 'summary_plus_tokens': analysis = detect_mdx_analysis(raw_text, title) return analysis['summary'] + ' ' + ' '.join(tokens) raise ValueError(variant_id) def spearman(rank_a_map, rank_b_map, keys): """Spearman rho on given key list. rank_*_map: {key: 1-based rank}.""" n = len(keys) if n < 2: return None d_sq = sum((rank_a_map[k] - rank_b_map[k]) ** 2 for k in keys) return round(1 - (6 * d_sq) / (n * (n * n - 1)), 4) # ============================================================ # 지표 계산 # ============================================================ def full32_metrics(mdx_vec, frame_vecs, fids, answer_fid=None): cosines = np.array([float(cos_sim(mdx_vec, frame_vecs[i])) for i in range(len(fids))]) order = np.argsort(-cosines) ranked = [(fids[i], float(cosines[i])) for i in order] out = { 'top5': [{'frame_id': fid, 'cosine': round(c, 4)} for fid, c in ranked[:5]], 'top1_cosine': round(ranked[0][1], 4), 'top2_cosine': round(ranked[1][1], 4), 'top1_top2_gap': round(ranked[0][1] - ranked[1][1], 4), 'top1_top5_gap': round(ranked[0][1] - ranked[4][1], 4), 'cosine_mean_32': round(float(cosines.mean()), 4), 'cosine_std_32': round(float(cosines.std()), 4), 'cosine_max_32': round(float(cosines.max()), 4), 'cosine_min_32': round(float(cosines.min()), 4), } if answer_fid: ordered_fids = [fid for fid, _ in ranked] rank = ordered_fids.index(answer_fid) + 1 # percentile: 1위 = 100%, 32위 = 0% pct = round(100 * (len(fids) - rank) / (len(fids) - 1), 1) answer_cos = float(cosines[fids.index(answer_fid)]) out['answer_rank_in_32'] = rank out['answer_percentile'] = pct out['answer_cosine'] = round(answer_cos, 4) # gap to top1 out['answer_gap_to_top1'] = round(ranked[0][1] - answer_cos, 4) return out def v1_top5_internal_metrics(mdx_vec, frame_vecs, fids, v1_top5, answer_fid=None): """V1 Top-5 내부에서의 cosine rerank 진단.""" fid_to_idx = {fid: i for i, fid in enumerate(fids)} v1_fids = [r['frame_id'] for r in v1_top5] cos_map = {fid: float(cos_sim(mdx_vec, frame_vecs[fid_to_idx[fid]])) for fid in v1_fids} cos_ordered = sorted(v1_fids, key=lambda fid: -cos_map[fid]) v1_rank_map = {fid: idx + 1 for idx, fid in enumerate(v1_fids)} cos_rank_map = {fid: idx + 1 for idx, fid in enumerate(cos_ordered)} sp = spearman(v1_rank_map, cos_rank_map, v1_fids) top1_cos = cos_map[cos_ordered[0]] top2_cos = cos_map[cos_ordered[1]] if len(cos_ordered) >= 2 else None gap = round(top1_cos - top2_cos, 4) if top2_cos is not None else None out = { 'v1_top5_fids': v1_fids, 'cosine_rerank_order': cos_ordered, 'cosine_values': {fid: round(cos_map[fid], 4) for fid in v1_fids}, 'top1_top2_gap_in_top5': gap, 'spearman_v1_vs_cosine_within_top5': sp, } if answer_fid and answer_fid in v1_fids: out['answer_v1_rank_in_top5'] = v1_rank_map[answer_fid] out['answer_cosine_rank_in_top5'] = cos_rank_map[answer_fid] return out def pairwise_stats(vecs): n = len(vecs) arr = [] for i in range(n): for j in range(i + 1, n): arr.append(float(cos_sim(vecs[i], vecs[j]))) a = np.array(arr) return { 'pair_count': len(arr), 'mean': round(float(a.mean()), 4), 'std': round(float(a.std()), 4), 'median': round(float(np.median(a)), 4), 'min': round(float(a.min()), 4), 'max': round(float(a.max()), 4), 'p90': round(float(np.quantile(a, 0.9)), 4), } def near_duplicate_pairs(vecs, fids, threshold=0.9): pairs = [] for i in range(len(vecs)): for j in range(i + 1, len(vecs)): c = float(cos_sim(vecs[i], vecs[j])) if c >= threshold: pairs.append({'a': fids[i], 'b': fids[j], 'cosine': round(c, 4)}) pairs.sort(key=lambda x: -x['cosine']) return pairs def length_stats(texts): lens = [len(t) for t in texts] a = np.array(lens) return { 'count': len(lens), 'mean_chars': round(float(a.mean()), 1), 'median_chars': round(float(np.median(a)), 1), 'min_chars': int(a.min()), 'max_chars': int(a.max()), 'std_chars': round(float(a.std()), 1), } # ============================================================ # 리포트 생성 # ============================================================ def interpret_result(diag): """가설 A~E 판정 요약 문장.""" lines = [] pw = diag['frame_content_quality']['pairwise_cosine_stats'] if pw['mean'] > 0.8: lines.append( f"- **가설 B (frame content 품질) 의심**: 프레임 간 평균 pairwise cosine = {pw['mean']:.3f} " f"— 모든 프레임이 서로 '비슷'해서 cosine 이 구분 신호를 주기 어려움" ) elif pw['mean'] > 0.7: lines.append( f"- **가설 B 약간 의심**: pairwise cosine 평균 = {pw['mean']:.3f} — 구분이 다소 약함" ) else: lines.append( f"- 가설 B 약함: pairwise cosine 평균 = {pw['mean']:.3f} — 프레임 간 구분은 가능한 수준" ) # TARGET 섹션의 정답 rank (summary variant 기준) summary_v = diag['variant_results'].get('summary', {}) target_answer_ranks = [] for sid, data in summary_v.items(): ans_rank = data['full_32'].get('answer_rank_in_32') if ans_rank is not None: target_answer_ranks.append((sid, ans_rank)) if target_answer_ranks: above_top5 = sum(1 for _, r in target_answer_ranks if r <= 5) lines.append( f"- **정답 프레임의 32-전체 cosine 랭킹 (summary)**: " + ', '.join(f"{sid}={r}위" for sid, r in target_answer_ranks) + f" (Top-5 내 {above_top5}/{len(target_answer_ranks)})" ) if above_top5 < len(target_answer_ranks): lines.append( f" → 일부 섹션에서 정답이 32-전체 cosine Top-5 밖 — " f"**가설 A/C 의심**: 입력(summary) 또는 모델이 정답 신호를 못 잡음" ) # V1 Top-5 내 Spearman 평균 spearmans = [] for sid, data in summary_v.items(): s = data['v1_top5'].get('spearman_v1_vs_cosine_within_top5') if s is not None: spearmans.append(s) if spearmans: m = np.mean(spearmans) lines.append( f"- **V1 Top-5 내 V1↔cosine Spearman 평균 = {m:.3f}** " f"(1=완전 일치, -1=완전 반대) " f"→ V1 순서와 cosine 순서가 {'매우 유사' if m > 0.7 else '유사' if m > 0.3 else '거의 무관'}" ) if m > 0.7: lines.append( f" → **가설 E 의심**: V1 이 이미 semantic 유사도로 후보를 정렬해버려 " f"V2 rerank 가 새 정보를 주기 어려움" ) # Variant 간 비교 (top1_top2_gap 평균) lines.append('') lines.append('**Variant 별 32-전체 Top1-Top2 평균 gap** (클수록 구분력 강함):') for vid, _ in INPUT_VARIANTS: gaps = [ d['full_32']['top1_top2_gap'] for d in diag['variant_results'].get(vid, {}).values() ] if gaps: lines.append(f" - `{vid}`: mean gap = {np.mean(gaps):.4f}") return '\n'.join(lines) def write_md(diag): lines = [] lines.append('# V2 실패 원인 진단 리포트') lines.append('') lines.append(f"_생성: {diag['meta']['timestamp']} · 모델: `{diag['meta']['model']}`_") lines.append('') lines.append( 'V2(의미 기반 rerank) 가 TARGET 4 섹션 모두에서 V1 1위를 바꾸지 못한 이유를 ' '가설 A~E 로 분리 진단한 결과.' ) lines.append('') # 해석 lines.append('## 1. 진단 해석 (요약)') lines.append('') lines.append(interpret_result(diag)) lines.append('') # 입력 품질 lines.append('## 2. 입력 품질') lines.append('') lines.append('### 2.1 Frame content (32 프레임)') lines.append('') fc = diag['frame_content_quality'] lines.append( f"- 글자 수: 평균 **{fc['length_stats']['mean_chars']}** · 중앙값 {fc['length_stats']['median_chars']} " f"· 최소 {fc['length_stats']['min_chars']} / 최대 {fc['length_stats']['max_chars']}" ) pw = fc['pairwise_cosine_stats'] lines.append( f"- 프레임 간 pairwise cosine 분포 ({pw['pair_count']} 쌍): " f"평균 **{pw['mean']}** · 중앙값 {pw['median']} · p90 {pw['p90']} " f"· 범위 [{pw['min']}, {pw['max']}]" ) if fc['near_duplicates']: lines.append(f"- 근접 중복 쌍 (cosine ≥ 0.9): **{len(fc['near_duplicates'])}개**") for pair in fc['near_duplicates'][:5]: lines.append(f" - `{pair['a']}` ↔ `{pair['b']}` cosine={pair['cosine']}") else: lines.append('- 근접 중복 쌍 (cosine ≥ 0.9): 없음') lines.append('') # MDX input variants lines.append('### 2.2 MDX 섹션별 입력 (variant 비교)') lines.append('') lines.append('| 섹션 | summary 글자 | full 글자 | title 글자 | summary+tokens 글자 |') lines.append('|---|---:|---:|---:|---:|') for sid, data in diag['mdx_input_stats'].items(): lines.append( f"| **{sid}** | {data['summary']} | {data['full_text']} | " f"{data['title_only']} | {data['summary_plus_tokens']} |" ) lines.append('') # Full-32 진단 (summary variant) lines.append('## 3. 32-전체 cosine 진단 (현재 V2 입력 = summary)') lines.append('') lines.append('| 섹션 | 정답 rank | %ile | 정답 cosine | top1 | top1-top2 gap | 32 평균 cosine |') lines.append('|---|---:|---:|---:|---:|---:|---:|') for sid, data in diag['variant_results']['summary'].items(): f32 = data['full_32'] ans_rank = f32.get('answer_rank_in_32', '—') ans_pct = f32.get('answer_percentile', '—') ans_cos = f32.get('answer_cosine', '—') lines.append( f"| **{sid}** | {ans_rank} | {ans_pct} | {ans_cos} | " f"{f32['top1_cosine']} | {f32['top1_top2_gap']} | {f32['cosine_mean_32']} |" ) lines.append('') # V1 Top-5 내부 lines.append('## 4. V1 Top-5 내부 cosine 진단 (현재 V2 입력 = summary)') lines.append('') lines.append('| 섹션 | V1 순서 | cosine 순서 | Spearman | Top1-Top2 gap |') lines.append('|---|---|---|---:|---:|') # Build frame_id → frame_num lookup fid_to_num = diag['frame_id_to_number'] for sid, data in diag['variant_results']['summary'].items(): t5 = data['v1_top5'] v1_order = ' → '.join(str(fid_to_num.get(fid, '?')) for fid in t5['v1_top5_fids']) cos_order = ' → '.join(str(fid_to_num.get(fid, '?')) for fid in t5['cosine_rerank_order']) sp = t5.get('spearman_v1_vs_cosine_within_top5') gap = t5.get('top1_top2_gap_in_top5') lines.append(f"| **{sid}** | {v1_order} | {cos_order} | {sp} | {gap} |") lines.append('') # Variant ablation lines.append('## 5. 입력 variant ablation') lines.append('') lines.append('TARGET 4 섹션의 **정답 rank (32-전체)** variant 별 비교.') lines.append('') lines.append('| 섹션 | summary | full_text | title_only | summary+tokens |') lines.append('|---|---:|---:|---:|---:|') target_sids = [s for s in diag['variant_results']['summary'] if diag['variant_results']['summary'][s]['full_32'].get('answer_rank_in_32') is not None] for sid in target_sids: row = [f"**{sid}**"] for vid, _ in INPUT_VARIANTS: ar = diag['variant_results'][vid][sid]['full_32'].get('answer_rank_in_32', '—') row.append(str(ar)) lines.append('| ' + ' | '.join(row) + ' |') lines.append('') lines.append('TARGET 4 섹션의 **Top1-Top2 gap (32-전체)** variant 별 비교 — 크면 구분력 강함.') lines.append('') lines.append('| 섹션 | summary | full_text | title_only | summary+tokens |') lines.append('|---|---:|---:|---:|---:|') for sid in target_sids: row = [f"**{sid}**"] for vid, _ in INPUT_VARIANTS: g = diag['variant_results'][vid][sid]['full_32']['top1_top2_gap'] row.append(f"{g:.4f}") lines.append('| ' + ' | '.join(row) + ' |') lines.append('') # 다음 단계 lines.append('## 6. 다음 단계') lines.append('') lines.append( '이 리포트의 수치를 근거로 주범 가설을 확정한 뒤 해당 축만 개선 — ' '전체 리팩터 전에 원인 분리가 목적.' ) lines.append('') lines.append( '- 가설 B (frame content 품질) 이 주범이면: `analysis.md` "내용 설명" 을 재작성해야 함. ' '분량을 늘리고 구조적 특징을 더 구체적으로 적어 구분도 확보.' ) lines.append( '- 가설 A (MDX summary) 이 주범이면: `detect_mdx.build_summary` 를 개선 또는 variant 교체.' ) lines.append( '- 가설 E (V1 Top-5 범위) 가 주범이면: rerank 범위를 Top-10 / Top-15 로 확장, ' '또는 V2 를 32-전체 rerank 로 변경.' ) lines.append( '- 가설 C (모델) 은 A/B/D/E 배제 후에만 검토 (과잉 실험 방지).' ) MD_PATH.write_text('\n'.join(lines), encoding='utf-8') # ============================================================ # 메인 # ============================================================ def main(): # 1. V1 결과 로드 v1 = yaml.safe_load((HERE / 'mdx_matching_result.yaml').read_text(encoding='utf-8')) answer_map = v1['meta']['answer_map'] v1_sections = v1['mdx_sections'] # 2. 프레임 로드 + content 임베딩 frames = load_32_frames() idx_data, frame_to_short = load_frame_index() fids = list(frames.keys()) frame_contents = [frames[fid].get('content', '') for fid in fids] frame_num_map = {fid: int(frame_to_short[fid]) for fid in fids} print("[V2 진단] 32 프레임 content 임베딩...") frame_vecs = embed_texts(frame_contents) # 3. 프레임 content 품질 분석 frame_content_quality = { 'length_stats': length_stats(frame_contents), 'pairwise_cosine_stats': pairwise_stats(frame_vecs), 'near_duplicates': near_duplicate_pairs(frame_vecs, fids, threshold=0.9), } # 4. 토큰 로드 (summary_plus_tokens variant 용) tokens_yaml = yaml.safe_load((HERE / 'normalized_text_tokens.yaml').read_text(encoding='utf-8')) section_tokens = {sid: tokens_yaml['mdx'][sid].get('unique_tokens', []) for sid in v1_sections} # 5. variant 별 진단 mdx_input_stats = {} variant_results = {vid: {} for vid, _ in INPUT_VARIANTS} for sid in v1_sections: title, raw_text = extract_mdx_raw(sid) tokens = section_tokens.get(sid, []) # 입력 길이 통계 mdx_input_stats[sid] = {} for vid, _ in INPUT_VARIANTS: inp = build_input(vid, title, raw_text, tokens) mdx_input_stats[sid][vid] = len(inp) # variant 별 임베딩 + 진단 answer_num = answer_map.get(sid) answer_fid = None if answer_num is not None: # frame_num → frame_id 역매핑 for fid, num in frame_num_map.items(): if num == answer_num: answer_fid = fid break for vid, _ in INPUT_VARIANTS: inp = build_input(vid, title, raw_text, tokens) mdx_vec = embed_texts([inp])[0] full_32 = full32_metrics(mdx_vec, frame_vecs, fids, answer_fid=answer_fid) v1_top5_list = v1_sections[sid]['rank_by_matching_score'][:TOP_K] v1_top5_d = v1_top5_internal_metrics( mdx_vec, frame_vecs, fids, v1_top5_list, answer_fid=answer_fid, ) variant_results[vid][sid] = { 'full_32': full_32, 'v1_top5': v1_top5_d, } # 6. 집계 + 출력 diag = { 'meta': { 'timestamp': datetime.datetime.now().isoformat(timespec='seconds'), 'model': 'jhgan/ko-sroberta-multitask', 'top_k': TOP_K, 'answer_map': answer_map, 'holdout_sections': v1['meta']['holdout_sections'], 'variants': {vid: desc for vid, desc in INPUT_VARIANTS}, 'hypotheses': { 'A': '입력 품질 — MDX summary 가 부실', 'B': 'Frame content 품질 — 프레임 간 구분 안 됨', 'C': '모델 — ko-sroberta 가 BIM/DX 도메인에 약함', 'D': '입력 조합 — variant 간 격차', 'E': 'rerank 범위 — V1 Top-5 로 좁혀 leverage 없음', }, }, 'frame_content_quality': frame_content_quality, 'frame_id_to_number': frame_num_map, 'mdx_input_stats': mdx_input_stats, 'variant_results': variant_results, } YAML_PATH.write_text( yaml.safe_dump(diag, allow_unicode=True, sort_keys=False, width=1000), encoding='utf-8', ) write_md(diag) print("=" * 70) print("V2 진단 완료") print("=" * 70) print(f" yaml: {YAML_PATH}") print(f" md: {MD_PATH}") print() pw = frame_content_quality['pairwise_cosine_stats'] print(f" frame content pairwise cosine mean: {pw['mean']} (≥0.8 이면 가설 B 의심)") # TARGET 평균 지표 spearmans = [ variant_results['summary'][sid]['v1_top5']['spearman_v1_vs_cosine_within_top5'] for sid in variant_results['summary'] ] spearmans = [s for s in spearmans if s is not None] if spearmans: print(f" V1↔cosine Spearman (Top-5 내부) 평균: {np.mean(spearmans):.3f}") target_sids_l = [s for s in variant_results['summary'] if variant_results['summary'][s]['full_32'].get('answer_rank_in_32') is not None] if target_sids_l: ranks = [variant_results['summary'][s]['full_32']['answer_rank_in_32'] for s in target_sids_l] print(f" TARGET 정답 32-전체 rank (summary): {ranks}") if __name__ == '__main__': main()