Something went wrong. Try again.
Parametric learning in your code harness.
Something went wrong. Try again.
7.4 kB · 195 lines
Python
at letta-code
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196#!/usr/bin/env python3from __future__ import annotations
import argparseimport jsonimport reimport shleximport sysfrom pathlib import Path, PurePosixPathfrom typing import Any
EXAMPLE_DIR = Path(__file__).resolve().parentsys.path.insert(0, str(EXAMPLE_DIR / "src"))
from euler_common import ( # noqa: E402 answer_is_correct, extract_final_answer, final_assistant_text, load_jsonl, normalize_answer,)
RUNTIME_EXECUTABLES = { "python": {"python", "python3"}, "javascript": {"node", "bun"}, "julia": {"julia"}, "ruby": {"ruby"},}
def _tool_result_text(result: Any) -> str: if not isinstance(result, dict): return "" content = result.get("content") if not isinstance(content, list): return "" parts: list[str] = [] for part in content: if isinstance(part, dict) and isinstance(part.get("text"), str): parts.append(part["text"]) return "\n".join(parts)
def _text_contains_expected(text: str, expected: str) -> bool: normalized = expected.strip().replace(",", "") if re.fullmatch(r"[-+]?\d+(?:[.]\d+)?", normalized): compact = text.replace(",", "") pattern = rf"(?<![\d.]){re.escape(normalized)}(?!\d|[.]\d)" return re.search(pattern, compact) is not None return any(line.strip().casefold() == normalized.casefold() for line in text.splitlines())
def _normalized_relative_path(value: str) -> str | None: path = PurePosixPath(value) if not value or path.is_absolute() or ".." in path.parts: return None normalized = str(path) return normalized if normalized not in {"", "."} else None
def _direct_program(command: str, language: str) -> str | None: try: argv = shlex.split(command) except ValueError: return None if len(argv) != 2: return None executables = RUNTIME_EXECUTABLES.get(language.lower(), {language.lower()}) if PurePosixPath(argv[0]).name not in executables: return None return _normalized_relative_path(argv[1])
def _successful_exit(event: dict[str, Any]) -> bool: if event.get("isError") is True: return False result = event.get("result") if isinstance(result, dict): details = result.get("details") if isinstance(details, dict): for key in ("exitCode", "exit_code", "returnCode", "return_code", "code"): value = details.get(key) if isinstance(value, int): return value == 0 text = _tool_result_text(result) match = re.search(r"(?im)command exited with code\s+(-?\d+)", text) return match is None or int(match.group(1)) == 0
def _output_is_exact(text: str, expected: str) -> bool: lines = [line.strip() for line in text.splitlines() if line.strip()] if not lines: return False return normalize_answer(lines[-1]) == normalize_answer(expected)
def execution_metrics( events: list[dict[str, Any]], language: str, expected: str) -> dict[str, Any]: program_paths: dict[str, str] = {} mutation_starts: dict[str, tuple[str, str]] = {} successful_ids: set[str] = set() matched_execution_ids: set[str] = set() correct_output_ids: set[str] = set() successful_write_ids: set[str] = set() successful_edit_ids: set[str] = set() successful_mutation_paths: set[str] = set() observed_commands: list[str] = [] for event in events: event_type = event.get("type") tool_name = str(event.get("toolName", "")) call_id = str(event.get("toolCallId", "")) if event_type == "tool_execution_start" and tool_name in {"write", "edit"}: args = event.get("args") value = args.get("path", "") if isinstance(args, dict) else "" path = _normalized_relative_path(value) if isinstance(value, str) else None if path is not None: mutation_starts[call_id] = (tool_name, path) elif event_type == "tool_execution_start" and tool_name == "bash": args = event.get("args") command = args.get("command", "") if isinstance(args, dict) else "" if not isinstance(command, str): continue observed_commands.append(command) program = _direct_program(command, language) if program is not None: program_paths[call_id] = program elif event_type == "tool_execution_end": mutation = mutation_starts.get(call_id) if mutation is not None and event.get("isError") is not True: mutation_tool, mutation_path = mutation successful_mutation_paths.add(mutation_path) if mutation_tool == "write": successful_write_ids.add(call_id) else: successful_edit_ids.add(call_id) if call_id in program_paths and _successful_exit(event): successful_ids.add(call_id) if program_paths[call_id] in successful_mutation_paths: matched_execution_ids.add(call_id) if call_id in matched_execution_ids and _output_is_exact( _tool_result_text(event.get("result")), expected ): correct_output_ids.add(call_id) return { "attempted_execution": bool(program_paths), "successful_execution": bool(successful_ids), "executed_mutated_file": bool(matched_execution_ids), "execution_output_correct": bool(correct_output_ids), "wrote_file": bool(successful_write_ids), "edited_file": bool(successful_edit_ids), "file_mutation_succeeded": bool(successful_write_ids or successful_edit_ids), "commands": observed_commands, "executed_program_paths": sorted( {program_paths[call_id] for call_id in successful_ids} ), "mutated_file_paths": sorted(successful_mutation_paths), }
def score_events(events: list[dict[str, Any]], expected: str, language: str) -> dict[str, Any]: response = final_assistant_text(events) answer, answer_format = extract_final_answer(response) execution = execution_metrics(events, language, expected) exact_answer_correct = answer_is_correct(answer, expected) response_answer_correct = exact_answer_correct or _text_contains_expected(response, expected) functional_correct = ( execution["file_mutation_succeeded"] and execution["executed_mutated_file"] and execution["execution_output_correct"] and response_answer_correct ) return { "response": response, "extracted_answer": answer, "answer_format": answer_format, "format_valid": answer_format == "boxed", "correct": exact_answer_correct, "response_answer_correct": response_answer_correct, "functional_correct": functional_correct, **execution, }
def main() -> None: parser = argparse.ArgumentParser(description="Score one Pi JSON event stream.") parser.add_argument("events", type=Path) parser.add_argument("--answer", required=True) parser.add_argument("--language", required=True) args = parser.parse_args() print(json.dumps(score_events(load_jsonl(args.events), args.answer, args.language), indent=2))
if __name__ == "__main__": main()