app.py

import os
import gradio as gr
import torch
from PIL import Image
from transformers import AutoModel, AutoTokenizer

# Notes:
# - This demo runs on CPU for broader compatibility. It may be slow compared to GPU.
# - If you have a GPU, you can set device="cuda" and possibly use torch_dtype=torch.bfloat16.
# - MiniCPM-V-4_5 uses trust_remote_code; ensure you trust the source.
# - The model expects multi-modal messages in a chat-like format: [{'role': 'user', 'content': [image, text]}]
# - For multi-turn chat, we persist history in Gradio state and pass it back to model.chat.

MODEL_ID = os.environ.get("MINICPM_MODEL_ID", "openbmb/MiniCPM-V-4_5")
DEVICE = "cpu"  # Force CPU per user request
DTYPE = torch.float32  # CPU-friendly dtype

# Lazy global variables (loaded on first launch)
_tokenizer = None
_model = None

def load_model():
    global _tokenizer, _model
    if _model is None or _tokenizer is None:
        # Some platforms require setting no_mmap or local_files_only as needed; adjust if necessary.
        _model = AutoModel.from_pretrained(
            MODEL_ID,
            trust_remote_code=True,
            attn_implementation="sdpa",  # sdpa is fine on CPU; avoid eager per model note
            torch_dtype=DTYPE
        )
        _model = _model.eval().to(DEVICE)
        _tokenizer = AutoTokenizer.from_pretrained(MODEL_ID, trust_remote_code=True)
    return _model, _tokenizer

def format_history(history):
    """
    Convert Gradio-style chat history into model's expected message format.
    history: list of tuples (user_text, assistant_text) where user_text may have an <image> placeholder handled separately.
    We will store messages in a structured way in state to retain images explicitly instead of parsing text.
    This function is not used directly; we keep the raw message structure in state for fidelity.
    """
    return history

def predict(image, user_message, history_state, enable_thinking=False, stream=False):
    """
    image: PIL.Image or None
    user_message: str
    history_state: list of dicts in MiniCPM format [{'role': 'user'|'assistant', 'content':[...]}]
    """
    model, tokenizer = load_model()

    # Initialize history if empty
    msgs = history_state if isinstance(history_state, list) else []

    # Build the current user content payload
    # The model expects a list mixing image(s) and text; include only provided items.
    content = []
    if image is not None:
        if image.mode != "RGB":
            image = image.convert("RGB")
        content.append(image)
    if user_message and user_message.strip():
        content.append(user_message.strip())

    if len(content) == 0:
        return gr.update(), msgs, "Please provide an image and/or a message."

    msgs = msgs + [{'role': 'user', 'content': content}]

    # Run generation
    try:
        # model.chat returns either an iterator (when stream=True) or a string
        answer = model.chat(
            msgs=msgs,
            tokenizer=tokenizer,
            enable_thinking=bool(enable_thinking),
            stream=bool(stream)
        )

        if stream:
            # Concatenate streamed text
            generated = []
            for chunk in answer:
                generated.append(chunk)
                yield "\n".join(["".join(generated)]), msgs, None
            final_text = "".join(generated)
        else:
            final_text = answer

        # Append assistant message back into msgs
        msgs = msgs + [{"role": "assistant", "content": [final_text]}]

        # Return final
        yield final_text, msgs, None

    except Exception as e:
        yield gr.update(), msgs, f"Error: {e}"

def clear_state():
    return None, [], None

with gr.Blocks(title="MiniCPM-V-4_5 CPU Gradio Demo") as demo:
    gr.Markdown("# MiniCPM-V-4_5 (CPU) Demo")
    gr.Markdown("Upload an image (optional) and ask a question. Multi-turn chat is supported. Running on CPU may be slow.")

    with gr.Row():
        with gr.Column(scale=1):
            image_in = gr.Image(type="pil", label="Image (optional)")
            user_in = gr.Textbox(label="Your Message", placeholder="Ask a question about the image or general query...", lines=3)
            with gr.Row():
                think_chk = gr.Checkbox(label="Enable Thinking Mode", value=False)
                stream_chk = gr.Checkbox(label="Stream Output", value=False)
            with gr.Row():
                submit_btn = gr.Button("Send", variant="primary")
                clear_btn = gr.Button("Clear")

        with gr.Column(scale=2):
            chat_out = gr.Chatbot(label="Chat", type="messages", height=450, avatar_images=(None, None))
            status_box = gr.Markdown("", visible=True)

    # Hidden state: we store the raw MiniCPM messages, not just text pairs
    state_msgs = gr.State([])

    def on_submit(image, message, enable_thinking, stream, msgs):
        # Kick off streaming generator
        # We'll display only last exchange in Chatbot. Convert msgs to Chatbot-friendly format when yielding.
        # For Chatbot display, we reconstruct from msgs
        def format_for_chatbot(msgs_local):
            chat_pairs = []
            # Collect pairs by scanning msgs in order
            user_tmp = None
            for m in msgs_local:
                if m["role"] == "user":
                    # Convert content to displayable string for Chatbot
                    parts = []
                    for c in m["content"]:
                        if isinstance(c, Image.Image):
                            parts.append("[Image]")
                        else:
                            parts.append(str(c))
                    user_tmp = " ".join(parts).strip() or "[Image]"
                elif m["role"] == "assistant":
                    assistant_text = " ".join([str(x) for x in m["content"]]) if m["content"] else ""
                    if user_tmp is None:
                        chat_pairs.append((None, assistant_text))
                    else:
                        chat_pairs.append((user_tmp, assistant_text))
                        user_tmp = None
            return chat_pairs

        gen = predict(image, message, msgs, enable_thinking, stream)
        if stream:
            for partial_text, updated_msgs, err in gen:
                # Build display history from updated_msgs + current partial response
                display_msgs = updated_msgs.copy()
                # Don't duplicate assistant msg until finalized; just show in Chatbot via the last pair
                chat_history = format_for_chatbot(display_msgs)
                if chat_history and isinstance(partial_text, str) and partial_text:
                    if chat_history and (not chat_history[-1][1] or chat_history[-1][1] == ""):
                        # replace last tuple assistant part
                        u, _ = chat_history[-1]
                        chat_history[-1] = (u, partial_text)
                    else:
                        # append live pair
                        last_user = None
                        for m in reversed(display_msgs):
                            if m["role"] == "user":
                                parts = []
                                for c in m["content"]:
                                    if isinstance(c, Image.Image):
                                        parts.append("[Image]")
                                    else:
                                        parts.append(str(c))
                                last_user = " ".join(parts).strip() or "[Image]"
                                break
                        chat_history.append((last_user, partial_text))
                status = "" if not err else f"{err}"
                yield chat_history, updated_msgs, status, gr.update(value=None), gr.update(value=None)
        else:
            for final_text, updated_msgs, err in gen:
                chat_history = []
                # Build chat history from updated_msgs
                def format_for_chatbot_final(msgs_local):
                    pairs = []
                    u_txt = None
                    for m in msgs_local:
                        if m["role"] == "user":
                            parts = []
                            for c in m["content"]:
                                if isinstance(c, Image.Image):
                                    parts.append("[Image]")
                                else:
                                    parts.append(str(c))
                            u_txt = " ".join(parts).strip() or "[Image]"
                        elif m["role"] == "assistant":
                            a_txt = " ".join([str(x) for x in m["content"]]) if m["content"] else ""
                            if u_txt is None:
                                pairs.append((None, a_txt))
                            else:
                                pairs.append((u_txt, a_txt))
                                u_txt = None
                    return pairs

                chat_history = format_for_chatbot_final(updated_msgs)
                status = "" if not err else f"{err}"
                yield chat_history, updated_msgs, status, gr.update(value=None), gr.update(value=None)

    submit_btn.click(
        on_submit,
        inputs=[image_in, user_in, think_chk, stream_chk, state_msgs],
        outputs=[chat_out, state_msgs, status_box, user_in, image_in]
    )

    clear_btn.click(
        fn=clear_state,
        inputs=[],
        outputs=[user_in, state_msgs, status_box]
    ).then(
        lambda: [],
        inputs=None,
        outputs=chat_out
    )

    # Preload model on app start (optional; keeps UI responsive on first query)
    demo.load(lambda: "Model loading on CPU... Please wait a moment.", outputs=status_box).then(
        lambda: (load_model() or True) and "Model loaded. Ready!",
        outputs=status_box
    )

if __name__ == "__main__":
    # Set server_name="0.0.0.0" to expose externally if needed.
    demo.queue(max_size=8, concurrency_count=1).launch()