vh
750675e391
Both are required in compose because they are coupled: MAX_TOKENS is checked before the forward pass, so an oversized call is a clear 422 instead of reaching the cap as a 503. Pre-deploy floor is nvidia-smi Free >= 15,400 MiB on GPU 1 (card peak 9,876 + scriberr 5,496).
260 lines
14 KiB
Python
260 lines
14 KiB
Python
"""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.
|
|
maxone ONE binary question with the state filling MAX_TOKENS: --repeats x 200, one token more is 422
|
|
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, decisions=None):
|
|
"""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 = decisions or [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_maxone(url, h, base_state, max_tokens, repeats):
|
|
"""The other extreme of the largest call: ONE binary question, the state filling the rest of MAX_TOKENS."""
|
|
one = [{"id": "only", "question": "Does the policy allow the exception described?", "options": YESNO}]
|
|
state, tokens = find_max_state(url, h, base_state, max_tokens, decisions=one)
|
|
over = post(url + "/decide/shared", {"state": state + " extra words", "decisions": one}, h)
|
|
runs = []
|
|
for _ in range(repeats):
|
|
s, j, ms = post(url + "/decide", {"id": "only", "state": state, **{k: one[0][k] for k in ("question", "options")}}, h)
|
|
runs.append({"status": s, "e2e_ms": round(ms, 1), "input_tokens": j.get("input_tokens"),
|
|
"code": (j.get("error") or {}).get("code"), "t_end": time.time()})
|
|
return {"state_chars": len(state), "tokens": tokens, "one_more_is": [over[0], (over[1].get("error") or {}).get("message")],
|
|
"runs": runs, "pass": all(r["status"] == 200 for r in runs) and over[0] == 422}
|
|
|
|
|
|
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),
|
|
"maxone": lambda: check_maxone(url, h, long_state, health["max_tokens"], args.repeats),
|
|
"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()
|