BLIP-2 による Visual Question Answering（ソースコードと実行結果）

REM Python をシステム領域にインストール
winget install --scope machine --id Python.Python.3.12 -e --silent
REM Python のパス設定
set "PYTHON_PATH=C:\Program Files\Python312"
set "PYTHON_SCRIPTS_PATH=C:\Program Files\Python312\Scripts"
echo "%PATH%" | find /i "%PYTHON_PATH%" >nul
if errorlevel 1 setx PATH "%PATH%;%PYTHON_PATH%" /M >nul
echo "%PATH%" | find /i "%PYTHON_SCRIPTS_PATH%" >nul
if errorlevel 1 setx PATH "%PATH%;%PYTHON_SCRIPTS_PATH%" /M >nul

winget install --scope machine Codeium.Windsurf -e --silent

# BLIP-2による画像質問応答システム
# BLIP-2: Bootstrapping Language-Image Pre-training with Frozen Image Encoders and Large Language Models
# 出典: Li, J., Li, D., Savarese, S., & Hoi, S. (2023). arXiv preprint arXiv:2301.12597.
# 機能: 画像とテキストの理解により画像に関する質問に自然言語で回答

import torch
from transformers import Blip2Processor, Blip2ForConditionalGeneration
import cv2
import tkinter as tk
from tkinter import filedialog
import urllib.request
from PIL import Image
import os
import warnings

# モデル定義
MODELS = [
    ("Salesforce/blip2-opt-2.7b", "BLIP-2 with OPT-2.7B"),
    ("Salesforce/blip2-flan-t5-xl", "BLIP-2 with Flan-T5-XL")
]

DEFAULT_MODEL = "Salesforce/blip2-opt-2.7b"
MAX_SIZE = 384

# 生成パラメータ
MAX_NEW_TOKENS = 50
NUM_BEAMS = 5
REPETITION_PENALTY = 1.5
LENGTH_PENALTY = 1.0
NO_REPEAT_NGRAM_SIZE = 3

SAMPLE_URLS = [
    "https://raw.githubusercontent.com/opencv/opencv/master/samples/data/fruits.jpg",
    "https://raw.githubusercontent.com/opencv/opencv/master/samples/data/messi5.jpg",
    "https://raw.githubusercontent.com/opencv/opencv/master/samples/data/aero3.jpg",
    "https://upload.wikimedia.org/wikipedia/commons/3/3a/Cat03.jpg"
]

QUESTIONS = [
    "What is in this image?",
    "Describe this image in detail.",
    "What are the colors and atmosphere of this image?",
    "What objects can you see in this image?",
    "What is happening in this image?",
    "Describe the main subject of this image."
]

print("BLIP-2 Visual Question Answering システム")

device = "cuda" if torch.cuda.is_available() else "cpu"
print(f"使用デバイス: {device}")

print("\n利用可能なBLIP-2モデル:")
for i, (model_name, description) in enumerate(MODELS, 1):
    print(f"{i}. {description}")
print("Enter: デフォルト(BLIP-2 with OPT-2.7B)を使用")

choice = input("\nモデル番号 (Enterでデフォルト): ")
if choice == "":
    model_name = DEFAULT_MODEL
    model_description = "BLIP-2 with OPT-2.7B"
else:
    try:
        idx = int(choice) - 1
        if 0 <= idx < len(MODELS):
            model_name = MODELS[idx][0]
            model_description = MODELS[idx][1]
        else:
            print("無効な選択です。デフォルトを使用します。")
            model_name = DEFAULT_MODEL
            model_description = "BLIP-2 with OPT-2.7B"
    except ValueError:
        print("無効な選択です。デフォルトを使用します。")
        model_name = DEFAULT_MODEL
        model_description = "BLIP-2 with OPT-2.7B"

print(f"\n選択されたモデル: {model_description}")
print("モデルを読み込み中...")

try:
    warnings.filterwarnings("ignore", category=FutureWarning)

    processor = Blip2Processor.from_pretrained(model_name)

    if device == "cuda":
        model = Blip2ForConditionalGeneration.from_pretrained(
            model_name,
            torch_dtype=torch.float16,
            device_map="auto"
        )
    else:
        model = Blip2ForConditionalGeneration.from_pretrained(model_name)
        model.to(device)

    print("モデルの読み込み完了\n")

except Exception as e:
    print(f"モデル読み込み中にエラーが発生しました: {e}")
    exit(1)

results = []

def preprocess_image(image_input):
    """画像の前処理"""
    if isinstance(image_input, str):
        image = Image.open(image_input).convert('RGB')
    else:
        image = Image.fromarray(cv2.cvtColor(image_input, cv2.COLOR_BGR2RGB))

    if max(image.size) > MAX_SIZE:
        image.thumbnail((MAX_SIZE, MAX_SIZE), Image.Resampling.LANCZOS)

    return image

def generate_answer(image, question):
    """BLIP-2による回答生成"""
    try:
        inputs = processor(image, question, return_tensors="pt")

        if device == "cuda":
            inputs = {k: v.to(device, torch.float16 if v.dtype == torch.float32 else v.dtype)
                     for k, v in inputs.items()}
        else:
            inputs = {k: v.to(device) for k, v in inputs.items()}

        with torch.no_grad():
            outputs = model.generate(
                **inputs,
                max_new_tokens=MAX_NEW_TOKENS,
                min_length=5,
                do_sample=False,
                num_beams=NUM_BEAMS,
                repetition_penalty=REPETITION_PENALTY,
                length_penalty=LENGTH_PENALTY,
                early_stopping=True,
                no_repeat_ngram_size=NO_REPEAT_NGRAM_SIZE,
                pad_token_id=processor.tokenizer.pad_token_id,
                eos_token_id=processor.tokenizer.eos_token_id
            )

        generated_text = processor.decode(outputs[0], skip_special_tokens=True)

        # プロンプト除去
        if question in generated_text:
            answer = generated_text.replace(question, "").strip()
        else:
            answer = generated_text.strip()

        if len(answer.strip()) < 3:
            answer = "I apologize, but I couldn't generate an appropriate response."

        return answer

    except Exception as e:
        print(f"回答生成中にエラーが発生しました: {e}")
        return "An error occurred. Please try with a different image or question."

def select_question():
    """質問の選択"""
    print("\n質問を選択:")
    for i, q in enumerate(QUESTIONS, 1):
        print(f"{i}. {q}")
    print(f"{len(QUESTIONS)+1}. 自由入力")

    while True:
        try:
            q_choice = input("選択: ")
            q_choice = int(q_choice)
            if 1 <= q_choice <= len(QUESTIONS):
                return QUESTIONS[q_choice-1]
            elif q_choice == len(QUESTIONS)+1:
                return input("Enter your question: ")
            else:
                print(f"1から{len(QUESTIONS)+1}の間で選択してください。")
        except ValueError:
            print("数字を入力してください。")

def process_image(img_input, img_path=None):
    """画像処理メイン"""
    image = preprocess_image(img_path or img_input)
    question = select_question()

    print("回答生成中...")
    answer = generate_answer(image, question)
    result = f"\nQuestion: {question}\nAnswer: {answer}\n"
    print(result)
    results.append(result)

    display_img = cv2.imread(img_path) if img_path else img_input
    cv2.imshow('Image - Press any key to close', display_img)
    cv2.waitKey(0)

def main():
    print("\n画像入力方法を選択:")
    print("0: 画像ファイル")
    print("1: カメラ")
    print("2: サンプル画像")

    choice = input("選択: ")

    if choice == '0':
        root = tk.Tk()
        root.withdraw()
        file_paths = filedialog.askopenfilenames(
            title="画像ファイルを選択",
            filetypes=[("画像ファイル", "*.jpg *.jpeg *.png *.bmp *.tiff")]
        )
        root.destroy()

        for file_path in file_paths:
            print(f"\n処理中: {os.path.basename(file_path)}")
            process_image(None, file_path)

    elif choice == '1':
        cap = cv2.VideoCapture(0)
        if not cap.isOpened():
            print("カメラを開けませんでした。")
            return

        print("\nスペースキー: 撮影して分析")
        print("Qキー: 終了")

        while True:
            ret, frame = cap.read()
            if not ret:
                print("カメラからフレームを取得できませんでした。")
                break

            cv2.imshow('Camera - Press SPACE to capture, Q to quit', frame)
            key = cv2.waitKey(1) & 0xFF

            if key == ord(' '):
                print("\n画像を撮影しました")
                process_image(frame.copy())
            elif key == ord('q'):
                break
        cap.release()

    elif choice == '2':
        print("\nサンプル画像:")
        sample_names = ["fruits.jpg", "messi5.jpg", "aero3.jpg", "Cat03.jpg"]
        for i, name in enumerate(sample_names, 1):
            print(f"{i}. {name}")

        while True:
            try:
                sample_choice = int(input("選択: ")) - 1
                if 0 <= sample_choice < len(sample_names):
                    break
                else:
                    print(f"1から{len(sample_names)}の間で選択してください。")
            except ValueError:
                print("数字を入力してください。")

        url = SAMPLE_URLS[sample_choice]
        filename = f"sample_{sample_choice}.jpg"

        print(f"サンプル画像をダウンロード中: {sample_names[sample_choice]}")
        try:
            urllib.request.urlretrieve(url, filename)
            process_image(None, filename)
            if os.path.exists(filename):
                os.remove(filename)
        except Exception as e:
            print(f"サンプル画像のダウンロードに失敗しました: {e}")

    cv2.destroyAllWindows()

if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        print("\nプログラムが中断されました。")
    except Exception as e:
        print(f"予期しないエラーが発生しました: {e}")

    # 結果保存
    if results:
        with open('result.txt', 'w', encoding='utf-8') as f:
            f.write("=== BLIP-2 VQA 結果 ===\n")
            f.write(f"使用モデル: {model_name}\n")
            f.write(f"デバイス: {device}\n")
            f.write("=" * 50 + "\n")
            for result in results:
                f.write(result)
        print(f"\n結果をresult.txtに保存しました ({len(results)}件)")

    print("\nプログラムを終了します")