Buckets:
| """Evaluate direct-answer accuracy of the unmodified UniAR-SFT baseline.""" | |
| from __future__ import annotations | |
| import argparse | |
| import json | |
| from pathlib import Path | |
| import torch | |
| from transformers import AutoProcessor | |
| from uniar import UniARForConditionalGeneration, configure_processor | |
| from .dataset import ZebraTrajectoryDataset | |
| from .scoring import score_answer | |
| from .trajectory import render_messages, resolve_image_path | |
| DIRECT_SYSTEM = "Inspect the image and solve the problem. Return only the final answer." | |
| def _load_rgb(root: Path, relative: str): | |
| from PIL import Image | |
| with Image.open(resolve_image_path(root, relative)) as image: | |
| return image.convert("RGB").copy() | |
| def _target(trajectory: dict) -> str: | |
| return next( | |
| event["text"] for event in trajectory["assistant"] if event["role"] == "answer" | |
| ) | |
| def evaluate(args: argparse.Namespace) -> dict: | |
| from qwen_vl_utils import process_vision_info | |
| output_dir = args.output_root / args.output_name | |
| output_dir.mkdir(parents=True, exist_ok=False) | |
| processor = configure_processor( | |
| AutoProcessor.from_pretrained(args.model_path, use_fast=True) | |
| ) | |
| model = UniARForConditionalGeneration.from_pretrained( | |
| args.model_path, | |
| torch_dtype=torch.bfloat16, | |
| attn_implementation=args.attn, | |
| low_cpu_mem_usage=True, | |
| ).to("cuda").eval() | |
| im_end_id = processor.tokenizer.convert_tokens_to_ids("<|im_end|>") | |
| dataset = ZebraTrajectoryDataset(args.samples_root) | |
| dataset.records = dataset.records[: args.limit_samples] | |
| rows = [] | |
| with (output_dir / "predictions.jsonl").open("w", encoding="utf-8") as handle: | |
| for index, trajectory in enumerate(dataset.records): | |
| messages, _ = render_messages( | |
| trajectory, lambda path: _load_rgb(dataset.root, path) | |
| ) | |
| if args.prompt_mode == "direct": | |
| prompt = messages[:-1] | |
| prompt[0]["content"][0]["text"] = DIRECT_SYSTEM | |
| prompt[-1]["content"].append( | |
| { | |
| "type": "text", | |
| "text": "\nDo not show reasoning steps. Output only the final answer.", | |
| } | |
| ) | |
| else: | |
| # Native released understanding behavior: user image/question, | |
| # then an ordinary assistant generation prompt. | |
| prompt = [messages[1]] | |
| text = processor.apply_chat_template( | |
| prompt, tokenize=False, add_generation_prompt=True | |
| ) | |
| images, videos = process_vision_info(prompt) | |
| inputs = processor( | |
| text=[text], images=images, videos=videos, return_tensors="pt" | |
| ).to("cuda") | |
| inputs.pop("mm_token_type_ids", None) | |
| ids = model.generate( | |
| **inputs, | |
| do_sample=False, | |
| max_new_tokens=args.max_new_tokens, | |
| eos_token_id=[im_end_id, processor.tokenizer.pad_token_id], | |
| pad_token_id=processor.tokenizer.pad_token_id, | |
| ) | |
| generated = ids[0, inputs.input_ids.shape[1] :] | |
| decoded = processor.tokenizer.decode( | |
| generated, skip_special_tokens=False, clean_up_tokenization_spaces=False | |
| ) | |
| score = score_answer(decoded, _target(trajectory)) | |
| row = { | |
| "sample_id": trajectory["sample_id"], | |
| "decoded": decoded, | |
| "generated_tokens": generated.tolist(), | |
| "score": score, | |
| } | |
| rows.append(row) | |
| handle.write(json.dumps(row, ensure_ascii=False) + "\n") | |
| print( | |
| f"baseline {index + 1}/{len(dataset)} {trajectory['sample_id']} " | |
| f"correct={score['correct']} output={score['candidate']!r}", | |
| flush=True, | |
| ) | |
| summary = { | |
| "samples": len(rows), | |
| "correct": sum(row["score"]["correct"] for row in rows), | |
| "accuracy": sum(row["score"]["correct"] for row in rows) / len(rows), | |
| "exact": sum(row["score"]["exact"] for row in rows), | |
| } | |
| (output_dir / "summary.json").write_text( | |
| json.dumps(summary, indent=2) + "\n", encoding="utf-8" | |
| ) | |
| print(json.dumps(summary), flush=True) | |
| return summary | |
| def build_parser() -> argparse.ArgumentParser: | |
| parser = argparse.ArgumentParser(description=__doc__) | |
| parser.add_argument("--model-path", type=Path, default=Path("models/UniAR-SFT")) | |
| parser.add_argument("--samples-root", type=Path, default=Path("data/Zebra-CoT-samples")) | |
| parser.add_argument("--output-root", type=Path, required=True) | |
| parser.add_argument("--attn", default="flash_attention_2") | |
| parser.add_argument("--max-new-tokens", type=int, default=512) | |
| parser.add_argument("--limit-samples", type=int, default=16) | |
| parser.add_argument("--output-name", default="baseline") | |
| parser.add_argument("--prompt-mode", choices=["direct", "native"], default="direct") | |
| return parser | |
| def main() -> None: | |
| evaluate(build_parser().parse_args()) | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 5.2 kB
- Xet hash:
- 9aacdee26dd30513f757154fd5a6104b61c3f31b2ea933f7eef1576fbf02ab11
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.