backend-QWEN2-VL.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "provenance": [],
      "gpuType": "T4"
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    },
    "accelerator": "GPU"
  },
  "cells": [
    {
      "cell_type": "code",
      "metadata": {
        "id": "5BvNgFgk5dE2"
      },
      "source": [
        "# !pip install anvil-uplink"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "EML6wBYQ5fiM"
      },
      "source": [
        "import anvil.server"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "-cuHx3s3Vm52"
      },
      "source": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "MA9-qSCOSckw"
      },
      "source": [
        "anvil.server.connect(\"server_A54UXZNMAOF44RJNVYRNG6UE-QIJV2LK4TV7ORECE\")"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [],
      "metadata": {
        "id": "eqlYtPYGFOIK"
      }
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "Z9FGGe-2-V79"
      },
      "source": [
        "# !pip install git+https://github.com/huggingface/transformers\n",
        "\n",
        "# !pip install qwen_vl_utils\n"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "import torch"
      ],
      "metadata": {
        "id": "0PaHS-egb4KL"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [],
      "metadata": {
        "id": "VOXPxtylJK72"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "# Initialize an empty list to store the conversation history\n",
        "conversation_history = []"
      ],
      "metadata": {
        "id": "3gQyuiKeCifQ"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "images"
      ],
      "metadata": {
        "id": "RV5ahj_7FX82"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "from PIL import Image\n",
        "import io\n",
        "\n",
        "# Define a function to receive the image\n",
        "@anvil.server.callable\n",
        "def process_uploaded_image(image_file):\n",
        "    # Convert the Anvil image file to a PIL Image\n",
        "    image_bytes = io.BytesIO(image_file.get_bytes())\n",
        "    image = Image.open(image_bytes)\n",
        "\n",
        "    # Process the image as needed (e.g., save, display, analyze, etc.)\n",
        "    image.save('/content/test.jpg')  # Example: Display the image\n",
        "\n",
        "    # Return a response (optional)\n",
        "    return \"Image uploaded and processed successfully!\"\n"
      ],
      "metadata": {
        "id": "bgHYDJY-FY9T",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "219bc127-5196-4d4b-c6c2-a4c99f17afdb"
      },
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Warning: a callable with the name 'process_uploaded_image' has already been registered (previously by '__main__.process_uploaded_image' now by '__main__.process_uploaded_image').\n"
          ]
        }
      ]
    },
    {
      "cell_type": "markdown",
      "source": [],
      "metadata": {
        "id": "Dpb7ZlHwMlPk"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "from PIL import Image\n",
        "img=Image.open('/content/test.jpg')\n",
        "@anvil.server.callable\n",
        "def prompt_text(query,img=img):\n",
        "    from transformers import Qwen2VLForConditionalGeneration, AutoTokenizer, AutoProcessor\n",
        "    from qwen_vl_utils import process_vision_info\n",
        "    import torch\n",
        "    torch.cuda.empty_cache()\n",
        "    model = Qwen2VLForConditionalGeneration.from_pretrained(\n",
        "    \"Qwen/Qwen2-VL-2B-Instruct\", torch_dtype=\"auto\", device_map=\"auto\")\n",
        "    processor = AutoProcessor.from_pretrained(\"Qwen/Qwen2-VL-2B-Instruct-GPTQ-Int4\")\n",
        "\n",
        "\n",
        "    # Append the new user input to the conversation history\n",
        "    conversation_history.append(\n",
        "        {\n",
        "            \"role\": \"user\",\n",
        "            \"content\": [\n",
        "                {\"type\": \"image\", \"image\": img},\n",
        "                {\"type\": \"text\", \"text\": query}\n",
        "\n",
        "            ],\n",
        "        }\n",
        "    )\n",
        "\n",
        "    # Prepare the messages with the full conversation history\n",
        "    messages = conversation_history[:]\n",
        "\n",
        "    # Process the messages\n",
        "    text = processor.apply_chat_template(\n",
        "        messages, tokenize=False, add_generation_prompt=True\n",
        "    )\n",
        "    image_inputs, video_inputs = process_vision_info(messages)\n",
        "    inputs = processor(\n",
        "        text=[text],\n",
        "        images=image_inputs,\n",
        "        videos=video_inputs,\n",
        "        padding=True,\n",
        "        return_tensors=\"pt\",\n",
        "    )\n",
        "    inputs = inputs.to('cuda')\n",
        "\n",
        "    # Inference: Generation of the output\n",
        "    generated_ids = model.generate(**inputs, max_new_tokens=128)\n",
        "    generated_ids_trimmed = [\n",
        "        out_ids[len(in_ids):] for in_ids, out_ids in zip(inputs.input_ids, generated_ids)\n",
        "    ]\n",
        "    output_text = processor.batch_decode(\n",
        "        generated_ids_trimmed, skip_special_tokens=True, clean_up_tokenization_spaces=False\n",
        "    )\n",
        "\n",
        "    # Append the model's response to the conversation history\n",
        "    conversation_history.append(\n",
        "        {\n",
        "            \"role\": \"system\",\n",
        "            \"content\": {\"type\": \"text\", \"text\": output_text[0]},\n",
        "        }\n",
        "    )\n",
        "    ans=output_text[0]\n",
        "    del generated_ids, generated_ids_trimmed, output_text\n",
        "    del image_inputs, video_inputs, inputs, text, messages\n",
        "    del model, processor\n",
        "    return ans"
      ],
      "metadata": {
        "id": "T-jILwLaMlty"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "torch.cuda.empty_cache()"
      ],
      "metadata": {
        "id": "GLBOaRRKG8Y4"
      },
      "execution_count": null,
      "outputs": []
    }
  ]
}