app.py

import gradio as gr
from transformers import AutoModelForCausalLM, AutoTokenizer
from PIL import Image
import re
import copy
import secrets
from pathlib import Path

# Constants
BOX_TAG_PATTERN = r"<box>([\s\S]*?)</box>"
PUNCTUATION = "!\"#$%&'()*+,-./:;<=>?@[\\]^_`{|}~"

# Initialize model and tokenizer
tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen-VL-Chat-Int4", trust_remote_code=True)
model = AutoModelForCausalLM.from_pretrained("Qwen/Qwen-VL-Chat-Int4", device_map="auto", trust_remote_code=True).eval()

def format_text(text):
    """Format text for rendering in the chat UI."""
    lines = text.split("\n")
    lines = [line for line in lines if line != ""]
    count = 0
    for i, line in enumerate(lines):
        if "```" in line:
            count += 1
            items = line.split("`")
            if count % 2 == 1:
                lines[i] = f'<pre><code class="language-{items[-1]}">'
            else:
                lines[i] = f"<br></code></pre>"
        else:
            if i > 0:
                if count % 2 == 1:
                    line = line.replace("`", r"\`")
                    line = line.replace("<", "&lt;")
                    line = line.replace(">", "&gt;")
                    line = line.replace(" ", "&nbsp;")
                    line = line.replace("*", "&ast;")
                    line = line.replace("_", "&lowbar;")
                    line = line.replace("-", "&#45;")
                    line = line.replace(".", "&#46;")
                    line = line.replace("!", "&#33;")
                    line = line.replace("(", "&#40;")
                    line = line.replace(")", "&#41;")
                    line = line.replace("$", "&#36;")
                lines[i] = "<br>" + line
    text = "".join(lines)
    return text

def get_chat_response(chatbot, task_history):
    """Generate a response using the model."""
    chat_query = chatbot[-1][0]
    query = task_history[-1][0]
    history_cp = copy.deepcopy(task_history)
    full_response = ""

    history_filter = []
    pic_idx = 1
    pre = ""
    for i, (q, a) in enumerate(history_cp):
        if isinstance(q, (tuple, list)):
            q = f'Picture {pic_idx}: <img>{q[0]}</img>'
            pre += q + '\n'
            pic_idx += 1
        else:
            pre += q
            history_filter.append((pre, a))
            pre = ""
    history, message = history_filter[:-1], history_filter[-1][0]
    
    inputs = tokenizer.encode_plus(message, return_tensors='pt')
    outputs = model.generate(inputs['input_ids'], max_length=150, num_beams=4, length_penalty=2.0, early_stopping=True)
    response = tokenizer.decode(outputs[0], skip_special_tokens=True)
    
    task_history.append((message, response))
    chatbot.append((format_text(message), format_text(response)))

    return chatbot, task_history


def handle_text_input(history, task_history, text):
    """Handle text input from the user."""
    task_text = text
    if len(text) >= 2 and text[-1] in PUNCTUATION and text[-2] not in PUNCTUATION:
        task_text = text[:-1]
    history = history + [(format_text(text), None)]
    task_history = task_history + [(task_text, None)]
    return history, task_history, ""

def handle_file_upload(history, task_history, file):
    """Handle file upload from the user."""
    history = history + [((file.name,), None)]
    task_history = task_history + [((file.name,), None)]
    return history, task_history

def clear_input():
    """Clear the user input."""
    return gr.update(value="")

def clear_history(task_history):
    """Clear the chat history."""
    task_history.clear()
    return []

def handle_regeneration(chatbot, task_history):
    """Handle the regeneration of the last response."""
    print("Regenerate clicked")
    print("Before:", task_history, chatbot)
    if not task_history:
        return chatbot
    item = task_history[-1]
    if item[1] is None:
        return chatbot
    task_history[-1] = (item[0], None)
    chatbot_item = chatbot.pop(-1)
    if chatbot_item[0] is None:
        chatbot[-1] = (chatbot[-1][0], None)
    else:
        chatbot.append((chatbot_item[0], None))
    print("After:", task_history, chatbot)
    return get_chat_response(chatbot, task_history)

chatbot = []
task_history = []

def main_function(text, image):
    global chatbot, task_history
    if text:
        chatbot, task_history = handle_text_input(chatbot, task_history, text)
    if image:
        chatbot, task_history = handle_file_upload(chatbot, task_history, image)
    chatbot, task_history = get_chat_response(chatbot, task_history)
    formatted_response = chatbot[-1][1]  # Get the latest response from the chatbot
    return formatted_response

def clear_history_fn():
    global chatbot, task_history
    chatbot.clear()
    task_history.clear()
    return "History cleared."

# Custom CSS
css = '''
    .gradio-container {
        max-width: 800px !important;
    }
'''

with gr.Blocks(css=css) as demo:
    gr.Markdown("# Qwen-VL-Chat Bot")
    gr.Markdown(
        "## Developed by Keyvan Hardani (Keyvven on [Twitter](https://twitter.com/Keyvven))\n"
        "Special thanks to [@Artificialguybr](https://twitter.com/artificialguybr) for the inspiration from his code.\n"
        "### Qwen-VL: A Multimodal Large Vision Language Model by Alibaba Cloud\n"
    )
    chat_interface = gr.Interface(
        fn=main_function,
        inputs=[
            gr.components.Textbox(lines=2, label='Input'),  # Update here
            gr.components.Image(type='filepath', label='Upload Image')  # Update here
        ],
        outputs='text',
        live=True,
        layout='vertical',
        theme=None,
        css=css
    ).launch()
    gr.Markdown("### Key Features:\n- **Strong Performance**: Surpasses existing LVLMs on multiple English benchmarks including Zero-shot Captioning and VQA.\n- **Multi-lingual Support**: Supports English, Chinese, and multi-lingual conversation.\n- **High Resolution**: Utilizes 448*448 resolution for fine-grained recognition and understanding.")
    
    demo.add_button("🧹 Clear History", clear_history_fn)

demo.launch(share=True)