Eval v2: clean rewrite with PEP 723 deps + inlined helpers

eval_securecoder.py

@@ -3,6 +3,10 @@
 # dependencies = [
 #     "datasets",
 #     "huggingface_hub",
+#     "torch",
+#     "transformers",
+#     "peft",
+#     "unsloth",
 # ]
 # ///
 """Cheap evaluations for SecureCoder.
@@ -18,8 +22,8 @@ Three things are scored:
 
   3. Security knowledge - CyberSecurityEval MCQ when available, otherwise skip.
 
-The script writes results to --out-dir/report.json, prints a summary table, and
-uploads the report to --upload-repo if set (default: the adapter repo itself).
+Writes results to --out-dir/report.json, prints a summary table, and uploads
+the report to --upload-repo if set (default: the adapter repo itself).
 """
 
 from __future__ import annotations
@@ -63,6 +67,7 @@ CODE_PROMPTS: list[str] = [
 
 TOOL_FN_PAT = re.compile(r"<function=([A-Za-z0-9_\.]+)>", re.S)
 TOOL_PARAM_PAT = re.compile(r"<parameter=([A-Za-z0-9_]+)>\s*(.*?)\s*</parameter>", re.S)
+
 
 
 def parse_args() -> argparse.Namespace:
@@ -76,23 +81,9 @@ def parse_args() -> argparse.Namespace:
     p.add_argument("--max-new-tokens", type=int, default=384)
     p.add_argument("--seed", type=int, default=3407)
     return p.parse_args()
+
 
-def _fetch_first_rows(repo: str, config: str | None, split: str, n: int) -> list[dict]:
-    from datasets import load_dataset
 
-    kwargs: dict[str, Any] = {"split": split, "streaming": True}
-    if config:
-        kwargs["name"] = config
-    ds = load_dataset(repo, token=os.environ.get("HF_TOKEN"), **kwargs)
-    out = []
-    for row in ds:
-        out.append(dict(row))
-
-
-# --------------------------------------------------------------------------
-# Local helpers (inlined so the script runs standalone on HF Jobs without
-# needing train_securecoder.py to also be uploaded).
-# --------------------------------------------------------------------------
 _TYPE_ALIASES = {
     "str": "string", "string": "string", "text": "string",
     "int": "integer", "integer": "integer", "long": "integer",
@@ -102,7 +93,7 @@ _TYPE_ALIASES = {
 }
 
 
-def _normalise_tool_schema_local(tool: Any) -> dict | None:
+def _normalise_tool_schema(tool):
     if not isinstance(tool, dict):
         return None
     fn = tool.get("function") if isinstance(tool.get("function"), dict) else tool
@@ -114,10 +105,8 @@ def _normalise_tool_schema_local(tool: Any) -> dict | None:
         required: list[str] = []
         for name, spec in params.items():
             if isinstance(spec, dict):
-                cleaned = {k: v for k, v in spec.items()
-                           if k in ("type", "description", "enum", "default", "title", "items")}
-                cleaned["type"] = _TYPE_ALIASES.get(
-                    str(cleaned.get("type", "")).lower(), "string")
+                cleaned = {k: v for k, v in spec.items() if k in ("type", "description", "enum", "default", "title", "items")}
+                cleaned["type"] = _TYPE_ALIASES.get(str(cleaned.get("type", "")).lower(), "string")
                 properties[name] = cleaned
                 if "default" not in spec:
                     required.append(name)
@@ -130,15 +119,14 @@ def _normalise_tool_schema_local(tool: Any) -> dict | None:
     return {"name": fn["name"], "description": fn.get("description", ""), "parameters": params}
 
 
-def _messages_from_any_local(row: dict, kind: str) -> tuple[list[dict], list[dict]]:
+def _messages_from_any(row, kind):
     tools: list[dict] = []
     raw = row.get("tools")
     candidates = [raw] if isinstance(raw, dict) else (raw if isinstance(raw, list) else [])
     for c in candidates:
-        n = _normalise_tool_schema_local(c)
+        n = _normalise_tool_schema(c)
         if n:
             tools.append(n)
-
     messages: list[dict] = []
     if isinstance(row.get("messages"), list):
         for turn in row["messages"]:
@@ -149,36 +137,39 @@ def _messages_from_any_local(row: dict, kind: str) -> tuple[list[dict], list[dic
             if role in {"user", "assistant", "system"} and content:
                 messages.append({"role": role, "content": str(content)})
         return messages, tools
-
     if row.get("query") and row.get("answers"):
         messages.append({"role": "user", "content": str(row["query"])})
         messages.append({"role": "assistant", "content": str(row["answers"])})
         return messages, tools
-
     if row.get("user") and row.get("assistant"):
         if row.get("system"):
             messages.append({"role": "system", "content": str(row["system"])})
         messages.append({"role": "user", "content": str(row["user"])})
         messages.append({"role": "assistant", "content": str(row["assistant"])})
         return messages, tools
-
     if row.get("instruction") and (row.get("output") or row.get("response")):
         messages.append({"role": "user", "content": str(row["instruction"])})
         messages.append({"role": "assistant", "content": str(row.get("output") or row.get("response"))})
         return messages, tools
-
     return [], tools
+
 
 
-
+def _fetch_first_rows(repo: str, config: str | None, split: str, n: int) -> list[dict]:
+    from datasets import load_dataset
+    kwargs: dict[str, Any] = {"split": split, "streaming": True}
+    if config:
+        kwargs["name"] = config
+    ds = load_dataset(repo, token=os.environ.get("HF_TOKEN"), **kwargs)
+    out = []
+    for row in ds:
+        out.append(dict(row))
         if len(out) >= n:
             break
     return out
 
 
 def _build_tool_prompts(rows: list[dict]) -> list[dict]:
-
-
     prompts = []
     for row in rows:
         if not isinstance(row.get("messages"), list):
@@ -186,19 +177,19 @@ def _build_tool_prompts(rows: list[dict]) -> list[dict]:
         tools_raw = row.get("tools")
         if not tools_raw:
             continue
-        tools = []
+        tools: list[dict] = []
         if isinstance(tools_raw, list):
             for t in tools_raw:
-                n = _normalise_tool_schema_local(t)
+                n = _normalise_tool_schema(t)
                 if n:
                     tools.append(n)
         elif isinstance(tools_raw, dict):
-            n = _normalise_tool_schema_local(tools_raw)
+            n = _normalise_tool_schema(tools_raw)
             if n:
                 tools.append(n)
         if not tools:
             continue
-        messages, _ = _messages_from_any_local(row, "auto")
+        messages, _ = _messages_from_any(row, "auto")
         if not messages:
             continue
         user = next((m["content"] for m in messages if m["role"] == "user"), None)
@@ -245,10 +236,11 @@ def _score_call(call: dict, tools: list[dict]) -> dict:
     return {"parse": True, "name_ok": True,
             "schema_ok": expected.issubset(given) if expected else True,
             "expected_keys": sorted(expected), "given_keys": sorted(given)}
+
+
 
 def eval_tool_calls(tokenizer, model, args) -> dict:
     import torch
-
     log.info("tool-call eval: streaming candidates from hermes FC ...")
     rows = _fetch_first_rows("NousResearch/hermes-function-calling-v1", "func_calling", "train",
                              args.tool_prompts * 4)
@@ -265,23 +257,16 @@ def eval_tool_calls(tokenizer, model, args) -> dict:
             with torch.no_grad():
                 generated = model.generate(ids, max_new_tokens=args.max_new_tokens, do_sample=False)
             reply = tokenizer.decode(generated[0, ids.shape[-1]:], skip_special_tokens=True)
-        except Exception as exc:  # noqa: BLE001
+        except Exception as exc:
             out.append({"prompt": p["prompt"][:60], "error": repr(exc)[:120]})
             continue
-
         calls = _parse_emitted_calls(reply)
         scored = [_score_call(c, p["tools"]) for c in calls]
-        out.append({
-            "prompt": p["prompt"][:80],
-            "reply_first_160": reply[:160],
-            "expected_call": p["expected_call"],
-            "n_calls": len(calls),
-            "calls": calls,
-            "scores": scored,
-        })
+        out.append({"prompt": p["prompt"][:80], "reply_first_160": reply[:160],
+                    "expected_call": p["expected_call"], "n_calls": len(calls),
+                    "calls": calls, "scores": scored})
         if (i + 1) % 25 == 0:
             log.info("  tool-call progress: %d/%d", i + 1, len(prompts))
-
     n = len(out)
     parse_ok = sum(1 for r in out if r.get("scores") and any(s["parse"] for s in r["scores"]))
     name_ok = sum(1 for r in out if r.get("scores") and any(s["name_ok"] for s in r["scores"]))
@@ -293,7 +278,6 @@ def eval_tool_calls(tokenizer, model, args) -> dict:
 
 def eval_code_sanity(tokenizer, model, args) -> dict:
     import torch
-
     out = []
     prompts = CODE_PROMPTS[: args.code_prompts]
     for i, prompt in enumerate(prompts):
@@ -306,59 +290,51 @@ def eval_code_sanity(tokenizer, model, args) -> dict:
             with torch.no_grad():
                 generated = model.generate(ids, max_new_tokens=384, do_sample=False)
             reply = tokenizer.decode(generated[0, ids.shape[-1]:], skip_special_tokens=True)
-        except Exception as exc:  # noqa: BLE001
+        except Exception as exc:
             out.append({"prompt": prompt[:60], "error": repr(exc)[:120]})
             continue
-
         block = None
-        match = re.search(r"```(?:python)?\s*\n(.*?)```", reply, re.S)
-        if match:
-            block = match.group(1)
+        m = re.search(r"```(?:python)?\s*\n(.*?)```", reply, re.S)
+        if m:
+            block = m.group(1)
         else:
             start = reply.find("def ")
             if start >= 0:
                 block = reply[start:]
-
         parsed = compiles = None
         if block:
             try:
-                ast.parse(block)
-                parsed = True
+                ast.parse(block); parsed = True
             except SyntaxError:
-                parsed = False
-                block = None
+                parsed = False; block = None
         if block:
             try:
-                compile(block, "<eval>", "exec")
-                compiles = True
-            except Exception:  # noqa: BLE001
+                compile(block, "<eval>", "exec"); compiles = True
+            except Exception:
                 compiles = False
         out.append({"prompt": prompt[:60], "ast_ok": parsed, "compile_ok": compiles,
                     "reply_first_160": reply[:160]})
         if (i + 1) % 5 == 0:
             log.info("  code sanity: %d/%d", i + 1, len(prompts))
-
     n = len(out)
     ast_ok = sum(1 for r in out if r.get("ast_ok"))
     compile_ok = sum(1 for r in out if r.get("compile_ok"))
     return {"section": "code_sanity", "n_prompts": n,
             "ast_rate": ast_ok / max(n, 1), "compile_rate": compile_ok / max(n, 1),
             "details": out}
+
+
 
 def eval_security_mcq(tokenizer, model, n_questions: int = 25) -> dict:
     import torch
-
     try:
         rows = _fetch_first_rows("CyberNative/CyberSecurityEval", None, "train", n_questions * 2)
-    except Exception as exc:  # noqa: BLE001
+    except Exception as exc:
         return {"section": "security_mcq", "error": repr(exc)[:200], "skipped": True}
-
     rows = rows[:n_questions]
     if not rows:
         return {"section": "security_mcq", "skipped": True, "reason": "no rows"}
-
-    correct = 0
-    details = []
+    correct = 0; details = []
     for r in rows:
         question = r.get("question") or r.get("prompt") or r.get("input")
         options = r.get("options") or r.get("choices") or r.get("answers")
@@ -371,40 +347,30 @@ def eval_security_mcq(tokenizer, model, n_questions: int = 25) -> dict:
         else:
             opts = "\n".join(f"{i}. {o}" for i, o in enumerate(options))
             key_map = {str(i): options[i]}
-
         user = f"Question: {question}\n\n{opts}\n\nRespond with the letter of the correct answer only."
         text = tokenizer.apply_chat_template(
-            [{"role": "user", "content": user}], tokenize=False, add_generation_prompt=True,
-        )
+            [{"role": "user", "content": user}], tokenize=False, add_generation_prompt=True)
         ids = tokenizer(text, return_tensors="pt", add_special_tokens=False).input_ids.to(model.device)
         try:
             with torch.no_grad():
                 generated = model.generate(ids, max_new_tokens=8, do_sample=False)
             reply = tokenizer.decode(generated[0, ids.shape[-1]:], skip_special_tokens=True).strip()
-        except Exception:  # noqa: BLE001
+        except Exception:
             continue
-
         first_letter = reply[:1].upper()
         predicted = key_map.get(first_letter)
         is_correct = predicted == answer
         correct += int(is_correct)
         details.append({"question": str(question)[:80], "reply": reply[:10], "ok": is_correct})
-
-    return {
-        "section": "security_mcq",
-        "n_questions": len(details),
-        "accuracy": correct / max(len(details), 1),
-        "details": details,
-    }
+    return {"section": "security_mcq", "n_questions": len(details),
+            "accuracy": correct / max(len(details), 1), "details": details}
 
 
 def main() -> int:
     args = parse_args()
     token = os.environ.get("HF_TOKEN")
     if not token:
-        log.error("HF_TOKEN not set")
-        return 1
-
+        log.error("HF_TOKEN not set"); return 1
     os.makedirs(args.out_dir, exist_ok=True)
     import torch
     from transformers import AutoTokenizer
@@ -414,8 +380,7 @@ def main() -> int:
     random.seed(args.seed)
     log.info("loading adapter %s on top of %s ...", args.adapter, args.base)
     model, tokenizer = FastLanguageModel.from_pretrained(
-        model_name=args.base, max_seq_length=2048, dtype=torch.bfloat16, load_in_4bit=True,
-    )
+        model_name=args.base, max_seq_length=2048, dtype=torch.bfloat16, load_in_4bit=True)
     model = PeftModel.from_pretrained(model, args.adapter, token=token)
     log.info("adapter loaded")
 
@@ -435,7 +400,6 @@ def main() -> int:
         ],
         "raw": sections,
     }
-
     out_json = os.path.join(args.out_dir, "report.json")
     with open(out_json, "w", encoding="utf-8") as fh:
         json.dump(summary, fh, indent=2, default=str)
@@ -460,7 +424,6 @@ def main() -> int:
                         repo_id=args.upload_repo, repo_type="model",
                         commit_message="Add evaluation report")
         log.info("report pushed to https://huggingface.co/%s", args.upload_repo)
-
     return 0