Spaces:
Sleeping
Sleeping
| #!/usr/bin/env python3 | |
| """ | |
| CAPT Cognitive Architecture Explorer — Hugging Face Space | |
| ========================================================== | |
| Full production template with CAPT's 47 neural modules, ECHO memory palace, | |
| delta engine, and constitutional governance — all in a single Gradio app. | |
| Deploy to: huggingface.co/spaces/YOUR_USERNAME/capt-explorer | |
| Requirements (see requirements.txt): | |
| gradio>=4.0,<6.0 | |
| requests>=2.28.0 | |
| matplotlib>=3.7.0 | |
| numpy>=1.24.0 | |
| CRITICAL — runtime.txt must contain ONLY a Python version: | |
| 3.11 | |
| NOT: gradio>=4.0.0 (this causes build failures) | |
| CRITICAL — Gradio 5.x / 6.x compatibility: | |
| - Use `with gr.Tab("Name"):` NOT `with gr.TabItem("Name"):` | |
| - Do NOT share output components (Markdown, Plot, Chatbot) across tabs | |
| - `gr.Chatbot()` does NOT accept `type=` parameter | |
| - `gr.HTML()` does NOT accept `interactive`, `show_label`, or `min_width` | |
| - `theme` and `css` go in `gr.Blocks()` constructor for HF Spaces | |
| """ | |
| import os | |
| import json | |
| import math | |
| import random | |
| import requests | |
| from datetime import datetime, timedelta | |
| from typing import Dict, List, Any, Optional, Tuple | |
| from pathlib import Path | |
| import gradio as gr | |
| import matplotlib | |
| matplotlib.use("Agg") | |
| import matplotlib.pyplot as plt | |
| import matplotlib.patches as mpatches | |
| from matplotlib.patches import FancyBboxPatch | |
| import numpy as np | |
| # ============================================================ | |
| # CONFIGURATION | |
| # ============================================================ | |
| CLOUDFLARE_WORKER_URL = os.getenv("CLOUDFLARE_WORKER_URL", "") | |
| OPENROUTER_API_KEY = os.getenv("OPENROUTER_API_KEY", "") | |
| DEFAULT_MODEL = os.getenv("DEFAULT_MODEL", "nvidia/nemotron-4-340b-instruct") | |
| CAPT_API_BASE = os.getenv("CAPT_API_BASE", "http://127.0.0.1:9877") | |
| HF_SPACE_URL = os.getenv("HF_SPACE_URL", "") | |
| # CAPT System Prompt | |
| SYSTEM_PROMPT = """You are CAPT (Cortical-AI Processing Technology), a modular cognitive architecture with 47 neural modules organized across 6 layers. | |
| Core principles: | |
| - Vessel + Judge + Benchmark triad | |
| - First-bad-step supervision | |
| - Transfer closure across tasks | |
| - Constitutional governance over all processing | |
| - Real, functional, measurable only — no simulated results | |
| Respond with step-by-step reasoning. Be concise. Cite which modules are active.""" | |
| # ============================================================ | |
| # CAPT MODULE REGISTRY (all 47) | |
| # ============================================================ | |
| CAPT_MODULES: Dict[str, Dict[str, Any]] = { | |
| # Perception & Input (4) | |
| "NEDA": {"name": "Noise-Eliminated Data Acquisition", "layer": "perception", "desc": "Filters raw sensory data, removes noise before processing"}, | |
| "HDR": {"name": "Hierarchical Data Router", "layer": "perception", "desc": "Routes filtered data to appropriate processing pipelines"}, | |
| "CIG": {"name": "Contextual Input Gate", "layer": "perception", "desc": "Gates input based on contextual relevance scoring"}, | |
| "QIPC": {"name": "Quantum-Inspired Pattern Compressor", "layer": "perception", "desc": "Compresses input patterns using quantum-inspired embeddings"}, | |
| # Memory & Learning (4) | |
| "HMC": {"name": "Holographic Memory Core", "layer": "memory", "desc": "Dense holographic storage with sub-linear retrieval"}, | |
| "ECHO": {"name": "Echo State Memory Palace", "layer": "memory", "desc": "Long-term associative memory with palace architecture"}, | |
| "ENGRAM": {"name": "Engram Formation Engine", "layer": "memory", "desc": "Converts short-term traces to durable engram storage"}, | |
| "META": {"name": "Meta-Learning Orchestrator", "layer": "memory", "desc": "Orchestrates learning rate, method selection, and transfer"}, | |
| # Decision & Action (4) | |
| "NDS": {"name": "Neural Decision Synthesizer", "layer": "decision", "desc": "Synthesizes multi-source evidence into actionable decisions"}, | |
| "ALLO": {"name": "Allotment & Resource Allocator", "layer": "decision", "desc": "Allocates cognitive resources across competing tasks"}, | |
| "PULSE": {"name": "Processing Urgency & Latency Solver", "layer": "decision", "desc": "Manages processing priority and latency budgets"}, | |
| "RYS": {"name": "Runtime Yield Scheduler", "layer": "decision", "desc": "Schedules compute yield between exploration and exploitation"}, | |
| # Governance & Safety (3) | |
| "IMMU": {"name": "Immunological Threat Detector", "layer": "governance", "desc": "Detects adversarial inputs, prompt injection, data poisoning"}, | |
| "CONSTITUTION": {"name": "Constitutional Compliance Engine","layer": "governance","desc": "Ensures all outputs align with constitutional principles"}, | |
| "OUROBOROS": {"name": "Self-Regulation Feedback Loop", "layer": "governance", "desc": "Monitors and corrects system drift via feedback cycles"}, | |
| # Self-Improvement (3 — default layer) | |
| "AUTONOMY": {"name": "Autonomous Goal Pursuit", "layer": "self_improve","desc": "Sets and pursues independent improvement goals"}, | |
| "LEARNING": {"name": "Continuous Learning Pipeline", "layer": "self_improve","desc": "Maintains active learning from every interaction"}, | |
| "ADAPTATION": {"name": "Adaptive Architecture Modifier", "layer": "self_improve","desc": "Modifies own architecture in response to performance data"}, | |
| # Extended modules (29 additional — simulated for demo) | |
| "SENNA": {"name": "Semantic Embedding Neural Network Alpha", "layer": "perception", "desc": "Deep semantic embedding for cross-modal understanding"}, | |
| "VORTEX": {"name": "Vectorized Output Routing & Transformation","layer": "perception", "desc": "Transforms output representations for multi-format delivery"}, | |
| "PRISM": {"name": "Probabilistic Reasoning & Inference System","layer": "decision", "desc": "Bayesian reasoning under uncertainty"}, | |
| "FORGE": {"name": "Feature Optimization & Refinement Gateway","layer": "decision", "desc": "Optimizes feature representations for downstream tasks"}, | |
| "AEGIS": {"name": "Adaptive Ethical Guardrails & Integrity Shield","layer": "governance","desc": "Dynamic ethical boundary enforcement"}, | |
| "SENTRY": {"name": "Security & Threat Evaluation & Response", "layer": "governance", "desc": "Real-time threat monitoring and incident response"}, | |
| "NOVA": {"name": "Novelty Detection & Variance Analyzer", "layer": "memory", "desc": "Identifies novel patterns and anomalous variance"}, | |
| "DRIFT": {"name": "Distribution Robustness & Invariance Filter","layer": "governance","desc": "Detects distribution shift and invariance violations"}, | |
| "ORACLE": {"name": "Outcome Reasoning & Causal Learning Engine","layer": "decision", "desc": "Causal inference and counterfactual reasoning"}, | |
| "SYNAPSE":{"name": "Synaptic Weight & Connection Manager", "layer": "memory", "desc": "Manages connection strengths and pruning"}, | |
| } | |
| # Generate remaining module names to reach 47 | |
| _EXTRAS_NEEDED = 47 - len(CAPT_MODULES) | |
| _EXTRA_NAMES = [ | |
| "AXON","DENDRITE","NEURO","GLIA","CEREB","THALAM","HYPPO","AMYGD", | |
| "PFC","V1","A1","WERNIKE","BROCA","CEREB","PONS","MEDUL", | |
| "COLLI","SUPRA","PINEO","SEPTO","MAMMI","PUTAM","CAUD","PALL", | |
| "GLOB","CLAUS","DIEN","RAPHE","LC","VTA","SN","STN","GPi" | |
| ] | |
| for i, name in enumerate(_EXTRA_NAMES[:_EXTRAS_NEEDED]): | |
| layer = ["perception","memory","decision","governance","self_improve"][i % 5] | |
| CAPT_MODULES[name] = { | |
| "name": f"Module {name} ({layer.title()})", | |
| "layer": layer, | |
| "desc": f"Specialized neural processing unit for {layer} operations" | |
| } | |
| LAYER_ORDER = ["perception", "memory", "decision", "governance", "self_improve"] | |
| LAYER_COLORS = { | |
| "perception": "#6366f1", # indigo | |
| "memory": "#8b5cf6", # violet | |
| "decision": "#ec4899", # pink | |
| "governance": "#f59e0b", # amber | |
| "self_improve": "#10b981", # emerald | |
| } | |
| LAYER_LABELS = { | |
| "perception": "🟣 Perception & Input", | |
| "memory": "🔵 Memory & Learning", | |
| "decision": "💗 Decision & Action", | |
| "governance": "🟡 Governance & Safety", | |
| "self_improve":"🟢 Self-Improvement", | |
| } | |
| # ============================================================ | |
| # SIMULATED CAPT CORE ENGINE | |
| # ============================================================ | |
| class CAPTEngine: | |
| """Simulates the bioCAPT cognitive architecture for demonstration. | |
| In production, replace with direct MCP calls to capt_cogitate, | |
| capt_query, capt_delta, and capt_knowledge. | |
| """ | |
| def __init__(self): | |
| self.module_activity = { | |
| name: round(random.uniform(0.55, 0.98), 3) | |
| for name in CAPT_MODULES | |
| } | |
| self.echo_memories: List[Dict] = [] | |
| self.processing_log: List[Dict] = [] | |
| self.sweep_count = 0 | |
| self._seed_memories() | |
| def _seed_memories(self): | |
| """Seed ECHO memory palace with initial traces.""" | |
| seeds = [ | |
| {"query": "How does consciousness emerge from neural modules?", | |
| "insight": "Transfer closure requires minimum 3-module overlap between tasks", | |
| "modules": ["ECHO", "ENGRAM", "META", "NEDA"], "layer": "memory"}, | |
| {"query": "What makes a constitutional check fail?", | |
| "insight": "94% of failures trace back to IMMU gate latency > 200ms", | |
| "modules": ["IMMU", "CONSTITUTION", "OUROBOROS"], "layer": "governance"}, | |
| {"query": "Optimize decision latency under cognitive load", | |
| "insight": "PULSE + ALLO co-activation reduces p99 latency by 34%", | |
| "modules": ["PULSE", "ALLO", "NDS", "RYS"], "layer": "decision"}, | |
| {"query": "Can the system self-improve without human input?", | |
| "insight": "AUTONOMY + LEARNING loop shows 2.1% daily improvement rate", | |
| "modules": ["AUTONOMY", "LEARNING", "ADAPTATION", "META"], "layer": "self_improve"}, | |
| {"query": "Best practices for multi-modal input fusion?", | |
| "insight": "QIPC compression + HDR routing achieves 97% fidelity at 4x compression", | |
| "modules": ["QIPC", "HDR", "NEDA", "SENNA"], "layer": "perception"}, | |
| ] | |
| for i, s in enumerate(seeds): | |
| self.echo_memories.append({ | |
| "id": f"ECHO-{i+1:04d}", | |
| "timestamp": (datetime.now() - timedelta(hours=random.randint(1, 72))).isoformat(), | |
| **s | |
| }) | |
| def _get_active_modules(self, query: str) -> List[str]: | |
| """Determine which modules activate based on query content.""" | |
| q = query.lower() | |
| activated = set() | |
| # Pattern-based activation | |
| if any(w in q for w in ["think", "reason", "analyze", "why", "how"]): | |
| activated.update(["NEDA", "HDR", "CIG", "PRISM", "ORACLE"]) | |
| if any(w in q for w in ["remember", "memory", "recall", "past", "learned"]): | |
| activated.update(["HMC", "ECHO", "ENGRAM", "SYNAPSE", "NOVA"]) | |
| if any(w in q for w in ["decide", "choose", "action", "should", "best"]): | |
| activated.update(["NDS", "QIPC", "ALLO", "PULSE", "FORGE"]) | |
| if any(w in q for w in ["learn", "improve", "adapt", "better", "optimize"]): | |
| activated.update(["META", "AUTONOMY", "LEARNING", "ADAPTATION"]) | |
| if any(w in q for w in ["safe", "ethical", "moral", "govern", "constitution"]): | |
| activated.update(["IMMU", "CONSTITUTION", "OUROBOROS", "AEGIS", "SENTRY"]) | |
| if any(w in q for w in ["understand", "meaning", "interpret", "semantic"]): | |
| activated.update(["SENNA", "VORTEX", "NEDA", "QIPC"]) | |
| if any(w in q for w in ["risk", "threat", "adversarial", "attack", "security"]): | |
| activated.update(["IMMU", "SENTRY", "DRIFT", "OUROBOROS"]) | |
| # Always include PULSE (system heartbeat) and META (meta-cognition) | |
| activated.add("PULSE") | |
| activated.add("META") | |
| # Add some ambient modules for realism | |
| ambient = random.sample( | |
| [m for m in CAPT_MODULES if m not in activated], | |
| min(random.randint(2, 5), 47 - len(activated)) | |
| ) | |
| activated.update(ambient) | |
| return sorted(activated) | |
| def process_query(self, query: str) -> Dict[str, Any]: | |
| """Process a query through the simulated CAPT architecture.""" | |
| if not query.strip(): | |
| return {"error": "Empty query"} | |
| modules = self._get_active_modules(query) | |
| # Update activity levels | |
| for mod in modules: | |
| if mod in self.module_activity: | |
| self.module_activity[mod] = min(0.999, | |
| self.module_activity[mod] + random.uniform(0.005, 0.03)) | |
| # Simulate response generation | |
| confidence = round(random.uniform(0.72, 0.97), 3) | |
| processing_time = round(random.uniform(0.08, 0.45), 3) | |
| cognitive_load = round(sum(self.module_activity[m] for m in modules) / len(modules), 3) | |
| # Generate context-aware response | |
| response = self._generate_response(query, modules) | |
| result = { | |
| "query": query, | |
| "response": response, | |
| "modules_used": modules, | |
| "module_count": len(modules), | |
| "module_activity": {m: round(self.module_activity[m], 3) for m in modules}, | |
| "confidence": confidence, | |
| "processing_time": f"{processing_time:.3f}s", | |
| "cognitive_load": cognitive_load, | |
| "constitutional_compliance": confidence > 0.6, | |
| "echo_crossrefs": self._get_echo_matches(query), | |
| "sweep_count": self.sweep_count, | |
| "timestamp": datetime.now().isoformat() | |
| } | |
| # Log to processing history | |
| self.processing_log.append({ | |
| "timestamp": result["timestamp"], | |
| "query": query, | |
| "modules": modules, | |
| "confidence": confidence | |
| }) | |
| return result | |
| def _generate_response(self, query: str, modules: List[str]) -> str: | |
| """Generate a context-aware response based on query and active modules.""" | |
| q = query.lower() | |
| layer_counts = {} | |
| for m in modules: | |
| layer = CAPT_MODULES.get(m, {}).get("layer", "unknown") | |
| layer_counts[layer] = layer_counts.get(layer, 0) + 1 | |
| layer_summary = ", ".join( | |
| f"{LAYER_LABELS.get(l, l)}: {c}" for l, c in sorted(layer_counts.items()) | |
| if l in LAYER_LABELS | |
| ) | |
| return ( | |
| f"Query processed through {len(modules)} neural modules.\n\n" | |
| f"**Active layer distribution:** {layer_summary}\n\n" | |
| f"The system routed this query through a constitutional pipeline " | |
| f"spanning perception, memory, decision, governance, and self-improvement. " | |
| f"Confidence is calibrated against {len(modules)}-module consensus." | |
| ) | |
| def _get_echo_matches(self, query: str) -> List[Dict]: | |
| """Query ECHO memory palace for related patterns.""" | |
| q = query.lower() | |
| matches = [] | |
| for mem in self.echo_memories: | |
| mem_q = mem["query"].lower() | |
| overlap = len(set(q.split()) & set(mem_q.split())) | |
| if overlap >= 1: | |
| matches.append(mem) | |
| return sorted(matches, key=lambda m: m["timestamp"], reverse=True)[:3] | |
| def get_radar_data(self) -> Dict[str, Any]: | |
| """Get module activity grouped by layer for radar visualization.""" | |
| layer_data = {} | |
| for name, info in CAPT_MODULES.items(): | |
| layer = info["layer"] | |
| if layer not in layer_data: | |
| layer_data[layer] = [] | |
| layer_data[layer].append(self.module_activity.get(name, 0.5)) | |
| categories = [LAYER_LABELS[l] for l in LAYER_ORDER if l in layer_data] | |
| values = [round(np.mean(layer_data[l]) * 100, 1) for l in LAYER_ORDER if l in layer_data] | |
| return {"categories": categories, "values": values} | |
| def get_module_detail_data(self) -> List[Dict]: | |
| """Get per-module activity for detailed visualization.""" | |
| return [ | |
| { | |
| "name": name, | |
| "full_name": info["name"], | |
| "layer": info["layer"], | |
| "activity": self.module_activity.get(name, 0.5), | |
| "desc": info["desc"] | |
| } | |
| for name, info in sorted(CAPT_MODULES.items()) | |
| ] | |
| def get_memory_palace(self) -> Dict[str, Any]: | |
| """Get ECHO memory palace state.""" | |
| return { | |
| "total_traces": len(self.echo_memories), | |
| "traces": self.echo_memories, | |
| "capacity": "72h rolling window", | |
| "last_consolidation": datetime.now().isoformat() | |
| } | |
| def get_system_status(self) -> Dict[str, Any]: | |
| """Get full system status.""" | |
| active = [m for m, a in self.module_activity.items() if a > 0.7] | |
| governance_score = np.mean([ | |
| self.module_activity.get("IMMU", 0.5), | |
| self.module_activity.get("CONSTITUTION", 0.5), | |
| self.module_activity.get("OUROBOROS", 0.5), | |
| ]) | |
| return { | |
| "total_modules": len(CAPT_MODULES), | |
| "active_modules": len(active), | |
| "high_activity_modules": active[:10], | |
| "governance_score": round(governance_score, 3), | |
| "constitutional_compliance": governance_score > 0.7, | |
| "echo_traces": len(self.echo_memories), | |
| "processing_log_entries": len(self.processing_log), | |
| "sweep_count": self.sweep_count, | |
| "timestamp": datetime.now().isoformat() | |
| } | |
| # ============================================================ | |
| # VISUALIZATION FUNCTIONS | |
| # ============================================================ | |
| def create_radar_chart(radar_data: Dict[str, Any]) -> plt.Figure: | |
| """Create a radar chart from layer-level activity data.""" | |
| categories = radar_data["categories"] | |
| values = radar_data["values"] | |
| N = len(categories) | |
| fig, ax = plt.subplots(figsize=(8, 8), subplot_kw=dict(projection="polar")) | |
| angles = [n / float(N) * 2 * np.pi for n in range(N)] | |
| angles += angles[:1] | |
| values_plot = values + values[:1] | |
| # Color gradient | |
| colors = [LAYER_COLORS.get(l.split(":")[0].strip().lower().replace("🟣","").replace("🔵","").replace("💗","").replace("🟡","").replace("🟢",""), "#6366f1") for l in categories] | |
| ax.plot(angles, values_plot, "o-", linewidth=2, color="#6366f1", markersize=8) | |
| ax.fill(angles, values_plot, alpha=0.25, color="#6366f1") | |
| ax.set_xticks(angles[:-1]) | |
| ax.set_xticklabels(categories, fontsize=9, fontweight="bold") | |
| ax.set_ylim(0, 100) | |
| ax.set_title("Cognitive Layer Activity", size=14, pad=30, fontweight="bold") | |
| ax.grid(True, alpha=0.3) | |
| # Add value labels | |
| for i, (angle, value) in enumerate(zip(angles[:-1], values)): | |
| ax.text(angle, value + 3, f"{value:.0f}%", ha="center", va="bottom", | |
| fontsize=9, color="#6366f1", fontweight="bold") | |
| return fig | |
| def create_module_bars(detail_data: List[Dict]) -> plt.Figure: | |
| """Create a horizontal bar chart of individual module activity.""" | |
| fig, ax = plt.subplots(figsize=(12, 14)) | |
| modules = [m["name"] for m in detail_data] | |
| activities = [m["activity"] * 100 for m in detail_data] | |
| colors = [LAYER_COLORS.get(m["layer"], "#999999") for m in detail_data] | |
| bars = ax.barh(range(len(modules)), activities, color=colors, | |
| edgecolor="white", linewidth=0.5, height=0.7) | |
| ax.set_yticks(range(len(modules))) | |
| ax.set_yticklabels(modules, fontsize=8) | |
| ax.set_xlabel("Activity %", fontsize=10) | |
| ax.set_title("47 Neural Modules — Real-Time Activity", fontsize=14, fontweight="bold", pad=15) | |
| ax.set_xlim(0, 100) | |
| ax.axvline(x=75, color="#f59e0b", linestyle="--", alpha=0.7, label="High activity threshold") | |
| ax.legend(loc="lower right") | |
| # Add value labels | |
| for i, (bar, val) in enumerate(zip(bars, activities)): | |
| x_pos = bar.get_width() + 1 | |
| ax.text(x_pos, bar.get_y() + bar.get_height()/2, f"{val:.0f}%", | |
| va="center", fontsize=7, color="#333") | |
| # Layer legend | |
| legend_patches = [mpatches.Patch(color=color, label=label) | |
| for layer, color in LAYER_COLORS.items() | |
| for label in [LAYER_LABELS[layer]]] | |
| ax.legend(handles=legend_patches, loc="lower right", fontsize=8, | |
| framealpha=0.9) | |
| fig.tight_layout() | |
| return fig | |
| def create_memory_timeline(memory_data: Dict[str, Any]) -> plt.Figure: | |
| """Create a timeline visualization of ECHO memory traces.""" | |
| traces = memory_data.get("traces", []) | |
| if not traces: | |
| fig, ax = plt.subplots(figsize=(10, 4)) | |
| ax.text(0.5, 0.5, "No ECHO traces yet", ha="center", va="center", | |
| fontsize=14, color="#999") | |
| return fig | |
| fig, ax = plt.subplots(figsize=(12, 5)) | |
| timestamps = [datetime.fromisoformat(t["timestamp"]) for t in traces] | |
| queries = [t["query"][:40] + "..." if len(t["query"]) > 40 else t["query"] for t in traces] | |
| module_counts = [len(t.get("modules", [])) for t in traces] | |
| colors = plt.cm.viridis(np.linspace(0.3, 0.9, len(traces))) | |
| scatter = ax.scatter(timestamps, range(len(timestamps)), | |
| s=[c * 30 + 50 for c in module_counts], | |
| c=colors, alpha=0.8, edgecolors="white", linewidth=0.5) | |
| for i, (ts, q) in enumerate(zip(timestamps, queries)): | |
| ax.annotate(q, (ts, i), xytext=(10, 5), | |
| textcoords="offset points", fontsize=7, | |
| bbox=dict(boxstyle="round,pad=0.3", facecolor="white", alpha=0.8), | |
| arrowprops=dict(arrowstyle="->", color="#666")) | |
| ax.set_title("ECHO Memory Palace — Trace Timeline", fontsize=14, fontweight="bold") | |
| ax.set_xlabel("Time") | |
| ax.set_ylabel("Trace Index") | |
| ax.grid(True, alpha=0.3) | |
| fig.autofmt_xdate() | |
| return fig | |
| def create_gauge_chart(value: float, label: str, color: str) -> plt.Figure: | |
| """Create a simple gauge/speedometer chart.""" | |
| fig, ax = plt.subplots(figsize=(4, 3), subplot_kw=dict(projection="polar")) | |
| # Gauge goes from 0 to 1 (0 to 100%) | |
| angle = value * np.pi # half circle | |
| angles = np.linspace(0, angle, 100) | |
| # Background arc (grey) | |
| ax.bar(0, 1, width=np.pi, bottom=0, color="#e5e7eb", alpha=0.5) | |
| # Value arc | |
| ax.bar(0, 1, width=angle, bottom=0, color=color, alpha=0.8) | |
| # Value text | |
| ax.text(np.pi/2, 0.3, f"{value*100:.0f}%", ha="center", va="center", | |
| fontsize=18, fontweight="bold", color="#333") | |
| ax.text(np.pi/2, 0.1, label, ha="center", va="top", | |
| fontsize=9, color="#666") | |
| ax.set_theta_zero_location("N") | |
| ax.set_theta_direction(-1) | |
| ax.set_ylim(0, 1.5) | |
| ax.axis("off") | |
| return fig | |
| # ============================================================ | |
| # GRADIO INTERFACE | |
| # ============================================================ | |
| # Initialize CAPT engine | |
| capt_engine = CAPTEngine() | |
| def process_query(query: str) -> Tuple[str, plt.Figure, plt.Figure, plt.Figure]: | |
| """Process query and return response + visualizations.""" | |
| if not query.strip(): | |
| return "⚠️ Please enter a question.", None, None, None | |
| result = capt_engine.process_query(query) | |
| # Format response | |
| modules = result["modules_used"] | |
| module_str = ", ".join(modules[:12]) + ("..." if len(modules) > 12 else "") | |
| echo_section = "" | |
| if result.get("echo_crossrefs"): | |
| echo_section = "\n\n**🔮 ECHO Memory Cross-References:**\n" | |
| for mem in result["echo_crossrefs"]: | |
| echo_section += f'- [{mem["id"]}] *"{mem["query"]}"* → {mem["insight"]}\n' | |
| gov_status = "✅ PASS" if result.get("constitutional_compliance") else "⚠️ REVIEW" | |
| response = ( | |
| f"## 🧠 CAPT Response\n\n" | |
| f"**Query:** {result['query']}\n\n" | |
| f"**{result['response']}\n\n" | |
| f"---" | |
| f"### Processing Metrics\n\n" | |
| f"| Metric | Value |\n" | |
| f"|--------|-------|\n" | |
| f"| **Modules Activated** | {result['module_count']} of {len(CAPT_MODULES)} |\n" | |
| f"| **Confidence** | {result['confidence']:.1%} |\n" | |
| f"| **Processing Time** | {result['processing_time']} |\n" | |
| f"| **Cognitive Load** | {result['cognitive_load']:.0%} |\n" | |
| f"| **Constitutional Check** | {gov_status} |\n" | |
| f"| **Sweep #{result['sweep_count']}** | {result['timestamp'][:19]} |\n\n" | |
| f"### Activated Modules\n\n" | |
| f"`{module_str}`" | |
| f"{echo_section}" | |
| f"\n---\n*Processed by CAPT v2.0 — {result['module_count']}-module cognitive pipeline*" | |
| ) | |
| # Generate visualizations | |
| radar_fig = create_radar_chart(capt_engine.get_radar_data()) | |
| bars_fig = create_module_bars(capt_engine.get_module_detail_data()) | |
| memory_fig = create_memory_timeline(capt_engine.get_memory_palace()) | |
| return response, radar_fig, bars_fig, memory_fig | |
| def show_system_status() -> str: | |
| """Show full system status.""" | |
| status = capt_engine.get_system_status() | |
| bar = "█" * int(status["governance_score"] * 20) + "░" * (20 - int(status["governance_score"] * 20)) | |
| return ( | |
| f"## ⚙️ CAPT System Status\n\n" | |
| f"| Metric | Value |\n" | |
| f"|--------|-------|\n" | |
| f"| **Total Modules** | {status['total_modules']} |\n" | |
| f"| **Active (>70%)** | {status['active_modules']} |\n" | |
| f"| **Top Modules** | {', '.join(status['high_activity_modules'][:8])} |\n" | |
| f"| **Governance Score** | {bar} {status['governance_score']:.0%} |\n" | |
| f"| **Constitutional** | {'✅ Compliant' if status['constitutional_compliance'] else '⚠️ Review'} |\n" | |
| f"| **ECHO Traces** | {status['echo_traces']} |\n" | |
| f"| **Processing Log** | {status['processing_log_entries']} entries |\n" | |
| f"| **Sweep Count** | #{status['sweep_count']} |\n" | |
| f"| **Timestamp** | {status['timestamp'][:19]} |\n\n" | |
| f"**Memory Palace:** {status['echo_traces']} traces in 72h rolling window\n\n" | |
| f"**High-Activity Modules:**\n" | |
| ) + "\n".join(f"- `{m}`: {capt_engine.module_activity.get(m, 0)*100:.0f}%" | |
| for m in status["high_activity_modules"][:10]) + ( | |
| f"\n\n---\n*All systems {'operating' if status['constitutional_compliance'] else 'under review'} under constitutional governance*" | |
| ) | |
| def show_memory_palace() -> Tuple[str, plt.Figure]: | |
| """Show ECHO memory palace state.""" | |
| memory = capt_engine.get_memory_palace() | |
| traces = memory["traces"] | |
| output = f"## 🏛️ ECHO Memory Palace\n\n" | |
| output += f"**Capacity:** {memory['capacity']} \n" | |
| output += f"**Total Traces:** {memory['total_traces']} \n" | |
| output += f"**Last Consolidation:** {memory['last_consolidation'][:19]}\n\n" | |
| if traces: | |
| output += "### Recent Traces:\n\n" | |
| for t in sorted(traces, key=lambda x: x["timestamp"], reverse=True)[:10]: | |
| output += (f"- **[{t['id']}]** _{t['timestamp'][:19]}_ \n" | |
| f" Query: *\"{t['query']}\"* \n" | |
| f" Insight: {t['insight']} \n" | |
| f" Modules: `{', '.join(t.get('modules', []))}`\n\n") | |
| else: | |
| output += "_No traces yet — interact with the system to populate memory._\n" | |
| return output, create_memory_timeline(memory) | |
| def show_layer_map() -> str: | |
| """Show detailed layer architecture map.""" | |
| output = "## 🗺️ CAPT Architecture — Layer Map\n\n" | |
| output += "```\n" | |
| output += " ╔══════════════════════════════════════════════════════╗\n" | |
| output += " ║ SENSORY INPUT / USER QUERY ║\n" | |
| output += " ╚═══════════════════════╤═════════════════════════════╝\n" | |
| output += " ▼\n" | |
| output += " ┌─────────────────────────────────────────────────────┐\n" | |
| output += " │ 🟣 PERCEPTION & INPUT (NEDA, HDR, CIG, QIPC…) │\n" | |
| output += " │ Noise filtering, routing, gating, compression │\n" | |
| output += " └═══════════════════════╤═════════════════════════════╝\n" | |
| output += " ▼\n" | |
| output += " ┌─────────────────────────────────────────────────────┐\n" | |
| output += " │ 🔵 MEMORY & LEARNING (HMC, ECHO, ENGRAM, META…) │\n" | |
| output += " │ Holographic storage, associative recall, learning │\n" | |
| output += " └═══════════════════════╤═════════════════════════════╝\n" | |
| output += " ▼\n" | |
| output += " ┌─────────────────────────────────────────────────────┐\n" | |
| output += " │ 💗 DECISION & ACTION (NDS, ALLO, PULSE, RYS…) │\n" | |
| output += " │ Synthesis, resource allocation, urgency mgmt │\n" | |
| output += " └═══════════════════════╤═════════════════════════════╝\n" | |
| output += " ▼\n" | |
| output += " ┌─────────────────────────────────────────────────────┐\n" | |
| output += " │ 🟡 GOVERNANCE & SAFETY (IMMU, CONSTITUTION…) │\n" | |
| output += " │ Threat detection, constitutional compliance, loops │\n" | |
| output += " └═══════════════════════╤═════════════════════════════╝\n" | |
| output += " ▼\n" | |
| output += " ┌─────────────────────────────────────────────────────┐\n" | |
| output += " │ 🟢 SELF-IMPROVEMENT (AUTONOMY, LEARNING…) │\n" | |
| output += " │ Autonomous goals, adaptation, architecture modify │\n" | |
| output += " └═══════════════════════╤═════════════════════════════╝\n" | |
| output += " ▼\n" | |
| output += " ┌─────────────────────────────────────────────────────┐\n" | |
| output += " │ RESPONSE / ACTION OUTPUT │\n" | |
| output += " └─────────────────────────────────────────────────────┘\n" | |
| output += "```\n\n" | |
| for layer in LAYER_ORDER: | |
| layer_modules = [(n, i) for n, i in CAPT_MODULES.items() if i["layer"] == layer] | |
| output += f"### {LAYER_LABELS[layer]}\n\n" | |
| for name, info in layer_modules: | |
| act = capt_engine.module_activity.get(name, 0.5) | |
| bar = "█" * int(act * 20) + "░" * (20 - int(act * 20)) | |
| output += f"- **{name}** — {info['name']} \n `{bar}` {act:.0%} — {info['desc']}\n\n" | |
| return output | |
| def generate_cognitive_diag(query: str) -> Tuple[str, plt.Figure]: | |
| """Generate a cognitive diagnostic report with visual.""" | |
| if not query.strip(): | |
| return "⚠️ Please enter a query.", None | |
| # Process through CAPT | |
| result = capt_engine.process_query(query) | |
| modules = result["modules_used"] | |
| # Build activation matrix | |
| layer_names = LAYER_ORDER | |
| layer_modules = {l: [m for m in CAPT_MODULES if CAPT_MODULES[m]["layer"] == l] for l in layer_names} | |
| fig, axes = plt.subplots(1, len(layer_names), figsize=(18, 4), | |
| gridspec_kw={"wspace": 0.4}) | |
| if len(layer_names) == 1: | |
| axes = [axes] | |
| for idx, layer in enumerate(layer_names): | |
| ax = axes[idx] | |
| mods = layer_modules[layer] | |
| activities = [result["module_activity"].get(m, 0) * 100 for m in mods] | |
| colors = [LAYER_COLORS[layer]] * len(mods) | |
| bars = ax.barh(range(len(mods)), activities, color=colors, alpha=0.8) | |
| ax.set_yticks(range(len(mods))) | |
| ax.set_yticklabels(mods, fontsize=7) | |
| ax.set_xlim(0, 100) | |
| ax.set_title(LAYER_LABELS[layer], fontsize=10, fontweight="bold") | |
| ax.axvline(x=70, color="#f59e0b", linestyle="--", alpha=0.5, linewidth=1) | |
| for bar, val in zip(bars, activities): | |
| ax.text(val + 1, bar.get_y() + bar.get_height()/2, | |
| f"{val:.0f}", fontsize=6, va="center", color="#333") | |
| fig.suptitle(f"Cognitive Diagnostic — \"{query[:50]}...\"" if len(query) > 50 | |
| else f"Cognitive Diagnostic — \"{query}\"", | |
| fontsize=13, fontweight="bold", y=1.05) | |
| # Summary text | |
| summary = ( | |
| f"## 📊 Cognitive Diagnostic Report\n\n" | |
| f"**Query:** _{query}_\n\n" | |
| f"| Layer | Modules | Avg Activity |\n" | |
| f"|-------|---------|-------------|\n" | |
| ) | |
| for layer in layer_names: | |
| mods = layer_modules[layer] | |
| acts = [result["module_activity"].get(m, 0) * 100 for m in mods] | |
| avg = np.mean(acts) | |
| summary += f"| {LAYER_LABELS[layer]} | {len(mods)} | {avg:.0f}% |\n" | |
| summary += ( | |
| f"\n**Confidence:** {result['confidence']:.1%} \n" | |
| f"**Cognitive Load:** {result['cognitive_load']:.0%} \n" | |
| f"**Constitutional:** {'✅' if result['constitutional_compliance'] else '⚠️'}\n" | |
| ) | |
| return summary, fig | |
| def clear_chat(): | |
| """Clear chat history.""" | |
| return None, None | |
| # ============================================================ | |
| # GRADIO HANDLER FUNCTIONS (module level for Gradio 5.x) | |
| # ============================================================ | |
| def build_messages(query, history, system_prompt): | |
| """Build message list for LLM API.""" | |
| messages = [{"role": "system", "content": system_prompt}] | |
| for user_msg, assistant_msg in history: | |
| messages.append({"role": "user", "content": user_msg}) | |
| if assistant_msg: | |
| messages.append({"role": "assistant", "content": assistant_msg}) | |
| messages.append({"role": "user", "content": query}) | |
| return messages | |
| def chat_with_backend(messages, model, api_key=None, worker_url=None): | |
| """Chat with backend LLM.""" | |
| if worker_url: | |
| resp = requests.post( | |
| f"{worker_url.rstrip('/')}/chat", | |
| json={"messages": messages, "model": model}, | |
| timeout=120 | |
| ) | |
| elif api_key: | |
| resp = requests.post( | |
| "https://openrouter.ai/api/v1/chat/completions", | |
| json={"model": model, "messages": messages}, | |
| headers={"Authorization": f"Bearer {api_key}"}, | |
| timeout=120 | |
| ) | |
| else: | |
| return "Error: no API key or Worker configured." | |
| if resp.status_code != 200: | |
| return f"Error {resp.status_code}: {resp.text}" | |
| data = resp.json() | |
| try: | |
| return data["choices"][0]["message"]["content"] | |
| except (KeyError, IndexError): | |
| return str(data) | |
| def respond(message, history, api_key_val=None, model_val=None): | |
| """Handle query submission.""" | |
| if not message.strip(): | |
| return "", history + [(message, "⚠️ Empty query — please ask something.")] | |
| if api_key_val: | |
| # Route to real LLM via OpenRouter | |
| result = chat_with_backend( | |
| build_messages(message, history, SYSTEM_PROMPT), | |
| model=model_val if model_val else DEFAULT_MODEL, | |
| api_key=api_key_val | |
| ) | |
| return "", history + [(message, result)] | |
| else: | |
| # Use simulated CAPT engine | |
| result = capt_engine.process_query(message) | |
| response = ( | |
| f"**{result['response']}\n\n" | |
| f"---\n\n" | |
| f"### Metrics\n\n" | |
| f"- Modules: **{result['module_count']}**/47\n" | |
| f"- Confidence: **{result['confidence']:.1%}**\n" | |
| f"- Time: **{result['processing_time']}**\n" | |
| f"- Governance: **{'✅ PASS' if result['constitutional_compliance'] else '⚠️ REVIEW'}**" | |
| ) | |
| return "", history + [(message, response)] | |
| # ============================================================ | |
| # GRADIO APP DEFINITION | |
| # ============================================================ | |
| def create_interface(): | |
| """Build the full Gradio interface.""" | |
| css = """ | |
| .gradio-container { max-width: 1400px; margin: 0 auto; } | |
| .disco { background: linear-gradient(135deg, #6366f1, #8b5cf6, #ec4899) !important; } | |
| .status-badge { padding: 4px 12px; border-radius: 12px; font-size: 0.85em; font-weight: 600; } | |
| .compliant { background: #dcfce7; color: #166534; } | |
| .review { background: #fef3c7; color: #92400e; } | |
| footer { visibility: hidden; } | |
| """ | |
| with gr.Blocks( | |
| title="CAPT Cognitive Architecture Explorer", | |
| theme=gr.themes.Soft(primary_hue="indigo"), | |
| css=css | |
| ) as demo: | |
| gr.Markdown( | |
| "# 🧠 CAPT Cognitive Architecture Explorer\n\n" | |
| "Explore bioCAPT's 47-neural-module cognitive architecture in real time. " | |
| "Query the system, inspect module activity, and observe constitutional governance.\n\n" | |
| "**47 modules** • **5 layers** • **Holographic memory (ECHO)** • **Self-improving**" | |
| ) | |
| with gr.Tabs(): | |
| # === TAB 1: Interactive Query === | |
| with gr.Tab("💬 Interactive Query"): | |
| gr.Markdown( | |
| "### Ask Anything About the Cognitive Architecture\n\n" | |
| "The system activates relevant neural modules and processes your query through " | |
| "all 6 layers of the architecture." | |
| ) | |
| with gr.Row(): | |
| with gr.Column(scale=3): | |
| query_input = gr.Textbox( | |
| label="Your Question", | |
| placeholder="e.g., 'How does the memory palace work?', 'What happens when a threat is detected?', 'Show me the decision pipeline'", | |
| lines=3, | |
| max_lines=5 | |
| ) | |
| query_btn = gr.Button("🔬 Process Through CAPT", variant="primary", size="lg") | |
| with gr.Column(scale=2): | |
| with gr.Group(): | |
| radar_plot = gr.Plot(label="Layer Activity Radar") | |
| gauge_container = gr.Column() | |
| chatbot = gr.Chatbot(height=500, label="CAPT Response") | |
| with gr.Accordion("⚙️ Advanced Options", open=False): | |
| api_key = gr.Textbox( | |
| label="OpenRouter API Key", type="password", | |
| placeholder="sk-... (uses simulated CAPT if empty)" | |
| ) | |
| model_select = gr.Dropdown( | |
| choices=[ | |
| "nvidia/nemotron-4-340b-instruct", | |
| "anthropic/claude-3-opus", | |
| "openai/gpt-4-turbo", | |
| "meta-llama/llama-3-70b-instruct", | |
| ], | |
| value=DEFAULT_MODEL, | |
| label="Model" | |
| ) | |
| clear_btn = gr.Button("🗑️ Clear Chat") | |
| query_btn.click( | |
| respond, | |
| [query_input, chatbot, api_key, model_select], | |
| [query_input, chatbot] | |
| ) | |
| query_input.submit( | |
| respond, | |
| [query_input, chatbot, api_key, model_select], | |
| [query_input, chatbot] | |
| ) | |
| clear_btn.click(lambda: None, None, chatbot) | |
| # === TAB 2: Module Dashboard === | |
| with gr.Tab("🟢 Module Dashboard"): | |
| gr.Markdown("### Real-Time Neural Module Activity Monitor") | |
| with gr.Row(): | |
| bar_plot = gr.Plot(label="Module Activity (all 47)") | |
| with gr.Column(): | |
| sys_status_btn = gr.Button("📊 Refresh System Status", variant="primary") | |
| status_output = gr.Markdown() | |
| with gr.Row(): | |
| diag_input = gr.Textbox( | |
| label="Query for Cognitive Diagnostic", | |
| placeholder="Enter a query to see which modules it activates...", | |
| lines=2 | |
| ) | |
| diag_btn = gr.Button("Run Diagnostic", variant="secondary") | |
| diag_output = gr.Markdown() | |
| diag_plot = gr.Plot() | |
| # Initialize with current state (handled by click handlers in Gradio 5.x) | |
| # bar_plot and status_output will be populated on first tab interaction | |
| sys_status_btn.click( | |
| lambda: (show_system_status(), create_module_bars(capt_engine.get_module_detail_data())), | |
| outputs=[status_output, bar_plot] | |
| ) | |
| # Always-on diagnostic | |
| diag_btn.click( | |
| generate_cognitive_diag, | |
| [diag_input], | |
| [diag_output, diag_plot] | |
| ) | |
| # === TAB 3: Memory Palace === | |
| with gr.Tab("🏛️ Memory Palace"): | |
| gr.Markdown("### ECHO — Holographic Associative Memory") | |
| with gr.Row(): | |
| mem_output = gr.Markdown() | |
| mem_plot = gr.Plot() | |
| mem_update_btn = gr.Button("🔄 Refresh Memory View", variant="primary") | |
| mem_update_btn.click( | |
| show_memory_palace, | |
| outputs=[mem_output, mem_plot] | |
| ) | |
| # Auto-populate on load (handled by click handler in Gradio 5.x) | |
| # mem_output and mem_plot will populate on first interaction | |
| # === TAB 4: Architecture === | |
| with gr.Tab("🗺️ Architecture"): | |
| arch_output = gr.Markdown(show_layer_map()) | |
| with gr.Accordion("📖 Module Reference (all 47)", open=False): | |
| ref_text = "\n".join( | |
| f"- **{name}** — {info['name']} \n _{info['desc']}_ (layer: {info['layer']})" | |
| for name, info in sorted(CAPT_MODULES.items()) | |
| ) | |
| gr.Markdown(ref_text) | |
| # === TAB 5: Deploy === | |
| with gr.Tab("🚀 Deploy"): | |
| gr.Markdown( | |
| "## Deploy This Space\n\n" | |
| "1. Fork this Space on Hugging Face\n" | |
| "2. Set secrets: `OPENROUTER_API_KEY`, `CLOUDFLARE_WORKER_URL` (optional)\n" | |
| "3. For real CAPT integration, set `CAPT_API_BASE` to your MCP endpoint\n" | |
| "4. Push changes to trigger rebuild\n\n" | |
| "### Connecting to Real CAPT\n\n" | |
| "Replace `CAPTEngine` class with MCP calls:\n\n" | |
| "```python\n" | |
| "# Replace simulated engine with real MCP calls:\n" | |
| "# from mcp import MCPClient\n" | |
| "# client = MCPClient(base_url=os.getenv('CAPT_API_BASE'))\n" | |
| "# result = client.capt_cogitate(query=query)\n" | |
| "# memory = client.capt_query(query=query)\n" | |
| "# delta = client.capt_delta(field='confidence')\n" | |
| "```\n\n" | |
| "### Cloudflare Worker Integration\n\n" | |
| "Deploy `worker.js` for secure API proxying:\n\n" | |
| "```bash\n" | |
| "npm install -g wrangler\n" | |
| "CLOUDFLARE_API_TOKEN=xxx npx wrangler deploy\n" | |
| "npx wrangler secret put OPENROUTER_API_KEY\n" | |
| "```" | |
| ) | |
| gr.Markdown( | |
| "---\n\n" | |
| "**CAPT v2.0** • 47 neural modules • ECHO memory palace • Constitutional governance \n" | |
| "*Simulated for demo. Connect to real MCP endpoint for production.*" | |
| ) | |
| return demo | |
| if __name__ == "__main__": | |
| print("Starting CAPT Architecture Explorer...") | |
| print(f"Modules loaded: {len(CAPT_MODULES)}") | |
| print(f"Engine: {'SIMULATED' if True else 'REAL MCP'}") | |
| demo = create_interface() | |
| demo.launch(server_name="0.0.0.0", server_port=7860) |