feat(intern-decision): stack, DNS and GPU 3 acceptance for the SemIf replacement
stacks/intern-decision: compose (GPU 1, :8033, hard VRAM cap as the single .env knob, healthcheck, Homepage group 'AI - Eval & Retrieval'), .env.example and README. dns: intern-decision.fv.internal -> fv-ml1 (synced to ana/esh/nh3). acceptance on fv-ml1 GPU 3, 3 fresh processes: bit-identical to the Jev bench's native rows (pooled 240/259, Wyrd 79/84, 0/560 flips, Δp 0), negative control 10/122/14, 0 flips across restarts; largest accepted request 200 at a 10,134 MiB card peak under a 9.25 GiB cap; 503 and recovery proven at a tight cap. GPU 1 deploy held: nvidia-smi Free on GPU 1 is 15,442 MiB.
This commit is contained in:
@@ -0,0 +1,243 @@
|
||||
"""intern-decision-serve acceptance: the service-level checks the bench harness does not cover.
|
||||
Stdlib only, so it runs on fv-ml1's host python or on nh3-dev against a live instance.
|
||||
|
||||
python3 checks.py --url URL --token-file T --out out.json --checks auth,chunking,queue,maxreq
|
||||
python3 checks.py --url URL --token-file T --out oom.json --checks oom
|
||||
|
||||
auth 401 without / with a wrong token on both POSTs; /health open
|
||||
chunking 20 decisions in one /decide/shared equal the same decisions asked as 16 + 4
|
||||
queue past MAX_QUEUE requests in progress, the rest get 429 busy (and the admitted all answer)
|
||||
maxreq the largest request the API accepts: MAX_DECISIONS decisions x 16 options, the state
|
||||
grown until each 16-question call sits just under MAX_TOKENS (found from the model's own
|
||||
422 message). Sent --repeats times; status, tokens and /health's allocator peak recorded.
|
||||
fit under the cap, the largest 16-question call that answers 200 (binary search, in tokens)
|
||||
oom the over-cap request (maxreq's body when it ran) -> 503 out_of_memory, then reserved memory
|
||||
back to its resting value and an ordinary request answered 200. Earlier text: a long request against a deliberately tight cap -> 503 out_of_memory, then reserved
|
||||
memory back to its resting value and an ordinary request answered 200.
|
||||
Every request's status, timing and the fields that prove the point are kept in --out.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import re
|
||||
import threading
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from pathlib import Path
|
||||
|
||||
HERE = Path(__file__).resolve().parent
|
||||
YESNO = [{"id": "yes", "description": "Yes"}, {"id": "no", "description": "No"}]
|
||||
|
||||
|
||||
def post(url, body, headers, timeout=600):
|
||||
req = urllib.request.Request(url, data=json.dumps(body).encode(),
|
||||
headers={"Content-Type": "application/json", **headers})
|
||||
t = time.perf_counter()
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=timeout) as r:
|
||||
status, raw = r.status, r.read()
|
||||
except urllib.error.HTTPError as e:
|
||||
status, raw = e.code, e.read()
|
||||
ms = (time.perf_counter() - t) * 1000
|
||||
try:
|
||||
return status, json.loads(raw), ms
|
||||
except ValueError:
|
||||
return status, {"raw": raw[:300].decode(errors="replace")}, ms
|
||||
|
||||
|
||||
def get(url):
|
||||
with urllib.request.urlopen(url, timeout=30) as r:
|
||||
return r.status, json.loads(r.read())
|
||||
|
||||
|
||||
def criteria(n, options=YESNO, prefix="c"):
|
||||
return [{"id": f"{prefix}{i}", "question": f"Does the evidence mention item number {i}?", "options": options}
|
||||
for i in range(n)]
|
||||
|
||||
|
||||
def check_auth(url, h):
|
||||
out = {}
|
||||
body = {"id": "a", "state": "s", "question": "q?", "options": YESNO}
|
||||
for path, b in (("/decide", body), ("/decide/shared", {"state": "s", "decisions": [body]})):
|
||||
out[path] = {"no_token": post(url + path, b, {})[0],
|
||||
"wrong_token": post(url + path, b, {"Authorization": "Bearer " + "x" * 48})[0],
|
||||
"right_token": post(url + path, b, h)[0]}
|
||||
out["health_no_token"] = get(url + "/health")[0]
|
||||
out["pass"] = all(v["no_token"] == 401 and v["wrong_token"] == 401 and v["right_token"] == 200
|
||||
for k, v in out.items() if k.startswith("/")) and out["health_no_token"] == 200
|
||||
return out
|
||||
|
||||
|
||||
def check_chunking(url, h, state):
|
||||
crits = [{"id": f"d{i}", "question": q, "options": opts} for i, (q, opts) in enumerate([
|
||||
(f"Is item {i} a {kind}?", [{"id": "yes", "description": "Yes"}, {"id": "no", "description": "No"},
|
||||
{"id": "unclear", "description": "Cannot tell from the evidence"}])
|
||||
for i, kind in enumerate(["policy", "exception", "deadline", "person", "number"] * 4)])]
|
||||
s_all, j_all, ms_all = post(url + "/decide/shared", {"state": state, "decisions": crits}, h)
|
||||
s_a, j_a, _ = post(url + "/decide/shared", {"state": state, "decisions": crits[:16]}, h)
|
||||
s_b, j_b, _ = post(url + "/decide/shared", {"state": state, "decisions": crits[16:]}, h)
|
||||
split = (j_a.get("results", []) + j_b.get("results", [])) if s_a == s_b == 200 else []
|
||||
rows = []
|
||||
for whole, part in zip(j_all.get("results", []), split):
|
||||
rows.append({"id": whole["id"], "top_whole": whole["top"], "top_split": part["top"],
|
||||
"max_dp": max(abs(x - y) for x, y in zip(whole["probabilities"], part["probabilities"])),
|
||||
"call_whole": whole["call"], "sha_equal": whole["prompt_sha256"] == part["prompt_sha256"]})
|
||||
t = j_all.get("timing", {})
|
||||
return {"status": [s_all, s_a, s_b], "timing": t, "rows": rows, "e2e_ms": round(ms_all, 1),
|
||||
"pass": s_all == s_a == s_b == 200 and t.get("calls") == 2 and t.get("questions_per_call") == [16, 4]
|
||||
and len(rows) == 20 and all(r["top_whole"] == r["top_split"] and r["max_dp"] == 0 and r["sha_equal"]
|
||||
for r in rows)}
|
||||
|
||||
|
||||
def check_queue(url, h, state, max_queue, extra):
|
||||
n = max_queue + extra
|
||||
body = {"state": state, "decisions": criteria(16)}
|
||||
barrier = threading.Barrier(n)
|
||||
results = [None] * n
|
||||
|
||||
def one(i):
|
||||
barrier.wait()
|
||||
s, j, ms = post(url + "/decide/shared", body, h)
|
||||
results[i] = (s, (j.get("error") or {}).get("code"), round(ms, 1))
|
||||
|
||||
threads = [threading.Thread(target=one, args=(i,)) for i in range(n)]
|
||||
for t in threads:
|
||||
t.start()
|
||||
for t in threads:
|
||||
t.join()
|
||||
statuses = [r[0] for r in results]
|
||||
return {"sent": n, "max_queue": max_queue, "ok": statuses.count(200), "busy_429": statuses.count(429),
|
||||
"codes_429": sorted({r[1] for r in results if r[0] == 429}), "other": [r for r in results if r[0] not in (200, 429)],
|
||||
"pass": statuses.count(429) >= 1 and statuses.count(200) + statuses.count(429) == n
|
||||
and statuses.count(200) <= max_queue}
|
||||
|
||||
|
||||
def question_block(i):
|
||||
"""Zero-padded numbers: Qwen tokenises digits one by one, so every 16-question call of a
|
||||
64-decision request has the same token count and each one sits at the limit."""
|
||||
return {"id": f"q{i:02d}", "question": f"Which of the listed categories best describes clause {i:02d} of the policy?",
|
||||
"options": [{"id": f"k{j:02d}", "description": f"Category {j:02d}: the clause concerns topic number {j:02d}"}
|
||||
for j in range(16)]}
|
||||
|
||||
|
||||
def find_max_state(url, h, base_state, max_tokens, n_decisions=16):
|
||||
"""Grow the state (repeating base_state) until the request is refused with 422, using the token
|
||||
count the model's own 422 reports; return the longest state the API accepts and its tokens per call.
|
||||
Sized with the WHOLE n_decisions request, so the largest call in it is the one that decides."""
|
||||
decisions = [question_block(i) for i in range(n_decisions)]
|
||||
text = (base_state + "\n\n") * 4
|
||||
|
||||
def tokens_for(chars):
|
||||
s, j, _ = post(url + "/decide/shared", {"state": text[:chars], "decisions": decisions}, h)
|
||||
if s == 200:
|
||||
return s, j["timing"]["input_tokens"][0]
|
||||
if s == 503:
|
||||
return s, None
|
||||
m = re.search(r"has (\d+) tokens", (j.get("error") or {}).get("message", ""))
|
||||
return s, int(m.group(1)) if m else None
|
||||
|
||||
lo, hi = 1000, len(text)
|
||||
s, tok = tokens_for(hi)
|
||||
assert s == 422 and tok and tok > max_tokens, (s, tok)
|
||||
best = None
|
||||
while lo < hi:
|
||||
mid = (lo + hi + 1) // 2
|
||||
s, tok = tokens_for(mid)
|
||||
if s in (200, 503): # accepted by the API (503 = accepted, then refused by the VRAM cap)
|
||||
lo, best = mid, tok
|
||||
elif s == 422:
|
||||
hi = mid - 1
|
||||
else:
|
||||
raise RuntimeError(f"unexpected {s} while sizing the max request: {tok}")
|
||||
return text[:lo], best
|
||||
|
||||
|
||||
MAX_BODY = {}
|
||||
|
||||
|
||||
def check_maxreq(url, h, base_state, max_tokens, max_decisions, repeats):
|
||||
state, tokens = find_max_state(url, h, base_state, max_tokens, max_decisions)
|
||||
decisions = [question_block(i) for i in range(max_decisions)]
|
||||
body = {"state": state, "decisions": decisions}
|
||||
MAX_BODY.update(body)
|
||||
size = len(json.dumps(body).encode())
|
||||
runs = []
|
||||
for _ in range(repeats):
|
||||
s, j, ms = post(url + "/decide/shared", body, h)
|
||||
health = get(url + "/health")[1]["model"]
|
||||
runs.append({"status": s, "e2e_ms": round(ms, 1), "code": (j.get("error") or {}).get("code"),
|
||||
"calls": (j.get("timing") or {}).get("calls"), "input_tokens": (j.get("timing") or {}).get("input_tokens"),
|
||||
"reserved_gib_after": health.get("reserved_gib"), "max_reserved_gib": health.get("max_reserved_gib"),
|
||||
"t_end": time.time()})
|
||||
return {"state_chars": len(state), "tokens_per_call": tokens, "decisions": max_decisions, "options_each": 16,
|
||||
"body_bytes": size, "runs": runs}
|
||||
|
||||
|
||||
def check_fit(url, h, base_state):
|
||||
"""Under the cap: the largest 16-question call (16 options each) that still answers 200, in tokens."""
|
||||
decisions = [question_block(i) for i in range(16)]
|
||||
text = (base_state + "\n\n") * 4
|
||||
lo, hi, best, probes = 1000, len(text), None, []
|
||||
while lo < hi:
|
||||
mid = (lo + hi + 1) // 2
|
||||
s, j, _ = post(url + "/decide/shared", {"state": text[:mid], "decisions": decisions}, h)
|
||||
probes.append([mid, s, j["timing"]["input_tokens"][0] if s == 200 else None])
|
||||
if s == 200:
|
||||
lo, best = mid, j["timing"]["input_tokens"][0]
|
||||
else:
|
||||
hi = mid - 1
|
||||
return {"largest_ok_state_chars": lo, "largest_ok_tokens": best, "probes": probes}
|
||||
|
||||
|
||||
def check_oom(url, h, long_state):
|
||||
"""The over-cap request: the largest one (maxreq) if it ran, else 16 criteria over the long state."""
|
||||
before = get(url + "/health")[1]["model"]
|
||||
body = dict(MAX_BODY) or {"state": long_state, "decisions": criteria(16)}
|
||||
s, j, ms = post(url + "/decide/shared", body, h)
|
||||
after = get(url + "/health")[1]["model"]
|
||||
s2, j2, _ = post(url + "/decide", {"id": "after", "state": "The deploy passed.", "question": "Did it pass?",
|
||||
"options": YESNO}, h)
|
||||
again = get(url + "/health")[1]["model"]
|
||||
return {"oom_status": s, "oom_error": j.get("error"), "oom_ms": round(ms, 1),
|
||||
"reserved_gib": {"before": before.get("reserved_gib"), "after_oom": after.get("reserved_gib"),
|
||||
"after_next_request": again.get("reserved_gib")},
|
||||
"next_request_status": s2, "next_top": j2.get("top"),
|
||||
"pass": s == 503 and (j.get("error") or {}).get("code") == "out_of_memory" and s2 == 200
|
||||
and after.get("reserved_gib") is not None and after["reserved_gib"] <= before["reserved_gib"] + 0.5}
|
||||
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument("--url", required=True)
|
||||
ap.add_argument("--token-file", required=True)
|
||||
ap.add_argument("--out", required=True)
|
||||
ap.add_argument("--checks", default="auth,chunking,queue,maxreq")
|
||||
ap.add_argument("--long-state-file", default=str(HERE / "long_state.txt"))
|
||||
ap.add_argument("--max-queue", type=int, default=32)
|
||||
ap.add_argument("--queue-extra", type=int, default=16)
|
||||
ap.add_argument("--repeats", type=int, default=3)
|
||||
args = ap.parse_args()
|
||||
url = args.url.rstrip("/")
|
||||
h = {"Authorization": "Bearer " + Path(args.token_file).read_text().strip()}
|
||||
long_state = Path(args.long_state_file).read_text()
|
||||
health = get(url + "/health")[1]
|
||||
report = {"url": url, "started_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "health": health}
|
||||
for c in args.checks.split(","):
|
||||
t = time.time()
|
||||
report[c] = {"auth": lambda: check_auth(url, h),
|
||||
"chunking": lambda: check_chunking(url, h, long_state[:4000]),
|
||||
"queue": lambda: check_queue(url, h, long_state, args.max_queue, args.queue_extra),
|
||||
"maxreq": lambda: check_maxreq(url, h, long_state, health["max_tokens"], health["max_decisions"],
|
||||
args.repeats),
|
||||
"fit": lambda: check_fit(url, h, long_state),
|
||||
"oom": lambda: check_oom(url, h, long_state)}[c]()
|
||||
report[c]["t_start"], report[c]["t_end"] = t, time.time()
|
||||
print(c, json.dumps({k: v for k, v in report[c].items() if k not in ("rows", "runs")})[:600], flush=True)
|
||||
report["finished_utc"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
|
||||
Path(args.out).write_text(json.dumps(report, indent=1))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user