# --------------------------------------------------------
# InternVL
# Copyright (c) 2024 OpenGVLab
# Licensed under The MIT License [see LICENSE for details]
# --------------------------------------------------------
"""
A model worker executes the model.
"""
import argparse
import asyncio
import base64
import json
import os
import threading
import time
import uuid
from functools import partial
from io import BytesIO
from threading import Thread
import requests
import torch
import torchvision.transforms as T
import uvicorn
from constants import IMAGENET_MEAN, IMAGENET_STD, WORKER_HEART_BEAT_INTERVAL
from fastapi import BackgroundTasks, FastAPI, Request
from fastapi.responses import StreamingResponse
from PIL import Image
from torchvision.transforms.functional import InterpolationMode
from transformers import (AutoModelForCausalLM, AutoTokenizer,
TextIteratorStreamer)
from utils import build_logger, pretty_print_semaphore, server_error_msg
worker_id = str(uuid.uuid4())[:6]
logger = build_logger('model_worker', f'model_worker_{worker_id}.log')
global_counter = 0
model_semaphore = None
def load_image_from_base64(image):
return Image.open(BytesIO(base64.b64decode(image)))
def build_transform(input_size):
MEAN, STD = IMAGENET_MEAN, IMAGENET_STD
transform = T.Compose([
T.Lambda(lambda img: img.convert('RGB') if img.mode != 'RGB' else img),
T.Resize((input_size, input_size), interpolation=InterpolationMode.BICUBIC),
T.ToTensor(),
T.Normalize(mean=MEAN, std=STD)
])
return transform
def find_closest_aspect_ratio(aspect_ratio, target_ratios, width, height, image_size):
best_ratio_diff = float('inf')
best_ratio = (1, 1)
area = width * height
for ratio in target_ratios:
target_aspect_ratio = ratio[0] / ratio[1]
ratio_diff = abs(aspect_ratio - target_aspect_ratio)
if ratio_diff < best_ratio_diff:
best_ratio_diff = ratio_diff
best_ratio = ratio
elif ratio_diff == best_ratio_diff:
if area > 0.5 * image_size * image_size * ratio[0] * ratio[1]:
best_ratio = ratio
return best_ratio
def dynamic_preprocess(image, min_num=1, max_num=6, image_size=448, use_thumbnail=False):
orig_width, orig_height = image.size
aspect_ratio = orig_width / orig_height
# calculate the existing image aspect ratio
target_ratios = set(
(i, j) for n in range(min_num, max_num + 1) for i in range(1, n + 1) for j in range(1, n + 1) if
i * j <= max_num and i * j >= min_num)
target_ratios = sorted(target_ratios, key=lambda x: x[0] * x[1])
# find the closest aspect ratio to the target
target_aspect_ratio = find_closest_aspect_ratio(
aspect_ratio, target_ratios, orig_width, orig_height, image_size)
# calculate the target width and height
target_width = image_size * target_aspect_ratio[0]
target_height = image_size * target_aspect_ratio[1]
blocks = target_aspect_ratio[0] * target_aspect_ratio[1]
# resize the image
resized_img = image.resize((target_width, target_height))
processed_images = []
for i in range(blocks):
box = (
(i % (target_width // image_size)) * image_size,
(i // (target_width // image_size)) * image_size,
((i % (target_width // image_size)) + 1) * image_size,
((i // (target_width // image_size)) + 1) * image_size
)
# split the image
split_img = resized_img.crop(box)
processed_images.append(split_img)
assert len(processed_images) == blocks
if use_thumbnail and len(processed_images) != 1:
thumbnail_img = image.resize((image_size, image_size))
processed_images.append(thumbnail_img)
return processed_images
def heart_beat_worker(controller):
while True:
time.sleep(WORKER_HEART_BEAT_INTERVAL)
controller.send_heart_beat()
class ModelWorker:
def __init__(self, controller_addr, worker_addr, worker_id, model_path, model_name,
load_8bit, device, context_len=8192):
self.controller_addr = controller_addr
self.worker_addr = worker_addr
self.worker_id = worker_id
if model_path.endswith('/'):
model_path = model_path[:-1]
if model_name is None:
model_paths = model_path.split('/')
if model_paths[-1].startswith('checkpoint-'):
self.model_name = model_paths[-2] + '_' + model_paths[-1]
else:
self.model_name = model_paths[-1]
else:
self.model_name = model_name
logger.info(f'Loading the model {self.model_name} on worker {worker_id} ...')
self.tokenizer = AutoTokenizer.from_pretrained(model_path, trust_remote_code=True, use_fast=False)
if device == 'auto':
os.environ['CUDA_LAUNCH_BLOCKING'] = '1'
# This can make distributed deployment work properly
self.model = AutoModelForCausalLM.from_pretrained(
model_path,
load_in_8bit=load_8bit,
torch_dtype=torch.float16,
device_map='auto',
trust_remote_code=True).eval()
else:
self.model = AutoModelForCausalLM.from_pretrained(
model_path,
load_in_8bit=load_8bit,
torch_dtype=torch.float16,
trust_remote_code=True).eval()
if not load_8bit and not device == 'auto':
self.model = self.model.cuda()
self.image_size = self.model.config.force_image_size
self.context_len = context_len
self.register_to_controller()
self.heart_beat_thread = threading.Thread(
target=heart_beat_worker, args=(self,))
self.heart_beat_thread.start()
def register_to_controller(self):
logger.info('Register to controller')
url = self.controller_addr + '/register_worker'
data = {
'worker_name': self.worker_addr,
'check_heart_beat': True,
'worker_status': self.get_status()
}
r = requests.post(url, json=data)
assert r.status_code == 200
def send_heart_beat(self):
logger.info(f'Send heart beat. Models: {[self.model_name]}. '
f'Semaphore: {pretty_print_semaphore(model_semaphore)}. '
f'global_counter: {global_counter}')
url = self.controller_addr + '/receive_heart_beat'
while True:
try:
ret = requests.post(url, json={
'worker_name': self.worker_addr,
'queue_length': self.get_queue_length()}, timeout=5)
exist = ret.json()['exist']
break
except requests.exceptions.RequestException as e:
logger.error(f'heart beat error: {e}')
time.sleep(5)
if not exist:
self.register_to_controller()
def get_queue_length(self):
if model_semaphore is None:
return 0
else:
return args.limit_model_concurrency - model_semaphore._value + (len(
model_semaphore._waiters) if model_semaphore._waiters is not None else 0)
def get_status(self):
return {
'model_names': [self.model_name],
'speed': 1,
'queue_length': self.get_queue_length(),
}
@torch.inference_mode()
def generate_stream(self, params):
system_message = params['prompt'][0]['content']
send_messages = params['prompt'][1:]
max_input_tiles = params['max_input_tiles']
temperature = params['temperature']
top_p = params['top_p']
max_new_tokens = params['max_new_tokens']
repetition_penalty = params['repetition_penalty']
do_sample = True if temperature > 0.0 else False
global_image_cnt = 1
history, pil_images, max_input_tile_list = [], [], []
for message in send_messages:
if message['role'] == 'user':
prefix = ''
if 'image' in message:
max_input_tile_temp = []
for image_str in message['image']:
pil_images.append(load_image_from_base64(image_str))
prefix += f'Image-{global_image_cnt}: <image>\n\n'
global_image_cnt += 1
max_input_tile_temp.append(max(1, max_input_tiles // len(message['image'])))
if len(max_input_tile_temp) > 0:
max_input_tile_list.append(max_input_tile_temp)
content = prefix + message['content']
history.append([content, ])
else:
history[-1].append(message['content'])
question, history = history[-1][0], history[:-1]
# Create a new list to store processed sublists
flattened_list = []
# Iterate through all but the last sublist in max_input_tile_list and process them
for sublist in max_input_tile_list[:-1]:
processed_sublist = [1] * len(sublist) # Change each element in the sublist to 1
flattened_list.extend(processed_sublist) # Flatten the processed sublist and add to the new list
# If max_input_tile_list is not empty, add the last sublist to the new list
if max_input_tile_list:
flattened_list.extend(max_input_tile_list[-1])
max_input_tile_list = flattened_list
assert len(max_input_tile_list) == len(pil_images), 'The number of max_input_tile_list and pil_images should be the same.'
logger.info(f'max_input_tile_list: {max_input_tile_list}')
old_system_message = self.model.system_message
self.model.system_message = system_message
image_tiles = []
transform = build_transform(input_size=self.image_size)
if len(pil_images) > 0:
for current_max_input_tiles, pil_image in zip(max_input_tile_list, pil_images):
if self.model.config.dynamic_image_size:
tiles = dynamic_preprocess(
pil_image, image_size=self.image_size, max_num=current_max_input_tiles,
use_thumbnail=self.model.config.use_thumbnail)
else:
tiles = [pil_image]
image_tiles += tiles
pixel_values = [transform(item) for item in image_tiles]
pixel_values = torch.stack(pixel_values).to(self.model.device, dtype=torch.float16)
logger.info(f'Split images to {pixel_values.shape}')
else:
pixel_values = None
streamer = TextIteratorStreamer(self.tokenizer, skip_prompt=True, skip_special_tokens=False, timeout=10)
generation_config = dict(
num_beams=1,
max_new_tokens=max_new_tokens,
do_sample=do_sample,
temperature=temperature,
repetition_penalty=repetition_penalty,
max_length=self.context_len,
top_p=top_p,
streamer=streamer,
)
logger.info(history)
logger.info(f'Generation config: {generation_config}')
try:
thread = Thread(target=self.model.chat, kwargs=dict(
tokenizer=self.tokenizer,
pixel_values=pixel_values,
question=question,
history=history,
return_history=False,
generation_config=generation_config,
))
thread.start()
generated_text = ''
for new_text in streamer:
generated_text += new_text
yield json.dumps({'text': generated_text.replace(self.model.conv_template.sep, ''),
'error_code': 0}).encode() + b'\0'
self.model.system_message = old_system_message
except:
torch.cuda.empty_cache()
def generate_stream_gate(self, params):
try:
for x in self.generate_stream(params):
yield x
except ValueError as e:
print('Caught ValueError:', e)
ret = {
'text': server_error_msg,
'error_code': 1,
}
yield json.dumps(ret).encode() + b'\0'
except torch.cuda.CudaError as e:
print('Caught torch.cuda.CudaError:', e)
ret = {
'text': server_error_msg,
'error_code': 1,
}
yield json.dumps(ret).encode() + b'\0'
except Exception as e:
print('Caught Unknown Error', e)
ret = {
'text': server_error_msg,
'error_code': 1,
}
yield json.dumps(ret).encode() + b'\0'
app = FastAPI()
def release_model_semaphore(fn=None):
model_semaphore.release()
if fn is not None:
fn()
@app.post('/worker_generate_stream')
async def generate_stream(request: Request):
global model_semaphore, global_counter
global_counter += 1
params = await request.json()
if model_semaphore is None:
model_semaphore = asyncio.Semaphore(args.limit_model_concurrency)
await model_semaphore.acquire()
worker.send_heart_beat()
generator = worker.generate_stream_gate(params)
background_tasks = BackgroundTasks()
background_tasks.add_task(partial(release_model_semaphore, fn=worker.send_heart_beat))
return StreamingResponse(generator, background=background_tasks)
@app.post('/worker_get_status')
async def get_status(request: Request):
return worker.get_status()
if __name__ == '__main__':
parser = argparse.ArgumentParser()
parser.add_argument('--host', type=str, default='0.0.0.0')
parser.add_argument('--port', type=int, default=21002)
parser.add_argument('--worker-address', type=str, default='http://localhost:21002')
parser.add_argument('--controller-address', type=str, default='http://localhost:21001')
parser.add_argument('--model-path', type=str, default='facebook/opt-350m')
parser.add_argument('--model-name', type=str)
parser.add_argument('--device', type=str, default='cuda')
parser.add_argument('--limit-model-concurrency', type=int, default=5)
parser.add_argument('--stream-interval', type=int, default=1)
parser.add_argument('--load-8bit', action='store_true')
args = parser.parse_args()
logger.info(f'args: {args}')
worker = ModelWorker(args.controller_address,
args.worker_address,
worker_id,
args.model_path,
args.model_name,
args.load_8bit,
args.device)
uvicorn.run(app, host=args.host, port=args.port, log_level='info')