From 404a7071fb539d57295a30dedc1e2e5f3dd739fa Mon Sep 17 00:00:00 2001 From: Adolfo Reyna Date: Wed, 12 Aug 2026 10:17:47 -0400 Subject: [PATCH] feat: real-time reasoning speech, dynamic voice tags, and unified web UI bubbles --- apple_llm.py | 11 +- bot.py | 30 +++- dual_engine.py | 262 +++++++++++++++++++++++++++++++ hermes_llm.py | 373 +++++++++++++++++++++++++++++++++++++++----- spoken_text.py | 17 +- test_dual_engine.py | 28 ++++ test_spoken_text.py | 1 + web_server.py | 106 +++++++------ 8 files changed, 734 insertions(+), 94 deletions(-) create mode 100644 dual_engine.py create mode 100644 test_dual_engine.py diff --git a/apple_llm.py b/apple_llm.py index a9b3fb1..c94155c 100644 --- a/apple_llm.py +++ b/apple_llm.py @@ -169,9 +169,9 @@ class MacOSLLM(FrameProcessor): async def _connect(self): available, reason = probe_apple_llm() logger.info(f"macOS LLM engine: {reason}") - if LLM_HELPER_PATH.exists() and "FoundationModels available" in reason: + if LLM_HELPER_PATH.exists() and available and ("FoundationModels" in reason or "Apple Intelligence" in reason): self._use_swift = True - logger.info("Using Swift FoundationModels engine.") + logger.info("Using native Swift macOS Apple Intelligence / FoundationModels engine.") else: self._use_swift = False logger.info(f"Loading MLX model {self._model_name} on Apple Silicon...") @@ -199,6 +199,9 @@ class MacOSLLM(FrameProcessor): await self.cancel_task(task) async def _run_turn(self, utterance: str): + if not self._use_swift and (self._mlx_model is None or self._mlx_tokenizer is None): + await self._connect() + self._history.append({"role": "user", "content": utterance}) await self.push_frame(LLMFullResponseStartFrame()) @@ -224,6 +227,10 @@ class MacOSLLM(FrameProcessor): chunk = data["delta"] chunks.append(chunk) await self.push_frame(LLMTextFrame(chunk)) + elif "content" in data and not chunks: + chunk = data["content"] + chunks.append(chunk) + await self.push_frame(LLMTextFrame(chunk)) elif "text" in data and not chunks: chunk = data["text"] chunks.append(chunk) diff --git a/bot.py b/bot.py index 96e71be..b5453fc 100644 --- a/bot.py +++ b/bot.py @@ -80,7 +80,8 @@ say them: - Spell out things that only make sense visually. Say "line forty-two of bot dot py" rather than pasting a path. - Use your available tools (listing directories, searching, reading files, shell execution) whenever the user asks about files, commands, CLI tools (such as Paseo), or workspace tasks. -- You can change your own voice! If the user asks to list available voices or switch voice, run `python bin/voice_tool.py list` or `python bin/voice_tool.py set ` (voices: af_heart, af_bella, am_michael, am_fenrir, am_puck, bf_emma, bm_george, Moira, Daniel). +- You can dynamically change your spoken voice mid-response! Use markdown tags like `[Voice:af_bella]` or `[Voice:am_michael]` inline to switch voices (e.g. `[Voice:af_bella] Hello from Bella! [Voice:am_michael] And hello from Michael!`). The active voice will persist until you change it again. +- You can change your default voice! If the user asks to list available voices or switch voice permanently, run `python bin/voice_tool.py list` or `python bin/voice_tool.py set ` (voices: af_heart, af_bella, am_michael, am_fenrir, am_puck, bf_emma, bm_george, Moira, Daniel). - You can change your AI model on the fly! If the user asks to list available models or change model, run `python bin/model_tool.py list` or `python bin/model_tool.py set ` (models: luna, gemma, deepseek, gpt-oss, sonnet, etc.). - You can reset or start a fresh conversation session! If the user asks to start a fresh session, reset the conversation, or clear session context, run `python bin/session_tool.py reset`. - You can switch Hermes agent profiles! If the user asks to list Hermes profiles or switch profile, run `python bin/profile_tool.py list` or `python bin/profile_tool.py set `. @@ -287,6 +288,18 @@ def parse_args() -> argparse.Namespace: action="store_true", help="Disable the Companion Web Chat UI server.", ) + parser.add_argument( + "--dual-engine", + action="store_true", + default=False, + help="Enable dual-engine mode: instant macOS foundation model (<400ms) + deep Hermes reasoning.", + ) + parser.add_argument( + "--no-dual-engine", + action="store_false", + dest="dual_engine", + help="Disable dual-engine mode and run single engine directly.", + ) parser.add_argument("--log-level", default="INFO") return parser.parse_args() @@ -520,17 +533,28 @@ def build_llm( model = (model_manager.load_saved_model() if model_manager else None) or getattr(args, "hermes_model", "hermes-3") if args.llm_engine in ("hermes", "ollama"): from hermes_llm import HermesLLM, probe_hermes + from dual_engine import DualEngineProcessor available, reason = probe_hermes(model) logger.info(f"LLM: Hermes ({reason})") - # Hermes handles persona, personality, and memory natively. - return HermesLLM( + + deep_llm = HermesLLM( model=model, cwd=args.cwd, session_name="Voice Agent", observer=observer, ) + if getattr(args, "dual_engine", False): + from apple_llm import MacOSLLM, probe_apple_llm + fast_available, fast_reason = probe_apple_llm() + if fast_available: + logger.info(f"Dual-Engine: Pairing Hermes with fast macOS Foundation Model ({fast_reason})") + fast_llm = MacOSLLM(model=args.mlx_model, system_prompt="You are a fast voice assistant.") + return DualEngineProcessor(fast_llm=fast_llm, deep_llm=deep_llm, observer=observer) + + return deep_llm + if args.llm_engine in ("apple", "macos"): from apple_llm import MacOSLLM, probe_apple_llm diff --git a/dual_engine.py b/dual_engine.py new file mode 100644 index 0000000..c750022 --- /dev/null +++ b/dual_engine.py @@ -0,0 +1,262 @@ +import asyncio +import time +from typing import Callable, Optional +from loguru import logger + +from pipecat.frames.frames import ( + CancelFrame, + EndFrame, + Frame, + InterruptionFrame, + LLMContextFrame, + LLMFullResponseEndFrame, + LLMFullResponseStartFrame, + LLMTextFrame, + StartFrame, +) +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor + + +class DualEngineProcessor(FrameProcessor): + """Dual-Engine Orchestrator. + + Combines a fast local engine (macOS Foundation Model / Apple MLX) for instant + sub-400ms voice feedback with a deep engine (Hermes Agent / Luna / Gemma) for + deep reasoning, tool execution, and workspace memory. + """ + + def __init__( + self, + *, + fast_llm: Optional[FrameProcessor] = None, + deep_llm: FrameProcessor, + observer: Optional[Callable[[str], None]] = None, + **kwargs, + ): + super().__init__(**kwargs) + self._fast_llm = fast_llm + self._deep_llm = deep_llm + self._observer = observer + + self._current_user_text: str = "" + self._fast_task: Optional[asyncio.Task] = None + self._deep_task: Optional[asyncio.Task] = None + self._fast_spoken: bool = False + self._deep_spoken: bool = False + self._last_tool_phrase: str = "" + + if hasattr(self._deep_llm, "_on_tool_event"): + self._deep_llm._on_tool_event = self.handle_tool_signal + + def handle_tool_signal(self, detail: str): + if not detail or self._deep_spoken: + return + + detail_lower = detail.lower() + if "read" in detail_lower or "view" in detail_lower or "cat" in detail_lower: + phrase = "Inspecting project files." + elif "search" in detail_lower or "grep" in detail_lower or "find" in detail_lower: + phrase = "Searching the codebase." + elif "exec" in detail_lower or "run" in detail_lower or "command" in detail_lower: + phrase = "Running command." + else: + phrase = "Working on that." + + if phrase == self._last_tool_phrase: + return + self._last_tool_phrase = phrase + + logger.info(f"🗣 [DualEngine Voice Signal]: {phrase!r} (from tool event: {detail[:60]!r})") + asyncio.create_task(self._speak_tool_update(phrase)) + + async def _speak_tool_update(self, phrase: str): + try: + await self.push_frame(LLMFullResponseStartFrame()) + await self.push_frame(LLMTextFrame(phrase)) + await self.push_frame(LLMFullResponseEndFrame()) + except Exception as e: + logger.debug(f"Tool voice update error: {e}") + + async def setup(self, task_manager): + await super().setup(task_manager) + if self._fast_llm and hasattr(self._fast_llm, "setup"): + await self._fast_llm.setup(task_manager) + if self._deep_llm and hasattr(self._deep_llm, "setup"): + await self._deep_llm.setup(task_manager) + + def set_task_manager(self, task_manager): + super().set_task_manager(task_manager) + if self._fast_llm and hasattr(self._fast_llm, "set_task_manager"): + self._fast_llm.set_task_manager(task_manager) + if self._deep_llm and hasattr(self._deep_llm, "set_task_manager"): + self._deep_llm.set_task_manager(task_manager) + + def link(self, processor: "FrameProcessor"): + super().link(processor) + if self._fast_llm and hasattr(self._fast_llm, "link"): + self._fast_llm.link(processor) + if self._deep_llm and hasattr(self._deep_llm, "link"): + self._deep_llm.link(processor) + + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + if isinstance(frame, StartFrame): + if self._fast_llm: + await self._fast_llm.process_frame(frame, direction) + await self._deep_llm.process_frame(frame, direction) + await self.push_frame(frame, direction) + + elif isinstance(frame, (EndFrame, CancelFrame)): + await self._cancel_active_tasks() + if self._fast_llm: + await self._fast_llm.process_frame(frame, direction) + await self._deep_llm.process_frame(frame, direction) + await self.push_frame(frame, direction) + + elif isinstance(frame, InterruptionFrame): + await self._cancel_active_tasks() + if self._fast_llm: + await self._fast_llm.process_frame(frame, direction) + await self._deep_llm.process_frame(frame, direction) + await self.push_frame(frame, direction) + + elif isinstance(frame, LLMContextFrame): + text = self._extract_user_text(frame.context) + if text: + await self.start_dual_turn(text) + else: + await self.push_frame(frame, direction) + + else: + await self.push_frame(frame, direction) + + def start_turn_direct(self, text: str): + utterance = text.strip() + if not utterance: + return + asyncio.create_task(self.start_dual_turn(utterance)) + + async def start_dual_turn(self, text: str): + utterance = text.strip() + if not utterance: + return + + await self._cancel_active_tasks() + self._current_user_text = utterance + self._fast_spoken = False + self._deep_spoken = False + self._suppress_deep = False + + logger.info(f"⚡ [DualEngine] Starting turn for prompt: {utterance!r}") + t0 = time.perf_counter() + + # Start deep Hermes processing in background + self._deep_task = asyncio.create_task(self._run_deep_path(utterance, t0)) + + # Dispatch fast-path acknowledgment concurrently + if self._fast_llm and hasattr(self._fast_llm, "_run_turn"): + self._fast_task = asyncio.create_task(self._run_fast_path(utterance, t0)) + + async def _run_fast_path(self, utterance: str, t0: float): + try: + fast_prompt = ( + "You are a fast voice assistant.\n" + "Rules:\n" + "1. If the prompt is a simple greeting or fully answered by a short sentence, " + "end your answer with [COMPLETE].\n" + "2. If it requires deep search/code/tools, use a soft natural human filler " + '(e.g., "Ah, let me check that...", "Hmm, let me look into that.") and end with [NEEDS_DEEP].\n' + "3. Keep output under 15 words.\n\n" + f"User prompt: {utterance!r}" + ) + chunks: list[str] = [] + if hasattr(self._fast_llm, "_run_turn_cli"): + await self._fast_llm._run_turn_cli(fast_prompt, chunks) + elif hasattr(self._fast_llm, "_run_turn"): + await self._fast_llm._run_turn(fast_prompt) + + t1 = time.perf_counter() + raw_text = " ".join(chunks).strip() + + is_complete = "[COMPLETE]" in raw_text + cleaned_text = raw_text.replace("[COMPLETE]", "").replace("[NEEDS_DEEP]", "").strip() + + if cleaned_text and not self._deep_spoken: + self._fast_spoken = True + if is_complete: + self._suppress_deep = True + logger.info(f"⚡ [DualEngine Speculative Routing]: Query marked COMPLETE by fast model. Suppressing redundant deep response.") + + logger.info(f"⏱ [DualEngine Fast-Path ({int((t1-t0)*1000)}ms)]: {cleaned_text!r} (Complete: {is_complete})") + + try: + import web_server + web_server.broadcast_event("fast_reply", { + "text": cleaned_text, + "is_complete": is_complete, + }) + except Exception: + pass + + await self.push_frame(LLMFullResponseStartFrame()) + await self.push_frame(LLMTextFrame(cleaned_text)) + await self.push_frame(LLMFullResponseEndFrame()) + + except asyncio.CancelledError: + pass + except Exception as e: + logger.debug(f"DualEngine fast-path error: {e}") + + async def _run_deep_path(self, utterance: str, t0: float): + try: + # If fast model marked turn COMPLETE, run deep Hermes in background history mode + if self._suppress_deep: + logger.info("Hermes deep path running silently in background history sync mode...") + + if hasattr(self._deep_llm, "_run_turn"): + try: + await self._deep_llm._run_turn(utterance, suppress_output=self._suppress_deep) + except TypeError: + await self._deep_llm._run_turn(utterance) + + t1 = time.perf_counter() + self._deep_spoken = True + logger.info(f"⏱ [DualEngine Deep-Path ({int((t1-t0)*1000)}ms)] turn complete.") + + try: + import web_server + web_server.broadcast_event("profiling", { + "mode": "Dual-Engine (Fast + Deep)", + "total_ms": int((t1 - t0) * 1000), + }) + except Exception: + pass + + except asyncio.CancelledError: + pass + except Exception as e: + logger.error(f"DualEngine deep-path error: {e}") + + async def _cancel_active_tasks(self): + for task in (self._fast_task, self._deep_task): + if task and not task.done(): + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + self._fast_task = None + self._deep_task = None + + def _extract_user_text(self, context) -> str: + if not context or not hasattr(context, "messages"): + return "" + for msg in reversed(context.messages): + if isinstance(msg, dict) and msg.get("role") == "user": + content = msg.get("content", "") + if isinstance(content, str): + return content + elif isinstance(content, list): + return " ".join([c.get("text", "") for c in content if isinstance(c, dict)]) + return "" diff --git a/hermes_llm.py b/hermes_llm.py index e888074..896c3bf 100644 --- a/hermes_llm.py +++ b/hermes_llm.py @@ -12,6 +12,8 @@ import re import shutil from pathlib import Path import aiohttp +from typing import Callable, Optional +import time from loguru import logger import env_setup @@ -26,6 +28,8 @@ from pipecat.frames.frames import ( LLMFullResponseStartFrame, LLMTextFrame, StartFrame, + TextFrame, + TTSSpeakFrame, ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor @@ -81,6 +85,38 @@ def _clean_spoken_text(text: str) -> str: return " ".join(lines).strip() +class StreamParser: + """Parses streaming tokens, separating ... reasoning blocks from spoken text.""" + + def __init__(self): + self.in_think = False + + def feed(self, chunk: str) -> tuple[str, str]: + thinking = "" + spoken = "" + buf = chunk + while buf: + if not self.in_think: + if "" in buf: + parts = buf.split("", 1) + spoken += parts[0] + self.in_think = True + buf = parts[1] + else: + spoken += buf + buf = "" + else: + if "" in buf: + parts = buf.split("", 1) + thinking += parts[0] + self.in_think = False + buf = parts[1] + else: + thinking += buf + buf = "" + return thinking, spoken + + def find_hermes_cli() -> str | None: candidates = [ shutil.which("hermes"), @@ -95,20 +131,37 @@ def find_hermes_cli() -> str | None: return shutil.which("hermes") -async def check_hermes_server_active(port: int = 8642) -> tuple[bool, str]: - """Check if Hermes gateway server daemon is responding to health requests.""" - url = f"http://localhost:{port}/api/health" +def get_hermes_api_key() -> str: + env_file = Path.home() / ".hermes" / ".env" + if env_file.exists(): + try: + with open(env_file, "r") as f: + for line in f: + if line.startswith("API_SERVER_KEY="): + return line.split("=", 1)[1].strip().strip('"').strip("'") + except Exception: + pass + return os.environ.get("API_SERVER_KEY", "") + + +async def check_hermes_server_active(port: int = 9119) -> tuple[bool, str]: + """Check if Hermes OpenAI gateway endpoint (v1/models) is responding with authorization.""" + url_models = f"http://127.0.0.1:{port}/v1/models" + key = get_hermes_api_key() + headers = {"Authorization": f"Bearer {key}"} if key else {} try: - async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=1.5)) as session: - async with session.get(url) as resp: + async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=1.0)) as session: + async with session.get(url_models, headers=headers) as resp: if resp.status == 200: - return True, f"Hermes server active on http://localhost:{port}" + return True, f"Hermes Gateway API active on http://127.0.0.1:{port}/v1" + elif resp.status == 401: + return False, "Hermes Gateway API requires API_SERVER_KEY" except Exception: pass - return False, "Hermes server daemon not active" + return False, "Hermes Gateway API server not active" -async def ensure_hermes_server(port: int = 8642) -> tuple[bool, str]: +async def ensure_hermes_server(port: int = 9119) -> tuple[bool, str]: """Ensure Hermes gateway server daemon or CLI binary is available.""" active, msg = await check_hermes_server_active(port) if active: @@ -127,8 +180,6 @@ def probe_hermes(model: str | None = None) -> tuple[bool, str]: m_str = f" with model {model}" if model else "" return True, f"Hermes available ({cli}){m_str}" return False, "Hermes CLI binary not found (install hermes or ensure it is in PATH)" - - class HermesLLM(FrameProcessor): """Runs user turns through Hermes CLI or Gateway API using persistent session tracking.""" @@ -137,10 +188,12 @@ class HermesLLM(FrameProcessor): *, model: str | None = None, cwd: str | Path | None = None, - port: int = 8642, + port: int = 9119, session_name: str = "Voice Agent", observer=None, + on_tool_event: Optional[Callable[[str], None]] = None, use_server: bool = False, + keep_open: bool = True, **kwargs, ): super().__init__(**kwargs) @@ -149,13 +202,19 @@ class HermesLLM(FrameProcessor): self._port = port self._session_name = session_name self._on_reply = observer + self._on_tool_event = on_tool_event self._turn_task: asyncio.Task | None = None self._history: list[dict[str, str]] = [] self._cli_path = find_hermes_cli() or "hermes" self._use_server = use_server + self._keep_open = keep_open self._session_renamed = False self._http_session: aiohttp.ClientSession | None = None + self._proc: asyncio.subprocess.Process | None = None + self._proc_lock = asyncio.Lock() + self._stderr_task: asyncio.Task | None = None + # Keep the conversation lineage with the workspace. A single global # session file can make two voice-agent workspaces resume each other's # Hermes conversations. @@ -172,6 +231,8 @@ class HermesLLM(FrameProcessor): self._session_id = None self._session_renamed = False self._history.clear() + if self._proc: + asyncio.create_task(self._stop_persistent_proc()) try: if self._session_state_file.exists(): self._session_state_file.unlink() @@ -186,6 +247,8 @@ class HermesLLM(FrameProcessor): self._session_id = disk_sid self._session_renamed = False self._history.clear() + if self._proc: + asyncio.create_task(self._stop_persistent_proc()) async def _get_http_session(self) -> aiohttp.ClientSession: if self._http_session is None or self._http_session.closed: @@ -250,6 +313,53 @@ class HermesLLM(FrameProcessor): except Exception as e: logger.debug(f"Could not rename Hermes session: {e}") + async def _ensure_persistent_proc(self): + async with self._proc_lock: + ok, _ = await check_hermes_server_active(self._port) + if ok: + return + + cmd = [self._cli_path, "serve", "--port", str(self._port), "--skip-build"] + try: + env = {**os.environ, "PYTHONUNBUFFERED": "1", "FORCE_COLOR": "0", "NO_COLOR": "1"} + self._proc = await asyncio.create_subprocess_exec( + *cmd, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + cwd=str(self._cwd), + env=env, + ) + logger.info(f"Auto-started Hermes server (`hermes serve --port {self._port} --skip-build`, PID: {self._proc.pid})") + + for _ in range(50): + ready, _ = await check_hermes_server_active(self._port) + if ready: + logger.info(f"Hermes server active and ready on port {self._port}.") + break + await asyncio.sleep(0.1) + except Exception as e: + logger.warning(f"Could not auto-start Hermes server daemon (`hermes serve --skip-build`): {e}") + + async def _stop_persistent_proc(self): + async with self._proc_lock: + proc = self._proc + self._proc = None + if self._stderr_task and not self._stderr_task.done(): + self._stderr_task.cancel() + self._stderr_task = None + if proc and proc.returncode is None: + try: + if proc.stdin and not proc.stdin.is_closing(): + proc.stdin.close() + proc.terminate() + await asyncio.wait_for(proc.wait(), timeout=1.5) + except Exception: + try: + proc.kill() + except Exception: + pass + logger.info("Persistent Hermes process terminated cleanly.") + async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) @@ -257,8 +367,11 @@ class HermesLLM(FrameProcessor): await self.push_frame(frame, direction) available, reason = probe_hermes(self._model) logger.info(f"Hermes LLM engine initialized: {reason}") + if self._keep_open: + asyncio.create_task(self._ensure_persistent_proc()) elif isinstance(frame, (EndFrame, CancelFrame)): await self._cancel_turn() + await self._stop_persistent_proc() await self._close_http_session() await self.push_frame(frame, direction) elif isinstance(frame, InterruptionFrame): @@ -325,76 +438,237 @@ class HermesLLM(FrameProcessor): if new_model != self._model: logger.info(f"Hermes LLM switching active model: {self._model} -> {new_model}") self._model = new_model + if self._proc: + asyncio.create_task(self._stop_persistent_proc()) except Exception as e: logger.debug(f"Could not read model settings: {e}") - async def _run_turn(self, utterance: str): + async def _run_turn(self, utterance: str, suppress_output: bool = False): self._sync_disk_model() self._sync_disk_session() self._history.append({"role": "user", "content": utterance}) - await self.push_frame(LLMFullResponseStartFrame()) + if not suppress_output: + await self.push_frame(LLMFullResponseStartFrame()) chunks: list[str] = [] - if self._use_server: + if self._keep_open: + await self._ensure_persistent_proc() + + t0 = time.perf_counter() + server_ok, _ = await check_hermes_server_active(self._port) + mode_str = f"API ({self._port})" if (server_ok or self._use_server) else "CLI" + + try: + import web_server + web_server.broadcast_event("hermes_status", { + "mode": mode_str, + "is_api": server_ok or self._use_server, + "port": self._port, + }) + except Exception: + pass + + if server_ok or self._use_server: await self._run_turn_server(utterance, chunks) else: await self._run_turn_cli(utterance, chunks) - # The transport-specific runners may fall back from one to the other; - # emit exactly one response terminator for the whole turn. - await self.push_frame(LLMFullResponseEndFrame()) + t1 = time.perf_counter() + total_ms = int((t1 - t0) * 1000) + + if not suppress_output: + await self.push_frame(LLMFullResponseEndFrame()) + + logger.info(f"⏱ [PROFILING] Hermes LLM ({mode_str}): Turn completed in {total_ms}ms ({total_ms/1000:.2f}s)") + try: + import web_server + web_server.broadcast_event("profiling", { + "mode": mode_str, + "total_ms": total_ms, + "model": self._model or "default", + }) + except Exception: + pass full_reply = _clean_spoken_text(" ".join(chunks)) if full_reply: self._history.append({"role": "assistant", "content": full_reply}) logger.info(f"Hermes LLM ({self._model or 'default'}): {full_reply}") - if self._on_reply: - self._on_reply(full_reply) + if not suppress_output: + try: + import web_server + web_server.broadcast_event("reply", {"text": full_reply}) + except Exception: + pass + if self._on_reply: + self._on_reply(full_reply) async def _run_turn_server(self, utterance: str, chunks: list[str]): - """Run turn via Hermes Server / Gateway HTTP API if available.""" + """Run turn via Hermes OpenAI-compatible Gateway API using SSE streaming (stream: true).""" try: ok, _ = await check_hermes_server_active(self._port) if not ok: - raise RuntimeError("Hermes server daemon unavailable") + raise RuntimeError("Hermes Gateway API unavailable") + + key = get_hermes_api_key() + headers = {"Authorization": f"Bearer {key}", "Content-Type": "application/json"} if key else {"Content-Type": "application/json"} + url = f"http://127.0.0.1:{self._port}/v1/chat/completions" + + messages = [dict(m) for m in self._history[-10:]] + if not messages or messages[-1].get("content") != utterance: + messages.append({"role": "user", "content": utterance}) - session_id = self._session_id or "voice-agent" - url = f"http://localhost:{self._port}/api/sessions/{session_id}/chat" payload = { - "message": utterance, + "model": "hermes-agent" if not self._model or self._model.lower() in ("default", "none", "") else self._model, + "messages": messages, + "stream": True, } - if self._model and self._model.lower() not in ("default", "none", ""): - payload["model"] = self._model session = await self._get_http_session() - async with session.post(url, json=payload) as resp: - if resp.status == 200: - data = await resp.json() - text_val = data.get("reply") or data.get("text") or data.get("content", "") - cleaned = _clean_spoken_text(str(text_val)) - if cleaned: - chunks.append(cleaned) - await self.push_frame(LLMTextFrame(cleaned)) + async with session.post(url, json=payload, headers=headers) as resp: + if resp.status == 200: + sentence_buffer = "" + reasoning_buffer = "" + parser = StreamParser() + + async for raw_line in resp.content: + line = raw_line.decode("utf-8").strip() + if not line or line.startswith(":"): + continue + if line == "data: [DONE]": + break + if line.startswith("data: "): + try: + data = json.loads(line[6:]) + choices = data.get("choices", []) + if choices: + delta = choices[0].get("delta", {}) + + # Handle thinking/reasoning deltas immediately + reasoning = delta.get("reasoning") or delta.get("thought") + if reasoning: + reasoning_buffer += str(reasoning) + words = reasoning_buffer.strip().split() + if "\n" in reasoning_buffer or any(p in reasoning_buffer for p in (".", "!", "?")) or len(words) >= 4: + reasoning_phrase = reasoning_buffer.strip() + reasoning_buffer = "" + cleaned_reasoning = _clean_spoken_text(reasoning_phrase) + if cleaned_reasoning: + if not cleaned_reasoning.endswith((".", "!", "?")): + cleaned_reasoning += "." + chunks.append(cleaned_reasoning) + # Push TTSSpeakFrame so Kokoro TTS synthesizes & plays audio IMMEDIATELY + await self.push_frame(TTSSpeakFrame(cleaned_reasoning)) + try: + import web_server + web_server.broadcast_event("thinking", {"text": str(reasoning)}) + except Exception: + pass + + # Handle tool call deltas + tool_calls = delta.get("tool_calls") + if tool_calls: + for tc in tool_calls: + fn = tc.get("function", {}) + tool_name = fn.get("name", "tool") + tool_args = fn.get("arguments", "") + if self._on_tool_event: + self._on_tool_event(f"Executing {tool_name}") + try: + import web_server + web_server.broadcast_event("tool", {"name": tool_name, "detail": tool_args}) + except Exception: + pass + + content = delta.get("content", "") + if content: + think_text, spoken_text = parser.feed(content) + if think_text: + reasoning_buffer += think_text + words = reasoning_buffer.strip().split() + if "\n" in reasoning_buffer or any(p in reasoning_buffer for p in (".", "!", "?")) or len(words) >= 4: + reasoning_phrase = reasoning_buffer.strip() + reasoning_buffer = "" + cleaned_reasoning = _clean_spoken_text(reasoning_phrase) + if cleaned_reasoning: + if not cleaned_reasoning.endswith((".", "!", "?")): + cleaned_reasoning += "." + chunks.append(cleaned_reasoning) + # Push TTSSpeakFrame so Kokoro TTS synthesizes & plays audio IMMEDIATELY + await self.push_frame(TTSSpeakFrame(cleaned_reasoning)) + try: + import web_server + web_server.broadcast_event("thinking", {"text": think_text}) + except Exception: + pass + + if spoken_text: + sentence_buffer += spoken_text + while any(p in sentence_buffer for p in (".", "!", "?", "\n")): + idxs = [sentence_buffer.find(p) for p in (".", "!", "?", "\n") if sentence_buffer.find(p) != -1] + split_idx = min(idxs) + 1 + sentence = sentence_buffer[:split_idx].strip() + sentence_buffer = sentence_buffer[split_idx:] + + cleaned_sent = _clean_spoken_text(sentence) + if cleaned_sent: + if not cleaned_sent.endswith((".", "!", "?")): + cleaned_sent += "." + chunks.append(cleaned_sent) + await self.push_frame(LLMTextFrame(cleaned_sent)) + try: + import web_server + web_server.broadcast_event("partial_reply", {"text": cleaned_sent}) + except Exception: + pass + except json.JSONDecodeError: + pass + + if reasoning_buffer.strip(): + cleaned_r_rem = _clean_spoken_text(reasoning_buffer) + if cleaned_r_rem: + if not cleaned_r_rem.endswith((".", "!", "?")): + cleaned_r_rem += "." + chunks.append(cleaned_r_rem) + await self.push_frame(TTSSpeakFrame(cleaned_r_rem)) + + if sentence_buffer.strip(): + cleaned_rem = _clean_spoken_text(sentence_buffer) + if cleaned_rem: + if not cleaned_rem.endswith((".", "!", "?")): + cleaned_rem += "." + chunks.append(cleaned_rem) + await self.push_frame(LLMTextFrame(cleaned_rem)) try: import web_server - web_server.broadcast_event("partial_reply", {"text": cleaned}) + web_server.broadcast_event("partial_reply", {"text": cleaned_rem}) except Exception: pass - else: - err_text = await resp.text() - logger.error(f"Hermes server HTTP {resp.status}: {err_text}") - raise RuntimeError(f"HTTP {resp.status}") + else: + err_text = await resp.text() + logger.error(f"Hermes Gateway API HTTP {resp.status}: {err_text}") + raise RuntimeError(f"HTTP {resp.status}") except asyncio.CancelledError: - logger.info("Hermes server turn cancelled mid-response.") + logger.info("Hermes Gateway turn cancelled mid-response.") raise except Exception as e: - logger.warning(f"Hermes server error ({e}), falling back to CLI...") + logger.warning(f"Hermes Gateway error ({e}), falling back to CLI...") + try: + import web_server + web_server.broadcast_event("hermes_status", { + "mode": "CLI", + "is_api": False, + "port": self._port, + }) + except Exception: + pass await self._run_turn_cli(utterance, chunks) + async def _run_turn_cli(self, utterance: str, chunks: list[str]): """Run turn via Hermes CLI using persistent session tracking.""" - cmd = [self._cli_path, "chat", "-q", utterance, "-Q", "--source", "voice"] + cmd = [self._cli_path, "chat", "-q", utterance, "-Q", "--source", "voice", "--reasoning", "none"] if self._session_id: cmd.extend(["-r", self._session_id]) if self._model and self._model.lower() not in ("default", "none", ""): @@ -423,7 +697,14 @@ class HermesLLM(FrameProcessor): # Extract session_id emitted on stderr self._remember_session_id(cleaned) - if cleaned and cleaned not in ("[0m", "0m", "]") and not cleaned.startswith(">"): + if ( + cleaned + and cleaned not in ("[0m", "0m", "]") + and not cleaned.startswith(">") + and not cleaned.startswith("↻") + and not "Resumed session" in cleaned + and not cleaned.lower().startswith("session_id:") + ): logger.info(f"Hermes Tool: {cleaned}") try: import web_server @@ -431,6 +712,12 @@ class HermesLLM(FrameProcessor): except Exception: pass + if self._on_tool_event: + try: + self._on_tool_event(cleaned) + except Exception as exc: + logger.debug(f"on_tool_event error: {exc}") + # Keep tool progress visible in logs and the Companion Web UI, but # do not send implementation details through the spoken channel. diff --git a/spoken_text.py b/spoken_text.py index 7c06f76..78ad8ca 100644 --- a/spoken_text.py +++ b/spoken_text.py @@ -29,10 +29,12 @@ _LIST_MARKER = re.compile(r"^[ \t]*[-*•]\s+", re.MULTILINE) # Identifiers read better as words: "sample_rate" -> "sample rate". _UNDERSCORE_WORD = re.compile(r"(?<=\w)_(?=\w)") _EXTRA_SPACE = re.compile(r"[ \t]{2,}") +_CONTROL_TAGS = re.compile(r"\[(COMPLETE|NEEDS_DEEP|STATUS:[^\]]+)\]", re.IGNORECASE) +_VOICE_TAG = re.compile(r"\[Voice:\s*([a-zA-Z0-9_\-]+)\]", re.IGNORECASE) class SpokenTextFilter(MarkdownTextFilter): - """Markdown filtering, plus the leftovers that matter when read aloud.""" + """Markdown filtering, plus voice tag parsing and leftovers that matter when read aloud.""" def __init__(self, voice_manager=None, **kwargs): super().__init__(**kwargs) @@ -41,8 +43,21 @@ class SpokenTextFilter(MarkdownTextFilter): async def filter(self, text: str) -> str: if self._voice_manager: self._voice_manager.sync_voice() + + # Intercept and set active voice on [Voice:VoiceName] tags + match = _VOICE_TAG.search(text) + if match: + new_voice = match.group(1) + text = _VOICE_TAG.sub("", text) + if self._voice_manager: + try: + self._voice_manager.set_voice(new_voice) + except Exception: + pass + text = _TIMES.sub(" times ", text) text = await super().filter(text) + text = _CONTROL_TAGS.sub("", text) text = _STRIKETHROUGH.sub(r"\1", text) text = _LIST_MARKER.sub("", text) text = _UNDERSCORE_WORD.sub(" ", text) diff --git a/test_dual_engine.py b/test_dual_engine.py new file mode 100644 index 0000000..f890c8e --- /dev/null +++ b/test_dual_engine.py @@ -0,0 +1,28 @@ +import asyncio +import sys +from pathlib import Path + +WORKSPACE = Path("/Users/adolforeyna/Projects/VoiceAgent1") +sys.path.insert(0, str(WORKSPACE)) + +import env_setup +env_setup.setup_environment_path() + +from dual_engine import DualEngineProcessor +from hermes_llm import HermesLLM +from apple_llm import MacOSLLM + +async def test_dual_engine_orchestrator(): + print("Testing DualEngineProcessor initialization and dispatch...") + + deep_llm = HermesLLM(cwd=WORKSPACE, keep_open=True) + fast_llm = MacOSLLM() + + orchestrator = DualEngineProcessor(fast_llm=fast_llm, deep_llm=deep_llm) + assert orchestrator._fast_llm == fast_llm + assert orchestrator._deep_llm == deep_llm + + print("PASS: DualEngineProcessor initialized and verified!") + +if __name__ == "__main__": + asyncio.run(test_dual_engine_orchestrator()) diff --git a/test_spoken_text.py b/test_spoken_text.py index 3114b1e..9eeeffc 100644 --- a/test_spoken_text.py +++ b/test_spoken_text.py @@ -16,6 +16,7 @@ CASES = [ ("Multiply 3 * 4.", "Multiply 3 times 4."), ("A plain sentence.", "A plain sentence."), ("The well-known trade-off is fine.", "The well-known trade-off is fine."), + ("[Voice:Bella] Hello from Bella!", "Hello from Bella!"), ] async def main(): diff --git a/web_server.py b/web_server.py index 3287ded..632f177 100644 --- a/web_server.py +++ b/web_server.py @@ -729,6 +729,12 @@ HTML_INDEX = """

VoiceAgent Companion

+
+ Hermes: API (9119) +
+
+ Latency: -- ms +
Model: Loading...
@@ -845,6 +851,18 @@ HTML_INDEX = """ } else if (data.type === 'status_change') { if (data.model) { modelNameEl.textContent = data.model; activeModelId = data.model; } if (data.voice) voiceNameEl.textContent = data.voice; + } else if (data.type === 'hermes_status') { + const hermesEl = document.getElementById('hermesModeText'); + if (hermesEl && data.mode) { + hermesEl.textContent = data.mode; + hermesEl.style.color = data.is_api ? '#10b981' : '#f59e0b'; + } + } else if (data.type === 'profiling') { + const latEl = document.getElementById('latencyText'); + if (latEl && data.total_ms !== undefined) { + latEl.textContent = `${data.total_ms} ms (${data.mode || ''})`; + latEl.style.color = data.total_ms < 1500 ? '#10b981' : '#818cf8'; + } } else if (data.type === 'show_file') { if (data.name && data.content) { drawerFileName.textContent = data.name; @@ -863,6 +881,10 @@ HTML_INDEX = """ function renderEvent(data) { if (data.type === 'heard' && data.text) { appendUserMessage(data.text, data.at); + } else if (data.type === 'fast_reply' && data.text) { + appendFastReply(data.text, data.is_complete, data.at); + } else if (data.type === 'thinking' && data.text) { + appendThinkingStep(data.text, data.at); } else if (data.type === 'partial_reply' && data.text) { appendPartialReply(data.text, data.at); } else if (data.type === 'reply' && data.text) { @@ -872,42 +894,25 @@ HTML_INDEX = """ } } - function sendMessage() { - const text = userInputEl.value.trim(); - if (!text) return; - userInputEl.value = ''; + let currentThinkingContent = null; - fetch('/api/send', { - method: 'POST', - headers: {'Content-Type': 'application/json'}, - body: JSON.stringify({text: text}) - }) - .then(r => r.json()) - .then(res => { - if (res.error) alert(res.error); - }) - .catch(err => console.error('Failed to send message:', err)); - } - - async function resetSession() { - if (!confirm('Start a fresh Hermes conversation session?')) return; - try { - const res = await fetch('/api/session/reset', { method: 'POST' }); - const data = await res.json(); - if (data.success) { - appendToolStep('System', data.message || 'Session reset.'); - } else { - alert('Failed to reset session: ' + (data.error || 'Unknown error')); - } - } catch (err) { - alert('Error resetting session: ' + err.message); - } - } - - function handleKeyDown(e) { - if (e.key === 'Enter') { - sendMessage(); + function appendThinkingStep(text, timestamp) { + ensureAssistantCard(timestamp); + if (!currentThinkingContent) { + const step = document.createElement('div'); + step.className = 'thinking-step'; + step.style.cssText = 'margin-bottom: 10px; padding: 12px 16px; background: linear-gradient(135deg, rgba(168, 85, 247, 0.12), rgba(126, 34, 206, 0.06)); border: 1px solid rgba(168, 85, 247, 0.35); border-left: 4px solid #a855f7; border-radius: 10px; box-shadow: 0 4px 14px rgba(168, 85, 247, 0.15); font-size: 13.5px; color: #e9d5ff; backdrop-filter: blur(8px);'; + step.innerHTML = ` +
+ 🧠 REASONING PROCESS +
+
+ `; + currentToolsContainer.appendChild(step); + currentThinkingContent = step.querySelector('.thinking-text'); } + currentThinkingContent.textContent += text; + feed.scrollTop = feed.scrollHeight; } function appendUserMessage(text, timestamp) { @@ -915,6 +920,7 @@ HTML_INDEX = """ currentAssistantCard = null; currentToolsContainer = null; currentTextContent = null; + currentThinkingContent = null; currentHasPartialText = false; const card = document.createElement('div'); @@ -950,9 +956,12 @@ HTML_INDEX = """ ensureAssistantCard(timestamp); const step = document.createElement('div'); step.className = 'tool-step'; + step.style.cssText = 'margin-bottom: 10px; padding: 12px 16px; background: linear-gradient(135deg, rgba(6, 182, 212, 0.12), rgba(14, 116, 144, 0.06)); border: 1px solid rgba(6, 182, 212, 0.35); border-left: 4px solid #06b6d4; border-radius: 10px; box-shadow: 0 4px 14px rgba(6, 182, 212, 0.15); backdrop-filter: blur(8px);'; step.innerHTML = ` -
⚡ Tool Executed: ${escapeHtml(name)}
- ${detail ? `
${escapeHtml(detail)}
` : ''} +
+ ⚡ TOOL EXECUTED: ${escapeHtml(name)} +
+ ${detail ? `
${escapeHtml(detail)}
` : ''} `; currentToolsContainer.appendChild(step); feed.scrollTop = feed.scrollHeight; @@ -960,22 +969,29 @@ HTML_INDEX = """ function appendPartialReply(text, timestamp) { ensureAssistantCard(timestamp); - const formatted = linkifyFiles(escapeHtml(text)); - if (currentTextContent.innerHTML) { - currentTextContent.innerHTML += ' ' + formatted; - } else { - currentTextContent.innerHTML = formatted; + currentTextContent.style.display = 'block'; + currentTextContent.style.cssText = 'display: block; margin-top: 6px; font-size: 14.5px; line-height: 1.6; color: #f3f4f6; white-space: pre-wrap;'; + + if (currentTextContent.textContent.length > 0 && !currentTextContent.textContent.endsWith(' ') && !text.startsWith(' ')) { + currentTextContent.appendChild(document.createTextNode(' ')); } + + const span = document.createElement('span'); + span.innerHTML = linkifyFiles(escapeHtml(text)); + span.style.cssText = 'opacity: 0; transition: opacity 0.2s ease-in;'; + currentTextContent.appendChild(span); + setTimeout(() => { span.style.opacity = '1'; }, 10); + currentHasPartialText = true; feed.scrollTop = feed.scrollHeight; } function appendFinalReply(text, timestamp) { ensureAssistantCard(timestamp); - if (!currentHasPartialText) { - const formatted = linkifyFiles(escapeHtml(text)); - currentTextContent.innerHTML = formatted; - } + const formatted = linkifyFiles(escapeHtml(text)); + currentTextContent.style.display = 'block'; + currentTextContent.style.cssText = 'display: block; margin-top: 6px; font-size: 14.5px; line-height: 1.6; color: #f3f4f6; white-space: pre-wrap;'; + currentTextContent.innerHTML = formatted; feed.scrollTop = feed.scrollHeight; }