大規模言語モデルに目を与える

こちらの続きです。

https://qiita.com/taka_yayoi/items/b88d47bcfabf2f6bfa48

元々はこちらを実現したかったのでした。カメラモジュールがあるので、それをマルチモーダルモデルの入力にしたらどうだろうと、よくある発想かもしれませんが実装したら面白かったです。

ベースにしたのはこちらです。

https://qiita.com/taka_yayoi/items/8f81d96753ed645f3e26

元記事は、テキストのプロンプトと画像のURLを受け取る実装になっていましたが、Raspberry Piの画像を直接受け取る形に変更します。

カメラモジュールの操作に関してはこちらが非常に助けになりました。ありがとうございます。

https://kosh.dev/article/14/

# Databricks側の実装

モデル入力のインタフェースを変更します。画像をBase64エンコードを受け取るようにします。[こちら](https://stackoverflow.com/questions/29104107/upload-image-using-post-form-data-in-python-requests)を参考にさせてもらいました。

```py
import mlflow
import pandas as pd

from transformers import LlavaNextProcessor, LlavaNextForConditionalGeneration
import torch
from PIL import Image
import requests

import io
import base64                  

class Model(mlflow.pyfunc.PythonModel):
    def __init__(self):
        self.processor = LlavaNextProcessor.from_pretrained(model_id)

        self.model = LlavaNextForConditionalGeneration.from_pretrained(
            model_id,
            torch_dtype=torch.float16, 
            low_cpu_mem_usage=True,
            cache_dir="/Volumes/users/takaaki_yayoi/llava_test")
        
        self.model.to("cuda:0")

    def predict(self, context, model_input):
        processor = self.processor
        model = self.model

        results = []

        for mi in model_input:

            im_b64 = mi[0]
            prompt = mi[1]

            # convert it into bytes  
            img_bytes = base64.b64decode(im_b64.encode('utf-8'))

            # convert bytes data to PIL Image object
            image = Image.open(io.BytesIO(img_bytes))
    
            # Prepare inputs
            inputs = processor(prompt, image, return_tensors='pt').to(0, torch.float16)

            # Generate and store response
            output = model.generate(**inputs, max_new_tokens=150, do_sample=False)
            result = (processor.decode(output[0], skip_special_tokens=True))

            results.append(result)

        return results


# 関数をモデルとして保存
with mlflow.start_run():
    mlflow.pyfunc.log_model(
        "model", 
        python_model=Model(),
        pip_requirements=['transformers==4.39.0', 'mlflow==2.11.3', 'tensorflow', 'torch', 'Image', 'requests'],
        signature=signature
        )
    run_id = mlflow.active_run().info.run_id
```

これでGPUサービングエンドポイントにデプロイします。[前回同様](https://qiita.com/taka_yayoi/items/b88d47bcfabf2f6bfa48#%E3%83%A2%E3%83%87%E3%83%AB%E3%82%B5%E3%83%BC%E3%83%93%E3%83%B3%E3%82%B0%E3%82%A8%E3%83%B3%E3%83%89%E3%83%9D%E3%82%A4%E3%83%B3%E3%83%88%E3%81%AE%E7%A2%BA%E8%AA%8D)にURLとパーソナルアクセストークンを準備しておきます。
![Screenshot 2024-09-03 at 14.31.08.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/8179dc8b-976e-0aa7-6dd3-c30ac0146bfd.png)

# Raspberry Pi側の実装

前回のものを拡張して、カメラの画像をモデルサービングエンドポイントに送信するようにします。あと、[こちら](https://forums.raspberrypi.com/viewtopic.php?t=367558)を参考にpicamera2を使えるように仮想環境を作り直しました。

```sh
python -m venv --system-site-packages env
source env/bin/activate
```
```sh
/home/yayoi/Documents/env/bin/pip install openai requests scipy pyopenjtalk
```

コードの実態はこちら。2.4Hzの通信をしているので、画像をそのまま送信するとそれがボトルネックになってしまっていたので、画像サイズを小さくしています。

```py:multimodal.py
import os

# 音声発話向けライブラリ
import pyopenjtalk
from scipy.io import wavfile
import numpy as np

import subprocess

# カメラ向けライブラリ
from picamera2 import Picamera2

import requests
import json
import re
import base64
from io import BytesIO
from PIL import Image

# レスポンスの読み上げを格納する音声ファイル
wave_file = "/home/yayoi/Documents/dbx/talk.wav"

# テキストを発話
def synthesize_voice(text):
    x, sr = pyopenjtalk.tts(text)
    wavfile.write(wave_file, sr, x.astype(np.int16))

    cmd = f'/usr/bin/aplay {wave_file}'
    process = (subprocess.Popen(cmd, stdout=subprocess.PIPE,
                           shell=True).communicate()[0]).decode('utf-8')

# PIL ImageをBase64に変換
def pil_to_base64(img, format="jpeg"):
    buffer = BytesIO()
    img.save(buffer, format)
    img_str = base64.b64encode(buffer.getvalue()).decode("ascii")

    return img_str

# How to get your Databricks token: https://docs.databricks.com/en/dev-tools/auth/pat.html
DATABRICKS_TOKEN = os.environ.get('DATABRICKS_TOKEN')
# Alternatively in a Databricks notebook you can use this:
# DATABRICKS_TOKEN = dbutils.notebook.entry_point.getDbutils().notebook().getContext().apiToken().get()

print("##### camera initialize.")

# 画像保存先
img_file = "/home/yayoi/Documents/capture.jpg"

# 撮影
camera = Picamera2()
config_capture = camera.create_still_configuration()
camera.configure(config_capture)
camera.start()

image = camera.capture_image()
image = image.resize((160, 128)) # サイズ変更
im_b64 = pil_to_base64(image) # Base64エンコード

print("##### photo captured.")

prompt = '
 <image>\n画像を簡潔に説明してください
'

# 画像を保存
image.save(img_file, 'JPEG')

data = {
  "inputs": [(im_b64, prompt)]
}

#print(data)

headers = {"Context-Type": "text/json", "Authorization": f"Bearer {DATABRICKS_TOKEN}"}

print("##### sending request.")


response = requests.post(
  url="<エンドポイントのURL>", json=data, headers=headers
)

#response = json.dumps(response.json(), ensure_ascii=False)

response_list = response.json()["predictions"]
response_text = re.sub(r"\[?.*\]", "", response_list[0])
print(response_text)

synthesize_voice(response_text)
```

# 動作確認

動かします。

```sh
/home/yayoi/Documents/env/bin/python /home/yayoi/Documents/multimodal.py 
```

<iframe width="560" height="315" src="https://www.youtube.com/embed/apbymFTIewQ?si=-NOJO8seMOqMZYWg" title="YouTube video player" frameborder="0" allow="accelerometer; autoplay; clipboard-write; encrypted-media; gyroscope; picture-in-picture; web-share" referrerpolicy="strict-origin-when-cross-origin" allowfullscreen></iframe>

キャプチャされた画像はこちらです。

![capture.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/61a1e67f-ee15-4649-f9fe-ab1ce44a6efa.jpeg)

やっぱり、色々なインタフェースと組み合わせるとさらに興味深いものになりますね。

### はじめてのDatabricks

[はじめてのDatabricks](https://qiita.com/taka_yayoi/items/8dc72d083edb879a5e5d)

### Databricks無料トライアル

[Databricks無料トライアル](https://databricks.com/jp/try-databricks)