Skip to content
Open
Show file tree
Hide file tree
Changes from 3 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -61,3 +61,4 @@ docs/让*
tests/run_*.sh
tests/launch_*.py
*.launch.log
.codegraph/
81 changes: 81 additions & 0 deletions skillopt/model/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@

from skillopt.model import azure_openai as _openai
from skillopt.model import claude_backend as _claude
from skillopt.model import hermes_backend as _hermes
from skillopt.model import minimax_backend as _minimax
from skillopt.model import qwen_backend as _qwen
from skillopt.model.backend_config import ( # noqa: F401
Expand Down Expand Up @@ -55,6 +56,10 @@ def set_backend(name: str | None) -> str:
set_optimizer_backend("openai_chat")
set_target_backend("minimax_chat")
return "minimax_chat"
if normalized in {"hermes", "hermes_chat"}:
set_optimizer_backend("hermes_chat")
set_target_backend("hermes_chat")
return "hermes_chat"
raise ValueError(f"Unsupported legacy backend: {name!r}")


Expand All @@ -74,6 +79,8 @@ def get_backend_name() -> str:
return "qwen_chat"
if optimizer == "openai_chat" and target == "minimax_chat":
return "minimax_chat"
if optimizer == "hermes_chat" and target == "hermes_chat":
return "hermes_chat"
return f"{optimizer}+{target}"


Expand Down Expand Up @@ -105,6 +112,15 @@ def chat_optimizer(
reasoning_effort=reasoning_effort,
timeout=timeout,
)
if get_optimizer_backend() == "hermes_chat":
return _hermes.chat_optimizer(
system=system,
user=user,
max_completion_tokens=max_completion_tokens,
retries=retries,
stage=stage,
timeout=timeout,
)
return _openai.chat_optimizer(
system=system,
user=user,
Expand Down Expand Up @@ -153,6 +169,15 @@ def chat_target(
stage=stage,
reasoning_effort=reasoning_effort,
)
if get_target_backend() == "hermes_chat":
return _hermes.chat_target(
system=system,
user=user,
max_completion_tokens=max_completion_tokens,
retries=retries,
stage=stage,
timeout=timeout,
)
if not is_target_chat_backend():
raise NotImplementedError(
"chat_target is only supported with target_backend=openai_chat, claude_chat, qwen_chat, or minimax_chat. "
Expand Down Expand Up @@ -204,6 +229,17 @@ def chat_optimizer_messages(
return_message=return_message,
timeout=timeout,
)
if get_optimizer_backend() == "hermes_chat":
return _hermes.chat_optimizer_messages(
messages=messages,
max_completion_tokens=max_completion_tokens,
retries=retries,
stage=stage,
tools=tools,
tool_choice=tool_choice,
return_message=return_message,
timeout=timeout,
)
return _openai.chat_optimizer_messages(
messages=messages,
max_completion_tokens=max_completion_tokens,
Expand Down Expand Up @@ -263,6 +299,17 @@ def chat_target_messages(
tool_choice=tool_choice,
return_message=return_message,
)
if get_target_backend() == "hermes_chat":
return _hermes.chat_target_messages(
messages=messages,
max_completion_tokens=max_completion_tokens,
retries=retries,
stage=stage,
tools=tools,
tool_choice=tool_choice,
return_message=return_message,
timeout=timeout,
)
if not is_target_chat_backend():
raise NotImplementedError(
"chat_target_messages is only supported with target_backend=openai_chat, claude_chat, qwen_chat, or minimax_chat. "
Expand Down Expand Up @@ -294,6 +341,18 @@ def chat_messages_with_deployment(
return_message: bool = False,
timeout: int | None = None,
) -> tuple[Any, dict]:
if get_optimizer_backend() == "hermes_chat" or get_target_backend() == "hermes_chat":
return _hermes.chat_messages_with_deployment(
deployment=deployment,
messages=messages,
max_completion_tokens=max_completion_tokens,
retries=retries,
stage=stage,
tools=tools,
tool_choice=tool_choice,
return_message=return_message,
timeout=timeout,
)
return _openai.chat_messages_with_deployment(
deployment=deployment,
messages=messages,
Expand All @@ -318,6 +377,16 @@ def chat_with_deployment(
reasoning_effort: str | None = None,
timeout: int | None = None,
) -> tuple[str, dict]:
if get_optimizer_backend() == "hermes_chat" or get_target_backend() == "hermes_chat":
return _hermes.chat_with_deployment(
deployment=deployment,
system=system,
user=user,
max_completion_tokens=max_completion_tokens,
retries=retries,
stage=stage,
timeout=timeout,
)
return _openai.chat_with_deployment(
deployment=deployment,
system=system,
Expand Down Expand Up @@ -365,6 +434,17 @@ def get_token_summary() -> dict:
summary[stage]["prompt_tokens"] += values["prompt_tokens"]
summary[stage]["completion_tokens"] += values["completion_tokens"]
summary[stage]["total_tokens"] += values["total_tokens"]
hermes_summary = _hermes.get_token_summary()
for stage, values in hermes_summary.items():
if stage == "_total":
continue
if stage not in summary:
summary[stage] = values
continue
summary[stage]["calls"] += values["calls"]
summary[stage]["prompt_tokens"] += values["prompt_tokens"]
summary[stage]["completion_tokens"] += values["completion_tokens"]
summary[stage]["total_tokens"] += values["total_tokens"]
total = {
"calls": 0,
"prompt_tokens": 0,
Expand All @@ -387,6 +467,7 @@ def reset_token_tracker() -> None:
_claude.reset_token_tracker()
_qwen.reset_token_tracker()
_minimax.reset_token_tracker()
_hermes.reset_token_tracker()


def configure_azure_openai(
Expand Down
8 changes: 4 additions & 4 deletions skillopt/model/backend_config.py
Original file line number Diff line number Diff line change
Expand Up @@ -49,10 +49,10 @@ def _parse_int(value: str | None, default: int) -> int:
def set_optimizer_backend(backend: str) -> None:
global OPTIMIZER_BACKEND
OPTIMIZER_BACKEND = normalize_backend_name(backend or "openai_chat")
if OPTIMIZER_BACKEND not in {"openai_chat", "claude_chat", "qwen_chat", "minimax_chat"}:
if OPTIMIZER_BACKEND not in {"openai_chat", "claude_chat", "qwen_chat", "minimax_chat", "hermes_chat"}:
raise ValueError(
f"Unsupported optimizer backend: {OPTIMIZER_BACKEND!r}. "
"Supported values are 'openai_chat', 'claude_chat', 'qwen_chat', and 'minimax_chat'."
"Supported values are 'openai_chat', 'claude_chat', 'qwen_chat', 'minimax_chat', and 'hermes_chat'."
)
os.environ["OPTIMIZER_BACKEND"] = OPTIMIZER_BACKEND

Expand All @@ -64,10 +64,10 @@ def get_optimizer_backend() -> str:
def set_target_backend(backend: str) -> None:
global TARGET_BACKEND
TARGET_BACKEND = normalize_backend_name(backend or "openai_chat")
if TARGET_BACKEND not in {"openai_chat", "claude_chat", "qwen_chat", "minimax_chat", "codex_exec", "claude_code_exec"}:
if TARGET_BACKEND not in {"openai_chat", "claude_chat", "qwen_chat", "minimax_chat", "codex_exec", "claude_code_exec", "hermes_chat"}:
raise ValueError(
f"Unsupported target backend: {TARGET_BACKEND!r}. "
"Supported values are 'openai_chat', 'claude_chat', 'qwen_chat', 'minimax_chat', 'codex_exec', and 'claude_code_exec'."
"Supported values are 'openai_chat', 'claude_chat', 'qwen_chat', 'minimax_chat', 'codex_exec', 'claude_code_exec', and 'hermes_chat'."
)
os.environ["TARGET_BACKEND"] = TARGET_BACKEND

Expand Down
184 changes: 184 additions & 0 deletions skillopt/model/hermes_backend.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,184 @@
"""Hermes CLI chat backend for SkillOpt.

Chama `hermes --profile <name> chat -q "<prompt>"` como target/optimizer.
Mais simples que claude_backend: sem tools, imagens, ou attachments.
"""
from __future__ import annotations

import json
import os
import subprocess
import time
from typing import Any

from skillopt.model.common import CompatAssistantMessage, CompatToolCall, CompatToolFunction, default_model_for_backend, tracker

HERMES_BIN = os.environ.get("HERMES_BIN", "hermes")
HERMES_TARGET_PROFILE = os.environ.get("HERMES_TARGET_PROFILE", "default")
HERMES_OPTIMIZER_PROFILE = os.environ.get("HERMES_OPTIMIZER_PROFILE", "default")

OPTIMIZER_DEPLOYMENT = os.environ.get("OPTIMIZER_DEPLOYMENT", "default")
TARGET_DEPLOYMENT = os.environ.get("TARGET_DEPLOYMENT", "default")


def _call_hermes(prompt: str, profile: str, timeout: int | None = None) -> tuple[str, dict[str, int]]:
"""Call hermes CLI and return (response_text, token_info)."""
cmd = [HERMES_BIN, "--profile", profile, "chat", "-q", prompt]
t0 = time.time()
proc = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=timeout or 180,
env={**os.environ, "HERMES_NO_COLOR": "1"},
)
elapsed = time.time() - t0
if proc.returncode != 0:
stderr = (proc.stderr or "").strip()
raise RuntimeError(stderr or f"Hermes CLI exited with code {proc.returncode}")

text = (proc.stdout or "").strip()
tokens_in = len(prompt) // 4
tokens_out = len(text) // 4
return text, {
"prompt_tokens": tokens_in,
"completion_tokens": tokens_out,
"total_tokens": tokens_in + tokens_out,
}


def _build_prompt(system: str, user: str) -> str:
"""Build a prompt string from system + user messages."""
parts = []
if system:
parts.append(system)
if user:
parts.append(user)
return "\n\n".join(parts)


def chat_optimizer(system: str, user: str, max_completion_tokens: int = 16384, retries: int = 3, stage: str = "optimizer", timeout: int | None = None) -> tuple[str, dict[str, int]]:
"""Call Hermes as optimizer with profile=target."""
del max_completion_tokens
prompt = _build_prompt(system, user)
last_err = None
for attempt in range(retries):
try:
text, usage = _call_hermes(prompt, HERMES_OPTIMIZER_PROFILE, timeout=timeout)
tracker.record(stage, usage["prompt_tokens"], usage["completion_tokens"])
return text, usage
except Exception as e:
last_err = e
time.sleep(min(2 ** attempt, 10))
raise RuntimeError(f"Hermes optimizer backend failed after {retries} retries: {last_err}")


def chat_target(system: str, user: str, max_completion_tokens: int = 16384, retries: int = 3, stage: str = "target", timeout: int | None = None) -> tuple[str, dict[str, int]]:
"""Call Hermes as target with profile=target."""
del max_completion_tokens
prompt = _build_prompt(system, user)
last_err = None
for attempt in range(retries):
try:
text, usage = _call_hermes(prompt, HERMES_TARGET_PROFILE, timeout=timeout)
tracker.record(stage, usage["prompt_tokens"], usage["completion_tokens"])
return text, usage
except Exception as e:
last_err = e
time.sleep(min(2 ** attempt, 10))
raise RuntimeError(f"Hermes target backend failed after {retries} retries: {last_err}")


def chat_with_deployment(deployment: str, system: str, user: str, max_completion_tokens: int = 16384, retries: int = 3, stage: str = "custom", timeout: int | None = None) -> tuple[str, dict[str, int]]:
"""Call Hermes with a custom profile name as deployment."""
del max_completion_tokens
profile = deployment or HERMES_TARGET_PROFILE
prompt = _build_prompt(system, user)
last_err = None
for attempt in range(retries):
try:
text, usage = _call_hermes(prompt, profile, timeout=timeout)
tracker.record(stage, usage["prompt_tokens"], usage["completion_tokens"])
return text, usage
except Exception as e:
last_err = e
time.sleep(min(2 ** attempt, 10))
raise RuntimeError(f"Hermes backend (deployment={deployment}) failed after {retries} retries: {last_err}")


# ── Message-based variants (needed for tool-using benchmarks like spreadsheetbench) ──

def chat_optimizer_messages(messages: list[dict[str, Any]], max_completion_tokens: int = 16384, retries: int = 3, stage: str = "optimizer", *, tools: list[dict[str, Any]] | None = None, tool_choice: str | dict[str, Any] | None = None, return_message: bool = False, timeout: int | None = None) -> tuple[Any, dict[str, int]]:
"""Simplified: flatten messages to prompt text."""
del max_completion_tokens, tools, tool_choice, return_message
parts = []
for msg in messages:
role = msg.get("role", "user")
content = msg.get("content", "")
if isinstance(content, list):
texts = [c.get("text", "") for c in content if isinstance(c, dict) and c.get("type") == "text"]
content = "\n".join(texts)
parts.append(f"<{role}>\n{content}")
prompt = "\n".join(parts)
text, usage = _call_hermes(prompt, HERMES_OPTIMIZER_PROFILE, timeout=timeout)
tracker.record(stage, usage["prompt_tokens"], usage["completion_tokens"])
return text, usage


def chat_target_messages(messages: list[dict[str, Any]], max_completion_tokens: int = 16384, retries: int = 3, stage: str = "target", *, tools: list[dict[str, Any]] | None = None, tool_choice: str | dict[str, Any] | None = None, return_message: bool = False, timeout: int | None = None) -> tuple[Any, dict[str, int]]:
"""Simplified: flatten messages to prompt text."""
del max_completion_tokens, tools, tool_choice, return_message
parts = []
for msg in messages:
role = msg.get("role", "user")
content = msg.get("content", "")
if isinstance(content, list):
texts = [c.get("text", "") for c in content if isinstance(c, dict) and c.get("type") == "text"]
content = "\n".join(texts)
parts.append(f"<{role}>\n{content}")
prompt = "\n".join(parts)
text, usage = _call_hermes(prompt, HERMES_TARGET_PROFILE, timeout=timeout)
tracker.record(stage, usage["prompt_tokens"], usage["completion_tokens"])
return text, usage


def chat_messages_with_deployment(deployment: str, messages: list[dict[str, Any]], max_completion_tokens: int = 16384, retries: int = 3, stage: str = "custom", *, tools: list[dict[str, Any]] | None = None, tool_choice: str | dict[str, Any] | None = None, return_message: bool = False, timeout: int | None = None) -> tuple[Any, dict[str, int]]:
"""Simplified: flatten messages to prompt text."""
del max_completion_tokens, tools, tool_choice, return_message
profile = deployment or HERMES_TARGET_PROFILE
parts = []
for msg in messages:
role = msg.get("role", "user")
content = msg.get("content", "")
if isinstance(content, list):
texts = [c.get("text", "") for c in content if isinstance(c, dict) and c.get("type") == "text"]
content = "\n".join(texts)
parts.append(f"<{role}>\n{content}")
prompt = "\n".join(parts)
text, usage = _call_hermes(prompt, profile, timeout=timeout)
tracker.record(stage, usage["prompt_tokens"], usage["completion_tokens"])
return text, usage


def get_token_summary() -> dict[str, dict[str, int]]:
return tracker.summary()


def reset_token_tracker() -> None:
tracker.reset()


def set_reasoning_effort(effort: str | None) -> None:
pass # Not applicable for Hermes


def set_target_deployment(deployment: str) -> None:
global TARGET_DEPLOYMENT
TARGET_DEPLOYMENT = deployment or default_model_for_backend("hermes")
os.environ["TARGET_DEPLOYMENT"] = TARGET_DEPLOYMENT


def set_optimizer_deployment(deployment: str) -> None:
global OPTIMIZER_DEPLOYMENT
OPTIMIZER_DEPLOYMENT = deployment or default_model_for_backend("hermes")
os.environ["OPTIMIZER_DEPLOYMENT"] = OPTIMIZER_DEPLOYMENT
Loading