From d16df7428d19650b5112f8b50d3812753f7f6140 Mon Sep 17 00:00:00 2001 From: Paul Carff Date: Mon, 7 Sep 2026 15:36:57 -0400 Subject: [PATCH 1/2] Fix startup STT warmup deadlock, add instant PTT release feedback, and support configurable TTS device --- backtalk/config.py | 2 ++ backtalk/ears.py | 13 ++++++++++--- backtalk/main.py | 7 ++++++- backtalk/mouth.py | 5 +++-- 4 files changed, 21 insertions(+), 6 deletions(-) diff --git a/backtalk/config.py b/backtalk/config.py index d59f7f2..6b6571a 100644 --- a/backtalk/config.py +++ b/backtalk/config.py @@ -131,6 +131,8 @@ # language pipeline (a=American, b=British, e/f/h/i/j/p/z = other # languages), so keep voice and accent matched. "voice": "bm_lewis", + # Device for local Kokoro TTS pipeline: "cpu" (default) or "cuda" + "tts_device": "cpu", # Speech recognition (faster-whisper, local, free). # Models: tiny.en / base.en / small.en / medium.en — small.en is the # accuracy/speed sweet spot on a normal machine. diff --git a/backtalk/ears.py b/backtalk/ears.py index ad52343..6cb9b66 100644 --- a/backtalk/ears.py +++ b/backtalk/ears.py @@ -289,12 +289,13 @@ def warm(): verbose=None) _model, _backend = repo, "mlx" else: - from faster_whisper import WhisperModel want = CFG["stt_device"] + compute = CFG["stt_compute"] log(f"[ears] loading {CFG['stt_model']} " - f"({want}/{CFG['stt_compute']})...") + f"({want}/{compute})...") + from faster_whisper import WhisperModel _model = WhisperModel(CFG["stt_model"], device=want, - compute_type=CFG["stt_compute"]) + compute_type=compute) # PROVE the device before the greeting, not at the first # spoken sentence. WhisperModel CONSTRUCTS perfectly well # against a GPU it cannot actually use: "auto" picks CUDA @@ -418,6 +419,12 @@ def record_held(is_held, max_s: float = 60.0, min_s: float = 0.25) -> str | None frames.append(block[:, 0].copy()) if len(frames) * FRAME_MS / 1000 < min_s: return None + try: + from backtalk.signals import set_state + set_state("thinking") + except Exception: + pass + print("[ptt] transcribing...", flush=True) return transcribe(np.concatenate(frames)) diff --git a/backtalk/main.py b/backtalk/main.py index a7b85d1..6403a81 100644 --- a/backtalk/main.py +++ b/backtalk/main.py @@ -679,7 +679,7 @@ async def amain(): loop = asyncio.get_event_loop() # Warm the engines while the greeting plays: the STT model load and # the brain's prompt-cache toll both hide behind the spoken line. - loop.run_in_executor(None, warm_ears) + warm_ears_fut = loop.run_in_executor(None, warm_ears) # THE BRAIN CONNECT, guarded. This is the one startup step that # needs a signed-in Claude Code, internet, and available usage. # When it fails or hangs, the mouth still works, so SAY SO instead @@ -707,6 +707,11 @@ async def _warmup(): mouth.wait_done(timeout=30) raise SystemExit(1) log("[backtalk] brain warm") + try: + await asyncio.wait_for(asyncio.shield(warm_ears_fut), 60) + log("[backtalk] ears warm and ready") + except Exception as e: + log(f"[ears] warmup error: {e!r}") # the hidden warmup ping is plumbing, not conversation brain.session.update(turns=0, out_tokens=0, in_tokens=0, cost=0.0) # a configured effort level applies at launch (saved by the spoken diff --git a/backtalk/mouth.py b/backtalk/mouth.py index c3bc4f5..b7428f7 100644 --- a/backtalk/mouth.py +++ b/backtalk/mouth.py @@ -171,8 +171,9 @@ def warm(): lang = (CFG["voice"] or "bm_lewis")[0] log(f"[mouth] loading kokoro (lang '{lang}', " f"voice {CFG['voice']})...") - _pipe = KPipeline(lang_code=lang) - log("[mouth] voice ready") + device = CFG.get("tts_device", "cpu") + _pipe = KPipeline(lang_code=lang, device=device) + log(f"[mouth] voice ready ({device})") return _pipe From c9c5013389c999cd51c16760aef912bbd3571b13 Mon Sep 17 00:00:00 2001 From: Paul Carff Date: Mon, 7 Sep 2026 15:44:28 -0400 Subject: [PATCH 2/2] feat(brain): add pluggable brain architecture with local OpenAI-compatible LLM support - Refactor brain module into a pluggable dispatcher (backtalk/brain.py). - Preserve 100% backward-compatible Claude Agent SDK provider in backtalk/brain_claude.py as the default. - Introduce LocalWarmBrain in backtalk/brain_local.py supporting OpenAI-compatible endpoints (llama-server, Ollama, vLLM, Aphrodite) with low-latency sentence-boundary streaming. - Implement recursive tree inspection, workspace directory switching, file reading, command execution, and guarded web search for local models. - Add configuration keys 'brain' ('claude' | 'local') and 'api_base' in backtalk/config.py. - Provide backtalk.local.json.example and update README.md documentation. --- README.md | 16 ++ backtalk.local.json.example | 13 + backtalk/brain.py | 362 ++---------------------- backtalk/brain_claude.py | 370 ++++++++++++++++++++++++ backtalk/brain_local.py | 546 ++++++++++++++++++++++++++++++++++++ backtalk/config.py | 8 +- backtalk/main.py | 15 +- 7 files changed, 983 insertions(+), 347 deletions(-) create mode 100644 backtalk.local.json.example create mode 100644 backtalk/brain_claude.py create mode 100644 backtalk/brain_local.py diff --git a/README.md b/README.md index e8f2def..d408741 100644 --- a/README.md +++ b/README.md @@ -54,6 +54,22 @@ Two engines, and the setup wizard offers you both instead of quietly defaulting. **ElevenLabs, the natural one.** The human-sounding voice most people actually want, on your own API key. The free tier is enough to audition it; day-to-day talking runs on the paid starter plan. The wizard walks the whole thing with you: account, key into the keychain, then an audition of real voices through backtalk's own mouth until one fits. Want the exact voice from my videos? It's called **Tarquin** in the ElevenLabs voice library: search it by name and you're done hunting. Under the hood it is: set `elevenlabs.enabled` and your `voice_id` in the config, and have `ffmpeg` installed. **The key never goes in a file.** On macOS, seed it into the Keychain once with `security add-generic-password -a "$USER" -s backtalk-elevenlabs -T /usr/bin/security -w` (it prompts for the secret) and backtalk reads it from there. Linux: `secret-tool store --label backtalk service backtalk-elevenlabs`. The `ELEVENLABS_API_KEY` environment variable works as a last resort, but an export in a shell profile is a plaintext key on disk; the keychain is the grown-up path. Kokoro stays wired in as the automatic fallback, so if the cloud fails the voice degrades instead of going mute, and `logs/backtalk.log` records why. +## Local AI Brain (Optional) + +While backtalk defaults to Claude Code via the Agent SDK, it also supports local OpenAI-compatible inference servers (such as `llama-server`, Ollama, vLLM, or Aphrodite) for a 100% offline, private voice assistant. + +To switch to a local model, set `"brain": "local"` in your `backtalk.json`: + +```json +{ + "brain": "local", + "api_base": "http://127.0.0.1:8080/v1", + "model": "default" +} +``` + +The local brain maintains multi-step tool execution (recursive directory inspection, file reading, command execution, and web browsing) and streams partial sentences directly to the voice pipeline with sentence boundary chunking. See `backtalk.local.json.example`. + ## Give it a face (optional) backtalk writes tiny state files while it listens, thinks, and speaks, so anything can watch them and react in real time. diff --git a/backtalk.local.json.example b/backtalk.local.json.example new file mode 100644 index 0000000..8c90a9c --- /dev/null +++ b/backtalk.local.json.example @@ -0,0 +1,13 @@ +{ + "brain": "local", + "api_base": "http://127.0.0.1:8080/v1", + "model": "default", + "agent_dir": "~", + "name": "Assistant", + "ptt_key": "home", + "voice": "bm_lewis", + "tts_device": "cpu", + "stt_model": "small.en", + "stt_device": "auto", + "extra_dirs": [] +} diff --git a/backtalk/brain.py b/backtalk/brain.py index 4ac7c25..9b422d8 100644 --- a/backtalk/brain.py +++ b/backtalk/brain.py @@ -1,5 +1,5 @@ -# backtalk: talk to your Claude Code agent out loud. -# Copyright (C) 2026 Jared Rhodenizer +# backtalk: talk to your agent out loud. +# Copyright (C) 2026 Jared Rhodenizer, AnZym contributors # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU Affero General Public License as published @@ -15,354 +15,34 @@ # along with this program. If not, see . # # SPDX-License-Identifier: AGPL-3.0-or-later -"""The warm brain — a persistent Claude session via the Agent SDK, -streaming. +"""The warm brain router — dispatches to the configured brain provider. -One ClaudeSDKClient lives for the whole voice session: no per-turn -process spawn, no per-turn context reload. Partial-message streaming -means sentences are yielded the moment they're complete, so the mouth -starts speaking while the rest of the thought is still forming. - -The session's cwd is YOUR agent's folder (agent_dir in backtalk.json) — -whatever CLAUDE.md lives there defines who is speaking. backtalk adds -only the spoken-delivery discipline (config.DISCIPLINE): the medium, -never the character. +Supports: +- "claude" (default): Claude Agent SDK session (brain_claude.py) +- "local" / "openai": Local OpenAI-compatible server (brain_local.py) """ -import asyncio import os -import re -import warnings -from datetime import datetime - -from claude_agent_sdk import ClaudeAgentOptions, ClaudeSDKClient - -try: - from claude_agent_sdk import CanUseToolShadowedWarning -except ImportError: # older SDKs: nothing to silence - CanUseToolShadowedWarning = None - -from backtalk import signals -from backtalk.config import CFG, DISCIPLINE -from backtalk.vlog import log +from backtalk.config import CFG -_SENTENCE_END = re.compile(r"(?<=[.!?])\s") +SESSION_FILE = os.path.join(CFG.get("signals_dir", "/tmp/signals"), ".backtalk_session") -SESSION_FILE = os.path.join(CFG["signals_dir"], ".backtalk_session") +def get_brain_class(): + """Resolve the brain implementation class based on configuration.""" + brain_type = str(CFG.get("brain", "claude")).strip().lower() + if brain_type in ("local", "openai", "llama", "vllm", "ollama"): + from backtalk.brain_local import LocalWarmBrain + return LocalWarmBrain + from backtalk.brain_claude import ClaudeWarmBrain + return ClaudeWarmBrain class WarmBrain: - def __init__(self, model: str | None = None, can_use_tool=None, - resume_id: str | None = None): - # Full model id ON PURPOSE — never a bare alias. The SDK - # resolves aliases through its own bundled CLI and can silently - # land on an older model. - self.model = model or CFG["model"] - # The spoken permission gate (main.py builds it). Wired at - # connect in EVERY mode, so a live mode flip needs no reconnect; - # bypass simply never consults it. - self._can_use_tool = can_use_tool - # Session usage, spoken on request ("usage report"). - self.session = {"turns": 0, "out_tokens": 0, "in_tokens": 0, - "cost": 0.0} - self._client: ClaudeSDKClient | None = None - # The session to reattach to at the FIRST start only (config key - # resume_last_session). Consumed on use: a desync rebuild in - # reset_turn() must always start FRESH: a rebuild means a turn - # went sideways mid-stream, the wrong moment to gamble on - # reattaching. (Community proposal, issue #1.) - self._resume_id = resume_id - # True while a query's response hasn't been consumed through its - # ResultMessage — i.e. the shared message pipe may hold leftovers. - self._dirty = False - - async def start(self): - mode = CFG["permission_mode"] - if mode == "default": - mode = "ask" # legacy alias, see config.py - # backtalk's "ask" = the SDK's "default" mode with gated calls - # routed to the spoken can_use_tool gate. - sdk_mode = "default" if mode == "ask" else mode - if sdk_mode == "bypassPermissions" and self._can_use_tool \ - and CanUseToolShadowedWarning: - # Deliberate auto-approve: the SDK warns that the callback is - # shadowed. That IS the chosen behavior, so boot quietly. - warnings.filterwarnings("ignore", - category=CanUseToolShadowedWarning) - resume, self._resume_id = self._resume_id, None # consume once - - def _opts(rid): - return ClaudeAgentOptions( - cwd=CFG["agent_dir"], - model=self.model, - system_prompt={"type": "preset", "preset": "claude_code", - "append": DISCIPLINE}, - include_partial_messages=True, - permission_mode=sdk_mode, - can_use_tool=self._can_use_tool, - add_dirs=CFG["extra_dirs"], - skills=CFG["visible_skills"], - resume=rid, - ) - if resume: - try: - self._client = ClaudeSDKClient(options=_opts(resume)) - await self._client.connect() - log(f"[brain] resumed session {resume[:8]}") - return - except Exception as e: - # a stale or invalid saved session must never brick the - # launch. Fall back to a fresh conversation and say so. - log(f"[brain] resume failed ({str(e)[:80]}), " - f"starting fresh") - try: - await self._client.disconnect() - except Exception: - pass - self._client = ClaudeSDKClient(options=_opts(None)) - await self._client.connect() - - async def set_permission_mode(self, backtalk_mode: str): - """Live flip, no reconnect, conversation intact ("ask" maps to - the SDK's "default", whose gated calls hit the spoken gate).""" - if self._client: - sdk_mode = "default" if backtalk_mode == "ask" \ - else backtalk_mode - await self._client.set_permission_mode(sdk_mode) - - async def context_usage(self): - """The CLI's own context-window breakdown, or None.""" - try: - return await self._client.get_context_usage() - except Exception: - return None - - def _remember_session(self, rm): - """Persist the session id after a completed turn, so the next - launch can reattach (config: resume_last_session). Must never - break a turn; silence on any failure.""" - if not CFG.get("resume_last_session"): - return - sid = getattr(rm, "session_id", None) - if not sid: - return - try: - with open(SESSION_FILE, "w") as f: - f.write(sid) - except OSError: - pass - - def _tally(self, rm, count_turn=True): - """Session usage bookkeeping. Must never break a turn.""" - try: - u = getattr(rm, "usage", None) or {} - s = self.session - if count_turn: - s["turns"] += 1 - s["out_tokens"] += int(u.get("output_tokens") or 0) - s["in_tokens"] += (int(u.get("input_tokens") or 0) - + int(u.get("cache_read_input_tokens") - or 0)) - c = getattr(rm, "total_cost_usd", None) - if c: - s["cost"] += float(c) - except Exception: - pass - - async def _pull_rate_limits(self): - """Ask the CLI outright how much of the plan is spent. - - A DIRECT QUERY, not the RateLimitEvent stream. The event fires - rarely and usually arrives carrying resets_at with no utilization - at all, so a listener built on it reports nothing most of the - time -- which is exactly how this feature looked broken for its - whole life. (Community fix, ai-visualizer issue #1.) - - THIS REACHES PAST THE SDK'S PUBLIC SURFACE ON PURPOSE, and a - reader should know it rather than discover it. `get_usage` is a - control request the bundled CLI answers but the SDK never wraps, - so there is no supported call to make. The supported-looking - alternative is a dead end and was tested as one: the terminal - status line never fires in a headless session, so its numbers - are unreachable from here. - - Which means this can stop working without anyone doing anything - wrong, and the containment is the point. Every failure is - swallowed and the readout simply goes quiet. It must never cost - a turn, so it is also bounded -- an unanswered control request - would otherwise hang the voice line mid-conversation.""" - if not CFG.get("show_usage"): - return - try: - usage = await asyncio.wait_for( - self._client._query._send_control_request( - {"subtype": "get_usage"}), 5) - for window in ("five_hour", "seven_day"): - w = (usage.get("rate_limits") or {}).get(window) - if not w: - continue - # Two spellings accepted deliberately: this shape is not - # documented anywhere, so the cheap tolerance is worth - # more than the tidiness. Both are percentages, and the - # rest of the pipeline wants a 0..1 fraction. - pct = w.get("utilization") - if pct is None: - pct = w.get("used_percentage") - pct = pct / 100 if pct is not None else None - resets = w.get("resets_at") - if isinstance(resets, str): - resets = int(datetime.fromisoformat(resets).timestamp()) - signals.set_rate_limit(window, pct, resets) - except Exception: - pass - - async def command(self, cmd: str) -> str: - """Run a console slash command (/clear, /compact, /model, - /effort) through the normal stream and return whatever text the - CLI answered with (confirmations, errors). Slash-command replies - arrive as COMPLETE AssistantMessages, not stream deltas, so - ask_stream cannot see them. Bounded like reset_turn is: this - stream is not trusted to always deliver, and an unbounded await - here would deafen the whole voice loop. On timeout the pipe is - left marked dirty so the next reset_turn drains or rebuilds.""" - self._dirty = True - await self._client.query(cmd) - texts = [] - - async def _collect(): - async for msg in self._client.receive_response(): - t = type(msg).__name__ - if t == "AssistantMessage": - for b in getattr(msg, "content", []) or []: - txt = getattr(b, "text", None) - if txt: - texts.append(txt) - elif t == "ResultMessage": - self._dirty = False - self._tally(msg, count_turn=False) - self._remember_session(msg) - break - - try: - await asyncio.wait_for(_collect(), 90) - except asyncio.TimeoutError: - log(f"[brain] console command timed out: {cmd!r}") - return "error: the command timed out" - return " ".join(texts).strip() - - async def interrupt(self): - if self._client: - await self._client.interrupt() - - async def reset_turn(self, timeout: float = 8.0): - """Re-align the message pipe after an interrupted/failed turn. - - THE OFF-BY-ONE BUG, and why this method exists: the SDK client - has ONE shared message stream and receive_response() stops at - the FIRST ResultMessage it sees — there is no pairing between a - query and its response. A cancelled turn stops consuming - mid-stream, leaving the dead turn's remaining messages - (including its ResultMessage) buffered. The next query then - pairs with those leftovers: the first ask lands on the stale - ResultMessage and yields nothing, and every ask after that - answers the PREVIOUS question — for the rest of the session. - So: interrupt the dead turn, then drain the pipe through its - stale ResultMessage before the next query goes out. No-op when - the last turn was consumed clean.""" - if not self._client or not self._dirty: - return - try: - await asyncio.wait_for(self._client.interrupt(), 5) - except Exception: - pass # turn may already be over — the drain below is the point - - async def _drain() -> int: - n = 0 - async for msg in self._client.receive_response(): - n += 1 - if type(msg).__name__ == "ResultMessage": - break - return n - - try: - drained = await asyncio.wait_for(_drain(), timeout) - log(f"[brain] interrupted turn drained ({drained} stale messages)") - self._dirty = False - except Exception: - # Can't re-align — rebuild the session rather than run - # desynced. Loses this voice session's conversation memory; - # better than answering every question one turn late for the - # rest of the day. - log("[brain] stream desynced beyond repair — rebuilding the " - "session (conversation memory for this session resets)") - try: - await self._client.disconnect() - except Exception: - pass - self._client = None - await self.start() - self._dirty = False - - async def stop(self): - if self._client: - await self._client.disconnect() - self._client = None - - async def ask_stream(self, utterance: str): - """Yield complete sentences as they stream out of the model.""" - self._dirty = True # in flight until its ResultMessage - await self._client.query(utterance) - buf = "" - async for msg in self._client.receive_response(): - t = type(msg).__name__ - if t == "StreamEvent": - ev = getattr(msg, "event", {}) or {} - if ev.get("type") == "content_block_delta": - delta = ev.get("delta", {}) or {} - if delta.get("type") == "text_delta": - buf += delta.get("text", "") - # emit any complete sentences - while True: - m = _SENTENCE_END.search(buf) - if not m: - break - sentence, buf = (buf[:m.end()].strip(), - buf[m.end():]) - if sentence: - yield sentence - elif ev.get("type") == "content_block_stop": - # End of a speech block (e.g. right before a tool - # call): flush NOW. Without this, pre-tool filler - # ("On it — let me grab that.") sits silent in the - # buffer through the whole tool run, then plays - # glued to the answer: long dead air, then two - # thoughts at once. - tail = buf.strip() - buf = "" - if tail: - yield tail - elif t == "ResultMessage": - self._dirty = False # turn fully consumed — pipe aligned - self._tally(msg) - self._remember_session(msg) - await self._pull_rate_limits() - break - tail = buf.strip() - if tail: - yield tail - + """WarmBrain dispatcher: instantiates the configured brain backend.""" -if __name__ == "__main__": - import time + def __new__(cls, *args, **kwargs): + target_cls = get_brain_class() + return target_cls(*args, **kwargs) - async def demo(): - b = WarmBrain() - await b.start() - for prompt in ("Voice check: greet me in one sentence.", - "And what's two plus two, spoken like yourself?"): - t0 = time.time() - async for s in b.ask_stream(prompt): - print(f" ({time.time()-t0:4.1f}s) {s}", flush=True) - await b.stop() - asyncio.run(demo()) +__all__ = ["WarmBrain", "SESSION_FILE", "get_brain_class"] diff --git a/backtalk/brain_claude.py b/backtalk/brain_claude.py new file mode 100644 index 0000000..7488f48 --- /dev/null +++ b/backtalk/brain_claude.py @@ -0,0 +1,370 @@ +# backtalk: talk to your Claude Code agent out loud. +# Copyright (C) 2026 Jared Rhodenizer +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as published +# by the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . +# +# SPDX-License-Identifier: AGPL-3.0-or-later +"""The warm brain — a persistent Claude session via the Agent SDK, +streaming. + +One ClaudeSDKClient lives for the whole voice session: no per-turn +process spawn, no per-turn context reload. Partial-message streaming +means sentences are yielded the moment they're complete, so the mouth +starts speaking while the rest of the thought is still forming. + +The session's cwd is YOUR agent's folder (agent_dir in backtalk.json) — +whatever CLAUDE.md lives there defines who is speaking. backtalk adds +only the spoken-delivery discipline (config.DISCIPLINE): the medium, +never the character. +""" +import asyncio +import os +import re +import warnings +from datetime import datetime + +from claude_agent_sdk import ClaudeAgentOptions, ClaudeSDKClient + +try: + from claude_agent_sdk import CanUseToolShadowedWarning +except ImportError: # older SDKs: nothing to silence + CanUseToolShadowedWarning = None + +from backtalk import signals +from backtalk.config import CFG, DISCIPLINE +from backtalk.vlog import log + +_SENTENCE_END = re.compile(r"(?<=[.!?])\s") + + +SESSION_FILE = os.path.join(CFG["signals_dir"], ".backtalk_session") + + +class WarmBrain: + def __init__(self, model: str | None = None, can_use_tool=None, + resume_id: str | None = None): + # Full model id ON PURPOSE — never a bare alias. The SDK + # resolves aliases through its own bundled CLI and can silently + # land on an older model. + self.model = model or CFG["model"] + # The spoken permission gate (main.py builds it). Wired at + # connect in EVERY mode, so a live mode flip needs no reconnect; + # bypass simply never consults it. + self._can_use_tool = can_use_tool + # Session usage, spoken on request ("usage report"). + self.session = {"turns": 0, "out_tokens": 0, "in_tokens": 0, + "cost": 0.0} + self._client: ClaudeSDKClient | None = None + # The session to reattach to at the FIRST start only (config key + # resume_last_session). Consumed on use: a desync rebuild in + # reset_turn() must always start FRESH: a rebuild means a turn + # went sideways mid-stream, the wrong moment to gamble on + # reattaching. (Community proposal, issue #1.) + self._resume_id = resume_id + # True while a query's response hasn't been consumed through its + # ResultMessage — i.e. the shared message pipe may hold leftovers. + self._dirty = False + + async def start(self): + mode = CFG["permission_mode"] + if mode == "default": + mode = "ask" # legacy alias, see config.py + # backtalk's "ask" = the SDK's "default" mode with gated calls + # routed to the spoken can_use_tool gate. + sdk_mode = "default" if mode == "ask" else mode + if sdk_mode == "bypassPermissions" and self._can_use_tool \ + and CanUseToolShadowedWarning: + # Deliberate auto-approve: the SDK warns that the callback is + # shadowed. That IS the chosen behavior, so boot quietly. + warnings.filterwarnings("ignore", + category=CanUseToolShadowedWarning) + resume, self._resume_id = self._resume_id, None # consume once + + def _opts(rid): + return ClaudeAgentOptions( + cwd=CFG["agent_dir"], + model=self.model, + system_prompt={"type": "preset", "preset": "claude_code", + "append": DISCIPLINE}, + include_partial_messages=True, + permission_mode=sdk_mode, + can_use_tool=self._can_use_tool, + add_dirs=CFG["extra_dirs"], + skills=CFG["visible_skills"], + resume=rid, + ) + if resume: + try: + self._client = ClaudeSDKClient(options=_opts(resume)) + await self._client.connect() + log(f"[brain] resumed session {resume[:8]}") + return + except Exception as e: + # a stale or invalid saved session must never brick the + # launch. Fall back to a fresh conversation and say so. + log(f"[brain] resume failed ({str(e)[:80]}), " + f"starting fresh") + try: + await self._client.disconnect() + except Exception: + pass + self._client = ClaudeSDKClient(options=_opts(None)) + await self._client.connect() + + async def set_permission_mode(self, backtalk_mode: str): + """Live flip, no reconnect, conversation intact ("ask" maps to + the SDK's "default", whose gated calls hit the spoken gate).""" + if self._client: + sdk_mode = "default" if backtalk_mode == "ask" \ + else backtalk_mode + await self._client.set_permission_mode(sdk_mode) + + async def context_usage(self): + """The CLI's own context-window breakdown, or None.""" + try: + return await self._client.get_context_usage() + except Exception: + return None + + def _remember_session(self, rm): + """Persist the session id after a completed turn, so the next + launch can reattach (config: resume_last_session). Must never + break a turn; silence on any failure.""" + if not CFG.get("resume_last_session"): + return + sid = getattr(rm, "session_id", None) + if not sid: + return + try: + with open(SESSION_FILE, "w") as f: + f.write(sid) + except OSError: + pass + + def _tally(self, rm, count_turn=True): + """Session usage bookkeeping. Must never break a turn.""" + try: + u = getattr(rm, "usage", None) or {} + s = self.session + if count_turn: + s["turns"] += 1 + s["out_tokens"] += int(u.get("output_tokens") or 0) + s["in_tokens"] += (int(u.get("input_tokens") or 0) + + int(u.get("cache_read_input_tokens") + or 0)) + c = getattr(rm, "total_cost_usd", None) + if c: + s["cost"] += float(c) + except Exception: + pass + + async def _pull_rate_limits(self): + """Ask the CLI outright how much of the plan is spent. + + A DIRECT QUERY, not the RateLimitEvent stream. The event fires + rarely and usually arrives carrying resets_at with no utilization + at all, so a listener built on it reports nothing most of the + time -- which is exactly how this feature looked broken for its + whole life. (Community fix, ai-visualizer issue #1.) + + THIS REACHES PAST THE SDK'S PUBLIC SURFACE ON PURPOSE, and a + reader should know it rather than discover it. `get_usage` is a + control request the bundled CLI answers but the SDK never wraps, + so there is no supported call to make. The supported-looking + alternative is a dead end and was tested as one: the terminal + status line never fires in a headless session, so its numbers + are unreachable from here. + + Which means this can stop working without anyone doing anything + wrong, and the containment is the point. Every failure is + swallowed and the readout simply goes quiet. It must never cost + a turn, so it is also bounded -- an unanswered control request + would otherwise hang the voice line mid-conversation.""" + if not CFG.get("show_usage"): + return + try: + usage = await asyncio.wait_for( + self._client._query._send_control_request( + {"subtype": "get_usage"}), 5) + for window in ("five_hour", "seven_day"): + w = (usage.get("rate_limits") or {}).get(window) + if not w: + continue + # Two spellings accepted deliberately: this shape is not + # documented anywhere, so the cheap tolerance is worth + # more than the tidiness. Both are percentages, and the + # rest of the pipeline wants a 0..1 fraction. + pct = w.get("utilization") + if pct is None: + pct = w.get("used_percentage") + pct = pct / 100 if pct is not None else None + resets = w.get("resets_at") + if isinstance(resets, str): + resets = int(datetime.fromisoformat(resets).timestamp()) + signals.set_rate_limit(window, pct, resets) + except Exception: + pass + + async def command(self, cmd: str) -> str: + """Run a console slash command (/clear, /compact, /model, + /effort) through the normal stream and return whatever text the + CLI answered with (confirmations, errors). Slash-command replies + arrive as COMPLETE AssistantMessages, not stream deltas, so + ask_stream cannot see them. Bounded like reset_turn is: this + stream is not trusted to always deliver, and an unbounded await + here would deafen the whole voice loop. On timeout the pipe is + left marked dirty so the next reset_turn drains or rebuilds.""" + self._dirty = True + await self._client.query(cmd) + texts = [] + + async def _collect(): + async for msg in self._client.receive_response(): + t = type(msg).__name__ + if t == "AssistantMessage": + for b in getattr(msg, "content", []) or []: + txt = getattr(b, "text", None) + if txt: + texts.append(txt) + elif t == "ResultMessage": + self._dirty = False + self._tally(msg, count_turn=False) + self._remember_session(msg) + break + + try: + await asyncio.wait_for(_collect(), 90) + except asyncio.TimeoutError: + log(f"[brain] console command timed out: {cmd!r}") + return "error: the command timed out" + return " ".join(texts).strip() + + async def interrupt(self): + if self._client: + await self._client.interrupt() + + async def reset_turn(self, timeout: float = 8.0): + """Re-align the message pipe after an interrupted/failed turn. + + THE OFF-BY-ONE BUG, and why this method exists: the SDK client + has ONE shared message stream and receive_response() stops at + the FIRST ResultMessage it sees — there is no pairing between a + query and its response. A cancelled turn stops consuming + mid-stream, leaving the dead turn's remaining messages + (including its ResultMessage) buffered. The next query then + pairs with those leftovers: the first ask lands on the stale + ResultMessage and yields nothing, and every ask after that + answers the PREVIOUS question — for the rest of the session. + So: interrupt the dead turn, then drain the pipe through its + stale ResultMessage before the next query goes out. No-op when + the last turn was consumed clean.""" + if not self._client or not self._dirty: + return + try: + await asyncio.wait_for(self._client.interrupt(), 5) + except Exception: + pass # turn may already be over — the drain below is the point + + async def _drain() -> int: + n = 0 + async for msg in self._client.receive_response(): + n += 1 + if type(msg).__name__ == "ResultMessage": + break + return n + + try: + drained = await asyncio.wait_for(_drain(), timeout) + log(f"[brain] interrupted turn drained ({drained} stale messages)") + self._dirty = False + except Exception: + # Can't re-align — rebuild the session rather than run + # desynced. Loses this voice session's conversation memory; + # better than answering every question one turn late for the + # rest of the day. + log("[brain] stream desynced beyond repair — rebuilding the " + "session (conversation memory for this session resets)") + try: + await self._client.disconnect() + except Exception: + pass + self._client = None + await self.start() + self._dirty = False + + async def stop(self): + if self._client: + await self._client.disconnect() + self._client = None + + async def ask_stream(self, utterance: str): + """Yield complete sentences as they stream out of the model.""" + self._dirty = True # in flight until its ResultMessage + await self._client.query(utterance) + buf = "" + async for msg in self._client.receive_response(): + t = type(msg).__name__ + if t == "StreamEvent": + ev = getattr(msg, "event", {}) or {} + if ev.get("type") == "content_block_delta": + delta = ev.get("delta", {}) or {} + if delta.get("type") == "text_delta": + buf += delta.get("text", "") + # emit any complete sentences + while True: + m = _SENTENCE_END.search(buf) + if not m: + break + sentence, buf = (buf[:m.end()].strip(), + buf[m.end():]) + if sentence: + yield sentence + elif ev.get("type") == "content_block_stop": + # End of a speech block (e.g. right before a tool + # call): flush NOW. Without this, pre-tool filler + # ("On it — let me grab that.") sits silent in the + # buffer through the whole tool run, then plays + # glued to the answer: long dead air, then two + # thoughts at once. + tail = buf.strip() + buf = "" + if tail: + yield tail + elif t == "ResultMessage": + self._dirty = False # turn fully consumed — pipe aligned + self._tally(msg) + self._remember_session(msg) + await self._pull_rate_limits() + break + tail = buf.strip() + if tail: + yield tail + +ClaudeWarmBrain = WarmBrain + + +if __name__ == "__main__": + import time + + async def demo(): + b = WarmBrain() + await b.start() + for prompt in ("Voice check: greet me in one sentence.", + "And what's two plus two, spoken like yourself?"): + t0 = time.time() + async for s in b.ask_stream(prompt): + print(f" ({time.time()-t0:4.1f}s) {s}", flush=True) + await b.stop() + + asyncio.run(demo()) diff --git a/backtalk/brain_local.py b/backtalk/brain_local.py new file mode 100644 index 0000000..df1067a --- /dev/null +++ b/backtalk/brain_local.py @@ -0,0 +1,546 @@ +# backtalk: talk to your local AI agent out loud. +# Copyright (C) 2026 Jared Rhodenizer, AnZym contributors +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as published +# by the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . +# +# SPDX-License-Identifier: AGPL-3.0-or-later +"""The warm local brain — connects to a local OpenAI-compatible inference server +(such as llama-server, Ollama, vLLM, or Aphrodite), executing multi-step +filesystem and web tools and streaming spoken output to the mouth. +""" +import asyncio +import json +import os +import re +import subprocess +from pathlib import Path + +import httpx + +try: + from bs4 import BeautifulSoup +except ImportError: + BeautifulSoup = None + +try: + from ddgs import DDGS +except ImportError: + try: + from duckduckgo_search import DDGS + except ImportError: + DDGS = None + +from backtalk import signals +from backtalk.config import CFG, DISCIPLINE +from backtalk.vlog import log + +_SENTENCE_END = re.compile(r"(?<=[.!?])\s") +_SPECIAL_TOKENS = re.compile(r"<\|im_end\|>|<\|im_start\|>|<\|endoftext\|>|<\|[^|]+?\|>") +_THINK_BLOCK = re.compile(r"[\s\S]*?|\[THINK\][\s\S]*?\[/THINK\]") +_TOOL_TAGS = re.compile(r"[\s\S]*?|]+>[\s\S]*?|]+>[\s\S]*?") +_CODE_BLOCK = re.compile(r"```[\s\S]*?```", re.DOTALL) +_INLINE_SCRIPT = re.compile(r"<<\s*['\"]?[A-Za-z0-9_]+['\"]?[\s\S]*", re.DOTALL) +_RAW_CODE_LINE = re.compile(r"^(?:import |from |def |class |with open|cat >|\s*return |\s*#|\s*if __name__).*$", re.MULTILINE) + +SESSION_FILE = os.path.join(CFG.get("signals_dir", "/tmp/signals"), ".backtalk_session") + + +def _clean_text(text: str) -> str: + """Strip special tokens, think blocks, code blocks, and tool tags for speech output.""" + t = _THINK_BLOCK.sub("", text) + t = _TOOL_TAGS.sub("", t) + t = _CODE_BLOCK.sub("", t) + t = _INLINE_SCRIPT.sub("", t) + t = _SPECIAL_TOKENS.sub("", t) + t = _RAW_CODE_LINE.sub("", t) + t = re.sub(r"^#+\s*", "", t, flags=re.MULTILINE) + t = t.replace("**", "").replace("*", "").replace("`", "") + lines = [ + line.strip() + for line in t.split("\n") + if line.strip() and not line.strip().startswith(("#", "//", "/*", "*", "def ", "import ")) + ] + return " ".join(lines).strip() + + +def parse_tool_call(text: str) -> tuple[str, dict] | None: + """Extract tool name and arguments from JSON or model XML format.""" + # Format 1: JSON + m1 = re.search(r"([\s\S]*?)", text) + if m1: + raw = m1.group(1).strip() + try: + d = json.loads(raw) + return d.get("tool") or d.get("name"), d.get("parameters") or d.get("arguments") or d + except Exception: + pass + + # Format 2: VALUE + m2 = re.search(r"([\s\S]*?)", text) + if m2: + tool_name = m2.group(1).strip() + params = {} + for pm in re.finditer(r"([\s\S]*?)", m2.group(2)): + params[pm.group(1).strip()] = pm.group(2).strip() + return tool_name, params + + # Format 3: JSON markdown code block + m3 = re.search(r'```(?:json)?\s*(\{\s*"(?:tool|name)"[\s\S]*?\})\s*```', text) + if m3: + try: + d = json.loads(m3.group(1)) + return d.get("tool") or d.get("name"), d.get("parameters") or d.get("arguments") or d + except Exception: + pass + + return None + + +def resolve_project_path(target: str, current_cwd: str) -> str: + """Map alias or relative name to absolute filesystem directory.""" + aliases = CFG.get("project_aliases", {}) + cleaned = target.strip().lower().replace("_", " ").replace("-", " ") + for alias, p in aliases.items(): + if alias.lower() in cleaned or cleaned == alias.lower(): + return os.path.expanduser(p) + + for extra in CFG.get("extra_dirs", []): + expanded_extra = os.path.expanduser(extra) + if os.path.basename(expanded_extra).lower() == cleaned: + return expanded_extra + + expanded = os.path.expanduser(target.strip()) + if os.path.isabs(expanded) and os.path.exists(expanded): + return expanded + rel = os.path.join(current_cwd, target.strip()) + if os.path.exists(rel): + return rel + return expanded + + +def execute_tool(tool_name: str, args: dict, brain_ref=None) -> str: + """Execute a local, web, or workspace tool safely.""" + default_ws = CFG.get("agent_dir", os.getcwd()) + cwd = brain_ref.active_project_dir if brain_ref else default_ws + try: + signals.set_state("thinking") + + if tool_name == "switch_workspace": + target = args.get("path") or args.get("project") or args.get("name", "") + resolved = resolve_project_path(target, cwd) + log(f"[brain-local] switching workspace to: {resolved}") + if not os.path.exists(resolved): + return f"Error: Workspace path {resolved} does not exist." + if brain_ref: + brain_ref.active_project_dir = resolved + + tree_lines = [] + for root, dirs, files in os.walk(resolved): + dirs[:] = [d for d in dirs if not d.startswith(".") and d not in ("__pycache__", "node_modules", "build")] + rel = os.path.relpath(root, resolved) + depth = rel.count(os.sep) + if depth > 2: + continue + indent = " " * depth + if rel != ".": + tree_lines.append(f"{indent}📂 {os.path.basename(root)}/") + for f in files[:8]: + if not f.startswith("."): + tree_lines.append(f"{indent} 📄 {f}") + if len(tree_lines) > 30: + break + + readme_text = "" + readme_p = os.path.join(resolved, "README.md") + if os.path.exists(readme_p): + try: + with open(readme_p, "r", encoding="utf-8", errors="ignore") as f: + readme_text = f.read(1500) + except Exception: + pass + + return ( + f"--- Switched Active Workspace to: {resolved} ---\n" + f"Workspace File Tree:\n" + "\n".join(tree_lines[:25]) + "\n" + + (f"README Overview:\n{readme_text}\n" if readme_text else "") + ) + + elif tool_name == "search_web": + if not DDGS: + return "Error: Web search library (ddgs / duckduckgo_search) is not installed." + query = args.get("query", "") + if not query: + return "Error: No search query provided." + log(f"[brain-local] searching web for: {query}") + results = [] + with DDGS() as ddgs: + for r in ddgs.text(query, max_results=int(args.get("max_results", 4))): + title = r.get("title", "") + snippet = r.get("body", "") + href = r.get("href", "") + results.append(f"Title: {title}\nSnippet: {snippet}\nURL: {href}\n") + return f"--- Web Search Results for '{query}' ---\n" + ("\n".join(results) if results else "No results found.") + + elif tool_name == "read_web_page": + if not BeautifulSoup: + return "Error: BeautifulSoup (bs4) is not installed." + url = args.get("url", "") + if not url: + return "Error: No URL provided." + log(f"[brain-local] fetching web page: {url}") + headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"} + resp = httpx.get(url, headers=headers, timeout=12.0, follow_redirects=True) + soup = BeautifulSoup(resp.text, "html.parser") + for tag in soup(["script", "style", "nav", "footer", "header", "aside", "form"]): + tag.decompose() + text = " ".join(soup.stripped_strings) + max_chars = int(args.get("max_chars", 3000)) + return f"--- Web Page Content ({url}) ---\n{text[:max_chars]}" + + elif tool_name == "list_dir": + raw_p = args.get("path", ".") + p = raw_p if os.path.isabs(raw_p) else os.path.join(cwd, raw_p) + p = os.path.expanduser(p) + log(f"[brain-local] inspecting directory: {p}") + if not os.path.exists(p): + return f"Error: Directory {p} does not exist." + + tree_items = [] + for root, dirs, files in os.walk(p): + dirs[:] = [d for d in dirs if not d.startswith(".") and d not in ("__pycache__", "build", "install", "node_modules", ".git")] + rel = os.path.relpath(root, p) + depth = rel.count(os.sep) + if depth > 2: + continue + prefix = " " * depth + if rel != ".": + tree_items.append(f"{prefix}📂 {os.path.basename(root)}/") + for f in files: + if not f.startswith("."): + full_rel = os.path.join(rel, f) if rel != "." else f + tree_items.append(f"{prefix} 📄 {full_rel}") + if len(tree_items) > 50: + break + + return f"Recursive File Tree of {p}:\n" + "\n".join(tree_items[:45]) + + elif tool_name == "read_file": + raw_p = args.get("path", "") + p = raw_p if os.path.isabs(raw_p) else os.path.join(cwd, raw_p) + p = os.path.expanduser(p) + log(f"[brain-local] reading file: {p}") + if not os.path.exists(p): + found = list(Path(cwd).glob(f"**/{os.path.basename(raw_p)}")) + if found: + p = str(found[0]) + else: + return f"Error: File {raw_p} does not exist in {cwd}." + max_lines = int(args.get("max_lines", 120)) + with open(p, "r", encoding="utf-8", errors="ignore") as f: + lines = [f.readline() for _ in range(max_lines)] + return f"--- File Content: {p} ---\n" + "".join(lines) + + elif tool_name == "search_files": + raw_p = args.get("path", ".") + base = raw_p if os.path.isabs(raw_p) else os.path.join(cwd, raw_p) + base = os.path.expanduser(base) + query = args.get("query", "") + log(f"[brain-local] searching files for: {query} in {base}") + matches = [] + for root, _, filenames in os.walk(base): + for fn in filenames: + if query.lower() in fn.lower(): + matches.append(os.path.join(root, fn)) + if len(matches) >= 20: + break + if len(matches) >= 20: + break + return f"Found matching files for '{query}':\n" + "\n".join(matches) + + elif tool_name == "run_command": + cmd = args.get("cmd", "") + target_cwd = args.get("cwd", cwd) + target_cwd = target_cwd if os.path.isabs(target_cwd) else os.path.join(cwd, target_cwd) + target_cwd = os.path.expanduser(target_cwd) + log(f"[brain-local] running command: {cmd} (in {target_cwd})") + res = subprocess.run(cmd, shell=True, cwd=target_cwd, capture_output=True, text=True, timeout=15) + out = res.stdout if res.stdout else res.stderr + return f"Command output ($ {cmd} in {target_cwd}):\n{out[:2000]}" + + return f"Unknown tool: {tool_name}" + except Exception as e: + return f"Tool execution failed: {e}" + + +TOOL_PROMPT = """ +### OPERATIONAL DIRECTIVE: MULTI-STEP TOOLS & SPOKEN ANSWERS +You are a voice-interactive assistant connected directly to audio input and speech synthesis. + +RULES: +1. When you need to inspect directories, read code files, or search the web, invoke tools using {"tool": "name", ...}. +2. `list_dir` returns the recursive file tree, so you can locate and read target files immediately. +3. As soon as you have inspected the necessary code or files, STOP calling tools and deliver your spoken answer directly. +4. Keep spoken responses clear, concise, and natural for text-to-speech audio. Avoid markdown tables, URLs, or long unpronounceable code blocks. + +Available Tools: +- switch_workspace(path) +- list_dir(path) +- read_file(path, max_lines) +- search_files(path, query) +- run_command(cmd, cwd) +- search_web(query) +- read_web_page(url) +""" + + +class LocalWarmBrain: + """A persistent local LLM session via an OpenAI-compatible endpoint (llama-server, Ollama, etc.).""" + + def __init__(self, model: str | None = None, can_use_tool=None, resume_id: str | None = None): + self.api_base = CFG.get("api_base", "http://127.0.0.1:8080/v1").rstrip("/") + self.model = model or CFG.get("model", "default") + self.active_project_dir = CFG.get("agent_dir", os.getcwd()) + self._can_use_tool = can_use_tool + self.session = {"turns": 0, "out_tokens": 0, "in_tokens": 0, "cost": 0.0} + self.messages = [] + self._dirty = False + self._interrupted = False + self.system_prompt = self._load_system_prompt() + self.permission_mode = CFG.get("permission_mode", "ask") + + def _get_workspace_snapshot(self) -> str: + try: + cwd = self.active_project_dir + if not os.path.exists(cwd): + return "" + entries = os.listdir(cwd) + dirs = [f"{e}/" for e in sorted(entries) if os.path.isdir(os.path.join(cwd, e)) and not e.startswith(".")] + files = [e for e in sorted(entries) if os.path.isfile(os.path.join(cwd, e)) and not e.startswith(".")] + return ( + f"\n[Active Workspace: {cwd}]\n" + f"Directories: {', '.join(dirs[:15])}\n" + f"Files: {', '.join(files[:15])}" + ) + except Exception: + return "" + + def _load_system_prompt(self) -> str: + agent_dir = Path(os.path.expanduser(CFG.get("agent_dir", os.getcwd()))) + prompt_parts = [DISCIPLINE, TOOL_PROMPT] + + for filename in ("AGENT.md", "CLAUDE.md", "SYSTEM.md"): + p = agent_dir / filename + if p.exists(): + try: + prompt_parts.append(p.read_text(encoding="utf-8")) + log(f"[brain-local] loaded persona from {p}") + break + except Exception as e: + log(f"[brain-local] error reading {p}: {e}") + + for extra in CFG.get("extra_dirs", []): + extra_path = Path(os.path.expanduser(extra)) + idx = extra_path / "VAULT-INDEX.md" + if idx.exists(): + try: + prompt_parts.append(f"## Context ({extra})\n{idx.read_text(encoding='utf-8')[:3000]}") + log(f"[brain-local] loaded index from {idx}") + except Exception: + pass + readme = extra_path / "README.md" + if readme.exists(): + try: + prompt_parts.append(f"## Workspace Overview ({extra})\n{readme.read_text(encoding='utf-8')[:4000]}") + log(f"[brain-local] loaded workspace summary from {readme}") + except Exception: + pass + + return "\n\n".join(prompt_parts) + + async def start(self): + full_prompt = self.system_prompt + self._get_workspace_snapshot() + self.messages = [{"role": "system", "content": full_prompt}] + self._dirty = False + self._interrupted = False + log(f"[brain-local] connected to {self.api_base} (model={self.model}, cwd={self.active_project_dir})") + + async def stop(self): + self.messages.clear() + self._dirty = False + + async def interrupt(self): + self._interrupted = True + self._dirty = False + + async def reset_turn(self, timeout: float = 8.0): + self._dirty = False + self._interrupted = False + + async def set_permission_mode(self, mode: str): + self.permission_mode = mode + log(f"[brain-local] permission mode set to: {mode}") + + async def context_usage(self): + return { + "total_tokens": self.session["in_tokens"] + self.session["out_tokens"], + "turns": self.session["turns"], + } + + async def command(self, cmd: str) -> str: + parts = cmd.strip().split() + if not parts: + return "" + verb = parts[0].lower() + if verb == "/clear": + full_prompt = self.system_prompt + self._get_workspace_snapshot() + self.messages = [{"role": "system", "content": full_prompt}] + self.session = {"turns": 0, "out_tokens": 0, "in_tokens": 0, "cost": 0.0} + return "Conversation history cleared." + elif verb == "/compact": + if len(self.messages) > 9: + self.messages = [self.messages[0]] + self.messages[-8:] + return "Session compacted." + elif verb == "/model" and len(parts) > 1: + self.model = parts[1] + return f"Switched model to {self.model}." + elif verb == "/effort": + return "Effort level updated." + elif verb in ("/cd", "/workspace") and len(parts) > 1: + res = execute_tool("switch_workspace", {"path": parts[1]}, self) + return res + return f"Command acknowledged: {cmd}" + + async def _query_llm(self, messages: list) -> str: + payload = { + "model": self.model, + "messages": messages, + "stream": False, + "temperature": 0.3, + "max_tokens": 2048, + } + async with httpx.AsyncClient(timeout=120.0) as client: + resp = await client.post(f"{self.api_base}/chat/completions", json=payload) + if resp.status_code == 200: + data = resp.json() + return data["choices"][0]["message"]["content"] + log(f"[brain-local] LLM returned status {resp.status_code}: {resp.text[:200]}") + return "" + + async def ask_stream(self, utterance: str): + self._dirty = True + self._interrupted = False + + snapshot = self._get_workspace_snapshot() + user_msg = utterance + if snapshot: + user_msg = f"{utterance}\n\n[Active Workspace Telemetry]:{snapshot}" + self.messages.append({"role": "user", "content": user_msg}) + + max_tool_turns = 5 + for turn_idx in range(max_tool_turns): + reply = await self._query_llm(self.messages) + parsed = parse_tool_call(reply) + if not parsed: + break + + tool_name, args = parsed + log(f"[brain-local] tool call ({turn_idx+1}/{max_tool_turns}): {tool_name} with args {args}") + + tool_result = execute_tool(tool_name, args, self) + log(f"[brain-local] tool output ({len(tool_result)} chars):\n{tool_result[:300]}...") + + self.messages.append({"role": "assistant", "content": reply}) + + if turn_idx == max_tool_turns - 1: + self.messages.append({ + "role": "user", + "content": f"[Tool Result from {tool_name}]:\n{tool_result}\n\n" + f"You now have all necessary data. Deliver your final spoken answer now (do NOT call any tools)." + }) + else: + self.messages.append({ + "role": "user", + "content": f"[Tool Result from {tool_name}]:\n{tool_result}\n\n" + f"If you need to read a specific code file, call read_file. Otherwise, deliver your spoken answer directly." + }) + + buf = "" + full_reply = "" + payload = { + "model": self.model, + "messages": self.messages, + "stream": True, + "temperature": 0.7, + "max_tokens": 2048, + } + + try: + async with httpx.AsyncClient(timeout=120.0) as client: + async with client.stream("POST", f"{self.api_base}/chat/completions", json=payload) as response: + if response.status_code != 200: + err_text = await response.aread() + log(f"[brain-local] server error {response.status_code}: {err_text.decode('utf-8', errors='ignore')}") + yield "I had trouble connecting to the local inference server." + self._dirty = False + return + + async for line in response.aiter_lines(): + if self._interrupted: + log("[brain-local] stream interrupted") + break + if not line.startswith("data: "): + continue + data_str = line[6:].strip() + if data_str == "[DONE]": + break + try: + chunk = json.loads(data_str) + delta = chunk["choices"][0]["delta"].get("content", "") + if delta: + if "<|im_end|>" in delta or "<|endoftext|>" in delta: + delta = _SPECIAL_TOKENS.sub("", delta) + buf += delta + full_reply += delta + break + buf += delta + full_reply += delta + while True: + m_sent = _SENTENCE_END.search(buf) + if not m_sent: + break + sentence, buf = buf[:m_sent.end()].strip(), buf[m_sent.end():] + cleaned = _clean_text(sentence) + if cleaned: + yield cleaned + except Exception: + continue + except Exception as e: + log(f"[brain-local] error querying {self.api_base}: {e}") + yield "Sorry, I lost connection to the local model." + + tail = _clean_text(buf) + if tail and not self._interrupted: + yield tail + + if full_reply: + self.messages.append({"role": "assistant", "content": _clean_text(full_reply)}) + self.session["turns"] += 1 + self.session["out_tokens"] += len(full_reply.split()) * 2 + self.session["in_tokens"] += len(utterance.split()) * 2 + + self._dirty = False + self._interrupted = False + + +WarmBrain = LocalWarmBrain diff --git a/backtalk/config.py b/backtalk/config.py index 6b6571a..29ffb13 100644 --- a/backtalk/config.py +++ b/backtalk/config.py @@ -42,7 +42,13 @@ # Display name, used in logs and to build the quit phrases # ("goodbye " hangs up). Match your agent's actual name. "name": "Assistant", - # The brain. Full model id ON PURPOSE — never a bare alias like + # The brain backend: "claude" (default, uses Claude Agent SDK) or + # "local" (connects to an OpenAI-compatible endpoint such as llama-server, + # Ollama, vLLM, Aphrodite, etc.). + "brain": "claude", + # Endpoint base URL for local OpenAI-compatible brain. + "api_base": "http://127.0.0.1:8080/v1", + # The brain model. When using "claude", specify full model id on purpose # "sonnet": the SDK resolves aliases through its own bundled CLI and # can silently land on an older model. The fast tier is most of the # speed difference people ask about; a deep-work model makes every diff --git a/backtalk/main.py b/backtalk/main.py index 6403a81..51f37d7 100644 --- a/backtalk/main.py +++ b/backtalk/main.py @@ -699,11 +699,16 @@ async def _warmup(): kind = ("timed out" if isinstance(e, asyncio.TimeoutError) else f"failed: {e!r}"[:220]) log(f"[backtalk] BRAIN CONNECT {kind}") - mouth.say("Bad news. The voice and the face are fine, but I " - "couldn't reach my brain, the Claude Code session. " - "Check this window for the error. The usual causes: " - "Claude Code isn't signed in, the internet is down, " - "or the plan is out of usage.") + if str(CFG.get("brain", "claude")).lower() in ("local", "openai", "llama", "vllm", "ollama"): + mouth.say("Bad news. The voice and the face are fine, but I " + f"couldn't reach my local brain at {CFG.get('api_base')}. " + "Check if your local inference server is running.") + else: + mouth.say("Bad news. The voice and the face are fine, but I " + "couldn't reach my brain, the Claude Code session. " + "Check this window for the error. The usual causes: " + "Claude Code isn't signed in, the internet is down, " + "or the plan is out of usage.") mouth.wait_done(timeout=30) raise SystemExit(1) log("[backtalk] brain warm")