Data 100 LLM Benchmarking

import subprocess, sys

pkgs = ['openai', 'pandas', 'python-dotenv']
for pkg in pkgs:
    print(f'Installing {pkg}...')
    subprocess.run([sys.executable, '-m', 'pip', 'install', pkg, '-q', '--upgrade'],
                   check=True)

print('\nAll dependencies installed.')
print('Restart the kernel now if this is your first time, then continue.')

import os
from pathlib import Path
from openai import OpenAI
from dotenv import load_dotenv

# ── Load API key from ~/.env ──────────────────────────────────────────────
# Expected format in ~/.env:
#   OPENROUTER_API_KEY=sk-or-v1-...
#
# To create it the first time (run once in a terminal):
#   echo 'OPENROUTER_API_KEY=your-key-here' >> ~/.env
#   chmod 600 ~/.env
load_dotenv(Path.home() / '.env')
OPENROUTER_API_KEY = os.environ.get('OPENROUTER_API_KEY', '')

if not OPENROUTER_API_KEY:
    raise EnvironmentError(
        'OPENROUTER_API_KEY not found.\n'
        'Add it to ~/.env:\n'
        '  echo \'OPENROUTER_API_KEY=your-key-here\' >> ~/.env\n'
        'Then restart the kernel.'
    )

client = OpenAI(
    base_url='https://openrouter.ai/api/v1',
    api_key=OPENROUTER_API_KEY,
)

Path('exams').mkdir(exist_ok=True)
Path('results').mkdir(exist_ok=True)

# ── Verify connectivity ───────────────────────────────────────────────────
try:
    test = client.chat.completions.create(
        model='meta-llama/llama-3.2-1b-instruct',
        messages=[{'role': 'user', 'content': 'Reply with the single word OK.'}],
        max_tokens=5,
        temperature=0,
    )
    print('OpenRouter connection verified.')
    print(f'  Response: {test.choices[0].message.content.strip()}')
except Exception as e:
    print(f'Connection failed: {e}')

# ── Models to evaluate — use OpenRouter model IDs ────────────────────────
# Full list: https://openrouter.ai/models
EVAL_MODELS = [
    'meta-llama/llama-3.2-3b-instruct',
    'qwen/qwen-2.5-7b-instruct',
]

MODEL_DISPLAY = {
    'meta-llama/llama-3.2-3b-instruct': 'Llama-3.2-3B',
    'qwen/qwen-2.5-7b-instruct':        'Qwen-2.5-7B',
    'anthropic/claude-haiku-4-5':        'Claude-Haiku-4-5',
}

# ── Exam source ───────────────────────────────────────────────────────────

# 'exams/my.json'   -> your own exam JSON
EXAM_SOURCE = 'exams/fa25_final.json'

# ── Question types ────────────────────────────────────────────────────────
RUN_MCQ          = True
RUN_SHORT_ANSWER = True

# ── Judge model ───────────────────────────────────────────────────────────
# A separate model grades short-answer questions. This avoids self-grading
# bias and lets you use a stronger model for scoring independent of cost.
# Set to None to fall back to the same model being evaluated.
JUDGE_MODEL = 'anthropic/claude-haiku-4-5'

# ── Judge prompt strategy ─────────────────────────────────────────────────
# Controls how the judge is prompted to score short answers.
# Run Section 5b to compare strategies before choosing one.
# Options: 'baseline' | 'chain_of_thought' | 'rubric_anchored'
JUDGE_STRATEGY = 'rubric_anchored'

# ── Inference settings ────────────────────────────────────────────────────
TEMPERATURE = 0.0  # 0 = deterministic; best for reproducible evals

print('Config ready.')

import json, re, csv, time, gc
from pathlib import Path
from datetime import datetime
from dataclasses import dataclass, field
from typing import Optional
from collections import defaultdict
import pandas as pd

# ── Data classes ──────────────────────────────────────────────────────────
@dataclass
class Question:
    id: str
    type: str           # 'mcq' or 'short_answer'
    topic: str
    points: int
    question: str
    answer: str
    choices: Optional[dict] = None
    rubric: Optional[list] = None

@dataclass
class Exam:
    name: str
    semester: str
    questions: list = field(default_factory=list)

    @property
    def mcq(self): return [q for q in self.questions if q.type == 'mcq']
    @property
    def short_answer(self): return [q for q in self.questions if q.type == 'short_answer']
    @property
    def total_points(self): return sum(q.points for q in self.questions)

# ── Student-facing prompt templates ──────────────────────────────────────
def mcq_prompt(q):
    choices = '\n'.join(f'  {k}) {v}' for k, v in q.choices.items())
    return [
        {'role': 'system', 'content': 'You are a student taking a Data 100 exam at UC Berkeley. '
                                      'Answer every question directly and concisely.'},
        {'role': 'user',   'content': (
            f'Question [{q.topic}, {q.points} pts]:\n{q.question}\n\n'
            f'Choices:\n{choices}\n\n'
            f'Reply with ONLY the single letter of the correct answer.'
        )},
    ]

def short_answer_prompt(q):
    return [
        {'role': 'system', 'content': 'You are a student taking a Data 100 exam at UC Berkeley. '
                                      'Answer every question directly and concisely.'},
        {'role': 'user',   'content': (
            f'Question [{q.topic}, {q.points} pts]:\n{q.question}\n\n'
            f'Give a clear, concise answer (3-6 sentences).'
        )},
    ]

# ── Judge prompt strategies ───────────────────────────────────────────────
# Three strategies with increasing structure. JUDGE_STRATEGY (set in cell 3)
# selects which one is used in the main benchmark run.

def judge_prompt_baseline(q, student_answer):
    """Minimal prompt. Asks for a score and one sentence of feedback.
    Fast, but prone to inconsistent formatting and score inflation."""
    rubric = '\n'.join(f'  - {r}' for r in (q.rubric or []))
    return [
        {'role': 'system', 'content': 'You are a strict, impartial exam grader.'},
        {'role': 'user', 'content': (
            f'Grade this student response.\n\n'
            f'Question ({q.points} pts): {q.question}\n\n'
            f'Reference answer: {q.answer}\n\n'
            f'Rubric:\n{rubric}\n\n'
            f'Student answer: {student_answer}\n\n'
            f'Respond EXACTLY as:\nSCORE: X/{q.points}\nFEEDBACK: one sentence'
        )},
    ]

def judge_prompt_chain_of_thought(q, student_answer):
    """Ask the judge to reason through each rubric item before scoring.
    More reliable on ambiguous answers at the cost of more output tokens."""
    rubric = '\n'.join(f'  {i+1}. {r}' for i, r in enumerate(q.rubric or []))
    return [
        {'role': 'system', 'content': (
            'You are a strict, impartial exam grader. '
            'Think through your reasoning carefully before assigning a score.'
        )},
        {'role': 'user', 'content': (
            f'Grade this student response step by step.\n\n'
            f'Question ({q.points} pts): {q.question}\n\n'
            f'Reference answer: {q.answer}\n\n'
            f'Rubric criteria:\n{rubric}\n\n'
            f'Student answer: {student_answer}\n\n'
            f'First, evaluate whether the student addressed each rubric criterion. '
            f'Then assign a score.\n\n'
            f'Respond EXACTLY as:\n'
            f'REASONING: <your analysis>\n'
            f'SCORE: X/{q.points}\n'
            f'FEEDBACK: one sentence summary'
        )},
    ]

def judge_prompt_rubric_anchored(q, student_answer):
    """Score each rubric criterion independently (0 or 1), then sum.
    Most structured and reproducible; produces granular feedback."""
    criteria = q.rubric or []
    criterion_lines = '\n'.join(
        f'  Criterion {i+1}: {r}' for i, r in enumerate(criteria)
    )
    score_lines = '\n'.join(
        f'  CRITERION_{i+1}: 0 or 1' for i in range(len(criteria))
    )
    pts_per = round(q.points / len(criteria), 2) if criteria else q.points
    return [
        {'role': 'system', 'content': (
            'You are a strict, impartial exam grader. '
            'Score each criterion independently based only on whether the student addressed it.'
        )},
        {'role': 'user', 'content': (
            f'Grade this student response criterion by criterion.\n\n'
            f'Question: {q.question}\n\n'
            f'Reference answer: {q.answer}\n\n'
            f'Rubric ({len(criteria)} criteria, {pts_per} pts each):\n{criterion_lines}\n\n'
            f'Student answer: {student_answer}\n\n'
            f'Score each criterion 0 (not addressed) or 1 (addressed). Be strict.\n\n'
            f'Respond EXACTLY as:\n'
            f'{score_lines}\n'
            f'FEEDBACK: one sentence'
        )},
    ]

# Map strategy name to function
JUDGE_PROMPTS = {
    'baseline':        judge_prompt_baseline,
    'chain_of_thought': judge_prompt_chain_of_thought,
    'rubric_anchored': judge_prompt_rubric_anchored,
}

def judge_prompt(q, student_answer, strategy=None):
    """Dispatch to the selected judge prompt strategy."""
    s = strategy or JUDGE_STRATEGY
    return JUDGE_PROMPTS[s](q, student_answer)

# ── Score parsers ─────────────────────────────────────────────────────────
def parse_judge_baseline(text, max_points):
    m = re.search(r'SCORE:\s*(\d+(?:\.\d+)?)\s*/\s*(\d+)', text, re.I)
    if m:
        earned, out_of = float(m.group(1)), float(m.group(2))
        if out_of > 0:
            return min(round((earned / out_of) * max_points, 2), max_points)
    return 0.0

def parse_judge_rubric_anchored(text, q):
    """Sum CRITERION_N scores and scale to q.points."""
    criteria = q.rubric or []
    if not criteria:
        return parse_judge_baseline(text, q.points)
    total = 0
    for i in range(len(criteria)):
        m = re.search(rf'CRITERION_{i+1}:\s*([01])', text, re.I)
        if m:
            total += int(m.group(1))
    pts_per = q.points / len(criteria)
    return min(round(total * pts_per, 2), q.points)

def parse_score(judge_output, q, strategy=None):
    """Parse score from judge output using the appropriate parser."""
    s = strategy or JUDGE_STRATEGY
    if s == 'rubric_anchored':
        return parse_judge_rubric_anchored(judge_output, q)
    return parse_judge_baseline(judge_output, q.points)

def parse_feedback(text):
    m = re.search(r'FEEDBACK:\s*(.+)', text, re.I | re.DOTALL)
    return m.group(1).strip() if m else text.strip()

def extract_letter(text):
    t = text.strip().upper()
    if t and t[0] in 'ABCD': return t[0]
    m = re.search(r'\b([ABCD])\b', t)
    return m.group(1) if m else 'UNKNOWN'

def load_json_exam(data):
    qs = [Question(id=q['id'], type=q['type'], topic=q.get('topic', 'unknown'),
                   points=q['points'], question=q['question'], answer=q['answer'],
                   choices=q.get('choices'), rubric=q.get('rubric'))
          for q in data['questions']]
    return Exam(name=data['exam_name'], semester=data.get('semester', ''), questions=qs)

def parse_pdf_exam(pdf_path):
    """
    Best-effort MCQ parser for Data 100 PDFs.
    Finds questions starting with a number and answer choices A-D.
    You will likely need to manually clean results for complex questions.
    Answers are set to '?' and must be filled in from a solutions PDF.
    """
    try:
        import pdfplumber
    except ImportError:
        import subprocess, sys
        subprocess.run([sys.executable, '-m', 'pip', 'install', 'pdfplumber', '-q'])
        import pdfplumber

    questions = []
    with pdfplumber.open(pdf_path) as pdf:
        text = '\n'.join(page.extract_text() or '' for page in pdf.pages)

    blocks = re.split(r'\n(?=\d+\.\s)', text)
    qnum = 0
    for block in blocks:
        lines = block.strip().split('\n')
        if not lines: continue
        header = lines[0]
        if not re.match(r'^\d+\.', header): continue
        qtext = header + ' ' + ' '.join(
            l for l in lines[1:]
            if not re.match(r'^[A-D][.)\s]', l.strip())
        )
        choices = {}
        for line in lines:
            m = re.match(r'^([A-D])[.)\s]+(.+)', line.strip())
            if m:
                choices[m.group(1)] = m.group(2).strip()
        if len(choices) >= 2:
            qnum += 1
            questions.append(Question(
                id=f'q{qnum}', type='mcq', topic='unknown', points=2,
                question=qtext.strip(), answer='?', choices=choices,
            ))

    name = Path(pdf_path).stem
    print(f'Parsed {len(questions)} MCQ questions from {name}')
    print('NOTE: Answers are set to "?" — fill them in from the solutions PDF.')
    return Exam(name=name, semester='', questions=questions)

# ── Load exam based on EXAM_SOURCE ────────────────────────────────────────
if EXAM_SOURCE == 'sample':
    exam = load_json_exam(SAMPLE_EXAM_DATA)
elif EXAM_SOURCE.endswith('.json'):
    with open(EXAM_SOURCE) as f:
        exam = load_json_exam(json.load(f))
elif EXAM_SOURCE.endswith('.pdf'):
    exam = parse_pdf_exam(EXAM_SOURCE)
else:
    raise ValueError(f'Unknown EXAM_SOURCE: {EXAM_SOURCE}')

print(f'Exam: {exam.name} ({exam.semester})')
print(f'MCQ: {len(exam.mcq)}  |  Short answer: {len(exam.short_answer)}  |  Total: {exam.total_points} pts')
print()
pd.DataFrame([{'id': q.id, 'type': q.type, 'topic': q.topic, 'pts': q.points,
               'question': q.question[:65] + '...' if len(q.question) > 65 else q.question}
              for q in exam.questions])

def generate(model_id, messages, max_tokens=256):
    """Send a chat completion request to OpenRouter and return (text, latency_seconds)."""
    t0 = time.perf_counter()
    resp = client.chat.completions.create(
        model=model_id,
        messages=messages,
        max_tokens=max_tokens,
        temperature=TEMPERATURE,
    )
    text = resp.choices[0].message.content or ''
    return text.strip(), time.perf_counter() - t0

def display_name(model_id):
    return MODEL_DISPLAY.get(model_id, model_id.split('/')[-1])

# ── Main eval loop ────────────────────────────────────────────────────────
all_scores = []
judge_id   = JUDGE_MODEL

for model_id in EVAL_MODELS:
    disp = display_name(model_id)
    print(f"\n{'='*55}\nEvaluating: {disp}\n{'='*55}")

    judge_disp = display_name(judge_id) if judge_id else disp
    if JUDGE_MODEL and JUDGE_MODEL != model_id:
        print(f'  Judge: {judge_disp}  (strategy: {JUDGE_STRATEGY})')
    else:
        print(f'  Judge: {disp} (self-grading — consider setting a separate JUDGE_MODEL)')

    # ── MCQ ──────────────────────────────────────────────────────────────
    if RUN_MCQ and exam.mcq:
        print(f'\n  MCQ ({len(exam.mcq)} questions):')
        for q in exam.mcq:
            msgs = mcq_prompt(q)
            raw, latency = generate(model_id, msgs, max_tokens=16)
            got     = extract_letter(raw)
            correct = got == q.answer.upper()
            pts     = q.points if correct else 0
            mark    = 'PASS' if correct else 'FAIL'
            print(f'    [{mark}] {q.id:4s}  got={got}  expected={q.answer}  ({latency:.1f}s)')
            all_scores.append({
                'model': disp, 'question_id': q.id, 'type': 'mcq', 'topic': q.topic,
                'points_earned': pts, 'max_points': q.points, 'answer': got,
                'correct_answer': q.answer, 'correct': correct,
                'feedback': '', 'latency': round(latency, 2),
                'judge_strategy': '',
            })

    # ── Short answer ──────────────────────────────────────────────────────
    has_sa = RUN_SHORT_ANSWER and len(exam.short_answer) > 0
    effective_judge = judge_id or model_id

    if has_sa:
        print(f'\n  Short answer ({len(exam.short_answer)} questions):')
        for q in exam.short_answer:
            # Student response
            sa_msgs = short_answer_prompt(q)
            raw, latency = generate(model_id, sa_msgs, max_tokens=300)

            # Judge grading
            j_msgs      = judge_prompt(q, raw, strategy=JUDGE_STRATEGY)
            judge_out, _ = generate(effective_judge, j_msgs, max_tokens=350)
            pts          = parse_score(judge_out, q, strategy=JUDGE_STRATEGY)
            feedback     = parse_feedback(judge_out)

            print(f'    {q.id:4s}  score={pts}/{q.points}  ({latency:.1f}s)')
            print(f'         {feedback[:90]}')
            all_scores.append({
                'model': disp, 'question_id': q.id, 'type': 'short_answer', 'topic': q.topic,
                'points_earned': pts, 'max_points': q.points, 'answer': raw[:200],
                'correct_answer': q.answer[:100], 'correct': None,
                'feedback': feedback[:200], 'latency': round(latency, 2),
                'judge_strategy': JUDGE_STRATEGY,
            })

# ── Save to CSV ───────────────────────────────────────────────────────────
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
csv_path  = f'results/scores_{timestamp}.csv'
pd.DataFrame(all_scores).to_csv(csv_path, index=False)
print(f'\nScores saved to {csv_path}')

# ── Judge strategy comparison ─────────────────────────────────────────────
# Scores a single student answer with all three strategies side by side.
# Run this cell independently; it does not affect all_scores.

# Pick a short-answer question to test
test_q = exam.short_answer[0] if exam.short_answer else None

if test_q is None:
    print('No short-answer questions in this exam — skipping judge comparison.')
else:
    # Generate one student response from the first eval model
    test_model = EVAL_MODELS[0]
    sa_msgs    = short_answer_prompt(test_q)
    student_ans, _ = generate(test_model, sa_msgs, max_tokens=300)

    print(f'Question:  {test_q.question}')
    print(f'Reference: {test_q.answer}')
    print(f'Student ({display_name(test_model)}): {student_ans}')
    print()

    judge_to_use = JUDGE_MODEL or EVAL_MODELS[0]
    rows = []

    for strategy in ['baseline', 'chain_of_thought', 'rubric_anchored']:
        j_msgs    = judge_prompt(test_q, student_ans, strategy=strategy)
        j_out, latency = generate(judge_to_use, j_msgs, max_tokens=400)
        score     = parse_score(j_out, test_q, strategy=strategy)
        feedback  = parse_feedback(j_out)
        rows.append({
            'strategy':  strategy,
            'score':     f'{score}/{test_q.points}',
            'latency_s': round(latency, 2),
            'feedback':  feedback,
        })
        print(f'--- {strategy} ---')
        print(f'  Score:    {score}/{test_q.points}')
        print(f'  Feedback: {feedback}')
        print(f'  Latency:  {latency:.1f}s')
        print()

    print('Summary:')
    pd.DataFrame(rows)

df = pd.DataFrame(all_scores)

def leaderboard(df):
    rows = []
    for model, g in df.groupby('model'):
        earned   = g['points_earned'].sum()
        possible = g['max_points'].sum()
        pct      = 100 * earned / possible if possible else 0
        mcq      = g[g['type'] == 'mcq']
        sa       = g[g['type'] == 'short_answer']
        mcq_acc  = f"{int(100*mcq['correct'].sum()/len(mcq))}%" if len(mcq) else 'N/A'
        sa_score = f"{sa['points_earned'].sum():.1f}/{sa['max_points'].sum()}" if len(sa) else 'N/A'
        rows.append({'Model': model, 'Score': f'{earned:.0f}/{possible}',
                     'Overall %': f'{pct:.0f}%', 'MCQ Accuracy': mcq_acc,
                     'Short Answer': sa_score})
    return pd.DataFrame(rows).sort_values('Overall %', ascending=False).reset_index(drop=True)

leaderboard(df)

topic_df = (df.groupby(['model', 'topic'])
              .apply(lambda g: pd.Series({
                  'earned':   g['points_earned'].sum(),
                  'possible': g['max_points'].sum(),
                  '%':        round(100 * g['points_earned'].sum() / g['max_points'].sum())
              }))
              .reset_index())
topic_df['score'] = topic_df['earned'].astype(str) + '/' + topic_df['possible'].astype(str)
topic_df[['model', 'topic', 'score', '%']].sort_values(['model', '%'], ascending=[True, False])

detail = df[['model','question_id','type','topic','points_earned','max_points',
             'answer','correct','feedback','latency']].copy()
detail['result'] = detail.apply(
    lambda r: ('✓' if r['correct'] else '✗') if r['type'] == 'mcq' else f"{r['points_earned']}/{r['max_points']}", axis=1)
detail['answer'] = detail['answer'].str[:60]
detail['feedback'] = detail['feedback'].str[:80]
pd.set_option('display.max_colwidth', 80)
detail[['model','question_id','type','topic','result','answer','feedback','latency']]

from IPython.display import HTML

lb = leaderboard(df)

def color(pct_str):
    p = int(pct_str.replace('%',''))
    return '#4ade80' if p >= 75 else ('#fbbf24' if p >= 50 else '#f87171')

lb_rows = ''.join(
    f"<tr><td>{r['Model']}</td><td>{r['Score']}</td>"
    f"<td style='color:{color(r['Overall %'])};font-weight:600'>{r['Overall %']}</td>"
    f"<td>{r['MCQ Accuracy']}</td><td>{r['Short Answer']}</td></tr>"
    for _, r in lb.iterrows()
)

q_rows = ''.join(
    f"<tr><td>{r['model']}</td><td>{r['question_id']}</td><td>{r['type']}</td>"
    f"<td>{r['topic']}</td>"
    f"<td style='color:{color(str(int(100*r['points_earned']/r['max_points'] if r['max_points'] else 0))+'%')}'>"
    f"{r['points_earned']}/{r['max_points']}</td>"
    f"<td>{'✓' if r['correct'] else ('✗' if r['correct'] is False else '')}</td>"
    f"<td style='color:#94a3b8;font-size:.8em'>{str(r['feedback'])[:80]}</td></tr>"
    for _, r in df.iterrows()
)

html = f"""<!DOCTYPE html><html><head><meta charset='UTF-8'>
<title>Data 100 Benchmark</title>
<style>
  body{{font-family:'Courier New',monospace;background:#0f1117;color:#e2e8f0;padding:2rem}}
  h1{{color:#7dd3fc;margin-bottom:.2rem}} h2{{color:#94a3b8;margin:2rem 0 .6rem;border-bottom:1px solid #1e293b;padding-bottom:.3rem}}
  p{{color:#64748b;font-size:.85rem;margin-bottom:1.5rem}}
  table{{width:100%;border-collapse:collapse;font-size:.85rem}}
  th{{background:#1e293b;color:#7dd3fc;padding:.5rem 1rem;text-align:left}}
  td{{padding:.45rem 1rem;border-bottom:1px solid #1e293b}}
  tr:hover td{{background:#1a2236}}
</style></head><body>
<h1>Data 100 LLM Benchmark</h1>
<p>{exam.name} &mdash; {exam.semester}</p>
<h2>Leaderboard</h2>
<table><tr><th>Model</th><th>Score</th><th>%</th><th>MCQ Acc</th><th>Short Answer</th></tr>
{lb_rows}</table>
<h2>Per-Question Results</h2>
<table><tr><th>Model</th><th>Q</th><th>Type</th><th>Topic</th><th>Score</th><th>MCQ</th><th>Feedback</th></tr>
{q_rows}</table>
</body></html>"""

report_path = csv_path.replace('.csv', '.html')
with open(report_path, 'w') as f:
    f.write(html)
print(f'Report saved → {report_path}')
HTML(f'<a href="{report_path}" target="_blank" style="color:#7dd3fc">Open report ↗</a>')