hermes-relay/hermes_client.py

"""
Hermes Client — directe Python import (geen subprocess overhead).

Warme agent: eenmalig initialiseren, daarna hergebruiken voor elke query.
Dit bespaart de subprocess + Python startup overhead van 2-5 seconden per query.

Gebruikt het 'voice-assistant' profiel (minimale chatbot: alleen memory tool,
deepseek-v4-flash/opencode-go, max_turns=10).
"""
import io
import os
import sys
import time
import logging
from contextlib import redirect_stdout, redirect_stderr

logger = logging.getLogger(__name__)

# ── Eenmalige Hermes imports (bij module laden) ──────────────────────
_HERMES_LOADED = False
_cli = None

# Voice-assistant profiel home directory
_HERMES_HOME = "/root/.hermes/profiles/voice-assistant"


def _ensure_hermes_loaded():
    """Laad Hermes eenmalig bij eerste gebruik — dit is het zware werk."""
    global _HERMES_LOADED, _cli
    if _HERMES_LOADED:
        return

    start = time.time()

    # Forceer voice-assistant profiel via HERMES_HOME
    os.environ["HERMES_HOME"] = _HERMES_HOME

    # Zorg dat Hermes in de Python path zit
    hermes_path = "/usr/local/lib/hermes-agent/venv/lib/python3.11/site-packages"
    if hermes_path not in sys.path:
        sys.path.insert(0, hermes_path)

    # Force UTF-8
    os.environ.setdefault("PYTHONIOENCODING", "utf-8")

    # Laad .env via Hermes' eigen loader
    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(hermes_path), ".env")
    load_hermes_dotenv(hermes_home=_hermes_home, project_env=_project_env)

    # Load config (nu vanuit voice-assistant profiel)
    from hermes_cli.config import load_config
    config = load_config()

    # Gebruik messaging toolset zodat agent send_message tool kan gebruiken
    toolsets = ["messaging"]

    # YOLO mode + interactive mode
    os.environ["HERMES_YOLO_MODE"] = "1"
    os.environ["HERMES_INTERACTIVE"] = "1"
    os.environ["HERMES_SESSION_SOURCE"] = "relay"

    # Import HermesCLI
    from cli import HermesCLI

    # Maak CLI instantie (nog géén agent init — dat gebeurt pas bij eerste query)
    _cli = HermesCLI(
        model=config.get("model", {}).get("default", ""),
        toolsets=toolsets,
    )
    # Onderdruk banner/status output
    _cli.tool_progress_mode = "off"

    elapsed = time.time() - start
    logger.info(
        "Hermes loaded in %.2fs (toolsets: %s, profile: voice-assistant)",
        elapsed, toolsets
    )
    _HERMES_LOADED = True


def call_hermes(prompt):
    """
    Roep Hermes aan — hergebruikt een warme agent.

    De agent wordt één keer geïnitialiseerd (bij eerste query) en daarna
    hergebruikt. Na elke query wordt de conversation history gereset,
    zodat elke query een frisse start krijgt.
    """
    _ensure_hermes_loaded()

    # Geen prefix meer — voice-assistant profiel handelt dit af via config
    full_prompt = prompt.strip()

    start = time.time()

    # ── Agent initialiseren ──
    turn_route = _cli._resolve_turn_agent_config(full_prompt)

    if turn_route["signature"] != _cli._active_agent_route_signature:
        _cli.agent = None
    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:
        return "⚠️ Kon Hermes agent niet initialiseren."

    # Stil modus
    _cli.agent.quiet_mode = True
    _cli.agent.suppress_status_output = True
    _cli.agent.stream_delta_callback = None
    _cli.agent.tool_gen_callback = None

    # ── Query uitvoeren ──
    # Capture alles wat naar stdout/stderr gaat (session_id, banner, etc.)
    stdout_capture = io.StringIO()
    stderr_capture = io.StringIO()

    try:
        with redirect_stdout(stdout_capture), redirect_stderr(stderr_capture):
            # conversation_history wissen voor frisse start
            _cli.agent.messages = []
            result = _cli.agent.chat(full_prompt)
    except SystemExit:
        # Hermes roept sys.exit() aan in sommige error paths
        logger.warning("Hermes riep sys.exit() aan tijdens query")
        result = stdout_capture.getvalue().strip()
    except Exception as e:
        logger.error("Hermes error: %s", e, exc_info=True)
        return "⚠️ Hermes error: " + str(e)

    elapsed = time.time() - start

    # Fallback als Hermes None teruggeeft (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