※ 本記事はClaude Sonnet4を活用して作成しています。

はじめに

最近海外に駐在しており、その国では様々な買い物を全てネットで完結できる文化がすでに醸成されている状態でした。
そういった状況から、ネットショッピング文化を活用して買い物の完全デジタル管理を実現したいと考え、色々とアプリを作っています。今回は、スマホから気軽に購入商品の画像をアップロードするだけで、AIが自動的に食材名・価格・規格を抽出し、CSV形式でデータ蓄積するシステムを構築したので紹介します。

システム概要

目的: 北京での買い物データを完全管理し、料理メニュー考案・家計管理を効率化
フロー: Slackに画像投稿 → VLMで画像分析 → LLMでデータ構造化 → CSV保存

主要技術スタック

Flask: WebサーバーとしてSlackイベントを受信
OpenRouter API: VLM（Qwen 2.5-VL 7B）+ LLM（GPT-4o mini）
Slack Events API: スマホからの画像アップロード検知

実装のポイント

1. 二段階AI処理による高精度抽出

class FoodExtractor:
    def extract_and_structure(self, image: Image.Image) -> tuple:
        # Step 1: VLMで画像から商品情報を抽出
        raw_output = self.vlm.extract_food_info(image)
        # Step 2: LLMで構造化JSONに変換
        structured_output = self.llm.process_food_data(raw_output)
        return raw_output, structured_output

VLMプロンプト例:

この画像に表示されている食材や食品の名前、価格、規格を日本語で詳しく
リストアップしてください。中国語の商品名も併記してください。

2. 重複処理防止機能

Slackの仕様により同じファイルアップロードが複数回検知される問題への対策：

# 処理済み・処理中ファイルIDの管理
processed_files: Set[str] = set()
processing_files: Set[str] = set()

def is_file_already_processed(file_id: str) -> bool:
    return file_id in processed_files or file_id in processing_files

3. メモリ効率的な画像処理

長時間稼働するFlaskアプリでのメモリリーク対策：

def process_slack_image(file_info: dict, ...):
    try:
        # 画像サイズ制限
        if image.size[0] > max_size[0] or image.size[1] > max_size[1]:
            image.thumbnail(max_size, Image.Resampling.LANCZOS)
        
        # 処理実行
        extractor = FoodExtractor(openrouter_api_key)
        raw_output, structured_output = extractor.extract_and_structure(image)
        
    finally:
        # 確実なメモリ解放
        if image:
            image.close()
            del image
        gc.collect()

技術選択の理由

OpenRouter + Qwen 2.5-VL

無料で利用可能
処理速度が高速（ローカルLLMより圧倒的に速い）
中国語商品名の認識精度が良好

Slack インターフェース

スマホからの操作が簡単（画像アップロードのみ）
クロスプラットフォーム対応
既存のコミュニケーションツールとして馴染みがある

実際の使用感

メリット

✅ 操作が超簡単: スマホから画像をSlackにアップロードするだけ
✅ 精度が高い: VLM→LLMの二段階処理で構造化データが確実に生成
✅ データ蓄積: CSV形式で家計管理・メニュー考案に活用可能

改善点

⚠️ 処理完了通知がない: ユーザーは処理成功を確認できない
⚠️ エラー時の通知なし: 失敗時の原因が分からない

コード全体

完全なFlaskアプリケーション

import os
import json
import re
import requests
import pandas as pd
from PIL import Image
import base64
import io
from datetime import datetime
from typing import List, Dict, Set
from flask import Flask, request, jsonify
import logging
from dotenv import load_dotenv
import gc

# .envファイルを読み込み
load_dotenv()

# ログ設定
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = Flask(__name__)

# 処理済みファイルIDを追跡するセット（メモリ内）
processed_files: Set[str] = set()
processing_files: Set[str] = set()

class OpenRouterVLM:
    """OpenRouter APIを使用したVLMクラス"""
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://openrouter.ai/api/v1/chat/completions"
        self.vlm_model = "qwen/qwen-2.5-vl-7b-instruct"
    
    def image_to_base64(self, image: Image.Image) -> str:
        buffer = io.BytesIO()
        try:
            max_size = (1024, 1024)
            if image.size[0] > max_size[0] or image.size[1] > max_size[1]:
                image.thumbnail(max_size, Image.Resampling.LANCZOS)
            
            image.save(buffer, format='PNG')
            img_bytes = buffer.getvalue()
            base64_str = base64.b64encode(img_bytes).decode('utf-8')
            buffer.close()
            return base64_str
        except Exception as e:
            buffer.close()
            raise e
    
    def extract_food_info(self, image: Image.Image) -> str:
        try:
            base64_image = self.image_to_base64(image)
            
            messages = [{
                "role": "user",
                "content": [{
                    "type": "text",
                    "text": "この画像に表示されている食材や食品の名前、価格、規格を日本語で詳しくリストアップしてください。中国語の商品名も併記してください。"
                }, {
                    "type": "image_url",
                    "image_url": {"url": f"data:image/png;base64,{base64_image}"}
                }]
            }]
            
            response = requests.post(
                url=self.base_url,
                headers={
                    "Authorization": f"Bearer {self.api_key}",
                    "Content-Type": "application/json"
                },
                json={
                    "model": self.vlm_model,
                    "messages": messages,
                    "temperature": 0.1,
                    "max_tokens": 1000
                },
                timeout=60
            )
            
            if response.status_code == 200:
                result = response.json()
                return result['choices'][0]['message']['content']
            else:
                return f"VLM API Error: {response.status_code}"
        except Exception as e:
            return f"VLM処理エラー: {e}"
        finally:
            if 'base64_image' in locals():
                del base64_image
            gc.collect()

class OpenRouterLLM:
    """OpenRouter APIを使用したLLMクラス"""
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://openrouter.ai/api/v1/chat/completions"
        self.llm_model = "openai/gpt-4o-mini"
    
    def process_food_data(self, raw_text: str) -> str:
        prompt = f"""
以下のテキストから食材名と価格情報を抽出して、JSON形式で整理してください：

{{
  "items": [
    {{
      "name": "食材名（中国語）",
      "price": 価格（数値のみ）,
      "amount": "数量（例：200g）",
      "category": "食材カテゴリ"
    }}
  ]
}}

分析対象テキスト:
{raw_text}

JSONのみを返答してください。
"""
        
        try:
            response = requests.post(
                url=self.base_url,
                headers={
                    "Authorization": f"Bearer {self.api_key}",
                    "Content-Type": "application/json"
                },
                json={
                    "model": self.llm_model,
                    "messages": [{"role": "user", "content": prompt}],
                    "temperature": 0.1
                },
                timeout=30
            )
            
            if response.status_code == 200:
                result = response.json()
                return result['choices'][0]['message']['content']
            else:
                return f"LLM API Error: {response.status_code}"
        except Exception as e:
            return f"LLM処理エラー: {e}"

# 以下、Flask エンドポイントとヘルパー関数...
@app.route('/slack/events', methods=['POST'])
def handle_slack_event():
    try:
        data = request.json
        
        if data.get('type') == 'url_verification':
            return jsonify({'challenge': data['challenge']})
        
        if data.get('type') == 'event_callback':
            event = data.get('event', {})
            
            if event.get('type') == 'message' and event.get('files'):
                for file_info in event['files']:
                    if file_info.get('mimetype', '').startswith('image/'):
                        file_id = file_info.get('id')
                        
                        if file_id and not is_file_already_processed(file_id):
                            process_slack_image(file_info)
        
        return jsonify({'status': 'ok'})
    except Exception as e:
        logger.error(f"Slackイベント処理エラー: {e}")
        return jsonify({'error': str(e)}), 500

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=3000, debug=True)

エラーハンドリングの詳細

主要な対策項目

1. Slack重複イベント対策

# ファイルIDベースの重複防止
def is_file_already_processed(file_id: str) -> bool:
    return file_id in processed_files or file_id in processing_files

2. メモリリーク対策

# 画像オブジェクトの確実な削除
finally:
    if image:
        image.close()
        del image
    gc.collect()

3. API接続エラー対策

# タイムアウト設定とステータスコードチェック
response = requests.post(..., timeout=60)
if response.status_code != 200:
    return f"API Error: {response.status_code}"

4. 長期稼働時のメモリ管理

def cleanup_processed_files():
    if len(processed_files) > 1000:
        processed_files.clear()
        processing_files.clear()

まとめ

スマホ一つで買い物データを完全自動化できる実用的なシステムが完成しました。中国のデジタル決済文化に最適化されており、実際の生活で活用できています。

今後は処理完了通知機能やエラーレポート機能を追加し、ユーザビリティをさらに向上させる予定です。

環境変数設定

export SLACK_BOT_TOKEN='xoxb-your-slack-bot-token'
export OPENROUTER_API_KEY='sk-or-your-openrouter-api-key'
export CSV_FILENAME='food_data.csv'
export PORT=3000