0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@NKKTechGlobal(NKKTech Global)

FastAPI + Celeryで作る、非同期タスク処理を伴う重たいAIエージェントのバックエンド構築

0
Posted at

FastAPI + Celeryで作る、非同期タスク処理を伴う重たいAIエージェントのバックエンド構築

Feature (17).png

はじめに

2026年、AIエージェントは「単純な一問一答」から、複数のステップを踏んで推論・検索・ツール実行を行う「複雑なワークフロー」へと進化しました。しかし、こうした重たいタスクをAPIの1リクエスト内で完結させようとすると、HTTPタイムアウトやレスポンス遅延によるユーザー体験の悪化が避けられません。

本記事では、ハノイを拠点に高度なAIソリューションを開発する NKKTech Software の知見をもとに、FastAPI と Celery を組み合わせた、非同期タスク処理基盤の構築手法を解説します。

1. システムアーキテクチャ

リクエストを受け取るAPI層（FastAPI）と、重たいAI処理を実行するワーカー層（Celery）を分離し、メッセージブローカー（Redis）を介して通信します。

2. 技術スタックの選定理由

重たいAI推論や複雑なエージェントワークフローを支えるため、以下の「3種の神器」を採用しています。

  • FastAPI:
    • PythonのモダンなWebフレームワーク。
    • 非同期I/O（Async/Await） をネイティブサポートしており、モデルの推論待ちが発生しても他のリクエストをブロックせず、大量のタスク受付を最小限のリソース（低オーバーヘッド）で処理できます。
  • Celery:
    • Python製分散タスクキューのデファクトスタンダード。
    • 数十分に及ぶような重たいAI処理をバックグラウンドへ逃がすことができ、自動再試行（Retry）、タスクの優先度管理、ワーカーの横展開（スケールアウト） が容易です。
  • Redis:
    • メモリ内データストア。
    • Celeryのメッセージブローカー（タスクの運び役）および、処理結果の一時的な保存先（Result Backend）として、極めて高速に動作します。

3. 実装のポイント

① プロジェクト構造

FastAPIとCeleryを疎結合に保ち、メンテナンス性を高めるための標準的なディレクトリ構成です。

.
├── main.py          # FastAPI本体（リクエスト受付、状態確認API）
├── celery_app.py    # Celeryの初期化設定（Broker/Backend接続定義）
├── tasks.py         # 重たいAI処理のロジック（Celeryタスク）
├── requirements.txt # 依存ライブラリ
└── docker-compose.yml # Redis, API, Workerの環境一括構築用

② Celeryの設定 (celery_app.py)

Celeryのインスタンスを生成し、メッセージブローカー（タスクの運び役）および結果ストアとしてのRedisとの接続を定義します。

celery_app.py
from celery import Celery

# Celeryインスタンスの作成
# broker: タスクキューの送信先 (Redis)
# backend: タスクの実行結果やステータスを保存する場所 (Redis)
app = Celery(
    "ai_agent_tasks",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/0"
)

# 運用上の詳細設定
app.conf.update(
    result_expires=3600,        # 結果の保持期間（1時間）。AIの重たい処理結果を一時保存
    task_serializer='json',     # セキュリティと互換性のためにJSONを使用
    accept_content=['json'],
    result_serializer='json',
    task_track_started=True,    # タスクが開始されたことをステータスとして追跡
    worker_prefetch_multiplier=1 # AIタスクは重いため、1ワーカーが1タスクずつ集中して処理する設定
)

④ FastAPIエンドポイントの実装 (main.py)

FastAPI側では、重たいAI処理を直接実行せず、Celeryの delay() メソッドを使用してタスクをキューに投入します。これにより、クライアントには即座に task_id を返し、HTTPタイムアウトを完全に回避します。

main.py
from fastapi import FastAPI, HTTPException
from celery.result import AsyncResult
from tasks import run_heavy_ai_agent
from celery_app import app as celery_app

app = FastAPI(title="AI Agent Async API")

@app.post("/agent/run", status_code=202)
async def trigger_agent(query: str):
    """
    AIエージェントの実行をリクエストするエンドポイント。
    タスクをキューに投入し、即座に受付完了を返します。
    """
    # .delay() を呼ぶことで非同期にCelery Workerへタスクを送信
    task = run_heavy_ai_agent.delay(query)
    
    return {
        "task_id": task.id, 
        "status": "accepted",
        "message": "AIエージェントの処理を開始しました。"
    }

@app.get("/agent/status/{task_id}")
async def get_status(task_id: str):
    """
    指定されたtask_idの実行状態と結果を取得するエンドポイント。
    """
    # Redisからタスクの状態を取得
    task_result = AsyncResult(task_id, app=celery_app)
    
    response = {
        "task_id": task_id,
        "task_status": task_result.status, # PENDING, PROGRESS, SUCCESS, FAILURE 等
        "task_result": None,
        "progress_info": None
    }

    # タスクが進行中の場合、カスタムメタデータ（current_step等）を取得
    if task_result.status == 'PROGRESS':
        response["progress_info"] = task_result.info # tasks.pyでupdate_stateした内容
    
    # タスクが正常終了した場合、戻り値を取得
    elif task_result.status == 'SUCCESS':
        response["task_result"] = task_result.result
        
    # タスクが失敗した場合
    elif task_result.status == 'FAILURE':
        response["error"] = str(task_result.info)
        
    return response

4. エンタープライズ運用における3つの重要事項

AIエージェントを本番環境で運用する場合、単にタスクを動かすだけでなく、以下の「安定性」と「UX」への考慮が不可欠です。

1. 状態のポーリング vs WebSocket/SSE

小規模なシステムであれば、フロントエンドから数秒おきに get_status を叩くポーリング方式で十分です。しかし、2026年現在のモダンなAIアプリケーションでは、WebSocket や Server-Sent Events (SSE) を採用し、ワーカー側の進捗をリアルタイムでプッシュするのが一般的です。これにより、ユーザーはAIが「今何を考えているか」をストレスなく待つことができます。

2. リソース管理（GPU/CPU）

AI推論はメモリ消費が激しいため、ワーカーの並列数（Concurrency）を無制限にすると簡単にメモリ不足（OOM）でシステムがクラッシュします。

  • 並列数の制限: celery worker -c 2 のように、利用可能なビデオメモリ（VRAM）やRAMに合わせてワーカーの同時実行数を厳格に制限します。
  • 専用キューの運用: 軽量なタスク（メール送信等）と重たいAIタスクを同じワーカーで処理せず、AI専用の「GPUキュー」を用意して隔離運用するのが定石です。

3. Graceful Shutdown（正常なシャットダウン）

AIの推論中にワーカーが再起動（デプロイやオートスケーリング等）されると、処理中のタスクが消失するリスクがあります。

  • acks_late の有効化: タスクが完全に完了してから「確認（Ack）」を返す設定にします。
  • SIGTERMの処理: 終了シグナルを受け取った際に、現在のAI処理を可能な限り区切りの良いところまで完結させてから終了する実装が必要です。

5. まとめ：スケーラブルなAI基盤へ

FastAPIとCeleryを組み合わせることで、「リクエストの受付（API層）」と「計算コストの高い実行（Worker層）」を完全に疎結合にできます。

これにより、フロントエンドは即座に受付完了レスポンスを受け取り、バックグラウンドでAIがじっくりと「思考」を続ける、現代的なAIアプリケーションの理想的なアーキテクチャが完成します。将来的なリクエスト増に対しても、ワーカーを増やすだけで容易にスケールアウトが可能です。

🚀 NKKTech Software：高度なAIバックエンドをハノイから提供

私たちは、ベトナム・ハノイを拠点に、日本品質のマネジメントと大規模なAIシステム構築の実績を持つプロフェッショナル集団です。

  • FastAPI / Celeryを用いた大規模非同期処理基盤の構築
  • LangGraph / LangChainを用いた複雑なAIエージェントの実装
  • インフラコストを最適化したGPU/CPUハイブリッド運用の設計

AIエージェントの商用化や、レスポンス遅延・スケーラビリティの問題でお困りでしたら、ぜひ一度ご相談ください。

確かなエンジニアリングで、AIの可能性をビジネス価値へ変えます。

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?