""" Hermes Client — True warm agent for zero-latency queries. Architecture: - Agent is initialized ONCE at module load (or first use) and kept warm - Only reinitialize if model/provider/runtime actually changes (route signature) - No conversation history reset - each query is naturally stateless - All config via environment variables for deployment flexibility """ import os import sys import time import logging import signal import atexit import threading from typing import Optional, Dict, Any, Tuple, Union logger = logging.getLogger(__name__) # ── Configuration via environment variables ───────────────────────────── HERMES_HOME = os.environ.get("HERMES_RELAY_HERMES_HOME", "/root/.hermes/profiles/voice-assistant") HERMES_SITE_PACKAGES = os.environ.get("HERMES_RELAY_SITE_PACKAGES", "/usr/local/lib/hermes-agent/venv/lib/python3.11/site-packages") HERMES_MODEL = os.environ.get("HERMES_RELAY_MODEL", "") # Empty = use profile default HERMES_TOOLSETS = os.environ.get("HERMES_RELAY_TOOLSETS", "messaging").split(",") HERMES_MAX_TURNS = int(os.environ.get("HERMES_RELAY_MAX_TURNS", "10")) HERMES_TIMEOUT = int(os.environ.get("HERMES_RELAY_TIMEOUT", "120")) LOG_LEVEL = os.environ.get("HERMES_RELAY_LOG_LEVEL", "INFO").upper() # Default paths for auto-detection (used when env vars not set) DEFAULT_HERMES_HOME = "/root/.hermes/profiles/voice-assistant" DEFAULT_SITE_PACKAGES = "/usr/local/lib/hermes-agent/venv/lib/python3.11/site-packages" # ── Module-level state (true warm agent) ──────────────────────────────── _HERMES_LOADED = False _CLI = None _ACTIVE_ROUTE_SIGNATURE = None _SHUTTING_DOWN = False # Thread lock for protecting global state modifications _STATE_LOCK = threading.Lock() def _setup_logging() -> None: """Configure logging once.""" logging.basicConfig( level=getattr(logging, LOG_LEVEL, logging.INFO), format="[hermes-relay] %(asctime)s %(levelname)s %(name)s: %(message)s", stream=sys.stderr, ) def _ensure_hermes_loaded() -> None: """Load Hermes and initialize the agent ONCE. This is the heavy lifting.""" global _HERMES_LOADED, _CLI, _ACTIVE_ROUTE_SIGNATURE # Fast path - already loaded if _HERMES_LOADED: return # Slow path - acquire lock and double-check with _STATE_LOCK: if _HERMES_LOADED: return start = time.time() _setup_logging() # Force voice-assistant profile via HERMES_HOME os.environ["HERMES_HOME"] = HERMES_HOME or DEFAULT_HERMES_HOME # Ensure Hermes is in Python path site_packages = HERMES_SITE_PACKAGES or DEFAULT_SITE_PACKAGES if site_packages not in sys.path: sys.path.insert(0, site_packages) # Force UTF-8 os.environ.setdefault("PYTHONIOENCODING", "utf-8") # Load .env via Hermes' own loader (loads profile .env + project .env) from hermes_cli.env_loader import load_hermes_dotenv from hermes_constants import get_hermes_home _hermes_home = get_hermes_home() _project_env = os.path.join(os.path.dirname(site_packages), ".env") load_hermes_dotenv(hermes_home=_hermes_home, project_env=_project_env) # Load config from voice-assistant profile from hermes_cli.config import load_config config = load_config() # YOLO mode + interactive mode (required for agent tools) os.environ["HERMES_YOLO_MODE"] = "1" os.environ["HERMES_INTERACTIVE"] = "1" os.environ["HERMES_SESSION_SOURCE"] = "relay" # Import HermesCLI from cli import HermesCLI # Determine model (env override > profile default) model = HERMES_MODEL or config.get("model", {}).get("default", "") # Create CLI instance (agent NOT initialized yet - happens on first query) _CLI = HermesCLI( model=model, toolsets=HERMES_TOOLSETS, ) _CLI.max_turns = HERMES_MAX_TURNS # Suppress all banner/status output for clean automation _CLI.tool_progress_mode = "off" _CLI.verbose = False _CLI.streaming_enabled = False elapsed = time.time() - start logger.info( "Hermes client loaded in %.2fs (toolsets=%s, model=%s, max_turns=%d, profile=voice-assistant)", elapsed, HERMES_TOOLSETS, model or "profile-default", HERMES_MAX_TURNS ) _HERMES_LOADED = True def _initialize_agent() -> bool: """ Initialize the agent if not already initialized, or if route signature changed. Returns True on success, False on failure. """ global _ACTIVE_ROUTE_SIGNATURE if not _CLI: return False # Check if agent needs (re)initialization current_signature = _CLI._resolve_turn_agent_config("")["signature"] if _CLI.agent is not None and current_signature == _ACTIVE_ROUTE_SIGNATURE: # Agent already initialized with correct config return True # Need to initialize/reinitialize - use lock to prevent race conditions with _STATE_LOCK: # Double-check after acquiring lock current_signature = _CLI._resolve_turn_agent_config("")["signature"] if _CLI.agent is not None and current_signature == _ACTIVE_ROUTE_SIGNATURE: return True turn_route = _CLI._resolve_turn_agent_config("") init_ok = _CLI._init_agent( model_override=turn_route["model"], runtime_override=turn_route["runtime"], request_overrides=turn_route.get("request_overrides"), ) if not init_ok: logger.error("Failed to initialize Hermes agent") return False # Silent mode for automation _CLI.agent.quiet_mode = True _CLI.agent.suppress_status_output = True _CLI.agent.stream_delta_callback = None _CLI.agent.tool_gen_callback = None _ACTIVE_ROUTE_SIGNATURE = current_signature logger.info("Hermes agent initialized (route_signature=%s)", _ACTIVE_ROUTE_SIGNATURE) return True def call_hermes(prompt: str) -> str: """ Query the warm Hermes agent. The agent is initialized once at startup and reused for all queries. Only reinitializes if model/provider/base_url actually changes. """ if _SHUTTING_DOWN: return "⚠️ Service shutting down, please retry." _ensure_hermes_loaded() if not _initialize_agent(): return "⚠️ Kon Hermes agent niet initialiseren." full_prompt = prompt.strip() if not full_prompt: return "⚠️ Lege prompt ontvangen." start = time.time() try: # Each query is a fresh conversation - reset messages # This is intentional: we want stateless queries for the relay _CLI.agent.messages = [] result = _CLI.agent.chat(full_prompt) except KeyboardInterrupt: logger.warning("Hermes query interrupted") raise # Re-raise to allow proper shutdown except SystemExit as e: # Hermes calls sys.exit() in some error paths logger.warning("Hermes called sys.exit() during query: %s", e) return "⚠️ Hermes error: process exited unexpectedly" except Exception as e: logger.error("Hermes error: %s", e, exc_info=True) # Return user-friendly message but log full traceback return f"⚠️ Hermes error: {type(e).__name__}" elapsed = time.time() - start # Fallback if Hermes returns None (error path) if result is None: logger.warning("Hermes returned None — possible error during query") return "⚠️ Geen antwoord van Hermes. Probeer opnieuw." logger.info("Hermes responded in %.1fs (%d chars)", elapsed, len(result)) return result def health_check() -> Dict[str, Any]: """ Check if Hermes agent is healthy and responsive. Returns a dict with status details. This is a LIGHTWEIGHT check - no actual LLM query is made. For deep health checking, use the /health endpoint with detail=true. """ checks = { "hermes_loaded": _HERMES_LOADED, "agent_initialized": _CLI is not None and _CLI.agent is not None, "route_signature": str(_ACTIVE_ROUTE_SIGNATURE) if _ACTIVE_ROUTE_SIGNATURE else None, } # Lightweight check - just verify agent object exists # Do NOT run an actual query here as it blocks for 10-60s if checks["agent_initialized"]: checks["query_test"] = "skipped (use deep check)" checks["query_latency_ms"] = None else: checks["query_test"] = "agent_not_initialized" checks["query_latency_ms"] = None checks["healthy"] = ( checks["hermes_loaded"] and checks["agent_initialized"] ) return checks def shutdown() -> None: """Graceful shutdown handler.""" global _SHUTTING_DOWN with _STATE_LOCK: _SHUTTING_DOWN = True logger.info("Shutdown initiated") if _CLI and _CLI.agent: try: # Give agent a chance to clean up if hasattr(_CLI.agent, "shutdown"): _CLI.agent.shutdown() except Exception as e: logger.error("Error during agent shutdown: %s", e, exc_info=True) logger.info("Shutdown complete") # Register shutdown handlers signal.signal(signal.SIGTERM, lambda s, f: shutdown()) signal.signal(signal.SIGINT, lambda s, f: shutdown()) atexit.register(shutdown)