#!/usr/bin/env python3 """ WBS-9.6 Phase 1 & 2: LLM Radar Trust Tier + Loading Order Phase 1: Trust 라벨 시스템 정의 Phase 2: 5-tier 로딩 순서 구현 """ import yaml import json from pathlib import Path from typing import Dict, List, Tuple from datetime import datetime import sys class LLMRadarPhase12: """LLM Radar Trust Tier System (Phase 1 & 2)""" def __init__(self): self.trust_tier_spec = Path("spec/llm_radar_trust_tiers_v1.yaml") self.context_cache = {} self.load_order_sequence = [] self.results = { "timestamp": datetime.now().isoformat(), "phase_1": {}, "phase_2": {}, "summary": {} } def load_trust_tier_spec(self) -> Dict: """Trust tier 스펙 로드""" if not self.trust_tier_spec.exists(): print(f"[ERROR] Trust tier spec not found: {self.trust_tier_spec}") return {} with open(self.trust_tier_spec, encoding='utf-8') as f: spec = yaml.safe_load(f) return spec def phase_1_build_trust_labels(self) -> Dict: """Phase 1: 모든 문서에 trust 라벨 지정""" spec = self.load_trust_tier_spec() if not spec: return {"status": "FAILED", "error": "Spec not loaded"} # Trust tier 정보 추출 trust_tiers = spec.get("trust_tier_system", {}) doc_classification = spec.get("document_classification", {}) trust_labels = {} # 각 분류에서 문서 추출 for category, config in doc_classification.items(): tier = config.get("tier") documents = config.get("documents", []) if tier not in trust_tiers: print(f"[WARNING] Unknown tier: {tier} for category {category}") continue tier_info = trust_tiers[tier] trust_level = tier_info.get("trust_level", 0) for doc in documents: trust_labels[doc] = { "tier": tier, "trust_level": trust_level, "category": category, "priority": tier_info.get("loading_priority", 0) } self.results["phase_1"]["total_documents"] = len(trust_labels) self.results["phase_1"]["tier_distribution"] = self._count_by_tier(trust_labels) self.results["phase_1"]["trust_labels"] = trust_labels print(f"[Phase 1] Trust labels created for {len(trust_labels)} documents") return { "status": "SUCCESS", "documents_labeled": len(trust_labels), "tiers": list(set(label["tier"] for label in trust_labels.values())) } def _count_by_tier(self, labels: Dict) -> Dict: """Tier별 문서 개수 계산""" counts = {} for label in labels.values(): tier = label["tier"] counts[tier] = counts.get(tier, 0) + 1 return counts def phase_2_build_loading_order(self) -> Dict: """Phase 2: 5-tier 로딩 순서 정의""" spec = self.load_trust_tier_spec() if not spec: return {"status": "FAILED", "error": "Spec not loaded"} loading_strategy = spec.get("loading_strategy", {}) trust_tiers = spec.get("trust_tier_system", {}) # 로딩 순서 정의 load_sequence = [] # Phase 1: Canonical (trust_level=100) canonical_docs = [ "spec/12_field_dictionary.yaml", "spec/14_raw_workbook_mapping.yaml", "spec/11_market_regime.yaml" ] load_sequence.append({ "phase": 1, "name": "Canonical References", "tier": "canonical", "trust_threshold": 100, "documents": canonical_docs, "action": "Always load" }) # Phase 2: Adapter (trust_level=80) adapter_docs = [ "spec/09_decision_flow.yaml", "spec/13_formula_registry.yaml" ] load_sequence.append({ "phase": 2, "name": "Adapter Bridges", "tier": "adapter", "trust_threshold": 80, "documents": adapter_docs, "action": "Load if no conflict with canonical" }) # Phase 3: Reference (trust_level=60) reference_docs = [ "docs/WBS_9_1_F14_MIGRATION_COMPLETE_2026_06_22.md", "docs/WBS_9_4_INCIDENT_RESPONSE_PLAYBOOK_2026_06_22.md" ] load_sequence.append({ "phase": 3, "name": "Reference Context", "tier": "reference", "trust_threshold": 60, "documents": reference_docs, "action": "Load as secondary context" }) # Phase 4: Search-Based load_sequence.append({ "phase": 4, "name": "Search-Based Context", "tier": "search", "trust_threshold": 50, "documents": [], # Dynamic - determined by query "action": "Retrieve by relevance + tier" }) # Phase 5: Fallback load_sequence.append({ "phase": 5, "name": "LLM Knowledge Fallback", "tier": "fallback", "trust_threshold": 0, "documents": [], # LLM internal "action": "Use LLM training data only" }) self.load_order_sequence = load_sequence self.results["phase_2"]["loading_phases"] = load_sequence self.results["phase_2"]["total_phases"] = len(load_sequence) print(f"[Phase 2] Loading order defined for {len(load_sequence)} phases") return { "status": "SUCCESS", "phases": len(load_sequence), "total_documents_to_load": sum(len(p.get("documents", [])) for p in load_sequence) } def generate_llm_context_builder_pseudo_code(self) -> str: """LLM context builder를 위한 의사 코드 생성""" pseudo_code = """ // LLM Radar Phase 1 & 2 Context Builder function buildContextWithTrustTiers(query, userContext) { context = [] loadedDocs = set() // Phase 1: Load Canonical (100% trust) for (doc in canonicalDocuments) { if (doc.exists()) { content = loadDocument(doc) context.append({ tier: "canonical", trust_level: 100, content: content, loaded_at: phase_1 }) loadedDocs.add(doc) } } // Phase 2: Load Adapter (80% trust) for (doc in adapterDocuments) { if (doc.exists() && doc not in loadedDocs) { content = loadDocument(doc) if (!conflictsWithCanonical(content, context)) { context.append({ tier: "adapter", trust_level: 80, content: content, loaded_at: phase_2 }) loadedDocs.add(doc) } } } // Phase 3: Load Reference (60% trust) for (doc in referenceDocuments) { if (doc.exists() && doc not in loadedDocs) { content = loadDocument(doc) context.append({ tier: "reference", trust_level: 60, content: content, loaded_at: phase_3 }) loadedDocs.add(doc) } } // Phase 4: Search-Based (50% trust) relevantDocs = searchDocuments(query, threshold=50) for (doc in relevantDocs) { if (doc not in loadedDocs) { content = loadDocument(doc) context.append({ tier: "search", trust_level: 50, content: content, relevance_score: doc.score, loaded_at: phase_4 }) loadedDocs.add(doc) } } // Phase 5: Fallback (0% trust - use LLM knowledge) if (context.isEmpty()) { context.append({ tier: "fallback", trust_level: 0, source: "llm_training_data", loaded_at: phase_5 }) } return context } // Conflict detection function conflictsWithCanonical(adapterDoc, canonicalContext) { for (canonical in canonicalContext) { if (contradicts(adapterDoc, canonical)) { logWarning("Adapter contradicts canonical") return true } } return false } """ return pseudo_code def generate_report(self) -> Dict: """전체 리포트 생성""" print("\n" + "="*80) print("WBS-9.6 Phase 1 & 2: LLM Radar Trust Tier System") print("="*80) # Phase 1 phase1_result = self.phase_1_build_trust_labels() print(f"\n[Phase 1 Result] {phase1_result}") # Phase 2 phase2_result = self.phase_2_build_loading_order() print(f"[Phase 2 Result] {phase2_result}") # Summary self.results["summary"] = { "phase_1_status": phase1_result.get("status"), "phase_2_status": phase2_result.get("status"), "total_trust_labels": self.results["phase_1"].get("total_documents", 0), "loading_phases_defined": self.results["phase_2"].get("total_phases", 0), "next_phase": "Phase 3: Dependency Graph", "target_completion": "2026-08-15", "error_rate_target": "50% reduction from baseline" } print("\n[Summary]") print(f" Phase 1 (Trust Labels): {self.results['phase_1'].get('total_documents', 0)} docs labeled") print(f" Phase 2 (Load Order): {self.results['phase_2'].get('total_phases', 0)} phases defined") print(f" Tiers: {', '.join(self.results['phase_1'].get('tier_distribution', {}).keys())}") return self.results def save_report(self, output_file: str = None) -> None: """리포트 저장""" if not output_file: output_file = f"Temp/llm_radar_phase12_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json" Path(output_file).parent.mkdir(parents=True, exist_ok=True) with open(output_file, 'w', encoding='utf-8') as f: json.dump(self.results, f, indent=2, ensure_ascii=False) print(f"\n[Save] Report saved: {output_file}") if __name__ == "__main__": radar = LLMRadarPhase12() radar.generate_report() radar.save_report() # 의사 코드 출력 print("\n" + "="*80) print("Pseudo Code: LLM Context Builder with Trust Tiers") print("="*80) print(radar.generate_llm_context_builder_pseudo_code())