python"""
side_by_side_eval.py
High-level harness:
- tasks: list of dicts {name, prompt, test_file_contents}
- apis: simple wrappers for OpenAI (GPT-5) and Anthropic (Claude Opus 4.1)
- run: for each task, call each model, write code, run pytest, collect metrics
NOTE: replace API_KEY_* with your keys and confirm official endpoints/params per vendor docs.
"""
import os
import json
import subprocess
import time
from typing import Dict, Any
import requests
# === CONFIG - fill these from your environment ===
# === Simple API wrappers (check vendor docs for exact endpoints/params) ===
def call_gpt5(prompt: str, max_tokens=1024) -> Dict:
url = "https://api.cometapi.com/v1/responses" # example; confirm actual endpoint headers = {"Authorization": f"Bearer {CometAPI_API_KEY}"}
body = {
"model": "gpt-5",
"input": prompt,
"max_output_tokens": max_tokens
}
t0 = time.time()
r = requests.post(url, headers=headers, json=body, timeout=60)
latency = time.time() - t0
r.raise_for_status()
resp = r.json()
# token info might be in resp depending on API; adapt as needed
return {"text": resp if "output_text" in resp else resp, "raw": resp, "latency": latency}
def call_claude(prompt: str, max_tokens=1024) -> Dict:
url = "https://api.cometapi.com/v1/chat/completions" # example; confirm actual endpoint headers = {"x-api-key": CometAPI_API_KEY}
body = {
"model": "claude-opus-4-1-20250805", "prompt": prompt,
"max_tokens_to_sample": max_tokens
}
t0 = time.time()
r = requests.post(url, headers=headers, json=body, timeout=60)
latency = time.time() - t0
r.raise_for_status()
resp = r.json()
return {"text": resp.get("completion", ""), "raw": resp, "latency": latency}
# === Test runner ===
def run_task(task: Dict, model_fn, model_name: str):
"""Run a single task: call model, write file, run pytest, collect result."""
prompt = task
result = model_fn(prompt, max_tokens=task.get("max_tokens", 2048))
code_text = result
# write task files into temporary folder
tmpdir = f"runs/{task}/{model_name}"
os.makedirs(tmpdir, exist_ok=True)
code_file = os.path.join(tmpdir, "submission.py")
with open(code_file, "w") as f:
f.write(code_text)
# write tests
test_file = os.path.join(tmpdir, "test_submission.py")
with open(test_file, "w") as f:
f.write(task)
# run pytest in subprocess with timeout
try:
proc = subprocess.run(
,
stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
timeout=30
)
passed = proc.returncode == 0
output = proc.stdout.decode()
except subprocess.TimeoutExpired:
passed = False
output = "pytest timeout"
return {
"model": model_name,
"task": task,
"passed": passed,
"latency": result,
"tokens_estimate": result.get("usage", {}),
"stdout": output,
"code": code_text
}
# === Example tasks: simple function to implement ===
TASKS = [
{
"name": "is_prime",
"prompt": "Implement a Python function `is_prime(n: int) -> bool` with proper docstring and edge case handling.",
"test_code": """
import submission
def test_prime():
assert submission.is_prime(2)
assert submission.is_prime(13)
assert not submission.is_prime(1)
assert not submission.is_prime(0)
assert not submission.is_prime(-7)
assert not submission.is_prime(15)
""",
"max_tokens": 256
}
]
# === Runner ===
if __name__ == "__main__":
results = []
for task in TASKS:
for model_fn, name in :
res = run_task(task, model_fn, name)
print(json.dumps(res, indent=2))
results.append(res)
# save to file
with open("results.json", "w") as f:
json.dump(results, f, indent=2)
