Initial release: MoME — Mixture of Memory Experts plugin for Hermes Agent

2026-05-26 22:19:49 +03:00
commit 89f1bae5e0
10 changed files with 8887 additions and 0 deletions
@@ -0,0 +1,116 @@
+# MoME — Mixture of Memory Experts for Hermes Agent
+
+Sparse-gated personal memory with an online-learning router. Replaces monolithic memory context with expert-routed retrieval — only relevant memory experts are activated per query.
+
+## Features
+
+- **4 Experts**: `identity`, `knowledge`, `projects`, `preferences`
+- **Sparse Activation**: Only 1–2 experts are queried per turn (determined by the router)
+- **Online Learning**: SGD classifier router learns which expert to route to based on usage patterns
+- **Local-Only**: 100% offline, no API keys needed
+- **Auto-Learning**: Regex-based fact extraction from user queries
+- **Persistent**: JSON-based storage, survives restarts
+
+## Installation
+
+```bash
+# 1. Clone the plugin into Hermes plugins directory
+git clone https://github.com/emil28092005/hermes-plugin-mome.git \
+    ~/.hermes/hermes-agent/plugins/memory/mome
+
+# 2. Install dependencies
+pip install numpy scikit-learn
+
+# 3. Activate via Hermes memory setup
+hermes memory setup
+```
+
+Select `mome` from the list of available memory providers.
+
+## Usage
+
+Once activated, MoME provides three tools:
+
+### `mome_search`
+Search memory across experts. The router automatically selects which experts to query.
+
+```text
+mome_search(query="what projects am I working on?", top_k=3)
+mome_search(query="Python skills", expert="knowledge", top_k=5)
+```
+
+### `mome_store`
+Store a fact directly into a specific expert.
+
+```text
+mome_store(expert="identity", fact="I prefer dark mode")
+```
+
+### `mome_status`
+Show expert sizes and router training state.
+
+```text
+mome_status()
+```
+
+## Architecture
+
+```
+User Query
+    │
+    ▼
+┌─────────────┐
+│   Router    │ ← SGDClassifier (online learning)
+│  (predict)  │
+└──────┬──────┘
+       │ top-2 experts selected
+       ▼
+┌──────┴──────┐
+│   Experts   │
+│  identity   │ ─── cosine similarity search
+│  knowledge  │
+│  projects   │
+│ preferences │
+└──────┬──────┘
+       │ relevant memories
+       ▼
+┌─────────────┐
+│   Context   │ → injected into system prompt
+└─────────────┘
+```
+
+### Fact Extraction (Auto-Learning)
+
+On each turn, MoME automatically extracts facts from user queries via regex:
+
+- `меня зовут X` → identity
+- `я живу в X` → identity
+- `работаю над X` → projects
+- `у меня проект X` → projects
+- `знаю/использую X` → knowledge
+- `нравится X` → preferences
+
+The router learns over time which experts to activate for which types of queries.
+
+## Dependencies
+
+- Python ≥ 3.10
+- `numpy` — vector operations
+- `scikit-learn` — SGD router classifier
+
+Zero external API dependencies. Works fully offline.
+
+## Development
+
+```bash
+# Test the engine standalone
+python -c "from plugins.memory.mome.engine import MomeEngine; e = MomeEngine('/tmp/test_mome'); e.store_fact('identity', 'Test fact'); print(e.query('test'))"
+```
+
+## Author
+
+**Emil Shanaty** — [github.com/emil28092005](https://github.com/emil28092005)
+
+## License
+
+MIT
@@ -0,0 +1,283 @@
+"""
+MoME — Mixture of Memory Experts plugin for Hermes Agent.
+
+Sparse-gated personal memory with online-learning router.
+Replaces monolithic memory context with expert-routed retrieval.
+
+Activate::
+    hermes memory setup
+    # then select "mome" from the list
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import threading
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+from agent.memory_provider import MemoryProvider
+
+from .engine import EXPERT_NAMES, MomeEngine
+
+logger = logging.getLogger(__name__)
+
+# ─── Tool Schemas ─────────────────────────────────────────────────────────
+
+MEMORY_SEARCH_SCHEMA = {
+    "name": "mome_search",
+    "description": "Search MoME memory experts. Returns relevant facts routed to the right expert.",
+    "parameters": {
+        "type": "object",
+        "properties": {
+            "query": {
+                "type": "string",
+                "description": "What to search for.",
+            },
+            "expert": {
+                "type": "string",
+                "enum": EXPERT_NAMES + ["all"],
+                "description": "Which expert to search (default: auto-routed).",
+            },
+            "top_k": {
+                "type": "integer",
+                "description": "Max results (default: 3).",
+            },
+        },
+        "required": ["query"],
+    },
+}
+
+MEMORY_STORE_SCHEMA = {
+    "name": "mome_store",
+    "description": "Store a fact in a specific MoME memory expert.",
+    "parameters": {
+        "type": "object",
+        "properties": {
+            "expert": {
+                "type": "string",
+                "enum": EXPERT_NAMES,
+                "description": "Target expert.",
+            },
+            "fact": {
+                "type": "string",
+                "description": "The fact to remember.",
+            },
+        },
+        "required": ["expert", "fact"],
+    },
+}
+
+MEMORY_STATUS_SCHEMA = {
+    "name": "mome_status",
+    "description": "Show MoME memory status — expert sizes and router learning state.",
+    "parameters": {"type": "object", "properties": {}},
+}
+
+
+# ─── MoME Provider ─────────────────────────────────────────────────────────
+
+class MomeProvider(MemoryProvider):
+    """MoME — Mixture of Memory Experts for Hermes Agent.
+
+    Sparse-gated personal memory with:
+    - 4 experts (identity, knowledge, projects, preferences)
+    - Online-learning SGD classifier router
+    - Regex-based fact extraction from user queries
+    """
+
+    def __init__(self):
+        self._engine: Optional[MomeEngine] = None
+        self._hermes_home: Optional[Path] = None
+        self._prefetch_result = ""
+        self._prefetch_lock = threading.Lock()
+        self._prefetch_thread: Optional[threading.Thread] = None
+
+    @property
+    def name(self) -> str:
+        return "mome"
+
+    def is_available(self) -> bool:
+        return True  # 100% local, no API keys needed
+
+    def get_config_schema(self) -> List[Dict[str, Any]]:
+        return [
+            {
+                "key": "store_dir",
+                "description": (
+                    "Directory for MoME memory storage "
+                    "(relative to HERMES_HOME)"
+                ),
+                "default": "mome_store",
+                "required": False,
+            },
+        ]
+
+    def initialize(self, session_id: str, **kwargs) -> None:
+        hermes_home = kwargs.get(
+            "hermes_home",
+            os.environ.get("HERMES_HOME", ""),
+        )
+        if hermes_home:
+            self._hermes_home = Path(hermes_home)
+        else:
+            self._hermes_home = Path.home() / ".hermes"
+
+        store_dir = self._hermes_home / "mome_store"
+        self._engine = MomeEngine(store_dir)
+
+        logger.info(
+            "MoME initialized: %s (%d experts, %d total facts)",
+            store_dir,
+            len(self._engine.experts),
+            sum(e.count() for e in self._engine.experts.values()),
+        )
+
+    def system_prompt_block(self) -> str:
+        if not self._engine:
+            return ""
+        stats = self._engine.get_stats()
+        total = sum(stats.values())
+        return (
+            "## MoME Memory\n"
+            f"Active. Experts: "
+            f"{', '.join(f'{k}={v}' for k, v in stats.items())} "
+            f"total={total}\n"
+            "Use `mome_search` to find memories, "
+            "`mome_store` to save facts, "
+            "`mome_status` for expert info."
+        )
+
+    def prefetch(self, query: str, *, session_id: str = "") -> str:
+        if not self._engine:
+            return ""
+        context = self._engine.query(query)
+        if context:
+            return f"## MoME Memory Context\n{context}"
+        return ""
+
+    def queue_prefetch(self, query: str, *, session_id: str = "") -> None:
+        if not self._engine:
+            return
+
+        def _run() -> None:
+            try:
+                context = self._engine.query(query)
+                with self._prefetch_lock:
+                    self._prefetch_result = context
+            except Exception as e:
+                logger.debug("MoME prefetch failed: %s", e)
+
+        self._prefetch_thread = threading.Thread(
+            target=_run, daemon=True, name="mome-prefetch",
+        )
+        self._prefetch_thread.start()
+
+    def sync_turn(
+        self,
+        user_content: str,
+        assistant_content: str,
+        *,
+        session_id: str = "",
+    ) -> None:
+        if not self._engine:
+            return
+        try:
+            self._engine.extract_and_store(user_content)
+        except Exception as e:
+            logger.debug("MoME sync failed: %s", e)
+
+    def get_tool_schemas(self) -> List[Dict[str, Any]]:
+        return [
+            MEMORY_SEARCH_SCHEMA,
+            MEMORY_STORE_SCHEMA,
+            MEMORY_STATUS_SCHEMA,
+        ]
+
+    def handle_tool_call(
+        self,
+        tool_name: str,
+        args: Dict[str, Any],
+        **kwargs,
+    ) -> str:
+        if not self._engine:
+            return json.dumps({"error": "MoME not initialized"})
+
+        if tool_name == "mome_search":
+            return self._handle_search(args)
+        elif tool_name == "mome_store":
+            return self._handle_store(args)
+        elif tool_name == "mome_status":
+            return self._handle_status()
+
+        return json.dumps({"error": f"Unknown tool: {tool_name}"})
+
+    def _handle_search(self, args: Dict[str, Any]) -> str:
+        query = args.get("query", "")
+        expert_filter = args.get("expert", "all")
+        top_k = int(args.get("top_k", 3))
+
+        if expert_filter == "all":
+            selected = self._engine.router.predict(query, top_k=2)
+            results = []
+            for name, conf in selected:
+                memories = self._engine.experts[name].search(query, top_k=top_k)
+                for m in memories:
+                    results.append({
+                        "expert": name,
+                        "memory": m,
+                        "confidence": round(conf, 2),
+                    })
+            return json.dumps({"results": results, "count": len(results)})
+
+        elif expert_filter in self._engine.experts:
+            memories = self._engine.experts[expert_filter].search(
+                query, top_k=top_k,
+            )
+            results = [
+                {"expert": expert_filter, "memory": m}
+                for m in memories
+            ]
+            return json.dumps({"results": results, "count": len(results)})
+
+        return json.dumps({"error": f"Unknown expert: {expert_filter}"})
+
+    def _handle_store(self, args: Dict[str, Any]) -> str:
+        expert = args.get("expert", "")
+        fact = args.get("fact", "")
+        if expert in self._engine.experts and fact:
+            self._engine.experts[expert].write(fact)
+            return json.dumps({
+                "result": f"Stored in [{expert}]",
+                "fact": fact,
+            })
+        return json.dumps({
+            "error": f"Invalid expert '{expert}' or empty fact",
+        })
+
+    def _handle_status(self) -> str:
+        stats = self._engine.get_stats()
+        router_status = (
+            "trained"
+            if self._engine.router._fitted
+            else "untrained (default routing)"
+        )
+        return json.dumps({
+            "expert_counts": stats,
+            "total": sum(stats.values()),
+            "router": router_status,
+            "experts": EXPERT_NAMES,
+        })
+
+    def shutdown(self) -> None:
+        if self._prefetch_thread and self._prefetch_thread.is_alive():
+            self._prefetch_thread.join(timeout=3.0)
+        self._engine = None
+        logger.info("MoME shut down")
+
+
+def register(ctx) -> None:
+    """Register MoME as a Hermes memory provider plugin."""
+    ctx.register_memory_provider(MomeProvider())
@@ -0,0 +1,236 @@
+"""
+MoME Engine — Mixture of Memory Experts core.
+Sparse-gated personal memory with online-learning router.
+
+Requires: numpy, scikit-learn
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import re
+import threading
+import time
+import uuid
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+import numpy as np
+from sklearn.linear_model import SGDClassifier
+
+logger = logging.getLogger(__name__)
+
+# ─── Expert Names ──────────────────────────────────────────────────────────
+
+EXPERT_NAMES = ["identity", "knowledge", "projects", "preferences"]
+
+# ─── Tiny Embedder ─────────────────────────────────────────────────────────
+
+class TinyEmbedder:
+    """Минимальный bag-of-words эмбеддер без внешних зависимостей."""
+
+    def __init__(self, dim: int = 384):
+        self.dim = dim
+        self.word_vectors: Dict[str, np.ndarray] = {}
+        self.rng = np.random.RandomState(42)
+
+    def _get_word_vec(self, word: str) -> np.ndarray:
+        if word not in self.word_vectors:
+            v = self.rng.randn(self.dim).astype(np.float32)
+            v /= np.linalg.norm(v) + 1e-8
+            self.word_vectors[word] = v
+        return self.word_vectors[word]
+
+    def embed(self, text: str) -> np.ndarray:
+        words = re.findall(r'\w+', text.lower())
+        if not words:
+            return np.zeros(self.dim, dtype=np.float32)
+        vecs = [self._get_word_vec(w) for w in words]
+        vec = np.mean(vecs, axis=0).astype(np.float32)
+        vec /= np.linalg.norm(vec) + 1e-8
+        return vec
+
+
+# ─── Memory Expert ─────────────────────────────────────────────────────────
+
+class MemoryExpert:
+    """Хранилище памяти эксперта с векторным поиском."""
+
+    def __init__(self, name: str, embedder: TinyEmbedder, store_dir: Path):
+        self.name = name
+        self.embedder = embedder
+        self.path = store_dir / f"{name}.json"
+        self.memories: List[Dict[str, Any]] = []
+        self._lock = threading.Lock()
+        self.load()
+
+    def load(self) -> None:
+        if self.path.exists():
+            try:
+                with open(self.path) as f:
+                    data = json.load(f)
+                    for m in data:
+                        m["embedding"] = np.array(m["embedding"], dtype=np.float32)
+                    self.memories = data
+            except Exception:
+                self.memories = []
+
+    def save(self) -> None:
+        data = []
+        for m in self.memories:
+            entry = dict(m)
+            entry["embedding"] = m["embedding"].tolist()
+            data.append(entry)
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        with open(self.path, "w") as f:
+            json.dump(data, f, ensure_ascii=False, indent=2)
+
+    def write(self, text: str) -> None:
+        with self._lock:
+            embedding = self.embedder.embed(text)
+            self.memories.append({
+                "id": str(uuid.uuid4()),
+                "text": text,
+                "embedding": embedding,
+                "timestamp": time.time(),
+                "access_count": 0,
+            })
+            self.save()
+
+    def search(self, query: str, top_k: int = 3) -> List[str]:
+        with self._lock:
+            if not self.memories:
+                return []
+            q_emb = self.embedder.embed(query)
+            scores = [float(np.dot(q_emb, m["embedding"])) for m in self.memories]
+            top_idx = np.argsort(scores)[::-1][:top_k]
+            results = []
+            for i in top_idx:
+                self.memories[i]["access_count"] += 1
+                results.append(self.memories[i]["text"])
+            self.save()
+            return results
+
+    def all(self) -> List[str]:
+        with self._lock:
+            return [m["text"] for m in self.memories]
+
+    def count(self) -> int:
+        with self._lock:
+            return len(self.memories)
+
+    def delete(self, text_contains: str) -> None:
+        with self._lock:
+            self.memories = [m for m in self.memories if text_contains not in m["text"]]
+            self.save()
+
+
+# ─── Online Router ─────────────────────────────────────────────────────────
+
+class MemoryRouter:
+    """Online-learning роутер экспертов через SGDClassifier."""
+
+    def __init__(self, embedder: TinyEmbedder):
+        self.embedder = embedder
+        self.classifier = SGDClassifier(
+            loss='log_loss', penalty='l2', alpha=0.001,
+            learning_rate='adaptive', eta0=0.01,
+            warm_start=True, random_state=42,
+        )
+        self._fitted = False
+        self._classes = np.array(EXPERT_NAMES)
+
+    def predict(self, query: str, top_k: int = 2) -> List[tuple[str, float]]:
+        emb = self.embedder.embed(query).reshape(1, -1)
+        if not self._fitted:
+            return [(name, 0.5) for name in EXPERT_NAMES[:top_k]]
+        probs = self.classifier.predict_proba(emb)[0]
+        top_indices = np.argsort(probs)[::-1][:top_k]
+        return [(EXPERT_NAMES[i], float(probs[i])) for i in top_indices]
+
+    def update(self, query: str, feedback: Dict[str, float]) -> None:
+        emb = self.embedder.embed(query).reshape(1, -1)
+        best_expert = max(feedback, key=feedback.get)
+        target = np.array([best_expert])
+        if not self._fitted:
+            self.classifier.partial_fit(emb, target, classes=self._classes)
+            self._fitted = True
+        else:
+            self.classifier.partial_fit(emb, target)
+
+
+# ─── MoME Engine ───────────────────────────────────────────────────────────
+
+class MomeEngine:
+    """Ядро MoME: роутер + эксперты + извлечение фактов."""
+
+    def __init__(self, store_dir: Path):
+        self.store_dir = store_dir
+        self.store_dir.mkdir(parents=True, exist_ok=True)
+        self.embedder = TinyEmbedder(dim=384)
+        self.router = MemoryRouter(self.embedder)
+        self.experts = {
+            name: MemoryExpert(name, self.embedder, store_dir)
+            for name in EXPERT_NAMES
+        }
+
+    def query(self, text: str, top_k: int = 2) -> str:
+        """Получить релевантный контекст. Возвращает форматированную строку."""
+        selected = self.router.predict(text, top_k=top_k)
+        parts = []
+        for name, confidence in selected:
+            results = self.experts[name].search(text, top_k=3)
+            if results:
+                lines = "\n".join(f"• {r}" for r in results)
+                parts.append(f"[{name.upper()}] (conf: {confidence:.2f}):\n{lines}")
+        return "\n\n".join(parts) if parts else ""
+
+    def store_fact(self, expert: str, fact: str) -> bool:
+        """Сохранить факт в указанного эксперта."""
+        if expert in self.experts and fact:
+            self.experts[expert].write(fact)
+            return True
+        return False
+
+    def learn(self, query: str, expert_ratings: Dict[str, float]) -> None:
+        """Обучить роутер на feedback."""
+        self.router.update(query, expert_ratings)
+
+    def extract_and_store(self, query: str) -> int:
+        """Извлечь факты из запроса через regex и сохранить."""
+        facts = []
+
+        name_match = re.search(r'(?:меня\s+зовут|мо[её]\s+имя)\s+([^,\.!?]+)', query, re.IGNORECASE)
+        if name_match:
+            facts.append(("identity", f"Меня зовут {name_match.group(1).strip()}."))
+
+        city_match = re.search(r'(?:я\s+(?:из|живу\s+в)\s+)([^,\.!?]+)', query, re.IGNORECASE)
+        if city_match:
+            facts.append(("identity", f"Я живу в {city_match.group(1).strip()}."))
+
+        work_match = re.search(r'(?:занимаюсь|работаю\s+над|делаю|пишу)\s+([^,\.!?]+)', query, re.IGNORECASE)
+        if work_match:
+            facts.append(("projects", f"Работает над {work_match.group(1).strip()}."))
+
+        project_match = re.search(r'(?:проект|мой\s+проект|у\s+меня\s+(?:есть\s+)?проект)\s+([^,\.!?]+)', query, re.IGNORECASE)
+        if project_match:
+            facts.append(("projects", f"Проект: {project_match.group(1).strip()}."))
+
+        skill_match = re.search(r'(?:знаю|умею|использую|пишу\s+на|стек|технологии?)\s*[\:\-]?\s*([^,\.!?]+)', query, re.IGNORECASE)
+        if skill_match:
+            facts.append(("knowledge", f"Знает/использует: {skill_match.group(1).strip()}."))
+
+        like_match = re.search(r'(?:нравится|люблю|предпочитаю)\s+([^,\.!?]+)', query, re.IGNORECASE)
+        if like_match:
+            facts.append(("preferences", f"Предпочитает {like_match.group(1).strip()}."))
+
+        stored = 0
+        for expert, fact in facts:
+            if self.store_fact(expert, fact):
+                stored += 1
+                logger.info("  💾 [%s] запомнил: %s", expert, fact[:60])
+        return stored
+
+    def get_stats(self) -> Dict[str, int]:
+        return {name: expert.count() for name, expert in self.experts.items()}
@@ -0,0 +1,12 @@
+name: mome
+description: "MoME — Mixture of Memory Experts. Sparse-gated personal memory with online-learning router for Hermes Agent."
+version: 0.1.0
+author: Emil Shanaty
+homepage: https://github.com/emil28092005/hermes-plugin-mome
+repository: https://github.com/emil28092005/hermes-plugin-mome
+tags: [memory, mome, moe, experts, personalization, routing]
+license: MIT
+python:
+  requirements:
+    - numpy>=1.24
+    - scikit-learn>=1.3