Something went wrong. Try again.
Parametric learning in your code harness.
Something went wrong. Try again.
5.3 kB · 132 lines
Python
at letta-code
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133"""Shared Tinker environment for Project Euler evaluation and GRPO."""from __future__ import annotations
from typing import Any
from tinker_cookbook import renderersfrom tinker_cookbook.rl.problem_env import ProblemEnvfrom tinker_cookbook.rl.types import Action, ActionExtra, StepResult
from euler_common import Problem, answer_is_correct, extract_boxed
STRATEGY_INSTRUCTIONS = { "canonical": "Choose the approach you consider most reliable.", "derive": "Derive the mathematics before writing code.", "bruteforce_then_optimize": "Begin with brute force, then optimize it enough to run.", "tests_first": "Check the small examples before computing the full result.", "recurrence": "Look for a recurrence, invariant, or dynamic-programming formulation.", "readable": "Prefer clear and auditable code.",}
LANGUAGE_FENCES = { "python": ("```python",), "javascript": ("```javascript", "```js"), "julia": ("```julia",), "ruby": ("```ruby",), "go": ("```go",), "rust": ("```rust",),}
def build_tinker_question(problem: Problem, language: str, strategy: str) -> str: instruction = STRATEGY_INSTRUCTIONS.get(strategy, strategy) return ( f"Solve Project Euler problem {problem.problem_id}. Use {language}. " f"{instruction} Include the code and finish with the exact answer as " f"\\boxed{{answer}}. Be concise and put the final answer before the token limit.\n\n" f"{problem.prompt}" )
class EulerEnv(ProblemEnv): """Single-turn exact-answer environment shared by baseline eval and GRPO."""
def __init__( self, problem: Problem, language: str, strategy: str, renderer: renderers.Renderer, format_coef: float = 0.1, example_id: str | None = None, ) -> None: super().__init__(renderer, format_coef=format_coef) self.problem = problem self.language = language self.strategy = strategy self.example_id = example_id
def get_question(self) -> str: return build_tinker_question(self.problem, self.language, self.strategy)
def check_answer(self, sample_str: str) -> bool: return answer_is_correct(extract_boxed(sample_str), self.problem.answer)
def check_format(self, sample_str: str) -> bool: lowered = sample_str.lower() return extract_boxed(sample_str) is not None and any( marker in lowered for marker in LANGUAGE_FENCES.get(self.language, ("```",)) )
def get_reference_answer(self) -> str: return self.problem.answer
async def step(self, action: Action, *, extra: ActionExtra | None = None) -> StepResult: result = await super().step(action, extra=extra) stop_reason = (extra or {}).get("stop_reason", "stop") prefix = f"condition/p{self.problem.problem_id:04d}/{self.language}/{self.strategy}" result.metrics.update( { f"{prefix}/attempt": 1.0, f"{prefix}/correct": float(result.metrics.get("correct", 0.0)), f"{prefix}/format": float(result.metrics.get("format", 0.0)), f"{prefix}/truncated": float(stop_reason == "length"), "max_tokens_reached": float(stop_reason == "length"), } ) result.logs.update( { "example_id": self.example_id, "problem_id": self.problem.problem_id, "language": self.language, "strategy": self.strategy, "expected_answer": self.problem.answer, "extracted_answer": extract_boxed(renderers.get_text_content(self.renderer.parse_response(action)[0])), } ) return result
def aggregate_condition_metrics(metrics_list: list[dict[str, float | int]]) -> dict[str, Any]: """Aggregate one-hot condition metrics emitted by :class:`EulerEnv`.""" totals: dict[str, dict[str, float]] = {} for metrics in metrics_list: for key, value in metrics.items(): if not key.startswith("condition/"): continue condition, metric = key.rsplit("/", 1) totals.setdefault(condition, {}).setdefault(metric, 0.0) totals[condition][metric] += float(value) rows: list[dict[str, Any]] = [] for condition, values in sorted(totals.items()): _, problem, language, strategy = condition.split("/", 3) attempts = int(values.get("attempt", 0)) correct = int(values.get("correct", 0)) formatted = int(values.get("format", 0)) truncated = int(values.get("truncated", 0)) rows.append( { "problem_id": int(problem.removeprefix("p")), "language": language, "strategy": strategy, "attempts": attempts, "correct": correct, "exact_rate": correct / attempts if attempts else 0.0, "formatted": formatted, "format_rate": formatted / attempts if attempts else 0.0, "truncated": truncated, "truncation_rate": truncated / attempts if attempts else 0.0, } ) return {"conditions": rows}