初始化项目，由ModelHub XC社区提供模型
Model: Santhoshini/iol-solver-v2 Source: Original Platform
This commit is contained in:
ModelHub XC
846
script.py
Normal file
846
script.py
Normal file
@@ -0,0 +1,846 @@
|
||||
#!/usr/bin/env python3
|
||||
"""IOL-AI 2026 submission: Qwen2.5-14B-Instruct (bnb-4bit via
|
||||
unsloth/Qwen2.5-14B-Instruct-bnb-4bit), offline-only dependency install,
|
||||
a decomposition-and-verification prompt augmented with a deterministic
|
||||
symbolic-evidence layer, item-count-aware token budgeting, a fail-open
|
||||
closed-answer-space constraint for match_letters, and guaranteed
|
||||
explanations.
|
||||
|
||||
History, briefly: every piece below was individually diagnosed against a
|
||||
real failure on real Linguini/IOL problems (a markdown-formatted answer
|
||||
marker, a COMPUTE-line bleeding into the answer list, an "is:" prefix
|
||||
surviving into a near-miss answer, a match_letters bijection violation,
|
||||
truncation on multi-item problems) before being combined here. Nothing in
|
||||
this file is speculative -- every module states the specific failure it
|
||||
closes.
|
||||
|
||||
Compliance: fully offline before any Hugging Face import, MODEL_ID=".",
|
||||
reads only /tmp/data/test.csv, writes only submission.csv with
|
||||
id/pred/explanation, float16 (the T4 is Turing, no native bfloat16), the
|
||||
30-minute budget is respected with a real safety margin, every row is
|
||||
guaranteed a submission.csv entry even under a crash or a timeout.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import atexit
|
||||
import os
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
SCRIPT_STARTED_AT = time.monotonic()
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Offline mode, set before any Hugging Face import. Restored on exit (see
|
||||
# _restore_offline_env_vars below) -- if the evaluation harness ever runs
|
||||
# this script in-process rather than as an isolated subprocess, leftover
|
||||
# offline-mode env vars could otherwise affect a later, unrelated
|
||||
# huggingface_hub call made by the harness itself after this script exits.
|
||||
# ---------------------------------------------------------------------------
|
||||
_ORIGINAL_HF_HUB_OFFLINE = os.environ.get("HF_HUB_OFFLINE")
|
||||
_ORIGINAL_TRANSFORMERS_OFFLINE = os.environ.get("TRANSFORMERS_OFFLINE")
|
||||
|
||||
|
||||
def _restore_offline_env_vars() -> None:
|
||||
"""Restores HF_HUB_OFFLINE/TRANSFORMERS_OFFLINE to their exact
|
||||
pre-script state on exit, via atexit so it fires regardless of how or
|
||||
where the script exits. Costs nothing; cannot make anything worse."""
|
||||
for key, original in (("HF_HUB_OFFLINE", _ORIGINAL_HF_HUB_OFFLINE),
|
||||
("TRANSFORMERS_OFFLINE", _ORIGINAL_TRANSFORMERS_OFFLINE)):
|
||||
if original is None:
|
||||
os.environ.pop(key, None)
|
||||
else:
|
||||
os.environ[key] = original
|
||||
|
||||
|
||||
atexit.register(_restore_offline_env_vars)
|
||||
os.environ["HF_HUB_OFFLINE"] = "1"
|
||||
os.environ["TRANSFORMERS_OFFLINE"] = "1"
|
||||
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Configuration -- env-var overridable, sensible defaults otherwise.
|
||||
# ---------------------------------------------------------------------------
|
||||
SCRIPT_DIR = Path(__file__).resolve().parent
|
||||
INPUT_CSV = Path(os.environ.get("IOL_INPUT", "/tmp/data/test.csv"))
|
||||
OUTPUT_CSV = Path(os.environ.get("IOL_OUTPUT", "submission.csv"))
|
||||
MODEL_ID = os.environ.get("IOL_MODEL_ID", ".")
|
||||
|
||||
TIME_LIMIT_S = float(os.environ.get("IOL_TIME_LIMIT_S", 30 * 60))
|
||||
SETUP_BUFFER_S = float(os.environ.get("IOL_SETUP_BUFFER_S", 420)) # 14B bnb-4bit is ~8-9GB, slow to load
|
||||
EXIT_RESERVE_S = float(os.environ.get("IOL_EXIT_RESERVE_S", 60))
|
||||
|
||||
TOKENS_CAP_FLOOR = int(os.environ.get("IOL_TOKENS_FLOOR", 640))
|
||||
TOKENS_CAP_CEIL = int(os.environ.get("IOL_TOKENS_CEIL", 1536))
|
||||
TOKENS_PER_ITEM = int(os.environ.get("IOL_TOKENS_PER_ITEM", 48))
|
||||
TOKENS_PER_ITEM_BASE = int(os.environ.get("IOL_TOKENS_PER_ITEM_BASE", 256))
|
||||
|
||||
|
||||
def elapsed_seconds() -> float:
|
||||
return time.monotonic() - SCRIPT_STARTED_AT
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Crash safety: two independent write paths, neither depending on the
|
||||
# other, neither depending on anything that might have just failed.
|
||||
# ---------------------------------------------------------------------------
|
||||
def write_submission_csv(rows_list: list[dict]) -> None:
|
||||
"""Stdlib csv, not pandas -- avoids a real, documented pandas/numpy ABI
|
||||
crash ('TypeError: Cannot convert numpy.ndarray to numpy.ndarray' inside
|
||||
pandas' Index construction) that a top-scoring public IOL-AI 2026
|
||||
submission hit in this exact sandbox. Atomic: writes to a temp file
|
||||
then os.replace()s it into place, so a reader can never observe a
|
||||
partially-written file mid-save."""
|
||||
import csv
|
||||
tmp_path = OUTPUT_CSV.with_suffix(OUTPUT_CSV.suffix + ".tmp")
|
||||
with tmp_path.open("w", newline="", encoding="utf-8") as f:
|
||||
writer = csv.DictWriter(f, fieldnames=["id", "pred", "explanation"])
|
||||
writer.writeheader()
|
||||
for row in rows_list:
|
||||
writer.writerow(row)
|
||||
os.replace(tmp_path, OUTPUT_CSV)
|
||||
|
||||
|
||||
def emergency_submission_csv(reason: str, rows_so_far: list[dict] | None = None) -> None:
|
||||
"""Last-resort guarantee: no matter WHERE the script dies, a valid
|
||||
submission.csv exists before the process exits -- the single fix for
|
||||
the pattern where a crash with nothing written turns a scoreable zero
|
||||
into a hard evaluation failure. Independent of write_submission_csv:
|
||||
uses only the standard library, so it cannot fail for the same reason
|
||||
a pandas-based path might."""
|
||||
import csv
|
||||
import json
|
||||
try:
|
||||
if rows_so_far:
|
||||
write_submission_csv(rows_so_far)
|
||||
return
|
||||
ids: list[str] = []
|
||||
try:
|
||||
with INPUT_CSV.open(newline="", encoding="utf-8") as f:
|
||||
for row in csv.DictReader(f):
|
||||
if row.get("id"):
|
||||
ids.append(row["id"])
|
||||
except Exception:
|
||||
pass
|
||||
rows = [{"id": i, "pred": json.dumps([""]),
|
||||
"explanation": f"EMERGENCY FALLBACK: {str(reason)[:150]}"} for i in ids]
|
||||
write_submission_csv(rows)
|
||||
except Exception:
|
||||
try:
|
||||
with OUTPUT_CSV.open("w") as f:
|
||||
f.write("id,pred,explanation\n")
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Offline dependency install.
|
||||
# ---------------------------------------------------------------------------
|
||||
def ensure_dependencies() -> None:
|
||||
"""Split deliberately: torch is NOT force-upgraded (a multi-GB
|
||||
CUDA-specific wheel; forcing -U risks pulling a build mismatched with
|
||||
the sandbox's actual driver -- a worse failure than a missing
|
||||
package). bitsandbytes needs no upgrade evidence behind it.
|
||||
transformers/accelerate/tokenizers have a CONFIRMED version-related
|
||||
failure behind them -- those are the only ones forced."""
|
||||
subprocess.run([sys.executable, "-m", "pip", "install", "-q",
|
||||
"torch>=2.2", "bitsandbytes", "pandas"], check=True)
|
||||
subprocess.run([sys.executable, "-m", "pip", "install", "-q", "-U",
|
||||
"transformers>=4.43", "accelerate>=0.30", "tokenizers"], check=True)
|
||||
|
||||
|
||||
try:
|
||||
ensure_dependencies()
|
||||
except Exception as exc:
|
||||
emergency_submission_csv(f"pip install failed: {exc}")
|
||||
raise
|
||||
|
||||
import re
|
||||
import json
|
||||
import unicodedata
|
||||
import ast as pyast
|
||||
import pandas as pd
|
||||
import torch
|
||||
from difflib import SequenceMatcher
|
||||
from collections import defaultdict
|
||||
from transformers import AutoTokenizer, AutoModelForCausalLM
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Model loading.
|
||||
# ---------------------------------------------------------------------------
|
||||
def load_model():
|
||||
"""Fast tokenizer first; on failure, falls back to use_fast=False --
|
||||
bypasses TokenizerFast.from_file() entirely, which is exactly the call
|
||||
that fails on a tokenizer.json saved by a newer tokenizers library than
|
||||
the sandbox has."""
|
||||
try:
|
||||
tok = AutoTokenizer.from_pretrained(MODEL_ID)
|
||||
print("Tokenizer loaded (fast).", flush=True)
|
||||
except Exception as exc:
|
||||
print(f"Fast tokenizer failed ({exc}); falling back to use_fast=False.", flush=True)
|
||||
tok = AutoTokenizer.from_pretrained(MODEL_ID, use_fast=False)
|
||||
print("Tokenizer loaded (slow fallback).", flush=True)
|
||||
|
||||
model = AutoModelForCausalLM.from_pretrained(
|
||||
MODEL_ID, torch_dtype=torch.float16, device_map="auto",
|
||||
).eval()
|
||||
print(f"Model loaded | memory footprint: {round(model.get_memory_footprint() / 1e9, 1)} GB | "
|
||||
f"quantized: {getattr(model.config, 'quantization_config', None) is not None}", flush=True)
|
||||
return tok, model
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Query parsing: widened patterns + honest "unknown count" fallback.
|
||||
# ---------------------------------------------------------------------------
|
||||
def parse_items(query: str) -> tuple[str, list[str], bool]:
|
||||
"""Returns (preamble, items, count_known). count_known=False means no
|
||||
pattern matched -- we do NOT guess a count, we let the model's own
|
||||
answer list stand rather than risk truncating real content."""
|
||||
item_pat = re.compile(r"(?m)^\s*(\d+)\s*[.\)]\s*(.*)$")
|
||||
matches = list(item_pat.finditer(query))
|
||||
if matches:
|
||||
preamble = query[:matches[0].start()].strip()
|
||||
items = []
|
||||
for i, m in enumerate(matches):
|
||||
end = matches[i + 1].start() if i + 1 < len(matches) else len(query)
|
||||
text = re.sub(r"^\s*\d+\s*[.\)]\s*", "", query[m.start():end].strip())
|
||||
items.append(text)
|
||||
return preamble, items, True
|
||||
|
||||
rng = re.search(r"[\(\[]?\s*(\d+)\s*(?:[-\u2013\u2014:]|to)\s*(\d+)\s*[\)\]]?", query, flags=re.IGNORECASE)
|
||||
if rng:
|
||||
lo, hi = int(rng.group(1)), int(rng.group(2))
|
||||
if 0 < hi - lo < 100:
|
||||
items = []
|
||||
for k in range(lo, hi + 1):
|
||||
line_match = re.search(rf"(?m)^.*\(\s*{k}\s*\).*$", query)
|
||||
if line_match:
|
||||
clue = re.sub(rf"\(\s*{k}\s*\)", "", line_match.group(0)).strip()
|
||||
clue = re.sub(r"\|\s*\|", "|", clue)
|
||||
clue = re.sub(r"\s{2,}", " ", clue).strip(" |")
|
||||
items.append(clue if clue else f"the numbered item {k} from the examples above")
|
||||
else:
|
||||
items.append(f"the numbered item {k} from the examples above")
|
||||
return query.strip(), items, True
|
||||
|
||||
csv_nums = re.findall(r"(?m)^\s*(\d+)\s*,\s*(\d+(?:\s*,\s*\d+)*)\s*$", query)
|
||||
if csv_nums:
|
||||
all_nums = re.findall(r"\d+", " ".join(csv_nums[0]))
|
||||
return query.strip(), [f"the numbered item {n}" for n in all_nums], True
|
||||
|
||||
return query.strip(), [], False
|
||||
|
||||
|
||||
TASK_GUIDANCE = {
|
||||
"translation": "give the translated form only, in the language asked.",
|
||||
"fill_blanks": "give only the missing form for each blank.",
|
||||
"match_letters": "give only the option letter (for example A, B, C).",
|
||||
"text_to_num": "give the number in digits.",
|
||||
"num_to_text": "give the number written out in words, in the language asked.",
|
||||
}
|
||||
DEFAULT_GUIDANCE = "give exactly what the instruction asks, nothing else."
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Symbolic preprocessing layer -- pure standard library, no new
|
||||
# dependencies, deterministic, CPU-only, negligible runtime. Survived a
|
||||
# multi-round falsification pass: only the two evidence objects that (a)
|
||||
# compute something a fast read is likely to miss by construction and (b)
|
||||
# cannot mislead when wrong (worst case is silence, never false
|
||||
# confidence) were kept. Augments the raw context; never replaces it.
|
||||
# ---------------------------------------------------------------------------
|
||||
def extract_forms_from_context(context: str) -> list[str]:
|
||||
"""Pulls candidate unknown-language 'forms' for reduplication's
|
||||
per-word self-check ONLY. Pipe-delimited lines contribute ONLY their
|
||||
FIRST field -- including gloss/meaning fields would let ordinary
|
||||
English words trigger false reduplication hits. Lines with more than 3
|
||||
pipes are skipped defensively -- Hadza (a confirmed IOL 2026 language)
|
||||
is a click language, and '|' is sometimes used informally to
|
||||
transcribe click consonants, which would misparse as our delimiter."""
|
||||
forms = []
|
||||
for line in context.splitlines():
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
pipe_count = line.count("|")
|
||||
if 0 < pipe_count <= 3:
|
||||
first_field = re.sub(r"^\s*\d+\s*[.\)]\s*", "", line.split("|")[0].strip()).strip()
|
||||
if first_field:
|
||||
forms.append(first_field)
|
||||
elif pipe_count == 0:
|
||||
for t in line.split():
|
||||
t_clean = re.sub(r"^\s*\d+\s*[.\)]\s*", "", t).strip(".,;:")
|
||||
if t_clean and len(t_clean) > 1:
|
||||
forms.append(t_clean)
|
||||
seen, unique_forms = set(), []
|
||||
for f in forms:
|
||||
if f not in seen:
|
||||
seen.add(f)
|
||||
unique_forms.append(f)
|
||||
return unique_forms
|
||||
|
||||
|
||||
def extract_explicit_pairs(context: str) -> list[tuple[str, str]]:
|
||||
"""Genuine (input, output) pairs from pipe-delimited rows -- e.g.
|
||||
fill_blanks' 'given | derived | gloss' structure. The ONLY source of
|
||||
pairs fed to transformation-family detection: forms from DIFFERENT
|
||||
rows are never cross-compared, which would manufacture spurious
|
||||
'transformations' between unrelated words."""
|
||||
pairs = []
|
||||
for line in context.splitlines():
|
||||
line = line.strip()
|
||||
if not (0 < line.count("|") <= 3):
|
||||
continue
|
||||
fields = [re.sub(r"^\s*\d+\s*[.\)]\s*", "", f.strip()).strip() for f in line.split("|")]
|
||||
fields = [f for f in fields if f]
|
||||
if len(fields) >= 2:
|
||||
pairs.append((fields[0], fields[1]))
|
||||
return pairs
|
||||
|
||||
|
||||
def edit_signature(a: str, b: str):
|
||||
"""A clean single-region transformation signature, or None if the
|
||||
difference is scattered (too noisy to call one transformation), OR if
|
||||
there is no genuine shared stem of at least 2 characters -- without
|
||||
this check, two totally unrelated words with zero characters in
|
||||
common were being accepted as a fake signature, since SequenceMatcher
|
||||
returns a single 'replace' opcode for a total mismatch too."""
|
||||
sm = SequenceMatcher(None, a, b, autojunk=False)
|
||||
all_ops = sm.get_opcodes()
|
||||
ops = [op for op in all_ops if op[0] != "equal"]
|
||||
if not ops or len(ops) > 2:
|
||||
return None
|
||||
equal_len = sum((i2 - i1) for tag, i1, i2, j1, j2 in all_ops if tag == "equal")
|
||||
if equal_len < 2:
|
||||
return None
|
||||
tag, i1, i2, j1, j2 = ops[0]
|
||||
removed, inserted = a[i1:i2], b[j1:j2]
|
||||
if i1 == 0:
|
||||
pos = "prefix"
|
||||
elif i2 == len(a):
|
||||
pos = "suffix"
|
||||
else:
|
||||
pos = "infix"
|
||||
return (pos, removed, inserted)
|
||||
|
||||
|
||||
def find_transformation_families(pairs: list[tuple[str, str]]) -> list[str]:
|
||||
"""Clusters GENUINELY PAIRED forms (same row only) sharing an
|
||||
identical clean edit signature. Emits a family only if 2+ separate
|
||||
given pairs share it -- one occurrence is worse than silence."""
|
||||
groups = defaultdict(list)
|
||||
for a, b in pairs:
|
||||
if not a or not b or a == b:
|
||||
continue
|
||||
sig = edit_signature(a, b)
|
||||
if sig:
|
||||
groups[sig].append((a, b))
|
||||
|
||||
families = []
|
||||
for sig, grp in groups.items():
|
||||
unique_pairs = list(dict.fromkeys(grp))
|
||||
if len(unique_pairs) >= 2:
|
||||
pos, removed, inserted = sig
|
||||
removed_disp = removed if removed else "(nothing)"
|
||||
inserted_disp = inserted if inserted else "(nothing)"
|
||||
examples = "; ".join(f"{a}->{b}" for a, b in unique_pairs[:4])
|
||||
families.append((len(unique_pairs),
|
||||
f"{pos} change: '{removed_disp}' -> '{inserted_disp}' (seen in: {examples})"))
|
||||
families.sort(key=lambda x: -x[0])
|
||||
return [f for _, f in families]
|
||||
|
||||
|
||||
def detect_reduplication(forms: list[str]) -> list[str]:
|
||||
"""Flags a word only if it contains an exact adjacent doubled
|
||||
substring (length >= 2). Emits nothing if absent."""
|
||||
findings = []
|
||||
for w in forms:
|
||||
n = len(w)
|
||||
found = False
|
||||
for length in range(2, n // 2 + 1):
|
||||
for start in range(0, n - 2 * length + 1):
|
||||
chunk = w[start:start + length]
|
||||
nxt = w[start + length:start + 2 * length]
|
||||
if chunk == nxt:
|
||||
findings.append(f"reduplication in '{w}': '{chunk}' repeated")
|
||||
found = True
|
||||
break
|
||||
if found:
|
||||
break
|
||||
return findings
|
||||
|
||||
|
||||
def build_symbolic_evidence(context: str) -> str:
|
||||
"""Returns "" if no supported transformation family and no
|
||||
reduplication is found -- augments the prompt only with real,
|
||||
multi-supported evidence. Never replaces context."""
|
||||
forms = extract_forms_from_context(context)
|
||||
pairs = extract_explicit_pairs(context)
|
||||
families = find_transformation_families(pairs) if pairs else []
|
||||
redup = detect_reduplication(forms) if forms else []
|
||||
|
||||
lines = []
|
||||
if families:
|
||||
lines.append("Transformation families found (patterns supported by multiple examples):")
|
||||
for f in families[:3]:
|
||||
lines.append(f"- {f}")
|
||||
if redup:
|
||||
lines.append("Reduplication detected:")
|
||||
for r in redup[:2]:
|
||||
lines.append(f"- {r}")
|
||||
if not lines:
|
||||
return ""
|
||||
return ("\n\nSYMBOLIC EVIDENCE (deterministically computed from the examples above; "
|
||||
"may be incomplete -- verify against the examples, do not trust blindly):\n"
|
||||
+ "\n".join(lines))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Closed-answer-space pre-constraint, match_letters only. Deterministic,
|
||||
# read-only, zero generate() calls of its own. Extracts the closed set of
|
||||
# option letters genuinely present in the context (always explicitly
|
||||
# given -- "A. water", "B. child", ...) and states that exact set as a
|
||||
# soft hint in the prompt. FAIL-OPEN: if extraction isn't clean and
|
||||
# unambiguous, the hint is skipped -- baseline behavior for that row is
|
||||
# byte-identical to not having this module at all.
|
||||
# ---------------------------------------------------------------------------
|
||||
def extract_match_letter_options(context: str) -> list[str] | None:
|
||||
"""Returns a sorted list of option letters if extraction is CLEAN and
|
||||
UNAMBIGUOUS, else None. Deliberately strict: must never guess."""
|
||||
found = set()
|
||||
for line in context.splitlines():
|
||||
for m in re.finditer(r"(?:^|\s)([A-Z])[.\)]\s+\S", line):
|
||||
found.add(m.group(1))
|
||||
if not found:
|
||||
return None
|
||||
letters = sorted(found)
|
||||
expected = [chr(ord("A") + i) for i in range(len(letters))]
|
||||
if letters != expected:
|
||||
return None
|
||||
if not (2 <= len(letters) <= 26):
|
||||
return None
|
||||
return letters
|
||||
|
||||
|
||||
def build_messages(context: str, query: str, task_type: str) -> tuple[list[dict], int | None]:
|
||||
"""The proven decomposition-and-verification scaffold, augmented with
|
||||
the symbolic evidence layer and the match_letters closed-option hint.
|
||||
Nothing else about the reasoning instructions has changed since the
|
||||
version that scored 0.083/0.0296/0.2323 on the real leaderboard."""
|
||||
preamble, items, count_known = parse_items(query)
|
||||
guidance = TASK_GUIDANCE.get(task_type, DEFAULT_GUIDANCE)
|
||||
symbolic_evidence = build_symbolic_evidence(context)
|
||||
|
||||
system = (
|
||||
"You solve puzzles about a language you have never seen. Everything you "
|
||||
"need is in the examples below. Use only the examples, not outside "
|
||||
"knowledge of any language. You may meet a task type you have never "
|
||||
"seen -- read the instruction and examples, and answer in the same "
|
||||
"form they use."
|
||||
)
|
||||
number_note = ""
|
||||
if task_type == "text_to_num":
|
||||
number_note = (
|
||||
"\n\nAlso add one more line after your answers, exactly like this:\n"
|
||||
"COMPUTE: expr1 | expr2\n"
|
||||
"where each expr is a plain arithmetic expression (digits, +, -, *, "
|
||||
"parentheses only) for that item's value, one per answer, matching "
|
||||
"the rule you found."
|
||||
)
|
||||
options_note = ""
|
||||
if task_type == "match_letters":
|
||||
options = extract_match_letter_options(context)
|
||||
if options:
|
||||
options_note = (
|
||||
f"\n\nThe only valid answers are: {', '.join(options)}. "
|
||||
f"Do not use any other letter."
|
||||
)
|
||||
|
||||
if count_known:
|
||||
n_items = len(items)
|
||||
slots = "\n\n".join(f"Question {i + 1}: {it}\nAnswer {i + 1}:" for i, it in enumerate(items))
|
||||
user = (
|
||||
f"EXAMPLES:\n{context.strip()}"
|
||||
f"{symbolic_evidence}\n\n"
|
||||
f"--- The examples end here. The questions begin below. ---\n\n"
|
||||
f"For each question: find the rule that explains ALL the examples above "
|
||||
f"(not just one). Check it against every example before answering. "
|
||||
f"For this task type, {guidance}\n\n"
|
||||
f"{preamble}\n\n{slots}\n\n"
|
||||
f"After answering all {n_items} questions, finish with exactly one line, "
|
||||
f"all {n_items} answers in order separated by ' | ':\n"
|
||||
f"FINAL ANSWERS: answer1 | answer2"
|
||||
f"{number_note}"
|
||||
f"{options_note}"
|
||||
)
|
||||
else:
|
||||
n_items = None
|
||||
user = (
|
||||
f"EXAMPLES:\n{context.strip()}"
|
||||
f"{symbolic_evidence}\n\n"
|
||||
f"--- The examples end here. The question begins below. ---\n\n"
|
||||
f"Find the rule that explains ALL the examples above (not just one). "
|
||||
f"Check it against every example before answering. "
|
||||
f"For this task type, {guidance}\n\n"
|
||||
f"{preamble}\n\n"
|
||||
f"Answer every item asked above, in order, one per answer. Finish "
|
||||
f"with exactly one line, all your answers in order separated by ' | ':\n"
|
||||
f"FINAL ANSWERS: answer1 | answer2"
|
||||
f"{number_note}"
|
||||
f"{options_note}"
|
||||
)
|
||||
return [{"role": "system", "content": system}, {"role": "user", "content": user}], n_items
|
||||
|
||||
|
||||
def build_repair_messages(query: str, n_items: int | None, bad_text: str) -> list[dict]:
|
||||
n_desc = f"exactly {n_items}" if n_items is not None else "one per item asked"
|
||||
system = "You reformat answers. Output nothing except the requested line."
|
||||
user = (
|
||||
f"Question:\n{query.strip()}\n\n"
|
||||
f"A previous attempt produced:\n{bad_text[:600]}\n\n"
|
||||
f"Extract or restate {n_desc} final answers, in order, as ONE line:\n"
|
||||
f"FINAL ANSWERS: answer1 | answer2"
|
||||
)
|
||||
return [{"role": "system", "content": system}, {"role": "user", "content": user}]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Safe arithmetic: no exec(), no eval() of arbitrary code.
|
||||
# ---------------------------------------------------------------------------
|
||||
_ALLOWED_BINOPS = (pyast.Add, pyast.Sub, pyast.Mult)
|
||||
|
||||
|
||||
def safe_arithmetic(expr: str) -> float | int | None:
|
||||
try:
|
||||
tree = pyast.parse(expr.strip(), mode="eval")
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
def _eval(node):
|
||||
if isinstance(node, pyast.Expression):
|
||||
return _eval(node.body)
|
||||
if isinstance(node, pyast.Constant) and isinstance(node.value, (int, float)):
|
||||
return node.value
|
||||
if isinstance(node, pyast.BinOp) and isinstance(node.op, _ALLOWED_BINOPS):
|
||||
left, right = _eval(node.left), _eval(node.right)
|
||||
if left is None or right is None:
|
||||
return None
|
||||
if isinstance(node.op, pyast.Add):
|
||||
return left + right
|
||||
if isinstance(node.op, pyast.Sub):
|
||||
return left - right
|
||||
if isinstance(node.op, pyast.Mult):
|
||||
return left * right
|
||||
if isinstance(node, pyast.UnaryOp) and isinstance(node.op, pyast.USub):
|
||||
v = _eval(node.operand)
|
||||
return -v if v is not None else None
|
||||
return None
|
||||
|
||||
return _eval(tree)
|
||||
|
||||
|
||||
def clean_answer(a: str) -> str:
|
||||
"""Broadened: strips "Answer N:", "is:", "the answer is:", "final
|
||||
answer:" prefixes, applies NFC Unicode normalization and collapses
|
||||
internal whitespace runs -- both target exact-match killers the
|
||||
organizers' own documented normalization does not cover (Unicode form,
|
||||
internal whitespace)."""
|
||||
a = re.sub(r"(?i)^\s*(the\s+)?(final\s+)?answer\s*\d*\s*(is)?\s*:\s*", "", a).strip()
|
||||
a = re.sub(r"(?i)^\s*is\s*:\s*", "", a).strip()
|
||||
a = a.strip("* ")
|
||||
a = unicodedata.normalize("NFC", a)
|
||||
a = re.sub(r"\s{2,}", " ", a)
|
||||
return a.strip(" .\"'\u201c\u201d\u2018\u2019")
|
||||
|
||||
|
||||
def extract(text: str) -> tuple[list[str], int | None]:
|
||||
"""Fixed against three real bugs found on real Linguini output:
|
||||
(1) markdown-bold marker with content on the NEXT line, not same line;
|
||||
(2) a following COMPUTE: line bleeding into the answer list;
|
||||
(3) NO marker found + answers dumped on one pipe-separated line --
|
||||
splits each fallback line further by "|" instead of treating the
|
||||
whole line as one answer."""
|
||||
m = list(re.finditer(r"final answers?\s*:?\s*\**", text, flags=re.IGNORECASE))
|
||||
if m:
|
||||
tail = text[m[-1].end():]
|
||||
stop = re.search(r"(?i)compute\s*:", tail)
|
||||
if stop:
|
||||
tail = tail[:stop.start()]
|
||||
tail = tail.replace("**", " ").strip()
|
||||
candidate = " ".join(tail.splitlines())
|
||||
parts = [clean_answer(p) for p in candidate.split("|") if p.strip()]
|
||||
if parts:
|
||||
return parts, m[-1].start()
|
||||
lines = [ln.strip() for ln in text.splitlines() if ln.strip()]
|
||||
fallback = []
|
||||
for ln in lines:
|
||||
ln_clean = re.sub(r"^\s*\d+\s*[.\)]\s*", "", ln)
|
||||
if "|" in ln_clean:
|
||||
fallback.extend(clean_answer(p) for p in ln_clean.split("|") if p.strip())
|
||||
else:
|
||||
fallback.append(clean_answer(ln_clean))
|
||||
return fallback, None
|
||||
|
||||
|
||||
def extract_compute_overrides(text: str, n_answers: int) -> dict[int, str]:
|
||||
m = re.search(r"compute\s*:\s*(.+)", text, flags=re.IGNORECASE)
|
||||
if not m:
|
||||
return {}
|
||||
exprs = [e.strip() for e in m.group(1).split("|")]
|
||||
overrides = {}
|
||||
for i, e in enumerate(exprs[:n_answers]):
|
||||
val = safe_arithmetic(e)
|
||||
if val is not None:
|
||||
overrides[i] = str(int(val)) if float(val).is_integer() else str(val)
|
||||
return overrides
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Generation: defensive against both chat-template return shapes (the
|
||||
# sandbox's transformers version may return a bare tensor from
|
||||
# apply_chat_template rather than a dict), and against a constrained
|
||||
# decoding attempt failing for any reason.
|
||||
# ---------------------------------------------------------------------------
|
||||
def generate(tok, model, messages: list[dict], max_new_tokens: int, constraint_fn=None) -> str:
|
||||
"""constraint_fn: optional prefix_allowed_tokens_fn, default None means
|
||||
byte-identical behavior to an unconstrained call. If a constrained
|
||||
attempt fails for ANY reason, falls back to a fully UNCONSTRAINED
|
||||
generation (not a retry with the same broken kwarg) -- the two
|
||||
concerns (dict-vs-tensor API shape, constrained-vs-unconstrained) are
|
||||
isolated from each other so a failure in one never masks as the
|
||||
other."""
|
||||
def _try_generate(gen_kwargs):
|
||||
try:
|
||||
enc = tok.apply_chat_template(
|
||||
messages, add_generation_prompt=True, return_tensors="pt", return_dict=True,
|
||||
).to(model.device)
|
||||
input_len = enc["input_ids"].shape[-1]
|
||||
with torch.no_grad():
|
||||
out = model.generate(**enc, **gen_kwargs)
|
||||
except Exception:
|
||||
ids = tok.apply_chat_template(
|
||||
messages, add_generation_prompt=True, return_tensors="pt",
|
||||
).to(model.device)
|
||||
input_len = ids.shape[-1]
|
||||
with torch.no_grad():
|
||||
out = model.generate(ids, **gen_kwargs)
|
||||
return out, input_len
|
||||
|
||||
base_kwargs = {"max_new_tokens": max_new_tokens, "do_sample": False}
|
||||
if constraint_fn is not None:
|
||||
try:
|
||||
out, input_len = _try_generate({**base_kwargs, "prefix_allowed_tokens_fn": constraint_fn})
|
||||
except Exception:
|
||||
out, input_len = _try_generate(base_kwargs)
|
||||
else:
|
||||
out, input_len = _try_generate(base_kwargs)
|
||||
return tok.decode(out[0][input_len:], skip_special_tokens=True).strip()
|
||||
|
||||
|
||||
# Adapted from a top-1 public submission's prefix_allowed_tokens_fn
|
||||
# technique -- but scoped correctly for OUR prompt architecture. Their
|
||||
# version applies across an ENTIRE generation because their prompt has no
|
||||
# reasoning phase (plain newline-per-answer output). Ours does have a
|
||||
# reasoning phase (decomposition + "FINAL ANSWERS:" marker); applying a
|
||||
# letter-only constraint there would silently break the model's ability to
|
||||
# reason at all. Scoped here to ONLY the repair call, whose entire
|
||||
# expected output is already a short answer line. Fail-open throughout.
|
||||
_LETTER_CONSTRAINT_CACHE: dict = {}
|
||||
|
||||
|
||||
def build_letter_constraint_fn(tok, valid_letters: list[str]):
|
||||
cache_key = (id(tok), tuple(sorted(valid_letters)))
|
||||
if cache_key in _LETTER_CONSTRAINT_CACHE:
|
||||
return _LETTER_CONSTRAINT_CACHE[cache_key]
|
||||
try:
|
||||
allowed_chars = set(valid_letters) | set(" |\n\t\r")
|
||||
eos = tok.eos_token_id
|
||||
pieces = []
|
||||
for token_id in range(len(tok)):
|
||||
if token_id == eos:
|
||||
continue
|
||||
piece = tok.decode([token_id], skip_special_tokens=False)
|
||||
if piece and all(c in allowed_chars for c in piece):
|
||||
pieces.append(token_id)
|
||||
allowed_ids = ([eos] if eos is not None else []) + pieces
|
||||
|
||||
def allowed(_batch_id, _input_ids):
|
||||
return allowed_ids if allowed_ids else list(range(len(tok)))
|
||||
|
||||
_LETTER_CONSTRAINT_CACHE[cache_key] = allowed
|
||||
return allowed
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
EXPLANATION_SYSTEM = (
|
||||
"Summarize the following reasoning into a few short bullet points: the "
|
||||
"rule or pattern found in the data and the key evidence for the answer. "
|
||||
"Be concise and structured -- do not repeat the full reasoning."
|
||||
)
|
||||
EXPLANATION_FALLBACK = "Answer derived from patterns found in the examples above."
|
||||
|
||||
|
||||
def dynamic_tokens_cap(n_items: int | None, time_based_cap: int) -> int:
|
||||
"""Item-count-aware token budget, evidenced by a top-1 public
|
||||
submission citing truncation on multi-item problems as "a pure
|
||||
unforced loss". Combined with, not replacing, the time-based
|
||||
adaptation via min() -- a multi-item problem gets more room, but never
|
||||
more than time allows."""
|
||||
if not n_items:
|
||||
return time_based_cap
|
||||
item_based_cap = max(TOKENS_CAP_FLOOR, min(TOKENS_CAP_CEIL, n_items * TOKENS_PER_ITEM + TOKENS_PER_ITEM_BASE))
|
||||
return min(time_based_cap, item_based_cap)
|
||||
|
||||
|
||||
def process_row(tok, model, row: dict, n_rows: int, n_done: int, per_row_budget: float) -> tuple[dict, bool]:
|
||||
"""Processes one row. Returns (result_row, ok) -- ok=False means a
|
||||
fallback row was produced after an exception, not a real answer."""
|
||||
try:
|
||||
remaining = TIME_LIMIT_S - elapsed_seconds()
|
||||
budget_left_rows = max(n_rows - n_done, 1)
|
||||
row_budget = remaining / budget_left_rows
|
||||
time_based_cap = 1280 if row_budget > per_row_budget else 640
|
||||
|
||||
task_type = row.get("task_type", "")
|
||||
messages, n_items = build_messages(row["context"], row["query"], task_type)
|
||||
tokens_cap = dynamic_tokens_cap(n_items, time_based_cap)
|
||||
text = generate(tok, model, messages, tokens_cap)
|
||||
answers, marker_pos = extract(text)
|
||||
|
||||
if task_type == "text_to_num":
|
||||
overrides = extract_compute_overrides(text, len(answers))
|
||||
for idx, val in overrides.items():
|
||||
if idx < len(answers):
|
||||
answers[idx] = val
|
||||
|
||||
# Repair only on TRUE extraction failure (no marker / nothing found)
|
||||
# -- not a mere count difference, since extra answers are harmless
|
||||
# and our own count guess may be the thing that's wrong.
|
||||
if (marker_pos is None or not answers) and remaining > SETUP_BUFFER_S:
|
||||
repair_constraint = None
|
||||
if task_type == "match_letters":
|
||||
repair_options = extract_match_letter_options(row["context"])
|
||||
if repair_options:
|
||||
repair_constraint = build_letter_constraint_fn(tok, repair_options)
|
||||
repair_text = generate(tok, model, build_repair_messages(row["query"], n_items, text),
|
||||
128, constraint_fn=repair_constraint)
|
||||
rep, rep_pos = extract(repair_text)
|
||||
if rep:
|
||||
answers, marker_pos = rep, rep_pos
|
||||
|
||||
if n_items is not None:
|
||||
if len(answers) < n_items:
|
||||
answers = answers + [answers[-1] if answers else ""] * (n_items - len(answers))
|
||||
elif len(answers) > n_items and marker_pos is None:
|
||||
answers = answers[:n_items]
|
||||
# else: marker found, more answers than our guess -> keep them all
|
||||
|
||||
if not answers:
|
||||
answers = [""]
|
||||
|
||||
# Explanation: dedicated call if time is comfortable, else a cheap
|
||||
# truncated fallback -- never blank, never a second full generation
|
||||
# under time pressure.
|
||||
remaining_after = TIME_LIMIT_S - elapsed_seconds()
|
||||
budget_left_after = max(n_rows - n_done - 1, 0)
|
||||
comfortable = remaining_after > (budget_left_after + 1) * per_row_budget * 1.3
|
||||
if comfortable:
|
||||
try:
|
||||
explanation = generate(
|
||||
tok, model,
|
||||
[{"role": "system", "content": EXPLANATION_SYSTEM},
|
||||
{"role": "user", "content": text}], 300,
|
||||
) or EXPLANATION_FALLBACK
|
||||
except Exception:
|
||||
explanation = EXPLANATION_FALLBACK
|
||||
else:
|
||||
snippet = re.sub(r"\s{2,}", " ", text[:300]).strip()
|
||||
explanation = snippet if snippet else EXPLANATION_FALLBACK
|
||||
|
||||
return {"id": row["id"], "pred": json.dumps(answers, ensure_ascii=False),
|
||||
"explanation": explanation}, True
|
||||
|
||||
except Exception as exc:
|
||||
try:
|
||||
_, fallback_items, fk = parse_items(row["query"])
|
||||
n_fallback = len(fallback_items) if fk else 1
|
||||
except Exception:
|
||||
n_fallback = 1
|
||||
print(f"ROW ERROR on {row.get('id', '?')}: {exc}", flush=True)
|
||||
return {"id": row["id"], "pred": json.dumps([""] * n_fallback, ensure_ascii=False),
|
||||
"explanation": EXPLANATION_FALLBACK}, False
|
||||
|
||||
|
||||
def main() -> None:
|
||||
if not INPUT_CSV.exists():
|
||||
emergency_submission_csv(f"input CSV not found: {INPUT_CSV}")
|
||||
raise FileNotFoundError(f"Missing input CSV: {INPUT_CSV}")
|
||||
|
||||
try:
|
||||
# Read test.csv and checkpoint a placeholder submission FIRST,
|
||||
# before the slowest and most failure-prone step (model loading,
|
||||
# ~7-9 min for this checkpoint size) even starts -- matching a
|
||||
# top-1 public submission's proven order. If loading hangs or gets
|
||||
# killed, something valid already exists on disk.
|
||||
df = pd.read_csv(INPUT_CSV, dtype=str).fillna("")
|
||||
placeholder_rows = [{"id": rid, "pred": json.dumps([""]),
|
||||
"explanation": "Placeholder written before model load."}
|
||||
for rid in df["id"].tolist()]
|
||||
write_submission_csv(placeholder_rows)
|
||||
print(f"Pre-load checkpoint written for {len(placeholder_rows)} rows.", flush=True)
|
||||
|
||||
tok, model = load_model()
|
||||
except Exception as exc:
|
||||
emergency_submission_csv(f"tokenizer/model load or test.csv read failed: {exc}")
|
||||
raise
|
||||
|
||||
n_rows = len(df)
|
||||
actual_setup_elapsed = elapsed_seconds()
|
||||
per_row_budget = max(20, (TIME_LIMIT_S - actual_setup_elapsed) / max(n_rows, 1))
|
||||
print(f"Setup took {actual_setup_elapsed:.0f}s (estimated {SETUP_BUFFER_S:.0f}s) | "
|
||||
f"per_row_budget={per_row_budget:.0f}s for {n_rows} rows", flush=True)
|
||||
|
||||
rows: list[dict] = []
|
||||
processed_ids: set[str] = set()
|
||||
|
||||
try:
|
||||
for _, row in df.iterrows():
|
||||
result_row, _ok = process_row(tok, model, row, n_rows, len(rows), per_row_budget)
|
||||
rows.append(result_row)
|
||||
processed_ids.add(row["id"])
|
||||
write_submission_csv(rows)
|
||||
print(f"{len(rows)}/{n_rows} elapsed={elapsed_seconds():.0f}s", flush=True)
|
||||
|
||||
if elapsed_seconds() > TIME_LIMIT_S - EXIT_RESERVE_S:
|
||||
print("Time budget nearly exhausted, stopping early.", flush=True)
|
||||
break
|
||||
|
||||
# Guarantee one row per test.csv id, even under a timeout.
|
||||
for _, row in df.iterrows():
|
||||
if row["id"] in processed_ids:
|
||||
continue
|
||||
try:
|
||||
_, fallback_items, fk = parse_items(row["query"])
|
||||
n_fallback = len(fallback_items) if fk else 1
|
||||
except Exception:
|
||||
n_fallback = 1
|
||||
rows.append({"id": row["id"], "pred": json.dumps([""] * n_fallback, ensure_ascii=False),
|
||||
"explanation": EXPLANATION_FALLBACK})
|
||||
|
||||
write_submission_csv(rows)
|
||||
print(f"DONE. Wrote {len(rows)} rows in {elapsed_seconds():.0f}s.", flush=True)
|
||||
|
||||
except Exception as exc:
|
||||
# Final safety net: even if something escapes every inner
|
||||
# try/except above, whatever rows were collected so far still get
|
||||
# written.
|
||||
emergency_submission_csv(f"main loop failed: {exc}", rows_so_far=rows if rows else None)
|
||||
print(f"FATAL, but submission.csv was written with {len(rows)} rows. Error: {exc}", flush=True)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user