diff --git a/tools/probe_models.py b/tools/probe_models.py new file mode 100644 index 0000000..c5b34c7 --- /dev/null +++ b/tools/probe_models.py @@ -0,0 +1,409 @@ +#!/usr/bin/env python3 +"""I-01 模型能力自动探测:每晚自动 + 异常告警方向,最小闭环。 + +读 ~/.dsh/settings.yaml 枚举全部分组与模型,对每个模型跑三组探测: +- 基线:一次最小对话请求(200 + 非空回复) +- 思考:按模型的 thinkingFormat/reasoningEfforts 配置发带推理参数请求,验证 reasoning 字段 +- 视觉:内置 64px 纯色 PNG(base64 内嵌)问主色 + +铁律:脚本只读 settings.yaml(open(mode='r'),代码层面无写入路径); +任何写入需求只进报告建议,由人执行。报告不打印 API key。 + +产出:tools/probe-reports/probe-YYYY-MM-DD.json(机读)+ 同名 .md(人读:表格 + 与上次 diff)。 + +用法: + python tools/probe_models.py --all # 全量探测 + 写报告 + python tools/probe_models.py --all --compare # 全量探测 + 写报告 + 打印与上次对比 + python tools/probe_models.py --provider b # 只跑一组 + python tools/probe_models.py --check-readonly # 自证:grep 本源码无 settings 写入路径 + +Windows 计划任务(每晚 02:00)示例: + schtasks /create /tn "DSH模型探测" /tr "python C:\\Users\\12914\\Desktop\\dsh\\tools\\probe_models.py --all" /sc daily /st 02:00 +""" +from __future__ import annotations + +import argparse +import base64 +import datetime as _dt +import io +import json +import os +import struct +import sys +import time +import urllib.request +import zlib +from pathlib import Path + +HERE = Path(__file__).resolve().parent +REPORT_DIR = HERE / "probe-reports" +SETTINGS = Path(os.path.expanduser("~/.dsh/settings.yaml")) + +TIMEOUT = 30 # 单请求超时(秒),失败隔离:单模型失败不中断全量 +RED_PNG_B64: str | None = None + + +def red_png_b64() -> str: + """64px 纯红 PNG,纯标准库生成(zlib+struct),无外部文件依赖。""" + global RED_PNG_B64 + if RED_PNG_B64 is None: + w = h = 64 + raw = b"".join(b"\x00" + b"\xff\x00\x00" * w for _ in range(h)) + + def chunk(typ: bytes, data: bytes) -> bytes: + c = typ + data + return struct.pack(">I", len(data)) + c + struct.pack(">I", zlib.crc32(c) & 0xFFFFFFFF) + + png = (b"\x89PNG\r\n\x1a\n" + + chunk(b"IHDR", struct.pack(">IIBBBBB", w, h, 8, 2, 0, 0, 0)) + + chunk(b"IDAT", zlib.compress(raw)) + + chunk(b"IEND", b"")) + RED_PNG_B64 = base64.b64encode(png).decode() + return RED_PNG_B64 + + +def load_settings() -> dict: + """只读加载 settings.yaml(本函数是全脚本唯一的 settings 打开点,mode='r')。""" + try: + import yaml # type: ignore + with open(SETTINGS, "r", encoding="utf-8") as fh: + return yaml.safe_load(fh) or {} + except ImportError: + pass + # 无 PyYAML 时的最小缩进解析:只取 providers 下 api/apiKeyEnv/baseURL/models 必要字段 + import re + text = None + with open(SETTINGS, "r", encoding="utf-8") as fh: + text = fh.read() + providers: dict = {} + cur_p = cur_m = None + cur: dict = {} + for line in text.splitlines(): + m = re.match(r"^ ([a-z0-9-]+):\s*$", line) + if m: + if cur_p and cur: + providers[cur_p] = cur + cur_p, cur, cur_m = m.group(1), {"models": []}, None + continue + if cur_p is None: + continue + m2 = re.match(r"^ (displayName|apiKeyEnv|api|baseURL):\s*(.+?)\s*$", line) + if m2: + cur[m2.group(1)] = m2.group(2) + continue + m3 = re.match(r"^ - id:\s*(\S+)", line) + if m3: + cur_m = {"id": m3.group(1), "input": ["text"], "compat": {}, "reasoningEfforts": None} + cur["models"].append(cur_m) + continue + if cur_m is not None: + m4 = re.match(r"^ (thinkingFormat|supportsDeveloperRole):\s*(\S+)", line) + if m4: + cur_m["compat"][m4.group(1)] = m4.group(2) + continue + m5 = re.match(r"^ input:\s*\[(.+)\]", line) + if m5: + cur_m["input"] = [x.strip() for x in m5.group(1).split(",")] + continue + if re.match(r"^ reasoningEfforts:\s*$", line): + cur_m["reasoningEfforts"] = {} + continue + m6 = re.match(r"^ (off|low|medium|high):\s*(\S+)?", line) + if m6 and isinstance(cur_m.get("reasoningEfforts"), dict): + cur_m["reasoningEfforts"][m6.group(1)] = (m6.group(2) or "").lower() == "null" and None or m6.group(2) + if cur_p and cur: + providers[cur_p] = cur + return {"llm-pi-ai": {"providers": providers}} + + +def iter_models(cfg: dict): + provs = ((cfg.get("llm-pi-ai") or {}).get("providers") or {}) + for pname, p in provs.items(): + for m in p.get("models") or []: + yield pname, p, m + + +def _post(url: str, key: str, payload: dict, timeout: int = TIMEOUT) -> tuple[int, dict, str]: + body = json.dumps(payload, ensure_ascii=False).encode() + req = urllib.request.Request( + url, data=body, + headers={"Content-Type": "application/json", "Authorization": f"Bearer {key}"}, + method="POST", + ) + t0 = time.time() + try: + with urllib.request.urlopen(req, timeout=timeout) as resp: + raw = resp.read().decode("utf-8", "replace") + return resp.status, json.loads(raw), f"{(time.time() - t0) * 1000:.0f}ms" + except Exception as exc: + ms = f"{(time.time() - t0) * 1000:.0f}ms" + code = getattr(exc, "code", "") + return -1, {"_error": f"{type(exc).__name__}: {str(exc)[:200]}", "_code": code}, ms + + +def probe_baseline(base: str, key: str, model: str, api: str): + if api == "openai-responses": + payload = {"model": model, "input": "只回复:OK"} + st, data, ms = _post(base.rstrip("/") + "/responses", key, payload) + text = "" + try: + out = data.get("output") or [] + for item in out: + for c in item.get("content") or []: + text += c.get("text", "") + except Exception: + pass + ok = st == 200 and bool(text.strip()) + return {"pass": ok, "ms": ms, "status": st, + "err": "" if ok else str(data.get("_error", data))[:200]} + payload = {"model": model, "messages": [{"role": "user", "content": "只回复:OK"}], + "max_tokens": 16} + st, data, ms = _post(base.rstrip("/") + "/chat/completions", key, payload) + text = "" + try: + text = (data.get("choices") or [{}])[0].get("message", {}).get("content", "") + except Exception: + pass + ok = st == 200 and bool((text or "").strip()) + return {"pass": ok, "ms": ms, "status": st, + "err": "" if ok else str(data.get("_error", data))[:200]} + + +def probe_thinking(base: str, key: str, model: str, api: str, compat: dict, efforts: dict | None): + """按 thinkingFormat 发推理参数;验证 reasoning 字段返回。无声明 → skip。""" + fmt = (compat or {}).get("thinkingFormat") + if not fmt or not efforts: + return {"pass": None, "ms": "-", "status": "-", "err": "skip: 无 thinking 声明"} + hi = (efforts or {}).get("high") + extra: dict = {} + if fmt == "zai": + extra = {"enable_thinking": True, **({"reasoning_effort": hi} if hi else {})} + elif fmt == "qwen": + extra = {"enable_thinking": True, **({"reasoning_effort": hi or "xhigh"} if True else {})} + else: + extra = {"enable_thinking": True} + if api == "openai-responses": + payload = {"model": model, "input": "5-2=3,还剩3;2*3-3=3,新买3;3+3=? 只给出数字答案。", **extra} + st, data, ms = _post(base.rstrip("/") + "/responses", key, payload) + raw = json.dumps(data, ensure_ascii=False) + has_reason = "reasoning" in raw + text = "" + try: + for item in data.get("output") or []: + for c in item.get("content") or []: + text += c.get("text", "") + except Exception: + pass + ok = st == 200 and has_reason and ("6" in text) + return {"pass": ok, "ms": ms, "status": st, + "err": "" if ok else f"reasoning={'有' if has_reason else '无'} 答={'6' in text} " + str(data.get("_error", ""))[:120]} + payload = {"model": model, + "messages": [{"role": "user", "content": "5-2=3,还剩3;2*3-3=3,新买3;3+3=? 只给出数字答案。"}], + "max_tokens": 256, **extra} + st, data, ms = _post(base.rstrip("/") + "/chat/completions", key, payload) + raw = json.dumps(data, ensure_ascii=False) + has_reason = "reasoning" in raw + text = "" + try: + text = (data.get("choices") or [{}])[0].get("message", {}).get("content", "") + except Exception: + pass + ok = st == 200 and has_reason and ("6" in text) + return {"pass": ok, "ms": ms, "status": st, + "err": "" if ok else f"reasoning={'有' if has_reason else '无'} 答={'6' in text} " + str(data.get("_error", ""))[:120]} + + +def probe_vision(base: str, key: str, model: str, api: str, inputs: list): + """视觉:内嵌 64px 纯红 PNG 问主色。未声明 image → skip。""" + if "image" not in (inputs or []): + return {"pass": None, "ms": "-", "status": "-", "err": "skip: 未声明 image 输入"} + img = {"type": "image_url", "image_url": {"url": "data:image/png;base64," + red_png_b64()}} + if api == "openai-responses": + payload = {"model": model, "input": [{"role": "user", "content": [ + {"type": "input_text", "text": "这张纯色图的主色是什么?只用一个词回答。"}, + {"type": "input_image", "image_url": "data:image/png;base64," + red_png_b64()}, + ]}]} + st, data, ms = _post(base.rstrip("/") + "/responses", key, payload) + text = json.dumps(data, ensure_ascii=False) + ok = st == 200 and ("红" in text or "red" in text.lower()) + return {"pass": ok, "ms": ms, "status": st, + "err": "" if ok else str(data.get("_error", text[:200]))[:200]} + payload = {"model": model, "max_tokens": 32, "messages": [ + {"role": "user", "content": [ + {"type": "text", "text": "这张纯色图的主色是什么?只用一个词回答。"}, + img, + ]}]} + st, data, ms = _post(base.rstrip("/") + "/chat/completions", key, payload) + text = "" + try: + text = (data.get("choices") or [{}])[0].get("message", {}).get("content", "") + except Exception: + pass + ok = st == 200 and ("红" in (text or "") or "red" in (text or "").lower()) + return {"pass": ok, "ms": ms, "status": st, + "err": "" if ok else (str(data.get("_error", text))[:200])} + + +def run(provider_filter: str | None = None): + cfg = load_settings() + rows = [] + for pname, p, m in iter_models(cfg): + if provider_filter and pname != provider_filter: + continue + mid = m.get("id", "?") + api = p.get("api", "openai-completions") + base = p.get("baseURL", "") + key = os.environ.get(p.get("apiKeyEnv", ""), "") + if not key: + rows.append({"provider": pname, "model": mid, "api": api, + "baseline": {"pass": None, "ms": "-", "status": "-", "err": "skip: 无 key(未设 " + p.get("apiKeyEnv", "?") + ")"}, + "thinking": {"pass": None, "ms": "-", "status": "-", "err": "skip: 无 key"}, + "vision": {"pass": None, "ms": "-", "status": "-", "err": "skip: 无 key"}}) + continue + compat = m.get("compat") or {} + efforts = m.get("reasoningEfforts") + inputs = m.get("input") or ["text"] + try: + b = probe_baseline(base, key, mid, api) + except Exception as exc: + b = {"pass": False, "ms": "-", "status": -1, "err": f"{type(exc).__name__}: {exc}"[:200]} + try: + t = probe_thinking(base, key, mid, api, compat, efforts) + except Exception as exc: + t = {"pass": False, "ms": "-", "status": -1, "err": f"{type(exc).__name__}: {exc}"[:200]} + try: + v = probe_vision(base, key, mid, api, inputs) + except Exception as exc: + v = {"pass": False, "ms": "-", "status": -1, "err": f"{type(exc).__name__}: {exc}"[:200]} + rows.append({"provider": pname, "model": mid, "api": api, + "baseline": b, "thinking": t, "vision": v}) + return rows + + +def verdict(row: dict) -> str: + vals = [row[k].get("pass") for k in ("baseline", "thinking", "vision")] + if all(v is True or v is None for v in vals) and any(v is True for v in vals): + skips = sum(1 for v in vals if v is None) + return "pass" if skips == 0 else "pass-skip" + if any(v is False for v in vals): + return "fail" + return "skip" + + +def write_reports(rows: list) -> tuple[Path, Path]: + REPORT_DIR.mkdir(parents=True, exist_ok=True) + day = _dt.date.today().isoformat() + jpath = REPORT_DIR / f"probe-{day}.json" + mpath = REPORT_DIR / f"probe-{day}.md" + payload = {"date": day, "total": len(rows), + "pass": 0, "fail": 0, "skip": 0, + "rows": [{**r, "verdict": verdict(r)} for r in rows]} + for r in payload["rows"]: + if r["verdict"].startswith("pass"): + payload["pass"] += 1 + elif r["verdict"] == "fail": + payload["fail"] += 1 + else: + payload["skip"] += 1 + # 与上次对比(能力变化告警方向) + prev = sorted(REPORT_DIR.glob("probe-*.json")) + prev = [p for p in prev if p.name != jpath.name] + diff_lines: list[str] = [] + if prev: + last = json.loads(prev[-1].read_text(encoding="utf-8")) + by_key = {(r["provider"], r["model"]): r.get("verdict") for r in last.get("rows", [])} + for r in payload["rows"]: + old = by_key.get((r["provider"], r["model"])) + if old and old != r["verdict"]: + diff_lines.append(f'- {r["provider"]}/{r["model"]}: {old} → **{r["verdict"]}**') + payload["diff_from"] = prev[-1].name if prev else None + payload["diff"] = diff_lines + jpath.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") + + alert = "⚠️ 能力变化告警" if diff_lines else "无能力变化" + if payload["fail"]: + alert += f'|{payload["fail"]} 个模型失败' + lines = [f"# 模型探测报告 {day}", "", f"> {alert}", "", + f"总 {payload['total']} · 通过 {payload['pass']} · 失败 {payload['fail']} · 全跳过 {payload['skip']}", + "", "| 分组 | 模型 | 基线 | 思考 | 视觉 | 结论 |", + "|---|---|---|---|---|---|"] + sym = {True: "✅", False: "❌", None: "⏭️"} + for r in payload["rows"]: + b, t, v = r["baseline"], r["thinking"], r["vision"] + lines.append(f"| {r['provider']} | {r['model']} | {sym[b['pass']]}{b['ms']} " + f"| {sym[t['pass']]}{t['ms']} | {sym[v['pass']]}{v['ms']} | {r['verdict']} |") + fails = [r for r in payload["rows"] if r["verdict"] == "fail"] + if fails: + lines += ["", "## 失败明细(错误摘要,不含 key)"] + for r in fails: + for k in ("baseline", "thinking", "vision"): + if r[k].get("pass") is False: + lines.append(f"- {r['provider']}/{r['model']} {k}: [{r[k].get('status')}] {r[k].get('err', '')[:150]}") + if diff_lines: + lines += ["", f"## 与上次对比({payload['diff_from']})"] + diff_lines + else: + lines += ["", "## 与上次对比", "无变化" if prev else "(首次报告,无基线)"] + lines += ["", "## 配置建议(需人执行,本脚本不写配置)", + "失败/变化的模型请对照 2026-09-06 手动结论复核后再改 settings.yaml。"] + mpath.write_text("\n".join(lines) + "\n", encoding="utf-8") + return jpath, mpath + + +def check_readonly() -> int: + """自证无 settings 写入路径:源码中不得出现写模式 open/settings 写入调用。""" + src = Path(__file__).read_text(encoding="utf-8") + bad = [] + in_checker = False + for i, line in enumerate(src.splitlines(), 1): + s = line.strip() + if s.startswith("def check_readonly"): + in_checker = True + elif in_checker and s.startswith("def "): + in_checker = False + if in_checker: + continue # 跳过检查器自身的字面量(否则自指误报) + if s.startswith("#") or s.startswith('"""') or s.startswith("'''"): + continue + if "SETTINGS" in line and ("\"w\"" in line or "'w'" in line or "\"a\"" in line or "'a'" in line): + bad.append((i, line.strip())) + if "settings.yaml" in line.lower() and (".write" in line or "yaml.dump" in line or "safe_dump" in line): + bad.append((i, line.strip())) + if bad: + print("发现疑似写入路径:") + for i, l in bad: + print(f" L{i}: {l}") + return 1 + opens = [l.strip() for l in src.splitlines() if "open(SETTINGS" in l] + print("settings 打开点(应仅 mode='r'):") + for l in opens: + print(f" {l}") + print("readonly-check: PASS(无写入路径)") + return 0 + + +def main(argv: list[str]) -> int: + ap = argparse.ArgumentParser(description="DSH 模型能力自动探测(只报告不写配置)") + ap.add_argument("--all", action="store_true", help="全量探测") + ap.add_argument("--provider", default="", help="只跑指定分组") + ap.add_argument("--compare", action="store_true", help="打印与上次报告的对比") + ap.add_argument("--check-readonly", action="store_true", help="自证无 settings 写入路径") + args = ap.parse_args(argv) + if args.check_readonly: + return check_readonly() + if not args.all and not args.provider: + ap.print_help() + return 2 + rows = run(args.provider or None) + jp, mp = write_reports(rows) + print(f"报告:{jp}\n报告:{mp}") + if args.compare: + data = json.loads(jp.read_text(encoding="utf-8")) + print(f"对比基线:{data.get('diff_from') or '无(首次)'}") + for d in data.get("diff", []) or ["无变化"]: + print(f" {d}") + fails = sum(1 for r in rows if verdict(r) == "fail") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main(sys.argv[1:]))