A 284B model was built for data centers. A 2021 Mac runs it.

"""Independent eval runner for DeepSeek V4 Flash q2 served by ds4-server.

Usage:
  python3 harness.py aime       # 30 AIME 2025 problems, thinking mode
  python3 harness.py humaneval  # 164 HumanEval tasks, non-thinking, greedy

Results append to results/<suite>.jsonl (resumable: done ids are skipped).
"""

import csv
import json
import random
import sys
import time
import urllib.request
from pathlib import Path

from graders import extract_code, extract_mcq_letter, grade_aime, run_humaneval_check

BASE = Path(__file__).parent
SERVER = "http://127.0.0.1:8000/v1/chat/completions"

AIME_INSTRUCTION = (
    "Please reason step by step, and put your final answer within \\boxed{}. "
    "The answer is an integer between 0 and 999."
)
HUMANEVAL_INSTRUCTION = (
    "Complete the following Python function. Reply with the complete, "
    "self-contained implementation (including the function signature and any "
    "imports it needs) in a single ```python code block. Do not include tests "
    "or example usage.\n\n```python\n{prompt}```"
)


def chat(payload, timeout=3600):
    req = urllib.request.Request(
        SERVER,
        data=json.dumps(payload).encode(),
        headers={"Content-Type": "application/json"},
    )
    start = time.monotonic()
    with urllib.request.urlopen(req, timeout=timeout) as resp:
        body = json.load(resp)
    elapsed = time.monotonic() - start
    msg = body["choices"][0]["message"]
    return {
        "content": msg.get("content") or "",
        "reasoning": msg.get("reasoning_content") or "",
        "finish_reason": body["choices"][0].get("finish_reason"),
        "usage": body.get("usage", {}),
        "wall_s": round(elapsed, 2),
    }


def load_done(path):
    if not path.exists():
        return set()
    return {json.loads(line)["id"] for line in path.open() if line.strip()}


def run_aime(out_path, done):
    problems = json.load((BASE / "data" / "aime25.json").open())
    for p in problems:
        pid = f"aime25-{p['id']}"
        if pid in done:
            continue
        r = chat(
            {
                "model": "deepseek-v4-flash",
                "messages": [
                    {"role": "user", "content": p["problem"] + "\n\n" + AIME_INSTRUCTION}
                ],
                "max_tokens": 24576,
            }
        )
        graded_text = r["content"] if r["content"].strip() else r["reasoning"]
        passed = grade_aime(graded_text, p["answer"])
        yield out_path, {
            "id": pid,
            "passed": passed,
            "expected": p["answer"],
            "finish_reason": r["finish_reason"],
            "usage": r["usage"],
            "wall_s": r["wall_s"],
            "content": r["content"],
            "reasoning_chars": len(r["reasoning"]),
        }


def run_humaneval(out_path, done):
    tasks = [
        json.loads(line)
        for line in (BASE / "data" / "HumanEval.jsonl").open()
        if line.strip()
    ]
    for t in tasks:
        tid = t["task_id"]
        if tid in done:
            continue
        r = chat(
            {
                "model": "deepseek-v4-flash",
                "messages": [
                    {
                        "role": "user",
                        "content": HUMANEVAL_INSTRUCTION.format(prompt=t["prompt"]),
                    }
                ],
                "think": False,
                "temperature": 0,
                "max_tokens": 4096,
            }
        )
        code = extract_code(r["content"])
        passed, detail = run_humaneval_check(code, t["test"], t["entry_point"])
        yield out_path, {
            "id": tid,
            "passed": passed,
            "detail": detail if not passed else "ok",
            "finish_reason": r["finish_reason"],
            "usage": r["usage"],
            "wall_s": r["wall_s"],
            "content": r["content"],
        }


GPQA_INSTRUCTION = (
    "Answer the following multiple-choice question. Think briefly if needed, then "
    "give your final answer on the last line in the exact form 'Answer: X' where X "
    "is one of A, B, C, D."
)


def run_gpqa(out_path, done):
    with (BASE / "data" / "gpqa_diamond.csv").open(newline="") as f:
        rows = list(csv.DictReader(f))
    for i, row in enumerate(rows):
        qid = f"gpqa-diamond-{i}"
        if qid in done:
            continue
        choices = [
            ("correct", row["Correct Answer"].strip()),
            ("wrong", row["Incorrect Answer 1"].strip()),
            ("wrong", row["Incorrect Answer 2"].strip()),
            ("wrong", row["Incorrect Answer 3"].strip()),
        ]
        random.Random(i).shuffle(choices)
        letters = "ABCD"
        expected = letters[[k for k, _ in choices].index("correct")]
        options = "\n".join(f"{letters[j]}) {text}" for j, (_, text) in enumerate(choices))
        prompt = f"{GPQA_INSTRUCTION}\n\n{row['Question'].strip()}\n\n{options}"
        r = chat(
            {
                "model": "deepseek-v4-flash",
                "messages": [{"role": "user", "content": prompt}],
                "think": False,
                "temperature": 0,
                "max_tokens": 3072,
            }
        )
        got = extract_mcq_letter(r["content"])
        yield out_path, {
            "id": qid,
            "passed": got == expected,
            "expected": expected,
            "given": got,
            "finish_reason": r["finish_reason"],
            "usage": r["usage"],
            "wall_s": r["wall_s"],
            "content": r["content"],
        }


def main():
    suite = sys.argv[1] if len(sys.argv) > 1 else ""
    runners = {"aime": run_aime, "humaneval": run_humaneval, "gpqa": run_gpqa}
    if suite not in runners:
        sys.exit(f"usage: harness.py [{'|'.join(runners)}] [result-suffix]")
    suffix = f"-{sys.argv[2]}" if len(sys.argv) > 2 else ""
    out_path = BASE / "results" / f"{suite}{suffix}.jsonl"
    out_path.parent.mkdir(exist_ok=True)
    done = load_done(out_path)
    if done:
        print(f"resuming: {len(done)} items already done", flush=True)
    n_pass = n_run = 0
    for path, record in runners[suite](out_path, done):
        with path.open("a") as f:
            f.write(json.dumps(record) + "\n")
        n_run += 1
        n_pass += bool(record["passed"])
        usage = record["usage"]
        gen = usage.get("completion_tokens", 0)
        tps = round(gen / record["wall_s"], 2) if record["wall_s"] else 0
        print(
            f"[{record['id']}] {'PASS' if record['passed'] else 'FAIL'} "
            f"gen={gen}tok wall={record['wall_s']}s ({tps} t/s) "
            f"running: {n_pass}/{n_run}",
            flush=True,
        )
    print(f"suite={suite} new_pass={n_pass}/{n_run}", flush=True)


if __name__ == "__main__":
    main()

"""Grading logic for the independent AIME 2025 / HumanEval eval."""

import re
import subprocess
import sys
import tempfile

BOXED_RE = re.compile(r"\\boxed\{([^{}]*)\}")
FENCE_RE = re.compile(r"```(?:python)?\s*\n(.*?)```", re.DOTALL)
INT_RE = re.compile(r"-?\d[\d,]*")


def extract_aime_answer(text):
    """Return the model's integer answer: last \\boxed{} if present, else last integer."""
    if not text:
        return None
    boxed = BOXED_RE.findall(text)
    candidates = boxed if boxed else INT_RE.findall(text)
    for cand in reversed(candidates):
        digits = cand.strip().replace(",", "")
        try:
            return int(digits)
        except ValueError:
            continue
    return None


def grade_aime(text, expected):
    got = extract_aime_answer(text)
    return got is not None and got == int(expected)


ANSWER_LETTER_RE = re.compile(
    r"(?:answer|correct option|choice)\b[^A-D]{0,20}\b([A-D])\b", re.IGNORECASE
)
STANDALONE_LETTER_RE = re.compile(r"\(?\b([A-D])\b\)?")


def extract_mcq_letter(text):
    """Return the model's A-D choice: prefer an 'Answer: X' style match, else the
    last standalone capital letter A-D."""
    if not text:
        return None
    answered = ANSWER_LETTER_RE.findall(text)
    if answered:
        return answered[-1].upper()
    standalone = [m for m in STANDALONE_LETTER_RE.findall(text) if m.isupper()]
    return standalone[-1] if standalone else None


def extract_code(text):
    """Return the last fenced code block, or the raw text if no fence exists."""
    if not text:
        return ""
    blocks = FENCE_RE.findall(text)
    return blocks[-1].strip() if blocks else text.strip()


def run_humaneval_check(code, test_src, entry_point, timeout=15):
    """Execute candidate code against the HumanEval test in a subprocess.

    Returns (passed: bool, detail: str).
    """
    program = "\n".join(
        [
            "from typing import *",
            "import math",
            "",
            code,
            "",
            test_src,
            f"check({entry_point})",
            "print('HUMANEVAL_OK')",
        ]
    )
    with tempfile.TemporaryDirectory() as sandbox:
        try:
            proc = subprocess.run(
                [sys.executable, "-c", program],
                cwd=sandbox,
                capture_output=True,
                text=True,
                timeout=timeout,
            )
        except subprocess.TimeoutExpired:
            return False, "timeout"
    if proc.returncode == 0 and "HUMANEVAL_OK" in proc.stdout:
        return True, "ok"
    return False, (proc.stderr or proc.stdout).strip()[-500:]

import unittest

from graders import extract_aime_answer, grade_aime, extract_code, run_humaneval_check


class TestAimeExtraction(unittest.TestCase):
    def test_boxed_answer(self):
        self.assertEqual(extract_aime_answer(r"The answer is \boxed{70}."), 70)

    def test_last_boxed_wins(self):
        text = r"First we get \boxed{12}, but correcting: \boxed{588}"
        self.assertEqual(extract_aime_answer(text), 588)

    def test_boxed_with_spaces_and_commas(self):
        self.assertEqual(extract_aime_answer(r"\boxed{ 1,024 }"), 1024)

    def test_no_boxed_falls_back_to_last_integer(self):
        self.assertEqual(extract_aime_answer("So the final answer is 204."), 204)

    def test_no_answer(self):
        self.assertIsNone(extract_aime_answer("I could not solve this."))

    def test_grade(self):
        self.assertTrue(grade_aime(r"\boxed{70}", "70"))
        self.assertFalse(grade_aime(r"\boxed{71}", "70"))
        self.assertFalse(grade_aime("no idea", "70"))


class TestCodeExtraction(unittest.TestCase):
    def test_python_fence(self):
        text = "Here you go:\n```python\ndef f():\n    return 1\n```\nDone."
        self.assertEqual(extract_code(text), "def f():\n    return 1")

    def test_last_fence_wins(self):
        text = "```python\ndef f():\n    return 1\n```\nOops, fix:\n```python\ndef f():\n    return 2\n```"
        self.assertEqual(extract_code(text), "def f():\n    return 2")

    def test_bare_fence(self):
        text = "```\ndef f():\n    return 3\n```"
        self.assertEqual(extract_code(text), "def f():\n    return 3")

    def test_no_fence_returns_raw(self):
        text = "def f():\n    return 4"
        self.assertEqual(extract_code(text), "def f():\n    return 4")


class TestHumanEvalCheck(unittest.TestCase):
    TEST_SRC = (
        "def check(candidate):\n"
        "    assert candidate(2) == 4\n"
        "    assert candidate(3) == 9\n"
    )

    def test_pass(self):
        ok, detail = run_humaneval_check("def sq(x):\n    return x * x", self.TEST_SRC, "sq")
        self.assertTrue(ok, detail)

    def test_wrong_answer_fails(self):
        ok, _ = run_humaneval_check("def sq(x):\n    return x + x", self.TEST_SRC, "sq")
        self.assertFalse(ok)

    def test_exception_fails(self):
        ok, _ = run_humaneval_check("def sq(x):\n    raise ValueError", self.TEST_SRC, "sq")
        self.assertFalse(ok)

    def test_infinite_loop_times_out(self):
        ok, detail = run_humaneval_check(
            "def sq(x):\n    while True:\n        pass", self.TEST_SRC, "sq", timeout=3
        )
        self.assertFalse(ok)
        self.assertIn("timeout", detail)

    def test_typing_names_available(self):
        code = "def ident(x: List[int]) -> List[int]:\n    return x"
        test = "def check(candidate):\n    assert candidate([1]) == [1]\n"
        ok, detail = run_humaneval_check(code, test, "ident")
        self.assertTrue(ok, detail)


if __name__ == "__main__":
    unittest.main()


class TestMcqExtraction(unittest.TestCase):
    def test_answer_prefix(self):
        from graders import extract_mcq_letter
        self.assertEqual(extract_mcq_letter("Some reasoning.\n\nAnswer: C"), "C")

    def test_answer_bold(self):
        from graders import extract_mcq_letter
        self.assertEqual(extract_mcq_letter("The answer is **B**."), "B")

    def test_parenthesized(self):
        from graders import extract_mcq_letter
        self.assertEqual(extract_mcq_letter("Therefore (D) is correct."), "D")

    def test_last_standalone_letter_fallback(self):
        from graders import extract_mcq_letter
        self.assertEqual(extract_mcq_letter("Options A and B fail, so C."), "C")

    def test_no_letter(self):
        from graders import extract_mcq_letter
        self.assertIsNone(extract_mcq_letter("I am not sure."))

    def test_ignores_lowercase_article_a(self):
        from graders import extract_mcq_letter
        self.assertEqual(extract_mcq_letter("This is a tricky one. Answer: B"), "B")