Upload emergence_engine.py

emergence_engine.py

@@ -0,0 +1,1081 @@
+"""
+emergence_engine.py — MAIA 창발성 발현 엔진 v1.0
+특허 제7호: 다차원 가능성 공간 탐색 및 창발적 조합 생성 엔진
+
+[독립항 1] 5단계 창발적 조합 생성 방법:
+  (a) 브루트포스 전수 탐색 + 스코어링
+  (b) 볼츠만 분포 기반 확률적 샘플링 (온도 파라미터 T)
+  (c) 지식그래프 연결 탐색 (미발견 경로)
+  (d) 교차 융합 보너스 매트릭스
+  (e) 창발성 임계점 판정
+
+[종속항 3] 엔트로피 기반 다양성 보장
+[종속항 4] 다중 도메인 적용
+
+(주) 비드래프트 / AETHER Proto-AGI
+"""
+
+import math
+import logging
+import hashlib
+import time
+from typing import List, Dict, Any, Tuple, Optional, Set
+from dataclasses import dataclass, field
+from itertools import combinations, product
+from collections import Counter, defaultdict
+
+import numpy as np
+
+# ── 상수 ──
+DEFAULT_TEMPERATURE = 1.0       # 볼츠만 분포 기본 온도
+MIN_EMERGENCE_THRESHOLD = 0.65  # 창발성 임계점 기본값
+MIN_ENTROPY = 1.5               # 최소 엔트로피 (비트)
+MAX_BRUTE_FORCE_COMBOS = 50000  # 전수 탐색 최대 조합 수
+DEFAULT_SAMPLE_SIZE = 100       # 기본 샘플링 크기
+
+LAYER_ORDER = ["INPUT", "TRANSFORMATION", "CONTROL", "FABRICATION", "CONTEXT", "VALUE"]
+
+logger = logging.getLogger(__name__)
+
+
+# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+# 1. 데이터 구조
+# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+
+@dataclass
+class IdeaVector:
+    """가능성 공간 내 아이디어 벡터"""
+    item_name: str
+    category: str
+    layer: str
+    layer_index: int
+    position: np.ndarray        # N차원 좌표
+    relevance_score: float = 0.0
+    metadata: Dict[str, Any] = field(default_factory=dict)
+
+    @property
+    def id(self) -> str:
+        return hashlib.md5(f"{self.layer}:{self.category}:{self.item_name}".encode()).hexdigest()[:8]
+
+
+@dataclass
+class Combination:
+    """아이디어 조합 (2개 이상 아이디어의 융합)"""
+    items: List[IdeaVector]
+    score: float = 0.0
+    fusion_bonus: float = 0.0
+    complexity: float = 0.0
+    is_emergent: bool = False
+    emergence_level: str = "LOW"
+    novelty_score: float = 0.0
+    graph_paths: List[List[str]] = field(default_factory=list)
+    metadata: Dict[str, Any] = field(default_factory=dict)
+
+    @property
+    def layers_involved(self) -> Set[str]:
+        return set(item.layer for item in self.items)
+
+    @property
+    def n_layers(self) -> int:
+        return len(self.layers_involved)
+
+    @property
+    def description(self) -> str:
+        parts = [f"[{item.layer}] {item.category} > {item.item_name}" for item in self.items]
+        return " ⊕ ".join(parts)
+
+
+@dataclass
+class EmergenceResult:
+    """창발성 판정 결과"""
+    complexity: float
+    threshold: float
+    is_emergent: bool
+    emergence_level: str  # HIGH, MODERATE, LOW
+    confidence: float     # 0.0 ~ 1.0
+    contributing_factors: Dict[str, float] = field(default_factory=dict)
+
+
+# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+# 2. [독립항 전제] N차원 가능성 공간 (Possibility Space)
+# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+
+class PossibilitySpace:
+    """N차원(N≥5) 가능성 공간
+    
+    각 차원은 기술 레이어(INPUT, TRANSFORMATION, CONTROL, FABRICATION, CONTEXT, VALUE)에 대응.
+    아이디어는 N차원 벡터로 공간에 배치되며, 레이어 소속 정도를 연속 좌표로 표현.
+    """
+
+    def __init__(self, dimensions: List[str] = None):
+        self.dimensions = dimensions or LAYER_ORDER
+        self.n_dims = len(self.dimensions)
+        assert self.n_dims >= 5, f"가능성 공간은 최소 5차원이어야 합니다 (현재: {self.n_dims})"
+        self.ideas: List[IdeaVector] = []
+        self._dim_index = {d: i for i, d in enumerate(self.dimensions)}
+        logger.info(f"✅ 가능성 공간 초기화: {self.n_dims}차원 {self.dimensions}")
+
+    def place_idea(self, item_name: str, category: str, layer: str,
+                   relevance_score: float = 0.5, cross_layer_affinity: Dict[str, float] = None) -> IdeaVector:
+        """아이디어를 N차원 공간에 배치
+        
+        주(primary) 레이어에 높은 좌표, 교차 친화도에 따라 다른 차원에도 분산 배치.
+        이를 통해 "순수 INPUT" 아이디어와 "INPUT+CONTROL 융합" 아이디어의 위치가 구별됨.
+        """
+        position = np.zeros(self.n_dims)
+        primary_idx = self._dim_index.get(layer, 0)
+        position[primary_idx] = relevance_score
+
+        # 교차 레이어 친화도 반영 (부(secondary) 차원에 분산)
+        if cross_layer_affinity:
+            for other_layer, affinity in cross_layer_affinity.items():
+                if other_layer in self._dim_index:
+                    idx = self._dim_index[other_layer]
+                    position[idx] = max(position[idx], affinity * 0.5)
+
+        # L2 정규화 (단위 초구면에 배치)
+        norm = np.linalg.norm(position)
+        if norm > 0:
+            position = position / norm
+
+        idea = IdeaVector(
+            item_name=item_name, category=category, layer=layer,
+            layer_index=primary_idx, position=position,
+            relevance_score=relevance_score
+        )
+        self.ideas.append(idea)
+        return idea
+
+    def place_from_categories(self, categories, relevance_scores: Dict):
+        """categories.py의 카테고리/관련성 스코어로부터 대량 배치"""
+        count = 0
+        for cat in categories:
+            cat_name = cat.name_ko if hasattr(cat, 'name_ko') else str(cat)
+            layer = getattr(cat, 'layer', 'TRANSFORMATION')
+            cat_data = relevance_scores.get(cat_name, {})
+            cat_score = cat_data.get("category_score", 0.5)
+            item_scores = cat_data.get("items", {})
+
+            items = cat.items if hasattr(cat, 'items') and cat.items else []
+            for item in items:
+                item_key = item.get('name', str(item)) if isinstance(item, dict) else str(item)
+                item_score = item_scores.get(item_key, 0.5)
+                combined = (cat_score + item_score) / 2
+                self.place_idea(item_key, cat_name, layer, relevance_score=combined)
+                count += 1
+
+        logger.info(f"✅ 가능성 공간 배치 완료: {count}개 아이디어 → {self.n_dims}차원 공간")
+        return count
+
+    def distance(self, idea_a: IdeaVector, idea_b: IdeaVector) -> float:
+        """두 아이디어 간 유클리드 거리 (멀수록 이종 결합)"""
+        return float(np.linalg.norm(idea_a.position - idea_b.position))
+
+    def cosine_similarity(self, idea_a: IdeaVector, idea_b: IdeaVector) -> float:
+        """두 아이디어 간 코사인 유사도"""
+        dot = np.dot(idea_a.position, idea_b.position)
+        norm_a = np.linalg.norm(idea_a.position)
+        norm_b = np.linalg.norm(idea_b.position)
+        if norm_a == 0 or norm_b == 0:
+            return 0.0
+        return float(dot / (norm_a * norm_b))
+
+    def get_ideas_by_layer(self, layer: str) -> List[IdeaVector]:
+        return [idea for idea in self.ideas if idea.layer == layer]
+
+    @property
+    def total_ideas(self) -> int:
+        return len(self.ideas)
+
+
+# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+# 3. [독립항 (a)] 브루트포스 전수 탐색 + 스코어링
+# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+
+class BruteForceExplorer:
+    """가능성 공간의 전수 탐색 및 스코어링 엔진
+    
+    모든 2-way (또는 k-way) 조합을 열거하고 각 조합의 융합 점수를 산출.
+    대규모 공간의 경우 레이어별 대표 아이디어를 선별 후 탐색.
+    """
+
+    def __init__(self, space: PossibilitySpace, fusion_matrix: 'FusionBonusMatrix'):
+        self.space = space
+        self.fusion_matrix = fusion_matrix
+
+    def enumerate_all_combinations(self, k: int = 2,
+                                    max_combos: int = MAX_BRUTE_FORCE_COMBOS,
+                                    cross_layer_only: bool = True) -> List[Combination]:
+        """전수 탐색: k-way 조합 열거 및 스코어링
+        
+        Args:
+            k: 조합 크기 (기본 2-way)
+            max_combos: 최대 조합 수 (메모리/시간 제한)
+            cross_layer_only: True면 이종 레이어 조합만 탐색 (동일 레이어 내 조합 제외)
+        
+        Returns:
+            점수 내림차순 정렬된 조합 리스트
+        """
+        ideas = self.space.ideas
+        if not ideas:
+            logger.warning("가능성 공간에 아이디어가 없습니다.")
+            return []
+
+        # 대규모 공간: 레이어별 상위 아이디어만 선별
+        if len(list(combinations(range(len(ideas)), k))) > max_combos:
+            ideas = self._select_representative_ideas(ideas, per_layer=max(5, max_combos // 100))
+            logger.info(f"대규모 공간 → 대표 아이디어 {len(ideas)}개로 축소")
+
+        scored = []
+        combo_count = 0
+        for combo_tuple in combinations(ideas, k):
+            if combo_count >= max_combos:
+                break
+            combo_items = list(combo_tuple)
+
+            # 이종 레이어 필터
+            if cross_layer_only:
+                layers = set(item.layer for item in combo_items)
+                if len(layers) < 2:
+                    continue
+
+            score = self._score_combination(combo_items)
+            fusion_bonus = self.fusion_matrix.compute_bonus(combo_items)
+
+            combo = Combination(
+                items=combo_items,
+                score=score,
+                fusion_bonus=fusion_bonus,
+            )
+            scored.append(combo)
+            combo_count += 1
+
+        # 점수 내림차순 정렬
+        scored.sort(key=lambda c: c.score + c.fusion_bonus, reverse=True)
+        logger.info(f"✅ 전수 탐색 완료: {combo_count}개 조합 스코어링")
+        return scored
+
+    def _score_combination(self, items: List[IdeaVector]) -> float:
+        """조합의 기본 점수 = 관련성 평균 + 공간적 거리 보너스"""
+        relevance_avg = np.mean([item.relevance_score for item in items])
+
+        # 공간적 거리: 멀수록 (이종 융합일수록) 보너스
+        if len(items) >= 2:
+            distances = [self.space.distance(a, b) for a, b in combinations(items, 2)]
+            distance_bonus = np.mean(distances) * 0.3  # 거리의 30%를 보너스로
+        else:
+            distance_bonus = 0.0
+
+        return float(relevance_avg + distance_bonus)
+
+    def _select_representative_ideas(self, ideas: List[IdeaVector], per_layer: int = 10) -> List[IdeaVector]:
+        """각 레이어에서 관련성 상위 아이디어만 선별"""
+        by_layer = defaultdict(list)
+        for idea in ideas:
+            by_layer[idea.layer].append(idea)
+        selected = []
+        for layer in LAYER_ORDER:
+            layer_ideas = sorted(by_layer.get(layer, []), key=lambda x: x.relevance_score, reverse=True)
+            selected.extend(layer_ideas[:per_layer])
+        return selected
+
+
+# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+# 4. [독립항 (b)] 볼츠만 분포 기반 확률적 샘플링
+# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+
+class BoltzmannSampler:
+    """볼츠만 분포 P(x) = exp(-E(x)/kT) / Z 기반 확률적 샘플링
+    
+    온도 파라미터 T로 탐색 다양성을 동적 조절:
+      - 높은 T (>2.0): 탐색(exploration) 모드 — 저점수 아이디어도 높은 확률로 선택
+      - 중간 T (~1.0): 균형 모드
+      - 낮은 T (<0.5): 착취(exploitation) 모드 — 고점수 아이디어에 집중
+    """
+
+    def __init__(self, temperature: float = DEFAULT_TEMPERATURE):
+        self.temperature = temperature
+        self._validate_temperature()
+
+    def _validate_temperature(self):
+        if self.temperature <= 0:
+            raise ValueError(f"온도 T는 양수여야 합니다 (현재: {self.temperature})")
+
+    def set_temperature(self, temperature: float):
+        """온도 파라미터 동적 조절"""
+        self.temperature = temperature
+        self._validate_temperature()
+        logger.info(f"🌡️ 볼츠만 온도 변경: T={self.temperature:.2f}")
+
+    def compute_probabilities(self, combinations: List[Combination]) -> np.ndarray:
+        """각 조합의 볼츠만 확률 P(x) 계산
+        
+        P(x) = exp(-E(x) / T) / Z
+        여기서 E(x) = -score (높은 점수 = 낮은 에너지)
+              Z = Σ exp(-E(x) / T)  (분배함수)
+        """
+        if not combinations:
+            return np.array([])
+
+        scores = np.array([c.score + c.fusion_bonus for c in combinations])
+        energies = -scores  # 높은 점수 = 낮은 에너지
+
+        # 로그-확률 계산 (수치 안정성)
+        log_probs = -energies / self.temperature
+        log_probs -= np.max(log_probs)  # 오버플로 방지
+        probs = np.exp(log_probs)
+
+        # 분배함수 Z로 정규화
+        Z = probs.sum()
+        if Z > 0:
+            probs /= Z
+        else:
+            probs = np.ones(len(probs)) / len(probs)
+
+        return probs
+
+    def sample(self, combinations: List[Combination], n_samples: int = DEFAULT_SAMPLE_SIZE,
+               replace: bool = False) -> List[Combination]:
+        """볼츠만 분포에 따른 확률적 샘플링
+        
+        Args:
+            combinations: 전수 탐색 결과
+            n_samples: 샘플링할 조합 수
+            replace: True면 복원 추출 (동일 조합 중복 선택 가능)
+        
+        Returns:
+            볼츠만 확률에 따라 선택된 조합 리스트
+        """
+        if not combinations:
+            return []
+
+        n_samples = min(n_samples, len(combinations)) if not replace else n_samples
+        probs = self.compute_probabilities(combinations)
+
+        indices = np.random.choice(
+            len(combinations), size=n_samples, replace=replace, p=probs
+        )
+
+        sampled = [combinations[i] for i in indices]
+        logger.info(
+            f"✅ 볼츠만 샘플링 완료: T={self.temperature:.2f}, "
+            f"{len(combinations)}개 중 {n_samples}개 선택"
+        )
+        return sampled
+
+    def adaptive_sample(self, combinations: List[Combination],
+                        n_samples: int = DEFAULT_SAMPLE_SIZE,
+                        explore_ratio: float = 0.3) -> List[Combination]:
+        """적응적 샘플링: 착취(저온) + 탐색(고온) 혼합
+        
+        전체 샘플의 (1-explore_ratio)는 낮은 온도로, explore_ratio는 높은 온도로 샘플링.
+        이를 통해 고점수 조합을 확보하면서도 의외의 조합을 발견할 기회를 보장.
+        """
+        n_exploit = int(n_samples * (1 - explore_ratio))
+        n_explore = n_samples - n_exploit
+
+        # 착취 (낮은 T)
+        self.set_temperature(self.temperature * 0.3)
+        exploited = self.sample(combinations, n_exploit)
+
+        # 탐색 (높은 T)
+        self.set_temperature(self.temperature * 10.0)  # 원래의 ~3배
+        explored = self.sample(combinations, n_explore)
+
+        # 온도 복원
+        self.set_temperature(DEFAULT_TEMPERATURE)
+
+        result = exploited + explored
+        logger.info(f"✅ 적응적 샘플링: 착취 {n_exploit}개 + 탐색 {n_explore}개")
+        return result
+
+
+# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+# 5. [독립항 (c)] 지식그래프 연결 탐색
+# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+
+class EmergenceKnowledgeGraph:
+    """기술 항목 간 지식그래프 구축 및 미발견 경로 탐색
+    
+    노드: 카테고리, 기술 항목
+    엣지: contains(소속), related(관련성), cross_layer(교차 레이어)
+    미발견 경로: 직접 연결이 없는 노드 간 간접 경로 → 창발적 조합 후보
+    """
+
+    def __init__(self):
+        # 인접 리스트 기반 그래프 (networkx 의존 제거)
+        self.nodes: Dict[str, Dict[str, Any]] = {}
+        self.edges: Dict[str, List[Tuple[str, str, float]]] = defaultdict(list)
+        self._adjacency: Dict[str, Set[str]] = defaultdict(set)
+
+    def add_node(self, node_id: str, **attrs):
+        self.nodes[node_id] = attrs
+
+    def add_edge(self, source: str, target: str, relation: str = "related", weight: float = 1.0):
+        self.edges[source].append((target, relation, weight))
+        self._adjacency[source].add(target)
+        self._adjacency[target].add(source)  # 양방향
+
+    def build_from_space(self, space: PossibilitySpace, cross_layer_bonus: Dict = None):
+        """가능성 공간의 아이디어들로 지식그래프 구축"""
+        # 1. 레이어 → 노드
+        for layer in space.dimensions:
+            self.add_node(f"LAYER:{layer}", type="layer", name=layer)
+
+        # 2. 카테고리 → 노드 + 레이어 엣지
+        categories_seen = set()
+        for idea in space.ideas:
+            cat_id = f"CAT:{idea.category}"
+            if cat_id not in categories_seen:
+                self.add_node(cat_id, type="category", name=idea.category, layer=idea.layer)
+                self.add_edge(f"LAYER:{idea.layer}", cat_id, "contains", 1.0)
+                categories_seen.add(cat_id)
+
+            # 3. 항목 → 노드 + 카테고리 엣지
+            item_id = f"ITEM:{idea.id}"
+            self.add_node(item_id, type="item", name=idea.item_name,
+                         category=idea.category, layer=idea.layer)
+            self.add_edge(cat_id, item_id, "contains", idea.relevance_score)
+
+        # 4. 교차 레이어 엣지 (이종 카테고리 간 연결)
+        if cross_layer_bonus:
+            for (layer_a, layer_b), bonus in cross_layer_bonus.items():
+                self.add_edge(f"LAYER:{layer_a}", f"LAYER:{layer_b}", "cross_layer", bonus)
+
+        # 5. 가능성 공간 내 근접 아이디어 간 연결
+        self._connect_nearby_ideas(space, similarity_threshold=0.3)
+
+        logger.info(
+            f"✅ 지식그래프 구축: {len(self.nodes)}개 노드, "
+            f"{sum(len(v) for v in self.edges.values())}개 엣지"
+        )
+
+    def _connect_nearby_ideas(self, space: PossibilitySpace, similarity_threshold: float = 0.3):
+        """가능성 공간에서 유사한 아이디어 간 엣지 추가"""
+        items = [idea for idea in space.ideas]
+        for i, a in enumerate(items):
+            for j, b in enumerate(items):
+                if j <= i:
+                    continue
+                sim = space.cosine_similarity(a, b)
+                if sim > similarity_threshold:
+                    self.add_edge(f"ITEM:{a.id}", f"ITEM:{b.id}", "similar", sim)
+
+    def discover_hidden_paths(self, max_hops: int = 3, max_paths: int = 500) -> List[Dict]:
+        """기존 노드 간 미발견 경로 탐색 (BFS 기반)
+        
+        직접 연결(엣지)이 없지만 간접 경로가 존재하는 아이템 쌍을 발견.
+        이러한 간접 경로는 창발적 아이디어 융합의 후보가 됨.
+        
+        Returns:
+            [{"source": str, "target": str, "path": List[str], "novelty_score": float}, ...]
+        """
+        item_nodes = [nid for nid, attrs in self.nodes.items() if attrs.get("type") == "item"]
+        hidden_paths = []
+
+        for source in item_nodes:
+            direct_neighbors = self._adjacency.get(source, set())
+            # BFS로 max_hops 이내 도달 가능한 노드 탐색
+            reachable = self._bfs(source, max_hops)
+
+            for target, path in reachable.items():
+                if target == source:
+                    continue
+                if self.nodes.get(target, {}).get("type") != "item":
+                    continue
+                # 직접 연결이 없는 노드만 (미발견 경로)
+                if target not in direct_neighbors:
+                    # 신규도 = 1 / 경로 길이 (짧은 간접 경로가 더 의미 있음)
+                    novelty = 1.0 / len(path)
+                    # 이종 레이어 보너스
+                    source_layer = self.nodes[source].get("layer", "")
+                    target_layer = self.nodes[target].get("layer", "")
+                    if source_layer != target_layer:
+                        novelty *= 1.5  # 이종 레이어 간 미발견 경로 우대
+
+                    hidden_paths.append({
+                        "source": source,
+                        "target": target,
+                        "source_name": self.nodes[source].get("name", source),
+                        "target_name": self.nodes[target].get("name", target),
+                        "source_layer": source_layer,
+                        "target_layer": target_layer,
+                        "path": path,
+                        "hops": len(path) - 1,
+                        "novelty_score": novelty
+                    })
+
+                    if len(hidden_paths) >= max_paths:
+                        break
+            if len(hidden_paths) >= max_paths:
+                break
+
+        hidden_paths.sort(key=lambda x: x["novelty_score"], reverse=True)
+        logger.info(f"✅ 미발견 경로 탐색: {len(hidden_paths)}개 발견")
+        return hidden_paths
+
+    def _bfs(self, start: str, max_depth: int) -> Dict[str, List[str]]:
+        """BFS로 시작 노드에서 max_depth 이내 도달 가능한 모든 노드와 경로 반환"""
+        visited = {start: [start]}
+        queue = [(start, [start], 0)]
+        while queue:
+            current, path, depth = queue.pop(0)
+            if depth >= max_depth:
+                continue
+            for neighbor in self._adjacency.get(current, set()):
+                if neighbor not in visited:
+                    new_path = path + [neighbor]
+                    visited[neighbor] = new_path
+                    queue.append((neighbor, new_path, depth + 1))
+        return visited
+
+    def enrich_combinations(self, combinations: List[Combination]) -> List[Combination]:
+        """조합에 지식그래프 경로 정보를 추가"""
+        for combo in combinations:
+            item_ids = [f"ITEM:{item.id}" for item in combo.items]
+            paths_found = []
+            for a_id, b_id in zip(item_ids, item_ids[1:]):
+                reachable = self._bfs(a_id, 3)
+                if b_id in reachable:
+                    paths_found.append(reachable[b_id])
+            combo.graph_paths = paths_found
+            if paths_found:
+                combo.novelty_score = np.mean([1.0 / len(p) for p in paths_found])
+        return combinations
+
+
+# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+# 6. [독립항 (d)] 교차 융합 보너스 매트릭스
+# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+
+class FusionBonusMatrix:
+    """이종 계층 간 거리에 비례한 교차 융합 보너스 매트릭스
+    
+    수학적 보장: bonus[i][j] = |index_i - index_j| / (N-1) * max_bonus
+    → 먼 레이어 간 조합일수록 높은 보너스 (진정한 이종 융합 우대)
+    """
+
+    def __init__(self, layers: List[str] = None, max_bonus: float = 0.15):
+        self.layers = layers or LAYER_ORDER
+        self.n_layers = len(self.layers)
+        self.max_bonus = max_bonus
+        self._layer_index = {l: i for i, l in enumerate(self.layers)}
+        self.matrix = self._build_matrix()
+
+    def _build_matrix(self) -> np.ndarray:
+        """거리 비례 융합 보너스 매트릭스 구축
+        
+        matrix[i][j] = |i - j| / (N-1) * max_bonus
+        대각선(i==j) = 0 (동일 레이어 보너스 없음)
+        """
+        n = self.n_layers
+        matrix = np.zeros((n, n))
+        for i in range(n):
+            for j in range(n):
+                if i != j:
+                    distance = abs(i - j) / (n - 1)  # 정규화 거리 [0, 1]
+                    matrix[i][j] = distance * self.max_bonus
+        return matrix
+
+    def get_bonus(self, layer_a: str, layer_b: str) -> float:
+        """두 레이어 간 융합 보너스 조회"""
+        idx_a = self._layer_index.get(layer_a, 0)
+        idx_b = self._layer_index.get(layer_b, 0)
+        return float(self.matrix[idx_a][idx_b])
+
+    def compute_bonus(self, items: List[IdeaVector]) -> float:
+        """조합 내 모든 아이템 쌍의 융합 보너스 합산"""
+        total_bonus = 0.0
+        for a, b in combinations(items, 2):
+            total_bonus += self.get_bonus(a.layer, b.layer)
+        return total_bonus
+
+    def to_dict(self) -> Dict:
+        """매트릭스를 딕셔너리로 변환 (직렬화용)"""
+        result = {}
+        for i, layer_a in enumerate(self.layers):
+            for j, layer_b in enumerate(self.layers):
+                if i < j:
+                    result[(layer_a, layer_b)] = float(self.matrix[i][j])
+        return result
+
+    def __repr__(self):
+        header = "     " + " ".join(f"{l[:5]:>6}" for l in self.layers)
+        rows = []
+        for i, layer in enumerate(self.layers):
+            row_vals = " ".join(f"{self.matrix[i][j]:6.3f}" for j in range(self.n_layers))
+            rows.append(f"{layer[:5]:>5} {row_vals}")
+        return header + "\n" + "\n".join(rows)
+
+
+# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+# 7. [독립항 (e)] 창발성 임계점 (Emergence Threshold) 판정
+# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+
+class EmergenceThresholdJudge:
+    """조합 복잡도가 창발성 임계점을 초과하는지 판정
+    
+    복잡도 = f(참여 레이어 수, 교차 융합 보너스, 지식그래프 경로 신규도, 공간적 거리)
+    임계점 초과 시 해당 조합을 "창발 발현(Emergent)" 으로 판정.
+    """
+
+    def __init__(self, threshold: float = MIN_EMERGENCE_THRESHOLD,
+                 fusion_matrix: FusionBonusMatrix = None):
+        self.threshold = threshold
+        self.fusion_matrix = fusion_matrix or FusionBonusMatrix()
+
+    def calculate_complexity(self, combo: Combination, space: PossibilitySpace = None) -> float:
+        """조합의 복잡도 계산
+        
+        complexity = α * layer_diversity + β * fusion_bonus + γ * spatial_distance + δ * novelty
+        
+        α=0.30: 참여 레이어 다양성 (N_layers / N_total_layers)
+        β=0.25: 교차 융합 보너스 정규화
+        γ=0.25: 공간적 거리 (멀수록 이종 융합)
+        δ=0.20: 지식그래프 경로 신규도
+        """
+        # (1) 레이어 다양성
+        n_layers = combo.n_layers
+        layer_diversity = n_layers / len(LAYER_ORDER)
+
+        # (2) 교차 융합 보너스 (정규화)
+        max_possible_bonus = self.fusion_matrix.max_bonus * (n_layers * (n_layers - 1) / 2)
+        fusion_normalized = combo.fusion_bonus / max(max_possible_bonus, 0.01)
+
+        # (3) 공간적 거리 (평균)
+        if space and len(combo.items) >= 2:
+            distances = [space.distance(a, b) for a, b in combinations(combo.items, 2)]
+            spatial_distance = np.mean(distances)
+        else:
+            spatial_distance = 0.5
+
+        # (4) 지식그래프 경로 신규도
+        novelty = combo.novelty_score if combo.novelty_score > 0 else 0.3
+
+        # 가중 합산
+        complexity = (
+            0.30 * layer_diversity +
+            0.25 * min(fusion_normalized, 1.0) +
+            0.25 * min(spatial_distance, 1.0) +
+            0.20 * min(novelty, 1.0)
+        )
+
+        return float(complexity)
+
+    def judge(self, combo: Combination, space: PossibilitySpace = None) -> EmergenceResult:
+        """창발성 임계점 초과 여부 판정"""
+        complexity = self.calculate_complexity(combo, space)
+        is_emergent = complexity > self.threshold
+
+        if complexity > self.threshold * 1.5:
+            level = "HIGH"
+        elif is_emergent:
+            level = "MODERATE"
+        else:
+            level = "LOW"
+
+        confidence = min(complexity / self.threshold, 2.0) / 2.0
+
+        result = EmergenceResult(
+            complexity=complexity,
+            threshold=self.threshold,
+            is_emergent=is_emergent,
+            emergence_level=level,
+            confidence=confidence,
+            contributing_factors={
+                "layer_diversity": combo.n_layers / len(LAYER_ORDER),
+                "fusion_bonus": combo.fusion_bonus,
+                "novelty_score": combo.novelty_score,
+                "n_layers": combo.n_layers,
+            }
+        )
+
+        # 조합에도 결과 반영
+        combo.complexity = complexity
+        combo.is_emergent = is_emergent
+        combo.emergence_level = level
+
+        return result
+
+    def judge_batch(self, combinations: List[Combination],
+                    space: PossibilitySpace = None) -> List[Tuple[Combination, EmergenceResult]]:
+        """대량 판정"""
+        results = []
+        emergent_count = 0
+        for combo in combinations:
+            result = self.judge(combo, space)
+            results.append((combo, result))
+            if result.is_emergent:
+                emergent_count += 1
+
+        logger.info(
+            f"✅ 창발성 판정 완료: {len(combinations)}개 중 {emergent_count}개 창발 발현 "
+            f"(임계점: {self.threshold:.2f})"
+        )
+        return results
+
+
+# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+# 8. [종속항 3] 엔트로피 기반 다양성 보장
+# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+
+class EntropyDiversityGuard:
+    """탐색 경로의 다양성을 엔트로피로 측정하고 최소 수준을 보장
+    
+    Shannon 엔트로피 H = -Σ p_i * log2(p_i)
+    최대 엔트로피 = log2(N_layers) ≈ 2.58 (6레이어)
+    최소 엔트로피 기본값 = 1.5 비트 (적어도 3개 레이어에 분산)
+    """
+
+    def __init__(self, min_entropy: float = MIN_ENTROPY, layers: List[str] = None):
+        self.min_entropy = min_entropy
+        self.layers = layers or LAYER_ORDER
+        self.max_entropy = math.log2(len(self.layers))
+
+    def measure_entropy(self, combinations: List[Combination]) -> float:
+        """현재 조합 집합의 레이어 분포 엔트로피 측정"""
+        if not combinations:
+            return 0.0
+
+        layer_counts = Counter()
+        for combo in combinations:
+            for item in combo.items:
+                layer_counts[item.layer] += 1
+
+        total = sum(layer_counts.values())
+        if total == 0:
+            return 0.0
+
+        probs = [layer_counts.get(l, 0) / total for l in self.layers]
+        # Shannon 엔트로피 (base-2, 비트 단위)
+        entropy_val = 0.0
+        for p in probs:
+            if p > 0:
+                entropy_val -= p * math.log2(p)
+
+        return entropy_val
+
+    def is_diverse_enough(self, combinations: List[Combination]) -> bool:
+        return self.measure_entropy(combinations) >= self.min_entropy
+
+    def ensure_diversity(self, combinations: List[Combination],
+                         all_combinations: List[Combination]) -> Tuple[List[Combination], float]:
+        """최소 엔트로피 미달 시 저빈도 레이어 아이디어를 추가 교체
+        
+        Returns:
+            (다양성 보장된 조합 리스트, 최종 엔트로피)
+        """
+        current = list(combinations)
+        current_entropy = self.measure_entropy(current)
+
+        if current_entropy >= self.min_entropy:
+            return current, current_entropy
+
+        logger.info(
+            f"⚠️ 엔트로피 부족: {current_entropy:.2f} < {self.min_entropy:.2f}, 다양성 보강 시작"
+        )
+
+        # 저빈도 레이어 식별
+        layer_counts = Counter()
+        for combo in current:
+            for item in combo.items:
+                layer_counts[item.layer] += 1
+        underrepresented = [l for l in self.layers if layer_counts.get(l, 0) < 2]
+
+        # 저빈도 레이어의 조합을 all_combinations에서 추가
+        additions = []
+        for combo in all_combinations:
+            if combo in current:
+                continue
+            if any(item.layer in underrepresented for item in combo.items):
+                additions.append(combo)
+            if len(additions) >= len(underrepresented) * 3:
+                break
+
+        # 현재 리스트의 가장 낮은 점수 조합을 교체
+        current.sort(key=lambda c: c.score + c.fusion_bonus)
+        n_replace = min(len(additions), len(current) // 4)  # 최대 25% 교체
+        if n_replace > 0:
+            current = current[n_replace:] + additions[:n_replace]
+
+        final_entropy = self.measure_entropy(current)
+        logger.info(f"✅ 다양성 보강 완료: 엔트로피 {current_entropy:.2f} → {final_entropy:.2f}")
+        return current, final_entropy
+
+    def get_diversity_report(self, combinations: List[Combination]) -> Dict:
+        """다양성 보고서 생성"""
+        entropy_val = self.measure_entropy(combinations)
+        layer_counts = Counter()
+        for combo in combinations:
+            for item in combo.items:
+                layer_counts[item.layer] += 1
+
+        return {
+            "entropy": entropy_val,
+            "max_entropy": self.max_entropy,
+            "min_entropy": self.min_entropy,
+            "diversity_ratio": entropy_val / self.max_entropy if self.max_entropy > 0 else 0,
+            "is_diverse": entropy_val >= self.min_entropy,
+            "layer_distribution": dict(layer_counts),
+            "underrepresented_layers": [l for l in self.layers if layer_counts.get(l, 0) < 2],
+        }
+
+
+# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+# 9. 통합 엔진: MAIAEmergenceEngine
+# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+
+class MAIAEmergenceEngine:
+    """MAIA 창발성 발현 엔진 — 특허 제7호 독립항 (a)~(e) 통합 실행
+    
+    파이프라인:
+    1. 가능성 공간 구축 (N≥5 차원)
+    2. (a) 브루트포스 전수 탐색 + 스코어링
+    3. (b) 볼츠만 분포 확률적 샘플링
+    4. (c) 지식그래프 미발견 경로 탐색 + 조합 보강
+    5. (d) 교차 융합 보너스 매트릭스 적용
+    6. (e) 창발성 임계점 판정
+    7. [종속항3] 엔트로피 기반 다양성 보장
+    """
+
+    def __init__(self,
+                 temperature: float = DEFAULT_TEMPERATURE,
+                 emergence_threshold: float = MIN_EMERGENCE_THRESHOLD,
+                 min_entropy: float = MIN_ENTROPY,
+                 max_bonus: float = 0.15):
+        # 모듈 초기화
+        self.space = PossibilitySpace(LAYER_ORDER)
+        self.fusion_matrix = FusionBonusMatrix(LAYER_ORDER, max_bonus)
+        self.explorer = BruteForceExplorer(self.space, self.fusion_matrix)
+        self.sampler = BoltzmannSampler(temperature)
+        self.knowledge_graph = EmergenceKnowledgeGraph()
+        self.judge = EmergenceThresholdJudge(emergence_threshold, self.fusion_matrix)
+        self.diversity_guard = EntropyDiversityGuard(min_entropy, LAYER_ORDER)
+
+        # 실행 결과 저장
+        self.results: Dict[str, Any] = {}
+        self._initialized = False
+
+        logger.info("🚀 MAIA 창발성 발현 엔진 초기화 완료")
+
+    def initialize(self, categories, relevance_scores: Dict,
+                   cross_layer_bonus: Dict = None):
+        """엔진 초기화: 가능성 공간 구축 + 지식그래프 빌드"""
+        # 1. 가능성 공간에 아이디어 배치
+        n_placed = self.space.place_from_categories(categories, relevance_scores)
+
+        # 2. 지식그래프 구축
+        self.knowledge_graph.build_from_space(self.space, cross_layer_bonus)
+
+        self._initialized = True
+        self.results["initialization"] = {
+            "n_ideas": n_placed,
+            "n_dimensions": self.space.n_dims,
+            "n_graph_nodes": len(self.knowledge_graph.nodes),
+        }
+        return n_placed
+
+    def run(self, n_samples: int = DEFAULT_SAMPLE_SIZE,
+            k_way: int = 2,
+            max_brute_force: int = MAX_BRUTE_FORCE_COMBOS,
+            explore_ratio: float = 0.3) -> List[Combination]:
+        """전체 파이프라인 실행
+        
+        Args:
+            n_samples: 최종 출력 조합 수
+            k_way: 조합 크기 (2-way, 3-way 등)
+            max_brute_force: 전수 탐색 최대 조합 수
+            explore_ratio: 볼츠만 탐색 비율
+        
+        Returns:
+            창발성 판정 완료된 최종 조합 리스트 (점수 내림차순)
+        """
+        assert self._initialized, "먼저 initialize()를 호출하세요."
+        start_time = time.time()
+
+        # Step (a): 브루트포스 전수 탐색
+        logger.info("━━ Step (a): 브루트포스 전수 탐색 ━━")
+        all_combinations = self.explorer.enumerate_all_combinations(
+            k=k_way, max_combos=max_brute_force, cross_layer_only=True
+        )
+        self.results["brute_force"] = {"total_combinations": len(all_combinations)}
+
+        if not all_combinations:
+            logger.warning("전수 탐색 결과 없음")
+            return []
+
+        # Step (d): 교차 융합 보너스 적용 (스코어에 이미 반영됨)
+        logger.info("━━ Step (d): 교차 융합 보너스 매트릭스 적용 ━━")
+        # (이미 enumerate_all_combinations 내부에서 fusion_bonus 계산됨)
+
+        # Step (b): 볼츠만 분포 확률적 샘플링
+        logger.info("━━ Step (b): 볼츠만 분포 확률적 샘플링 ━━")
+        sampled = self.sampler.adaptive_sample(
+            all_combinations, n_samples=n_samples * 2, explore_ratio=explore_ratio
+        )
+        self.results["boltzmann_sampling"] = {
+            "temperature": self.sampler.temperature,
+            "input_count": len(all_combinations),
+            "sampled_count": len(sampled),
+        }
+
+        # Step (c): 지식그래프 미발견 경로로 조합 보강
+        logger.info("━━ Step (c): 지식그래프 미발견 경로 탐색 ━━")
+        hidden_paths = self.knowledge_graph.discover_hidden_paths(max_hops=3, max_paths=200)
+        sampled = self.knowledge_graph.enrich_combinations(sampled)
+        self.results["knowledge_graph"] = {"hidden_paths_found": len(hidden_paths)}
+
+        # [종속항3] 엔트로피 기반 다양성 보장
+        logger.info("━━ 엔트로피 기반 다양성 보장 ━━")
+        sampled, final_entropy = self.diversity_guard.ensure_diversity(sampled, all_combinations)
+        self.results["entropy"] = self.diversity_guard.get_diversity_report(sampled)
+
+        # Step (e): 창발성 임계점 판정
+        logger.info("━━ Step (e): 창발성 임계점 판정 ━━")
+        judged = self.judge.judge_batch(sampled, self.space)
+
+        # 창발 발현 조합만 필터 + 점수순 정렬
+        emergent_combos = [
+            combo for combo, result in judged if result.is_emergent
+        ]
+        emergent_combos.sort(
+            key=lambda c: c.complexity * (c.score + c.fusion_bonus), reverse=True
+        )
+
+        # 최종 n_samples개 선택
+        final = emergent_combos[:n_samples]
+
+        elapsed = time.time() - start_time
+        self.results["summary"] = {
+            "total_time_sec": round(elapsed, 2),
+            "input_ideas": self.space.total_ideas,
+            "brute_force_combos": len(all_combinations),
+            "boltzmann_sampled": len(sampled),
+            "emergent_combos": len(emergent_combos),
+            "final_output": len(final),
+            "emergence_rate": f"{len(emergent_combos) / max(len(sampled), 1) * 100:.1f}%",
+        }
+
+        logger.info(
+            f"🎯 MAIA 창발성 엔진 완료: {self.space.total_ideas}개 아이디어 → "
+            f"{len(all_combinations)}개 조합 → {len(emergent_combos)}개 창발 → "
+            f"최종 {len(final)}개 (소요시간: {elapsed:.1f}초)"
+        )
+
+        return final
+
+    def get_results_summary(self) -> Dict:
+        """실행 결과 요약"""
+        return self.results
+
+    def format_for_llm_prompt(self, combinations: List[Combination], top_n: int = 30) -> str:
+        """LLM 프롬프트용 창발적 조합 텍스트 포맷팅
+        
+        기존 app.py의 researcher_comprehensive_chunked()에 전달할 창발 후보 텍스트
+        """
+        lines = [f"■■■ MAIA 창발성 엔진 출력 (상위 {min(top_n, len(combinations))}개) ■■■\n"]
+        for i, combo in enumerate(combinations[:top_n], 1):
+            layers = ", ".join(sorted(combo.layers_involved))
+            lines.append(f"[창발 #{i:02d}] 복잡도={combo.complexity:.3f} | 레이어: {layers}")
+            lines.append(f"  조합: {combo.description}")
+            lines.append(f"  융합 보너스: {combo.fusion_bonus:.3f} | 창발 수준: {combo.emergence_level}")
+            if combo.graph_paths:
+                lines.append(f"  지식그래프 경로: {len(combo.graph_paths)}개 미발견 경로")
+            lines.append("")
+        return "\n".join(lines)
+
+
+# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+# 10. 편의 함수 (기존 코드 연동용)
+# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+
+def run_emergence_engine(categories, relevance_scores: Dict,
+                         cross_layer_bonus: Dict = None,
+                         temperature: float = 1.0,
+                         threshold: float = 0.65,
+                         n_output: int = 50) -> Tuple[List[Combination], Dict]:
+    """원스텝 실행 함수 (기존 app.py에서 호출용)
+    
+    Usage:
+        from emergence_engine import run_emergence_engine
+        combos, report = run_emergence_engine(categories, relevance_scores, 
+                                              CROSS_LAYER_EMERGENCE_BONUS)
+        prompt_text = engine.format_for_llm_prompt(combos)
+    """
+    engine = MAIAEmergenceEngine(
+        temperature=temperature,
+        emergence_threshold=threshold,
+    )
+    engine.initialize(categories, relevance_scores, cross_layer_bonus)
+    results = engine.run(n_samples=n_output)
+    return results, engine.get_results_summary()
+
+
+# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+# 11. 셀프테스트
+# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+
+def _self_test():
+    """엔진 셀프테스트 (categories.py 없이 독립 실행)"""
+    logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
+
+    # 테스트용 더미 데이터
+    from dataclasses import dataclass as dc
+
+    @dc
+    class DummyCat:
+        name_ko: str; name_en: str; items: list; layer: str; layer_description: str = ""
+
+    test_categories = [
+        DummyCat("센서 기능", "Sensor", ["압력센서", "온도��서", "가속도센서", "라이다", "초음파센서"], "INPUT"),
+        DummyCat("AI 및 ML", "AI/ML", ["트랜스포머", "RAG", "연합학습", "강화학습", "지식그래프"], "TRANSFORMATION"),
+        DummyCat("시스템 제어", "Control", ["PID제어", "퍼지제어", "강건제어", "적응제어"], "CONTROL"),
+        DummyCat("제조 방식", "Fabrication", ["3D프린팅", "레이저가공", "사출성형", "CNC가공"], "FABRICATION"),
+        DummyCat("환경 상호작용", "Environment", ["대기정화", "수질정화", "탄소포집", "재활용"], "CONTEXT"),
+        DummyCat("에너지 관리", "Energy", ["배터리", "태양전지", "연료전지", "에너지하베스팅", "무선충전"], "VALUE"),
+    ]
+
+    test_scores = {}
+    for cat in test_categories:
+        items_scores = {item: 0.3 + 0.5 * (i / len(cat.items)) for i, item in enumerate(cat.items)}
+        test_scores[cat.name_ko] = {
+            "category_score": 0.6,
+            "items": items_scores,
+            "layer": cat.layer,
+        }
+
+    print("=" * 60)
+    print("MAIA 창발성 발현 엔진 셀프테스트")
+    print("=" * 60)
+
+    engine = MAIAEmergenceEngine(temperature=1.0, emergence_threshold=0.55)
+    n = engine.initialize(test_categories, test_scores)
+    print(f"\n배치 완료: {n}개 아이디어")
+
+    results = engine.run(n_samples=20, k_way=2)
+    print(f"\n최종 창발 조합: {len(results)}개")
+
+    for i, combo in enumerate(results[:5], 1):
+        print(f"\n[#{i}] {combo.description}")
+        print(f"     복잡도={combo.complexity:.3f} | 레이어={combo.n_layers}개 | "
+              f"수준={combo.emergence_level} | 점수={combo.score:.3f}")
+
+    print("\n" + "=" * 60)
+    summary = engine.get_results_summary()
+    for key, val in summary.items():
+        print(f"  {key}: {val}")
+
+    # 융합 보너스 매트릭스 출력
+    print("\n교차 융합 보너스 매트릭스:")
+    print(engine.fusion_matrix)
+
+    # 다양성 보고서
+    report = engine.diversity_guard.get_diversity_report(results)
+    print(f"\n엔트로피: {report['entropy']:.2f} / {report['max_entropy']:.2f} "
+          f"(다양성 {'✅' if report['is_diverse'] else '❌'})")
+
+    print("\n✅ 셀프테스트 완료")
+
+
+if __name__ == "__main__":
+    _self_test()