369 lines
15 KiB
Python
369 lines
15 KiB
Python
"""Check the numeric claims in public texts (README, blog posts, guide, drafts) against the validation reports.
|
|
|
|
Code finds the claims and the candidate evidence; a TypeSafe System One model (Jev) only judges whether a passage
|
|
supports, contradicts or says nothing about a claim (the "double checking citations" pattern,
|
|
https://docs.typesafe.ai/cookbooks/citation_check). Only this repository's own documentation text is sent; no data,
|
|
no sample sheets, no vendor files. Needs ``TYPESAFE_API_KEY`` in the environment, or in
|
|
``~/.config/typesafe/credentials`` (mode 600), for the judgment step.
|
|
|
|
python scripts/docs/check_claims.py README.md blog/*.md # check, write claims_report.md
|
|
python scripts/docs/check_claims.py --dry-run linkedin_draft.txt # extraction and retrieval only, no API calls
|
|
|
|
Verdicts: verified (a passage supports the claim, confidence >= threshold), contradicted, unsupported (passages found
|
|
but none addresses the claim), no_source (no passage contains any of the claim's numbers), review (low confidence).
|
|
Exit code 1 when any claim is contradicted or has no source, so the script can gate a publishing step.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
API_URL = "https://api.typesafe.ai/v1/systemone"
|
|
MODEL = "jev-latest"
|
|
AUTO_ACCEPT = 0.8 # cookbook default; evaluate on our own claims before trusting it
|
|
TOP_K = 3
|
|
DEFAULT_SOURCES = ["docs/validation", "docs/known_unknowns.md", "SIMURGARRAY_FUNCTIONAL_SPEC.md"]
|
|
CACHE = ROOT / ".cache" / "claims"
|
|
|
|
RELATION = {
|
|
"type": "choice",
|
|
"criteria": {
|
|
"supports": "The section states the claim's numbers and meaning, or directly implies that the claim is true",
|
|
"contradicts": "The section gives a different value for the same quantity, or implies the claim is false",
|
|
"says_nothing": "The section does not address what the claim asserts, either way",
|
|
},
|
|
}
|
|
|
|
# ------------------------------------------------------------------ numbers
|
|
|
|
_NUM = re.compile(r"(?<![\w.])(\d{1,3}(?:[ ,]\d{3})+|\d+(?:\.\d+)?)(\s?%)?")
|
|
|
|
|
|
def numbers(text: str) -> set[str]:
|
|
"""Normalised number tokens of a text, with percent/fraction twins: '99.34%' -> {'99.34', '0.9934'}."""
|
|
out: set[str] = set()
|
|
for m in _NUM.finditer(text):
|
|
raw, pct = m.group(1), m.group(2)
|
|
v = raw.replace(",", "").replace(" ", "")
|
|
try:
|
|
f = float(v)
|
|
except ValueError:
|
|
continue
|
|
out.add(_fmt(f))
|
|
if pct:
|
|
out.add(_fmt(f / 100))
|
|
elif 0 < f < 1:
|
|
out.add(_fmt(f * 100))
|
|
return out
|
|
|
|
|
|
def _fmt(f: float) -> str:
|
|
s = f"{f:.6f}".rstrip("0").rstrip(".")
|
|
return s or "0"
|
|
|
|
|
|
_MONTHS = "January|February|March|April|May|June|July|August|September|October|November|December"
|
|
_DATE = re.compile(
|
|
rf"\b({_MONTHS})\s+\d{{4}}\b|\b({_MONTHS})\s+\d{{1,2}}(,\s*\d{{4}})?|\b(19|20)\d{{2}}\b|\d{{4}}-\d{{2}}-\d{{2}}"
|
|
r"|\b(version|v)\s?\d+(\.\d+)+\b"
|
|
)
|
|
|
|
|
|
def significant(nums: set[str]) -> set[str]:
|
|
"""Numbers worth matching: drop tiny integers that appear everywhere (0 to 12)."""
|
|
return {n for n in nums if not (n.isdigit() and int(n) <= 12)}
|
|
|
|
|
|
def claim_numbers(text: str) -> set[str]:
|
|
"""Significant numbers of a claim; dates and years are context, not claims."""
|
|
return significant(numbers(_DATE.sub(" ", text)))
|
|
|
|
|
|
# ------------------------------------------------------------------ claims
|
|
|
|
|
|
@dataclass
|
|
class Claim:
|
|
path: str
|
|
line: int
|
|
text: str
|
|
context: str
|
|
nums: set[str]
|
|
|
|
|
|
def _strip_md(s: str) -> str:
|
|
s = re.sub(r"<[^>]+>", " ", s)
|
|
s = re.sub(r"!\[[^\]]*\]\([^)]*\)", " ", s)
|
|
s = re.sub(r"\[([^\]]*)\]\([^)]*\)", r"\1", s)
|
|
s = re.sub(r"https?://\S+", " ", s)
|
|
return re.sub(r"\s+", " ", s.replace("**", "").replace("`", "")).strip()
|
|
|
|
|
|
def extract_claims(path: Path, text: str) -> list[Claim]:
|
|
"""Sentences of paragraphs and bullets, and table rows, that carry significant numbers. Lines of one paragraph
|
|
or bullet are joined first, so a sentence wrapped over several lines stays whole. Code blocks and commands are
|
|
skipped."""
|
|
claims: list[Claim] = []
|
|
heading, table_head, in_code = "", "", False
|
|
buf: list[str] = []
|
|
start = 0
|
|
|
|
def add(unit: str, line: int) -> None:
|
|
nums = claim_numbers(unit)
|
|
if nums and not re.search(r"\b(simurg|uv|pip|bash|docker|conda) ", unit):
|
|
ctx = " / ".join(x for x in (heading, table_head) if x)
|
|
claims.append(Claim(str(path), line, unit, ctx, nums))
|
|
|
|
def flush() -> None:
|
|
if buf:
|
|
para = _strip_md(" ".join(buf))
|
|
for u in re.split(r"(?<=[.!?])\s+(?=[A-Z])", para):
|
|
if u:
|
|
add(u, start)
|
|
buf.clear()
|
|
|
|
for i, raw in enumerate([*text.splitlines(), ""], 1):
|
|
ln = raw.rstrip()
|
|
if ln.lstrip().startswith("```"):
|
|
flush()
|
|
in_code = not in_code
|
|
continue
|
|
if in_code:
|
|
continue
|
|
if not ln.strip():
|
|
flush()
|
|
table_head = ""
|
|
continue
|
|
if ln.startswith("#"):
|
|
flush()
|
|
heading = _strip_md(ln.lstrip("#"))
|
|
continue
|
|
if ln.lstrip().startswith("|"):
|
|
flush()
|
|
if re.fullmatch(r"\s*\|[\s:|-]+\|?\s*", ln):
|
|
continue
|
|
if not table_head:
|
|
table_head = _strip_md(ln)
|
|
continue
|
|
add(_strip_md(ln), i)
|
|
continue
|
|
bullet = re.match(r"\s*([*>-]|\d+\.)\s+", ln)
|
|
if bullet:
|
|
flush()
|
|
ln = ln[bullet.end() :]
|
|
if not buf:
|
|
start = i
|
|
buf.append(ln.strip())
|
|
return claims
|
|
|
|
|
|
# ------------------------------------------------------------------ evidence
|
|
|
|
|
|
@dataclass
|
|
class Passage:
|
|
path: str
|
|
line: int
|
|
text: str
|
|
nums: set[str] = field(default_factory=set)
|
|
|
|
|
|
def load_passages(sources: list[Path]) -> list[Passage]:
|
|
"""Paragraphs and table rows (with their header) of the source documents."""
|
|
out: list[Passage] = []
|
|
files = [p for s in sources for p in (sorted(s.rglob("*.md")) if s.is_dir() else [s])]
|
|
for f in files:
|
|
rel = str(f.relative_to(ROOT)) if f.is_relative_to(ROOT) else str(f)
|
|
heading, table_head = "", ""
|
|
buf: list[str] = []
|
|
start = 0
|
|
for i, ln in enumerate([*f.read_text(encoding="utf-8").splitlines(), ""], 1):
|
|
is_row = ln.lstrip().startswith("|")
|
|
if (ln.startswith("#") or is_row or not ln.strip()) and buf:
|
|
t = (heading + ": " if heading else "") + " ".join(buf)
|
|
out.append(Passage(rel, start, t, numbers(t)))
|
|
buf = []
|
|
if ln.startswith("#"):
|
|
heading, table_head = _strip_md(ln.lstrip("#")), ""
|
|
elif is_row:
|
|
if re.fullmatch(r"\s*\|[\s:|-]+\|?\s*", ln):
|
|
continue
|
|
if not table_head:
|
|
table_head = _strip_md(ln)
|
|
continue
|
|
t = f"{heading}: {table_head} || {_strip_md(ln)}"
|
|
out.append(Passage(rel, i, t, numbers(t)))
|
|
elif not ln.strip():
|
|
table_head = ""
|
|
else:
|
|
if not buf:
|
|
start = i
|
|
buf.append(_strip_md(ln))
|
|
return out
|
|
|
|
|
|
def _shared(a: set[str], b: set[str], tol: float = 0.03) -> int:
|
|
"""Numbers of ``a`` found in ``b`` exactly, or within 3 % for values of 10 and more (rounded figures such as
|
|
'about 550 MB' for 545 and 551 MB)."""
|
|
fb = [float(x) for x in b]
|
|
n = 0
|
|
for x in a:
|
|
if x in b:
|
|
n += 1
|
|
continue
|
|
v = float(x)
|
|
if v >= 10 and any(w >= 10 and abs(v - w) <= tol * max(v, w) for w in fb):
|
|
n += 1
|
|
return n
|
|
|
|
|
|
def candidates(claim: Claim, passages: list[Passage], k: int = TOP_K) -> list[Passage]:
|
|
"""Passages sharing the most of the claim's numbers (exactly or approximately), then the most words."""
|
|
words = set(re.findall(r"[a-z]{4,}", claim.text.lower()))
|
|
scored = []
|
|
for p in passages:
|
|
shared = _shared(claim.nums, p.nums)
|
|
if shared:
|
|
overlap = len(words & set(re.findall(r"[a-z]{4,}", p.text.lower())))
|
|
scored.append((shared, overlap, p))
|
|
scored.sort(key=lambda t: (-t[0], -t[1], t[2].path, t[2].line))
|
|
return [p for _, _, p in scored[:k]]
|
|
|
|
|
|
# ------------------------------------------------------------------ judgment
|
|
|
|
|
|
def build_request(claim: Claim, cands: list[Passage]) -> dict[str, object]:
|
|
state = {
|
|
"claim": claim.text,
|
|
"claim_context": claim.context,
|
|
"sections": [{"source": f"{p.path}:{p.line}", "text": p.text[:1500]} for p in cands],
|
|
}
|
|
ask = "How does `sections[{i}].text` relate to `claim` (read `claim` together with `claim_context`)?"
|
|
questions = {f"s{i}": {**RELATION, "instructions": ask.format(i=i)} for i in range(len(cands))}
|
|
return {"model": MODEL, "state": state, "questions": questions}
|
|
|
|
|
|
def call_api(payload: dict[str, object], key: str, retries: int = 5) -> dict[str, object]:
|
|
body = json.dumps(payload).encode()
|
|
digest = hashlib.sha256(body).hexdigest()
|
|
cached = CACHE / f"{digest}.json"
|
|
if cached.exists():
|
|
return json.loads(cached.read_text()) # type: ignore[no-any-return]
|
|
headers = {
|
|
"Authorization": f"Bearer {key}",
|
|
"Content-Type": "application/json",
|
|
"User-Agent": "simurgarray-claims/1",
|
|
}
|
|
req = urllib.request.Request(API_URL, data=body, method="POST", headers=headers)
|
|
for attempt in range(retries):
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=60) as r:
|
|
res = json.loads(r.read())
|
|
CACHE.mkdir(parents=True, exist_ok=True)
|
|
cached.write_text(json.dumps(res))
|
|
return res # type: ignore[no-any-return]
|
|
except urllib.error.HTTPError as e:
|
|
if e.code in (429, 529) and attempt < retries - 1:
|
|
time.sleep(2**attempt)
|
|
continue
|
|
raise RuntimeError(f"TypeSafe API error {e.code}: {e.read()[:200]!r}") from e
|
|
raise RuntimeError("TypeSafe API: retries exhausted")
|
|
|
|
|
|
def verdict(answers: dict[str, dict[str, object]], threshold: float = AUTO_ACCEPT) -> tuple[str, str | None, float]:
|
|
"""Combine per-section answers: any confident 'supports' wins; else a confident 'contradicts'; else a confident
|
|
'says_nothing' from every section; anything else goes to a person."""
|
|
best: dict[str, tuple[float, str]] = {}
|
|
for qid, a in answers.items():
|
|
ch, conf = str(a["choice"]), float(a["confidence"]) # type: ignore[arg-type]
|
|
if conf >= threshold and conf > best.get(ch, (-1.0, ""))[0]:
|
|
best[ch] = (conf, qid)
|
|
if "supports" in best:
|
|
return "verified", best["supports"][1], best["supports"][0]
|
|
support = max((float(a["probabilities"].get("supports", 0.0)) for a in answers.values()), default=0.0) # type: ignore[attr-defined]
|
|
if "contradicts" in best:
|
|
if support >= 0.3: # another section leans towards support: a person decides
|
|
return "review", best["contradicts"][1], best["contradicts"][0]
|
|
return "contradicted", best["contradicts"][1], best["contradicts"][0]
|
|
confident = [float(a["confidence"]) >= threshold and a["choice"] == "says_nothing" for a in answers.values()] # type: ignore[arg-type]
|
|
if confident and all(confident):
|
|
return "unsupported", None, min(float(a["confidence"]) for a in answers.values()) # type: ignore[arg-type]
|
|
return "review", None, max((float(a["confidence"]) for a in answers.values()), default=0.0) # type: ignore[arg-type]
|
|
|
|
|
|
# ------------------------------------------------------------------ report
|
|
|
|
|
|
def _key_from_file(path: Path) -> str:
|
|
"""``TYPESAFE_API_KEY=...`` from a private local file (never from the repository)."""
|
|
if path.exists():
|
|
for ln in path.read_text().splitlines():
|
|
if ln.startswith("TYPESAFE_API_KEY="):
|
|
return ln.split("=", 1)[1].strip()
|
|
return ""
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
|
|
ap.add_argument("files", nargs="+", type=Path, help="texts to check (markdown or plain text)")
|
|
ap.add_argument("--sources", nargs="+", type=Path, default=[ROOT / s for s in DEFAULT_SOURCES])
|
|
ap.add_argument("--out", type=Path, default=Path("claims_report.md"))
|
|
ap.add_argument("--threshold", type=float, default=AUTO_ACCEPT)
|
|
ap.add_argument("--dry-run", action="store_true", help="extract claims and evidence, no API calls")
|
|
a = ap.parse_args(argv)
|
|
key = os.environ.get("TYPESAFE_API_KEY", "") or _key_from_file(Path.home() / ".config" / "typesafe" / "credentials")
|
|
if not a.dry_run and not key:
|
|
print("TYPESAFE_API_KEY is not set (use --dry-run to check extraction without the API).", file=sys.stderr)
|
|
return 2
|
|
passages = load_passages(a.sources)
|
|
rows: list[tuple[str, Claim, Passage | None, float, str]] = []
|
|
for f in a.files:
|
|
for c in extract_claims(f, f.read_text(encoding="utf-8")):
|
|
cands = candidates(c, passages)
|
|
if not cands:
|
|
rows.append(("no_source", c, None, 1.0, ""))
|
|
continue
|
|
if a.dry_run:
|
|
rows.append(("planned", c, cands[0], 0.0, f"{len(cands)} candidate(s)"))
|
|
continue
|
|
res = call_api(build_request(c, cands), key)
|
|
answers = res["answers"] # type: ignore[index]
|
|
v, qid, conf = verdict(answers, a.threshold) # type: ignore[arg-type]
|
|
src = cands[int(qid[1:])] if qid else cands[0]
|
|
probs = ", ".join(f"{q}: {answers[q]['choice']} {answers[q]['confidence']:.2f}" for q in answers) # type: ignore[index]
|
|
rows.append((v, c, src, conf, probs))
|
|
order = ["contradicted", "no_source", "review", "unsupported", "verified", "planned"]
|
|
counts = {v: sum(r[0] == v for r in rows) for v in order}
|
|
lines = ["# Claim check", "", " · ".join(f"{v} {n}" for v, n in counts.items() if n), ""]
|
|
for v in order:
|
|
sel = [r for r in rows if r[0] == v]
|
|
if not sel:
|
|
continue
|
|
lines += [f"## {v} ({len(sel)})", ""]
|
|
for _, c, src, conf, extra in sel:
|
|
lines.append(f"* `{c.path}:{c.line}` {c.text}")
|
|
if src:
|
|
lines.append(f" * evidence `{src.path}:{src.line}` (confidence {conf:.2f}) {src.text[:220]}")
|
|
if extra:
|
|
lines.append(f" * {extra}")
|
|
lines.append("")
|
|
a.out.write_text("\n".join(lines))
|
|
print(f"{len(rows)} claims: " + ", ".join(f"{v} {n}" for v, n in counts.items() if n) + f" → {a.out}")
|
|
return 1 if counts["contradicted"] or counts["no_source"] else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|