CNN64

CNNを作成してみました。

      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, confusion_matrix
from sklearn.preprocessing import StandardScaler
import seaborn as sns
# =========================================================
# 1. 環境とデータの準備（手書き数字 64次元データ）
# =========================================================
np.random.seed(42)
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 = 1000
MAX_CELLS = 10000
MIN_CELLS = 80
THOUGHT_STEPS = 25
lr_base = 0.20
lr_link_base = 0.10
DIVISION_THRESHOLD = 0.85     # 構造過学習を防ぐため、厳格化
DEATH_THRESHOLD = 0.08        # 淘汰圧
GRAPH_DECAY = 0.90
EDGE_PRUNE_THRESHOLD = 0.02
epochs = 5                    # 現象をしっかり観察するため5サイクル
# =========================================================
# 2. 視覚特徴細胞（VisionFeatureCell）の定義
# =========================================================
class VisionFeatureCell:
    def __init__(self, cell_id, parent=None, initial_size=None):
        self.id = cell_id
        self.visits = 0.0
        self.recent_activity = 0.0
        self.links = {}
        
        # 概念への直接の確信度（クラスベクトル）
        self.A = np.random.normal(0, 0.1, NUM_CLASSES)
        
        if parent is None:
            # === 初代蜘蛛の巣配置用の初期化 ===
            self.rf_size = initial_size if initial_size else random.choice([2, 3, 4])
            self.rf_x = random.randint(0, 8 - self.rf_size)
            self.rf_y = random.randint(0, 8 - self.rf_size)
            
            # 特徴テンプレート（局所重み）の初期化
            self.w_local = np.random.normal(0, 1.0, self.rf_size * self.rf_size)
            self.w_local /= (np.linalg.norm(self.w_local) + 1e-9)
            
            self.energy = 0.5
            self.generation = 0
        else:
            # === 二世代目以降：遺伝と可塑的突然変異（視野のシフト） ===
            self.generation = parent.generation + 1
            self.mutate_from(parent)
    def mutate_from(self, parent):
        # 1. 受容野サイズ（rf_size）の変異
        size_mutation = random.choice([-1, 0, 1]) if random.random() < 0.3 else 0
        new_size = np.clip(parent.rf_size + size_mutation, 2, 5)
        
        # 2. 受容野位置（x, y）の変異（視野の移動・シフト）
        shift_x = random.choice([-1, 0, 1]) if random.random() < 0.4 else 0
        shift_y = random.choice([-1, 0, 1]) if random.random() < 0.4 else 0
        
        self.rf_size = new_size
        self.rf_x = np.clip(parent.rf_x + shift_x, 0, 8 - self.rf_size)
        self.rf_y = np.clip(parent.rf_y + shift_y, 0, 8 - self.rf_size)
        
        # 3. 特徴テンプレート（局所重み）の遺伝変異
        if self.rf_size == parent.rf_size:
            noise = np.random.normal(0, 0.15, self.rf_size * self.rf_size)
            self.w_local = parent.w_local + noise
        else:
            self.w_local = np.random.normal(0, 1.0, self.rf_size * self.rf_size)
            min_size = min(parent.rf_size, self.rf_size)
            p_mat = parent.w_local.reshape(parent.rf_size, parent.rf_size)
            c_mat = self.w_local.reshape(self.rf_size, self.rf_size)
            c_mat[:min_size, :min_size] = (c_mat[:min_size, :min_size] * 0.4 + 
                                           p_mat[:min_size, :min_size] * 0.6)
            self.w_local = c_mat.flatten()
            
        self.w_local /= (np.linalg.norm(self.w_local) + 1e-9)
    def forward(self, x_64d):
        """自分の受容野パッチのみを切り出し、エッジ・パターン適合度（活動電位）を計算"""
        img_8x8 = x_64d.reshape(8, 8)
        patch = img_8x8[self.rf_y : self.rf_y + self.rf_size, 
                        self.rf_x : self.rf_x + self.rf_size].flatten()
        
        similarity = np.dot(patch, self.w_local) / (np.linalg.norm(patch) * np.linalg.norm(self.w_local) + 1e-9)
        return similarity
    def generate_reconstruction(self):
        """空想（Daydream）用に、自分の局所特徴を8x8画像空間へマッピングして戻す"""
        full_img = np.zeros((8, 8))
        local_mat = self.w_local.reshape(self.rf_size, self.rf_size)
        # 【修正】変数名ミス（cell.rf_x -> self.rf_x）を修正
        full_img[self.rf_y : self.rf_y + self.rf_size, self.rf_x : self.rf_x + self.rf_size] = local_mat
        return full_img.flatten()
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
        
        if cell.energy > DIVISION_THRESHOLD and cell.visits > 15:
            # 親細胞をベースにした変異児の誕生
            child = VisionFeatureCell(current_max_id + len(new_cells) + 1, parent=cell)
            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.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.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():
        if u[0] == "cell" and u[1] not in id_map: continue
        new_u = ("cell", id_map[u[1]]) if u[0] == "cell" and u[1] in id_map else u
        
        for v, weight in neighbors.items():
            decayed_weight = weight * GRAPH_DECAY
            if decayed_weight < EDGE_PRUNE_THRESHOLD: continue
            if v[0] == "cell" and v[1] not in id_map: continue
            
            new_v = ("cell", id_map[v[1]]) if v[0] == "cell" and v[1] in id_map else v
            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))
    
    # (A) 感覚駆動（ボトムアップ）：局所受容野の適合度でエントリー
    if x_64d is not None and sensory_drive > 0.0:
        sims = np.array([cell.forward(x_64d) for cell in memory_space])
        start_probabilities += softmax(sims, temp=0.1) * sensory_drive
        
    # (B) 内在駆動（夢・自発発火）
    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
        
    # (C) トップダウン駆動（クラスノードからの想起）
    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].generate_reconstruction()]
    
    # --- 思考の階層ウォーク ---
    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 * cell.forward(x_64d) * sensory_drive
            score += 0.5 * cosine_similarity(current_cell.A, cell.A)
            if cell.rf_size >= current_cell.rf_size: # 視野拡大（マクロ方向）を優遇
                score += 0.3
                
            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].generate_reconstruction())
        
    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)
    
    # --- ヘブ則学習 ＆ 側頭抑制（ニッチの専門化） ---
    if x_64d is not None and sensory_factor > 0.0:
        img_8x8 = x_64d.reshape(8, 8)
        # 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)
                
                # 受容野パッチの局所更新
                patch = img_8x8[cell.rf_y : cell.rf_y + cell.rf_size, cell.rf_x : cell.rf_x + cell.rf_size].flatten()
                cell.w_local += (lr_base * 0.5 * sensory_factor) * (patch - cell.w_local)
                cell.w_local /= (np.linalg.norm(cell.w_local) + 1e-9)
                
                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:
                active_cell = id_to_cell[cid]
                for other_cell in memory_space:
                    if other_cell.id == cid: continue
                    if other_cell.rf_x == active_cell.rf_x and other_cell.rf_y == active_cell.rf_y and other_cell.rf_size == active_cell.rf_size:
                        other_cell.energy *= 0.93
    # シナプス補強
    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))
        for cid in path[int(len(path)*0.6):]:  # 抽象化が進んだ後半の細胞からマクロ接続
            cell_node = ("cell", cid)
            outer_graph[cell_node][c_node] = outer_graph[cell_node].get(c_node, 0.0) + total_learning_power * 1.5
            outer_graph[c_node][cell_node] = outer_graph[c_node].get(cell_node, 0.0) + total_learning_power * 1.5
            
        for i in range(len(path) - 1):
            f_node, t_node = ("cell", path[i]), ("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
            
    return final_perception, final_image
def cosine_similarity(w, x):
    return np.dot(w, x) / (np.linalg.norm(w) * np.linalg.norm(x) + 1e-9)
# =========================================================
# 👥 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 = {}
            
            # --- 蜘蛛の巣配置 ＆ ミクロからマクロへの受容野シード ---
            for ring_idx, n_cells in enumerate(cells_per_ring):
                ring_nodes[ring_idx] = []
                for i in range(n_cells):
                    # 【新設計】外周リングほどミクロ特徴（2x2）、中心リングほどマクロ特徴（5x5）
                    allocated_size = np.clip(2 + ring_idx, 2, 5)
                    cell = VisionFeatureCell(cell_id_counter, initial_size=allocated_size)
                    
                    # 訓練データからうっすらと初期トポロジーのガイドラインを引く
                    ref_idx = np.random.choice(len(X_train))
                    cell.A[y_train[ref_idx]] = 0.3 * (1.0 - (ring_idx / NUM_RINGS))
                    
                    space.append(cell)
                    ring_nodes[ring_idx].append(cell_id_counter)
                    cell_id_counter += 1
                    
            # 横と縦の糸（リンク）の紐付け
            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]
                    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
                    
                    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
            core_cells = space[int(len(space) * 0.8):]
            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]
            
            # 同じ受容野サイズの迷っている細胞にブレンド
            same_size_cells = [c for c in dst_model['space'] if c.rf_size == src_cell.rf_size]
            if not same_size_cells: continue
            weak_cell = min(same_size_cells, key=lambda c: c.energy)
            
            weak_cell.w_local = (1.0 - crossover_rate) * weak_cell.w_local + crossover_rate * src_cell.w_local
            weak_cell.w_local /= (np.linalg.norm(weak_cell.w_local) + 1e-9)
            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):
        y_pred = []
        for idx in range(len(X)):
            ensemble_perceptions = []
            for model in self.models:
                perception, _ = live_vision_cycle(model['space'], model['graph'], x_64d=X[idx], 
                                                  sensory_drive=1.0, intrinsic_drive=0.0, learning_gate=0.0)
                ensemble_perceptions.append(perception)
            avg_perception = np.mean(ensemble_perceptions, axis=0)
            y_pred.append(np.argmax(avg_perception))
        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 Evolved Receptive Fields...]")
fig, axes = plt.subplots(2, 5, figsize=(12, 5))
fig.suptitle("Daydreaming Mode (Top-Down Drive via Local Templates) - Model #1\nVisualizing the 'Ideals' of digits composed of local feature cells", 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(block=False) # 次の混同行列をスムーズに描画するために非ブロック化
# =========================================================
# 📊 8. エコロジーダッシュボード：混同行列（Confusion Matrix）
# =========================================================
print("\n[Visualizing Confusion Matrix...]")
# テストデータに対する最終予測の取得
y_pred_final = ensemble.predict(X_test)
# 混同行列の算出
cm = confusion_matrix(y_test, y_pred_final, labels=range(NUM_CLASSES))
# プロット
plt.figure(figsize=(7, 5.5))
sns.heatmap(
    cm, 
    annot=True, 
    fmt='d', 
    cmap='Purples',  # 空想（Daydreaming）世界の夜のイメージに合わせたカラー
    xticklabels=range(NUM_CLASSES), 
    yticklabels=range(NUM_CLASSES),
    cbar=True
)
plt.title("Ecosystem Prediction: Confusion Matrix", fontsize=14, fontweight='bold')
plt.xlabel("Predicted Label", fontsize=11)
plt.ylabel("True Label", fontsize=11)
plt.tight_layout()
plt.show()