MiniGPT-4のAPIを実装する。　プログラムでマルチモーダルを自由に操作する。

こちら、v2へアップデートされたのでこのままでは動きません。以下参考記事です。

MiniGPT-4のデモは動かしていましたが、プログラムから自由に呼び出して画像の説明をさせたかったので、APIを実装しました。画像ファイルでも、カメラキャプチャ画像でも自在に説明できます。

環境設定

過去記事でMiniGPT-4を動かしていれば、以下のFastAPI関係の追加で動きます。前提としてCUDA-Toolkitが正しく動いている必要があります。

pip install fastapi
pip install pydantic

オリジナルのDemoコードはgradioで動いています。このＤemoコードを利用してAPIを実装します。本来であればminigpt4ディレクトリ内のコードを解析して再構築するべきでしょうが、時間節約のためにＤemoコードからgradioを削除し、FastAPIによるASGI化でPOST/GETで容易にアクセスできるようにしました。

プログラムの構造と今回の実装の構成

MiniGPT-4の推論過程は3つに別れています。
１　画像のアップロード
２　質問をする
３　回答を得る
マルチモーダルなので画像だけではなくチャット会話機能も持ち合わせているため、アップロードと質問/回答が独立しています。今回は各機能を個別に確認できる仕様でサーバ側とクライアント側のプログラムを作成しています。また、1回の操作で回答まで得られるアプリコードも作成しました。FastAPI側はサーバなので独立したPCで操作が可能です。

APIサーバ側プログラム

import argparse
import os
import random
import numpy as np
import torch
import torch.backends.cudnn as cudnn
from minigpt4.common.config import Config
from minigpt4.common.dist_utils import get_rank
from minigpt4.common.registry import registry
from minigpt4.conversation.conversation import Chat, CONV_VISION_Vicuna0, CONV_VISION_LLama2
# imports modules for registration
from minigpt4.datasets.builders import *
from minigpt4.models import *
from minigpt4.processors import *
from minigpt4.runners import *
from minigpt4.tasks import *

def parse_args():
    parser = argparse.ArgumentParser(description="Demo")
    parser.add_argument("--cfg-path", required=True, help="path to configuration file.")
    parser.add_argument("--gpu-id", type=int, default=0, help="specify the gpu to load the model.")
    parser.add_argument( "--options", nargs="+",
                                                        help="override some settings in the used config, the key-value pair "
                                                                    "in xxx=yyy format will be merged into config file (deprecate), "
                                                                    "change to --cfg-options instead.", )
    args = parser.parse_args()
    return args

def setup_seeds(config):
    seed = config.run_cfg.seed + get_rank()
    random.seed(seed)
    np.random.seed(seed)
    torch.manual_seed(seed)
    cudnn.benchmark = False
    cudnn.deterministic = True

# ===========   Model Initialization  ===========
conv_dict = {'pretrain_vicuna0': CONV_VISION_Vicuna0,  'pretrain_llama2': CONV_VISION_LLama2}
args = parse_args()
cfg = Config(args)
model_config = cfg.model_cfg
model_config.device_8bit = args.gpu_id
model_cls = registry.get_model_class(model_config.arch)
model = model_cls.from_config(model_config).to('cuda:{}'.format(args.gpu_id))
CONV_VISION = conv_dict[model_config.model_type]
vis_processor_cfg = cfg.datasets_cfg.cc_sbu_align.vis_processor.train
vis_processor = registry.get_processor_class(vis_processor_cfg.name).from_config(vis_processor_cfg)
chat = Chat(model, vis_processor, device='cuda:{}'.format(args.gpu_id))
print('Initialization Finished')

# ===============     FastAPI  ==============
from fastapi import FastAPI, UploadFile
from fastapi.responses import HTMLResponse
from PIL import Image
from io import BytesIO
from pydantic import BaseModel
#from typing import Optional, List, Dict

img_list=[]
chatbot=[]
chat_state=""

app = FastAPI()

class ChatRequest(BaseModel):      #@app.post("/ask/")用
    user_message:  str
    chatbot: list
    
class AnswerRequest(BaseModel):#@app.post("/answer/")用
     user_message : str
     chatbot:list
     num_beams:int
     temperature:float

@app.post("/llm_reset/")
def  reset():
    chat_state.messages = []
    if img_list is not None:
        img_list=[]
        chatbot=[]
    return {"message": "Reset", "chatbot": chatbot, "chat_state":chat_state, "img_list":img_list}

@app.post("/uploadfile/")
def uploadfile(file: UploadFile ):
    global  chat_state, img_list
    if file:
            image_data = file.file.read()
            gr_img = Image.open(BytesIO(image_data))  # バイナリデータをPIL形式に変換
    else:
        return {"message":"Error","chatbot": chatbot,"llm_message":"non","chat_state":chat_state ,"img_list":img_list}#"Error"
    chat_state = CONV_VISION.copy()
    img_list=[]
    chatbot=[]
    llm_message = chat.upload_img(gr_img, chat_state, img_list)
    return {"message": "Image received","chatbot": chatbot,"chat_state":chat_state ,"img_list":img_list}

@app.post("/ask/")
def ask_question(chat_request: ChatRequest):
    global  chat_state ,img_list
    user_message = chat_request.user_message
    chatbot = chat_request.chatbot
    if len(user_message) == 0:
        return  {"message": "ask_non","chatbot":chatbot, "chat_state":chat_state,"img_list":img_list}
    chat.ask(user_message, chat_state)
    chatbot = chatbot + [[user_message, None]]
    return {"message": "ask_completed","chatbot":chatbot, "chat_state":chat_state,"img_list":img_list}

@app.post("/answer/")
def answer(Ans_request: AnswerRequest):
    global chat_state,img_list
    user_message = Ans_request.user_message
    chatbot = Ans_request.chatbot
    num_beams = Ans_request.num_beams
    temperature = Ans_request.temperature

    llm_message = chat.answer(conv=chat_state,
                              img_list=img_list,
                              num_beams=num_beams,
                              temperature=temperature,
                              max_new_tokens=300,
                              max_length=2000)[0]
    chatbot[-1][1] = llm_message
    print("chatbot=",chatbot)
    return {"message": "ask_completed", "chatbot":chatbot,  "chat_state":chat_state, "sysem":llm_message,"img_list":img_list}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8001)

前半はモデルのイニシャライズなどです。後半にFastAPIによるインターフェースがあります。

@app.post("/llm_reset/")
リセットです。イメージが残っている場合はチャットのみのリセットになります。
@app.post("/uploadfile/")
画像ファイルのアップロードです。今回のAPIではファイルパスから画像データを読み込んでいます。バイナリデータで受け渡しをする場合は以下を変更します。

            image_data = file.file.read()
            gr_img = Image.open(BytesIO(image_data))  # バイナリデータをPIL形式に変換

@app.post("/ask/")
質問をする時のAPIです。class ChatRequest(BaseModel)に受け取取るデータの型定義が記述されています。
@app.post("/answer/")
回答を得るためのAPIです。この処理で初めてLLMによる推論が動きます。class AnswerRequest(BaseModel):に受け取取るデータの型定義が記述されています。

個別にテストする

アップロードのテスト

import requests
from PIL import Image
from io import BytesIO

# 送信するPIL形式の画像データ
image_file_path = '00016-331097358.png'

# FastAPIエンドポイントのURL
url = 'http://192.168.5.100:8001/uploadfile/'  # FastAPIサーバーのURLに合わせて変更してください

# ファイルをアップロードするためのリクエストを作成
files = {'file': open(image_file_path, 'rb')}

# POSTリクエストを送信
response = requests.post(url, files=files)

# レスポンスを表示     return {"message": "Image received ","chatbot": chatbot}
if response.status_code == 200:
    result = response.json()
    print("サーバーからの応答message:", result.get("message"))
    print("サーバーからの応答chatbot:", result.get("chatbot"))
    print("サーバーからの応答chat_state:", result.get("chat_state"))
    print("サーバーからの応答img_list:", result.get("img_list"))
else:
    print("リクエストが失敗しました。ステータスコード:", response.status_code)

簡単なアップロードテストのためのコードです。
@app.post("/uploadfile/")をテスト出来ます。正しくアップロードされるとサーバーからの応答messageが200になり、上手く動かない場合は404、422、500などが帰って来ます。

質問のテスト

import requests
from PIL import Image
from io import BytesIO

# 送信するデータを準備
data = {
    "user_message": "Please describe this picture in as much detail as possible.",
    "chatbot":[]
}
# MiniGPT-4 FastAPIエンドポイントのURL
url = 'http://0.0.0.0:8001/ask/'  # FastAPIサーバーのURLに合わせて変更してください

# POSTリクエストを送信
response = requests.post(url, json=data)
# レスポンスを表示 return {"message": "ask_completed ","chatbot":chatbot}

if response.status_code == 200:
    result = response.json()
    print("サーバーからの応答message:", result.get("message"))
    print("サーバーからの応答chatbot:", result.get("chatbot"))
    print("サーバーからの応答chat_state:", result.get("chat_state"))
    print("サーバーからの応答img_list:", result.get("img_list"))
else:
    print("リクエストが失敗しました。ステータスコード:", response.status_code)

LLMに対してテキストで質問を投げるコードです。レスポンスはアップロードテストと同様ですが、その他の応答も出力されるので、どのような処理がされているのか、わかると思います。

回答を得る

import requests
from PIL import Image
from io import BytesIO

# 送信するデータを準備
data = {
    "user_message": "Please describe this picture in as much detail as possible.",
    "chatbot":[['Please describe this picture in as much detail as possible.', None]],
    "num_beams":1,
    "temperature" :0.7,
}
# FastAPIエンドポイントのURL
url = 'http://192.168.5.100:8001/answer/'  # FastAPIサーバーのURLに合わせて変更してください

# POSTリクエストを送信
response = requests.post(url, json=data)

# レスポンスを表示
if response.status_code == 200:
    result = response.json()
    print("サーバーからの応答message:", result.get("message"))
    print("サーバーからの応答chat_state:", result.get("chat_state"))
    print("サーバーからの応答chatbot:", result.get("chatbot"))
    print("サーバーからの応答sysem:", result.get("sysem"))
    
    
else:
    print("リクエストが失敗しました。ステータスコード:", response.status_code)

質問に対する回答を問い合わせします。ここで推論が始まるので時間がかかります。

チャット

質問と回答を繰り返せばチャットになります。回答のリクエストを繰り返すと詳しい説明を得たり出来ます。

すべての処理を続けて行う

import requests
from PIL import Image
from io import BytesIO
import json
from time import sleep

def main():
    url_base="http://192.168.5.100:8001/"
    user_message="Please describe this picture in as much detail as possible."
    image_file_path = '00016-331097358.png'
    read_img(user_message,image_file_path,url_base)

#アップロードから回答を得るまで、纏めて処理する関数
def read_img(user_message,image_file_path,url_base):
    status_code, chat_state, chatbot, img_list= upload(image_file_path, url_base)
    print("Upload status_code:",status_code)
    status_code, chat_state, chatbot,  img_list= ask(user_message,chatbot, url_base)
    print("Ask status_code:",status_code)
    status_code, chat_state, chatbot, sysem, img_list = ans(user_message,chatbot, url_base)
    print("Ans status_code:",status_code)
    print(sysem)

#個別の関数
#upload
def upload(image_file_path,url_base):
    url = url_base+'uploadfile/'  # FastAPIサーバーのURLに合わせて変更してください
    files = {'file': open(image_file_path, 'rb')}
    # POSTリクエストを送信
    response = requests.post(url, files=files)
    if response.status_code == 200:
        result = response.json()
        return response.status_code, result.get("chat_state"), result.get("chatbot"), result.get("img_list")  
    else:
        return response.status_code, "non", "non","non"
#ask
def ask(user_message,chatbot,url_base):
    data = {
        "user_message": user_message,
        "chatbot": chatbot,
    }
    url = url_base+'ask/'  # FastAPIサーバーのURLに合わせて変更してください
    response = requests.post(url, json=data)
    if response.status_code == 200:
        result = response.json()
        return response.status_code, result.get("chat_state"), result.get("chatbot"),result.get("img_list")
    else:
        return response.status_code,"non","non","nin"
#ans
def ans(user_message,chatbot,url_base):
    data = {
        "user_message":user_message,
        "chatbot":chatbot,
        "num_beams":1,
        "temperature" :0.7,
    }
    url =url_base+'answer/'  # FastAPIサーバーのURLに合わせて変更してください
    response = requests.post(url, json=data)
    if response.status_code == 200:
        result = response.json()
        return response.status_code,  result.get("chat_state"), result.get("chatbot"), result.get("sysem"), result.get("img_list")
    else:
        return response.status_code,"non","non","non"

if __name__ == '__main__':
    main()

アップロード、質問、回答を順次実行すると回答まで得られます。上記コードではmainから
read_img(user_message,image_file_path,url_base)
を呼び出して一気に回答までさせるようにしました。日本語で回答を得るためには質問に「日本語で回答して」を入れてaskとanswerを呼び出せば出来ます。ただし、とても時間がかかるので、他の高速な日本語LLMを利用した方が良さそうです。

制限事項

コードを読んでいただければわかると思いますが、FastAPIサーバの各サービス間でchat_state とimg_listがglobal変数として共有されています。このため、複数のクライアントからの同時接続では問題が生じる可能性があります。回避するために各サービスのreturnでchat_state とimg_listを返しています。このリターン値を次に呼び出すサービスで受け取れるように変更する必要があります。サービスごとのテストを行うためにはサーバ側で持たざるを得ないので、現状のようになっています。
変更は容易か？
なんですが、img_listは単純です。型をlistとして定義してやればできます。問題はchat_stateで、どのような型指定をしても正しく受け渡しが出来ませんでした。どなたか方法をご教示いただけると助かります。

まとめ

すでにV2がリリースされ、OpenAIからはCharGPT4-V2が出てこのモデルを使う機会は少ないかなと思いますが、シンプルな画像情報取得用のマルチモーダルとしては、まだまだ利用価値はあるように思います。

おまけ。

いつもの画像をAPI経由で解説させた結果です。

回答を2回続けてリクエストしたときの回答です。
日本語訳もつけておきます（MniniGPT-4での翻訳ではありません）

英語の出力
The image is of a beautiful garden with a pink flower bed in the foreground, surrounded by a white gazebo and a path leading to a blue sky in the background. The flowers in the pink bed are in full bloom, with pink petals and blue stamens. The gazebo is covered in pink and white flowers, with a blue sky in the background. The path is lined with pink flowers and leads to the gazebo.
The colors used in the image are predominantly pastel, with shades of pink, white, and blue. The flowers in the pink bed are a deep pink color, while the gazebo is covered in a mix of pink and white flowers. The blue sky in the background provides a nice contrast to the pastel colors in the foreground.
The composition of the image is well balanced, with the gazebo and the path leading to it serving as the main focal points. The pink flowers in the foreground add a pop of color and create a sense of depth and dimensionality. The overall mood of the image is peaceful and serene, with the gentle curves of the gazebo and the winding path creating a sense of tranquility.

日本語
前景にピンク色の花壇があり、その周囲を白い展望台で囲み、背景に青い空に続く小道がある美しい庭園をイメージしています。ピンクの花壇には、ピンクの花びらと青い雄しべが咲き誇ります。青空を背景に、ガゼボはピンクと白の花で覆われています。道にはピンク色の花が並び、展望台へと続いています。 画像で使用されている色は主にパステルカラーで、ピンク、白、青の色合いが含まれています。ピンクのベッドの花は濃いピンク色で、ガゼボはピンクと白の花が混ざり合った花で覆われています。背景の青空が前景のパステルカラーと素晴らしいコントラストを生み出しています。 ガゼボとそこに続く道を主な焦点として、バランスのとれた画面構成となっています。前景のピンクの花が色彩にポップさを加え、奥行きと立体感を生み出します。イメージ全体の雰囲気は穏やかで穏やかで、東屋の緩やかな曲線と曲がりくねった小道が静けさを感じさせます。