File size: 5,592 Bytes
d997548 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 | """
Encompassment Engine - K6 as the CEO of knowledge.
K6 is not a counter. K6 is the central index that:
1. Registers every cell's pattern fingerprint
2. Computes centrality of each cell in the knowledge graph
3. Decides when births are warranted (new patterns)
4. Guides retrieval of relevant cells
5. Serves as the "summary dimension" of K1-K5
"""
from collections import defaultdict
from typing import Dict, List, Optional
import math
class EncompassmentEngine:
def __init__(self, max_index_size=10000):
self.pattern_index: Dict[str, List[str]] = defaultdict(list)
self.centrality: Dict[str, float] = {}
self.cell_metadata: Dict[str, dict] = {}
self.birth_history: List[dict] = []
self.max_index_size = max_index_size
self._total_registrations = 0
def register(self, cell_id: str, fp: str, other_K: List[float]):
if not fp:
return
if cell_id in self.cell_metadata:
old_fp = self.cell_metadata[cell_id].get("fp")
if old_fp == fp:
self._recompute_centrality(cell_id)
return
if old_fp and cell_id in self.pattern_index.get(old_fp, []):
self.pattern_index[old_fp].remove(cell_id)
self.pattern_index[fp].append(cell_id)
K1, K2, K3, K4, K5 = other_K[:5]
self.cell_metadata[cell_id] = {
"fp": fp, "K1": K1, "K3": K3, "K5": K5, "K2": K2, "K4": K4,
}
self._recompute_centrality(cell_id)
self._total_registrations += 1
if len(self.pattern_index) > self.max_index_size:
self._prune()
def _recompute_centrality(self, cell_id: str):
meta = self.cell_metadata.get(cell_id)
if not meta:
self.centrality[cell_id] = 0.0
return
fp = meta["fp"]
n_sharing = len(self.pattern_index.get(fp, []))
quality = (meta["K1"] + meta["K3"] + meta["K5"]) / 3.0
spread = min(1.0, math.log1p(n_sharing) / math.log1p(20))
c = 0.6 * quality + 0.4 * spread
self.centrality[cell_id] = max(0.0, min(1.0, c))
def _prune(self):
keep = {}
for fp, members in self.pattern_index.items():
if len(members) >= 2:
keep[fp] = members
else:
for m in members:
if self.centrality.get(m, 0.0) > 0.7:
keep[fp] = members
break
self.pattern_index = defaultdict(list, keep)
def K6(self, cell_id: str) -> float:
return float(self.centrality.get(cell_id, 0.0))
def K6_array(self, cell_ids: List[str]) -> List[float]:
return [self.K6(c) for c in cell_ids]
def should_birth(self, fp: str, hamming_fn, threshold: float = 0.55) -> bool:
if not fp:
return False
if fp in self.pattern_index and len(self.pattern_index[fp]) > 0:
return False
for existing_fp in self.pattern_index:
try:
sim = hamming_fn(fp, existing_fp)
except Exception:
continue
if sim > threshold:
return False
return True
def record_birth(self, cell_id: str, parent_id: str, fp: str, step: int):
self.birth_history.append({
"cell_id": cell_id, "parent_id": parent_id,
"fp": fp, "step": step,
})
def guide_retrieval(self, query_fp: str, hamming_fn, top_k: int = 5) -> List[str]:
if not query_fp:
return []
scores = []
for fp, members in self.pattern_index.items():
try:
sim = hamming_fn(query_fp, fp)
except Exception:
sim = 0.0
if sim < 0.1:
continue
for cid in members:
c = self.centrality.get(cid, 0.0)
scores.append((sim * (0.5 + 0.5 * c), cid))
scores.sort(reverse=True)
return [cid for _, cid in scores[:top_k]]
def summary(self) -> dict:
n_patterns = len(self.pattern_index)
n_cells = len(self.cell_metadata)
if self.centrality:
c_vals = list(self.centrality.values())
c_min, c_max = min(c_vals), max(c_vals)
c_mean = sum(c_vals) / len(c_vals)
else:
c_min = c_max = c_mean = 0.0
return {
"n_patterns": n_patterns,
"n_cells_indexed": n_cells,
"centrality_min": round(c_min, 4),
"centrality_max": round(c_max, 4),
"centrality_mean": round(c_mean, 4),
"births_total": len(self.birth_history),
"total_registrations": self._total_registrations,
}
if __name__ == "__main__":
def fake_hamming(a, b):
if not a or not b:
return 0.0
n = min(len(a), len(b))
return sum(1 for i in range(n) if a[i] == b[i]) / n
e = EncompassmentEngine()
e.register("cell-1", "aaaaaaaaaa", [0.9, 0, 0.2, 0, 0.9])
e.register("cell-2", "bbbbbbbbbb", [0.8, 0, 0.3, 0, 0.85])
e.register("cell-3", "aaaaaaaaaa", [0.95, 0, 0.1, 0, 0.95])
print("K6 cell-1:", e.K6("cell-1"))
print("K6 cell-2:", e.K6("cell-2"))
print("K6 cell-3:", e.K6("cell-3"))
print("should_birth(cccc):", e.should_birth("cccccccccc", fake_hamming, threshold=0.55))
print("should_birth(aaaa):", e.should_birth("aaaaaaaaaa", fake_hamming, threshold=0.55))
print("retrieval:", e.guide_retrieval("aaaaaaaaaa", fake_hamming, top_k=2))
print("summary:", e.summary())
|