spb/zyquo-router Public MIT

1#!/usr/bin/env python32#3#  verify.py4#  Zyquo Router5#6#  Author: Simon-Pierre Boucher7#  Mail: contact@spboucher.ai8#9#  Phase 7 verification harness. Drives EVERY chat model in the catalog10#  through the router's local endpoint using the official OpenAI Python SDK11#  (never the upstreams directly): non-streaming, streaming (SDK-parsed12#  chunk discipline), tools / vision / reasoning where the catalog says the13#  model supports them. Emits the compatibility matrix as Markdown.14#15#  Usage:  verify.py [--base http://127.0.0.1:8787/v1] [--providers xai,mistral]16#          [--models id1,id2] [--workers 6] [--out docs/VERIFICATION.md]17#18import argparse19import base6420import concurrent.futures21import json22import sys23import threading24import time25import urllib.request2627from openai import OpenAI, APIError, APIStatusError2829TOOLS = [{30    "type": "function",31    "function": {32        "name": "get_weather",33        "description": "Get current weather for a location",34        "parameters": {35            "type": "object",36            "properties": {"location": {"type": "string"}},37            "required": ["location"],38        },39    },40}]4142# 64×64 solid red PNG (1×1 images are rejected by some providers).43TINY_PNG = "iVBORw0KGgoAAAANSUhEUgAAAEAAAABACAIAAAAlC+aJAAAAb0lEQVR4nO3PAQkAAAyEwO9feoshgnABdLep8QUNyPEFDcjxBQ3I8QUNyPEFDcjxBQ3I8QUNyPEFDcjxBQ3I8QUNyPEFDcjxBQ3I8QUNyPEFDcjxBQ3I8QUNyPEFDcjxBQ3I8QUNyPEFDcjxBQ3IPanc8OLDQitxAAAAAElFTkSuQmCC"4445print_lock = threading.Lock()464748def log(msg):49    with print_lock:50        print(msg, flush=True)515253class ModelResult:54    def __init__(self, model_id, meta):55        self.id = model_id56        self.meta = meta or {}57        self.non_stream = None   # "PASS" / "FAIL: …" / "RATE-LIMITED"58        self.stream = None59        self.tools = "n/a"60        self.vision = "n/a"61        self.reasoning = "n/a"6263    @property64    def ok(self):65        checks = [self.non_stream, self.stream, self.tools, self.vision, self.reasoning]66        return all(c in (None, "n/a") or str(c).startswith(("PASS", "RATE-LIMITED", "SKIP")) for c in checks)676869def classify(err):70    if isinstance(err, APIStatusError) and err.status_code == 429:71        return "RATE-LIMITED"72    text = str(err)73    return "FAIL: " + text[:160].replace("\n", " ").replace("|", "/")747576def check_non_stream(client, result):77    r = client.chat.completions.create(78        model=result.id, max_tokens=32,79        messages=[{"role": "user", "content": "Reply with exactly: OK"}],80    )81    assert r.object == "chat.completion", f"object={r.object}"82    assert r.model == result.id, f"model echo {r.model}"83    assert r.choices[0].message.role == "assistant"84    assert r.choices[0].finish_reason in ("stop", "length"), f"finish={r.choices[0].finish_reason}"85    assert r.usage and r.usage.total_tokens > 0, "usage missing"86    return "PASS"878889def check_stream(client, result):90    text, finish, usage, got_role = "", None, None, False91    stream = client.chat.completions.create(92        model=result.id, max_tokens=64, stream=True,93        stream_options={"include_usage": True},94        messages=[{"role": "user", "content": "Count from 1 to 3, digits only."}],95    )96    first = True97    for chunk in stream:98        assert chunk.object == "chat.completion.chunk", f"chunk object={chunk.object}"99        if chunk.usage:100            usage = chunk.usage101        if not chunk.choices:102            continue103        delta = chunk.choices[0].delta104        if first and delta.role == "assistant":105            got_role = True106        first = False107        if chunk.choices[0].finish_reason:108            finish = chunk.choices[0].finish_reason109        text += delta.content or ""110    assert got_role, "no role delta on first chunk"111    assert finish in ("stop", "length"), f"finish={finish}"112    assert usage is not None and usage.total_tokens > 0, "usage chunk missing"113    return "PASS"114115116def check_tools(client, result, tool_choice="auto"):117    calls = {}118    stream = client.chat.completions.create(119        model=result.id, max_tokens=300, stream=True,120        tools=TOOLS, tool_choice=tool_choice,121        messages=[{"role": "user", "content": "What's the weather in Paris? Use the get_weather tool."}],122    )123    for chunk in stream:124        if not chunk.choices:125            continue126        for tc in chunk.choices[0].delta.tool_calls or []:127            entry = calls.setdefault(tc.index, {"id": None, "name": "", "args": ""})128            if tc.id:129                entry["id"] = tc.id130            if tc.function and tc.function.name:131                entry["name"] = tc.function.name132            if tc.function and tc.function.arguments:133                entry["args"] += tc.function.arguments134    assert calls, "no tool call streamed"135    call = calls[min(calls)]136    assert call["id"], "tool call id missing"137    assert call["name"] == "get_weather", f"name={call['name']}"138    try:139        args = json.loads(call["args"])140        note = ""141    except json.JSONDecodeError:142        # DashScope qwq repeats the complete arguments object per delta —143        # accept the first object but flag the upstream quirk.144        args, _ = json.JSONDecoder().raw_decode(call["args"])145        note = " (upstream repeats args)"146    assert "location" in args, f"args={args}"147    return "PASS" + note148149150def check_vision(client, result):151    r = client.chat.completions.create(152        model=result.id, max_tokens=1024,153        messages=[{"role": "user", "content": [154            {"type": "text", "text": "One word: what color is this image?"},155            {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{TINY_PNG}"}},156        ]}],157    )158    content = r.choices[0].message.content or ""159    assert content.strip(), "empty vision answer"160    return "PASS"161162163def check_reasoning(client, result):164    r = client.chat.completions.create(165        model=result.id, max_tokens=2048,166        extra_body={"reasoning_effort": "low"},167        messages=[{"role": "user", "content": "What is 17*23? Reply with the number only."}],168    )169    message = r.choices[0].message170    reasoning = getattr(message, "reasoning_content", None)171    if reasoning is None and message.model_extra:172        reasoning = message.model_extra.get("reasoning_content")173    details = (r.usage.completion_tokens_details if r.usage else None)174    reasoning_tokens = getattr(details, "reasoning_tokens", None) if details else None175    if (reasoning and reasoning.strip()) or (reasoning_tokens or 0) > 0:176        return "PASS"177    # The call accepted reasoning_effort and answered, but the provider keeps178    # reasoning server-side (OpenAI o-series/gpt-5*, some hosted models).179    assert (r.choices[0].message.content or "").strip(), "empty reasoning answer"180    return "PASS (hidden)"181182183def verify_model(base, entry):184    client = OpenAI(base_url=base, api_key="zyquo-verify", timeout=180, max_retries=0)185    meta = entry.get("x_zyquo") or {}186    result = ModelResult(entry["id"], meta)187188    # Deep-research models run for minutes — beyond any sane harness timeout.189    long_running = "deep-research" in result.id190191    for name, fn, gated in [192        ("non_stream", check_non_stream, True),193        ("stream", check_stream, not long_running),194        ("tools", check_tools, meta.get("tools")),195        ("vision", check_vision, meta.get("vision")),196        ("reasoning", check_reasoning, meta.get("reasoning") and not long_running),197    ]:198        if not gated:199            if long_running and name in ("stream", "reasoning"):200                setattr(result, name, "SKIP (long-running)")201            continue202        try:203            setattr(result, name, fn(client, result))204        except AssertionError as err:205            outcome = f"FAIL: {err}"206            # Weak tool-callers may ignore "auto" — one retry forcing the call.207            if name == "tools" and "no tool call" in str(err):208                try:209                    outcome = check_tools(client, result, tool_choice="required") + " (required)"210                except Exception as retry_err:  # noqa: BLE001211                    outcome = f"FAIL: no tool call with auto; required → {classify(retry_err)[:80]}"212            setattr(result, name, outcome)213        except (APIError, Exception) as err:  # noqa: BLE001 — harness must not die214            setattr(result, name, classify(err))215        time.sleep(0.3)216217    status = "OK " if result.ok else "!! "218    log(f"{status}{result.id}: ns={result.non_stream} st={result.stream} "219        f"tools={result.tools} vision={result.vision} reasoning={result.reasoning}")220    return result221222223def main():224    parser = argparse.ArgumentParser()225    parser.add_argument("--base", default="http://127.0.0.1:8787/v1")226    parser.add_argument("--providers", default="")227    parser.add_argument("--models", default="")228    parser.add_argument("--workers", type=int, default=6)229    parser.add_argument("--out", default="docs/VERIFICATION.md")230    args = parser.parse_args()231232    with urllib.request.urlopen(f"{args.base}/models") as response:233        catalog = json.load(response)["data"]234235    if args.providers:236        wanted = set(args.providers.split(","))237        catalog = [m for m in catalog if m["owned_by"] in wanted]238    if args.models:239        wanted = set(args.models.split(","))240        catalog = [m for m in catalog if m["id"] in wanted]241242    log(f"Verifying {len(catalog)} models via {args.base}")243    results = []244    with concurrent.futures.ThreadPoolExecutor(max_workers=args.workers) as pool:245        futures = [pool.submit(verify_model, args.base, entry) for entry in catalog]246        for future in concurrent.futures.as_completed(futures):247            results.append(future.result())248249    results.sort(key=lambda r: r.id)250    failed = [r for r in results if not r.ok]251252    lines = [253        "# Zyquo Router — Phase 7 Compatibility Matrix",254        "",255        f"Generated {time.strftime('%Y-%m-%d %H:%M')} by `scripts/verify.py` — every request",256        "went through the router's local endpoint using the official OpenAI Python SDK.",257        "",258        f"**{len(results)} models · {len(results) - len(failed)} green · {len(failed)} failing**",259        "",260        "| Model | Non-stream | Stream | Tools | Vision | Reasoning |",261        "|---|---|---|---|---|---|",262    ]263    for r in results:264        lines.append(f"| `{r.id}` | {r.non_stream} | {r.stream} | {r.tools} | {r.vision} | {r.reasoning} |")265    with open(args.out, "w") as handle:266        handle.write("\n".join(lines) + "\n")267    log(f"\n{len(results) - len(failed)}/{len(results)} green → {args.out}")268    sys.exit(1 if failed else 0)269270271if __name__ == "__main__":272    main()273