"""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()}