hermes-skills/hermes-skills-research
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Publish hermes-skills-research via gitea-publish skill

Browse Source
This commit is contained in:
figmar
2026-08-09 08:19:33 +08:00
commit bca5c4f6fa
67 changed files with 29416 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
grounded-citations/scripts/_hermes_home.py
+23
View File
@@ -0,0 +1,23 @@
"""Resolve HERMES_HOME for standalone skill scripts.
Skill scripts may run outside the Hermes process (system Python, nix env,
CI) where ``hermes_constants`` is not importable. This module provides the
same ``get_hermes_home()`` contract without requiring it on ``sys.path``.
When ``hermes_constants`` IS available it is used directly so profile
resolution and any future enhancements are picked up automatically.
"""
from __future__ import annotations
import os
from pathlib import Path
try:
from hermes_constants import get_hermes_home as get_hermes_home
except (ModuleNotFoundError, ImportError):
def get_hermes_home() -> Path:
"""Return the Hermes home directory (default: ``~/.hermes``)."""
val = os.environ.get("HERMES_HOME", "").strip()
return Path(val) if val else Path.home() / ".hermes"
grounded-citations/scripts/sources.py
+678
View File
@@ -0,0 +1,678 @@
#!/usr/bin/env python3
"""Citation ledger for grounded answers and documents.
Owns the ``url -> [n]`` mapping used by the ``grounded-citations`` skill.
Ids are assigned at retrieval time and never change, so a draft's ``[3]``
always resolves to the same page. The model only ever emits integers the
ledger handed it, which is what makes the citations verifiable.
Subcommands
-----------
reset start a clean ledger
add URL [URL ...] register source(s), print their ids
ingest FILE|- register every url found in JSON tool output
quote ID --text T --from FILE|- attach verbatim supporting evidence to a source
list show the ledger
render render a Sources block
verify DRAFT check a draft's citations against the ledger
Fact-checking is evidence-backed citation: ``quote`` only accepts text that
literally appears in the fetched page text you point it at, ``verify
--evidence`` requires every cited source to carry at least one such quote, and
``render --style evidence`` prints the quotes under each source so the reader
can check the chain themselves. Claims from model knowledge are declared with
an ``[unverified]`` marker rather than silently blended in.
Ledger path resolution (first wins):
--ledger PATH
$HERMES_CITATION_LEDGER
$HERMES_HOME/cache/citations/ledger.json
"""
from __future__ import annotations
import argparse
import json
import os
import re
import sys
import time
from pathlib import Path
from typing import Any, Iterable
sys.path.insert(0, str(Path(__file__).resolve().parent))
from _hermes_home import get_hermes_home # noqa: E402
SCHEMA_VERSION = 1
# A citation marker in prose: [12]. Markdown links ([text](url)) and
# reference-style labels are excluded by requiring digits only and no
# following "(" or ":".
_CITE_RE = re.compile(r"\[(\d{1,4})\](?![(:])")
_SOURCES_HEADER_RE = re.compile(r"^\s*(?:#{1,6}\s*)?(?:\*\*)?sources:?(?:\*\*)?\s*$", re.IGNORECASE)
_SOURCE_LINE_RE = re.compile(r"^\s*\[(\d{1,4})\]\s*[-–:]?\s*(\S+)")
_URL_IN_TEXT_RE = re.compile(r"https?://[^\s\"'<>)\]}]+")
_FENCE_RE = re.compile(r"^\s*(?:```|~~~)")
# Explicit declaration that a claim comes from model knowledge, not a source.
_UNVERIFIED_RE = re.compile(r"\[unverified\]", re.IGNORECASE)
# ---------------------------------------------------------------------------
# Ledger I/O
# ---------------------------------------------------------------------------
def resolve_ledger_path(explicit: str | None = None) -> Path:
if explicit:
return Path(explicit).expanduser()
env = os.environ.get("HERMES_CITATION_LEDGER", "").strip()
if env:
return Path(env).expanduser()
return get_hermes_home() / "cache" / "citations" / "ledger.json"
def normalize_url(url: str) -> str:
"""Canonicalize a URL for ledger identity.
Strips the fragment and a trailing slash so ``/page``, ``/page/`` and
``/page#section`` are one source. Query strings are significant and are
kept — they usually select different content.
"""
u = (url or "").strip()
if "#" in u:
u = u.split("#", 1)[0]
stripped = u.rstrip("/")
return stripped or u
def load_ledger(path: Path) -> dict[str, Any]:
if not path.exists():
return {"version": SCHEMA_VERSION, "sources": []}
try:
data = json.loads(path.read_text(encoding="utf-8"))
except (json.JSONDecodeError, OSError) as exc:
raise SystemExit(f"error: ledger at {path} is unreadable ({exc}); run `reset` to start over")
if not isinstance(data, dict) or not isinstance(data.get("sources"), list):
raise SystemExit(f"error: ledger at {path} has an unexpected shape; run `reset`")
return data
def save_ledger(path: Path, data: dict[str, Any]) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
tmp = path.with_suffix(path.suffix + f".tmp{os.getpid()}")
tmp.write_text(json.dumps(data, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
os.replace(tmp, path)
class _LedgerLock:
"""Best-effort cross-process lock (O_EXCL lockfile, stdlib only).
Parallel subagents can share one ledger via --ledger; without a lock two
concurrent ``add`` calls can assign the same id. Falls through after the
timeout rather than blocking a task forever — a stale lock must never
wedge the ledger.
"""
def __init__(self, path: Path, timeout: float = 5.0) -> None:
self.lock_path = path.with_suffix(path.suffix + ".lock")
self.timeout = timeout
self.fd: int | None = None
def __enter__(self) -> "_LedgerLock":
self.lock_path.parent.mkdir(parents=True, exist_ok=True)
deadline = time.monotonic() + self.timeout
while True:
try:
self.fd = os.open(str(self.lock_path), os.O_CREAT | os.O_EXCL | os.O_WRONLY)
return self
except FileExistsError:
if time.monotonic() >= deadline:
# Assume a stale lock from a crashed run.
try:
self.lock_path.unlink()
except OSError:
return self
continue
time.sleep(0.05)
def __exit__(self, *_exc: object) -> None:
if self.fd is not None:
try:
os.close(self.fd)
except OSError:
pass
try:
self.lock_path.unlink()
except OSError:
pass
# ---------------------------------------------------------------------------
# Core operations
# ---------------------------------------------------------------------------
def _by_url(sources: list[dict[str, Any]]) -> dict[str, dict[str, Any]]:
return {s["url"]: s for s in sources}
def add_sources(
path: Path,
urls: Iterable[str],
title: str | None = None,
accessed: str | None = None,
) -> list[dict[str, Any]]:
"""Register urls, returning their ledger entries (existing or new)."""
urls = [u for u in (str(u).strip() for u in urls) if u]
if not urls:
return []
with _LedgerLock(path):
data = load_ledger(path)
sources = data["sources"]
index = _by_url(sources)
out: list[dict[str, Any]] = []
changed = False
for raw in urls:
key = normalize_url(raw)
existing = index.get(key)
if existing is not None:
if title and not existing.get("title"):
existing["title"] = title
changed = True
out.append(existing)
continue
entry = {
"id": len(sources) + 1,
"url": key,
"title": (title or "").strip(),
"accessed": accessed or time.strftime("%Y-%m-%d"),
}
sources.append(entry)
index[key] = entry
out.append(entry)
changed = True
if changed:
save_ledger(path, data)
return out
def urls_from_json(payload: Any) -> list[tuple[str, str]]:
"""Walk arbitrary JSON tool output collecting (url, title) pairs.
Handles web_search (``data.web[]``), web_extract (``results[]``) and any
other nesting, in document order, deduped.
"""
found: list[tuple[str, str]] = []
seen: set[str] = set()
def walk(node: Any) -> None:
if isinstance(node, dict):
url = node.get("url") or node.get("link") or node.get("source_url")
if isinstance(url, str) and url.startswith(("http://", "https://")):
key = normalize_url(url)
if key not in seen:
seen.add(key)
raw_title = node.get("title") or node.get("name") or ""
found.append((url, raw_title if isinstance(raw_title, str) else ""))
for value in node.values():
walk(value)
elif isinstance(node, list):
for item in node:
walk(item)
walk(payload)
return found
# ---------------------------------------------------------------------------
# Evidence quotes (fact-checking)
# ---------------------------------------------------------------------------
def _normalize_ws(text: str) -> str:
"""Collapse all whitespace runs to single spaces for verbatim matching."""
return " ".join((text or "").split())
# Markdown artifacts that retrieval tools inject into otherwise-identical prose.
# ``web_extract`` returns markdown, so the most citation-worthy sentences are
# exactly the ones carrying inline links and emphasis around terms:
# "including _[ERAP1](https://…/erap1/)_, _[IL1A](…)_, have also been…"
# reads identically to the page a human sees. Matching has to see through that
# markup, or the skill pushes the agent toward weaker evidence fragments.
_MD_LINK_RE = re.compile(r"\[([^\]]*)\]\((?:[^()\s]|\([^()]*\))*\)")
_MD_NOISE_RE = re.compile(r"[*_`~]|\\(?=[^\w\s])")
def _match_key(text: str) -> str:
"""Canonicalize text for verbatim comparison.
Whitespace-, case-, and markdown-insensitive: inline links collapse to
their label, emphasis/code markers and backslash escapes are dropped. The
stored quote keeps whatever the caller passed, so the rendered evidence
block shows clean prose rather than extractor artifacts.
"""
collapsed = _MD_LINK_RE.sub(r"\1", text or "")
return _normalize_ws(_MD_NOISE_RE.sub("", collapsed)).casefold()
def quote_in_evidence(quote: str, evidence: str) -> bool:
"""True when ``quote`` appears verbatim in the fetched ``evidence`` text,
ignoring whitespace, case, and markdown markup on either side."""
q = _match_key(quote)
return bool(q) and q in _match_key(evidence)
def attach_quote(path: Path, source_id: int, quote: str, evidence: str) -> dict[str, Any]:
"""Attach a verbatim quote to a ledger entry after checking it against
the evidence text. Raises SystemExit on unknown id or non-verbatim text —
a quote the page does not contain is exactly the fabrication this guards
against."""
quote = (quote or "").strip()
if len(_normalize_ws(quote).split()) < 3:
raise SystemExit("error: quote too short — use at least 3 words of verbatim text")
if not quote_in_evidence(quote, evidence):
raise SystemExit(
"error: quote not found verbatim in the evidence text — "
"copy the exact wording from the fetched page, do not paraphrase"
)
with _LedgerLock(path):
data = load_ledger(path)
entry = next((s for s in data["sources"] if s["id"] == source_id), None)
if entry is None:
raise SystemExit(f"error: no source [{source_id}] in the ledger")
quotes = entry.setdefault("quotes", [])
norm = _match_key(quote)
if not any(_match_key(q.get("text", "")) == norm for q in quotes):
quotes.append({"text": quote, "added": time.strftime("%Y-%m-%d")})
save_ledger(path, data)
return entry
def render_sources(
sources: list[dict[str, Any]],
style: str = "markdown",
only: set[int] | None = None,
) -> str:
picked = [s for s in sources if only is None or s["id"] in only]
picked.sort(key=lambda s: s["id"])
if not picked:
return ""
lines: list[str] = []
if style == "bibtex":
for s in picked:
key = f"source{s['id']}"
title = s.get("title") or s["url"]
lines.append(
"@misc{%s,\n title = {%s},\n howpublished = {\\url{%s}},\n note = {Accessed %s}\n}"
% (key, title, s["url"], s.get("accessed", ""))
)
return "\n".join(lines)
if style == "footnotes":
for s in picked:
title = s.get("title")
suffix = f" — {title}" if title else ""
lines.append(f"[^{s['id']}]: {s['url']}{suffix}")
return "\n".join(lines)
header = "Sources:" if style == "plain" else "## Sources"
lines.append(header)
if style != "plain":
lines.append("")
for s in picked:
title = s.get("title")
suffix = f" — {title}" if title else ""
lines.append(f"[{s['id']}] {s['url']}{suffix}")
if style == "evidence":
for q in s.get("quotes", []):
lines.append(f' > "{q.get("text", "")}"')
return "\n".join(lines)
# ---------------------------------------------------------------------------
# Verification
# ---------------------------------------------------------------------------
def _split_draft(text: str) -> tuple[str, dict[int, str]]:
"""Split a draft into (prose, sources_block_map).
The sources block is everything after the last ``Sources`` header; its
``[n] url`` lines are parsed out so they aren't counted as prose citations.
Fenced code blocks are dropped from prose.
"""
lines = text.splitlines()
header_idx = -1
for i, line in enumerate(lines):
if _SOURCES_HEADER_RE.match(line):
header_idx = i
listed: dict[int, str] = {}
if header_idx >= 0:
for line in lines[header_idx + 1:]:
m = _SOURCE_LINE_RE.match(line)
if m:
url_match = _URL_IN_TEXT_RE.search(line)
listed[int(m.group(1))] = url_match.group(0) if url_match else m.group(2)
body_lines = lines[:header_idx]
else:
body_lines = lines
prose: list[str] = []
in_fence = False
for line in body_lines:
if _FENCE_RE.match(line):
in_fence = not in_fence
continue
if not in_fence:
prose.append(line)
return "\n".join(prose), listed
def _strip_sources_block(text: str) -> str:
"""Return the draft with its trailing Sources block removed.
Everything from the last Sources header onward goes; a draft with no such
header is returned unchanged. This is what makes ``render --replace-in``
idempotent instead of stacking duplicate blocks.
"""
lines = text.splitlines()
header_idx = -1
for i, line in enumerate(lines):
if _SOURCES_HEADER_RE.match(line):
header_idx = i
if header_idx < 0:
return text
return "\n".join(lines[:header_idx])
def _sentences(prose: str) -> list[str]:
"""Rough sentence split over prose lines, skipping headings and tables."""
out: list[str] = []
for line in prose.splitlines():
stripped = line.strip()
if not stripped or stripped.startswith("#") or stripped.startswith("|"):
continue
if stripped.startswith(">"):
stripped = stripped.lstrip("> ").strip()
for part in re.split(r"(?<=[.!?])\s+", stripped):
part = part.strip()
if len(part.split()) >= 4:
out.append(part)
return out
def verify_draft(
draft_path: Path,
sources: list[dict[str, Any]],
strict: bool = False,
min_coverage: float | None = None,
require_evidence: bool = False,
) -> tuple[int, list[str], list[str]]:
"""Return (exit_code, errors, warnings)."""
text = draft_path.read_text(encoding="utf-8")
prose, listed = _split_draft(text)
by_id = {s["id"]: s for s in sources}
errors: list[str] = []
warnings: list[str] = []
cited = [int(m) for m in _CITE_RE.findall(prose)]
cited_set = set(cited)
unknown = sorted(i for i in cited_set if i not in by_id)
if unknown:
errors.append(
"citations not in the ledger (hallucinated or renumbered): "
+ ", ".join(f"[{i}]" for i in unknown)
)
if cited_set and not listed:
errors.append("draft cites sources but has no `Sources:` block — run `render --cited-in`")
missing_from_block = sorted(cited_set - set(listed)) if listed else []
if missing_from_block:
errors.append(
"cited but absent from the Sources block: "
+ ", ".join(f"[{i}]" for i in missing_from_block)
)
for sid, url in sorted(listed.items()):
entry = by_id.get(sid)
if entry is None:
errors.append(f"Sources block lists [{sid}], which is not in the ledger")
continue
if normalize_url(url) != entry["url"]:
errors.append(
f"Sources block URL for [{sid}] does not match the ledger "
f"(block: {url} / ledger: {entry['url']}) — re-run `render`"
)
extra_in_block = sorted(set(listed) - cited_set)
if extra_in_block:
warnings.append(
"listed in Sources but never cited inline: "
+ ", ".join(f"[{i}]" for i in extra_in_block)
)
registered_uncited = sorted(set(by_id) - cited_set)
if registered_uncited:
warnings.append(
"registered in the ledger but not cited in this draft: "
+ ", ".join(f"[{i}]" for i in registered_uncited)
)
sentences = _sentences(prose)
cited_sentences = [s for s in sentences if _CITE_RE.search(s)]
unverified_sentences = [s for s in sentences if _UNVERIFIED_RE.search(s)]
covered = [s for s in sentences if _CITE_RE.search(s) or _UNVERIFIED_RE.search(s)]
coverage = (len(covered) / len(sentences)) if sentences else 0.0
if min_coverage is not None and sentences and coverage < min_coverage:
errors.append(
f"citation coverage {coverage:.0%} is below the required {min_coverage:.0%} "
f"({len(covered)}/{len(sentences)} sentences cited or marked [unverified])"
)
if require_evidence:
unevidenced = sorted(
i for i in cited_set if i in by_id and not by_id[i].get("quotes")
)
if unevidenced:
errors.append(
"cited sources carry no verbatim evidence quote (run `quote` with the "
"fetched page text): " + ", ".join(f"[{i}]" for i in unevidenced)
)
over_cited = [s for s in sentences if len(_CITE_RE.findall(s)) > 3]
if over_cited:
warnings.append(f"{len(over_cited)} sentence(s) carry more than 3 citations")
code = 1 if errors else (1 if (strict and warnings) else 0)
quoted = sum(1 for s in sources if s.get("quotes"))
stats = (
f"{len(sentences)} prose sentence(s), {len(covered)} with declared provenance "
f"({coverage:.0%}) — {len(cited_sentences)} cited, "
f"{len(unverified_sentences)} marked [unverified] (a sentence may be both); "
f"{len(cited_set)} distinct source(s) cited, "
f"{len(by_id)} in ledger ({quoted} with evidence quotes)"
)
warnings.insert(0, f"stats: {stats}")
return code, errors, warnings
# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def _parse_only(spec: str | None) -> set[int] | None:
if not spec:
return None
out: set[int] = set()
for chunk in spec.replace(" ", "").split(","):
if not chunk:
continue
if "-" in chunk:
lo, _, hi = chunk.partition("-")
out.update(range(int(lo), int(hi) + 1))
else:
out.add(int(chunk))
return out
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(
prog="sources.py", description="Citation ledger for grounded answers and documents."
)
parser.add_argument("--ledger", help="ledger file path (overrides env / default)")
sub = parser.add_subparsers(dest="cmd", required=True)
sub.add_parser("reset", help="start a clean ledger")
p_add = sub.add_parser("add", help="register source url(s), print their ids")
p_add.add_argument("urls", nargs="+")
p_add.add_argument("--title", help="title for the source (single-url calls)")
p_add.add_argument("--accessed", help="access date (default: today)")
p_add.add_argument("--json", action="store_true", help="emit JSON instead of ids")
p_ing = sub.add_parser("ingest", help="register every url in JSON tool output")
p_ing.add_argument("file", help="JSON file, or - for stdin")
p_q = sub.add_parser("quote", help="attach verbatim supporting evidence to a source")
p_q.add_argument("id", type=int, help="ledger id of the source the quote supports")
p_q.add_argument("--text", required=True, help="the exact quote, copied from the page")
p_q.add_argument(
"--from",
dest="evidence",
required=True,
help="file with the fetched page text (or - for stdin) the quote must appear in",
)
p_list = sub.add_parser("list", help="show the ledger")
p_list.add_argument("--json", action="store_true")
p_render = sub.add_parser("render", help="render a Sources block")
p_render.add_argument(
"--style", default="markdown", choices=["markdown", "plain", "footnotes", "bibtex", "evidence"]
)
p_render.add_argument("--only", help="ids to include, e.g. 1,3,5-7")
p_render.add_argument("--cited-in", help="include only ids cited in this draft file")
p_render.add_argument(
"--replace-in",
help="rewrite this draft's Sources block in place (implies --cited-in on it)",
)
p_ver = sub.add_parser("verify", help="check a draft's citations against the ledger")
p_ver.add_argument("draft")
p_ver.add_argument("--strict", action="store_true", help="treat warnings as failures")
p_ver.add_argument("--min-coverage", type=float, help="required cited-sentence share, e.g. 0.5")
p_ver.add_argument(
"--evidence",
action="store_true",
help="require every cited source to carry at least one verbatim quote",
)
args = parser.parse_args(argv)
path = resolve_ledger_path(args.ledger)
if args.cmd == "reset":
with _LedgerLock(path):
save_ledger(path, {"version": SCHEMA_VERSION, "sources": []})
print(f"ledger reset: {path}")
return 0
if args.cmd == "add":
title = args.title if len(args.urls) == 1 else None
entries = add_sources(path, args.urls, title=title, accessed=args.accessed)
if args.json:
print(json.dumps(entries, indent=2, ensure_ascii=False))
else:
for e in entries:
print(f"[{e['id']}] {e['url']}")
return 0
if args.cmd == "ingest":
raw = sys.stdin.read() if args.file == "-" else Path(args.file).read_text(encoding="utf-8")
try:
payload = json.loads(raw)
except json.JSONDecodeError as exc:
print(f"error: input is not valid JSON ({exc})", file=sys.stderr)
return 2
pairs = urls_from_json(payload)
if not pairs:
print("no urls found in input", file=sys.stderr)
return 1
for url, title in pairs:
entry = add_sources(path, [url], title=title or None)[0]
print(f"[{entry['id']}] {entry['url']}")
return 0
if args.cmd == "quote":
raw = (
sys.stdin.read()
if args.evidence == "-"
else Path(args.evidence).read_text(encoding="utf-8")
)
entry = attach_quote(path, args.id, args.text, raw)
print(f"[{entry['id']}] evidence attached ({len(entry.get('quotes', []))} quote(s))")
return 0
data = load_ledger(path)
sources = sorted(data["sources"], key=lambda s: s["id"])
if args.cmd == "list":
if args.json:
print(json.dumps(sources, indent=2, ensure_ascii=False))
elif not sources:
print(f"ledger is empty: {path}")
else:
for s in sources:
title = f" {s['title']}" if s.get("title") else ""
nq = len(s.get("quotes", []))
mark = f" ({nq} quote{'s' if nq != 1 else ''})" if nq else ""
print(f"[{s['id']}] {s['url']}{title}{mark}")
return 0
if args.cmd == "render":
only = _parse_only(args.only)
draft_for_ids = args.replace_in or args.cited_in
if draft_for_ids:
draft = Path(draft_for_ids).read_text(encoding="utf-8")
prose, _ = _split_draft(draft)
cited = {int(m) for m in _CITE_RE.findall(prose)}
only = cited if only is None else (only & cited)
block = render_sources(sources, style=args.style, only=only)
if not block:
print("no sources to render", file=sys.stderr)
return 1
if args.replace_in:
target = Path(args.replace_in)
body = _strip_sources_block(target.read_text(encoding="utf-8"))
target.write_text(body.rstrip("\n") + "\n\n" + block + "\n", encoding="utf-8")
print(f"Sources block rewritten in {target}")
return 0
print(block)
return 0
if args.cmd == "verify":
draft_path = Path(args.draft)
if not draft_path.is_file():
print(f"error: no such draft: {draft_path}", file=sys.stderr)
return 2
code, errors, warnings = verify_draft(
draft_path,
sources,
strict=args.strict,
min_coverage=args.min_coverage,
require_evidence=args.evidence,
)
for w in warnings:
prefix = "info" if w.startswith("stats: ") else "warn"
print(f"{prefix}: {w}")
for e in errors:
print(f"FAIL: {e}", file=sys.stderr)
print("citations OK" if code == 0 else "verification failed")
return code
return 2
if __name__ == "__main__":
sys.exit(main())