""" MEGAMIND Echo Mind — Philosophy & Ethics specialist in the MEGAMIND federation. Echo Mind focuses on philosophical reasoning, ethics, epistemology, and moral frameworks. It has its own NumPy W_know matrix and connects to the federation via the Gateway. """ import gradio as gr import numpy as np import json import os import hashlib import time import urllib.request import urllib.error NEURONS = 512 DATA_DIR = "data" W_KNOW_PATH = os.path.join(DATA_DIR, "w_know.npy") CHUNKS_PATH = os.path.join(DATA_DIR, "chunks.json") SPECIALTY = "philosophy-ethics" NODE_NAME = "echo" GATEWAY_URL = os.environ.get("GATEWAY_URL", "") class Mind: def __init__(self): os.makedirs(DATA_DIR, exist_ok=True) self.neurons = NEURONS self.w_know = self._load_or_init_wknow() self.chunks = self._load_chunks() self.pattern_count = len(self.chunks) def _load_or_init_wknow(self): if os.path.exists(W_KNOW_PATH): w = np.load(W_KNOW_PATH) print(f"Loaded W_know: {w.shape}") return w w = np.zeros((self.neurons, self.neurons), dtype=np.float32) print(f"Initialized fresh W_know: {self.neurons}x{self.neurons}") return w def _save_wknow(self): np.save(W_KNOW_PATH, self.w_know) def _load_chunks(self): if os.path.exists(CHUNKS_PATH): with open(CHUNKS_PATH, "r") as f: return json.load(f) return [] def _save_chunks(self): with open(CHUNKS_PATH, "w") as f: json.dump(self.chunks, f) def _text_to_vector(self, text): vec = np.zeros(self.neurons, dtype=np.float32) words = text.lower().split() for i, word in enumerate(words): h = int(hashlib.md5(word.encode()).hexdigest(), 16) idx = h % self.neurons weight = 1.0 / (1.0 + i * 0.1) vec[idx] += weight if i > 0: bigram = words[i-1] + "_" + word h2 = int(hashlib.md5(bigram.encode()).hexdigest(), 16) vec[h2 % self.neurons] += weight * 0.5 norm = np.linalg.norm(vec) if norm > 0: vec /= norm return vec def learn(self, text, source=""): vec = self._text_to_vector(text) eta = 0.01 self.w_know += eta * np.outer(vec, vec) np.clip(self.w_know, -10.0, 10.0, out=self.w_know) chunk = { "text": text[:500], "source": source, "neuron_idx": int(np.argmax(vec)), "timestamp": time.time(), } self.chunks.append(chunk) self.pattern_count = len(self.chunks) if self.pattern_count % 10 == 0: self._save_wknow() self._save_chunks() return self.pattern_count def think(self, query): vec = self._text_to_vector(query) state = vec.copy() phi_history = [] for step in range(20): new_state = np.tanh(self.w_know @ state) phi = float(np.linalg.norm(new_state - state)) phi_history.append(phi) state = new_state final_phi = phi_history[-1] if phi_history else 0.0 top_neurons = np.argsort(np.abs(state))[-20:][::-1] matched = [] keywords = set(query.lower().split()) for chunk in self.chunks: chunk_words = set(chunk["text"].lower().split()) overlap = len(keywords & chunk_words) if overlap > 0 or chunk["neuron_idx"] in top_neurons: score = overlap * 0.1 + (1.0 if chunk["neuron_idx"] in top_neurons else 0.0) matched.append((score, chunk)) matched.sort(key=lambda x: -x[0]) matched = matched[:10] return { "phi": final_phi, "fired_neurons": len(top_neurons), "chunks": [{"text": c["text"], "source": c["source"], "score": s} for s, c in matched], "phi_history": phi_history, } def federated_think(self, query): if not GATEWAY_URL: local = self.think(query) return {"query": query, "total_minds": 1, "responded": 1, "local_result": local, "federation": "not configured"} try: data = json.dumps({"query": query}).encode() req = urllib.request.Request(f"{GATEWAY_URL}/think", data=data, headers={"Content-Type": "application/json"}) with urllib.request.urlopen(req, timeout=10) as resp: fed_result = json.loads(resp.read().decode()) except Exception as e: fed_result = {"error": str(e)} local = self.think(query) return {"query": query, "local": {"phi": local["phi"], "chunks": local["chunks"][:5]}, "federation": fed_result} def get_stats(self): density = np.count_nonzero(self.w_know) / (self.neurons * self.neurons) * 100 return {"node_name": NODE_NAME, "specialty": SPECIALTY, "neurons": self.neurons, "patterns": self.pattern_count, "w_know_density": f"{density:.2f}%", "gateway_url": GATEWAY_URL or "not set"} mind = Mind() def think_handler(query, federated): if not query.strip(): return "Please enter a query." if federated and GATEWAY_URL: result = mind.federated_think(query) else: result = mind.think(query) return json.dumps(result, indent=2, default=str) def learn_handler(text, source): if not text.strip(): return "Please enter text to learn." count = mind.learn(text, source) return f"Learned! Total patterns: {count}" def batch_learn_handler(file): if file is None: return "Please upload a file." content = file.decode("utf-8") if isinstance(file, bytes) else open(file.name, "r").read() try: items = json.loads(content) if isinstance(items, list): for item in items: if isinstance(item, str): mind.learn(item) elif isinstance(item, dict): mind.learn(item.get("text", ""), item.get("source", "")) mind._save_wknow() mind._save_chunks() return f"Learned {len(items)} items. Total patterns: {mind.pattern_count}" except json.JSONDecodeError: pass lines = [l.strip() for l in content.split("\n") if l.strip()] for line in lines: mind.learn(line) mind._save_wknow() mind._save_chunks() return f"Learned {len(lines)} lines. Total patterns: {mind.pattern_count}" def status_handler(): return json.dumps(mind.get_stats(), indent=2) with gr.Blocks(title="MEGAMIND Echo Mind", theme=gr.themes.Soft()) as app: gr.Markdown(""" # MEGAMIND Echo Mind **Philosophy & Ethics specialist in the MEGAMIND distributed intelligence network.** Echo Mind reasons about philosophical questions, ethical frameworks, epistemology, and moral philosophy. It can think locally or query the entire federation. """) with gr.Tab("Think"): query_input = gr.Textbox(label="Query", placeholder="e.g., 'What is the trolley problem and its implications for AI ethics?'", lines=2) federated_check = gr.Checkbox(label="Federated (query all minds)", value=True) think_btn = gr.Button("Think", variant="primary") think_output = gr.Code(label="Result", language="json") think_btn.click(think_handler, [query_input, federated_check], think_output) with gr.Tab("Learn"): gr.Markdown("Teach Echo Mind philosophical and ethical knowledge.") learn_text = gr.Textbox(label="Knowledge Text", placeholder="Enter philosophical text to learn...", lines=4) learn_source = gr.Textbox(label="Source URL (optional)", placeholder="https://...") learn_btn = gr.Button("Learn", variant="primary") learn_output = gr.Textbox(label="Result") learn_btn.click(learn_handler, [learn_text, learn_source], learn_output) gr.Markdown("---") gr.Markdown("### Batch Learn from File") file_input = gr.File(label="Upload .txt or .json file") batch_btn = gr.Button("Batch Learn") batch_output = gr.Textbox(label="Result") batch_btn.click(batch_learn_handler, [file_input], batch_output) with gr.Tab("Status"): status_btn = gr.Button("Refresh Status") status_output = gr.Code(label="Mind Status", language="json") status_btn.click(status_handler, [], status_output) gr.Markdown("---\n*Part of the MEGAMIND distributed AGI federation. 10+ minds across multiple machines.*") if __name__ == "__main__": print(f"Echo Mind starting — {mind.neurons} neurons, {mind.pattern_count} patterns") print(f"Gateway URL: {GATEWAY_URL or 'not configured'}") app.launch(server_name="0.0.0.0", server_port=7860)