File size: 3,983 Bytes
4294de3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 | #!/usr/bin/env python3
"""Which server-side default makes Nex behave well for clients that pass no chat_template_kwargs (the gateway case)?
Follow-up to nex_tools_tpl.py: with the `high` shim, a no-kwargs one-word request came back entirely in
reasoning_content (the model never closed its forced-open think block). Two candidate defaults, greedy probes:
C1 high shim + `--chat-template-kwargs {"enable_thinking": false}` (thinking off unless a client asks)
C2 v2 shim (explicit enable_thinking=false wins; otherwise enable_thinking maps to high unless reasoning_effort is
set) + `--chat-template-kwargs {"reasoning_effort": "medium"}` (upstream's adaptive default)
Diagnostic only -> results/nex_seat_default_probe.json."""
import json, os, subprocess, sys, time
from types import SimpleNamespace
os.environ["AGNES_BIN"] = "/opt/llama-rocm/rocmfpx-724/build-hipvk/bin"
sys.path.insert(0, "/mnt/models/nex-n2.5-mini")
import nex_harness as H # noqa: E402
W = H.W
SRC = open(f"{W}/hf/chat_template.jinja", "rb").read()
SHIM_V2 = ("{%- if enable_thinking is defined and not enable_thinking %}{%- set reasoning_effort = 'none' %}"
"{%- elif reasoning_effort is not defined and enable_thinking is defined %}"
"{%- set reasoning_effort = 'high' %}{%- endif %}\n")
TPL_V2 = f"{W}/tpl/chat_template_enable_thinking_v2.jinja"
open(TPL_V2, "wb").write(SHIM_V2.encode() + SRC)
CONFIGS = [
("C1", [f"--chat-template-file", f"{W}/tpl/chat_template_enable_thinking.jinja",
"--chat-template-kwargs", json.dumps({"enable_thinking": False})]),
("C2", ["--chat-template-file", TPL_V2, "--chat-template-kwargs", json.dumps({"reasoning_effort": "medium"})]),
]
PROMPTS = [("correct-decline", "What is 17 times 23? Answer directly."),
("single-word", "Reply with the single word: ready"),
("multi-arg", "What's the weather in Paris in celsius?")]
KW = [("no-kwargs", None), ("enable_thinking=true", {"enable_thinking": True}),
("enable_thinking=false", {"enable_thinking": False})]
_orig = subprocess.Popen
report = {"shim_v2": SHIM_V2, "configs": {}}
for name, extra in CONFIGS:
def _popen(cmd, *a, _extra=extra, **k):
if cmd and str(cmd[0]).endswith("llama-server"):
cmd = list(cmd) + list(_extra)
return _orig(cmd, *a, **k)
subprocess.Popen = _popen
a = SimpleNamespace(model=f"{W}/out/Nex-N2.5-mini-Q4_0_ROCMFP4_STRIX_LEAN.gguf", dev="ROCm0", ctx=16384,
draft=None, mtp_infile=False, nmax=4, pmin=0.0, strict=False,
serverlog=f"{W}/logs/probe_seat_default_{name}.log")
try:
s = H.Server(a, 18653)
finally:
subprocess.Popen = _orig
rec = {"extra": extra, "results": {}}
try:
for kn, kw in KW:
for pn, prompt in PROMPTS:
body = {"messages": [{"role": "user", "content": prompt}], "tools": H.TOOLS, "tool_choice": "auto",
"temperature": 0, "top_k": 1, "max_tokens": 2048}
if kw is not None:
body["chat_template_kwargs"] = kw
try:
m = H.post(18653, "/v1/chat/completions", body)["choices"][0]["message"]
c = m.get("content") or ""
r = {"content": c[:120], "reasoning_len": len(m.get("reasoning_content") or ""),
"tool_calls": [(t.get("function") or {}).get("name") for t in m.get("tool_calls") or []],
"leaks": [x for x in H.LEAK if x in c]}
except Exception as e: # noqa: BLE001
r = {"error": repr(e)[:300]}
rec["results"][f"{kn}|{pn}"] = r
print(name, kn, pn, json.dumps(r)[:220], flush=True)
finally:
s.stop()
report["configs"][name] = rec
time.sleep(3)
json.dump(report, open(f"{W}/results/nex_seat_default_probe.json", "w"), indent=1)
print("NEX_SEAT_DEFAULT_PROBE_DONE")
|