初始化项目，由ModelHub XC社区提供模型
Model: KingpinFisk/iol-2026-qwen14b-entry Source: Original Platform
This commit is contained in:
ModelHub XC
318
script.py
Normal file
318
script.py
Normal file
@@ -0,0 +1,318 @@
|
||||
import os
|
||||
import json
|
||||
import pandas as pd
|
||||
import torch
|
||||
import re
|
||||
import time
|
||||
import unicodedata
|
||||
from transformers import AutoTokenizer, AutoModelForCausalLM
|
||||
|
||||
# Offline Environment Setup
|
||||
os.environ["HF_HUB_OFFLINE"] = "1"
|
||||
os.environ["TRANSFORMERS_OFFLINE"] = "1"
|
||||
MODEL_ID = "."
|
||||
|
||||
# Time Budget Guardrail (27.5 minutes max limit)
|
||||
T0 = time.time()
|
||||
DEADLINE = T0 + 1650
|
||||
|
||||
print("Loading tokenizer and 14B AWQ model offline...", flush=True)
|
||||
tok = AutoTokenizer.from_pretrained(MODEL_ID)
|
||||
|
||||
# Force GPU 0 placement ({"": 0})
|
||||
model = AutoModelForCausalLM.from_pretrained(
|
||||
MODEL_ID,
|
||||
torch_dtype=torch.float16,
|
||||
device_map={"": 0}
|
||||
).eval()
|
||||
|
||||
print("Reading mounted test set...", flush=True)
|
||||
df = pd.read_csv("/tmp/data/test.csv", dtype=str).fillna("")
|
||||
|
||||
# 1. Item-Count & Source Text Extraction (From Attached Script)
|
||||
_LINE_NUM = re.compile(r"^[ ]*(\d{1,3})[.)\]]", re.M)
|
||||
_LINE_LETTER = re.compile(r"^[ ]*([A-Z])[.)\]]\s", re.M)
|
||||
|
||||
def detect_n_items(query, context=""):
|
||||
q = query or ""
|
||||
line_nums = [int(m) for m in _LINE_NUM.findall(q)]
|
||||
if line_nums:
|
||||
return len(set(line_nums))
|
||||
lines = [l.strip() for l in q.splitlines() if l.strip()]
|
||||
if len(lines) > 1 and lines[0].endswith((":", ".")):
|
||||
return len(lines) - 1
|
||||
if context:
|
||||
c_nums = len(set(int(m) for m in _LINE_NUM.findall(context)))
|
||||
if c_nums > 1:
|
||||
return c_nums
|
||||
return max(len(lines), 1) if lines else 1
|
||||
|
||||
def extract_item_sources(query, n):
|
||||
q = query or ""
|
||||
out = []
|
||||
for ln in q.splitlines():
|
||||
s = ln.strip()
|
||||
if not s:
|
||||
continue
|
||||
m = re.match(r"^\(?(\d{1,3})\)?[.):\]]\s*(.+)$", s)
|
||||
if m:
|
||||
out.append(m.group(2).strip())
|
||||
if not out:
|
||||
lines = [l.strip() for l in q.splitlines() if l.strip()]
|
||||
if len(lines) > 1 and lines[0].endswith((":", ".")):
|
||||
out = lines[1:]
|
||||
out = [o.split("|")[0].strip() if "|" in o else o for o in out]
|
||||
out = [o for o in out if o]
|
||||
while len(out) < n:
|
||||
out.append(out[-1] if out else "?")
|
||||
return out[:n]
|
||||
|
||||
# 2. Hungarian Assignment Solver for match_letters (From Attached Script)
|
||||
_OPT_LINE = re.compile(r"^[ ]*([A-Za-z])[.)]\s+(.+)$", re.M)
|
||||
_ITEM_LINE = re.compile(r"^[ ]*(\d{1,3})[.)]\s+(.+)$", re.M)
|
||||
|
||||
def parse_matching_block(context):
|
||||
items = [(int(a), b.strip()) for a, b in _ITEM_LINE.findall(context or "")]
|
||||
opts = [(a, b.strip()) for a, b in _OPT_LINE.findall(context or "")]
|
||||
seen = set()
|
||||
items = [x for x in items if not (x[0] in seen or seen.add(x[0]))]
|
||||
seen = set()
|
||||
opts = [x for x in opts if not (x[0] in seen or seen.add(x[0]))]
|
||||
return items, opts
|
||||
|
||||
def best_assignment(score):
|
||||
n, m = len(score), len(score[0])
|
||||
try:
|
||||
from scipy.optimize import linear_sum_assignment
|
||||
import numpy as _np
|
||||
r, c = linear_sum_assignment(-_np.array(score))
|
||||
return list(c)
|
||||
except Exception:
|
||||
pass
|
||||
used, out = set(), [0] * n
|
||||
for i in range(n):
|
||||
j = max((j for j in range(m) if j not in used), key=lambda j: score[i][j], default=0)
|
||||
used.add(j)
|
||||
out[i] = j
|
||||
return out
|
||||
|
||||
def solve_matching(row, n):
|
||||
items, opts = parse_matching_block(row.get("context", ""))
|
||||
if len(items) < 2 or len(opts) < 2 or len(items) != n:
|
||||
return None
|
||||
letters = [o[0] for o in opts]
|
||||
cand_ids = []
|
||||
for L in letters:
|
||||
ids = set()
|
||||
for form in (L, " " + L):
|
||||
t = tok.encode(form, add_special_tokens=False)
|
||||
if t:
|
||||
ids.add(t[0])
|
||||
cand_ids.append(sorted(ids))
|
||||
|
||||
ctx = row["context"].strip()
|
||||
prompts_m = []
|
||||
for num, itext in items:
|
||||
msgs = [
|
||||
{"role": "system", "content": "You match items to their correct counterparts in a linguistics problem. Reply with one option letter only."},
|
||||
{"role": "user", "content": f"{ctx}\n\nWhich lettered option corresponds to item {num} ({itext})? Reply with the option letter only."},
|
||||
]
|
||||
prompts_m.append(tok.apply_chat_template(msgs, tokenize=False, add_generation_prompt=True))
|
||||
|
||||
score = []
|
||||
for p in prompts_m:
|
||||
enc = tok(p, return_tensors="pt", max_length=6144).to(model.device)
|
||||
with torch.no_grad():
|
||||
logits = model(**enc).logits[:, -1, :].float()
|
||||
logprobs = torch.log_softmax(logits, dim=-1)
|
||||
score.append([max(logprobs[0, i].item() for i in ids) for ids in cand_ids])
|
||||
col = best_assignment(score)
|
||||
return [letters[c] for c in col]
|
||||
|
||||
# 3. Task-Routed Prompting (From OUR Script)
|
||||
def get_task_prompt(task_type):
|
||||
base_prompt = (
|
||||
"You are an International Linguistics Olympiad gold medallist solving an unseen language puzzle. "
|
||||
"Strictly deduce all rules from the provided context only. Do not guess based on outside knowledge.\n\n"
|
||||
"Formatting constraints:\n"
|
||||
"<explanation>\nWrite short bullet points summarizing grammar rules and word-by-word literal translations.\n</explanation>\n"
|
||||
"<final_answers>\nAnswer 1\nAnswer 2\n</final_answers>\n"
|
||||
"CRITICAL RULE: DO NOT write question numbers (e.g. '1.', '17.') or bullets in your final answers. Output ONLY the answer text, one item per line."
|
||||
)
|
||||
|
||||
t_type = str(task_type).lower()
|
||||
if "num" in t_type:
|
||||
return base_prompt + (
|
||||
"\n\nSPECIAL TASK INSTRUCTIONS (NUMBERS):\n"
|
||||
"- Identify base system and arithmetic operations.\n"
|
||||
"- Output ONLY numerical digits for numeral queries (e.g. '9' or '11'). Do NOT add periods or spaces.\n"
|
||||
"EXAMPLE OUTPUT FORMAT:\n"
|
||||
"<explanation>\n- mer=1, thef=3.\n- 1+3=4.\n</explanation>\n"
|
||||
"<final_answers>\n4\n9\n</final_answers>"
|
||||
)
|
||||
elif "fill" in t_type:
|
||||
return base_prompt + (
|
||||
"\n\nSPECIAL TASK INSTRUCTIONS (FILL BLANKS):\n"
|
||||
"- Identify prefix/suffix patterns.\n"
|
||||
"- CRITICAL: If replacing `___`, output ONLY the missing letters/suffix that fill the blank, NOT the entire word or stem.\n"
|
||||
"EXAMPLE OUTPUT FORMAT:\n"
|
||||
"<explanation>\n- Suffix '-nyy' indicates plural.\n</explanation>\n"
|
||||
"<final_answers>\nnyy\n</final_answers>"
|
||||
)
|
||||
elif "match" in t_type:
|
||||
return base_prompt + (
|
||||
"\n\nSPECIAL TASK INSTRUCTIONS (MATCHING):\n"
|
||||
"- Match each item to its corresponding option letter or symbol.\n"
|
||||
"- Output ONLY the option letter per item (e.g. 'A', 'G', 'B'). Do NOT write full sentences.\n"
|
||||
"EXAMPLE OUTPUT FORMAT:\n"
|
||||
"<explanation>\n- Item 1 matches option A.\n</explanation>\n"
|
||||
"<final_answers>\nA\nG\n</final_answers>"
|
||||
)
|
||||
else:
|
||||
return base_prompt + (
|
||||
"\n\nSPECIAL TASK INSTRUCTIONS (TRANSLATION):\n"
|
||||
"- Align morphemes and perform word-by-word literal translation in <explanation> before writing final answers.\n"
|
||||
"- CRITICAL TENSE & PUNCTUATION ACCURACY: Check tense prefixes/infixes. Ensure questions end with '?' and statements end with standard punctuation.\n"
|
||||
"EXAMPLE OUTPUT FORMAT:\n"
|
||||
"<explanation>\n- 'nɤ'=you, 'ʒip'=sleep, 'tuʔ'=past.\n- Word-by-word: you sleep past question.\n</explanation>\n"
|
||||
"<final_answers>\nDid you sleep?\nDo I go?\n</final_answers>"
|
||||
)
|
||||
|
||||
# 4. Surgical Non-Destructive Answer Line Cleaner (From OUR Script)
|
||||
def clean_answer_line(line, task_type):
|
||||
raw_stripped = unicodedata.normalize('NFC', line.strip())
|
||||
raw_stripped = raw_stripped.replace(' ', ' ').replace('–', '-').replace('—', '-')
|
||||
raw_stripped = raw_stripped.replace('’', "'").replace('‘', "'").replace('“', '"').replace('”', '"')
|
||||
|
||||
if re.match(r"^(?:\d+[\.\:\)]\s*)?([A-Z])[\.]?$", raw_stripped):
|
||||
return re.sub(r"^(?:\d+[\.\:\)]\s*)?([A-Z])[\.]?$", r"\1", raw_stripped)
|
||||
|
||||
cleaned = re.sub(r"^\s*(?:\(?\d+[\.\:\)]|\(?\w[\.\)]|\-|\*|Item\s*\d+:?)\s*", "", raw_stripped, flags=re.IGNORECASE)
|
||||
cleaned = re.sub(r"[*`]", "", cleaned)
|
||||
cleaned = cleaned.strip('"\' ')
|
||||
cleaned = re.sub(r"\s+", " ", cleaned)
|
||||
|
||||
t_type = str(task_type).lower()
|
||||
if "num" in t_type or "fill" in t_type or "match" in t_type:
|
||||
cleaned = cleaned.rstrip(".")
|
||||
if "num" in t_type and re.search(r"\d+", cleaned):
|
||||
num_match = re.search(r"\b\d+\b", cleaned)
|
||||
if num_match:
|
||||
cleaned = num_match.group(0)
|
||||
else:
|
||||
cleaned = re.sub(r"\s+([\.\?\!\,])", r"\1", cleaned)
|
||||
if any(cleaned.lower().startswith(w) for w in ["do ", "did ", "does ", "is ", "are ", "was ", "were ", "what ", "who ", "where ", "whom "]):
|
||||
if not cleaned.endswith("?"):
|
||||
cleaned = cleaned.rstrip(".!") + "?"
|
||||
|
||||
return cleaned.strip()
|
||||
|
||||
def repair_bijection(answers):
|
||||
if len(answers) < 2 or not all(re.fullmatch(r"[A-Z]", a or "") for a in answers):
|
||||
return answers
|
||||
universe = [chr(ord("A") + i) for i in range(len(answers))]
|
||||
if len(set(answers)) == len(answers):
|
||||
return answers
|
||||
unused = [l for l in universe if l not in set(answers)]
|
||||
seen, out = set(), []
|
||||
for a in answers:
|
||||
if a in seen and unused:
|
||||
out.append(unused.pop(0))
|
||||
else:
|
||||
out.append(a)
|
||||
seen.add(a)
|
||||
return out
|
||||
|
||||
rows = []
|
||||
for idx, r in df.iterrows():
|
||||
if time.time() > DEADLINE:
|
||||
print("Deadline threshold reached. Exiting early to save valid submission.", flush=True)
|
||||
break
|
||||
|
||||
task_type = r.get('task_type', '')
|
||||
query_text = r['query'].strip()
|
||||
context_text = r['context'].strip()
|
||||
|
||||
num_items = detect_n_items(query_text, context_text)
|
||||
src_fallbacks = extract_item_sources(query_text, num_items)
|
||||
|
||||
explanation = "Deducted rules from context."
|
||||
answers = []
|
||||
|
||||
# Matching Tasks: Try Logit-Based Hungarian Assignment Solver First
|
||||
if "match" in str(task_type).lower():
|
||||
try:
|
||||
m_ans = solve_matching(r, num_items)
|
||||
if m_ans and len(m_ans) == num_items:
|
||||
answers = m_ans
|
||||
explanation = "Solved matching via logit assignment solver."
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Standard Path: Run Task-Routed CoT Generation
|
||||
if not answers:
|
||||
system_prompt = get_task_prompt(task_type)
|
||||
user_prompt = f"TASK TYPE: {task_type}\nCONTEXT:\n{context_text}\n\nQUERY:\n{query_text}"
|
||||
|
||||
messages = [
|
||||
{"role": "system", "content": system_prompt},
|
||||
{"role": "user", "content": user_prompt},
|
||||
]
|
||||
|
||||
prompt_text = tok.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)
|
||||
inputs = tok(prompt_text, return_tensors="pt").to(model.device)
|
||||
|
||||
with torch.no_grad():
|
||||
out = model.generate(**inputs, max_new_tokens=1536, do_sample=False, repetition_penalty=1.0)
|
||||
|
||||
input_length = inputs["input_ids"].shape[1]
|
||||
text = tok.decode(out[0][input_length:], skip_special_tokens=True).strip()
|
||||
text = re.sub(r"```(?:text|json)?", "", text).strip()
|
||||
|
||||
exp_match = re.search(r"<explanation>(.*?)</explanation>", text, flags=re.DOTALL | re.IGNORECASE)
|
||||
if exp_match:
|
||||
explanation = exp_match.group(1).strip()
|
||||
elif "<final_answers>" in text:
|
||||
explanation = text.split("<final_answers>")[0].replace("<explanation>", "").replace("</explanation>", "").strip()
|
||||
else:
|
||||
lines = [ln.strip() for ln in text.splitlines() if ln.strip()]
|
||||
explanation = " ".join(lines[:2]) if lines else "Deducted language patterns from context."
|
||||
|
||||
if "<final_answers>" in text:
|
||||
ans_text = text.split("<final_answers>")[-1].replace("</final_answers>", "").strip()
|
||||
raw_lines = [ln.strip() for ln in ans_text.splitlines() if ln.strip()]
|
||||
cleaned_lines = [clean_answer_line(ln, task_type) for ln in raw_lines if clean_answer_line(ln, task_type)]
|
||||
else:
|
||||
lines = [ln.strip() for ln in text.splitlines() if ln.strip()]
|
||||
candidate_answers = [
|
||||
clean_answer_line(ln, task_type) for ln in lines
|
||||
if len(ln) < 150 and not any(k in ln.lower() for k in ["<explanation", "</explanation", "special task", "context:"])
|
||||
]
|
||||
cleaned_lines = [c for c in candidate_answers if c]
|
||||
|
||||
if "match" in str(task_type).lower():
|
||||
cleaned_lines = repair_bijection(cleaned_lines)
|
||||
|
||||
# Bi-directional Alignment with Source-Text Fallback
|
||||
if num_items > 0:
|
||||
if len(cleaned_lines) > num_items:
|
||||
answers = cleaned_lines[:num_items]
|
||||
elif len(cleaned_lines) < num_items:
|
||||
# Fill missing slots with foreign source text fallback to capture partial chrF
|
||||
answers = cleaned_lines + src_fallbacks[len(cleaned_lines):num_items]
|
||||
else:
|
||||
answers = cleaned_lines
|
||||
else:
|
||||
answers = cleaned_lines if cleaned_lines else src_fallbacks
|
||||
|
||||
rows.append({
|
||||
"id": r["id"],
|
||||
"pred": json.dumps(answers, ensure_ascii=False),
|
||||
"explanation": explanation
|
||||
})
|
||||
|
||||
# Save submission.csv incrementally after every row
|
||||
pd.DataFrame(rows).to_csv("submission.csv", index=False, encoding="utf-8-sig")
|
||||
print(f"Processed {len(rows)}/{len(df)} problems in {time.time()-T0:.1f}s", flush=True)
|
||||
|
||||
print("wrote submission.csv successfully!", flush=True)
|
||||
Reference in New Issue
Block a user