初始化项目，由ModelHub XC社区提供模型
Model: JuliaKreutzerCohere/tiny-aya-global-prompt-userdetail Source: Original Platform
This commit is contained in:
ModelHub XC
291
script.py
Normal file
291
script.py
Normal file
@@ -0,0 +1,291 @@
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
|
||||
def _install_bundled_deps() -> None:
|
||||
"""Install transformers from bundled wheels (eval sandbox has no PyPI access)."""
|
||||
wheels_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "wheels")
|
||||
if not os.path.isdir(wheels_dir):
|
||||
return
|
||||
subprocess.run(
|
||||
[
|
||||
sys.executable,
|
||||
"-m",
|
||||
"pip",
|
||||
"install",
|
||||
"-q",
|
||||
"--no-index",
|
||||
f"--find-links={wheels_dir}",
|
||||
"transformers==4.56.2",
|
||||
],
|
||||
check=True,
|
||||
)
|
||||
|
||||
|
||||
_install_bundled_deps()
|
||||
|
||||
import re
|
||||
import csv
|
||||
import json
|
||||
import shutil
|
||||
import tempfile
|
||||
import torch
|
||||
from transformers import AutoTokenizer, AutoModelForCausalLM
|
||||
|
||||
# The repo is the working directory at run time, and there is no network.
|
||||
os.environ["HF_HUB_OFFLINE"] = "1"
|
||||
os.environ["TRANSFORMERS_OFFLINE"] = "1"
|
||||
MODEL_ID = "."
|
||||
MAX_NEW_TOKENS = 2000
|
||||
TEMPERATURE = 0.8
|
||||
TOP_P = 0.95
|
||||
MAX_ATTEMPTS = 5
|
||||
|
||||
|
||||
def load_tokenizer(model_id: str = "."):
|
||||
"""Load tokenizer, converting tokenizer.json for older tokenizers if needed."""
|
||||
tokenizer_path = os.path.join(model_id, "tokenizer.json")
|
||||
with open(tokenizer_path, encoding="utf-8") as handle:
|
||||
data = json.load(handle)
|
||||
|
||||
merges = data.get("model", {}).get("merges", [])
|
||||
if not merges or not isinstance(merges[0], list):
|
||||
return AutoTokenizer.from_pretrained(model_id)
|
||||
|
||||
# Older tokenizers expect merge pairs as "a b" strings, not ["a", "b"] lists.
|
||||
data["model"]["merges"] = [" ".join(piece) for piece in merges]
|
||||
tmpdir = tempfile.mkdtemp()
|
||||
for name in ("tokenizer_config.json", "special_tokens_map.json"):
|
||||
src = os.path.join(model_id, name)
|
||||
if os.path.isfile(src):
|
||||
shutil.copy(src, tmpdir)
|
||||
with open(os.path.join(tmpdir, "tokenizer.json"), "w", encoding="utf-8") as handle:
|
||||
json.dump(data, handle)
|
||||
return AutoTokenizer.from_pretrained(tmpdir)
|
||||
|
||||
|
||||
# Short durable contract only — procedure lives in the user turn with the problem.
|
||||
SYSTEM = (
|
||||
"You solve International Linguistics Olympiad (IOL) problems using only the "
|
||||
"CONTEXT and QUERY you are given. Do not rely on prior knowledge of the language.\n"
|
||||
"Always end with a line that says exactly `FINAL ANSWERS:`, then one bare answer "
|
||||
"per line for each QUERY item, in order — no numbering, quotes, or extra text."
|
||||
)
|
||||
|
||||
TASK_TYPE_HINTS = {
|
||||
"translation": (
|
||||
"return the translated form only, in the language the task asks for"
|
||||
),
|
||||
"fill_blanks": (
|
||||
"return only the missing form for each indicated blank "
|
||||
"(a word, part of a word, or phonetic transcription — match what CONTEXT uses)"
|
||||
),
|
||||
"match_letters": "return only the option letter (for example A, B, C)",
|
||||
"text_to_num": "return the number in digits",
|
||||
"num_to_text": "return the number written out in words, in the language asked",
|
||||
}
|
||||
|
||||
|
||||
def task_type_hint(task_type: str) -> str:
|
||||
return TASK_TYPE_HINTS.get(
|
||||
task_type,
|
||||
"return exactly what the instruction in QUERY asks for, nothing else",
|
||||
)
|
||||
|
||||
|
||||
def build_user_prompt(context: str, task_type: str, query: str) -> str:
|
||||
"""Detailed, instance-local instructions after the problem data (recency)."""
|
||||
return (
|
||||
f"CONTEXT:\n{context.strip()}\n\n"
|
||||
f"TASK TYPE: `{task_type}`\n\n"
|
||||
f"QUERY:\n{query.strip()}\n\n"
|
||||
"Instructions for this problem:\n"
|
||||
"1. Deduce the linguistic rules from the CONTEXT examples only.\n"
|
||||
"2. Apply those rules to every item in QUERY.\n"
|
||||
"3. Check the answer format requirements before finishing.\n"
|
||||
f"4. For this TASK TYPE (`{task_type}`): {task_type_hint(task_type)}.\n"
|
||||
"5. Draft answers, then verify they are complete (one answer for each QUERY "
|
||||
"item) and consistent with the rules and format.\n"
|
||||
"6. If needed, correct and refine.\n\n"
|
||||
"Finally, write a line that says exactly `FINAL ANSWERS:` and, below it, "
|
||||
"the answers to the QUERY items only (not CONTEXT), one answer per line "
|
||||
"in QUERY order — bare answers only, no numbering, no quotes, no extra text."
|
||||
)
|
||||
|
||||
|
||||
# Prefer a dedicated header line; also allow same-line answers after the colon.
|
||||
FINAL_ANSWERS_LINE_RE = re.compile(
|
||||
r"(?im)^[^\w\n]*final answers?[^\w\n]*:?[ \t]*(?=\n|$)|"
|
||||
r"(?im)^[^\w\n]*final answers?\s*:\s*"
|
||||
)
|
||||
FINAL_ANSWERS_INLINE_RE = re.compile(
|
||||
r"(?is)\bfinal answers?\s*:\s*"
|
||||
)
|
||||
|
||||
|
||||
def extract_raw_final(text: str) -> str:
|
||||
"""Return text after the last final-answers marker, or '' if none found."""
|
||||
line_matches = list(FINAL_ANSWERS_LINE_RE.finditer(text))
|
||||
if line_matches:
|
||||
return text[line_matches[-1].end() :]
|
||||
|
||||
inline_matches = list(FINAL_ANSWERS_INLINE_RE.finditer(text))
|
||||
if inline_matches:
|
||||
return text[inline_matches[-1].end() :]
|
||||
|
||||
return ""
|
||||
|
||||
|
||||
def expected_answer_count(query: str, task_type: str) -> int:
|
||||
if task_type == "match_letters":
|
||||
numbered = re.findall(r"^\s*\d+\.", query, re.MULTILINE)
|
||||
return len(numbered) or 1
|
||||
|
||||
if "blanks" in query.lower():
|
||||
range_match = re.search(r"\((\d+)-(\d+)\)", query)
|
||||
if range_match:
|
||||
return int(range_match.group(2)) - int(range_match.group(1)) + 1
|
||||
return len(re.findall(r"\(\d+\)", query)) or 1
|
||||
|
||||
numbered = re.findall(r"^\s*\d+[.)]", query, re.MULTILINE)
|
||||
return len(numbered) or 1
|
||||
|
||||
|
||||
def split_single_line_answer(text: str, expected: int, task_type: str) -> list[str]:
|
||||
text = text.strip()
|
||||
if expected <= 1:
|
||||
return [text]
|
||||
|
||||
def try_split(pattern: str) -> list[str] | None:
|
||||
parts = [part.strip() for part in re.split(pattern, text) if part.strip()]
|
||||
return parts if len(parts) == expected else None
|
||||
|
||||
if task_type == "match_letters":
|
||||
for pattern in (r"\s+", r",\s*", r";\s*"):
|
||||
if result := try_split(pattern):
|
||||
return result
|
||||
letters = re.findall(r"[A-Za-z]", text)
|
||||
if len(letters) == expected:
|
||||
return [letter.upper() for letter in letters]
|
||||
return [text]
|
||||
|
||||
if task_type in ("text_to_num", "num_to_text"):
|
||||
for pattern in (r",\s*", r";\s*", r"\s+"):
|
||||
if result := try_split(pattern):
|
||||
return result
|
||||
return [text]
|
||||
|
||||
for pattern in (r";\s*", r",\s*"):
|
||||
if result := try_split(pattern):
|
||||
return result
|
||||
return [text]
|
||||
|
||||
|
||||
def parse_answer_lines(text_after_marker: str, query: str, task_type: str) -> list[str]:
|
||||
"""Parse cleaned answer lines from the raw final-answers section."""
|
||||
answers = []
|
||||
for line in text_after_marker.splitlines():
|
||||
stripped_line = line.strip("`").strip()
|
||||
if stripped_line == "":
|
||||
continue
|
||||
|
||||
match_numbered_prefix = re.match(r"^\s*\d+[.)]\s+(.*)", stripped_line)
|
||||
if match_numbered_prefix:
|
||||
cleaned_line = match_numbered_prefix.group(1).strip()
|
||||
else:
|
||||
cleaned_line = stripped_line
|
||||
|
||||
cleaned_line = re.sub(r"\*\*", "", cleaned_line).strip()
|
||||
|
||||
if task_type == "match_letters":
|
||||
parts = [
|
||||
part.strip("().[]")
|
||||
for part in re.split(r"[\s,;]+", cleaned_line)
|
||||
if part.strip()
|
||||
]
|
||||
if not (
|
||||
len(parts) > 1
|
||||
and all(re.fullmatch(r"[A-Za-z]", part) for part in parts)
|
||||
):
|
||||
match_letter_word = re.match(
|
||||
r"^\s*(?:\(([A-Za-z])\)|\[([A-Za-z])\]|([A-Za-z]))\.?:?\s*(.*)$",
|
||||
cleaned_line,
|
||||
)
|
||||
if match_letter_word:
|
||||
letter = (
|
||||
match_letter_word.group(1)
|
||||
or match_letter_word.group(2)
|
||||
or match_letter_word.group(3)
|
||||
)
|
||||
cleaned_line = letter.upper()
|
||||
|
||||
if cleaned_line:
|
||||
answers.append(cleaned_line)
|
||||
|
||||
expected = expected_answer_count(query, task_type)
|
||||
if len(answers) == 1 and expected > 1:
|
||||
answers = split_single_line_answer(answers[0], expected, task_type)
|
||||
return answers
|
||||
|
||||
|
||||
def postprocess_answer(text, query, task_type):
|
||||
"""Keep only the content after the last 'FINAL ANSWERS' marker."""
|
||||
text_after_marker = extract_raw_final(text)
|
||||
if not text_after_marker.strip():
|
||||
return []
|
||||
return parse_answer_lines(text_after_marker, query, task_type)
|
||||
|
||||
|
||||
tok = load_tokenizer(MODEL_ID)
|
||||
model = AutoModelForCausalLM.from_pretrained(
|
||||
MODEL_ID, torch_dtype=torch.float16, device_map="auto"
|
||||
).eval()
|
||||
|
||||
with open("/tmp/data/test.csv", encoding="utf-8", newline="") as f:
|
||||
test_rows = list(csv.DictReader(f))
|
||||
|
||||
outputs_queries_types = []
|
||||
for r in test_rows:
|
||||
messages = [
|
||||
{"role": "system", "content": SYSTEM},
|
||||
{
|
||||
"role": "user",
|
||||
"content": build_user_prompt(r["context"], r["task_type"], r["query"]),
|
||||
},
|
||||
]
|
||||
ids = tok.apply_chat_template(
|
||||
messages, add_generation_prompt=True, return_tensors="pt",
|
||||
).to(model.device)
|
||||
|
||||
done = False
|
||||
attempts = 0
|
||||
text = ""
|
||||
while not done and attempts < MAX_ATTEMPTS:
|
||||
with torch.no_grad():
|
||||
out = model.generate(
|
||||
ids,
|
||||
max_new_tokens=MAX_NEW_TOKENS,
|
||||
do_sample=True,
|
||||
temperature=TEMPERATURE,
|
||||
top_p=TOP_P,
|
||||
)
|
||||
text = tok.decode(out[0][ids.shape[-1] :], skip_special_tokens=True).strip()
|
||||
attempts += 1
|
||||
if extract_raw_final(text).strip():
|
||||
done = True
|
||||
else:
|
||||
print(f"TRYING AGAIN...attempts #{attempts}/{MAX_ATTEMPTS}", flush=True)
|
||||
|
||||
outputs_queries_types.append((text, r["id"], r["query"], r["task_type"]))
|
||||
print(f"{len(outputs_queries_types)}/{len(test_rows)} done", flush=True)
|
||||
|
||||
rows = []
|
||||
for answer, row_id, query, task_type in outputs_queries_types:
|
||||
answers = postprocess_answer(answer, query, task_type)
|
||||
rows.append({"id": row_id, "pred": json.dumps(answers, ensure_ascii=False)})
|
||||
with open("submission.csv", "w", encoding="utf-8", newline="") as f:
|
||||
writer = csv.DictWriter(f, fieldnames=["id", "pred"])
|
||||
writer.writeheader()
|
||||
writer.writerows(rows)
|
||||
print("wrote submission.csv", flush=True)
|
||||
Reference in New Issue
Block a user