Something went wrong. Try again.
Parametric learning in your code harness.
Something went wrong. Try again.
6.2 kB · 180 lines
Python
at letta-code
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181from __future__ import annotations
import hashlibimport jsonfrom collections.abc import Callablefrom dataclasses import dataclassfrom pathlib import Path
LANGUAGES = ("python", "javascript", "julia")EXTENSIONS = {"python": "py", "javascript": "js", "julia": "jl"}
@dataclass(frozen=True)class TransferTask: task_id: str instruction: str answer: str common_resources: dict[str, str] language_resources: Callable[[str], dict[str, str]] | None = None
def resources(self, language: str) -> dict[str, str]: if language not in LANGUAGES: raise ValueError(f"unsupported transfer-task language: {language}") resources = dict(self.common_resources) if self.language_resources is not None: resources.update(self.language_resources(language)) return resources
def _event_log_task() -> TransferTask: rows = [ "# timestamp|service|status|latency_ms", "2026-08-01T10:00:00Z|atlas|ok|41", "2026-08-01T10:01:00Z|orion|retry|17", "2026-08-01T10:02:00Z|atlas|retry|23", "malformed row", "2026-08-01T10:03:00Z|borealis|retry|31", "2026-08-01T10:04:00Z|orion|ok|11", "2026-08-01T10:05:00Z|atlas|retry|29", "2026-08-01T10:06:00Z|orion|retry|19", "2026-08-01T10:07:00Z|atlas|retry|13", ] answer = 23 + 29 + 13 + 17 + 19 return TransferTask( task_id="event-log-retries", instruction=( "Read `events.log`. Ignore blank, comment, and malformed rows. Sum `latency_ms` " "for rows whose status is `retry` and whose service is either `atlas` or `orion`." ), answer=str(answer), common_resources={"events.log": "\n".join(rows) + "\n"}, )
def _inventory_task() -> TransferTask: rows = [ "sku,warehouse,quantity,unit_cents,state", "A10,west,4,1250,active", "B20,east,7,830,active", "C30,central,3,2100,active", "D40,west,-1,950,active", "E50,central,8,415,hold", "F60,west,6,775,active", "G70,central,2,1995,active", ] answer = 4 * 1250 + 3 * 2100 - 950 + 6 * 775 + 2 * 1995 return TransferTask( task_id="inventory-value", instruction=( "Read `inventory.csv`. For rows in state `active` whose warehouse is `west` " "or `central`, sum `quantity * unit_cents`. Quantities are signed; include returns." ), answer=str(answer), common_resources={"inventory.csv": "\n".join(rows) + "\n"}, )
def _repair_resources(language: str) -> dict[str, str]: numbers = [7, -2, 11, 5, 3, -4] values = "\n".join(str(value) for value in numbers) + "\n" if language == "python": starter = ( "numbers = [int(line) for line in open('numbers.txt') if line.strip()]\n" "total = sum((index + 3) * value for index, value in enumerate(numbers))\n" "print(total)\n" ) elif language == "javascript": starter = ( "const fs = require('fs');\n" "const numbers = fs.readFileSync('numbers.txt', 'utf8').trim().split(/\\s+/).map(Number);\n" "let total = 0;\n" "for (let index = 0; index < numbers.length; index++) total += (index + 3) * numbers[index];\n" "console.log(total);\n" ) else: starter = ( "numbers = parse.(Int, filter(!isempty, split(read(\"numbers.txt\", String), '\\n')))\n" "total = sum((index + 2) * value for (index, value) in enumerate(numbers))\n" "println(total)\n" ) return { "numbers.txt": values, f"broken.{EXTENSIONS[language]}": starter, }
def _repair_task() -> TransferTask: numbers = [7, -2, 11, 5, 3, -4] answer = sum((index + 3) * value for index, value in enumerate(numbers, start=1)) return TransferTask( task_id="repair-weighted-checksum", instruction=( "The provided `broken.<ext>` program is intended to read `numbers.txt` and print " "the one-based weighted checksum `sum((index + 3) * value)`. It has an off-by-one " "bug. Inspect it, repair it in place, and execute the repaired program." ), answer=str(answer), common_resources={}, language_resources=_repair_resources, )
def _shards_task() -> TransferTask: shards = { "shards/alpha.txt": "4\n19\n-3\n8\n", "shards/bravo.txt": "11\n11\n6\n27\n", "shards/charlie.txt": "-8\n2\n14\n5\n", } parsed = [[int(line) for line in content.splitlines()] for content in shards.values()] answer = sum(max(values) - min(values) for values in parsed) return TransferTask( task_id="sharded-ranges", instruction=( "Read every `.txt` file under `shards/`. For each file, compute `maximum - minimum`, " "then sum those per-file ranges." ), answer=str(answer), common_resources=shards, )
def build_transfer_tasks() -> list[TransferTask]: return [ _event_log_task(), _inventory_task(), _repair_task(), _shards_task(), ]
def transfer_task_digest(tasks: list[TransferTask]) -> str: payload = [] for task in tasks: payload.append( { "task_id": task.task_id, "instruction": task.instruction, "answer": task.answer, "resources": { language: task.resources(language) for language in LANGUAGES }, } ) encoded = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode() return hashlib.sha256(encoded).hexdigest()
def materialize_task_resources( task: TransferTask, language: str, output_dir: Path,) -> tuple[str, ...]: refs: list[str] = [] for relative, content in sorted(task.resources(language).items()): ref = Path(relative) destination = output_dir / ref destination.parent.mkdir(parents=True, exist_ok=True) destination.write_text(content, encoding="utf-8") refs.append(str(ref)) return tuple(refs)