(
model, tokenizer, image_processor, params, device
)
| 183 | |
| 184 | @torch.inference_mode() |
| 185 | def generate_stream( |
| 186 | model, tokenizer, image_processor, params, device |
| 187 | ): |
| 188 | prompt = params["prompt"] |
| 189 | images = params.get("images", None) |
| 190 | videos = params.get("videos", None) |
| 191 | temperature = float(params.get("temperature", 0.7)) |
| 192 | max_new_tokens = int(params.get("max_new_tokens", 1024)) |
| 193 | |
| 194 | num_queries = model.config.num_query_tokens |
| 195 | |
| 196 | stop_words = ["Human: ", "Assistant: ", "###", "\n\n"] |
| 197 | stop_words_ids = [tokenizer(stop_word, return_tensors='pt')['input_ids'].squeeze() for stop_word in stop_words] |
| 198 | stopping_criteria = StoppingCriteriaList([StoppingCriteriaSub(stops=stop_words_ids)]) |
| 199 | |
| 200 | generation_config = GenerationConfig( |
| 201 | bos_token_id=1, |
| 202 | do_sample=True, |
| 203 | temperature=temperature, |
| 204 | max_new_tokens=max_new_tokens, |
| 205 | stopping_criteria=stopping_criteria |
| 206 | ) |
| 207 | |
| 208 | pixel_values = None |
| 209 | if images is not None: |
| 210 | pixel_values = load_image(images).to(device) # only support one image |
| 211 | image_query = DEFAULT_IMG_START_TOKEN + DEFAULT_IMG_END_TOKEN |
| 212 | prompt = prompt.replace("<image>", image_query) |
| 213 | |
| 214 | elif videos is not None: |
| 215 | pixel_values = load_video(videos).to(device) |
| 216 | video_query = DEFAULT_VIDEO_START_TOKEN + DEFAULT_VIDEO_END_TOKEN |
| 217 | prompt = prompt.replace("<video>", video_query) |
| 218 | |
| 219 | model_inputs = tokenizer([prompt], return_tensors="pt") |
| 220 | model_inputs.pop("token_type_ids", None) |
| 221 | |
| 222 | if pixel_values is not None: |
| 223 | model_inputs["pixel_values"] = pixel_values |
| 224 | |
| 225 | generation_output = model.generate( |
| 226 | **model_inputs, |
| 227 | generation_config=generation_config, |
| 228 | return_dict_in_generate=True, |
| 229 | output_scores=True |
| 230 | ) |
| 231 | else: |
| 232 | generation_output = model.language_model.generate( |
| 233 | **model_inputs, |
| 234 | generation_config=generation_config, |
| 235 | return_dict_in_generate=True, |
| 236 | output_scores=True |
| 237 | ) |
| 238 | |
| 239 | preds = generation_output.sequences |
| 240 | outputs = tokenizer.batch_decode(preds, skip_special_tokens=True) |
| 241 | return outputs |
| 242 |
nothing calls this directly
no test coverage detected