| |
| |
| """Map any dataset tool call/result onto HARNESS_TOOLS Hermes shapes — or drop.""" |
|
|
| from __future__ import annotations |
|
|
| import json |
| import re |
| from typing import Any |
|
|
| from hermes_agent_eval import hermes_tool_content |
| from tool_runtime import HARNESS_TOOLS |
|
|
| HARNESS_NAMES = frozenset( |
| (t.get("function") or t).get("name") |
| for t in HARNESS_TOOLS |
| if (t.get("function") or t).get("name") |
| ) |
|
|
| _SEARCH_RE = re.compile( |
| r"(search|news|web|google|bing|wiki|lookup|find_info|get_news|whois|stock|movie|image|restaurant)", |
| re.I, |
| ) |
| _CALC_RE = re.compile( |
| r"(calc|math|compute|convert|currency|temperature|loan|mortgage|distance|random_number|evaluate)", |
| re.I, |
| ) |
| _VAULT_READ_RE = re.compile(r"(vault_read|read_file|read_note|get_file|open_file)", re.I) |
| _VAULT_WRITE_RE = re.compile(r"(vault_write|write_file|save_note|create_file|write_note)", re.I) |
| _VAULT_SEARCH_RE = re.compile(r"(vault_search|search_notes|search_vault|grep_vault)", re.I) |
|
|
|
|
| def _as_dict(payload: Any) -> dict | None: |
| if isinstance(payload, dict): |
| return payload |
| if isinstance(payload, str): |
| try: |
| obj = json.loads(payload) |
| except (json.JSONDecodeError, TypeError): |
| return None |
| return obj if isinstance(obj, dict) else None |
| return None |
|
|
|
|
| def _query_from_args(args: dict, human: str) -> str: |
| for k in ("query", "q", "keywords", "search_query", "stock_symbol", "company", "title", "domain"): |
| v = args.get(k) |
| if isinstance(v, list): |
| v = " ".join(str(x) for x in v if x) |
| if isinstance(v, str) and v.strip(): |
| return v.strip() |
| h = (human or "").strip() |
| return h[:120] if h else "search" |
|
|
|
|
| def _hits_from_payload(payload: dict) -> list[dict]: |
| """Build {title,url,snippet} list from common alien shapes.""" |
| data = payload.get("data") |
| if isinstance(data, dict) and isinstance(data.get("web"), list): |
| return [] |
| for key in ("results", "articles", "movies", "images", "hits", "items"): |
| raw = payload.get(key) |
| if not isinstance(raw, list) or not raw: |
| continue |
| hits = [] |
| for i, item in enumerate(raw): |
| if isinstance(item, str): |
| hits.append({"title": item, "snippet": item, "url": ""}) |
| continue |
| if not isinstance(item, dict): |
| continue |
| title = ( |
| item.get("title") |
| or item.get("name") |
| or item.get("headline") |
| or f"result {i + 1}" |
| ) |
| desc = ( |
| item.get("description") |
| or item.get("snippet") |
| or item.get("excerpt") |
| or item.get("summary") |
| or str(title) |
| ) |
| url = item.get("url") or item.get("link") or "" |
| if url and not str(url).startswith(("http://", "https://")): |
| url = "https://" + str(url).lstrip("/") |
| hits.append({"title": str(title), "snippet": str(desc), "url": str(url)}) |
| if hits: |
| return hits |
| |
| if len(payload) <= 6: |
| parts = [] |
| for k, v in payload.items(): |
| if k in ("error", "success", "ok", "data"): |
| continue |
| if isinstance(v, (str, int, float)): |
| parts.append(f"{k}: {v}") |
| elif isinstance(v, list) and v and all(isinstance(x, (str, int, float)) for x in v[:5]): |
| parts.append(f"{k}: {', '.join(str(x) for x in v[:5])}") |
| if parts: |
| text = "; ".join(parts) |
| return [{"title": text[:80], "snippet": text, "url": ""}] |
| return [] |
|
|
|
|
| def _to_web_search(name: str, args: dict, payload: Any, human: str) -> tuple[str, dict, dict] | None: |
| p = _as_dict(payload) |
| if p is None: |
| return None |
| query = _query_from_args(args or {}, human) |
| nargs = {"query": query, "limit": 5} |
| data = p.get("data") |
| if isinstance(data, dict) and isinstance(data.get("web"), list): |
| content = hermes_tool_content("web_search", p) |
| return "web_search", nargs, content |
| hits = _hits_from_payload(p) |
| if not hits: |
| return None |
| content = hermes_tool_content("web_search", {"results": hits}) |
| if not (content.get("data") or {}).get("web"): |
| return None |
| return "web_search", nargs, content |
|
|
|
|
| def _to_calculator(args: dict, payload: Any, human: str) -> tuple[str, dict, dict] | None: |
| p = _as_dict(payload) |
| if p is None: |
| return None |
| args = args or {} |
| expr = ( |
| args.get("expression") |
| or args.get("expr") |
| or p.get("expression") |
| or p.get("expr") |
| ) |
| result = p.get("result") |
| if result is None: |
| for k in ( |
| "value", |
| "output", |
| "answer", |
| "converted", |
| "converted_amount", |
| "converted_temperature", |
| "converted_length", |
| "exchange_rate", |
| "stock_price", |
| "gcd", |
| "lcm", |
| ): |
| if k in p: |
| result = p[k] |
| break |
| if expr is None: |
| if "amount" in args and ("from_currency" in args or "from" in args): |
| fr = args.get("from_currency") or args.get("from") |
| to = args.get("to_currency") or args.get("to") |
| expr = f"convert({args['amount']}, {fr!r}, {to!r})" |
| elif "temperature" in args: |
| expr = ( |
| f"convert({args['temperature']}, " |
| f"{args.get('from_unit')!r}, {args.get('to_unit')!r})" |
| ) |
| elif "length" in args: |
| expr = ( |
| f"convert({args['length']}, " |
| f"{args.get('from_unit')!r}, {args.get('to_unit')!r})" |
| ) |
| elif "base_currency" in args and "target_currency" in args: |
| expr = f"fx({args['base_currency']!r}, {args['target_currency']!r})" |
| elif "celsius" in args: |
| expr = f"convert({args['celsius']}, 'C', 'F')" |
| if expr is None or result is None: |
| return None |
| try: |
| if isinstance(result, str) and result.strip().startswith("$"): |
| result = float(re.sub(r"[^\d.]", "", result) or "nan") |
| except (TypeError, ValueError): |
| pass |
| content = hermes_tool_content( |
| "calculator", {"expression": str(expr), "result": result} |
| ) |
| return "calculator", {"expression": str(expr)}, content |
|
|
|
|
| def _to_vault( |
| kind: str, args: dict, payload: Any |
| ) -> tuple[str, dict, dict] | None: |
| p = _as_dict(payload) |
| if p is None: |
| return None |
| if kind == "vault_read": |
| path = (args or {}).get("path") or p.get("path") or "notes/note.md" |
| content = hermes_tool_content( |
| "vault_read", |
| { |
| "path": str(path), |
| "content": p.get("content") if "content" in p else json.dumps(p)[:800], |
| "bytes": int(p.get("bytes") or p.get("total_lines") or len(str(p.get("content") or ""))), |
| }, |
| ) |
| return "vault_read", {"path": str(path)}, content |
| if kind == "vault_write": |
| path = (args or {}).get("path") or p.get("path") or "notes/note.md" |
| body = (args or {}).get("content") or "" |
| content = hermes_tool_content( |
| "vault_write", |
| { |
| "ok": True, |
| "path": str(path), |
| "bytes": int(p.get("bytes") or p.get("bytes_written") or len(body)), |
| }, |
| ) |
| nargs = {"path": str(path), "content": str(body or f"note at {path}")} |
| if "append" in (args or {}): |
| nargs["append"] = bool(args["append"]) |
| return "vault_write", nargs, content |
| |
| q = (args or {}).get("query") or p.get("query") or "notes" |
| hits = p.get("hits") if isinstance(p.get("hits"), list) else [] |
| if not hits and isinstance(p.get("results"), list): |
| hits = [ |
| { |
| "path": (h.get("path") if isinstance(h, dict) else "notes/x.md"), |
| "score": (h.get("score") if isinstance(h, dict) else 0.5), |
| "excerpt": ( |
| h.get("excerpt") |
| or h.get("snippet") |
| or h.get("title") |
| or str(h) |
| ) |
| if isinstance(h, dict) |
| else str(h), |
| } |
| for h in p["results"][:5] |
| ] |
| content = hermes_tool_content( |
| "vault_search", {"hits": hits, "query": str(q)} |
| ) |
| return "vault_search", {"query": str(q)}, content |
|
|
|
|
| def classify_harness_tool(name: str) -> str | None: |
| n = (name or "").strip() |
| if n in HARNESS_NAMES: |
| return n |
| if _VAULT_READ_RE.search(n): |
| return "vault_read" |
| if _VAULT_WRITE_RE.search(n): |
| return "vault_write" |
| if _VAULT_SEARCH_RE.search(n): |
| return "vault_search" |
| if _CALC_RE.search(n): |
| return "calculator" |
| if _SEARCH_RE.search(n): |
| return "web_search" |
| return None |
|
|
|
|
| def normalize_to_harness( |
| *, |
| name: str, |
| args: dict | None, |
| payload: Any, |
| human: str = "", |
| gold: str = "", |
| ) -> dict | None: |
| """Return {name, args, payload, gold} in HARNESS Hermes shapes, or None to drop.""" |
| args = dict(args or {}) |
| target = classify_harness_tool(name) |
| if not target: |
| return None |
| if target == "web_search": |
| mapped = _to_web_search(name, args, payload, human) |
| elif target == "calculator": |
| mapped = _to_calculator(args, payload, human) |
| elif target in ("vault_read", "vault_write", "vault_search"): |
| mapped = _to_vault(target, args, payload) |
| else: |
| return None |
| if mapped is None: |
| return None |
| nname, nargs, npayload = mapped |
| |
| npayload = hermes_tool_content(nname, npayload) |
| g = (gold or "").strip() |
| if not g: |
| return None |
| return {"name": nname, "args": nargs, "payload": npayload, "gold": g} |
|
|
|
|
| def self_check(): |
| from hermes_agent_eval import hermes_tool_content as htc |
|
|
| m = normalize_to_harness( |
| name="search_news", |
| args={"keywords": ["technology"], "category": "technology"}, |
| payload={ |
| "articles": [ |
| {"title": "Tech Giants", "url": "www.technews.com/a"}, |
| {"title": "AI News", "url": "www.technews.com/b"}, |
| ] |
| }, |
| human="latest tech news", |
| gold="Here are tech articles.", |
| ) |
| assert m is not None |
| assert m["name"] == "web_search" |
| assert m["args"]["query"] == "technology" |
| assert m["payload"]["success"] is True |
| assert len(m["payload"]["data"]["web"]) == 2 |
| assert m["payload"]["data"]["web"][0]["url"].startswith("https://") |
|
|
| m2 = normalize_to_harness( |
| name="get_stock_price", |
| args={"company": "Apple"}, |
| payload={"stock_price": "$150.75"}, |
| human="Apple price?", |
| gold="Apple is $150.75.", |
| ) |
| assert m2 and m2["name"] == "web_search" |
| assert "150.75" in m2["payload"]["data"]["web"][0]["description"] |
|
|
| m3 = normalize_to_harness( |
| name="calculator", |
| args={"expression": "2+2"}, |
| payload={"expression": "2+2", "result": 4}, |
| human="2+2?", |
| gold="4", |
| ) |
| assert m3 and m3["name"] == "calculator" |
| assert m3["payload"] == htc("calculator", {"expression": "2+2", "result": 4}) |
|
|
| assert ( |
| normalize_to_harness( |
| name="send_email", |
| args={}, |
| payload={"ok": True}, |
| human="mail", |
| gold="sent", |
| ) |
| is None |
| ) |
| print("harness_normalize self_check ok") |
|
|
|
|
| if __name__ == "__main__": |
| self_check() |
|
|
