0

8かける8の画像認識

25
0
$$$$

画像認識

手書き数字の認識をしています。
8かける8のマスで0から9のラベルを判別します。
使うのは10枚の訓練データのみです。
性能は正解率が60パーセントです。

      from collections import defaultdict
import matplotlib.pyplot as plt
import numpy as np
import random
from sklearn.datasets import load_digits
from sklearn.metrics import accuracy_score
from sklearn.preprocessing import StandardScaler
# =========================================================
# 1. 環境とデータの準備(手書き数字 64次元データ)
# =========================================================
np.random.seed(42)
digits = load_digits()
X_raw, y_raw = digits.data, digits.target
# 生態系の変化を見やすくするため、400サンプルを抽出
total_indices = np.random.choice(len(X_raw), 400, replace=False)
X_400, y_400 = X_raw[total_indices], y_raw[total_indices]
scaler = StandardScaler()
X_scaled = scaler.fit_transform(X_400)
# 少数サンプルを訓練用に(残りをテスト環境に)
train_idx = []
for c in range(10):
    c_idxs = np.where(y_400 == c)[0]
    train_idx.extend(c_idxs[:1]) # 各クラス1枚ずつ
test_idx = [i for i in range(400) if i not in train_idx]
X_train, y_train = X_scaled[train_idx], y_400[train_idx]
X_test, y_test = X_scaled[test_idx], y_400[test_idx]
NUM_CLASSES = 10
INPUT_DIM = 64
# --- ハイパーパラメータ ---
INIT_CELLS = 680
MAX_CELLS = 10000
MIN_CELLS = 80
THOUGHT_STEPS = 3
lr_base = 0.30
lr_link_base = 0.10
DIVISION_THRESHOLD = 0.85     # 構造過学習を防ぐため、少し厳格化
DEATH_THRESHOLD = 0.08        # 淘汰圧を少し強める
GRAPH_DECAY = 0.90
EDGE_PRUNE_THRESHOLD = 0.02
epochs =6                   # 現象を観察するため、5サイクル回します
# =========================================================
# 2. 視覚記憶細胞(VisionMemoryCell)の定義
# =========================================================
class VisionMemoryCell:
    def __init__(self, cell_id):
        self.id = cell_id
        self.w_in = np.random.normal(0, 1.0, INPUT_DIM)  # 蜘蛛の巣初期化で再上書きされます
        self.A = np.random.normal(0, 0.1, NUM_CLASSES)
        self.links = {}
        self.visits = 0.0
        self.energy = 0.5
        self.recent_activity = 0.0
def cosine_similarity(w, x):
    return np.dot(w, x) / (np.linalg.norm(w) * np.linalg.norm(x) + 1e-9)
def softmax(x, temp=1.0):
    x = np.array(x) / temp
    x = x - np.max(x)
    e = np.exp(x)
    return e / (np.sum(e) + 1e-12)
# =========================================================
# 🧬 3. 代謝・新陳代謝コア(分裂コストの厳格化)
# =========================================================
def evolve_and_prune_space(memory_space, outer_graph):
    # 睡眠・自然減衰
    for cell in memory_space:
        cell.energy *= 0.92
        cell.recent_activity *= 0.50
        if cell.energy < 0.01: cell.energy = 0.01
        
    # --- 分裂フェーズ(ユーザーさんの提案によるコスト厳格化) ---
    new_cells = []
    current_max_id = max([c.id for c in memory_space]) if memory_space else 0
    for cell in memory_space:
        if len(memory_space) + len(new_cells) >= MAX_CELLS: break
        
        # 分裂条件をエネルギーかつ、 visits累積に設定
        if cell.energy > DIVISION_THRESHOLD and cell.visits > 15:
            child = VisionMemoryCell(current_max_id + len(new_cells) + 1)
            
            # 子は親の特性に少し変異(ノイズ)を乗せる
            child.w_in = np.clip(cell.w_in + np.random.normal(0, 0.05, INPUT_DIM), -3.0, 3.0)
            child.A = cell.A.copy()
            for k, v in cell.links.items(): child.links[k] = v * 0.5
            
            # 【ユーザーさんの設計思想】親の資源を大きく削り、子を未熟な状態からスタートさせる
            cell.energy *= 0.3
            cell.visits *= 0.5
            child.energy = 0.5  # 子の初期エネルギーは低い
            
            new_cells.append(child)
            
    memory_space.extend(new_cells)
    
    # --- 淘汰 & 外側マクログラフの再マッピング ---
    survived_space = []
    id_map = {}
    for cell in memory_space:
        if len(survived_space) > MIN_CELLS:
            if cell.energy < DEATH_THRESHOLD and cell.visits < 1.5:
                continue
        new_id = len(survived_space)
        id_map[cell.id] = new_id
        cell.id = new_id
        survived_space.append(cell)
        
    for cell in survived_space:
        cell.links = {id_map[old_id]: w * 0.85 for old_id, w in cell.links.items() if old_id in id_map and w > 0.05}
        
    new_graph = defaultdict(dict)
    for u, neighbors in outer_graph.items():
        new_u = ("cell", id_map[u[1]]) if u[0] == "cell" and u[1] in id_map else u
        if u[0] == "cell" and u[1] not in id_map: continue
        
        for v, weight in neighbors.items():
            decayed_weight = weight * GRAPH_DECAY
            if decayed_weight < EDGE_PRUNE_THRESHOLD: continue
            
            new_v = ("cell", id_map[v[1]]) if v[0] == "cell" and v[1] in id_map else v
            if v[0] == "cell" and v[1] not in id_map: continue
            
            new_graph[new_u][new_v] = decayed_weight
            
    return survived_space, new_graph
# =========================================================
# 🔄 4. 融合生命ダイナミクス(思考の歩行・側頭抑制)
# =========================================================
def live_vision_cycle(memory_space, outer_graph, x_64d=None, target_class=None, target_recall_class=None,
                      sensory_drive=1.0, intrinsic_drive=0.0, top_down_drive=0.0, learning_gate=1.0):
    if len(memory_space) == 0: return np.zeros(NUM_CLASSES), np.zeros(INPUT_DIM)
    id_to_cell = {c.id: c for c in memory_space}
    
    start_probabilities = np.zeros(len(memory_space))
    
    if x_64d is not None and sensory_drive > 0.0:
        sims = np.array([cosine_similarity(cell.w_in, x_64d) for cell in memory_space])
        start_probabilities += softmax(sims, temp=0.1) * sensory_drive
        
    if intrinsic_drive > 0.0:
        activity_weights = np.array([cell.energy * (cell.recent_activity + 0.05) for cell in memory_space])
        start_probabilities += (activity_weights / (np.sum(activity_weights) + 1e-12)) * intrinsic_drive
        
    if target_recall_class is not None and top_down_drive > 0.0:
        c_node = ("class", target_recall_class)
        for next_node, weight in outer_graph.get(c_node, {}).items():
            if next_node[0] == "cell" and next_node[1] in id_to_cell:
                start_probabilities[next_node[1]] += weight * top_down_drive
                
    if np.sum(start_probabilities) == 0: 
        start_probabilities = np.ones(len(memory_space)) / len(memory_space)
    start_probabilities /= np.sum(start_probabilities)
    
    current_cell_id = np.random.choice(list(id_to_cell.keys()), p=start_probabilities)
    path = [current_cell_id]
    
    collected_signals = [id_to_cell[current_cell_id].A.copy()]
    imagined_pixels = [id_to_cell[current_cell_id].w_in.copy()]
    
    # --- 思考がグラフを歩く(Graph Walk) ---
    for step in range(THOUGHT_STEPS - 1):
        current_cell = id_to_cell[current_cell_id]
        current_cell.recent_activity += sensory_drive * 1.0
        
        all_ids = list(id_to_cell.keys())
        candidates = set(np.random.choice(all_ids, min(25, len(all_ids)), replace=False))
        candidates.update(current_cell.links.keys())
        
        scores = []
        c_ids = []
        for cid in candidates:
            if cid in path or cid not in id_to_cell: continue
            cell = id_to_cell[cid]
            
            score = current_cell.links.get(cid, 0.0) * 2.5
            if x_64d is not None:
                score += 2.0 * cosine_similarity(cell.w_in, x_64d) * sensory_drive
            score += 0.5 * cosine_similarity(current_cell.A, cell.A)
            
            scores.append(score)
            c_ids.append(cid)
            
        if not scores: break
        probs = softmax(scores, temp=0.3)
        current_cell_id = np.random.choice(c_ids, p=probs)
        path.append(current_cell_id)
        collected_signals.append(id_to_cell[current_cell_id].A.copy())
        imagined_pixels.append(id_to_cell[current_cell_id].w_in.copy())
        
    final_perception = np.mean(np.stack(collected_signals), axis=0) if collected_signals else np.zeros(NUM_CLASSES)
    final_image = np.mean(np.stack(imagined_pixels), axis=0) if imagined_pixels else np.zeros(INPUT_DIM)
    
    sensory_factor = sensory_drive * learning_gate
    intrinsic_factor = intrinsic_drive * learning_gate
    
    inferred_class = np.argmax(final_perception)
    target = target_class if target_class is not None else inferred_class
    
    target_vec = np.zeros(NUM_CLASSES)
    target_vec[target] = 1.0
    
    # 報酬の決定(厳格化:専門化を促す)
    reward = min(1.0 / (np.linalg.norm(target_vec - final_perception) + 0.15), 2.5)
    
    # --- ヘブ則的な学習と、専門化のための側頭抑制(Lateral Inhibition) ---
    if x_64d is not None and sensory_factor > 0.0:
        # 1. 経路上の細胞の学習
        for cid in path:
            if cid in id_to_cell:
                cell = id_to_cell[cid]
                cell.visits += sensory_factor
                cell.energy = min(1.0, cell.energy + reward * 0.15 * sensory_factor)
                cell.w_in += (lr_base * 0.5 * sensory_factor) * (x_64d - cell.w_in)
                cell.A = (1.0 - lr_base * sensory_factor) * cell.A + (lr_base * sensory_factor) * target_vec
                
        # 2. 側頭抑制(同じ情報への過剰適合・重複増殖を防ぐ)
        # 歩いた細胞の近隣リンクにありながら、今回選ばれなかった細胞のエネルギーを少し削る
        for cid in path:
            if cid in id_to_cell:
                for neighbor_id in id_to_cell[cid].links.keys():
                    if neighbor_id not in path and neighbor_id in id_to_cell:
                        id_to_cell[neighbor_id].energy *= 0.95 # 側頭抑制によるエネルギー減衰
                
    total_learning_power = (sensory_factor * reward + intrinsic_factor * 0.2)
    if total_learning_power > 0.0:
        for i in range(len(path) - 1):
            u, v = path[i], path[i+1]
            if u in id_to_cell and v in id_to_cell:
                id_to_cell[u].links[v] = id_to_cell[u].links.get(v, 0.1) + (lr_link_base * total_learning_power)
                
        c_node = ("class", int(target))
        first_cell_node = ("cell", path[0])
        last_cell_node = ("cell", path[-1])
        
        for i in range(len(path) - 1):
            f_node = ("cell", path[i])
            t_node = ("cell", path[i+1])
            outer_graph[f_node][t_node] = outer_graph[f_node].get(t_node, 0.0) + total_learning_power
            outer_graph[t_node][f_node] = outer_graph[t_node].get(f_node, 0.0) + total_learning_power
            
        # クラスから細胞、細胞からクラスへの「双方向エッジ」を正確にマッピング
        outer_graph[first_cell_node][c_node] = outer_graph[first_cell_node].get(c_node, 0.0) + total_learning_power
        outer_graph[c_node][first_cell_node] = outer_graph[c_node].get(first_cell_node, 0.0) + total_learning_power
        
        outer_graph[last_cell_node][c_node] = outer_graph[last_cell_node].get(c_node, 0.0) + total_learning_power
        outer_graph[c_node][last_cell_node] = outer_graph[c_node].get(last_cell_node, 0.0) + total_learning_power
        
    return final_perception, final_image
# =========================================================
# 👥 5. アンサンブル・蜘蛛の巣トポロジー管理(ミーム交配の実装)
# =========================================================
class EnsembleVisionEcosystem:
    def __init__(self, n_estimators=5, total_cells=1000):
        self.n_estimators = n_estimators
        self.models = []
        NUM_RINGS = 5  # 蜘蛛の巣の階層数
        
        for _ in range(n_estimators):
            space = []
            graph = defaultdict(dict)
            
            # 密度の調整:外側(感覚層)ほど細胞を多く、内側(概念層)ほど少なく
            cells_per_ring = []
            remaining = total_cells
            for r in range(NUM_RINGS):
                if r == NUM_RINGS - 1:
                    cells_per_ring.append(remaining)
                else:
                    count = int(total_cells * (0.45 / (1.6**r)))
                    cells_per_ring.append(count)
                    remaining -= count
                    
            cell_id_counter = 0
            ring_nodes = {}
            
            # --- 1. 蜘蛛の巣の細胞配置と受容野初期化 ---
            for ring_idx, n_cells in enumerate(cells_per_ring):
                ring_nodes[ring_idx] = []
                for i in range(n_cells):
                    cell = VisionMemoryCell(cell_id_counter)
                    
                    # 外周(感覚層)ほど実際の数字に近い初期受容野を持たせる
                    sensory_proximity = 1.0 - (ring_idx / NUM_RINGS)
                    if sensory_proximity > 0.2:
                        ref_idx = np.random.choice(len(X_train))
                        base_pattern = X_train[ref_idx]
                        cell.w_in = base_pattern * sensory_proximity + np.random.normal(0, 0.3, INPUT_DIM) * (1.0 - sensory_proximity)
                        cell.A[y_train[ref_idx]] = 0.5 * sensory_proximity
                    
                    space.append(cell)
                    ring_nodes[ring_idx].append(cell_id_counter)
                    cell_id_counter += 1
                    
            # --- 2. 蜘蛛の巣の糸(リンク)の構築 ---
            id_to_cell = {c.id: c for c in space}
            for ring_idx in range(NUM_RINGS):
                nodes = ring_nodes[ring_idx]
                num_nodes = len(nodes)
                if num_nodes == 0: continue
                
                for i in range(num_nodes):
                    curr_id = nodes[i]
                    
                    # (a) 同心円の横の糸(横方向リンク)
                    next_id = nodes[(i + 1) % num_nodes]
                    id_to_cell[curr_id].links[next_id] = 0.4
                    id_to_cell[next_id].links[curr_id] = 0.4
                    
                    # (b) 放射方向の縦の糸(抽象化への階層リンク)
                    if ring_idx < NUM_RINGS - 1:
                        inner_nodes = ring_nodes[ring_idx + 1]
                        if len(inner_nodes) > 0:
                            inner_idx = int(i * len(inner_nodes) / num_nodes)
                            inner_id = inner_nodes[inner_idx]
                            
                            id_to_cell[curr_id].links[inner_id] = 0.6
                            id_to_cell[inner_id].links[curr_id] = 0.6
            self.models.append({'space': space, 'graph': graph})
    def train_cycle(self, X, y, phase="day"):
        for model in self.models:
            # 脳の個性を引き出すブートストラップサンプリング
            indices = np.random.choice(len(X), len(X), replace=True)
            for idx in indices:
                if phase == "day":
                    live_vision_cycle(model['space'], model['graph'], x_64d=X[idx], 
                                      target_class=y[idx], sensory_drive=1.0, intrinsic_drive=0.0)
                elif phase == "evening":
                    live_vision_cycle(model['space'], model['graph'], x_64d=X[idx], 
                                      target_class=y[idx], sensory_drive=0.3, intrinsic_drive=0.7)
    def sleep_cycle(self, steps=60):
        for model in self.models:
            for _ in range(steps):
                live_vision_cycle(model['space'], model['graph'], x_64d=None, 
                                  target_class=None, sensory_drive=0.0, intrinsic_drive=1.0)
    def cultural_crossover(self, crossover_rate=0.2):
        """異なる脳の間で、最も洗練されたイデア(ミーム)を交換する"""
        elite_cells_per_model = []
        
        for m_idx, model in enumerate(self.models):
            space = model['space']
            if not space: continue
            
            # 最深部(Ring 4に相当する後半10%のコア層)からエリートを選別
            core_start_idx = int(len(space) * 0.9)
            core_cells = space[core_start_idx:]
            if not core_cells: continue
            
            # エネルギーが高く、キャラクター(特定のクラスへの確信)が立っている細胞をスカウト
            elite_cell = max(core_cells, key=lambda c: (c.energy * np.max(c.A)))
            elite_cells_per_model.append((m_idx, elite_cell))
            
        # 集団間での文化伝播(シナプス感染)
        for src_idx, src_cell in elite_cells_per_model:
            target_indices = [i for i in range(self.n_estimators) if i != src_idx]
            if not target_indices: continue
            dst_idx = random.choice(target_indices)
            dst_model = self.models[dst_idx]
            
            dst_core_cells = dst_model['space'][int(len(dst_model['space']) * 0.9):]
            if not dst_core_cells: continue
            
            # 伝播先の脳の中で、最も迷っている(エネルギーの低い)細胞にミームをブレンドする
            weak_cell = min(dst_core_cells, key=lambda c: c.energy)
            
            # 知識の混ざり合い
            weak_cell.w_in = (1.0 - crossover_rate) * weak_cell.w_in + crossover_rate * src_cell.w_in
            weak_cell.A = (1.0 - crossover_rate) * weak_cell.A + crossover_rate * src_cell.A
            weak_cell.energy = max(weak_cell.energy, src_cell.energy * 0.8) # 文化刺激による活性化
    def evolve(self):
        for model in self.models:
            model['space'], model['graph'] = evolve_and_prune_space(model['space'], model['graph'])
    def predict(self, X, run_evolution=True):
        """
        推論を実行しつつ、その入力データにリアルタイムに適応して
        細胞を更新・分裂・淘汰させる(リアルタイム・オンライン学習モード)
        """
        y_pred = []
        for idx in range(len(X)):
            ensemble_perceptions = []
            
            # 各ベースモデル(脳)がデータに反応し、同時に学習・更新を行う
            for model in self.models:
                # 【変更点】
                # 1. target_class=None にすることで、自分の推論結果をターゲット(疑似ラベル)にする
                # 2. learning_gate=1.0 にして、推論時もヘブ則や側頭抑制を完全に作動させる
                perception, _ = live_vision_cycle(
                    model['space'], model['graph'], x_64d=X[idx], 
                    target_class=None, # 自身の推論(inferred_class)で自己学習
                    sensory_drive=1.0, 
                    intrinsic_drive=0.1, 
                    learning_gate=1.0  # ゲートを開放!推論時も動的に更新
                )
                ensemble_perceptions.append(perception)
            
            # アンサンブルとしての最終決定
            avg_perception = np.mean(ensemble_perceptions, axis=0)
            final_class = np.argmax(avg_perception)
            y_pred.append(final_class)
            
            # 【変更点】細胞を増やす・間引く(新陳代謝の実行)
            # サンプルごと(または数サンプルごと)に代謝を回すことで、推論中にリアルタイムに細胞が増減します
            if run_evolution and (idx + 1) % 10 == 0: # ここでは10サンプルごとに新陳代謝を実行
                self.evolve()
                
        return y_pred
    def get_average_cell_count(self):
        return int(np.mean([len(m['space']) for m in self.models]))
# =========================================================
# 🚀 6. 巡礼ループの実行
# =========================================================
print("=== 蜘蛛の巣ウェブ型・側頭抑制&ミーム交配システム 起動 ===")
ensemble = EnsembleVisionEcosystem(n_estimators=5, total_cells=INIT_CELLS)
for cycle in range(epochs):
    # ☀️ 昼
    ensemble.train_cycle(X_train, y_train, phase="day")
    # 🌆 夕
    ensemble.train_cycle(X_train, y_train, phase="evening")
    # 🌙 夜(夢)
    ensemble.sleep_cycle(steps=60)
    
    # 🌌 夢の最深部で「ミーム文化的交配」が発生
    ensemble.cultural_crossover(crossover_rate=0.2)
    
    # 🧬 新陳代謝(分裂コスト制限 & 淘汰)
    ensemble.evolve()
    
    # 📊 テスト
    y_pred = ensemble.predict(X_test)
    acc = accuracy_score(y_test, y_pred)
    
    avg_cells = ensemble.get_average_cell_count()
    print(f"Cycle {cycle+1:02d}/{epochs} | 平均現存細胞数: {avg_cells:>3} | アンサンブル正解率: {acc:.4f}")
# =========================================================
# 7. ダッシュボード:【 空想状態(Daydream) 】の可視化
# =========================================================
print("\n[Generating Daydream Images from the First Evolved Ecosystem...]")
fig, axes = plt.subplots(2, 5, figsize=(12, 5))
fig.suptitle("Daydreaming Mode (Top-Down Drive via Double Edges) - Model #1\nVisualizing the 'Ideals' of digits polished by Structural Control", fontsize=14)
first_model = ensemble.models[0]
for cls in range(10):
    _, daydream_image = live_vision_cycle(first_model['space'], first_model['graph'], 
                                          sensory_drive=0.0, intrinsic_drive=0.2, top_down_drive=1.0, 
                                          target_recall_class=cls, learning_gate=0.0)
    ax = axes[cls // 5, cls % 5]
    ax.imshow(daydream_image.reshape(8, 8), cmap="bone")
    ax.set_title(f"Concept: {cls}")
    ax.axis("off")
plt.tight_layout()
plt.show()
    

第二版(このコードは精度を保証しません)

      from collections import defaultdict
import matplotlib.pyplot as plt
import numpy as np
import random
from sklearn.datasets import load_digits
from sklearn.metrics import accuracy_score
from sklearn.preprocessing import StandardScaler
# =========================================================
# 1. 環境とデータの準備(手書き数字 64次元データ)
# =========================================================
np.random.seed(42)
digits = load_digits()
X_raw, y_raw = digits.data, digits.target
# 生態系の変化を見やすくするため、400サンプルを抽出
total_indices = np.random.choice(len(X_raw), 400, replace=False)
X_400, y_400 = X_raw[total_indices], y_raw[total_indices]
scaler = StandardScaler()
X_scaled = scaler.fit_transform(X_400)
# 少数サンプルを訓練用に(残りをテスト環境に)
train_idx = []
for c in range(10):
    c_idxs = np.where(y_400 == c)[0]
    train_idx.extend(c_idxs[:1]) # 各クラス1枚ずつ
test_idx = [i for i in range(400) if i not in train_idx]
X_train, y_train = X_scaled[train_idx], y_400[train_idx]
X_test, y_test = X_scaled[test_idx], y_400[test_idx]
NUM_CLASSES = 10
INPUT_DIM = 64
# --- ハイパーパラメータ ---
INIT_CELLS = 690
MAX_CELLS = 10000
MIN_CELLS = 80
THOUGHT_STEPS = 8
lr_base = 0.30
lr_link_base = 0.10
DIVISION_THRESHOLD = 0.85     # 構造過学習を防ぐため、少し厳格化
DEATH_THRESHOLD = 0.08        # 淘汰圧を少し強める
GRAPH_DECAY = 0.90
EDGE_PRUNE_THRESHOLD = 0.02
epochs = 6                   # 現象を観察するため、5サイクル回します
# =========================================================
# 2. 視覚記憶細胞(VisionMemoryCell)の定義
# =========================================================
class VisionMemoryCell:
    def __init__(self, cell_id):
        self.id = cell_id
        self.w_in = np.random.normal(0, 1.0, INPUT_DIM)  # 蜘蛛の巣初期化で再上書きされます
        self.A = np.random.normal(0, 0.1, NUM_CLASSES)
        self.links = {}
        self.visits = 0.0
        self.energy = 0.5
        self.recent_activity = 0.0
def cosine_similarity(w, x):
    return np.dot(w, x) / (np.linalg.norm(w) * np.linalg.norm(x) + 1e-9)
def softmax(x, temp=1.0):
    x = np.array(x) / temp
    x = x - np.max(x)
    e = np.exp(x)
    return e / (np.sum(e) + 1e-12)
# =========================================================
# 🧬 3. 代謝・新陳代謝コア(分裂コストの厳格化)
# =========================================================
def evolve_and_prune_space(memory_space, outer_graph):
    # 睡眠・自然減衰
    for cell in memory_space:
        cell.energy *= 0.92
        cell.recent_activity *= 0.50
        if cell.energy < 0.01: cell.energy = 0.01
        
    # --- 分裂フェーズ(ユーザーさんの提案によるコスト厳格化) ---
    new_cells = []
    current_max_id = max([c.id for c in memory_space]) if memory_space else 0
    for cell in memory_space:
        if len(memory_space) + len(new_cells) >= MAX_CELLS: break
        
        # 分裂条件をエネルギーかつ、 visits累積に設定
        if cell.energy > DIVISION_THRESHOLD and cell.visits > 15:
            child = VisionMemoryCell(current_max_id + len(new_cells) + 1)
            
            # 子は親の特性に少し変異(ノイズ)を乗せる
            child.w_in = np.clip(cell.w_in + np.random.normal(0, 0.05, INPUT_DIM), -3.0, 3.0)
            child.A = cell.A.copy()
            for k, v in cell.links.items(): child.links[k] = v * 0.5
            
            # 【ユーザーさんの設計思想】親の資源を大きく削り、子を未熟な状態からスタートさせる
            cell.energy *= 0.3
            cell.visits *= 0.5
            child.energy = 0.4  # 子の初期エネルギーは低い
            
            new_cells.append(child)
            
    memory_space.extend(new_cells)
    
    # --- 淘汰 & 外側マクログラフの再マッピング ---
    survived_space = []
    id_map = {}
    for cell in memory_space:
        if len(survived_space) > MIN_CELLS:
            if cell.energy < DEATH_THRESHOLD and cell.visits < 1.5:
                continue
        new_id = len(survived_space)
        id_map[cell.id] = new_id
        cell.id = new_id
        survived_space.append(cell)
        
    for cell in survived_space:
        cell.links = {id_map[old_id]: w * 0.85 for old_id, w in cell.links.items() if old_id in id_map and w > 0.05}
        
    new_graph = defaultdict(dict)
    for u, neighbors in outer_graph.items():
        new_u = ("cell", id_map[u[1]]) if u[0] == "cell" and u[1] in id_map else u
        if u[0] == "cell" and u[1] not in id_map: continue
        
        for v, weight in neighbors.items():
            decayed_weight = weight * GRAPH_DECAY
            if decayed_weight < EDGE_PRUNE_THRESHOLD: continue
            
            new_v = ("cell", id_map[v[1]]) if v[0] == "cell" and v[1] in id_map else v
            if v[0] == "cell" and v[1] not in id_map: continue
            
            new_graph[new_u][new_v] = decayed_weight
            
    return survived_space, new_graph
# =========================================================
# 🔄 4. 融合生命ダイナミクス(思考の歩行・側頭抑制)
# =========================================================
def live_vision_cycle(memory_space, outer_graph, x_64d=None, target_class=None, target_recall_class=None,
                      sensory_drive=1.0, intrinsic_drive=0.0, top_down_drive=0.0, learning_gate=1.0):
    if len(memory_space) == 0: return np.zeros(NUM_CLASSES), np.zeros(INPUT_DIM)
    id_to_cell = {c.id: c for c in memory_space}
    
    start_probabilities = np.zeros(len(memory_space))
    
    if x_64d is not None and sensory_drive > 0.0:
        sims = np.array([cosine_similarity(cell.w_in, x_64d) for cell in memory_space])
        start_probabilities += softmax(sims, temp=0.1) * sensory_drive
        
    if intrinsic_drive > 0.0:
        activity_weights = np.array([cell.energy * (cell.recent_activity + 0.05) for cell in memory_space])
        start_probabilities += (activity_weights / (np.sum(activity_weights) + 1e-12)) * intrinsic_drive
        
    if target_recall_class is not None and top_down_drive > 0.0:
        c_node = ("class", target_recall_class)
        for next_node, weight in outer_graph.get(c_node, {}).items():
            if next_node[0] == "cell" and next_node[1] in id_to_cell:
                start_probabilities[next_node[1]] += weight * top_down_drive
                
    if np.sum(start_probabilities) == 0: 
        start_probabilities = np.ones(len(memory_space)) / len(memory_space)
    start_probabilities /= np.sum(start_probabilities)
    
    current_cell_id = np.random.choice(list(id_to_cell.keys()), p=start_probabilities)
    path = [current_cell_id]
    
    collected_signals = [id_to_cell[current_cell_id].A.copy()]
    imagined_pixels = [id_to_cell[current_cell_id].w_in.copy()]
    
    # --- 思考がグラフを歩く(Graph Walk) ---
    for step in range(THOUGHT_STEPS - 1):
        current_cell = id_to_cell[current_cell_id]
        current_cell.recent_activity += sensory_drive * 1.0
        
        all_ids = list(id_to_cell.keys())
        candidates = set(np.random.choice(all_ids, min(25, len(all_ids)), replace=False))
        candidates.update(current_cell.links.keys())
        
        scores = []
        c_ids = []
        for cid in candidates:
            if cid in path or cid not in id_to_cell: continue
            cell = id_to_cell[cid]
            
            score = current_cell.links.get(cid, 0.0) * 2.5
            if x_64d is not None:
                score += 2.0 * cosine_similarity(cell.w_in, x_64d) * sensory_drive
            score += 0.5 * cosine_similarity(current_cell.A, cell.A)
            
            scores.append(score)
            c_ids.append(cid)
            
        if not scores: break
        probs = softmax(scores, temp=0.3)
        current_cell_id = np.random.choice(c_ids, p=probs)
        path.append(current_cell_id)
        collected_signals.append(id_to_cell[current_cell_id].A.copy())
        imagined_pixels.append(id_to_cell[current_cell_id].w_in.copy())
        
    final_perception = np.mean(np.stack(collected_signals), axis=0) if collected_signals else np.zeros(NUM_CLASSES)
    final_image = np.mean(np.stack(imagined_pixels), axis=0) if imagined_pixels else np.zeros(INPUT_DIM)
    
    sensory_factor = sensory_drive * learning_gate
    intrinsic_factor = intrinsic_drive * learning_gate
    
    inferred_class = np.argmax(final_perception)
    target = target_class if target_class is not None else inferred_class
    
    target_vec = np.zeros(NUM_CLASSES)
    target_vec[target] = 1.0
    
    # 報酬の決定(厳格化:専門化を促す)
    reward = min(1.0 / (np.linalg.norm(target_vec - final_perception) + 0.15), 2.5)
    
    # --- ヘブ則的な学習と、専門化のための側頭抑制(Lateral Inhibition) ---
    if x_64d is not None and sensory_factor > 0.0:
        # 1. 経路上の細胞の学習
        for cid in path:
            if cid in id_to_cell:
                cell = id_to_cell[cid]
                cell.visits += sensory_factor
                cell.energy = min(1.0, cell.energy + reward * 0.15 * sensory_factor)
                cell.w_in += (lr_base * 0.5 * sensory_factor) * (x_64d - cell.w_in)
                cell.A = (1.0 - lr_base * sensory_factor) * cell.A + (lr_base * sensory_factor) * target_vec
                
        # 2. 側頭抑制(同じ情報への過剰適合・重複増殖を防ぐ)
        # 歩いた細胞の近隣リンクにありながら、今回選ばれなかった細胞のエネルギーを少し削る
        for cid in path:
            if cid in id_to_cell:
                for neighbor_id in id_to_cell[cid].links.keys():
                    if neighbor_id not in path and neighbor_id in id_to_cell:
                        id_to_cell[neighbor_id].energy *= 0.95 # 側頭抑制によるエネルギー減衰
                
    total_learning_power = (sensory_factor * reward + intrinsic_factor * 0.2)
    if total_learning_power > 0.0:
        for i in range(len(path) - 1):
            u, v = path[i], path[i+1]
            if u in id_to_cell and v in id_to_cell:
                id_to_cell[u].links[v] = id_to_cell[u].links.get(v, 0.1) + (lr_link_base * total_learning_power)
                
        c_node = ("class", int(target))
        first_cell_node = ("cell", path[0])
        last_cell_node = ("cell", path[-1])
        
        for i in range(len(path) - 1):
            f_node = ("cell", path[i])
            t_node = ("cell", path[i+1])
            outer_graph[f_node][t_node] = outer_graph[f_node].get(t_node, 0.0) + total_learning_power
            outer_graph[t_node][f_node] = outer_graph[t_node].get(f_node, 0.0) + total_learning_power
            
        # クラスから細胞、細胞からクラスへの「双方向エッジ」を正確にマッピング
        outer_graph[first_cell_node][c_node] = outer_graph[first_cell_node].get(c_node, 0.0) + total_learning_power
        outer_graph[c_node][first_cell_node] = outer_graph[c_node].get(first_cell_node, 0.0) + total_learning_power
        
        outer_graph[last_cell_node][c_node] = outer_graph[last_cell_node].get(c_node, 0.0) + total_learning_power
        outer_graph[c_node][last_cell_node] = outer_graph[c_node].get(last_cell_node, 0.0) + total_learning_power
        
    return final_perception, final_image
# =========================================================
# 👥 5. アンサンブル・蜘蛛の巣トポロジー管理(ミーム交配の実装)
# =========================================================
class EnsembleVisionEcosystem:
    def __init__(self, n_estimators=5, total_cells=1000):
        self.n_estimators = n_estimators
        self.models = []
        NUM_RINGS = 5  # 蜘蛛の巣の階層数
        
        for _ in range(n_estimators):
            space = []
            graph = defaultdict(dict)
            
            # 密度の調整:外側(感覚層)ほど細胞を多く、内側(概念層)ほど少なく
            cells_per_ring = []
            remaining = total_cells
            for r in range(NUM_RINGS):
                if r == NUM_RINGS - 1:
                    cells_per_ring.append(remaining)
                else:
                    count = int(total_cells * (0.45 / (1.6**r)))
                    cells_per_ring.append(count)
                    remaining -= count
                    
            cell_id_counter = 0
            ring_nodes = {}
            
            # --- 1. 蜘蛛の巣の細胞配置と受容野初期化 ---
            for ring_idx, n_cells in enumerate(cells_per_ring):
                ring_nodes[ring_idx] = []
                for i in range(n_cells):
                    cell = VisionMemoryCell(cell_id_counter)
                    
                    # 外周(感覚層)ほど実際の数字に近い初期受容野を持たせる
                    sensory_proximity = 1.0 - (ring_idx / NUM_RINGS)
                    if sensory_proximity > 0.2:
                        ref_idx = np.random.choice(len(X_train))
                        base_pattern = X_train[ref_idx]
                        cell.w_in = base_pattern * sensory_proximity + np.random.normal(0, 0.3, INPUT_DIM) * (1.0 - sensory_proximity)
                        cell.A[y_train[ref_idx]] = 0.5 * sensory_proximity
                    
                    space.append(cell)
                    ring_nodes[ring_idx].append(cell_id_counter)
                    cell_id_counter += 1
                    
            # --- 2. 蜘蛛の巣の糸(リンク)の構築 ---
            id_to_cell = {c.id: c for c in space}
            for ring_idx in range(NUM_RINGS):
                nodes = ring_nodes[ring_idx]
                num_nodes = len(nodes)
                if num_nodes == 0: continue
                
                for i in range(num_nodes):
                    curr_id = nodes[i]
                    
                    # (a) 同心円の横の糸(横方向リンク)
                    next_id = nodes[(i + 1) % num_nodes]
                    id_to_cell[curr_id].links[next_id] = 0.4
                    id_to_cell[next_id].links[curr_id] = 0.4
                    
                    # (b) 放射方向の縦の糸(抽象化への階層リンク)
                    if ring_idx < NUM_RINGS - 1:
                        inner_nodes = ring_nodes[ring_idx + 1]
                        if len(inner_nodes) > 0:
                            inner_idx = int(i * len(inner_nodes) / num_nodes)
                            inner_id = inner_nodes[inner_idx]
                            
                            id_to_cell[curr_id].links[inner_id] = 0.6
                            id_to_cell[inner_id].links[curr_id] = 0.6
            self.models.append({'space': space, 'graph': graph})
    def train_cycle(self, X, y, phase="day"):
        for model in self.models:
            # 脳の個性を引き出すブートストラップサンプリング
            indices = np.random.choice(len(X), len(X), replace=True)
            for idx in indices:
                if phase == "day":
                    live_vision_cycle(model['space'], model['graph'], x_64d=X[idx], 
                                      target_class=y[idx], sensory_drive=1.0, intrinsic_drive=0.0)
                elif phase == "evening":
                    live_vision_cycle(model['space'], model['graph'], x_64d=X[idx], 
                                      target_class=y[idx], sensory_drive=0.3, intrinsic_drive=0.7)
    def sleep_cycle(self, steps=60):
        for model in self.models:
            for _ in range(steps):
                live_vision_cycle(model['space'], model['graph'], x_64d=None, 
                                  target_class=None, sensory_drive=0.0, intrinsic_drive=1.0)
    def cultural_crossover(self, crossover_rate=0.2):
        """異なる脳の間で、最も洗練されたイデア(ミーム)を交換する"""
        elite_cells_per_model = []
        
        for m_idx, model in enumerate(self.models):
            space = model['space']
            if not space: continue
            
            # 最深部(Ring 4に相当する後半10%のコア層)からエリートを選別
            core_start_idx = int(len(space) * 0.9)
            core_cells = space[core_start_idx:]
            if not core_cells: continue
            
            # エネルギーが高く、キャラクター(特定のクラスへの確信)が立っている細胞をスカウト
            elite_cell = max(core_cells, key=lambda c: (c.energy * np.max(c.A)))
            elite_cells_per_model.append((m_idx, elite_cell))
            
        # 集団間での文化伝播(シナプス感染)
        for src_idx, src_cell in elite_cells_per_model:
            target_indices = [i for i in range(self.n_estimators) if i != src_idx]
            if not target_indices: continue
            dst_idx = random.choice(target_indices)
            dst_model = self.models[dst_idx]
            
            dst_core_cells = dst_model['space'][int(len(dst_model['space']) * 0.9):]
            if not dst_core_cells: continue
            
            # 伝播先の脳の中で、最も迷っている(エネルギーの低い)細胞にミームをブレンドする
            weak_cell = min(dst_core_cells, key=lambda c: c.energy)
            
            # 知識の混ざり合い
            weak_cell.w_in = (1.0 - crossover_rate) * weak_cell.w_in + crossover_rate * src_cell.w_in
            weak_cell.A = (1.0 - crossover_rate) * weak_cell.A + crossover_rate * src_cell.A
            weak_cell.energy = max(weak_cell.energy, src_cell.energy * 0.8) # 文化刺激による活性化
    def evolve(self):
        for model in self.models:
            model['space'], model['graph'] = evolve_and_prune_space(model['space'], model['graph'])
    def predict(self, X, run_evolution=True):
        """
        推論を実行しつつ、その入力データにリアルタイムに適応して
        細胞を更新・分裂・淘汰させる(リアルタイム・オンライン学習モード)
        """
        y_pred = []
        for idx in range(len(X)):
            ensemble_perceptions = []
            
            # 各ベースモデル(脳)がデータに反応し、同時に学習・更新を行う
            for model in self.models:
                # 【変更点】
                # 1. target_class=None にすることで、自分の推論結果をターゲット(疑似ラベル)にする
                # 2. learning_gate=1.0 にして、推論時もヘブ則や側頭抑制を完全に作動させる
                perception, _ = live_vision_cycle(
                    model['space'], model['graph'], x_64d=X[idx], 
                    target_class=None, # 自身の推論(inferred_class)で自己学習
                    sensory_drive=1.0, 
                    intrinsic_drive=0.1, 
                    learning_gate=1.0  # ゲートを開放!推論時も動的に更新
                )
                ensemble_perceptions.append(perception)
            
            # アンサンブルとしての最終決定
            avg_perception = np.mean(ensemble_perceptions, axis=0)
            final_class = np.argmax(avg_perception)
            y_pred.append(final_class)
            
            # 【変更点】細胞を増やす・間引く(新陳代謝の実行)
            # サンプルごと(または数サンプルごと)に代謝を回すことで、推論中にリアルタイムに細胞が増減します
            if run_evolution and (idx + 1) % 10 == 0: # ここでは10サンプルごとに新陳代謝を実行
                self.evolve()
                
        return y_pred
    def get_average_cell_count(self):
        return int(np.mean([len(m['space']) for m in self.models]))
# =========================================================
# 🚀 6. 交互巡礼・適応ループの実行(訓練 ⇄ 推論の交互サイクル)
# =========================================================
print("=== 蜘蛛の巣ウェブ型・交互生態系システム 起動 ===")
ensemble = EnsembleVisionEcosystem(n_estimators=5, total_cells=INIT_CELLS)
# 全体のマクロサイクル数
MACRO_CYCLES = 1 
for m_cycle in range(MACRO_CYCLES):
    print(f"\n==================================================")
    print(f"🌀 マクロサイクル {m_cycle+1}/{MACRO_CYCLES} 開始")
    print(f"==================================================")
    
    # -----------------------------------------------------
    # 【フェーズA】訓練サイクル(基底知識のアップデート)
    # -----------------------------------------------------
    print("\n--- [Phase A: Training] 基礎訓練・睡眠・ミーム交配 ---")
    # ☀️ 昼
    ensemble.train_cycle(X_train, y_train, phase="day")
    # 🌆 夕
    ensemble.train_cycle(X_train, y_train, phase="evening")
    # 🌙 夜(夢)
    ensemble.sleep_cycle(steps=600)
    # 🌌 ミーム文化的交配
    ensemble.cultural_crossover(crossover_rate=0.2)
    # 🧬 新陳代謝
    ensemble.evolve()
    
    avg_cells = ensemble.get_average_cell_count()
    print(f"訓練完了 | 平均現存細胞数: {avg_cells}")
    # -----------------------------------------------------
    # 【フェーズB】推論&リアルタイム適応(5回連続パス)
    # -----------------------------------------------------
    print("\n--- [Phase B: Inference & Adaptation] 5回連続推論実験 ---")
    
    # このマクロサイクル内での推論ログ
    pass_logs = []
    
    for pass_idx in range(3):
        # 推論を実行しつつ、リアルタイムに自己更新(learning_gate=1.0)と代謝を実行
        y_pred = ensemble.predict(X_test, run_evolution=True)
        acc = accuracy_score(y_test, y_pred)
        
        avg_cells_infer = ensemble.get_average_cell_count()
        
        pass_logs.append({
            'Pass': pass_idx + 1,
            'Accuracy': acc,
            'Avg Cells': avg_cells_infer,
        })
        print(f"  Pass {pass_idx+1}/5 | 正解率: {acc:.4f} | 平均細胞数: {avg_cells_infer}")
        
    # --- 現在のサイクルのサマリー表示 ---
    print(f"\n【マクロサイクル {m_cycle+1} の実験結果】")
    print(f"  Pass  | Accuracy   | Avg Cells ")
    print("  -------------------------------")
    for log in pass_logs:
        print(f"   {log['Pass']:<4} | {log['Accuracy']:<10.4f} | {log['Avg Cells']:<10}")
print("\n=== すべての交互サイクルが完了しました ===")
# 🔄 6.5. 実験:同じテストデータを5回連続で流す(生態追跡)
# =========================================================
print("\n=== 実験: 同一テストセットを5回連続で推論(動的適応の観察) ===")
# 進化ログの格納用
pass_logs = []
# テストデータをシャffleせず、完全に同じ順序で2回流す
for pass_idx in range(2):
    # predict内部で学習(learning_gate=1.0)と、10サンプルごとのevolve()が走る
    y_pred = ensemble.predict(X_test, run_evolution=True)
    acc = accuracy_score(y_test, y_pred)
    
    # 現在のアンサンブルの平均細胞数を取得
    avg_cells = ensemble.get_average_cell_count()
    
    # グラフの総エッジ数(リンク数)を簡易計算
    total_edges = sum(len(m['graph']) for m in ensemble.models)
    
    pass_logs.append({
        'Pass': pass_idx + 1,
        'Accuracy': acc,
        'Avg Cells': avg_cells,
    })
    
    print(f"Pass {pass_idx+1}/5 | 正解率: {acc:.4f} | 平均細胞数: {avg_cells}")
# --- 結果のテーブル表示 ---
print("\n【実験結果サマリー】")
print(f"{'Pass':<6} | {'Accuracy':<10} | {'Avg Cells':<10}")
print("-" * 34)
for log in pass_logs:
    print(f"{log['Pass']:<6} | {log['Accuracy']:<10.4f} | {log['Avg Cells']:<10}")
# =========================================================
# 7. ダッシュボード:【 空想状態(Daydream) 】の可視化
# =========================================================
print("\n[Generating Daydream Images from the First Evolved Ecosystem...]")
fig, axes = plt.subplots(2, 5, figsize=(12, 5))
fig.suptitle("Daydreaming Mode (Top-Down Drive via Double Edges) - Model #1\nVisualizing the 'Ideals' of digits polished by Structural Control", fontsize=14)
first_model = ensemble.models[0]
for cls in range(10):
    _, daydream_image = live_vision_cycle(first_model['space'], first_model['graph'], 
                                          sensory_drive=0.0, intrinsic_drive=0.2, top_down_drive=1.0, 
                                          target_recall_class=cls, learning_gate=0.0)
    ax = axes[cls // 5, cls % 5]
    ax.imshow(daydream_image.reshape(8, 8), cmap="bone")
    ax.set_title(f"Concept: {cls}")
    ax.axis("off")
plt.tight_layout()
plt.show()

    
      from collections import defaultdict
import matplotlib.pyplot as plt
import numpy as np
import random
import sys
from sklearn.datasets import load_digits
from sklearn.metrics import accuracy_score, confusion_matrix, ConfusionMatrixDisplay
from sklearn.preprocessing import StandardScaler
# =========================================================
# 1. 環境とデータの準備(手書き数字 64次元データ)
# =========================================================
np.random.seed(42)
random.seed(42)
digits = load_digits()
X_raw, y_raw = digits.data, digits.target
total_indices = np.random.choice(len(X_raw), 400, replace=False)
X_400, y_400 = X_raw[total_indices], y_raw[total_indices]
scaler = StandardScaler()
X_scaled = scaler.fit_transform(X_400)
train_idx = []
for c in range(10):
    c_idxs = np.where(y_400 == c)[0]
    train_idx.extend(c_idxs[:1])
test_idx = [i for i in range(400) if i not in train_idx]
X_train, y_train = X_scaled[train_idx], y_400[train_idx]
X_test, y_test = X_scaled[test_idx], y_400[test_idx]
NUM_CLASSES = 10
INPUT_DIM = 64
# --- ハイパーパラメータ ---
INIT_CELLS = 350        
MAX_CELLS = 4000
MIN_CELLS = 80
THOUGHT_STEPS = 9
DIVISION_THRESHOLD = 0.5
DEATH_THRESHOLD = 0.09
epochs = 30  # デモ実行用に少しサイクル数を調整
# =========================================================
# 👥 2. 細胞の定義(4つの個性・役割分化の実装)
# =========================================================
class VisionMemoryCell:
    def __init__(self, cell_id):
        self.id = cell_id
        self.w_in = np.random.normal(0, 0.3, INPUT_DIM)
        self.A = np.ones(NUM_CLASSES) / NUM_CLASSES
        self.pos = np.random.uniform(-0.3, 0.2, 3)
        self.links = {} 
        
        self.visits = 0.0
        self.energy = 0.5
        self.recent_activity = 0.0
        
        self.role = random.choice(["Explorer", "Storage", "Bridge", "Default"])
        
        if self.role == "Explorer":
            self.lr = 0.35          
            self.a_decay = 0.70    
            self.div_rate = 1.2     
        elif self.role == "Storage":
            self.lr = 0.08          
            self.a_decay = 0.998    
            self.div_rate = 1.3
        elif self.role == "Bridge":
            self.lr = 0.20
            self.a_decay = 0.992
            self.div_rate = 1.3
            self.pos *= 0.5         
        else: # Default
            self.lr = 0.20
            self.a_decay = 0.995
            self.div_rate = 1.2
def cosine_similarity(w, x):
    return np.dot(w, x) / (np.linalg.norm(w) * np.linalg.norm(x) + 1e-9)
def softmax(x, temp=1.0):
    x = np.array(x) / temp
    x = x - np.max(x)
    e = np.exp(x)
    return e / (np.sum(e) + 1e-12)
# =========================================================
# 🌐 3. 空間トポロジー力学(構造記憶 & グリッド高速化)
# =========================================================
def update_topology_with_memory(memory_space, radius_connect=1.5, decay=0.92):
    for cell in memory_space:
        cell.links = {k: v * decay for k, v in cell.links.items() if v * decay > 0.04}
        
    num_cells = len(memory_space)
    for i in range(num_cells):
        for j in range(i + 1, num_cells):
            c1 = memory_space[i]
            c2 = memory_space[j]
            dist = np.linalg.norm(c1.pos - c2.pos)
            
            if dist < radius_connect:
                delta_w = (1.0 - (dist / radius_connect)) * 0.12
                c1.links[c2.id] = min(1.0, c1.links.get(c2.id, 0.0) + delta_w)
                c2.links[c1.id] = min(1.0, c2.links.get(c1.id, 0.0) + delta_w)
def update_cell_positions_fast(memory_space, path_history, grid_size=1.2):
    dt = 0.1
    num_cells = len(memory_space)
    if num_cells == 0: return
    forces = {c.id: np.zeros(3) for c in memory_space}
    
    grid = defaultdict(list)
    for cell in memory_space:
        grid_key = tuple(np.floor(cell.pos / grid_size).astype(int))
        grid[grid_key].append(cell)
        
    for grid_key, cells_in_grid in grid.items():
        for dx in [-1, 0, 1]:
            for dy in [-1, 0, 1]:
                for dz in [-1, 0, 1]:
                    neighbor_key = (grid_key[0]+dx, grid_key[1]+dy, grid_key[2]+dz)
                    if neighbor_key not in grid: continue
                    
                    for c1 in cells_in_grid:
                        for c2 in grid[neighbor_key]:
                            if c1.id >= c2.id: continue
                            
                            direction = c2.pos - c1.pos
                            distance = np.linalg.norm(direction) + 1e-5
                            unit_dir = direction / distance
                            
                            repulsion = 0.03 / (distance ** 2)
                            sim_A = cosine_similarity(c1.A, c2.A)
                            
                            bridge_factor = 1.3 if (c1.role == "Bridge" or c2.role == "Bridge") else 1.0
                            
                            if sim_A > 0.6:
                                attraction = 0.08 * (sim_A - 0.6) * distance * bridge_factor
                                force_mag = -repulsion + attraction
                            else:
                                force_mag = -(repulsion + 0.05 * (0.6 - sim_A))
                                
                            forces[c1.id] += unit_dir * force_mag
                            forces[c2.id] -= unit_dir * force_mag
    for path in path_history:
        for i in range(len(path) - 1):
            u, v = path[i], path[i+1]
            if u not in forces or v not in forces: continue
            c_u = next((c for c in memory_space if c.id == u), None)
            c_v = next((c for c in memory_space if c.id == v), None)
            if not c_u or not c_v: continue
            
            dir_uv = c_v.pos - c_u.pos
            dist_uv = np.linalg.norm(dir_uv) + 1e-5
            forces[u] += (dir_uv / dist_uv) * 0.05 * dist_uv
            forces[v] -= (dir_uv / dist_uv) * 0.05 * dist_uv
                
    for cell in memory_space:
        cell.pos += forces[cell.id] * dt
        cell.pos = np.clip(cell.pos, -4.0, 4.0)
# =========================================================
# 🔄 4. 融合生命ダイナミクス(環境スケール対応版)
# =========================================================
def live_vision_cycle(memory_space, x_64d=None, target_class=None, 
                      sensory_drive=1.0, intrinsic_drive=0.0, learning_gate=1.0,
                      lr_scale=1.0, role_lr_scales=None):
    if role_lr_scales is None:
        role_lr_scales = {"Explorer": 1.0, "Storage": 1.0, "Bridge": 1.0, "Default": 1.0}
    if len(memory_space) == 0: return np.zeros(NUM_CLASSES), np.zeros(INPUT_DIM), []
    id_to_cell = {c.id: c for c in memory_space}
    
    start_probabilities = np.zeros(len(memory_space))
    if x_64d is not None and sensory_drive > 0.0:
        sims = np.array([cosine_similarity(cell.w_in, x_64d) for cell in memory_space])
        start_probabilities += softmax(sims, temp=0.1) * sensory_drive
    if intrinsic_drive > 0.0:
        activity_weights = np.array([cell.energy * (cell.recent_activity + 0.05) for cell in memory_space])
        start_probabilities += (activity_weights / (np.sum(activity_weights) + 1e-12)) * intrinsic_drive
        
    if np.sum(start_probabilities) == 0: 
        start_probabilities = np.ones(len(memory_space)) / len(memory_space)
    start_probabilities /= np.sum(start_probabilities)
    
    current_cell_id = np.random.choice(list(id_to_cell.keys()), p=start_probabilities)
    path = [current_cell_id]
    
    collected_signals = [id_to_cell[current_cell_id].A.copy()]
    imagined_pixels = [id_to_cell[current_cell_id].w_in.copy()]
    
    for step in range(THOUGHT_STEPS - 1):
        current_cell = id_to_cell[current_cell_id]
        current_cell.recent_activity += sensory_drive * 1.0
        
        all_ids = list(id_to_cell.keys())
        
        if current_cell.role == "Explorer" and random.random() < 0.25:
            current_cell_id = random.choice(all_ids)
            path.append(current_cell_id)
            collected_signals.append(id_to_cell[current_cell_id].A.copy())
            imagined_pixels.append(id_to_cell[current_cell_id].w_in.copy())
            continue
            
        candidates = set(np.random.choice(all_ids, min(20, len(all_ids)), replace=False))
        candidates.update(current_cell.links.keys())
        
        scores = []
        c_ids = []
        for cid in candidates:
            if cid in path or cid not in id_to_cell: continue
            cell = id_to_cell[cid]
            
            score = current_cell.links.get(cid, 0.0) * 3.0
            if x_64d is not None:
                score += 2.0 * cosine_similarity(cell.w_in, x_64d) * sensory_drive
            
            score += 0.8 * cosine_similarity(current_cell.A, cell.A)
            
            if cell.role == "Bridge":
                score *= 2.5
                
            scores.append(score)
            c_ids.append(cid)
            
        if not scores: break
        probs = softmax(scores, temp=0.3)
        current_cell_id = np.random.choice(c_ids, p=probs)
        path.append(current_cell_id)
        
        collected_signals.append(id_to_cell[current_cell_id].A.copy())
        imagined_pixels.append(id_to_cell[current_cell_id].w_in.copy())
        
    final_perception = np.mean(np.stack(collected_signals), axis=0) if collected_signals else np.zeros(NUM_CLASSES)
    final_image = np.mean(np.stack(imagined_pixels), axis=0) if imagined_pixels else np.zeros(INPUT_DIM)
    
    sensory_factor = sensory_drive * learning_gate
    inferred_class = np.argmax(final_perception)
    target = target_class if target_class is not None else inferred_class
    
    target_vec = np.zeros(NUM_CLASSES)
    target_vec[target] = 1.0
    
    base_reward = min(1.0 / (np.linalg.norm(target_vec - final_perception) + 0.15), 2.5) * 1.5
    
    novelty_reward = 0.0
    for cid in path:
        if cid in id_to_cell and id_to_cell[cid].visits < 4.0:
            novelty_reward += 0.10
            
    info_gain = 0.0
    if len(collected_signals) > 1:
        first_error = np.linalg.norm(target_vec - collected_signals[0])
        final_error = np.linalg.norm(target_vec - final_perception)
        info_gain = max(0.0, first_error - final_error) * 2.0
        
    reward = base_reward + novelty_reward + info_gain
    
    for cid in path:
        if cid in id_to_cell:
            for neighbor_id in id_to_cell[cid].links.keys():
                if neighbor_id not in path and neighbor_id in id_to_cell:
                    id_to_cell[neighbor_id].energy *= 0.96
    if x_64d is not None and sensory_factor > 0.0:
        unique_visited_ids = set(path)
        
        for cid in unique_visited_ids:
            if cid not in id_to_cell: continue
            cell = id_to_cell[cid]
            
            if random.random() < 0.10: 
                continue 
            
            if cell.role == "Explorer":
                noise = np.random.normal(0, 0.12)
            else:
                noise = np.random.normal(0, 0.05)
            
            base_dynamic_lr = cell.lr * lr_scale * role_lr_scales.get(cell.role, 1.0)
            dynamic_lr = max(0.01, base_dynamic_lr + noise)
            
            cell.visits += sensory_factor
            cell.energy = min(1.0, cell.energy + reward * 0.12 * sensory_factor)
            
            cell.w_in += (dynamic_lr * 0.5 * sensory_factor) * (x_64d - cell.w_in)
            cell.A += (dynamic_lr * sensory_factor) * target_vec
            cell.A *= cell.a_decay
            cell.A /= (np.sum(cell.A) + 1e-12)
                                    
    return final_perception, final_image, path
# =========================================================
# 🧬 5. 代謝・新陳代謝コア(個性の遺伝と変異:div_scale対応)
# =========================================================
def evolve_and_prune_space(memory_space, div_scale=1.0):
    for cell in memory_space:
        cell.energy *= 0.90
        cell.recent_activity *= 0.50
        if cell.energy < 0.01: cell.energy = 0.01
        
    new_cells = []
    current_max_id = max([c.id for c in memory_space]) if memory_space else 0
    for cell in memory_space:
        if len(memory_space) + len(new_cells) >= MAX_CELLS: break
        
        effective_div_rate = cell.div_rate * div_scale
        if cell.energy > (DIVISION_THRESHOLD * (2.0 - effective_div_rate)) and cell.visits > 10:
            child = VisionMemoryCell(current_max_id + len(new_cells) + 1)
            child.w_in = np.clip(cell.w_in + np.random.normal(0, 0.05, INPUT_DIM), -3.0, 3.0)
            child.A = cell.A.copy()
            child.pos = cell.pos + np.random.normal(0, 0.1, 3)
            
            child.role = cell.role
            child.lr = cell.lr
            child.a_decay = cell.a_decay
            child.div_rate = cell.div_rate
            
            cell.energy *= 0.3
            cell.visits *= 0.5
            child.energy = 0.2
            new_cells.append(child)
            
    memory_space.extend(new_cells)
    
    survived_space = []
    id_map = {}
    for cell in memory_space:
        if len(survived_space) > MIN_CELLS:
            if cell.energy < DEATH_THRESHOLD and cell.visits < 1.0:
                continue
        id_map[cell.id] = len(survived_space)
        cell.id = id_map[cell.id]
        survived_space.append(cell)
        
    for cell in survived_space:
        cell.links = {id_map[old_id]: w for old_id, w in cell.links.items() if old_id in id_map}
        
    return survived_space
# =========================================================
# 👥 6. エコシステム・統合管理クラス
# =========================================================
class EnsembleVisionEcosystem:
    def __init__(self, n_estimators=3, total_cells=2500):
        self.n_estimators = n_estimators
        self.models = []
        
        self.lr_scale = 1.0
        self.div_scale = 1.0
        
        self.role_lr_scales = {
            "Explorer": 1.0,
            "Storage": 1.0,
            "Bridge": 1.0,
            "Default": 1.0
        }
        
        for _ in range(n_estimators):
            space = [VisionMemoryCell(i) for i in range(total_cells)]
            update_topology_with_memory(space) 
            self.models.append({'space': space})
            
    def train_cycle(self, X, y, phase="day"):
        for model in self.models:
            indices = np.random.choice(len(X), len(X), replace=True)
            path_history = []
            
            for idx in indices:
                s_drive = 1.0 if phase == "day" else 0.3
                i_drive = 0.0 if phase == "day" else 0.7
                
                _, _, path = live_vision_cycle(
                    model['space'], x_64d=X[idx], target_class=y[idx], 
                    sensory_drive=s_drive, intrinsic_drive=i_drive,
                    lr_scale=self.lr_scale, role_lr_scales=self.role_lr_scales
                )
                if path: path_history.append(path)
                
            update_cell_positions_fast(model['space'], path_history)
            update_topology_with_memory(model['space'])
            
    def sleep_cycle(self, steps=30):
        for model in self.models:
            path_history = []
            for _ in range(steps):
                _, _, path = live_vision_cycle(
                    model['space'], x_64d=None, target_class=None, 
                    sensory_drive=0.0, intrinsic_drive=1.0,
                    lr_scale=self.lr_scale, role_lr_scales=self.role_lr_scales
                )
                if path: path_history.append(path)
            update_cell_positions_fast(model['space'], path_history)
            update_topology_with_memory(model['space'])
            
    def evolve(self):
        for model in self.models:
            model['space'] = evolve_and_prune_space(model['space'], div_scale=self.div_scale)
            update_topology_with_memory(model['space'])
            
    def predict(self, X):
        y_pred = []
        for idx in range(len(X)):
            ensemble_perceptions = []
            ensemble_confidences = []
            
            for model in self.models:
                perception, _, _ = live_vision_cycle(
                    model['space'], x_64d=X[idx], 
                    sensory_drive=1.0, intrinsic_drive=0.0, learning_gate=0.0
                )
                confidence = np.max(perception)
                ensemble_perceptions.append(perception)
                ensemble_confidences.append(confidence)
                
            weighted_perceptions = np.zeros(NUM_CLASSES)
            total_conf = sum(ensemble_confidences) + 1e-12
            
            for p, c in zip(ensemble_perceptions, ensemble_confidences):
                weighted_perceptions += p * (c / total_conf)
                
            y_pred.append(np.argmax(weighted_perceptions))
        return y_pred
    def trigger_chaotic_rage(self):
        print("\n⚡ [EVENT] カオス・レイジ:未知のランダム大刺激が生態系を襲う!")
        for model in self.models:
            for cell in model['space']:
                cell.pos += np.random.uniform(-2.5, 2.5, 3)
                cell.pos = np.clip(cell.pos, -4.0, 4.0)
                cell.energy = min(1.0, cell.energy + 0.3)
                if random.random() < 0.4:
                    cell.lr = max(0.02, cell.lr * np.random.uniform(0.5, 2.0))
            update_topology_with_memory(model['space'])
    def trigger_hyper_focus(self, focus_digit, X_all, y_all):
        print(f"\n🧠 [EVENT] 残響ハイパーフォーカス:今学んでいる数字【 {focus_digit} 】に意識を集中させて再学習!")
        digit_indices = np.where(y_all == focus_digit)[0]
        if len(digit_indices) == 0:
            print("⚠ 対象データが見つかりません。")
            return
            
        for model in self.models:
            path_history = []
            for _ in range(3): 
                for idx in digit_indices:
                    _, _, path = live_vision_cycle(
                        model['space'], x_64d=X_all[idx], target_class=focus_digit,
                        sensory_drive=1.5, intrinsic_drive=0.1, lr_scale=self.lr_scale
                    )
                    if path: path_history.append(path)
            update_cell_positions_fast(model['space'], path_history)
            update_topology_with_memory(model['space'])
# =========================================================
# 7. 巡礼シミュレーションの実行(対話型マルチイベント版)
# =========================================================
print("=== 🌐 自己組織化認知生態系:マルチイベント介入モード 起動 ===")
ecosystem = EnsembleVisionEcosystem(n_estimators=3, total_cells=INIT_CELLS)
last_learned_digit = 0
# ノンブロッキング(自動デモ選択)を行うためのフラグ(必要に応じてFalseにすれば手動入力可能)
AUTO_DEMO = True 
for cycle in range(epochs):
    print(f"\n=========================================================")
    print(f"🔄 [Cycle {cycle+1:02d}/{epochs}] 開始前プロンプト")
    print(f"現在のステータス ➔ 学習率(lr): {ecosystem.lr_scale:.2f} | 分裂速度(div): {ecosystem.div_scale:.2f}")
    print(f"直近で意識が強い数字(今学んでいる数字) ➔ 【 {last_learned_digit} 】")
    print(f"---------------------------------------------------------")
    
    if not AUTO_DEMO:
        print("どのイベントを起こしますか? (数字キーを押して Enter)")
        print(" [1] ⚡ カオス・レイジ(ランダム大刺激・位置の攪乱と突然変異)")
        print(f" [2] 🧠 残響ハイパーフォーカス(今学んでいる数字『{last_learned_digit}』をもう一回集中して学ぶ)")
        print(" [3] 🛠️ 通常のパラメータ修正")
        print(" [Enter(空欄)] ⏳ 何もせず通常サイクルを実行")
        choice = input("👉 選択してください: ").strip()
    else:
        # 自動実行用にランダム確率、あるいは一定サイクルごとにイベントを自動発生
        if cycle == 4: choice = "1"
        elif cycle == 8: choice = "2"
        else: choice = ""
    if choice == "1":
        ecosystem.trigger_chaotic_rage()
    elif choice == "2":
        ecosystem.trigger_hyper_focus(last_learned_digit, X_train, y_train)
    elif choice == "3" and not AUTO_DEMO:
        try:
            new_lr_str = input(f" ➔ 新しい学習率スケール (現在: {ecosystem.lr_scale}): ").strip()
            if new_lr_str: ecosystem.lr_scale = float(new_lr_str)
            new_div_str = input(f" ➔ 新しい分裂速度スケール (現在: {ecosystem.div_scale}): ").strip()
            if new_div_str: ecosystem.div_scale = float(new_div_str)
            print("✨ パラメータを変更しました。")
        except ValueError:
            print("❌ 数値変換エラー。通常通り進みます。")
    else:
        print("⏳ 通常サイクル(介入なし)で進みます。")
        
    # --- 通常の計算サイクル実行 ---
    ecosystem.train_cycle(X_train, y_train, phase="day")
    ecosystem.train_cycle(X_train, y_train, phase="evening")
    ecosystem.sleep_cycle(steps=30)
    ecosystem.evolve()
    
    # === 評価 と 「2番目・3番目に自信がなかった数字」の自動判定 ===
    y_pred = []
    ambiguous_digits = [] # 迷った数字(2番目、3番目の候補)をためるリスト
    for idx in range(len(X_test)):
        ensemble_perceptions = []
        ensemble_confidences = []
        
        for model in ecosystem.models:
            perception, _, _ = live_vision_cycle(
                model['space'], x_64d=X_test[idx], 
                sensory_drive=1.0, intrinsic_drive=0.0, learning_gate=0.0
            )
            confidence = np.max(perception)
            ensemble_perceptions.append(perception)
            ensemble_confidences.append(confidence)
            
        weighted_perceptions = np.zeros(NUM_CLASSES)
        total_conf = sum(ensemble_confidences) + 1e-12
        for p, c in zip(ensemble_perceptions, ensemble_confidences):
            weighted_perceptions += p * (c / total_conf)
            
        # 予測結果(1位)を確定
        y_pred.append(np.argmax(weighted_perceptions))
        # --- 🔥 自信のなさをスコアリング ---
        top_indices = np.argsort(weighted_perceptions)[::-1] 
        margin = weighted_perceptions[top_indices[0]] - weighted_perceptions[top_indices[1]]
        
        if margin < 0.15:
            ambiguous_digits.append(top_indices[1])
            ambiguous_digits.append(top_indices[2])
    # === 評価スコアの確定とターゲット選定 ===
    acc = accuracy_score(y_test, y_pred)
    avg_cells = int(np.mean([len(m['space']) for m in ecosystem.models]))
    
    # 🌟 2位・3位の迷い(自信のなさ)を優先して次のハイパーフォーカスターゲットにする
    if len(ambiguous_digits) > 0:
        counts = np.bincount(ambiguous_digits, minlength=10)
        last_learned_digit = np.argmax(counts) # 最も細胞たちを惑わせた「苦手な数字」
        target_reason = "苦手克服モード"
    else:
        counts = np.bincount(y_pred, minlength=10)
        last_learned_digit = np.argmax(counts)
        target_reason = "通常(最多予測)"
        
    # 現在の進捗ログを出力
    print(f"➔ サイクル終了時: 正解率(Acc): {acc:.4f} | 平均細胞数: {avg_cells} | 次回選択理由: {target_reason}")
    
    # === 役割(Role)ごとの集計処理 ===
    roles_count = defaultdict(int)
    for model in ecosystem.models:
        for cell in model['space']:
            roles_count[cell.role] += 1
    print(f"   生存割合 ➔ Explorer: {roles_count['Explorer']//3}, Storage: {roles_count['Storage']//3}, Bridge: {roles_count['Bridge']//3}, Default: {roles_count['Default']//3}")
# =========================================================
# 8. ダッシュボード:3D空間上の細胞役割の可視化
# =========================================================
print("\n[Visualizing Evolved Neural Social Topology...]")
fig = plt.figure(figsize=(14, 6))
# --- サブプロット1: 3Dトポロジー ---
ax1 = fig.add_subplot(121, projection='3d')
ax1.set_title("Evolved Concept Space")
first_space = ecosystem.models[0]['space']
role_colors = {"Explorer": "cyan", "Storage": "orange", "Bridge": "magenta", "Default": "gray"}
role_markers = {"Explorer": "o", "Storage": "s", "Bridge": "^", "Default": "D"}
for role in ["Explorer", "Storage", "Bridge", "Default"]:
    r_cells = [c for c in first_space if c.role == role]
    if not r_cells: continue
    ax1.scatter([c.pos[0] for c in r_cells], [c.pos[1] for c in r_cells], [c.pos[2] for c in r_cells], 
                c=role_colors[role], marker=role_markers[role], s=40, label=role, edgecolors='black', alpha=0.6)
ax1.legend()
# =========================================================
# 9. 混同行列(Confusion Matrix)の描画
# =========================================================
final_y_pred = ecosystem.predict(X_test)
cm = confusion_matrix(y_test, final_y_pred, labels=list(range(10)))
# --- サブプロット2: 混同行列 ---
ax2 = fig.add_subplot(122)
disp = ConfusionMatrixDisplay(confusion_matrix=cm, display_labels=list(range(NUM_CLASSES)))
disp.plot(ax=ax2, cmap='Blues', values_format='d')
ax2.set_title("Final Ecosystem Confusion Matrix")
plt.tight_layout()
plt.show()
    
投稿日:20日前
更新日:19日前
数学の力で現場を変える アルゴリズムエンジニア募集 - Mathlog served by OptHub

この記事を高評価した人

高評価したユーザはいません

この記事に送られたバッジ

バッジはありません。

投稿者

Owl
1
288

コメント

他の人のコメント

コメントはありません。
読み込み中...
読み込み中