はじめに
WED株式会社で ML エンジニアをしています、ishi2kiです。
当社では、レシート買取アプリ「ONE」を開発・運用しています。 アプリユーザーは、レシート買取ミッションに参加し、対象商品のレシートをアップロードすることでお金がもらえます。
このアプリの運用課題として、ユーザーが同一のレシートを使い回していないか、人目で検品する必要があることが挙げられていました。
しかし、アップロードされるレシート枚数が多いと、その分検品にかかる人件費も増えてきてしまいます。そのため、生成 AI (Gemini) でレシート画像を比較し同じレシートか否かを判定 (重複判定) することで、検品作業を自動化することを目指しました。
要件は以下の通りです。
- 検品対象のレシートペアの重複判定を Gemini で行う。
- 許容できない処理コストになることを防ぐため、処理前に想定処理コストのチェックを行い、一定額以上の場合警告を出す。
- 非エンジニアの派遣社員が作業を行うため、Slack から処理を実行できるようにする。また、処理結果はスプレッドシートに出力する。
※レシートの印字を直接比較しないのは、以下のように写真の撮影範囲が異なる場合があるためです。
アーキテクチャ
Slack と重複判定プログラムの間にある Cloud Run Service は、本記事で紹介するプログラムの他にも、当社で開発したいくつかの Jobs や Airflow の実行に対応しており、Slack とプログラムを繋ぐ役割を担っています。
実装
想定処理コストのチェック
def calc_cost(pair_num: int) -> float:
"""
重複検知にかかるコストを概算する
Args:
pair_num (int): 重複ペア数
Returns:
float: コスト (ドル)
"""
# 処理コスト = (画像1枚あたりのコスト) * (画像2枚) * (重複ペア数) / (バッチ処理使用による50％オフ)
cost = COST_PER_IMAGE * 2 * pair_num / 2
return cost
正確に計算するなら、プロンプトの料金も加算する必要がありますが、画像の料金と比較したら無視できる程度なので計算式には入れていません。
画像 1 枚あたりのコスト (COST_PER_IMAGE) は、こちらに記載があります。
記事執筆当時は $0.00002 でした (5 万ペアの判定を行っても 1 ドルしかかかりません！！)。
Gemini による重複判定
今回の要件では、Gemini からの即時のレスポンスは必要としていません。
そこで、Batch Prediction を使うことにしました。
通常の Gemini の推論は 1 つのリクエストに対して 1 つのレスポンスが即時に返ってくるのに対し、Batch Prediction はまとめて複数のリクエストを投げ、全ての処理が終わった後にまとめてレスポンスが返ってきます。
使える場面は限られますが、Gemini の利用コストが半分になるというメリットがあります。
下は、Batch Prediction で重複判定を行うクラスです。
import json
import tempfile
import time
import vertexai
from fastapi import HTTPException
from vertexai.batch_prediction import BatchPredictionJob
import utils
class DuplicateDetector:
def __init__(self):
self.bucket = "my_bucket"
self.request_jsonl_path = "my_request_jsonl_path"
self.output_path = "my_output_path"
self.error_count = 0
def create_request_jsonl(self, target_receipts: list[tuple[str, str]]):
"""
リクエストをjsonlファイルにしてGCSに保存
Args:
target_receipts (list[tuple[str, str]]): 検品対象のレシート画像のペア (GCSのURI)
"""
prompt = """
Are these receipts same?
Output is only "YES" or "NO".
"""
with tempfile.NamedTemporaryFile(delete=True) as tf:
file_name = tf.name
with open(file_name, "a") as f:
for uri1, uri2 in target_receipts:
request_dict = {
"request": {
"contents": [
{
"role": "user",
"parts": [
{"text": prompt},
{"fileData": {"fileUri": uri1, "mimeType": "image/*"}},
{"fileData": {"fileUri": uri2, "mimeType": "image/*"}},
],
}
]
}
}
print(json.dumps(request_dict), file=f)
# jsonlファイルをGCSに保存
utils.copy_to_gcs(file_name, self.request_jsonl_path)
def request_predict(self):
"""
Batch Predictionの実行
Returns:
str: Batch Predictionの結果を指すパス
"""
vertexai.init(project="my_project", location="us-central1")
request_uri = f"gs://{self.bucket}/{self.request_jsonl_path}"
output_uri_prefix = f"gs://{self.bucket}/{self.output_path}"
batch_prediction_job = BatchPredictionJob.submit(
source_model="gemini-1.5-flash-002",
input_dataset=request_uri,
output_uri_prefix=output_uri_prefix,
)
# 処理が終わるまで待機
while not batch_prediction_job.has_ended:
time.sleep(5)
batch_prediction_job.refresh()
if not batch_prediction_job.has_succeeded:
raise HTTPException(status_code=500, detail="Batch prediction job failed")
return batch_prediction_job.output_location
Batch Prediction では、リクエストをまとめた jsonl ファイルを作成し、GCS に置く必要があります。また、今回のように画像をリクエストに含める場合は、画像も GCS に置いて URI をリクエストに含めます。
処理が終わったら、処理結果を格納する GCS のパスを受け取り、後続の処理につなげます。
スプレッドシートへの出力
指定されたスプレッドシートに新規シートを追加することで結果を出力します。
重複と判定されたレシートの、レシート ID (receipt_id) ペアとレシートの印字 (content) ペアを持った DataFrame を、set_with_dataframeを使って一括で書き込むことができます。
import gspread
from google.auth import default
from gspread_dataframe import set_with_dataframe
from pandas import DataFrame
def create_duplicates_sheet(duplicate_receipt_pairs: DataFrame, spreadsheet_url: str, new_sheet_name: str):
credentials, _ = default()
gc = gspread.authorize(credentials)
spreadsheet = gc.open_by_url(spreadsheet_url)
new_sheet = spreadsheet.add_worksheet(new_sheet_name, rows=1, cols=10)
# 結果 (DataFrame) をシートに書き込む
set_with_dataframe(new_sheet, duplicate_receipt_pairs)
なお、スプレッドシートをプログラムで操作するためには、プログラムを実行するサービスアカウントをファイルやフォルダの共有者に含める必要があるので注意してください。
運用効果
Gemini の判定ミスがあるため、完全な検品の自動化はせずに、一種のフィルターとしての運用を行うことにしました。
具体的には、以下の手順で検品を行うことにしました。
- 検品対象のレシートペア (同一の店舗、同一の購入時間などの条件を満たすペア) を収集する。
- 1 のペアに対して、重複判定プログラムで判定を行う。
- 2 で「重複ではない」とみなされたペアは検品は行わず、「重複である」とみなされたペアに対してのみ人目での検品を行う。
あるミッションでは、1 の検品対象のペアが 1 ヶ月で 165 件ありました。
これを全て人目で検品するとなると、それなりの時間がかかります。
しかし、重複判定プログラムでフィルターを行うことにより、検品対象を 7 ペアまで減らすことができました。
重複判定の結果例
1 つ目はパット見同じですが、購入時間とレジ担当者が違います。
まとめ
今回、Gemini を使った重複レシートの判定プログラムを作成しました。
撮影範囲が違うなど、印字での比較が難しいレシートペアでも正しく判定することができ、期待以上の性能を見せてくれました。
導入効果においても、検品が必要なペア数が 7/165 (4％) になり、負担を大幅に減らすことができました。
今後も Gemini などの生成 AI を積極的に使っていきたいです。
WED での他の Gemini 活用例
Discussion