File size: 4,435 Bytes
f3072b7 | 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 | """
AbstractionEngine — Vitalis FSI
The system doesn't just remember experiences.
It builds CONCEPTS from clusters of experiences.
This is the difference between memory and understanding.
Example:
50 individual auth patterns →
one abstract AUTH concept vector →
faster retrieval, analogical reasoning, cross-domain transfer
"""
import numpy as np
import os
import json
from vitalis_ide.math_core.kernel import VitalisKernel
from src.hippocampus import Hippocampus
class AbstractionEngine:
CLUSTER_THRESHOLD = 0.55
MIN_CLUSTER_SIZE = 2
def __init__(self):
self.kernel = VitalisKernel()
self.hippocampus = Hippocampus()
self.path = os.path.expanduser("~/.vitalis_workspace/abstractions.json")
self.abstract_path = os.path.expanduser("~/.vitalis_workspace/abstract_vectors.npy")
self._load()
def _load(self):
self.abstractions = {}
self.abstract_vectors = {}
if os.path.exists(self.path):
with open(self.path) as f:
self.abstractions = json.load(f)
if os.path.exists(self.abstract_path):
self.abstract_vectors = np.load(
self.abstract_path, allow_pickle=True).item()
def _save(self):
os.makedirs(os.path.dirname(self.path), exist_ok=True)
with open(self.path, 'w') as f:
json.dump(self.abstractions, f, indent=2)
np.save(self.abstract_path, self.abstract_vectors)
def abstract(self, vectors: dict) -> dict:
"""
Takes a dict of {slot: vector}.
Finds clusters. Builds concept vectors from each cluster.
Returns {concept_name: abstract_vector}.
"""
slots = list(vectors.keys())
visited = set()
concepts = {}
for i, slot_a in enumerate(slots):
if slot_a in visited:
continue
cluster = [slot_a]
vec_a = vectors[slot_a]
for slot_b in slots[i+1:]:
if slot_b in visited:
continue
vec_b = vectors[slot_b]
sim = self.kernel.similarity(vec_a, vec_b)
if sim > self.CLUSTER_THRESHOLD:
cluster.append(slot_b)
visited.add(slot_b)
if len(cluster) >= self.MIN_CLUSTER_SIZE:
# Bundle cluster into abstract concept
bundle = np.zeros(self.kernel.dim, dtype=np.int32)
for s in cluster:
bundle += vectors[s].astype(np.int32)
concept_vec = np.sign(bundle).astype(np.int8)
concept_vec[concept_vec == 0] = 1
concept_name = f"concept_{len(concepts)}"
concepts[concept_name] = {
"vector": concept_vec,
"members": cluster,
"size": len(cluster)
}
visited.add(slot_a)
return concepts
def run_abstraction_cycle(self, pattern_meta: dict) -> list:
"""
Called during dream cycles.
Takes current pattern library and forms higher-order concepts.
"""
vectors = {}
for slot in self.hippocampus.all_slots():
vec = self.hippocampus.recall(slot)
if vec is not None:
vectors[slot] = vec
if len(vectors) < self.MIN_CLUSTER_SIZE:
return []
new_concepts = self.abstract(vectors)
formed = []
for name, data in new_concepts.items():
full_name = f"abstract_{len(self.abstractions)}_{name}"
self.abstractions[full_name] = {
"members": data["members"],
"size": data["size"]
}
self.abstract_vectors[full_name] = data["vector"]
formed.append(full_name)
print(f"[ABSTRACTION] Concept formed: {full_name} ({data['size']} members)")
if formed:
self._save()
return formed
def query_abstractions(self, query_vec: np.ndarray, top_k: int = 3) -> list:
"""Search abstract concept space — faster than raw pattern search."""
results = []
for name, vec in self.abstract_vectors.items():
sim = self.kernel.similarity(query_vec, vec)
results.append((sim, name, self.abstractions.get(name, {})))
results.sort(reverse=True)
return results[:top_k]
|