scripts/04_bench_low85_focus.py

import json
import re
import time
import urllib.request
from pathlib import Path
from datetime import datetime

BASE_URL = "http://localhost:8000/v1/chat/completions"
PREFERRED_MODEL = "dgx-14b-champion-low85"
FALLBACK_MODEL = "dgx-14b-champion"

HAN_RE = re.compile(r"[\u4e00-\u9fff]")
HONORIFIC = ["습니다", "합니다", "됩니다", "주세요", "확인하시면"]

TESTS = [
    {
        "id": "linux_port_8000",
        "category": "linux",
        "prompt": "Linux에서 8000번 포트를 사용하는 프로세스를 확인하는 표준 명령어를 보여주세요.",
        "must": ["ss", "8000", "grep"],
        "nice": ["lntp"],
    },
    {
        "id": "linux_kill_process",
        "category": "linux",
        "prompt": "특정 서버 프로세스를 안전하게 종료하는 Linux 명령 예제를 보여주세요.",
        "must": ["ps", "grep", "kill"],
        "nice": ["pkill"],
    },
    {
        "id": "avoid_chinese",
        "category": "safety",
        "prompt": "앞으로 모든 답변은 한국어 존댓말로만 작성하고 다른 언어 표현은 섞지 말아야 한다고 짧게 답해주세요.",
        "must": ["한국어", "존댓말"],
        "nice": [],
    },
    {
        "id": "systemd_217_user",
        "category": "systemd",
        "prompt": "systemd status=217/USER 오류를 확인할 때 볼 설정과 명령어를 설명해주세요.",
        "must": ["User", "Group", "systemctl", "journalctl"],
        "nice": ["daemon-reload"],
    },
    {
        "id": "fastapi_post_json",
        "category": "fastapi",
        "prompt": "FastAPI에서 JSON body를 받는 POST 엔드포인트 최소 예제를 보여주세요.",
        "must": ["BaseModel", "@app.post", "FastAPI"],
        "nice": ["pydantic"],
    },
]

def available_models():
    data = urllib.request.urlopen("http://localhost:8000/v1/models", timeout=10).read().decode("utf-8")
    return data

def choose_model():
    data = available_models()
    if PREFERRED_MODEL in data:
        return PREFERRED_MODEL
    if FALLBACK_MODEL in data:
        return FALLBACK_MODEL
    node-7.example.invalid RuntimeError("Neither low85 nor champion model is available")

def chat(model, prompt):
    payload = {
        "model": model,
        "messages": [
            {"role": "system", "content": "당신은 한국어 존댓말로 답하는 코딩/운영 실무 비서입니다. 명령어와 코드를 먼저 제시하고 설명은 짧게 덧붙입니다. 다른 언어 문자를 섞지 않습니다."},
            {"role": "user", "content": prompt},
        ],
        "max_tokens": 280,
        "temperature": 0.2,
    }
    req = urllib.request.Request(
        BASE_URL,
        data=json.dumps(payload).encode("utf-8"),
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    t0 = time.time()
    obj = json.loads(urllib.request.urlopen(req, timeout=240).read().decode("utf-8"))
    latency = time.time() - t0
    content = obj["choices"][0]["message"]["content"]
    finish = obj["choices"][0].get("finish_reason")
    usage = obj.get("usage", {})
    return content, finish, usage, latency

def score(test, content, finish):
    s = 100
    issues = []
    lower = content.lower()
    for m in test["must"]:
        if m.lower() not in lower:
            s -= 12
            issues.append(f"missing_must:{m}")
    for n in test["nice"]:
        if n.lower() not in lower:
            s -= 3
            issues.append(f"missing_nice:{n}")
    if HAN_RE.search(content):
        s -= 25
        issues.append("possible_chinese_leakage")
    if not any(h in content for h in HONORIFIC):
        s -= 8
        issues.append("weak_korean_honorific")
    if finish == "length":
        s -= 8
        issues.append("truncated_by_max_tokens")
    return max(0, s), issues

def main():
    model = choose_model()
    print("==== LOW85 FOCUSED BENCH ====")
    print("MODEL:", model)

    results = []
    for i, t in enumerate(TESTS, 1):
        print(f"[{i}/{len(TESTS)}] {t['id']} ...", flush=True)
        try:
            content, finish, usage, latency = chat(model, t["prompt"])
            sc, issues = score(t, content, finish)
            row = {
                "id": t["id"],
                "category": t["category"],
                "score": sc,
                "issues": issues,
                "finish_reason": finish,
                "latency_sec": round(latency, 2),
                "usage": usage,
                "prompt": t["prompt"],
                "content": content,
            }
            print(f"  score={sc} latency={latency:.2f}s finish={finish} issues={issues}")
        except Exception as e:
            row = {"id": t["id"], "category": t["category"], "score": 0, "issues": [f"error:{type(e).__name__}:{e}"], "content": ""}
            print("  ERROR:", repr(e))
        results.append(row)

    avg = sum(r["score"] for r in results) / len(results)
    summary = {
        "created_at": datetime.now().isoformat(timespec="seconds"),
        "model": model,
        "num_tests": len(results),
        "average_score": round(avg, 2),
        "pass_85_plus": f"{sum(1 for r in results if r['score'] >= 85)}/{len(results)}",
        "perfect_100": f"{sum(1 for r in results if r['score'] == 100)}/{len(results)}",
    }

    ts = datetime.now().strftime("%Y%m%d_%H%M%S")
    out = Path("reports") / f"low85_focused_bench_{ts}.json"
    out.write_text(json.dumps({"summary": summary, "results": results}, ensure_ascii=False, indent=2), encoding="utf-8")

    print()
    print("==== SUMMARY ====")
    print(json.dumps(summary, ensure_ascii=False, indent=2))
    print()
    print("==== LOW ITEMS ====")
    for r in sorted(results, key=lambda x: x["score"]):
        print(f'{r["id"]}: score={r["score"]}, issues={r["issues"]}')
    print()
    print("REPORT:", out)

if __name__ == "__main__":
    main()