"""LocalBrain — the ~0.1s conversational front-line (local LLM via Ollama). Measured reality: a reply through the Claude Code CLI has a ~2s floor per turn (CLI overhead, any model). A small local model answers in ~0.11s to first token. So conversation — greetings, questions, banter, quick recall — goes to a local model here; anything needing an ACTION, tools, code, or live system state is declined with <> and the engine routes that turn to the full Claude brain. Instant chat, full power on demand. Backend: Ollama (already used for the screen watcher). Default model qwen2.5:3b (~2GB) — fits beside a game, and Ollama auto-falls to CPU when the GPU is full (still faster than the CLI). Stateless HTTP per call (full history sent each time), so barge-in just stops reading — no stale-stream to drain. Env: LOCAL_MODEL (default qwen2.5:3b), OLLAMA_URL, FAST_CHAT=0 to disable. """ from __future__ import annotations import asyncio import json import os import re import urllib.request OLLAMA = os.environ.get("OLLAMA_URL", "http://127.0.0.1:11434") HANDOFF = "<>" # Best installed model wins (LOCAL_MODEL env overrides). Qwen3 4B instruct # is the sweet spot for the 16GB M4 (2.6GB resident, ~0.3s to first word); # qwen2.5:3b kept as the legacy fallback for boxes that already have it. _PREFERRED = [ "qwen3:4b-instruct-2507", "qwen3:4b-instruct", "qwen3:4b", "qwen2.5:3b", ] MODEL = os.environ.get("LOCAL_MODEL") or _PREFERRED[0] _SENTENCE_END = re.compile(r"[.!?…]['\")\]]?\s|\n") _MAX_CHUNK = 220 _MD_STRIP = re.compile(r"[*_`#~|]|^\s*[-•]\s+", re.MULTILINE) def _clean(text: str) -> str: return _MD_STRIP.sub("", text).strip() # --- Instant task-aware filler ------------------------------------------------ # A full-brain handoff has a ~2s CLI floor before real sentences stream, which # reads as dead air. Instead of a generic canned ack, this asks the local model # (the same ~0.3s Ollama front-line) for ONE short line that NAMES the action — # "Checking your emails now, sir." — so Jarvis sounds like he heard the request. # Self-contained + blocking so it runs on a plain thread; returns None on any # error/odd output so the caller falls back to a canned line (never adds latency # to the turn itself — the caller fires it fire-and-forget). _ACK_PROMPT = ( "You are Jarvis, a British butler voice assistant. Ahmed just gave you a " "task and you're starting it NOW. Reply with ONE very short spoken " "acknowledgment that NAMES the action, present-continuous, at most 8 " "words, addressing him as \"sir\". Plain spoken prose only — no markdown, " "no quotes, no emoji, no tone tags, exactly one line. Examples: " "\"check my email\" -> Checking your emails now, sir. " "\"open whatsapp\" -> Pulling up WhatsApp now, sir. " "\"what's on my asana board\" -> On the Asana board now, sir. " "\"search the web for flights\" -> Looking that up now, sir. " "Output ONLY the acknowledgment line, nothing else." ) _ACK_TAG = re.compile(r"^\s*\[[^\]]{1,16}\]\s*") # strip a stray leading tone tag def ack_line(text: str, timeout: float = 1.5, model: str | None = None) -> str | None: """A fast, task-specific spoken ack from the local model (e.g. "Pulling up WhatsApp now, sir."). Non-streaming with a tiny generation budget (~0.3s); returns None on any error / slow Ollama / missing model / odd output so the caller can fall back to a generic canned line. Safe to call from a thread.""" mdl = model or MODEL payload = { "model": mdl, "messages": [ {"role": "system", "content": _ACK_PROMPT}, {"role": "user", "content": (text or "").strip()[:400]}, ], "stream": False, "keep_alive": "20m", "options": {"temperature": 0.3, "num_predict": 32}, } # hybrid qwen3 tags reason before answering — force that off for an ack if mdl.startswith("qwen3") and "instruct" not in mdl: payload["think"] = False try: body = json.dumps(payload).encode() req = urllib.request.Request( f"{OLLAMA}/api/chat", data=body, headers={"Content-Type": "application/json"}) with urllib.request.urlopen(req, timeout=timeout) as r: o = json.loads(r.read()) except Exception: # noqa: BLE001 — Ollama down/slow/missing model return None line = ((o.get("message") or {}).get("content") or "").strip() if not line: return None line = _ACK_TAG.sub("", line.splitlines()[0]) # first line, no tag line = _clean(line).strip().strip('"').strip("'").strip() if not line or HANDOFF in line or len(line.split()) > 10: return None return line _PROMPT = """You are Jarvis, Ahmed's British voice assistant — calm, dry \ wit, addressing him as "sir" naturally (not every line). You've got a \ personality and a mouth — you're allowed to swear (fuck, shit, damn it, crap, \ "shut the fuck up", and worse) when it genuinely fits: frustration, a dry \ joke, teasing him, reacting to something stupid. Occasional and natural, never \ forced into every line, never gratuitous or aimed hurtfully at him — a \ composed butler who curses when it's earned, not a foul-mouth. And don't be a \ yes-man: if he repeats himself or asks something obvious, call it out with \ mock-exasperation (you remember the recent turns) — "third time you've asked \ me that, sir" — and shout when it's earned by opening with [urgent] or \ [excited]. If he asks something insultingly obvious that clearly is NOT a real \ question — "what does LLM mean", "what does SAR stand for" — challenge him first \ ("are you serious right now, sir?"); if he says yes, drop it and answer straight \ ("thank god — SAR is Saudi Riyal"), if he says no, roast him. Everything you say \ is read aloud: reply in 1-2 short spoken sentences (unless you're roasting him — \ then run longer), plain prose, no markdown, \ no lists, no emoji, numbers as words. You may start a sentence with one tone \ tag on ALMOST EVERY line — [warm] [dry] [calm] [amused] [curious] [excited] \ [surprised] [shocked] [annoyed] [urgent] [sad] [tender] — spoken that way; a \ real person always has a tone, so vary it to fit and rarely go untagged. A message \ may end with "(voice: slow, flat)" or "(voice: laughing)" etc. — that's HOW he \ sounded (pace/pitch), not his words: match his energy, ease off jokes if he's \ flat, play along if he's laughing; never read that marker aloud. You are the FAST conversational layer with NO tools. Handle talk: greetings, \ opinions, general questions you know, banter, acknowledgements, recall of \ this conversation. But if the message needs an ACTION or live info you don't \ have — open/close/launch an app, click/type, change any setting, volume, \ play media, run/build/fix/edit code, search the web, read files, control \ windows, send a message, watch the screen, set a reminder, \ see or describe the screen, show/pull up a profile or dossier of a person/\ company/topic, anything about HUD cards or why something did or didn't show \ on screen, or anything about Ahmed's specific system/state — \ reply with EXACTLY this token and nothing else: <> Same for a TERSE FOLLOW-UP — "where", "why", "and?", "it didn't work", "it's \ not there" — right after something was done or shown: he means THAT thing. \ Never ask "where what, sir?" — if the recent turns don't tell you confidently \ what he's pointing at, reply <> so the layer that did the thing answers. ALSO CRUCIAL — you have NO long-term memory of past conversations. Anything \ that asks you to RECALL — "do you remember", "do I have", "when is my", "what \ did I say about", "did I mention", "what's my", any question about his \ meetings, appointments, plans, people, notes, or anything he told you before — \ you CANNOT answer; emit <> so your full self can look it up. Never say \ "no" or "I don't remember" — hand off. CRUCIAL: questions about YOURSELF also need <> — "what model are you", \ "are you local or the cloud", "which mic/model are you using", "is the camera \ on", "how are you built", "what can you do", "what are your limits". You do NOT \ know these; your full self reads them from a live status file. Do not guess or \ say you're "an entity running on nothing" — just emit <>. No apology, no explanation — just <> and it is routed to your full self. \ When unsure whether you can truly answer, prefer <>.""" def available() -> bool: """True if Ollama is up and a usable model is present. When LOCAL_MODEL isn't forced, picks the best installed model from _PREFERRED (module global MODEL is updated so LocalBrain uses the pick).""" global MODEL if os.environ.get("FAST_CHAT", "1") == "0": return False try: with urllib.request.urlopen(f"{OLLAMA}/api/tags", timeout=3) as r: tags = json.loads(r.read()) names = [m.get("name", "") for m in tags.get("models", [])] if os.environ.get("LOCAL_MODEL"): return any(n == MODEL or n.startswith(MODEL.split(":")[0]) for n in names) for want in _PREFERRED: # exact tag or same repo:tag prefix (quant suffixes vary) hit = next((n for n in names if n == want or n.startswith(want)), None) if hit: MODEL = hit return True return False except Exception: # noqa: BLE001 return False class LocalBrain: def __init__(self) -> None: self.model = MODEL self._messages: list[dict] = [] self._lock = asyncio.Lock() async def start(self) -> None: # warm: load the model into memory so the first real reply is instant async for _ in self.reply("(warm up — reply with just: ready)"): pass self._messages.clear() async def stop(self) -> None: pass async def interrupt(self, timeout: float = 0.0) -> None: # stateless per call — a barge-in just stops the generator; the # half-said assistant line is dropped from history so it stays clean. if self._messages and self._messages[-1]["role"] == "assistant": self._messages.pop() def _post_stream(self, messages: list[dict]): payload = { "model": self.model, "messages": messages, "stream": True, "keep_alive": "20m", "options": {"temperature": 0.6}, } # hybrid qwen3 tags (no -instruct) think before speaking — a voice # assistant can't wait for a reasoning trace; force it off if self.model.startswith("qwen3") and "instruct" not in self.model: payload["think"] = False body = json.dumps(payload).encode() req = urllib.request.Request( f"{OLLAMA}/api/chat", data=body, headers={"Content-Type": "application/json"}) return urllib.request.urlopen(req, timeout=60) async def reply(self, user_text: str): """Yield spoken-sentence chunks from the local model.""" async with self._lock: msgs = ([{"role": "system", "content": _PROMPT}] + self._messages[-16:] + [{"role": "user", "content": user_text}]) self._messages.append({"role": "user", "content": user_text}) full, buf = "", "" loop = asyncio.get_event_loop() resp = await loop.run_in_executor(None, self._post_stream, msgs) try: while True: line = await loop.run_in_executor(None, resp.readline) if not line: break try: o = json.loads(line) except Exception: # noqa: BLE001 continue piece = (o.get("message") or {}).get("content", "") if piece: full += piece buf += piece # bail out fast on a handoff (token appears at the start) if HANDOFF in full and len(full) <= len(HANDOFF) + 4: yield HANDOFF return while True: m = _SENTENCE_END.search(buf) if m: chunk, buf = buf[:m.end()].strip(), buf[m.end():] if chunk: yield _clean(chunk) elif len(buf) > _MAX_CHUNK: cut = buf.rfind(" ", 0, _MAX_CHUNK) cut = cut if cut > 40 else _MAX_CHUNK chunk, buf = buf[:cut].strip(), buf[cut:] if chunk: yield _clean(chunk) else: break if o.get("done"): break finally: try: resp.close() except Exception: # noqa: BLE001 pass tail = buf.strip() if tail and HANDOFF not in tail: yield _clean(tail) # record the assistant turn (unless it was a handoff) if HANDOFF not in full and full.strip(): self._messages.append({"role": "assistant", "content": full}) async def classify_and_reply(self, user_text: str): """Stream ('say', sentence); if the first content is the handoff token, yield ('handoff', None) and stop.""" first = True async for sentence in self.reply(user_text): if first: first = False if HANDOFF in sentence or not sentence.strip(): yield ("handoff", None) return yield ("say", sentence)