非同期処理の導入による生成AIプロダクトの処理速度の改善

非同期処理の導入による生成AIプロダク トの処理速度の改善 IT本部AI・データ戦略統括部データ基盤部 AIジャーニー推進グループ 黄 英智 © DeNA Co., Ltd. 1

自己紹介: 黄 英智/HarrisonEagle ● ● ● © DeNA Co., Ltd. 2023年新卒入社 ○ 入社以来、MLOps/LLM関係の仕事に多く参画 ○ 担当した仕事 ■ 内製機械学習基盤Hekatoncheirの開発/運用 ■ Pococha関連AIの運用 ■ 全社横断LLM基盤「SAI」の開発/運用 ■ 社内LLM案件の推進 趣味 ○ 筋トレ ○ ゴルフ ○ 格闘技 ○ 旅行 X: @harris0n3ag1e 2

今日話すこと ● ● © DeNA Co., Ltd. 生成AIプロダクト開発の難しさ ○ 処理精度と処理速度の両立 生成AIプロダクトのスケール性を改善する為のアプローチの紹介 ○ 非同期処理を利用した改善 ○ 非同期処理の信頼性を担保する仕組み 3

生成AIプロダクト開発の難しさ © DeNA Co., Ltd. 4

生成プロダクト開発の難しさ: 精度への要求 ● © DeNA Co., Ltd. 社内業務の効率化ソリューション、あるいはtoB向けのサービスを提供するに あたって、LLM処理に高い精度が求められる ○ QA作業など精度の高さが求められる作業については、AIでも少なくとも 90%以上の精度が求められる ■ 低い精度だと、結局必要となる人手の介入が多くなり、業務効率化 効果が薄くなる ○ 処理精度を高めるアプローチの多くは、時間がかかってしまう ■ リーズニングモデルの採用 ■ データの逐次処理 ■ 複雑なAIエージェントの構築 5

生成プロダクト開発の難しさ: 処理速度への要求 ● 処理精度を高めたいといっても、時間をかけすぎると、ユーザー体験が著し く落ちてしまう ○ 時間かけすぎると、業務効率化にとっては逆効果になる ○ 処理完了を1~2時間以上待つぐらいだったら、結局人手でやったほうが早 い 精度だけなく、処理を効率的に行い処理時間を減らすことも ユーザー体験を改善するにあたって非常に重要になる © DeNA Co., Ltd. 6

非同期処理によるプロダクト改善アプローチの 紹介 © DeNA Co., Ltd. 7

生成AIプロダクトにおける非同期処理の導入 ● 処理精度を犠牲にしない前提で、プロダクトの処理時間を削減するには、重 い処理を効率的に捌く仕組みが必要 ○ 逐次でデータごとに処理していた部分を、並列で行うようにする ○ ワークフローのうち、並列で実行できるパートがあったら、並列で実行 して、他のパートの処理を待つ時間を減らす 処理の非同期化&並列化によって、重いLLM処理を効率化する © DeNA Co., Ltd. 8

非同期処理の導入に考慮すべき事項 ● ● ● ● ● © DeNA Co., Ltd. 処理内容 ○ LLM API呼び出しだけで終わるのか？それともRAG検索など他の計算もあるのか? 出力結果の順番 ○ 要件によっては、データ出力の順番を保ってほしい場合がある 並列処理のWorker数の設定 ○ 並列で処理させる際、Worker数が全体の処理をどれぐらい効率化できるかを左右 する ○ 入れるデータの量、そしてリソース設定に応じて設定する ○ 必要に応じて処理サーバーのオートスケーリングを行う LLM API Callingのリトライ処理は堅固になっているのか? ○ OpenAI APIのRate Limitに引っかかった場合などの対応 ○ なんらかの問題で失敗すると処理結果のロスにつながる 結果を再取得できる仕組みは整っているのか？ 9

PythonのThreadPoolExecutorを使用する例 ● ● © DeNA Co., Ltd. ThreadPoolExecutorは、ス レッドプールを利用して、並 列処理を行うためのPythonの 標準クラス ○ 重い処理を並列化するに あたって一番簡単なアプ ローチ ○ LLM呼び出しなどの重い 処理を別スレッドで行う max_workersに指定された数 のスレッドが立ち上がる import concurrent.futures import time def worker(number): print(f"Task {number}: Starting") ... print(f"Task {number}: Finished") return result # max_workersで最大スレッド数を指定 with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor: # タスクを投入し、 Futureオブジェクトを受け取る future1 = executor.submit(worker, 1) future2 = executor.submit(worker, 2) future3 = executor.submit(worker, 3) print(f"Result 1: {future1.result()}") print(f"Result 2: {future2.result()}") print(f"Result 3: {future3.result()}") 10

FastAPI BackgroundTasksとは
import time
from fastapi import FastAPI, BackgroundTasks

●

●

© DeNA Co., Ltd.

FastAPIで、クライアント側
にレスポンスを返した後に
も、バックグラウンドで何
らかの処理を続行させるた
めの機能
大量なデータを捌くLLM処理
など、すぐにレスポンス返
せない重い処理を非同期化
するのに最適

app = FastAPI()
def send_welcome_email(email: str):
# メール送信をシミュレートする重いタスク
print(f"'{email}' 宛にメール送信を開始します ...")
time.sleep(5) # 5秒待機してメール送信処理を擬似的に表現
print(f"'{email}' 宛のメール送信が完了しました。 ")
@app.post("/register")
async def user_register(email: str, tasks:
BackgroundTasks):
# 重いメール送信処理をバックグラウンドタスクに登録
tasks.add_task(send_welcome_email, email)
# ユーザーにはすぐにレスポンスを返す
return {"message": f"ユーザー '{email}' の登録を受け付け
ました。"}

11

ThreadPoolExecutor + Background Tasksのアプローチ ● ● 重い処理は、ThreadPoolExecutorで LLM関連処理を並列化 処理した結果はすぐに返せないの で、Background TasksでLLM推論を 非同期化する ここでThreadPoolExecutor ベースのLLM並列処理を実行 © DeNA Co., Ltd. 12

ThreadPoolExecutor + Background Tasksのアプローチ ● ● © DeNA Co., Ltd. 処理結果は、後ほどクライアントが 取得できるようにRedisやFirestore などに保存する クライアントからは、処理開始リク エスト後、ポーリングで処理の状態 を監視し、処理完了後に結果をクラ イアント側に表示する 13

ThreadPoolExecutor + Background Tasksを実プロダクトに採用した結果 ● 精度要求が高いQA作業の効率化プロダクトにおいて、LLM処理を含 め重い処理をこの手法で並列化&非同期化し、LLM処理精度と処理速 度の両立を実現 ○ 処理に応じてWorker数の調整はしているが、max_workers=8の 場合で、大量データを捌くLLM処理では処理時間が60%以上削減 ○ さらに、Background Tasksで重い処理を非同期化すると同時に、 処理結果が再取得可能な設計にしたところ、重い処理が安定し、 ユーザーも処理状況がわかりやすくなり、ユーザー体験の改善が みられた © DeNA Co., Ltd. 14

ThreadPoolExecutor + Background Tasksの問題点 ● ● 並列処理についてはさらなる最適化余地がある ○ リソース設定に応じたWorker数の最適化 ○ 処理の性質に応じた最適化 Background Taskおよびその中で実行する並列タスクの数が非常に多い場合、 CPUへの負担がかかり、処理が不安定になる可能性がある ○ 処理の規模に応じた妥当なリソース管理、およびオートスケーリングの 設定などへの考慮が必要 処理の信頼性を保つために、Celeryなどの技術を用いて さらなる堅固な非同期処理基盤の構築が必要 © DeNA Co., Ltd. 15

Celeryとは ● ● ● © DeNA Co., Ltd. Pythonで書かれた非同期タスクキューライブラリ 中〜大規模なアプリケーションで、時間のかかる処理や定期的なタスク を効率的に実行する基盤の構築に使用される RabbitMQやRedisなど、さまざまなメッセージブローカーや結果バック エンドに対応しており、プロジェクトの要件に合わせて柔軟に構成変更 が可能 16

Celeryの基本的な使い方 非同期タスクの定義例 @celery_app.task(name="process_item", queue="llm") def process_item(batch_id: str, item: dict[str, Any]) -> dict[str, Any]: input_text = str(item.get("text", "")) result = run_llm(input_text) push_result(batch_id, result) incr_completed(batch_id) return result © DeNA Co., Ltd. 非同期タスクの実行リクエスト from myproject.tasks import process_item for item in req.items: process_item.delay(batch_id, item) 17

KEDA(Kubernetes Event-driven Autoscaling)とは ● ● © DeNA Co., Ltd. Kubernetes上でイベント駆動のスケーリングを可能にするためのオープン ソースコンポーネント CPUやメモリ使用率だけでなく、メッセージキュー、データベース、スト リーミングプラットフォーム、HTTPリクエストなど、様々な外部システムの メトリクスをトリガーとしてスケーリングできる。 ○ これらのイベントソースは「Scaler」と呼ばれている 18

Celery+KEDAによるオートスケーリング ● ● © DeNA Co., Ltd. Celery自身も、複数Workerに よる運用と、Workerのオート スケーリングをサポートして いる Kubernetesで運用する際、 KEDAでCelery非同期処理の待 機数に応じて、オートスケー リングをトリガーできる apiVersion: keda.sh/v1alpha1 kind: ScaledObject metadata: name: worker namespace: default spec: scaleTargetRef: kind: Deployment name: worker pollingInterval: 15 cooldownPeriod: 200 maxReplicaCount: 4 minReplicaCount: 1 triggers: - type: redis metadata: address: REDIS_HOST # Required host:port password: REDIS_PASSWORDired enableTLS: "false" # optional databaseIndex: "0" listName: celery listLength: "30" 19

CeleryベースのLLM非同期処理基盤 LLM非同期処理基盤による性能改善は検証済み、今後実導入予定 © DeNA Co., Ltd. 20

他に考えられるアプローチ ● ● © DeNA Co., Ltd. LLM処理を非同期化するには他にも色んな方法がある ○ 今回はThreadPoolExecutorを紹介したが、LLM API呼び出しのみの場 合、asyncio + AsyncAzureOpenAIの手もある ○ 非同期処理基盤の構築については、PubSub+Cloud Functions, あるいは CloudTasksを利用する手もある 運用する要件に応じて、適切な技術選定を行えば良い 21

まとめ © DeNA Co., Ltd. 22

まとめ ● ● © DeNA Co., Ltd. 生成AIプロダクトを成功させるには、処理の精度だけでなく、処理速度との 両立が必要 精度を保つ前提で処理速度を改善するには、LLM処理の並列化&非同期化が手 法として有効 ○ ThreadPoolExecutor + Background Tasks、およびCeleryなど、色んな手 法がある ○ 非同期処理を導入するにあたって、処理のリトライや安定性などの考慮 も必要 23

© DeNA Co., Ltd. 24 24