File size: 6,276 Bytes
c3e9bbf |
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 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 |
# 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)
|