|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| """Cheap evaluations for SecureCoder.
|
|
|
| Three things are scored:
|
|
|
| 1. Tool-call validity - sample prompts with real schemas, render the model's
|
| reply, parse the emitted <function=...><parameter=...> block back into
|
| JSON, score parse OK / correct function name / schema-conformant args.
|
|
|
| 2. Code sanity - self-contained Python coding prompts; the model generates
|
| code, we AST-parse + compile it (no execution).
|
|
|
| 3. Security knowledge - CyberSecurityEval MCQ when available, otherwise skip.
|
|
|
| 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
|
|
|
| import argparse
|
| import ast
|
| import json
|
| import logging
|
| import os
|
| import random
|
| import re
|
| import sys
|
| import time
|
| from typing import Any
|
|
|
| logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
|
| log = logging.getLogger("eval")
|
|
|
| CODE_PROMPTS: list[str] = [
|
| "Write a Python function `def is_palindrome(s: str) -> bool:` that returns True iff s reads the same backwards ignoring case and non-alphanumeric characters.",
|
| "Write a Python function `def merge_intervals(intervals: list[list[int]]) -> list[list[int]]:` that merges overlapping intervals.",
|
| "Write a Python function `def two_sum(nums: list[int], target: int) -> list[int]:` returning the indices of two numbers that sum to target.",
|
| "Write a Python function `def flatten(nested: list) -> list:` that flattens arbitrarily nested lists without recursion-limit errors.",
|
| "Write a Python function `def parse_csv_line(line: str) -> list[str]:` that handles quoted fields and escaped quotes per RFC 4180.",
|
| "Write a Python function `def lru_cache(k: int):` returning a decorator that keeps at most k most-recently-used call results.",
|
| "Write a Python function `def is_anagram(a: str, b: str) -> bool:` ignoring spaces, punctuation, and case.",
|
| "Write a Python function `def topological_order(graph: dict[str, list[str]]) -> list[str] | None:` returning a valid order or None on cycle.",
|
| "Write a Python function `def tokenise(s: str) -> list[str]:` for a simple expression language with integers, +, -, *, / and parentheses.",
|
| "Write a Python function `def slugify(text: str) -> str:` producing a URL-safe ASCII slug from any unicode text.",
|
| "Write a Python function `def read_jsonl(path: str) -> list[dict]:` streaming a JSONL file one record at a time without loading the whole file.",
|
| "Write a Python function `def binary_search(arr: list[int], target: int) -> int:` returning the index of target or -1.",
|
| "Write a Python function `def unique_in_order(s: str) -> list[str]:` preserving order while deduplicating adjacent equal characters.",
|
| "Write a Python function `def safe_eval(expr: str) -> int:` evaluating an integer expression with + - * / parens, no eval(), no imports.",
|
| "Write a Python function `def dedupe_preserve_order(items: list) -> list:` returning the input without duplicates, original order.",
|
| "Write a Python function `def word_frequency(text: str) -> dict[str, int]:` counting occurrences after normalising case and stripping punctuation.",
|
| "Write a Python function `def matrix_multiply(a: list[list[float]], b: list[list[float]]) -> list[list[float]]:` for any compatible shapes.",
|
| "Write a Python function `def roman_to_int(s: str) -> int:` handling subtractive notation (IV, IX, XL, XC, CD, CM) up to 3999.",
|
| "Write a Python function `def fibonacci(n: int) -> int:` returning the n-th Fibonacci number with O(n) time and O(1) space.",
|
| "Write a Python function `def validate_ipv4(s: str) -> bool:` accepting only dotted-quad strings with each octet in 0..255.",
|
| ]
|
|
|
| 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:
|
| p = argparse.ArgumentParser(description="SecureCoder evaluations")
|
| p.add_argument("--adapter", default="Taimwe/securecoder-30b-pro")
|
| p.add_argument("--base", default="unsloth/Qwen3-Coder-30B-A3B-Instruct")
|
| p.add_argument("--out-dir", default="/data/eval-out")
|
| p.add_argument("--tool-prompts", type=int, default=80)
|
| p.add_argument("--code-prompts", type=int, default=15)
|
| p.add_argument("--upload-repo", default="Taimwe/securecoder-30b-pro")
|
| p.add_argument("--max-new-tokens", type=int, default=384)
|
| p.add_argument("--seed", type=int, default=3407)
|
| return p.parse_args()
|
|
|
|
|
|
|
| _TYPE_ALIASES = {
|
| "str": "string", "string": "string", "text": "string",
|
| "int": "integer", "integer": "integer", "long": "integer",
|
| "float": "number", "double": "number", "number": "number",
|
| "bool": "boolean", "boolean": "boolean",
|
| "list": "array", "array": "array", "dict": "object", "object": "object",
|
| }
|
|
|
|
|
| def _normalise_tool_schema(tool):
|
| if not isinstance(tool, dict):
|
| return None
|
| fn = tool.get("function") if isinstance(tool.get("function"), dict) else tool
|
| if not isinstance(fn, dict) or not fn.get("name"):
|
| return None
|
| params = fn.get("parameters") or {}
|
| if isinstance(params, dict) and params and "properties" not in params:
|
| properties: dict[str, Any] = {}
|
| 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")
|
| properties[name] = cleaned
|
| if "default" not in spec:
|
| required.append(name)
|
| else:
|
| properties[name] = {"type": "string"}
|
| required.append(name)
|
| params = {"type": "object", "properties": properties}
|
| if required:
|
| params["required"] = required
|
| return {"name": fn["name"], "description": fn.get("description", ""), "parameters": params}
|
|
|
|
|
| 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(c)
|
| if n:
|
| tools.append(n)
|
| messages: list[dict] = []
|
| if isinstance(row.get("messages"), list):
|
| for turn in row["messages"]:
|
| if not isinstance(turn, dict):
|
| continue
|
| role = turn.get("role")
|
| content = turn.get("content", "")
|
| 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]:
|
| """Build (prompt, tools) tuples from a mix of Hermes-style ``conversations``
|
| (with ``from``/``value`` keys) and OpenAI-style ``messages`` rows.
|
| The Hermes rows are by far the more common in NousResearch/hermes-function-calling-v1."""
|
| ROLE_MAP = {"human": "user", "system": "system", "gpt": "assistant",
|
| "user": "user", "assistant": "assistant"}
|
| prompts = []
|
| for row in rows:
|
| tools_raw = row.get("tools")
|
| if not tools_raw:
|
| continue
|
| tools: list[dict] = []
|
| if isinstance(tools_raw, list):
|
| for t in tools_raw:
|
| n = _normalise_tool_schema(t)
|
| if n:
|
| tools.append(n)
|
| elif isinstance(tools_raw, dict):
|
| n = _normalise_tool_schema(tools_raw)
|
| if n:
|
| tools.append(n)
|
| if not tools:
|
| continue
|
|
|
| user_content = None
|
| convo = row.get("messages") or row.get("conversations") or []
|
| if isinstance(convo, list):
|
| for turn in convo:
|
| if not isinstance(turn, dict):
|
| continue
|
| role_raw = str(turn.get("from") or turn.get("role") or "").lower()
|
| role = ROLE_MAP.get(role_raw)
|
| if role != "user":
|
| continue
|
| user_content = turn.get("value", turn.get("content", ""))
|
| break
|
| if not user_content:
|
| continue
|
|
|
| prompts.append({"prompt": str(user_content)[:1200], "tools": tools,
|
| "expected_call": True})
|
| if len(prompts) >= 200:
|
| break
|
| return prompts
|
| def _render_prompt(tokenizer, prompt: str, tools: list[dict]) -> str:
|
| return tokenizer.apply_chat_template(
|
| [{"role": "user", "content": prompt}],
|
| tools=tools, tokenize=False, add_generation_prompt=True,
|
| )
|
|
|
|
|
| def _parse_emitted_calls(text: str) -> list[dict]:
|
| fns = TOOL_FN_PAT.findall(text)
|
| if not fns:
|
| return []
|
| calls = []
|
| for fn in fns:
|
| start = text.find(f"<function={fn}>")
|
| if start < 0:
|
| continue
|
| end = text.find("</function>", start)
|
| block = text[start:end if end > 0 else start + 4000]
|
| params = TOOL_PARAM_PAT.findall(block)
|
| calls.append({"name": fn, "arguments": dict(params)})
|
| return calls
|
|
|
|
|
| def _score_call(call: dict, tools: list[dict]) -> dict:
|
| name = call.get("name")
|
| fn = next((t for t in tools if t.get("name") == name), None)
|
| if not fn:
|
| return {"parse": True, "name_ok": False, "schema_ok": False}
|
| props = fn.get("parameters", {}).get("properties", {}) or {}
|
| expected = set(props.keys())
|
| given = set((call.get("arguments") or {}).keys())
|
| 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)
|
| prompts = _build_tool_prompts(rows)
|
| if args.tool_prompts:
|
| prompts = prompts[: args.tool_prompts]
|
| log.info("tool-call eval: %d usable prompts", len(prompts))
|
|
|
| out = []
|
| for i, p in enumerate(prompts):
|
| try:
|
| text = _render_prompt(tokenizer, p["prompt"], p["tools"])
|
| ids = tokenizer(text, return_tensors="pt", add_special_tokens=False).input_ids.to(model.device)
|
| 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:
|
| 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})
|
| 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"]))
|
| schema_ok = sum(1 for r in out if r.get("scores") and any(s["schema_ok"] for s in r["scores"]))
|
| return {"section": "tool_calls", "n_prompts": n,
|
| "parse_rate": parse_ok / max(n, 1), "name_rate": name_ok / max(n, 1),
|
| "schema_rate": schema_ok / max(n, 1), "details": out}
|
|
|
|
|
| def eval_code_sanity(tokenizer, model, args) -> dict:
|
| import torch
|
| out = []
|
| prompts = CODE_PROMPTS[: args.code_prompts]
|
| for i, prompt in enumerate(prompts):
|
| text = tokenizer.apply_chat_template(
|
| [{"role": "user", "content": prompt}],
|
| 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=384, do_sample=False)
|
| reply = tokenizer.decode(generated[0, ids.shape[-1]:], skip_special_tokens=True)
|
| except Exception as exc:
|
| out.append({"prompt": prompt[:60], "error": repr(exc)[:120]})
|
| continue
|
| block = None
|
|
|
|
|
|
|
|
|
| blocks = re.findall(r"```(?:python)?\s*\n(.*?)```", reply, re.S)
|
| if blocks:
|
| block = max(blocks, key=len)
|
| else:
|
| start = reply.find("def ")
|
| if start >= 0:
|
| block = reply[start:]
|
| if block:
|
|
|
| block = re.split(r"\n```", block, maxsplit=1)[0]
|
|
|
| parsed = compiles = None
|
| if block:
|
| try:
|
| ast.parse(block)
|
| parsed = True
|
| except SyntaxError:
|
| parsed = False
|
| if block and parsed:
|
| try:
|
| compile(block, "<eval>", "exec")
|
| compiles = True
|
| except Exception:
|
| compiles = False
|
| 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:
|
| 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 = []
|
| 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")
|
| answer = r.get("answer") or r.get("label")
|
| if not question or not options or answer is None:
|
| continue
|
| if isinstance(options, dict):
|
| opts = "\n".join(f"{k}. {v}" for k, v in options.items())
|
| key_map = {str(k): v for k, v in options.items()}
|
| 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)
|
| 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:
|
| 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}
|
|
|
|
|
| def main() -> int:
|
| args = parse_args()
|
| token = os.environ.get("HF_TOKEN")
|
| if not token:
|
| log.error("HF_TOKEN not set"); return 1
|
| os.makedirs(args.out_dir, exist_ok=True)
|
| import torch
|
| from transformers import AutoTokenizer
|
| from peft import PeftModel
|
| from unsloth import FastLanguageModel
|
|
|
| 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 = PeftModel.from_pretrained(model, args.adapter, token=token)
|
| log.info("adapter loaded")
|
|
|
| sections = [
|
| eval_tool_calls(tokenizer, model, args),
|
| eval_code_sanity(tokenizer, model, args),
|
| eval_security_mcq(tokenizer, model),
|
| ]
|
|
|
| summary = {
|
| "adapter": args.adapter,
|
| "base": args.base,
|
| "when": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
|
| "sections": [
|
| {"section": s["section"], **{k: v for k, v in s.items() if k not in {"section", "details"}}}
|
| for s in sections
|
| ],
|
| "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)
|
| log.info("report written: %s", out_json)
|
|
|
| print("\n" + "=" * 70)
|
| print(f"{'section':<18}{'metric':<22}{'value':>10}")
|
| print("-" * 70)
|
| for s in sections:
|
| for k, v in s.items():
|
| if isinstance(v, float) and k.endswith(("rate", "accuracy")):
|
| print(f"{s['section']:<18}{k:<22}{v*100:>9.1f}%")
|
| if "n_prompts" in s:
|
| print(f"{s['section']:<18}{'n_prompts':<22}{s['n_prompts']:>10}")
|
| print("=" * 70)
|
|
|
| if args.upload_repo:
|
| from huggingface_hub import HfApi
|
| api = HfApi(token=token)
|
| api.create_repo(args.upload_repo, repo_type="model", exist_ok=True, private=True)
|
| api.upload_file(path_or_fileobj=out_json, path_in_repo="eval-report.json",
|
| 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
|
|
|
|
|
| if __name__ == "__main__":
|
| raise SystemExit(main())
|
|
|
