Spaces:
Runtime error
Runtime error
File size: 4,873 Bytes
b3bb0f6 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 |
from transformers import Qwen2_5_VLForConditionalGeneration, AutoTokenizer, AutoProcessor
from qwen_vl_utils import process_vision_info
from smolagents import tool
import torch
@tool
def video_reasoner(file_path : str, query : str) -> str:
"""
This tool performs requested visual reasoning task on the provided video and returns the generated output.
Args:
file_path: Path of a local video file on which visual reasoning is to be done.
query: visual reasoning that is to be done.
"""
try:
# default: Load the model on the available device(s)
model = Qwen2_5_VLForConditionalGeneration.from_pretrained(
"Qwen/Qwen2.5-VL-7B-Instruct", torch_dtype="auto", device_map="auto"
)
# default processer
processor = AutoProcessor.from_pretrained("Qwen/Qwen2.5-VL-7B-Instruct")
messages = [
{
"role": "user",
"content": [
{
"type": "video",
"video": file_path,
"max_pixels": 360 * 360,
"fps": 0.3,
},
{"type": "text", "text": f"{query}\n\nAdditional instruction: Treat the two types of penguins as distinct species e.g. Adelie and Emperor Penguins are considered two different species of birds."},
],
}
]
# Preparation for inference
text = processor.apply_chat_template(
messages, tokenize=False, add_generation_prompt=True
)
image_inputs, video_inputs = process_vision_info(messages)
inputs = processor(
text=[text],
images=image_inputs,
videos=video_inputs,
padding=True,
return_tensors="pt",
)
inputs = inputs.to("cuda")
# Inference: Generation of the output
generated_ids = model.generate(**inputs, max_new_tokens=512)
generated_ids_trimmed = [
out_ids[len(in_ids) :] for in_ids, out_ids in zip(inputs.input_ids, generated_ids)
]
output_text = processor.batch_decode(
generated_ids_trimmed, skip_special_tokens=True, clean_up_tokenization_spaces=False
)
import gc
# After inference
del image_inputs
del video_inputs
del inputs
del model
del processor
gc.collect() # Force Python garbage collection
torch.cuda.empty_cache() # Clear cached memory
return output_text
except Exception as e:
return f'error occured: {e}'
@tool
def image_reasoner(file_path : str, query : str) -> str:
"""
This tool performs requested visual reasoning task on the provided image and returns the generated output.
Args:
file_path: Path of a local image file on which visual reasoning is to be done.
query: visual reasoning that is to be done.
"""
try:
# default: Load the model on the available device(s)
model = Qwen2_5_VLForConditionalGeneration.from_pretrained(
"Qwen/Qwen2.5-VL-7B-Instruct", torch_dtype="auto", device_map="auto"
)
# default processer
processor = AutoProcessor.from_pretrained("Qwen/Qwen2.5-VL-7B-Instruct")
messages = [
{
"role": "user",
"content": [
{
"type": "image",
"image": file_path,
},
{"type": "text", "text": f"{query}\n\nAdditional instruction: Review your answer for correctness."},
],
}
]
# Preparation for inference
text = processor.apply_chat_template(
messages, tokenize=False, add_generation_prompt=True
)
image_inputs, video_inputs = process_vision_info(messages)
inputs = processor(
text=[text],
images=image_inputs,
videos=video_inputs,
padding=True,
return_tensors="pt",
)
inputs = inputs.to("cuda")
# Inference: Generation of the output
generated_ids = model.generate(**inputs, max_new_tokens=512)
generated_ids_trimmed = [
out_ids[len(in_ids) :] for in_ids, out_ids in zip(inputs.input_ids, generated_ids)
]
output_text = processor.batch_decode(
generated_ids_trimmed, skip_special_tokens=True, clean_up_tokenization_spaces=False
)
import gc
# After inference
del image_inputs
del video_inputs
del inputs
del model
del processor
gc.collect() # Force Python garbage collection
torch.cuda.empty_cache() # Clear cached memory
return output_text
except Exception as e:
return f'error occured: {e}' |