#!/usr/bin/env python3 """Read-only native Tinker baseline evaluation for Project Euler.""" from __future__ import annotations import argparse import asyncio import json import logging import sys from dataclasses import dataclass, field from datetime import UTC, datetime from pathlib import Path from typing import Sequence import tinker EXAMPLE_DIR = Path(__file__).resolve().parent sys.path.insert(0, str(EXAMPLE_DIR / "src")) from euler_common import Problem, problem_map # noqa: E402 from euler_tinker import EulerEnv, aggregate_condition_metrics # noqa: E402 from tinker_cookbook import model_info, renderers # noqa: E402 from tinker_cookbook.eval.benchmarks import ( # noqa: E402 BenchmarkBuilder, BenchmarkConfig, BenchmarkResult, run_benchmark, ) from tinker_cookbook.tokenizer_utils import get_tokenizer # noqa: E402 def parse_problem_ids(spec: str) -> list[int]: """Parse comma-separated IDs and inclusive ranges such as ``1-5,8``.""" values: set[int] = set() for part in spec.split(","): item = part.strip() if not item: continue if "-" in item: start_text, end_text = item.split("-", 1) start, end = int(start_text), int(end_text) if start <= 0 or end < start: raise ValueError(f"invalid problem range: {item}") values.update(range(start, end + 1)) else: value = int(item) if value <= 0: raise ValueError(f"invalid problem id: {item}") values.add(value) if not values: raise ValueError("at least one problem id is required") return sorted(values) def comma_values(spec: str) -> list[str]: values = [value.strip() for value in spec.split(",") if value.strip()] if not values: raise ValueError("at least one value is required") return values @dataclass class EulerBenchmark(BenchmarkBuilder): problems: list[Problem] languages: list[str] strategies: list[str] name: str = "project_euler" recommended_timeout: float = 900 condition_summary: dict[str, object] = field(default_factory=lambda: {"conditions": []}) def make_envs( self, renderer: renderers.Renderer, _config: BenchmarkConfig, ) -> Sequence[EulerEnv]: return [ EulerEnv( problem, language, strategy, renderer, example_id=f"p{problem.problem_id:04d}-{language}-{strategy}", ) for problem in self.problems for language in self.languages for strategy in self.strategies ] def aggregate( self, rewards: list[float], metrics_list: list[dict[str, float | int]], ) -> BenchmarkResult: correct = sum(int(metrics.get("correct", 0)) for metrics in metrics_list) result = BenchmarkResult( name=self.name, score=correct / len(rewards) if rewards else 0.0, num_examples=len(rewards), num_correct=correct, metrics={ "format_rate": ( sum(float(metrics.get("format", 0)) for metrics in metrics_list) / len(rewards) if rewards else 0.0 ), }, ) self.condition_summary = aggregate_condition_metrics(metrics_list) return result def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--problems-file", type=Path, default=EXAMPLE_DIR / "data" / "problems.jsonl") parser.add_argument("--problems", default="1", help="IDs/ranges, for example 1-10,15") parser.add_argument("--languages", default="python,javascript,julia") parser.add_argument("--strategies", default="canonical") parser.add_argument("--fanout", type=int, default=2, help="Independent samples per condition") parser.add_argument("--model", default="Qwen/Qwen3.5-4B") parser.add_argument("--model-path", help="Optional tinker:// sampler checkpoint instead of the base model") parser.add_argument("--renderer", help="Defaults to the Cookbook recommendation for --model") parser.add_argument("--max-tokens", dest="max_output", type=int, default=4096) parser.add_argument("--context-window", type=int, default=65536) parser.add_argument("--temperature", type=float, default=1.0) parser.add_argument("--concurrency", type=int, default=16) parser.add_argument("--timeout", type=float, default=900) parser.add_argument("--output-dir", type=Path) return parser async def run(args: argparse.Namespace) -> dict[str, object]: if args.fanout <= 0: raise ValueError("--fanout must be positive") problem_ids = parse_problem_ids(args.problems) available = problem_map(args.problems_file.resolve()) problems: list[Problem] = [] for problem_id in problem_ids: problem = available.get(problem_id) if problem is None: raise ValueError(f"problem {problem_id} is absent from {args.problems_file}") if not problem.available: raise ValueError(f"problem {problem_id} is unavailable because resources are missing") problems.append(problem) languages = comma_values(args.languages) strategies = comma_values(args.strategies) renderer_name = args.renderer or model_info.get_recommended_renderer_name(args.model) renderer = renderers.get_renderer(renderer_name, get_tokenizer(args.model)) service_client = tinker.ServiceClient() sampling_client = service_client.create_sampling_client( model_path=args.model_path, base_model=None if args.model_path else args.model, ) timestamp = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ") output_dir = ( args.output_dir or EXAMPLE_DIR / "runs" / f"tinker-baseline-p{problem_ids[0]:04d}-p{problem_ids[-1]:04d}-{timestamp}" ).resolve() output_dir.mkdir(parents=True, exist_ok=True) benchmark = EulerBenchmark(problems, languages, strategies) generation_options = { "max_tokens": args.max_output, "max_generation_tokens": args.max_output, "max_trajectory_tokens": args.context_window, } config = BenchmarkConfig( concurrency=args.concurrency, timeout_seconds=args.timeout, context_window=args.context_window, temperature=args.temperature, num_samples=args.fanout, save_dir=str(output_dir), **generation_options, ) logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") result = await run_benchmark(benchmark, sampling_client, renderer, config) condition_summary = benchmark.condition_summary summary: dict[str, object] = { "model": args.model, "model_path": args.model_path, "renderer": renderer_name, "problems": problem_ids, "languages": languages, "strategies": strategies, "fanout": args.fanout, "max_tokens": args.max_output, "temperature": args.temperature, "score": result.score, "num_examples": result.num_examples, "num_correct": result.num_correct, "num_errors": result.num_errors, "num_truncated": result.num_truncated, "pass_at_k": result.pass_at_k, "metrics": result.metrics, **condition_summary, } (output_dir / "summary.json").write_text(json.dumps(summary, indent=2) + "\n", encoding="utf-8") print(json.dumps(summary, indent=2)) return summary def main() -> None: asyncio.run(run(build_parser().parse_args())) if __name__ == "__main__": main()