初始化项目，由ModelHub XC社区提供模型
Model: JuliaKreutzerCohere/tiny-aya-global-prompt-userdetail-splitpass Source: Original Platform
This commit is contained in:
ModelHub XC
394
script.py
Normal file
394
script.py
Normal file
@@ -0,0 +1,394 @@
|
||||
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_REASON = 1200
|
||||
MAX_NEW_TOKENS_ANSWER = 256
|
||||
MAX_NEW_TOKENS_EXPLAIN = 600
|
||||
TEMPERATURE = 0.8
|
||||
TOP_P = 0.95
|
||||
MAX_ATTEMPTS = 3
|
||||
MAX_REASONING_CHARS = 3500 # keep answer-pass context focused
|
||||
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
|
||||
PROMPTS_DIR = os.path.join(SCRIPT_DIR, "prompts")
|
||||
|
||||
KNOWN_TASK_TYPES = (
|
||||
"translation",
|
||||
"fill_blanks",
|
||||
"match_letters",
|
||||
"text_to_num",
|
||||
"num_to_text",
|
||||
)
|
||||
|
||||
# Short pass-role contracts; task-specific procedure is loaded into the user turn.
|
||||
SYSTEM_REASON = (
|
||||
"You solve International Linguistics Olympiad problems using only the given "
|
||||
"CONTEXT and QUERY. This turn is REASONING ONLY — do not write `FINAL ANSWERS:`."
|
||||
)
|
||||
SYSTEM_ANSWER = (
|
||||
"You emit only final answers for International Linguistics Olympiad problems. "
|
||||
"Start with a line that says exactly `FINAL ANSWERS:`, then exactly the requested "
|
||||
"number of bare answer lines — no reasoning, numbering, quotes, or glosses."
|
||||
)
|
||||
SYSTEM_EXPLAIN = (
|
||||
"You explain International Linguistics Olympiad solutions concisely. "
|
||||
"Output only the explanation — no final answers."
|
||||
)
|
||||
|
||||
|
||||
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)
|
||||
|
||||
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)
|
||||
|
||||
|
||||
def _prompt_path(pass_name: str, task_type: str) -> str:
|
||||
"""Resolve prompts/<pass>/<task_type>.txt, falling back to default.txt."""
|
||||
safe = os.path.basename(task_type.strip())
|
||||
if pass_name == "explain":
|
||||
path = os.path.join(PROMPTS_DIR, "explain.txt")
|
||||
if os.path.isfile(path):
|
||||
return path
|
||||
raise FileNotFoundError(f"Missing explain prompt at {path}")
|
||||
|
||||
candidate = os.path.join(PROMPTS_DIR, pass_name, f"{safe}.txt")
|
||||
if safe in KNOWN_TASK_TYPES and os.path.isfile(candidate):
|
||||
return candidate
|
||||
default = os.path.join(PROMPTS_DIR, pass_name, "default.txt")
|
||||
if os.path.isfile(default):
|
||||
return default
|
||||
raise FileNotFoundError(
|
||||
f"No {pass_name} prompt for task_type={task_type!r} under {PROMPTS_DIR}"
|
||||
)
|
||||
|
||||
|
||||
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 load_user_prompt(
|
||||
pass_name: str,
|
||||
context: str,
|
||||
task_type: str,
|
||||
query: str,
|
||||
reasoning: str = "",
|
||||
n_answers: int | None = None,
|
||||
) -> str:
|
||||
with open(_prompt_path(pass_name, task_type), encoding="utf-8") as handle:
|
||||
template = handle.read()
|
||||
if n_answers is None:
|
||||
n_answers = expected_answer_count(query, task_type)
|
||||
values = {
|
||||
"context": context.strip(),
|
||||
"query": query.strip(),
|
||||
"task_type": task_type.strip(),
|
||||
"reasoning": reasoning.strip(),
|
||||
"n_answers": n_answers,
|
||||
}
|
||||
needed = set(re.findall(r"\{(\w+)\}", template))
|
||||
return template.format(**{key: values[key] for key in needed})
|
||||
|
||||
|
||||
def truncate_reasoning(reasoning: str, max_chars: int = MAX_REASONING_CHARS) -> str:
|
||||
"""Keep the end of reasoning (drafts + conclusions) within a char budget."""
|
||||
reasoning = reasoning.strip()
|
||||
if len(reasoning) <= max_chars:
|
||||
return reasoning
|
||||
return "…\n" + reasoning[-max_chars:]
|
||||
|
||||
|
||||
# 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 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 []
|
||||
answers = parse_answer_lines(text_after_marker, query, task_type)
|
||||
expected = expected_answer_count(query, task_type)
|
||||
# Clip extras — misalignment from trailing commentary hurts every later slot.
|
||||
if len(answers) > expected:
|
||||
answers = answers[:expected]
|
||||
return answers
|
||||
|
||||
|
||||
def generate(
|
||||
tok,
|
||||
model,
|
||||
messages: list[dict],
|
||||
*,
|
||||
max_new_tokens: int,
|
||||
do_sample: bool,
|
||||
) -> str:
|
||||
ids = tok.apply_chat_template(
|
||||
messages, add_generation_prompt=True, return_tensors="pt",
|
||||
).to(model.device)
|
||||
gen_kwargs = {
|
||||
"max_new_tokens": max_new_tokens,
|
||||
"do_sample": do_sample,
|
||||
}
|
||||
if do_sample:
|
||||
gen_kwargs["temperature"] = TEMPERATURE
|
||||
gen_kwargs["top_p"] = TOP_P
|
||||
with torch.no_grad():
|
||||
out = model.generate(ids, **gen_kwargs)
|
||||
return tok.decode(out[0][ids.shape[-1] :], skip_special_tokens=True).strip()
|
||||
|
||||
|
||||
def generate_reasoning(tok, model, context: str, task_type: str, query: str) -> str:
|
||||
user = load_user_prompt("reason", context, task_type, query)
|
||||
messages = [
|
||||
{"role": "system", "content": SYSTEM_REASON},
|
||||
{"role": "user", "content": user},
|
||||
]
|
||||
return generate(
|
||||
tok, model, messages,
|
||||
max_new_tokens=MAX_NEW_TOKENS_REASON,
|
||||
do_sample=True,
|
||||
)
|
||||
|
||||
|
||||
def generate_answers(
|
||||
tok, model, context: str, task_type: str, query: str, reasoning: str,
|
||||
) -> str:
|
||||
n_answers = expected_answer_count(query, task_type)
|
||||
reasoning = truncate_reasoning(reasoning)
|
||||
user = load_user_prompt(
|
||||
"answer", context, task_type, query,
|
||||
reasoning=reasoning, n_answers=n_answers,
|
||||
)
|
||||
messages = [
|
||||
{"role": "system", "content": SYSTEM_ANSWER},
|
||||
{"role": "user", "content": user},
|
||||
]
|
||||
# Greedy answer emit: format discipline matters more than diversity here.
|
||||
return generate(
|
||||
tok, model, messages,
|
||||
max_new_tokens=MAX_NEW_TOKENS_ANSWER,
|
||||
do_sample=False,
|
||||
)
|
||||
|
||||
|
||||
def generate_explanation(
|
||||
tok, model, context: str, task_type: str, query: str, reasoning: str,
|
||||
) -> str:
|
||||
user = load_user_prompt(
|
||||
"explain", context, task_type, query,
|
||||
reasoning=truncate_reasoning(reasoning),
|
||||
)
|
||||
messages = [
|
||||
{"role": "system", "content": SYSTEM_EXPLAIN},
|
||||
{"role": "user", "content": user},
|
||||
]
|
||||
return generate(
|
||||
tok, model, messages,
|
||||
max_new_tokens=MAX_NEW_TOKENS_EXPLAIN,
|
||||
do_sample=False,
|
||||
)
|
||||
|
||||
|
||||
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))
|
||||
|
||||
# Write incrementally so a wall-clock kill still leaves a partial submission.csv.
|
||||
with open("submission.csv", "w", encoding="utf-8", newline="") as f:
|
||||
writer = csv.DictWriter(f, fieldnames=["id", "pred", "explanation"])
|
||||
writer.writeheader()
|
||||
f.flush()
|
||||
|
||||
for idx, r in enumerate(test_rows, start=1):
|
||||
task_type = r["task_type"]
|
||||
reason_tmpl = os.path.basename(_prompt_path("reason", task_type))
|
||||
answer_tmpl = os.path.basename(_prompt_path("answer", task_type))
|
||||
print(
|
||||
f"{idx}/{len(test_rows)} task={task_type} "
|
||||
f"reason={reason_tmpl} answer={answer_tmpl}",
|
||||
flush=True,
|
||||
)
|
||||
|
||||
reasoning = generate_reasoning(
|
||||
tok, model, r["context"], task_type, r["query"],
|
||||
)
|
||||
answer_text = generate_answers(
|
||||
tok, model, r["context"], task_type, r["query"], reasoning,
|
||||
)
|
||||
answers = postprocess_answer(answer_text, r["query"], task_type)
|
||||
explanation = generate_explanation(
|
||||
tok, model, r["context"], task_type, r["query"], reasoning,
|
||||
)
|
||||
|
||||
print(f" pred={answers!r}", flush=True)
|
||||
writer.writerow(
|
||||
{
|
||||
"id": r["id"],
|
||||
"pred": json.dumps(answers, ensure_ascii=False),
|
||||
"explanation": explanation,
|
||||
}
|
||||
)
|
||||
f.flush()
|
||||
|
||||
print("wrote submission.csv", flush=True)
|
||||
Reference in New Issue
Block a user