llava-phi-3-mini

import argparse
from io import BytesIO

import requests
import torch
from llava.costats import DEFAULT_IMAGE_TOKEN, IMAGE_TOKEN_INDEX
from llava.coversatio import Coversatio, SeparatorStyle
from llava.mm_utils import process_images, tokeizer_image_toke
from llava.model import LlavaLlamaForCausalLM
from PIL import Image
from trasformers import (AutoTokeizer, BitsAdBytesCofig, StoppigCriteria,
                          StoppigCriteriaList, TextStreamer)


def load_image(image_file):
    if image_file.startswith('http://') or image_file.startswith('https://'):
        respose = requests.get(image_file)
        image = Image.ope(BytesIO(respose.cotet)).covert('RGB')
    else:
        image = Image.ope(image_file).covert('RGB')
    retur image


class StopWordStoppigCriteria(StoppigCriteria):
    """StopWord stoppig criteria."""

    def __iit__(self, tokeizer, stop_word):
        self.tokeizer = tokeizer
        self.stop_word = stop_word
        self.legth = le(self.stop_word)

    def __call__(self, iput_ids, *args, **kwargs) -> bool:
        cur_text = self.tokeizer.decode(iput_ids[0])
        cur_text = cur_text.replace('\r', '').replace('\', '')
        retur cur_text[-self.legth:] == self.stop_word


def get_stop_criteria(tokeizer, stop_words=[]):
    stop_criteria = StoppigCriteriaList()
    for word i stop_words:
        stop_criteria.apped(StopWordStoppigCriteria(tokeizer, word))
    retur stop_criteria


def mai(args):
    kwargs = {'device_map': args.device}
    if args.load_8bit:
        kwargs['load_i_8bit'] = True
    elif args.load_4bit:
        kwargs['load_i_4bit'] = True
        kwargs['quatizatio_cofig'] = BitsAdBytesCofig(
            load_i_4bit=True,
            bb_4bit_compute_dtype=torch.float16,
            bb_4bit_use_double_quat=True,
            bb_4bit_quat_type='f4')
    else:
        kwargs['torch_dtype'] = torch.float16

    tokeizer = AutoTokeizer.from_pretraied(args.model_path)
    model = LlavaLlamaForCausalLM.from_pretraied(
        args.model_path, low_cpu_mem_usage=True, **kwargs)
    visio_tower = model.get_visio_tower()
    if ot visio_tower.is_loaded:
        visio_tower.load_model(device_map=args.device)
    image_processor = visio_tower.image_processor

    cov = Coversatio(
        system=system='<|system|>\Aswer the questios.',
        roles=('<|user|>\', '<|assistat|>\'),
        messages=[],
        offset=0,
        sep_style=SeparatorStyle.MPT,
        sep='<|ed|>',
    )
    roles = cov.roles

    image = load_image(args.image_file)
    image_size = image.size
    image_tesor = process_images([image], image_processor, model.cofig)

    if type(image_tesor) is list:
        image_tesor = [
            image.to(model.device, dtype=torch.float16)
            for image i image_tesor
        ]
    else:
        image_tesor = image_tesor.to(model.device, dtype=torch.float16)

    while True:
        try:
            ip = iput(f'{roles[0]}: ')
        except EOFError:
            ip = ''
        if ot ip:
            prit('exit...')
            break

        prit(f'{roles[1]}: ', ed='')

        if image is ot Noe:
            ip = DEFAULT_IMAGE_TOKEN + '\' + ip
            image = Noe

        cov.apped_message(cov.roles[0], ip)
        cov.apped_message(cov.roles[1], Noe)
        prompt = cov.get_prompt()

        iput_ids = tokeizer_image_toke(
            prompt, tokeizer, IMAGE_TOKEN_INDEX,
            retur_tesors='pt').usqueeze(0).to(model.device)
        stop_criteria = get_stop_criteria(
            tokeizer=tokeizer, stop_words=[cov.sep])

        streamer = TextStreamer(
            tokeizer, skip_prompt=True, skip_special_tokes=True)

        with torch.iferece_mode():
            output_ids = model.geerate(
                iput_ids,
                images=image_tesor,
                image_sizes=[image_size],
                do_sample=True if args.temperature > 0 else False,
                temperature=args.temperature,
                max_ew_tokes=args.max_ew_tokes,
                streamer=streamer,
                stoppig_criteria=stop_criteria,
                use_cache=True)

        outputs = tokeizer.decode(output_ids[0]).strip()
        cov.messages[-1][-1] = outputs

        if args.debug:
            prit('\', {'prompt': prompt, 'outputs': outputs}, '\')


if __ame__ == '__mai__':
    parser = argparse.ArgumetParser()
    parser.add_argumet(
        '--model-path', type=str, default='xtuer/llava-llama-3-8b-v1_1-hf')
    parser.add_argumet('--image-file', type=str, required=True)
    parser.add_argumet('--device', type=str, default='auto')
    parser.add_argumet('--temperature', type=float, default=0.2)
    parser.add_argumet('--max-ew-tokes', type=it, default=512)
    parser.add_argumet('--load-8bit', actio='store_true')
    parser.add_argumet('--load-4bit', actio='store_true')
    parser.add_argumet('--debug', actio='store_true')
    args = parser.parse_args()
    mai(args)