Something went wrong. Try again.
Parametric learning in your code harness.
Something went wrong. Try again.
7.8 kB · 206 lines
Python
at letta-code
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207#!/usr/bin/env python3"""Read-only native Tinker baseline evaluation for Project Euler."""from __future__ import annotations
import argparseimport asyncioimport jsonimport loggingimport sysfrom dataclasses import dataclass, fieldfrom datetime import UTC, datetimefrom pathlib import Pathfrom typing import Sequence
import tinker
EXAMPLE_DIR = Path(__file__).resolve().parentsys.path.insert(0, str(EXAMPLE_DIR / "src"))
from euler_common import Problem, problem_map # noqa: E402from euler_tinker import EulerEnv, aggregate_condition_metrics # noqa: E402from tinker_cookbook import model_info, renderers # noqa: E402from 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
@dataclassclass 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()