非同期基盤実装ガイド（Celery + Redis）

ステータス: ✅ 実装済み（2026-06-07 Phase 1-3） 作成日: 2026-05-26 関連: 01_DECISIONS/atelier-kyo-manager/2026-05-26_リファクタリング調査_T01-X01-ARCH.md

なぜ必要か

現在、以下の処理が全て同期（画面が固まる）:

• パイプライン実行: 画像収集→背景除去→AI生成（1商品で数十秒〜数分）
• スクレイピング: 価格取得・在庫チェック（1件3-5秒、数十件で数分）
• LLM呼び出し: AI分類・回答生成（3-10秒）
• バッチ処理: 複数商品のパイプライン（シーケンシャル実行）

商品数が50を超えたら、同期では実用上問題になる。

Celery + Redis 構成

[Flask App] → [Celery Client] → [Redis (Broker)] → [Celery Worker]
                                    ↓
                              [Redis (Result Backend)]

インフラ要件

• Redis: ローカル開発は無料（apt install redis）、本番はRedis Cloud無料枠またはVPS上
• Celery: pip install celery[redis]
• 追加プロセス: Celeryワーカー1プロセス（常時起動）

実装ステップ

Step 1: インストール

pip install celery[redis] redis
sudo apt install redis-server  # WSL2

Step 2: app/core/celery_app.py 作成

from celery import Celery
from app.config.config import AppConfig

def make_celery(app):
    celery = Celery(
        app.import_name,
        broker=app.config.get("CELERY_BROKER_URL", "redis://localhost:6379/0"),
        backend=app.config.get("CELERY_RESULT_BACKEND", "redis://localhost:6379/0"),
    )
    celery.conf.update(app.config)
    return celery

Step 3: app/__init__.py に統合

celery = make_celery(app)
app.celery = celery

Step 4: タスク定義（既存サービスをラップ）

# app/tasks/pipeline_tasks.py
from app import celery

@celery.task(bind=True)
def run_pipeline_task(self, product_id, site_key=""):
    from app.services.pipeline_service import PipelineService
    svc = PipelineService()
    return svc.run(product_id, site_key)

@celery.task(bind=True)
def fetch_stock_task(self, url):
    from app.services.price_scraper import PriceScraper
    scraper = PriceScraper()
    return scraper.fetch_cached(url)

Step 5: ルート修正（同期→非同期）

# Before: result = PipelineService().run(product_id)
# After:
task = run_pipeline_task.delay(product_id, site_key)
# 即座にtask.idを返す。進捗はフロントエンドがポーリング

Step 6: 進捗確認API

@bp.get("/api/tasks/<task_id>/status")
def task_status(task_id):
    result = celery.AsyncResult(task_id)
    return {"state": result.state, "result": result.result}

対象処理（優先順位）

優先度	処理	現在の所要時間	非同期化の効果
1	パイプライン実行	30秒-3分	画面即レスポンス
2	バッチパイプライン	N×30秒	並列化で3-5倍高速
3	全商品在庫チェック	N×3秒	バックグラウンド実行
4	AI説明文生成	5-10秒	画面即レスポンス
5	BUYMA価格スクレイピング	N×3秒	バックグラウンド実行

テスト方針

• Celeryタスクのテストは CELERY_ALWAYS_EAGER=True で同期実行
• 既存テストに影響なし（タスク定義がラップするだけ）
• 新規テスト: タスク状態遷移・エラー時のリトライ

参考

• Celery公式ドキュメント: https://docs.celeryq.dev/
• Flask + Celery: https://flask.palletsprojects.com/patterns/celery/
• 関連調査: U01 ai_llm_controller, S05 pipeline_service, S06 price_scraper