"""A Pipecat processor that puts Hermes in the LLM slot. Supports: 1. Hermes CLI (`hermes chat -q ... -Q`) with session tracking (`-r `). 2. Hermes Server / Gateway Daemon (`http://localhost:8642` or `http://localhost:4096`) if active. """ import asyncio import json import os import re import shutil from pathlib import Path import aiohttp from loguru import logger import env_setup from pipecat.frames.frames import ( CancelFrame, EndFrame, Frame, InterruptionFrame, LLMContextFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, LLMTextFrame, StartFrame, ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor _NOISE_TRANSCRIPTS = { "", ".", "thank you.", "thanks for watching!", "you", "bye.", "okay.", "[blank_audio]", "[silence]", } ANSI_ESCAPE = re.compile(r'\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])') SESSION_ID_REGEX = re.compile(r"\bsession_id:\s*([^\s]+)", re.IGNORECASE) def _strip_ansi(text: str) -> str: if not text: return "" return ANSI_ESCAPE.sub("", text).strip() def _clean_spoken_text(text: str) -> str: """Clean text for speech output and truncate fake turn generations.""" if not text: return "" cleaned_line = _strip_ansi(text).strip() # Filter out Hermes CLI session headers and status indicators if ( cleaned_line.startswith("↻") or "Resumed session" in cleaned_line or "session_id:" in cleaned_line.lower() ): return "" # Strip leading or inline role headers (e.g. "Assistant:") text = re.sub(r"(?i)\b(Assistant|assistant|Bot|bot):\s*", "", text) # Truncate if model hallucinates fake subsequent user turns for marker in ("\nUser:", "\nHuman:", "\nUser", "\nHuman"): if marker in text: text = text.split(marker)[0] # Remove markdown code blocks text = re.sub(r"```[\s\S]*?```", "", text) # Remove inline code ticks text = re.sub(r"`[^`]*`", "", text) # Remove markdown syntax characters text = re.sub(r"[\#\*\_\~]", "", text) # Flatten newlines into clear speech lines = [line.strip() for line in text.splitlines() if line.strip()] return " ".join(lines).strip() def find_hermes_cli() -> str | None: candidates = [ shutil.which("hermes"), os.path.expanduser("~/.hermes/bin/hermes"), os.path.expanduser("~/.local/bin/hermes"), "/opt/homebrew/bin/hermes", "/usr/local/bin/hermes", ] for candidate in candidates: if candidate and os.path.exists(candidate) and os.access(candidate, os.X_OK): return candidate return shutil.which("hermes") async def ensure_hermes_server(port: int = 8642) -> tuple[bool, str]: """Ensure Hermes gateway server daemon is available on port.""" url = f"http://localhost:{port}/api/health" try: async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=2.0)) as session: async with session.get(url) as resp: if resp.status == 200: return True, f"Hermes server active on http://localhost:{port}" except Exception: pass cli = find_hermes_cli() if not cli: return False, "Hermes CLI binary not found" return True, f"Hermes CLI ready ({cli})" def probe_hermes(model: str | None = None) -> tuple[bool, str]: cli = find_hermes_cli() if cli: 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.""" def __init__( self, *, model: str | None = None, cwd: str | Path | None = None, port: int = 8642, session_name: str = "Voice Agent", observer=None, use_server: bool = False, **kwargs, ): super().__init__(**kwargs) self._model = model self._cwd = Path(cwd or Path(__file__).parent).expanduser().resolve() self._port = port self._session_name = session_name self._on_reply = observer 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._session_renamed = False # Keep the conversation lineage with the workspace. A single global # session file can make two voice-agent workspaces resume each other's # Hermes conversations. self._session_state_file = self._cwd / ".hermes-voice-session.json" # Persisted session ID self._session_id: str | None = self._load_session_id() if self._session_id: logger.info(f"Loaded existing Hermes session ID: {self._session_id}") def _load_session_id(self) -> str | None: if self._session_state_file.exists(): try: data = json.loads(self._session_state_file.read_text()) sid = data.get("session_id") if sid and isinstance(sid, str): return sid.strip() except Exception as e: logger.debug(f"Could not load Hermes session state: {e}") return None def _save_session_id(self): try: self._session_state_file.parent.mkdir(parents=True, exist_ok=True) if self._session_id: self._session_state_file.write_text( json.dumps({"session_id": self._session_id}, indent=2) + "\n" ) elif self._session_state_file.exists(): self._session_state_file.unlink() except Exception as e: logger.warning(f"Could not save Hermes session state: {e}") def _remember_session_id(self, text: str): if not text: return match = SESSION_ID_REGEX.search(text) if match: new_sid = match.group(1).strip() if new_sid and new_sid != self._session_id: self._session_id = new_sid logger.info(f"Hermes active session tracking ID: {self._session_id}") self._save_session_id() if not self._session_renamed and self._session_name: asyncio.create_task(self._rename_session(new_sid)) async def _rename_session(self, session_id: str): self._session_renamed = True try: proc = await asyncio.create_subprocess_exec( self._cli_path, "sessions", "rename", session_id, self._session_name, stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.DEVNULL, ) await proc.wait() logger.info(f"Renamed Hermes session {session_id} to '{self._session_name}'") except Exception as e: logger.debug(f"Could not rename Hermes session: {e}") async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) if isinstance(frame, StartFrame): await self.push_frame(frame, direction) available, reason = probe_hermes(self._model) logger.info(f"Hermes LLM engine initialized: {reason}") elif isinstance(frame, (EndFrame, CancelFrame)): await self._cancel_turn() await self.push_frame(frame, direction) elif isinstance(frame, InterruptionFrame): await self._cancel_turn() await self.push_frame(frame, direction) elif isinstance(frame, LLMContextFrame): text = self._latest_user_text(frame.context) await self._maybe_start_turn(text) 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._maybe_start_turn(utterance)) def _latest_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): text_parts = [c.get("text", "") for c in content if isinstance(c, dict)] return " ".join(text_parts) return "" async def _maybe_start_turn(self, text: str): utterance = text.strip() if utterance.lower() in _NOISE_TRANSCRIPTS or len(utterance) < 2: logger.debug(f"Ignoring noise transcript: {utterance!r}") return await self._cancel_turn() logger.info(f"You: {utterance}") self._turn_task = asyncio.create_task(self._run_turn(utterance)) async def _cancel_turn(self): if not self._turn_task: return task, self._turn_task = self._turn_task, None if not task.done(): task.cancel() try: await task except asyncio.CancelledError: pass def set_model(self, model_name: str): if model_name != self._model: logger.info(f"Hermes LLM model set to: {model_name}") self._model = model_name def _sync_disk_model(self): config_file = self._cwd / "model_settings.json" if config_file.exists(): try: data = json.loads(config_file.read_text()) if "model" in data and isinstance(data["model"], str) and data["model"].strip(): new_model = data["model"].strip() if new_model != self._model: logger.info(f"Hermes LLM switching active model: {self._model} -> {new_model}") self._model = new_model except Exception as e: logger.debug(f"Could not read model settings: {e}") async def _run_turn(self, utterance: str): self._sync_disk_model() self._history.append({"role": "user", "content": utterance}) await self.push_frame(LLMFullResponseStartFrame()) chunks: list[str] = [] if 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()) 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) async def _run_turn_server(self, utterance: str, chunks: list[str]): """Run turn via Hermes Server / Gateway HTTP API if available.""" try: ok, _ = await ensure_hermes_server(self._port) if not ok: raise RuntimeError("Hermes server daemon unavailable") session_id = self._session_id or "voice-agent" url = f"http://localhost:{self._port}/api/sessions/{session_id}/chat" payload = { "message": utterance, } if self._model and self._model.lower() not in ("default", "none", ""): payload["model"] = self._model async with aiohttp.ClientSession() as 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)) try: import web_server web_server.broadcast_event("partial_reply", {"text": cleaned}) 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}") except asyncio.CancelledError: logger.info("Hermes server turn cancelled mid-response.") raise except Exception as e: logger.warning(f"Hermes server error ({e}), falling back to CLI...") 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"] if self._session_id: cmd.extend(["-r", self._session_id]) if self._model and self._model.lower() not in ("default", "none", ""): cmd.extend(["-m", self._model]) proc = None try: env = {**os.environ, "PYTHONUNBUFFERED": "1", "FORCE_COLOR": "0", "NO_COLOR": "1"} proc = await asyncio.create_subprocess_exec( *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, stdin=asyncio.subprocess.DEVNULL, cwd=str(self._cwd), env=env, ) async def _read_stderr(stream): while True: line = await stream.readline() if not line: break decoded = line.decode("utf-8", errors="replace").strip() cleaned = _strip_ansi(decoded) # Extract session_id emitted on stderr self._remember_session_id(cleaned) if cleaned and cleaned not in ("[0m", "0m", "]") and not cleaned.startswith(">"): logger.info(f"Hermes Tool: {cleaned}") try: import web_server web_server.broadcast_event("tool", {"name": "Hermes CLI", "detail": cleaned}) except Exception: pass # Keep tool progress visible in logs and the Companion Web UI, but # do not send implementation details through the spoken channel. stderr_task = asyncio.create_task(_read_stderr(proc.stderr)) while True: line = await proc.stdout.readline() if not line: break text_line = line.decode("utf-8", errors="replace") # Extract session_id if present in stdout as fallback self._remember_session_id(text_line) cleaned = _clean_spoken_text(text_line) if cleaned: chunks.append(cleaned) await self.push_frame(LLMTextFrame(cleaned)) try: import web_server web_server.broadcast_event("partial_reply", {"text": cleaned}) except Exception: pass await proc.wait() await stderr_task # If CLI failed due to an invalid session resume ID, clear session ID for next turn if proc.returncode != 0 and self._session_id and not chunks: logger.warning(f"Hermes CLI returned code {proc.returncode}, clearing session ID for retry...") self._session_id = None self._save_session_id() if not chunks: err_msg = "Done." chunks.append(err_msg) await self.push_frame(LLMTextFrame(err_msg)) except asyncio.CancelledError: logger.info("Hermes turn cancelled mid-response.") if proc and proc.returncode is None: try: proc.kill() except Exception: pass raise except Exception as e: logger.error(f"Hermes LLM CLI error: {e}") err_msg = "Sorry, I ran into an error generating a response." chunks.append(err_msg) await self.push_frame(LLMTextFrame(err_msg))