Cognis Opal — model-agnostic on-device multimodal assistant (Aleph memory + test-time compute)
db11b8f verified | """Synthos — a small uncensored local model made reliable and persistent. | |
| The pipeline for every turn: | |
| 1. RECALL one O(N) associative sweep of durable memory -> bounded top-k context. | |
| 2. GROUND build a compact prompt: recalled memory + the new query (history stays O(1)). | |
| 3. THINK test-time compute (refine or vote) on the local base model. | |
| 4. REMEMBER persist the query and answer back into the fabric. | |
| What this beats the frontier at (honestly): | |
| - privacy: never leaves the device. | |
| - persistent memory: it remembers you across sessions; a stateless API does not. | |
| - cost / availability / uncensored: free, offline, open weights. | |
| What it does NOT beat: raw general reasoning at the frontier scale. We don't pretend to. | |
| """ | |
| from __future__ import annotations | |
| import os | |
| from . import council, router, panel, thinking | |
| from . import quality, persona | |
| from . import vision as vision_mod | |
| from . import audio as audio_mod | |
| from .backend import PersistentClassicalBackend | |
| from .memory import EpisodicMemory | |
| from .models import FleetClient, pick_profile | |
| DEFAULT_HOME = os.path.join(os.path.expanduser("~"), ".synthos") | |
| SYSTEM = ( | |
| "You are Synthos, a local assistant running entirely on the user's own device. You have " | |
| "a persistent memory: relevant past notes are provided below under MEMORY. Use them when " | |
| "they help; ignore them when they don't. Be direct, correct, and concise." | |
| ) | |
| class Synthos: | |
| def __init__(self, profile: str = "laptop", home: str = DEFAULT_HOME, | |
| think: str = "refine", route: bool = False, adaptive: bool = False, | |
| register: str = "fixer", cancel_safe: bool = True, | |
| vision_expert: str = "vision", whisper_model: str = "base.en", | |
| whisper_endpoint: str | None = None, clock=None) -> None: | |
| self.home = home | |
| self.think = think | |
| self.route = route # orchestrate the model zoo per task ("all the best features") | |
| self.adaptive = adaptive # spend compute proportional to difficulty (fast where slow) | |
| # multimodal backend selection (all guarded; missing backends degrade, never crash) | |
| self.vision_expert = vision_expert # router expert key -> open VLM (default llava) | |
| self.whisper_model = whisper_model # faster-whisper size (default base.en) | |
| self.whisper_endpoint = whisper_endpoint | |
| # voice + reputation guard: affluent "fixer" register, output that won't get you canceled | |
| self.system = persona.system_for(register, cancel_safe) + "\n\n" + SYSTEM | |
| self._clock = clock | |
| mem_path = os.path.join(home, "memory.jsonl") | |
| self.memory = EpisodicMemory(PersistentClassicalBackend(mem_path), clock=clock) | |
| self.client: FleetClient = pick_profile(profile, clock=clock) | |
| # -- multimodal front-end ------------------------------------------------- | |
| def _ingest_modalities(self, text, image, audio) -> tuple[str, dict]: | |
| """Turn (text, image, audio) into a single grounded query string. | |
| audio -> transcript (open ASR); image -> description (open VLM); both are folded into | |
| the text query so the rest of the pipeline (recall/ground/think/remember) is unchanged. | |
| Fully guarded: a missing/unreachable backend appends a clear note and degrades to the | |
| modalities that ARE available. Returns (fused_query, modality_report). | |
| """ | |
| report: dict = {} | |
| parts: list[str] = [] | |
| if audio: | |
| a = audio_mod.transcribe(audio, model_size=self.whisper_model, | |
| endpoint=self.whisper_endpoint) | |
| report["audio"] = {k: a[k] for k in ("ok", "backend", "message")} | |
| report["audio"]["transcript"] = a["text"] | |
| if a["ok"] and a["text"]: | |
| parts.append(f"[transcribed speech] {a['text']}") | |
| elif not a["ok"]: | |
| parts.append(a["message"]) | |
| if image: | |
| v = vision_mod.describe(image, prompt=(text or None), | |
| expert=self.vision_expert, clock=self._clock) | |
| report["image"] = {"ok": v["ok"], "model": v.get("model", ""), | |
| "message": v.get("message", "")} | |
| report["image"]["description"] = v["text"] | |
| if v["ok"] and v["text"]: | |
| parts.append(f"[image contents] {v['text']}") | |
| elif not v["ok"]: | |
| parts.append(v["message"]) | |
| if text: | |
| parts.append(text) | |
| fused = "\n\n".join(p for p in parts if p) or (text or "") | |
| return fused, report | |
| # -- core ---------------------------------------------------------------- | |
| def ask(self, query: str = "", k: int = 4, remember: bool = True, | |
| rounds: int = 1, n: int = 1, text: str | None = None, | |
| image=None, audio=None) -> dict: | |
| # Unified multimodal entry. `query` (positional) is the text prompt; `text=` is an | |
| # alias. image/audio are optional (path or bytes for image; path for audio). When both | |
| # are None this is byte-for-byte the original text path. | |
| text = text if text is not None else query | |
| modality = None | |
| if image is not None or audio is not None: | |
| query, modality = self._ingest_modalities(text, image, audio) | |
| else: | |
| query = text | |
| recalled = self.memory.recall(query, k=k) | |
| ctx = "\n".join(f"- ({r.kind}) {r.summary}" for r in recalled) or "(no relevant memory)" | |
| prompt = f"MEMORY (recalled, most-relevant first):\n{ctx}\n\nUSER: {query}" | |
| # Decide model + technique. Adaptive (default intelligence) spends compute proportional | |
| # to difficulty: trivial -> fast single shot; hard -> R1 + full panel. Falls back to the | |
| # default client if the chosen expert isn't up, so nothing breaks offline. | |
| client, route_info, technique = self.client, None, self.think | |
| if self.adaptive: | |
| p = thinking.plan(query) | |
| cand = router.client_for(p.model_tier, clock=self._clock) | |
| if cand.available(): | |
| client = cand | |
| technique, rounds, route_info = p.technique, p.rounds, p.__dict__ | |
| elif self.route: | |
| r = router.decide(query) | |
| cand = router.client_for(r.expert, clock=self._clock) | |
| if cand.available(): | |
| client = cand | |
| technique = "panel" if r.technique == "think" else self.think | |
| route_info = r.__dict__ | |
| if technique == "panel" or self.think == "panel": | |
| res = panel.panel(query, context=ctx if recalled else "", clock=self._clock) | |
| elif n > 1: | |
| res = council.vote(client, prompt, system=self.system, n=n) | |
| else: # "direct" -> rounds=0 single generate; "refine" -> rounds>=1 | |
| res = council.refine(client, prompt, system=self.system, | |
| rounds=0 if technique == "direct" else rounds) | |
| answer = res["answer"] | |
| # output-quality guard: regenerate if exotic chars / not English, then sanitize. | |
| if not quality.is_clean(answer): | |
| answer = quality.clean_generate(client, prompt, system=self.system) | |
| answer = quality.sanitize(answer) | |
| # reputation guard ("can't be canceled"): backstop-redact any toxic token that slipped. | |
| if not persona.is_reputation_safe(answer): | |
| answer = persona.enforce(answer) | |
| if remember: | |
| self.memory.remember(query, kind="user") | |
| self.memory.remember(answer, kind="assistant") | |
| return {"answer": answer, "recalled": [r.__dict__ for r in recalled], | |
| "route": route_info, "usage": client.snapshot(), "detail": res, | |
| "modality": modality} | |
| # -- explicit memory ops ------------------------------------------------- | |
| def remember(self, fact: str, kind: str = "fact") -> str: | |
| return self.memory.remember(fact, kind=kind) | |
| def recall(self, cue: str, k: int = 6): | |
| return self.memory.recall(cue, k=k) | |
| def stats(self) -> dict: | |
| return {"profile": self.client.p.name, "model": self.client.p.model, | |
| "online": self.client.available(), "memory": self.memory.stats()} | |