Adolfo Reyna
807 lines
34 KiB
Python
807 lines
34 KiB
Python
"""A Pipecat processor that puts Hermes in the LLM slot.
|
|
|
|
Supports:
|
|
1. Hermes CLI (`hermes chat -q ... -Q`) with session tracking (`-r <session_id>`).
|
|
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 typing import Callable, Optional
|
|
import time
|
|
from loguru import logger
|
|
|
|
import env_setup
|
|
|
|
from pipecat.frames.frames import (
|
|
CancelFrame,
|
|
EndFrame,
|
|
Frame,
|
|
InterruptionFrame,
|
|
LLMContextFrame,
|
|
LLMFullResponseEndFrame,
|
|
LLMFullResponseStartFrame,
|
|
LLMTextFrame,
|
|
StartFrame,
|
|
TextFrame,
|
|
TTSSpeakFrame,
|
|
)
|
|
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)
|
|
|
|
|
|
|
|
VOICE_PROMPT_NUDGE = """[System Instruction / Voice & UI Context:
|
|
You are communicating with the user in a real-time voice conversation over microphone and TTS, while displaying formatted responses in the Companion Web UI.
|
|
- Keep spoken responses natural, concise, and conversational (1-2 sentences per turn unless details are requested).
|
|
- Use clean Markdown formatting (bolding, code blocks, bullet points) for readability in the Web UI.
|
|
- Write text meant to be read aloud using clear, natural phrasing and conversational contractions.
|
|
- Avoid repetitive filler openers like "Certainly!", "Absolutely!", or "Great question!".
|
|
- Perform any required tools or file operations silently without narrating step-by-step internal execution.]
|
|
|
|
"""
|
|
|
|
|
|
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()
|
|
|
|
|
|
class StreamParser:
|
|
"""Parses streaming tokens, separating <think>...</think> 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 "<think>" in buf:
|
|
parts = buf.split("<think>", 1)
|
|
spoken += parts[0]
|
|
self.in_think = True
|
|
buf = parts[1]
|
|
else:
|
|
spoken += buf
|
|
buf = ""
|
|
else:
|
|
if "</think>" in buf:
|
|
parts = buf.split("</think>", 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"),
|
|
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")
|
|
|
|
|
|
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.0)) as session:
|
|
async with session.get(url_models, headers=headers) as resp:
|
|
if resp.status == 200:
|
|
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 Gateway API server not active"
|
|
|
|
|
|
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:
|
|
return True, msg
|
|
|
|
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 = 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)
|
|
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._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.
|
|
self._session_state_file = self._cwd / ".hermes-voice-session.json"
|
|
|
|
self._nudge_sent = False
|
|
|
|
# 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 reset_session(self):
|
|
"""Reset active session so a fresh session starts on the next turn."""
|
|
logger.info("Resetting active Hermes session state...")
|
|
self._session_id = None
|
|
self._session_renamed = False
|
|
self._nudge_sent = 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()
|
|
except Exception as e:
|
|
logger.warning(f"Could not remove session file during reset: {e}")
|
|
|
|
def _sync_disk_session(self):
|
|
"""Sync in-memory session ID with disk file state prior to each turn."""
|
|
disk_sid = self._load_session_id()
|
|
if disk_sid != self._session_id:
|
|
logger.info(f"Hermes session state updated from disk: {self._session_id} -> {disk_sid}")
|
|
self._session_id = disk_sid
|
|
self._session_renamed = False
|
|
self._nudge_sent = 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:
|
|
self._http_session = aiohttp.ClientSession()
|
|
return self._http_session
|
|
|
|
async def _close_http_session(self):
|
|
if self._http_session and not self._http_session.closed:
|
|
await self._http_session.close()
|
|
self._http_session = None
|
|
|
|
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 _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)
|
|
|
|
if isinstance(frame, StartFrame):
|
|
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):
|
|
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
|
|
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, suppress_output: bool = False):
|
|
self._sync_disk_model()
|
|
self._sync_disk_session()
|
|
self._history.append({"role": "user", "content": utterance})
|
|
|
|
if not suppress_output:
|
|
await self.push_frame(LLMFullResponseStartFrame())
|
|
chunks: list[str] = []
|
|
spoken_chunks: list[str] = []
|
|
|
|
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, spoken_chunks)
|
|
else:
|
|
await self._run_turn_cli(utterance, chunks, spoken_chunks)
|
|
|
|
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(spoken_chunks)) if 'spoken_chunks' in locals() and spoken_chunks else _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 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], spoken_chunks: list[str] | None = None):
|
|
"""Run turn via Hermes OpenAI-compatible Gateway API using SSE streaming (stream: true)."""
|
|
if spoken_chunks is None:
|
|
spoken_chunks = chunks
|
|
try:
|
|
ok, _ = await check_hermes_server_active(self._port)
|
|
if not ok:
|
|
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})
|
|
|
|
if not self._nudge_sent and messages:
|
|
messages[0]["content"] = VOICE_PROMPT_NUDGE + messages[0]["content"]
|
|
self._nudge_sent = True
|
|
logger.info("Injecting voice & markdown interaction context nudge into Hermes session.")
|
|
|
|
payload = {
|
|
"model": "hermes-agent" if not self._model or self._model.lower() in ("default", "none", "") else self._model,
|
|
"messages": messages,
|
|
"stream": True,
|
|
}
|
|
|
|
session = await self._get_http_session()
|
|
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)
|
|
spoken_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)
|
|
spoken_chunks.append(cleaned_rem)
|
|
await self.push_frame(LLMTextFrame(cleaned_rem))
|
|
try:
|
|
import web_server
|
|
web_server.broadcast_event("partial_reply", {"text": cleaned_rem})
|
|
except Exception:
|
|
pass
|
|
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 Gateway turn cancelled mid-response.")
|
|
raise
|
|
except Exception as e:
|
|
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], spoken_chunks: list[str] | None = None):
|
|
"""Run turn via Hermes CLI using persistent session tracking."""
|
|
if spoken_chunks is None:
|
|
spoken_chunks = chunks
|
|
|
|
prompt_to_send = utterance
|
|
if not self._nudge_sent:
|
|
prompt_to_send = VOICE_PROMPT_NUDGE + utterance
|
|
self._nudge_sent = True
|
|
logger.info("Initializing Hermes turn with voice & markdown interaction context nudge.")
|
|
|
|
cmd = [self._cli_path, "chat", "-q", prompt_to_send, "-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(">")
|
|
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
|
|
web_server.broadcast_event("tool", {"name": "Hermes CLI", "detail": cleaned})
|
|
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.
|
|
|
|
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)
|
|
spoken_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))
|