app.py

# This app is inspired by:
# https://huggingface.co/spaces/ysharma/Microsoft_Phi-3-Vision-128k
# and ref: https://www.analyticsvidhya.com/blog/2023/12/building-a-multimodal-chatbot-with-gemini-and-gradio/

import os
import base64

import gradio as gr
from mistralai import Mistral

api_key = os.environ["MISTRAL_API_KEY"]

PLACEHOLDER = """In future, LISA will integrate multimodal model that brings together language and vision capabilities for chatting with papers."""


def encode_image(image_path):
    """Encode the image to base64."""
    try:
        with open(image_path, "rb") as image_file:
            return base64.b64encode(image_file.read()).decode("utf-8")
    except FileNotFoundError:
        print(f"Error: The file {image_path} was not found.")
        return None
    except Exception as e:  # Added general exception handling
        print(f"Error: {e}")
        return None


# def image_to_base64(image_path):
#     with open(image_path, "rb") as img:
#         encoded_string = base64.b64encode(img.read()).decode("utf-8")
#     return f"data:image/jpeg;base64,{encoded_string}"


def bot_streaming(message, history):
    print(f"message is - {message}")
    print(f"history is - {history}")
    if not message:
        raise gr.Error(
            "You need to upload an image for vision model to work. Close the error and try again with an Image."
        )
    if message["files"]:
        # message["files"][-1] is a Dict or just a string
        if type(message["files"][-1]) == dict:
            image = message["files"][-1]["path"]
        else:
            image = message["files"][-1]
    else:
        # if there's no image uploaded for this turn, look for images in the past turns
        # kept inside tuples, take the last one
        for hist in history:
            if type(hist[0]) == tuple:
                image = hist[0][0]
    try:
        if image is None:
            # Handle the case where image is None
            raise gr.Error(
                "You need to upload an image for vision model to work. Close the error and try again with an Image."
            )
    except NameError:
        # Handle the case where 'image' is not defined at all
        raise gr.Error(
            "You need to upload an image for vision model to work. Close the error and try again with an Image."
        )

    conversation = []
    flag = False
    for user, assistant in history:
        if assistant is None:
            # pass
            flag = True
            conversation.extend([{"role": "user", "content": ""}])
            continue
        if flag == True:
            conversation[0]["content"] = f"<|image_1|>\n{user}"
            conversation.extend([{"role": "assistant", "content": assistant}])
            flag = False
            continue
        conversation.extend(
            [
                {"role": "user", "content": user},
                {"role": "assistant", "content": assistant},
            ]
        )

    if len(history) == 0:
        conversation.append(
            {"role": "user", "content": f"<|image_1|>\n{message['text']}"}
        )
    else:
        conversation.append({"role": "user", "content": message["text"]})
    print(f"prompt is -\n{conversation}")
    base64_image = encode_image(image)


    # Specify model
    model = "pixtral-12b-2409"

    # Initialize the Mistral client
    client = Mistral(api_key=api_key)
    # inputs = processor(prompt, image, return_tensors="pt").to("cuda:0")

    # Generate a response from the model
    messages = [
        {
            "role": "user",
            "content": [
                {"type": "text", "text": "What's in this image?"},
                {
                    "type": "image_url",
                    "image_url": f"data:image/jpeg;base64,{base64_image}",
                },
            ],
        }
    ]

    # Stream, ref.: https://github.com/mistralai/client-python/blob/main/examples/chatbot_with_streaming.py
    stream_response = client.chat.stream(model=model, messages=messages)

    answer = ""
    for chunk in stream_response:
        response = chunk.data.choices[0].delta.content
        if response is not None:
            # print(response, end="", flush=True)
            answer += response
            yield answer

    # bulk inference:
    # Get the chat response
    # chat_response = client.chat.complete(
    #    model=model,
    #    messages=messages
    # )
    # Print the content of the response
    # print(chat_response.choices[0].message.content)
    # result = chat_response.choices[0].message.content
    # return result

    # streamer = TextIteratorStreamer(processor, **{"skip_special_tokens": True, "skip_prompt": True, 'clean_up_tokenization_spaces':False,})
    # generation_kwargs = dict(inputs, streamer=streamer, max_new_tokens=1024, do_sample=False, temperature=0.0, eos_token_id=processor.tokenizer.eos_token_id,)

    # thread = Thread(target=model.generate, kwargs=generation_kwargs)
    # thread.start()

    # for new_text in streamer:
    #    buffer += new_text
    #    yield buffer


chatbot = gr.Chatbot(scale=1, placeholder=PLACEHOLDER)
chat_input = gr.MultimodalTextbox(
    interactive=True,
    file_types=["image"],
    placeholder="Enter message or upload figure...",
    show_label=False,
)

with gr.Blocks(
    fill_height=True,
) as demo:
    gr.ChatInterface(
        fn=bot_streaming,
        title="LISA-Vision-test",
        examples=[
            {"text": "What does this figure describe?", "files": ["./sample1.png"]},
            {
                "text": "ocr the table in figure and put in Markdown format",
                "files": ["./sample2.png"],
            },
            {
                "text": "Explain this XRD figure to me in details.",
                "files": ["./sample3.png"],
            },
        ],
        description="Try VLM (Vision Language Model) to chat with characters. Upload an image and start chatting, or just try one of the examples below. If you don't upload an image, you'll get an error.",
        stop_btn="Stop Generation",
        multimodal=True,
        textbox=chat_input,
        chatbot=chatbot,
        cache_examples=False,
        examples_per_page=3,
    )

demo.queue(api_open=False)
demo.launch(share=False)