import argparse import itertools import json import os import random import math import re import time from functools import partial import torch import torchvision.transforms as T from torchvision.transforms.functional import InterpolationMode from transformers import AutoModel, AutoTokenizer from PIL import Image from tqdm import tqdm import sys current_dir = os.path.dirname(os.path.abspath(__file__)) root_dir = os.path.dirname(os.path.dirname(current_dir)) swift_path = os.path.join(root_dir, "third_party", "ms-swift-main") if swift_path not in sys.path: sys.path.append(swift_path) from swift.llm import ( PtEngine, RequestConfig, safe_snapshot_download, get_model_tokenizer, get_template, InferRequest ) from swift.tuners import Swift from scripts.tools.agentthink_data_generater_pipeline import generate_func_prompt from scripts.tools.tool_libraries_simple import FuncAgent func_agent = FuncAgent() ds_collections = { 'DriveLMMo1': { 'root': './data/DriveLMMo1_TEST.jsonl', 'max_new_tokens': 2000, 'min_new_tokens': 1, 'split': 'validation', 'image_root': './data/image2concat' } } def collate_fn(batches, tokenizer): # pixel_values = torch.cat([_['pixel_values'] for _ in batches], dim=0) images = [_['images'] for _ in batches] questions = [_['question'] for _ in batches] answers = [_['answer'] for _ in batches] data_ids = [_['data_id'] for _ in batches] return images, questions, answers, data_ids class DriveLMMo1Dataset(torch.utils.data.Dataset): def __init__(self, root, split, prompt, image_path, point_path=None, input_size=224, dynamic_image_size=False, use_thumbnail=False, max_num=6, ): with open(root, 'r') as f: self.data = [json.loads(line) for line in f.readlines()] # data_val = json.load(f) # merge all dataset # self.data = concatenate_datasets(sub_dataset_list) self.prompt = prompt self.input_size = input_size self.dynamic_image_size = dynamic_image_size self.use_thumbnail = use_thumbnail self.max_num = max_num self.image_path = image_path self.point_path = point_path def __len__(self): return len(self.data) def __getitem__(self, idx): data = self.data[idx] data_id = data['id'] question = data['conversations'][0]['value'].strip() image_file = os.path.join(self.image_path, data['image']) image = Image.open(image_file).convert('RGB') answer = data['conversations'][1]['value'].strip() if self.dynamic_image_size: pil_image = dynamic_preprocess(image, image_size=self.input_size, use_thumbnail=self.use_thumbnail, max_num=self.max_num) images = pil_image else: images = [image] return { 'question': self.prompt+'\n\n'+question, 'images': image_file, 'answer': answer, 'data_id': data_id } class InferenceSampler(torch.utils.data.sampler.Sampler): def __init__(self, size): self._size = int(size) assert size > 0 self._rank = torch.distributed.get_rank() self._world_size = torch.distributed.get_world_size() self._local_indices = self._get_local_indices(size, self._world_size, self._rank) @staticmethod def _get_local_indices(total_size, world_size, rank): shard_size = total_size // world_size left = total_size % world_size shard_sizes = [shard_size + int(r < left) for r in range(world_size)] begin = sum(shard_sizes[:rank]) end = min(sum(shard_sizes[:rank + 1]), total_size) return range(begin, end) def __iter__(self): yield from self._local_indices def __len__(self): return len(self._local_indices) def load_model(pretrained_model): """Load model and tokenizer""" model = pretrained_model template_type = None # None: default template_type default_system = None # None: default_system # Load models and conversation model, tokenizer = get_model_tokenizer(model) template_type = template_type or model.model_meta.template template = get_template(template_type, tokenizer, default_system=default_system) engine = PtEngine.from_model_template(model, template, max_batch_size=1) return engine, model, tokenizer def retry_torch_distributed_barrier(max_retries=3, delay_seconds=5): """ Attempts to execute torch.distributed.barrier() with a retry mechanism Args: max_retries (int): Maximum number of retry attempts delay_seconds (int): Delay in seconds between retry attempts """ retries = 0 while retries < max_retries: try: torch.distributed.barrier() # Exit the function upon successful execution return except Exception as e: retries += 1 print(f"torch.distributed.barrier() failed (retry {retries}/{max_retries}): {str(e)}") print(f"Retrying after {delay_seconds} seconds...") time.sleep(delay_seconds) # Raise exception if barrier still fails after max retries raise RuntimeError(f"torch.distributed.barrier() failed after {max_retries} retries") def evaluate_chat_model(): random.seed(args.seed) # prompt = "When answering the question based on the provided image, follow a structured and logical reasoning process. Organize your response using the format, ensuring each step builds upon the previous one and clearly explains how the image(s) contribute to the solution. Your answer should be structured as Reasoning Steps: (step by step reasoning) Final Answer: (final answer) \n Question: " tool_info_intro = generate_func_prompt() base_prompt = "When addressing the question presented in the image, please adhere to a structured and logical reasoning process. Ensure that your chain of reasoning consists of no more than six steps to maintain clarity and conciseness." tool_use_info = f'For each atomic step of the reasoning chain, you may choose the appropriate tools and their parameters (if blank, the value of "parameters" should be [""]). Refer this list:[{tool_info_intro}] as value of element "Tool" ' output_format_prompt = """Generate a sub-question for each action or reasoning step like(perception, prediction or planning information) as value of element "Sub". Also, generate answer for each Sub as value of element "Guess_Answer" if you make sure it is correct. Also, extract the key terms from the content related to the "Guess Answer" and list them as the value of the "keywords" element. For each keyword, consider 2-5 synonyms or alternative expressions. In addition, if you can not answer some sub-questions, make the element "Missing_flag" value “False”, otherwise, make it “True”. And "next_action" should be either "continue reasoning" or "conclude". Continue generating steps until the reasoning chain is complete. Final, add the final answer as the value of the "final_answer". Also, you should refer the final answer for extracting the key words and list these key words as the value of the "final_answer_keywords". For example: { "Question": "", "Chain": [ { "Tool": {"function_name":"open vocabulary detector", "parameters":["", ""]}, "Sub": "", "Guess_Answer": "", "key_words": ["words1", "words2", ...], "Missing_flag": "", "next_action": "continue reasoning" }, { "Tool": {"function_name":"depth_estimator", "parameters":["", ""]}, "Sub": "", "Guess_Answer": "", "key_words": ["words1", "words2", ...], "Missing_flag": "", "next_action": "conclude" } ], "final_answer_keywords": ["words1", "words2", ...], "final_answer": "" } STRICTLY FOLLOW THE JSON RESPONSE FORMAT. THE RESPONDE SHOULD START WITH "{". DO NOT START WITH "```json" OR ANYTHING ELSE.""" prompt = base_prompt + tool_use_info + output_format_prompt for ds_name in args.datasets: dataset = DriveLMMo1Dataset( root=ds_collections[ds_name]['root'], split=ds_collections[ds_name]['split'], prompt=prompt, image_path=ds_collections[ds_name]['image_root'], # image_meta = ds_collections[ds_name]["image_meta"], # input_size=image_size, dynamic_image_size=args.dynamic, # use_thumbnail=use_thumbnail, max_num=args.max_num ) dataloader = torch.utils.data.DataLoader( dataset=dataset, sampler=InferenceSampler(len(dataset)), batch_size=args.batch_size, num_workers=args.num_workers, pin_memory=True, drop_last=False, collate_fn=partial(collate_fn, tokenizer=tokenizer), ) outputs = [] for _, (images, questions, answers, data_ids) in tqdm(enumerate(dataloader)): # pixel_values = pixel_values.to(torch.bfloat16).cuda() generation_config = dict( num_beams=args.num_beams, max_new_tokens=ds_collections[ds_name]['max_new_tokens'], min_new_tokens=ds_collections[ds_name]['min_new_tokens'], do_sample=True if args.temperature > 0 else False, temperature=args.temperature, ) infer_requests = [ InferRequest(messages=[ {'role': 'user', 'content': f"{questions[0]}"} ], images=images), ] # breakpoint() resp_list = engine.infer(infer_requests, RequestConfig(max_tokens=2000, temperature=args.temperature)) pred = resp_list[0].choices[0].message.content # filter those wrong format try: with open(os.path.join(args.out_dir, 'filter.json'), 'w') as f: json.dump(pred, f, indent=4) except: continue preds = [pred] for question, pred, answer, data_id in zip(questions, preds, answers, data_ids): outputs.append({ 'question': question, 'answer': pred, 'gt_answers': answer, 'id': data_id }) # torch.distributed.barrier() retry_torch_distributed_barrier(max_retries=15, delay_seconds=5) world_size = torch.distributed.get_world_size() merged_outputs = [None for _ in range(world_size)] torch.distributed.all_gather_object(merged_outputs, json.dumps(outputs)) merged_outputs = [json.loads(_) for _ in merged_outputs] merged_outputs = [_ for _ in itertools.chain.from_iterable(merged_outputs)] if torch.distributed.get_rank() == 0: print(f'Evaluating {ds_name} ...') results_file = f'{ds_name}_{args.output_name}.json' output_path = os.path.join(args.out_dir, results_file) # breakpoint() with open(output_path, 'w') as f: json.dump(merged_outputs, f, indent=4) print('Results saved to {}'.format(output_path)) if __name__ == '__main__': parser = argparse.ArgumentParser() parser.add_argument('--checkpoint', type=str, default='') parser.add_argument('--datasets', type=str, default='DriveLMMo1') parser.add_argument('--batch-size', type=int, default=1) parser.add_argument('--num-workers', type=int, default=4) parser.add_argument('--num-beams', type=int, default=1) parser.add_argument('--temperature', type=float, default=0.0) parser.add_argument('--out-dir', type=str, default='results') parser.add_argument('--output_name', type=str, default='qwen_32B_swift') parser.add_argument('--seed', type=int, default=0) parser.add_argument('--dynamic', action='store_true', default=False) parser.add_argument('--max-num', type=int, default=12) parser.add_argument('--load-in-8bit', action='store_true') parser.add_argument('--load-in-4bit', action='store_true') parser.add_argument('--auto', action='store_true') args = parser.parse_args() if not os.path.exists(args.out_dir): os.makedirs(args.out_dir, exist_ok=True) args.datasets = args.datasets.split(',') print('datasets:', args.datasets) assert args.batch_size == 1, 'Only batch size 1 is supported' torch.distributed.init_process_group( backend='nccl', world_size=int(os.getenv('WORLD_SIZE', '1')), rank=int(os.getenv('RANK', '0')), ) torch.cuda.set_device(int(os.getenv('LOCAL_RANK', 0))) # model, tokenizer = load_model_and_tokenizer() # engine, model, tokenizer = load_model("qwen_vla/Qwen2.5-VL-32B-Instruct") engine, model, tokenizer = load_model(args.checkpoint) total_params = sum(p.numel() for p in model.parameters()) / 1e9 if total_params > 20 or args.dynamic: args.num_beams = 1 print(f'[test] total_params: {total_params}B, use num_beams: {args.num_beams}') else: print(f'[test] total_params: {total_params}B') print(f'[test] max_num: {args.max_num}') evaluate_chat_model()