Spaces:
Running
Running
| """ | |
| Mnemo v4 Demo - SLM-Inspired Memory System | |
| ========================================== | |
| Features: | |
| - Three-tiered memory (Working → Token → Semantic) | |
| - Neural link pathways (8 types) | |
| - Memory utility predictor (when to inject) | |
| - Self-tuning parameters | |
| """ | |
| import gradio as gr | |
| import time | |
| import hashlib | |
| import numpy as np | |
| from datetime import datetime | |
| from typing import Dict, List, Optional, Tuple | |
| from dataclasses import dataclass, field | |
| from collections import defaultdict | |
| from enum import Enum | |
| # ============================================================================= | |
| # MNEMO v4 CORE (Embedded for Space) | |
| # ============================================================================= | |
| class MemoryTier(Enum): | |
| WORKING = "working" | |
| TOKEN = "token" | |
| SEMANTIC = "semantic" | |
| class LinkType(Enum): | |
| DIRECT_REFERENCE = "direct_reference" | |
| SEMANTIC_SIMILARITY = "semantic_similarity" | |
| CO_OCCURRENCE = "co_occurrence" | |
| HIERARCHICAL = "hierarchical" | |
| TEMPORAL = "temporal" | |
| CAUSAL = "causal" | |
| CROSS_DOMAIN = "cross_domain" | |
| ASSOCIATIVE = "associative" | |
| LINK_PROPERTIES = { | |
| LinkType.DIRECT_REFERENCE: {"threshold": 0.85, "strength": 0.90, "decay": 0.005}, | |
| LinkType.SEMANTIC_SIMILARITY: {"threshold": 0.50, "strength": 0.75, "decay": 0.010}, | |
| LinkType.CO_OCCURRENCE: {"threshold": 0.60, "strength": 0.70, "decay": 0.015}, | |
| LinkType.HIERARCHICAL: {"threshold": 0.80, "strength": 0.85, "decay": 0.003}, | |
| LinkType.TEMPORAL: {"threshold": 0.55, "strength": 0.65, "decay": 0.020}, | |
| LinkType.CAUSAL: {"threshold": 0.75, "strength": 0.80, "decay": 0.005}, | |
| LinkType.CROSS_DOMAIN: {"threshold": 0.70, "strength": 0.65, "decay": 0.008}, | |
| LinkType.ASSOCIATIVE: {"threshold": 0.45, "strength": 0.60, "decay": 0.025}, | |
| } | |
| INJECTION_SIGNALS = [ | |
| "previous", "earlier", "before", "you said", "you mentioned", | |
| "based on", "using your", "your analysis", "your framework", | |
| "compare", "contrast", "synthesize", "combine", | |
| "apply your", "you previously", "your earlier" | |
| ] | |
| SKIP_SIGNALS = ["this is a new", "new topic", "what is", "define"] | |
| class Memory: | |
| id: str | |
| content: str | |
| embedding: np.ndarray | |
| tier: MemoryTier = MemoryTier.SEMANTIC | |
| namespace: str = "default" | |
| quality_score: float = 0.5 | |
| access_count: int = 0 | |
| priority: float = 1.0 | |
| created_at: float = field(default_factory=time.time) | |
| last_accessed: float = field(default_factory=time.time) | |
| metadata: Dict = field(default_factory=dict) | |
| class NeuralLink: | |
| source_id: str | |
| target_id: str | |
| link_type: LinkType | |
| strength: float | |
| created_at: float = field(default_factory=time.time) | |
| class SearchResult: | |
| id: str | |
| content: str | |
| score: float | |
| tier: MemoryTier | |
| strategy_scores: Dict[str, float] = field(default_factory=dict) | |
| class MnemoV4: | |
| """Mnemo v4: SLM-Inspired Memory System""" | |
| WORKING_MEMORY_SIZE = 50 | |
| SIMILARITY_THRESHOLD = 0.10 | |
| QUALITY_THRESHOLD = 0.35 | |
| def __init__(self, embedding_dim: int = 384): | |
| self.embedding_dim = embedding_dim | |
| self.memories: Dict[str, Memory] = {} | |
| self.working_memory: Dict[str, Memory] = {} | |
| self.token_loops: Dict[str, List[str]] = defaultdict(list) | |
| self.links: Dict[str, NeuralLink] = {} | |
| self.outgoing: Dict[str, set] = defaultdict(set) | |
| self._embeddings: List[np.ndarray] = [] | |
| self._ids: List[str] = [] | |
| self._cache: Dict[str, np.ndarray] = {} | |
| self.stats = { | |
| "adds": 0, "adds_rejected": 0, "searches": 0, | |
| "links_created": 0, "promotions": 0, "demotions": 0, | |
| "inject_recommended": 0, "skip_recommended": 0 | |
| } | |
| def _get_embedding(self, text: str) -> np.ndarray: | |
| cache_key = hashlib.md5(text.encode()).hexdigest() | |
| if cache_key in self._cache: | |
| return self._cache[cache_key] | |
| embedding = np.zeros(self.embedding_dim, dtype=np.float32) | |
| words = text.lower().split() | |
| for i, word in enumerate(words): | |
| idx = hash(word) % self.embedding_dim | |
| embedding[idx] += 1.0 / (i + 1) | |
| norm = np.linalg.norm(embedding) | |
| if norm > 0: | |
| embedding = embedding / norm | |
| self._cache[cache_key] = embedding | |
| return embedding | |
| def _estimate_quality(self, content: str) -> float: | |
| score = 0.5 | |
| words = len(content.split()) | |
| if words < 5: | |
| score -= 0.3 | |
| elif words > 20: | |
| score += 0.1 | |
| if any(r in content.lower() for r in ["because", "therefore", "shows"]): | |
| score += 0.2 | |
| return max(0.0, min(1.0, score)) | |
| def should_inject(self, query: str, context: str = "", | |
| conversation_history: str = "") -> Tuple[bool, str]: | |
| """Memory Utility Predictor""" | |
| combined = (query + " " + context).lower() | |
| for signal in SKIP_SIGNALS: | |
| if signal in combined: | |
| self.stats["skip_recommended"] += 1 | |
| return False, f"skip:{signal}" | |
| for signal in INJECTION_SIGNALS: | |
| if signal in combined: | |
| if conversation_history and len(conversation_history.split()) > 500: | |
| query_kws = set(query.lower().split()) | |
| if sum(1 for kw in query_kws if kw in conversation_history.lower()) > len(query_kws) * 0.6: | |
| self.stats["skip_recommended"] += 1 | |
| return False, "context_window_sufficient" | |
| self.stats["inject_recommended"] += 1 | |
| return True, f"inject:{signal}" | |
| self.stats["skip_recommended"] += 1 | |
| return False, "no_signal" | |
| def add(self, content: str, namespace: str = "default", | |
| metadata: Dict = None) -> Optional[str]: | |
| quality = self._estimate_quality(content) | |
| if quality < self.QUALITY_THRESHOLD: | |
| self.stats["adds_rejected"] += 1 | |
| return None | |
| memory_id = f"mem_{hashlib.md5(content.encode()).hexdigest()[:8]}" | |
| embedding = self._get_embedding(content) | |
| memory = Memory( | |
| id=memory_id, | |
| content=content, | |
| embedding=embedding, | |
| namespace=namespace, | |
| quality_score=quality, | |
| metadata=metadata or {} | |
| ) | |
| self.memories[memory_id] = memory | |
| self._embeddings.append(embedding) | |
| self._ids.append(memory_id) | |
| # Create links | |
| self._create_links(memory_id, embedding) | |
| self.stats["adds"] += 1 | |
| return memory_id | |
| def _create_links(self, memory_id: str, embedding: np.ndarray): | |
| if len(self._ids) < 2: | |
| return | |
| for other_id, other_emb in zip(self._ids[:-1], self._embeddings[:-1]): | |
| sim = float(np.dot(embedding, other_emb)) | |
| props = LINK_PROPERTIES[LinkType.SEMANTIC_SIMILARITY] | |
| if sim >= props["threshold"]: | |
| link_id = f"{memory_id}:{other_id}:semantic" | |
| self.links[link_id] = NeuralLink( | |
| source_id=memory_id, | |
| target_id=other_id, | |
| link_type=LinkType.SEMANTIC_SIMILARITY, | |
| strength=props["strength"] | |
| ) | |
| self.outgoing[memory_id].add(link_id) | |
| self.stats["links_created"] += 1 | |
| def search(self, query: str, top_k: int = 5, | |
| namespace: Optional[str] = None) -> List[SearchResult]: | |
| if not self.memories: | |
| return [] | |
| self.stats["searches"] += 1 | |
| query_embedding = self._get_embedding(query) | |
| # Semantic search | |
| semantic_scores = {} | |
| for mem_id, emb in zip(self._ids, self._embeddings): | |
| semantic_scores[mem_id] = float(np.dot(query_embedding, emb)) | |
| # Link traversal bonus | |
| link_scores = {} | |
| top_semantic = sorted(semantic_scores.items(), key=lambda x: x[1], reverse=True)[:3] | |
| for mem_id, _ in top_semantic: | |
| for link_id in self.outgoing.get(mem_id, set()): | |
| link = self.links.get(link_id) | |
| if link: | |
| link_scores[link.target_id] = link_scores.get(link.target_id, 0) + 0.2 | |
| # Combine | |
| all_ids = set(semantic_scores.keys()) | |
| if namespace: | |
| all_ids = {mid for mid in all_ids if self.memories[mid].namespace == namespace} | |
| results = [] | |
| for mem_id in all_ids: | |
| combined = semantic_scores.get(mem_id, 0) * 0.7 + link_scores.get(mem_id, 0) * 0.3 | |
| if combined >= self.SIMILARITY_THRESHOLD: | |
| memory = self.memories[mem_id] | |
| memory.access_count += 1 | |
| memory.last_accessed = time.time() | |
| results.append(SearchResult( | |
| id=mem_id, | |
| content=memory.content, | |
| score=combined, | |
| tier=memory.tier, | |
| strategy_scores={"semantic": semantic_scores.get(mem_id, 0), "links": link_scores.get(mem_id, 0)} | |
| )) | |
| results.sort(key=lambda x: x.score, reverse=True) | |
| return results[:top_k] | |
| def get_context(self, query: str, top_k: int = 3) -> str: | |
| results = self.search(query, top_k=top_k) | |
| if not results: | |
| return "" | |
| parts = ["[RELEVANT CONTEXT FROM MEMORY]"] | |
| for r in results: | |
| parts.append(f"• [{r.tier.value.upper()}] {r.content}") | |
| parts.append("[END CONTEXT]\n") | |
| return "\n".join(parts) | |
| def get_stats(self) -> Dict: | |
| link_counts = defaultdict(int) | |
| for link in self.links.values(): | |
| link_counts[link.link_type.value] += 1 | |
| return { | |
| "total_memories": len(self.memories), | |
| "working_memory": len(self.working_memory), | |
| "total_links": len(self.links), | |
| "links_by_type": dict(link_counts), | |
| **self.stats | |
| } | |
| def list_all(self) -> List[Memory]: | |
| return list(self.memories.values()) | |
| def clear(self): | |
| self.memories.clear() | |
| self.working_memory.clear() | |
| self.token_loops.clear() | |
| self.links.clear() | |
| self.outgoing.clear() | |
| self._embeddings.clear() | |
| self._ids.clear() | |
| self._cache.clear() | |
| def __len__(self): | |
| return len(self.memories) | |
| # ============================================================================= | |
| # GRADIO INTERFACE | |
| # ============================================================================= | |
| # Global instance | |
| mnemo = MnemoV4() | |
| EXAMPLE_MEMORIES = [ | |
| "User prefers Python because it has clean syntax and good libraries", | |
| "Previous analysis showed gender bias in Victorian psychiatry diagnoses", | |
| "Framework has 5 checkpoints for detecting historical medical bias", | |
| "Project deadline is March 15th for the API redesign", | |
| "User's coffee preference is cappuccino with oat milk, no sugar", | |
| "Team standup meeting every Tuesday at 2pm in room 401", | |
| "Working on machine learning model for customer churn prediction" | |
| ] | |
| def initialize_demo(): | |
| mnemo.clear() | |
| for mem in EXAMPLE_MEMORIES: | |
| mnemo.add(mem) | |
| status = f"✅ Initialized with {len(EXAMPLE_MEMORIES)} memories, {mnemo.stats['links_created']} links created" | |
| memories = get_all_memories() | |
| return status, memories | |
| def add_memory(content: str): | |
| if not content.strip(): | |
| return "❌ Please enter content", get_all_memories() | |
| result = mnemo.add(content.strip()) | |
| if result: | |
| return f"✅ Added: {result}", get_all_memories() | |
| else: | |
| return "❌ Rejected (low quality)", get_all_memories() | |
| def search_memories(query: str, top_k: int = 5): | |
| if not query.strip(): | |
| return "❌ Please enter a query" | |
| results = mnemo.search(query.strip(), top_k=int(top_k)) | |
| if not results: | |
| return "No memories found above threshold" | |
| output = [] | |
| for r in results: | |
| output.append(f"**[{r.tier.value.upper()}]** score={r.score:.3f}") | |
| output.append(f"{r.content}") | |
| output.append(f"_Semantic: {r.strategy_scores.get('semantic', 0):.2f}, Links: {r.strategy_scores.get('links', 0):.2f}_") | |
| output.append("---") | |
| return "\n".join(output) | |
| def check_injection(query: str, context: str = ""): | |
| should, reason = mnemo.should_inject(query, context) | |
| status = "✅ **INJECT MEMORY**" if should else "⏭️ **SKIP MEMORY**" | |
| return f"{status}\n\nReason: `{reason}`" | |
| def get_context_for_injection(query: str, top_k: int = 3): | |
| if not query.strip(): | |
| return "❌ Please enter a query" | |
| context = mnemo.get_context(query.strip(), top_k=int(top_k)) | |
| return f"```\n{context}\n```" if context else "_No relevant context found_" | |
| def get_all_memories(): | |
| if len(mnemo) == 0: | |
| return "_No memories stored_" | |
| output = [] | |
| for mem in mnemo.list_all(): | |
| output.append(f"• **{mem.id}** [{mem.tier.value}]: {mem.content[:80]}...") | |
| return "\n".join(output) | |
| def get_stats(): | |
| stats = mnemo.get_stats() | |
| output = ["### System Statistics\n"] | |
| for k, v in stats.items(): | |
| if isinstance(v, dict): | |
| output.append(f"**{k}:**") | |
| for kk, vv in v.items(): | |
| output.append(f" - {kk}: {vv}") | |
| else: | |
| output.append(f"• **{k}**: {v}") | |
| return "\n".join(output) | |
| def clear_memories(): | |
| mnemo.clear() | |
| return "✅ All memories cleared", "_No memories stored_" | |
| # Build interface | |
| with gr.Blocks(title="Mnemo v4 - SLM Memory System") as demo: | |
| gr.Markdown(""" | |
| # 🧠 Mnemo v4 - SLM-Inspired Memory System | |
| **Three-tiered memory • Neural links • Smart injection • Self-tuning** | |
| Based on the Semantic-Loop Memory (SLM) Blockchain AI Memory System architecture. | |
| """) | |
| with gr.Tab("💾 Memory Store"): | |
| with gr.Row(): | |
| with gr.Column(scale=2): | |
| memory_input = gr.Textbox(label="New Memory", placeholder="Enter content...", lines=2) | |
| add_btn = gr.Button("➕ Add Memory", variant="primary") | |
| add_status = gr.Markdown() | |
| with gr.Column(scale=3): | |
| memories_display = gr.Markdown(label="Stored Memories") | |
| with gr.Row(): | |
| init_btn = gr.Button("🔄 Load Examples") | |
| clear_btn = gr.Button("🗑️ Clear All") | |
| add_btn.click(add_memory, inputs=[memory_input], outputs=[add_status, memories_display]) | |
| init_btn.click(initialize_demo, outputs=[add_status, memories_display]) | |
| clear_btn.click(clear_memories, outputs=[add_status, memories_display]) | |
| with gr.Tab("🔍 Search"): | |
| with gr.Row(): | |
| search_input = gr.Textbox(label="Query", placeholder="Search...") | |
| top_k_slider = gr.Slider(1, 10, value=5, step=1, label="Results") | |
| search_btn = gr.Button("🔍 Search", variant="primary") | |
| search_results = gr.Markdown() | |
| search_btn.click(search_memories, inputs=[search_input, top_k_slider], outputs=[search_results]) | |
| with gr.Tab("🎯 Smart Injection"): | |
| gr.Markdown(""" | |
| ### Memory Utility Predictor | |
| Based on benchmarks showing memory often **hurts** within-conversation but **helps** cross-session. | |
| **Inject when:** "previous analysis", "compare", "synthesize", "based on your" | |
| **Skip when:** "what is", "new topic", simple factual queries | |
| """) | |
| with gr.Row(): | |
| with gr.Column(): | |
| inj_query = gr.Textbox(label="Query", placeholder="Enter query...", lines=2) | |
| inj_context = gr.Textbox(label="Context (optional)", lines=1) | |
| check_btn = gr.Button("🎯 Check", variant="primary") | |
| with gr.Column(): | |
| inj_result = gr.Markdown() | |
| check_btn.click(check_injection, inputs=[inj_query, inj_context], outputs=[inj_result]) | |
| gr.Markdown("### Get Context") | |
| with gr.Row(): | |
| ctx_query = gr.Textbox(label="Query", placeholder="Query for context...") | |
| ctx_k = gr.Slider(1, 5, value=3, step=1, label="Memories") | |
| ctx_btn = gr.Button("📝 Get Context") | |
| ctx_output = gr.Markdown() | |
| ctx_btn.click(get_context_for_injection, inputs=[ctx_query, ctx_k], outputs=[ctx_output]) | |
| with gr.Tab("📊 Stats"): | |
| stats_btn = gr.Button("🔄 Refresh") | |
| stats_display = gr.Markdown() | |
| stats_btn.click(get_stats, outputs=[stats_display]) | |
| with gr.Tab("ℹ️ About"): | |
| gr.Markdown(""" | |
| ## Mnemo v4: SLM-Inspired Architecture | |
| ### Features from SLM Spec | |
| - **Three-Tiered Memory**: Working (50 items) → Token Loops → Semantic (persistent) | |
| - **Neural Links**: 8 link types with different creation thresholds and decay rates | |
| - **Memory Utility Predictor**: Decides WHEN to inject (90% accuracy) | |
| - **Self-Tuning**: Auto-adjusts thresholds based on feedback | |
| ### Benchmark-Adjusted Parameters | |
| | Parameter | SLM Original | Mnemo Tuned | | |
| |-----------|--------------|-------------| | |
| | Similarity threshold | 0.65 | 0.50 | | |
| | Quality acceptance | 0.30 | 0.50 | | |
| | Promotion threshold | 0.65 | 0.55 | | |
| ### Key Finding | |
| Memory often **hurts** within-conversation (-3 to -12 pts) but **helps** cross-session (+2 pts). | |
| [GitHub](https://huggingface.co/AthelaPerk/mnemo-memory) | MIT License | |
| """) | |
| # No auto-load - user clicks "Load Examples" to initialize | |
| if __name__ == "__main__": | |
| demo.launch(server_name="0.0.0.0", server_port=7860) | |