mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-31 19:16:29 +00:00
feat(prompt-size): per-skill and per-toolset token-cost breakdown
`hermes prompt-size` reported skills as one <available_skills> block total and tools as one json-bytes total, so there was no way to see which installed skill or toolset actually dominates the fixed prompt budget. Add two additive breakdowns to compute_prompt_breakdown (hermes_cli/ prompt_size.py): - toolsets_breakdown: each resolved tool is attributed to its single canonical registry toolset (registry.get_tool_to_toolset_map), summed by group. Fully attributable — the grand total equals the existing tools.json_bytes minus JSON array framing (2*count bytes). - skills_breakdown: parsed from the rendered <available_skills> block, one entry per skill with two honest, distinct numbers — index_line_bytes (the always-on cost of listing the skill) and skill_md_bytes (on-disk SKILL.md size, the real read cost paid only on skill_view). Sorted largest-first by read cost. render_breakdown prints both as sorted "Toolsets by size" / "Skills by size" tables (skills capped at 20; --json carries them all). All existing keys and output are unchanged. Runs fully offline (dummy credentials, no network). Tests cover shapes, largest-first ordering, per-tool attribution reconciling to the total, namespaced-name parsing, and unmapped-skill handling. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
parent
4fe2fecf54
commit
8b9423444e
2 changed files with 255 additions and 3 deletions
|
|
@ -15,16 +15,35 @@ from __future__ import annotations
|
|||
|
||||
import json
|
||||
import re
|
||||
from typing import Any, Dict, List, Tuple
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional, Tuple
|
||||
|
||||
# The skills index is wrapped in this tag pair inside the stable tier.
|
||||
_SKILLS_BLOCK_RE = re.compile(r"<available_skills>.*?</available_skills>", re.DOTALL)
|
||||
|
||||
# A rendered skill entry inside <available_skills> is `` - name: desc`` (or
|
||||
# `` - name`` when the skill has no description). Category headers use two
|
||||
# leading spaces, so the four-space + ``- `` prefix isolates skill lines.
|
||||
_SKILL_LINE_PREFIX = " - "
|
||||
|
||||
# Cap the human-readable "Skills by size" table; ``--json`` always has them all.
|
||||
_SKILLS_TABLE_LIMIT = 20
|
||||
|
||||
|
||||
def _bytes(s: str) -> int:
|
||||
return len(s.encode("utf-8"))
|
||||
|
||||
|
||||
def _tool_name(tool: Any) -> str:
|
||||
"""Return the callable name of a tool schema (OpenAI ``function`` shape)."""
|
||||
if not isinstance(tool, dict):
|
||||
return ""
|
||||
fn = tool.get("function")
|
||||
if isinstance(fn, dict) and fn.get("name"):
|
||||
return str(fn["name"])
|
||||
return str(tool.get("name", ""))
|
||||
|
||||
|
||||
def _build_inspection_agent(platform: str) -> Any:
|
||||
"""Construct an offline AIAgent for prompt inspection.
|
||||
|
||||
|
|
@ -57,12 +76,118 @@ def _build_inspection_agent(platform: str) -> Any:
|
|||
)
|
||||
|
||||
|
||||
def _skill_md_paths_by_name() -> Dict[str, Path]:
|
||||
"""Map each installed skill's name to its ``SKILL.md`` path on disk.
|
||||
|
||||
Keyed by both the frontmatter ``name`` (what the index renders) and the
|
||||
skill directory name, so either resolves. Local skills win over external
|
||||
dirs (``get_all_skills_dirs`` yields local first), matching the index's own
|
||||
precedence. Used to attribute the real on-disk read cost per skill.
|
||||
"""
|
||||
from agent.skill_utils import (
|
||||
get_all_skills_dirs,
|
||||
iter_skill_index_files,
|
||||
parse_frontmatter,
|
||||
)
|
||||
|
||||
mapping: Dict[str, Path] = {}
|
||||
for skills_dir in get_all_skills_dirs():
|
||||
if not skills_dir.exists():
|
||||
continue
|
||||
for skill_file in iter_skill_index_files(skills_dir, "SKILL.md"):
|
||||
frontmatter_name = skill_file.parent.name
|
||||
try:
|
||||
frontmatter, _ = parse_frontmatter(
|
||||
skill_file.read_text(encoding="utf-8")
|
||||
)
|
||||
frontmatter_name = str(frontmatter.get("name") or frontmatter_name)
|
||||
except Exception:
|
||||
pass
|
||||
# setdefault keeps the first (local) occurrence on name collisions.
|
||||
mapping.setdefault(frontmatter_name, skill_file)
|
||||
mapping.setdefault(skill_file.parent.name, skill_file)
|
||||
return mapping
|
||||
|
||||
|
||||
def _compute_skills_breakdown(skills_block: str) -> List[Dict[str, Any]]:
|
||||
"""Per-skill byte breakdown parsed from the rendered ``<available_skills>``.
|
||||
|
||||
Two honest, distinct numbers per skill:
|
||||
|
||||
* ``index_line_bytes`` — the skill's actual rendered line in the always-on
|
||||
index (the fixed per-call cost of *listing* the skill).
|
||||
* ``skill_md_bytes`` — the on-disk size of the skill's ``SKILL.md`` (the
|
||||
real token cost paid only when the model loads it via ``skill_view``).
|
||||
``None`` when the name can't be mapped to a file (e.g. a plugin skill
|
||||
whose source lives outside the scanned skill dirs).
|
||||
|
||||
Sorted largest-first by ``skill_md_bytes`` (the read cost that dominates
|
||||
pruning decisions), tie-broken by name.
|
||||
"""
|
||||
name_to_path = _skill_md_paths_by_name()
|
||||
entries: List[Dict[str, Any]] = []
|
||||
for line in skills_block.splitlines():
|
||||
if not line.startswith(_SKILL_LINE_PREFIX):
|
||||
continue
|
||||
rest = line[len(_SKILL_LINE_PREFIX):]
|
||||
# ``name: desc`` — the first ``": "`` separates name from description.
|
||||
# Namespaced names (``codex:rescue``) have no space after their colon,
|
||||
# so partitioning on ``": "`` keeps the full name intact.
|
||||
name = rest.partition(": ")[0].strip()
|
||||
if not name:
|
||||
continue
|
||||
path = name_to_path.get(name)
|
||||
md_bytes: Optional[int] = None
|
||||
if path is not None:
|
||||
try:
|
||||
md_bytes = path.stat().st_size
|
||||
except OSError:
|
||||
md_bytes = None
|
||||
entries.append({
|
||||
"name": name,
|
||||
"index_line_bytes": _bytes(line),
|
||||
"skill_md_bytes": md_bytes,
|
||||
"path": str(path) if path is not None else "",
|
||||
})
|
||||
entries.sort(key=lambda e: (-(e["skill_md_bytes"] or 0), e["name"]))
|
||||
return entries
|
||||
|
||||
|
||||
def _compute_toolsets_breakdown(tools: List[Any]) -> List[Dict[str, Any]]:
|
||||
"""Per-toolset schema-byte breakdown of the resolved tool list.
|
||||
|
||||
Each tool is attributed to its single canonical toolset from the registry,
|
||||
so ``json_bytes`` sums are fully attributable: the grand total equals the
|
||||
sum of the individual tool serializations (which is the array total from
|
||||
``tools['json_bytes']`` minus JSON framing of ``2 * count`` bytes). Sorted
|
||||
largest-first by ``json_bytes``, tie-broken by toolset name.
|
||||
"""
|
||||
from tools.registry import registry
|
||||
|
||||
tool_to_toolset = registry.get_tool_to_toolset_map()
|
||||
groups: Dict[str, Dict[str, Any]] = {}
|
||||
for tool in tools:
|
||||
name = _tool_name(tool)
|
||||
toolset = tool_to_toolset.get(name) or "(unknown)"
|
||||
group = groups.setdefault(
|
||||
toolset, {"toolset": toolset, "tool_count": 0, "json_bytes": 0}
|
||||
)
|
||||
group["tool_count"] += 1
|
||||
group["json_bytes"] += _bytes(json.dumps(tool, ensure_ascii=False))
|
||||
out = list(groups.values())
|
||||
out.sort(key=lambda g: (-g["json_bytes"], g["toolset"]))
|
||||
return out
|
||||
|
||||
|
||||
def compute_prompt_breakdown(platform: str = "cli") -> Dict[str, Any]:
|
||||
"""Return a dict of prompt-size measurements for a fresh session.
|
||||
|
||||
Keys: ``system_prompt`` (chars/bytes), ``skills_index``, ``memory``,
|
||||
``user_profile``, ``tools`` (count + json bytes), and ``sections`` (a list
|
||||
of (label, chars, bytes) for the three prompt tiers).
|
||||
``user_profile``, ``tools`` (count + json bytes), ``sections`` (a list of
|
||||
(label, chars, bytes) for the three prompt tiers), ``skills_breakdown``
|
||||
(per-skill index-line + on-disk SKILL.md bytes, largest-first), and
|
||||
``toolsets_breakdown`` (per-toolset tool count + schema json bytes,
|
||||
largest-first). The last two answer "what should I disable to cut tokens?".
|
||||
"""
|
||||
from agent.system_prompt import build_system_prompt, build_system_prompt_parts
|
||||
|
||||
|
|
@ -114,6 +239,8 @@ def compute_prompt_breakdown(platform: str = "cli") -> Dict[str, Any]:
|
|||
"user_profile": {"chars": len(user_block), "bytes": _bytes(user_block)},
|
||||
"tools": {"count": len(tools), "json_bytes": _bytes(tools_json)},
|
||||
"sections": sections,
|
||||
"skills_breakdown": _compute_skills_breakdown(skills_index),
|
||||
"toolsets_breakdown": _compute_toolsets_breakdown(tools),
|
||||
}
|
||||
|
||||
|
||||
|
|
@ -143,6 +270,41 @@ def render_breakdown(data: Dict[str, Any]) -> str:
|
|||
lines.append("")
|
||||
tools = data["tools"]
|
||||
lines.append(f" Tool schemas : {tools['json_bytes']:>8,} B ({_fmt_kb(tools['json_bytes'])}, {tools['count']} tools)")
|
||||
|
||||
# Per-toolset schema cost — which toolset's tools cost the most to ship.
|
||||
toolsets = data.get("toolsets_breakdown") or []
|
||||
if toolsets:
|
||||
lines.append("")
|
||||
lines.append(" Toolsets by size (tool-schema JSON, largest first):")
|
||||
lines.append(f" {'toolset':<22} {'tools':>5} {'schema':>10}")
|
||||
for ts in toolsets:
|
||||
lines.append(
|
||||
f" {ts['toolset']:<22} {ts['tool_count']:>5} "
|
||||
f"{ts['json_bytes']:>8,} B ({_fmt_kb(ts['json_bytes'])})"
|
||||
)
|
||||
|
||||
# Per-skill cost — index line (always shipped) vs SKILL.md (read on load).
|
||||
skills = data.get("skills_breakdown") or []
|
||||
if skills:
|
||||
lines.append("")
|
||||
lines.append(
|
||||
" Skills by size (SKILL.md on-disk = read cost; index line = "
|
||||
"always-on cost, largest first):"
|
||||
)
|
||||
lines.append(f" {'skill':<28} {'SKILL.md':>10} {'index line':>10}")
|
||||
shown = skills[:_SKILLS_TABLE_LIMIT]
|
||||
for sk in shown:
|
||||
md = sk["skill_md_bytes"]
|
||||
md_str = f"{md:>8,} B" if md is not None else f"{'n/a':>10}"
|
||||
name = sk["name"]
|
||||
if len(name) > 28:
|
||||
name = name[:27] + "…"
|
||||
lines.append(
|
||||
f" {name:<28} {md_str} {sk['index_line_bytes']:>8,} B"
|
||||
)
|
||||
remaining = len(skills) - len(shown)
|
||||
if remaining > 0:
|
||||
lines.append(f" … and {remaining} more (use --json for the full list)")
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -2,6 +2,7 @@
|
|||
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
|
@ -9,6 +10,7 @@ import pytest
|
|||
from hermes_cli.prompt_size import (
|
||||
_SKILLS_BLOCK_RE,
|
||||
_build_inspection_agent,
|
||||
_compute_skills_breakdown,
|
||||
compute_prompt_breakdown,
|
||||
render_breakdown,
|
||||
)
|
||||
|
|
@ -155,6 +157,94 @@ def test_skills_block_regex_matches_tagged_block():
|
|||
assert m.group(0).endswith("</available_skills>")
|
||||
|
||||
|
||||
def test_toolsets_breakdown_reconciles_and_sorted(isolated_home):
|
||||
"""Per-toolset schema bytes attribute every tool exactly once.
|
||||
|
||||
Each resolved tool belongs to one registry toolset, so the grand total of
|
||||
per-toolset json bytes equals the whole-array total minus JSON framing
|
||||
(``2 * count`` bytes: brackets + ``", "`` separators between items).
|
||||
"""
|
||||
data = compute_prompt_breakdown("cli")
|
||||
toolsets = data["toolsets_breakdown"]
|
||||
assert toolsets # CLI always resolves at least terminal + file
|
||||
for ts in toolsets:
|
||||
assert set(ts) >= {"toolset", "tool_count", "json_bytes"}
|
||||
assert ts["tool_count"] >= 1
|
||||
assert ts["json_bytes"] > 0
|
||||
# Sorted largest-first.
|
||||
byte_sizes = [ts["json_bytes"] for ts in toolsets]
|
||||
assert byte_sizes == sorted(byte_sizes, reverse=True)
|
||||
# Every tool attributed to exactly one toolset.
|
||||
assert sum(ts["tool_count"] for ts in toolsets) == data["tools"]["count"]
|
||||
# Bytes reconcile to the existing whole-array total.
|
||||
grand = sum(ts["json_bytes"] for ts in toolsets)
|
||||
assert grand == data["tools"]["json_bytes"] - 2 * data["tools"]["count"]
|
||||
|
||||
|
||||
def test_skills_breakdown_shape_sorted_and_attributed(isolated_home):
|
||||
"""Per-skill breakdown reports index-line + on-disk SKILL.md bytes.
|
||||
|
||||
Seeded before the first build (skills prompt is cached per-process).
|
||||
"""
|
||||
_seed_skill(isolated_home, "small-skill", "short desc")
|
||||
_seed_skill(isolated_home, "big-skill", "a much longer description " * 20)
|
||||
data = compute_prompt_breakdown("cli")
|
||||
skills = data["skills_breakdown"]
|
||||
names = {s["name"] for s in skills}
|
||||
assert {"small-skill", "big-skill"} <= names
|
||||
for s in skills:
|
||||
assert set(s) >= {"name", "index_line_bytes", "skill_md_bytes", "path"}
|
||||
assert s["index_line_bytes"] > 0
|
||||
# Sorted largest-first by on-disk SKILL.md size.
|
||||
md_sizes = [s["skill_md_bytes"] or 0 for s in skills]
|
||||
assert md_sizes == sorted(md_sizes, reverse=True)
|
||||
# On-disk bytes match the real file; big-skill's SKILL.md is the larger.
|
||||
by_name = {s["name"]: s for s in skills}
|
||||
big = by_name["big-skill"]
|
||||
assert big["path"] and Path(big["path"]).stat().st_size == big["skill_md_bytes"]
|
||||
assert big["skill_md_bytes"] > by_name["small-skill"]["skill_md_bytes"]
|
||||
# Per-skill index lines are a subset of the whole <available_skills> block,
|
||||
# so they never exceed it (on-disk SKILL.md bytes are separate and don't).
|
||||
assert sum(s["index_line_bytes"] for s in skills) <= data["skills_index"]["bytes"]
|
||||
|
||||
|
||||
def test_skills_breakdown_unmapped_name_is_none():
|
||||
"""A skill line with no matching SKILL.md on disk reports None, not a crash."""
|
||||
block = (
|
||||
"<available_skills>\n"
|
||||
" demo:\n"
|
||||
" - phantom-skill: not on disk\n"
|
||||
"</available_skills>\n"
|
||||
)
|
||||
entries = _compute_skills_breakdown(block)
|
||||
assert len(entries) == 1
|
||||
assert entries[0]["name"] == "phantom-skill"
|
||||
assert entries[0]["skill_md_bytes"] is None
|
||||
assert entries[0]["path"] == ""
|
||||
assert entries[0]["index_line_bytes"] > 0
|
||||
|
||||
|
||||
def test_skills_breakdown_parses_namespaced_names():
|
||||
"""Namespaced names (``ns:skill``) survive the ``name: desc`` split."""
|
||||
block = (
|
||||
"<available_skills>\n"
|
||||
" plugins:\n"
|
||||
" - codex:rescue: rescue helper\n"
|
||||
"</available_skills>\n"
|
||||
)
|
||||
entries = _compute_skills_breakdown(block)
|
||||
assert [e["name"] for e in entries] == ["codex:rescue"]
|
||||
|
||||
|
||||
def test_render_includes_per_component_tables(isolated_home):
|
||||
"""The rendered report gains the two new sorted tables (additive)."""
|
||||
_seed_skill(isolated_home, "demo-skill", "a demo skill")
|
||||
data = compute_prompt_breakdown("cli")
|
||||
out = render_breakdown(data)
|
||||
assert "Toolsets by size" in out
|
||||
assert "Skills by size" in out
|
||||
|
||||
|
||||
def test_render_breakdown_is_plain_text(isolated_home):
|
||||
data = compute_prompt_breakdown("cli")
|
||||
out = render_breakdown(data)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue