mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-04-25 00:51:20 +00:00
143 lines
4.5 KiB
Python
143 lines
4.5 KiB
Python
"""Built-in boot-md hook — run ~/.hermes/BOOT.md on gateway startup.
|
|
|
|
This hook is always registered. It silently skips if no BOOT.md exists.
|
|
To activate, create ``~/.hermes/BOOT.md`` with instructions for the
|
|
agent to execute on every gateway restart.
|
|
|
|
Example BOOT.md::
|
|
|
|
# Startup Checklist
|
|
|
|
1. Check if any cron jobs failed overnight
|
|
2. Send a status update to Discord #general
|
|
3. If there are errors in /opt/app/deploy.log, summarize them
|
|
|
|
The agent runs in a background thread so it doesn't block gateway
|
|
startup. If nothing needs attention, it replies with [SILENT] to
|
|
suppress delivery.
|
|
"""
|
|
|
|
import logging
|
|
import os
|
|
import threading
|
|
|
|
logger = logging.getLogger("hooks.boot-md")
|
|
|
|
from hermes_constants import get_hermes_home
|
|
HERMES_HOME = get_hermes_home()
|
|
BOOT_FILE = HERMES_HOME / "BOOT.md"
|
|
|
|
|
|
def _build_boot_prompt(content: str) -> str:
|
|
"""Wrap BOOT.md content in a system-level instruction."""
|
|
return (
|
|
"You are running a startup boot checklist. Follow the BOOT.md "
|
|
"instructions below exactly.\n\n"
|
|
"---\n"
|
|
f"{content}\n"
|
|
"---\n\n"
|
|
"Execute each instruction. If you need to send a message to a "
|
|
"platform, use the send_message tool.\n"
|
|
"If nothing needs attention and there is nothing to report, "
|
|
"reply with ONLY: [SILENT]"
|
|
)
|
|
|
|
|
|
def _load_boot_model() -> str:
|
|
"""Read the configured default model for boot agent runs."""
|
|
try:
|
|
from hermes_cli.config import load_config
|
|
|
|
cfg = load_config() or {}
|
|
model_cfg = cfg.get("model", {})
|
|
if isinstance(model_cfg, str):
|
|
return model_cfg
|
|
if isinstance(model_cfg, dict):
|
|
return model_cfg.get("default") or model_cfg.get("model") or ""
|
|
except Exception as exc:
|
|
logger.debug("boot-md could not load configured model: %s", exc)
|
|
return ""
|
|
|
|
|
|
def _resolve_boot_agent_kwargs() -> dict:
|
|
"""Resolve model/provider credentials for the boot agent.
|
|
|
|
Gateway-created agents resolve runtime provider settings before constructing
|
|
AIAgent. The boot hook runs outside normal message handling, so it must do
|
|
the same explicitly instead of relying on AIAgent's legacy constructor
|
|
defaults.
|
|
"""
|
|
model = _load_boot_model()
|
|
runtime = {}
|
|
try:
|
|
from hermes_cli.runtime_provider import resolve_runtime_provider
|
|
|
|
runtime = resolve_runtime_provider(
|
|
requested=os.getenv("HERMES_INFERENCE_PROVIDER"),
|
|
) or {}
|
|
except Exception as exc:
|
|
logger.warning("boot-md could not resolve runtime provider: %s", exc)
|
|
|
|
if not model and runtime.get("provider"):
|
|
try:
|
|
from hermes_cli.models import get_default_model_for_provider
|
|
|
|
model = get_default_model_for_provider(runtime["provider"]) or ""
|
|
except Exception:
|
|
pass
|
|
|
|
return {
|
|
"model": model,
|
|
"api_key": runtime.get("api_key"),
|
|
"base_url": runtime.get("base_url"),
|
|
"provider": runtime.get("provider"),
|
|
"api_mode": runtime.get("api_mode"),
|
|
"command": runtime.get("command"),
|
|
"args": list(runtime.get("args") or []),
|
|
"credential_pool": runtime.get("credential_pool"),
|
|
}
|
|
|
|
|
|
def _run_boot_agent(content: str) -> None:
|
|
"""Spawn a one-shot agent session to execute the boot instructions."""
|
|
try:
|
|
from run_agent import AIAgent
|
|
|
|
prompt = _build_boot_prompt(content)
|
|
runtime_kwargs = _resolve_boot_agent_kwargs()
|
|
agent = AIAgent(
|
|
quiet_mode=True,
|
|
skip_context_files=True,
|
|
skip_memory=True,
|
|
max_iterations=20,
|
|
**runtime_kwargs,
|
|
)
|
|
result = agent.run_conversation(prompt)
|
|
response = result.get("final_response", "")
|
|
if response and "[SILENT]" not in response:
|
|
logger.info("boot-md completed: %s", response[:200])
|
|
else:
|
|
logger.info("boot-md completed (nothing to report)")
|
|
except Exception as e:
|
|
logger.error("boot-md agent failed: %s", e)
|
|
|
|
|
|
async def handle(event_type: str, context: dict) -> None:
|
|
"""Gateway startup handler — run BOOT.md if it exists."""
|
|
if not BOOT_FILE.exists():
|
|
return
|
|
|
|
content = BOOT_FILE.read_text(encoding="utf-8").strip()
|
|
if not content:
|
|
return
|
|
|
|
logger.info("Running BOOT.md (%d chars)", len(content))
|
|
|
|
# Run in a background thread so we don't block gateway startup.
|
|
thread = threading.Thread(
|
|
target=_run_boot_agent,
|
|
args=(content,),
|
|
name="boot-md",
|
|
daemon=True,
|
|
)
|
|
thread.start()
|