| 24 | return replaced_text |
| 25 | |
| 26 | class PromptEnhancerV2: |
| 27 | def __init__(self, models_root_path, device_map="auto", torch_dtype="bfloat16"): |
| 28 | if not logging.getLogger(__name__).handlers: |
| 29 | logging.basicConfig(level=logging.INFO) |
| 30 | self.logger = logging.getLogger(__name__) |
| 31 | |
| 32 | # dtype 兼容处理 |
| 33 | if torch_dtype == "bfloat16": |
| 34 | dtype = torch.bfloat16 |
| 35 | elif torch_dtype == "float16": |
| 36 | dtype = torch.float16 |
| 37 | else: |
| 38 | dtype = torch.float32 |
| 39 | |
| 40 | self.model = Qwen2_5_VLForConditionalGeneration.from_pretrained( |
| 41 | models_root_path, |
| 42 | torch_dtype=dtype, |
| 43 | attn_implementation="flash_attention_2", |
| 44 | device_map=device_map, |
| 45 | ) |
| 46 | self.processor = AutoProcessor.from_pretrained(models_root_path) |
| 47 | |
| 48 | @torch.inference_mode() |
| 49 | def predict( |
| 50 | self, |
| 51 | prompt_cot, |
| 52 | sys_prompt="请根据用户的输入,生成思考过程的思维链并改写提示词:", |
| 53 | temperature=0.0, |
| 54 | top_p=1.0, |
| 55 | max_new_tokens=2048, |
| 56 | device="cuda", |
| 57 | ): |
| 58 | org_prompt_cot = prompt_cot |
| 59 | try: |
| 60 | user_prompt_format = sys_prompt + "\n" + org_prompt_cot |
| 61 | messages = [ |
| 62 | { |
| 63 | "role": "user", |
| 64 | "content": [ |
| 65 | {"type": "text", "text": user_prompt_format}, |
| 66 | ], |
| 67 | } |
| 68 | ] |
| 69 | |
| 70 | text = self.processor.apply_chat_template( |
| 71 | messages, tokenize=False, add_generation_prompt=True |
| 72 | ) |
| 73 | image_inputs, video_inputs = process_vision_info(messages) |
| 74 | inputs = self.processor( |
| 75 | text=[text], |
| 76 | images=image_inputs, |
| 77 | videos=video_inputs, |
| 78 | padding=True, |
| 79 | return_tensors="pt", |
| 80 | ) |
| 81 | inputs = inputs.to(device) |
| 82 | |
| 83 | # 注意:原始代码固定 do_sample=False,top_k=5, top_p=0.9,这里保持一致 |