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?

@Maku-Inc(MAKU)
in

Python asyncioでAI APIを爆速化！並列呼び出しによる非同期処理でスループットを最大化

0
Last updated at Posted at 2026-11-01

Python asyncioでAI APIを爆速化！並列呼び出しによる非同期処理でスループットを最大化

近年、AI技術の発展は目覚ましく、OpenAIのChatGPTやGoogleのGeminiといった大規模言語モデル（LLM）のAPIがビジネスに深く組み込まれるようになりました。これらのAPIを活用することで、多様なAI機能をアプリケーションに手軽に統合できますが、その一方でパフォーマンスの課題に直面することも少なくありません。本記事では、Pythonのasyncioを用いた非同期処理によって、AI APIの並列呼び出しを高速化し、アプリケーションのスループットを最大化する方法について、現場の知見を交えながら解説します。

課題・背景

AI APIの利用が一般的になるにつれ、多くのエンジニアが「大量のテキストを処理したい」「複数のユーザーリクエストに同時に応答したい」といった場面で、処理速度のボトルネックに直面します。

一般的なAPI呼び出しは、リクエストを送信してからレスポンスを受け取るまで、その間の処理が「ブロック」されます。これは、ネットワークI/O（Input/Output）と呼ばれるデータの送受信待ちの時間であり、この間、Pythonプログラムは実質的に何もせず待機しています。特にAI APIは、モデルの推論時間やネットワークの遅延により、数秒から数十秒の応答時間がかかることも珍しくありません。

もし10個のAPIを順番に呼び出す場合、単純計算で「1回あたりの応答時間 × 10回」の時間が必要です。例えば、1回あたり5秒かかるAPIを10回呼び出すと、合計で50秒もかかってしまいます。これでは、ユーザー体験の低下やバッチ処理の大幅な遅延を招き、システムのボトルネックとなってしまいます。

同期処理の課題

同期処理では、前の処理が完了するまで次の処理が開始できないため、合計処理時間が長くなります。
このようなI/Oバウンドな処理において、CPUはほとんど利用されずアイドル状態です。この無駄な待機時間を有効活用し、複数のAPIリクエストを「並行」して実行することで、システム全体の応答性とスループットを向上させることが現代のAIアプリケーション開発では不可欠です。

実践・解決策

この課題を解決するのが、Pythonの標準ライブラリであるasyncioを活用した非同期処理です。asyncioは、コルーチンと呼ばれる軽量なタスクをイベントループ上で並行実行することで、I/O待機中に別の処理を進めることを可能にします。

具体的な解決策として、asyncioと非同期HTTPクライアントライブラリ（httpxやaiohttpなど）を組み合わせてAI APIを並列呼び出しする方法を紹介します。今回はhttpxを使用しますが、aiohttpでも同様の記述が可能です。

まずは、同期処理と非同期処理の速度差を体感するために、擬似的なAI API呼び出しをシミュレートするコードを見てみましょう。

AI API並列呼び出しの具体的な実装

それでは、同期処理と非同期処理でAI APIを複数回呼び出す際のパフォーマンスの違いを、具体的なコードで比較してみましょう。ここでは、より実践的にHTTPクライアントライブラリとして同期版ではrequests、非同期版ではhttpxを使用します。擬似的なAI APIとしては、無料で利用できるjsonplaceholder.typicode.comのエンドポイントを利用します。

同期処理でのAPI呼び出し

まずは、通常の同期処理で複数のAPIを呼び出す場合です。

import time
import requests # 同期リクエスト用のライブラリ

# 擬似的なAI APIエンドポイント (例: jsonplaceholderを使用)
FAKE_AI_API_URL = "https://jsonplaceholder.typicode.com/todos/{}"

def call_ai_api_sync_httpx(task_id: int, delay_s: float = 0.05):
    """requests を使って同期でAI APIを呼び出す関数"""
    url = FAKE_AI_API_URL.format(task_id)
    print(f"[{time.time():.2f}] Task {task_id}: Calling AI API at {url}...")
    try:
        # ネットワーク遅延をシミュレート (実際はAPIの応答時間によって変動)
        time.sleep(delay_s) 
        response = requests.get(url, timeout=5) # タイムアウト設定は重要
        response.raise_for_status() # HTTPエラー (4xx, 5xx) を検出
        data = response.json()
        print(f"[{time.time():.2f}] Task {task_id}: AI API responded with ID {data.get('id')}.")
        return f"Response for task {task_id}: {data.get('title', 'No Title')}"
    except requests.exceptions.RequestException as e:
        print(f"[{time.time():.2f}] Task {task_id}: HTTP request error: {e}")
        return f"Error for task {task_id}: {e}"

def main_sync(num_calls: int = 5):
    print("\n--- 同期処理の開始 (requests使用) ---")
    start_time = time.monotonic()
    responses = [call_ai_api_sync_httpx(i + 1) for i in range(num_calls)] # IDは1から始める
    end_time = time.monotonic()
    print(f"--- 同期処理の完了。所要時間: {end_time - start_time:.2f}秒 ---")
    return responses

if __name__ == "__main__":
    num_api_calls = 5
    sync_results = main_sync(num_api_calls)
    print("同期処理の結果:", sync_results)

このコードを実行すると、各API呼び出しが前の呼び出しの完了を待ってから開始されるため、合計実行時間は個々のAPI呼び出し時間の合計に近似します。

非同期処理でのAPI呼び出し

次に、asyncioとhttpxを組み合わせて非同期にAPIを呼び出す場合です。

import asyncio
import time
import httpx # 非同期リクエスト用のライブラリ

# 擬似的なAI APIエンドポイント (例: jsonplaceholderを使用)
FAKE_AI_API_URL = "https://jsonplaceholder.typicode.com/todos/{}"

async def call_ai_api_async_httpx(session: httpx.AsyncClient, task_id: int, delay_s: float = 0.05):
    """httpx を使って非同期でAI APIを呼び出す関数"""
    url = FAKE_AI_API_URL.format(task_id)
    print(f"[{time.time():.2f}] Task {task_id}: Calling AI API at {url}...")
    try:
        # ネットワーク遅延をシミュレート (実際はAPIの応答時間によって変動)
        await asyncio.sleep(delay_s)
        response = await session.get(url, timeout=5) # タイムアウト設定
        response.raise_for_status() # HTTPエラーを検出
        data = response.json()
        print(f"[{time.time():.2f}] Task {task_id}: AI API responded with ID {data.get('id')}.")
        return f"Response for task {task_id}: {data.get('title', 'No Title')}"
    except httpx.RequestError as e:
        print(f"[{time.time():.2f}] Task {task_id}: HTTP request error: {e}")
        return f"Error for task {task_id}: {e}"
    except httpx.HTTPStatusError as e:
        print(f"[{time.time():.2f}] Task {task_id}: HTTP status error: {e.response.status_code}")
        return f"Error for task {task_id}: {e.response.status_code}"

async def main_async(num_calls: int = 5):
    print("\n--- 非同期処理の開始 (httpx使用) ---")
    start_time = time.monotonic()
    async with httpx.AsyncClient() as client: # コネクションプールを効率的に利用
        tasks = [call_ai_api_async_httpx(client, i + 1) for i in range(num_calls)] # IDは1から始める
        responses = await asyncio.gather(*tasks) # 全てのタスクが完了するのを待つ
    end_time = time.monotonic()
    print(f"--- 非同期処理の完了。所要時間: {end_time - start_time:.2f}秒 ---")
    return responses

if __name__ == "__main__":
    num_api_calls = 5
    # 非同期処理を実行
    async_results = asyncio.run(main_async(num_api_calls))
    print("非同期処理の結果:", async_results)

このコードを実行すると、すべてのAPI呼び出しがほぼ同時に開始され、I/O待機中にCPUが他のタスクに切り替わるため、合計実行時間は最も時間がかかるAPI呼び出しの時間に近づきます。

実行結果と速度差の比較

上記2つのコード（if __name__ == "__main__":ブロックに同期・非同期両方のmain関数呼び出しを記述）を実行してみてください。
例えば、num_api_calls = 5、delay_s = 0.5で実行した場合、以下のような出力になるはずです（タイムスタンプは実行環境により変動）。

--- 同期処理の開始 (requests使用) ---
[1678886400.00] Task 1: Calling AI API at https://jsonplaceholder.typicode.com/todos/1...
[1678886400.50] Task 1: AI API responded with ID 1.
[1678886400.50] Task 2: Calling AI API at https://jsonplaceholder.typicode.com/todos/2...
[1678886401.00] Task 2: AI API responded with ID 2.
[1678886401.00] Task 3: Calling AI API at https://jsonplaceholder.typicode.com/todos/3...
[1678886401.50] Task 3: AI API responded with ID 3.
[1678886401.50] Task 4: Calling AI API at https://jsonplaceholder.typicode.com/todos/4...
[1678886402.00] Task 4: AI API responded with ID 4.
[1678886402.00] Task 5: Calling AI API at https://jsonplaceholder.typicode.com/todos/5...
[1678886402.50] Task 5: AI API responded with ID 5.
--- 同期処理の完了。所要時間: 2.51秒 ---
同期処理の結果: ['Response for task 1: delectus aut autem', ..., 'Response for task 5: fugiat est facilis corporis id cupiditate']

--- 非同期処理の開始 (httpx使用) ---
[1678886402.51] Task 1: Calling AI API at https://jsonplaceholder.typicode.com/todos/1...
[1678886402.51] Task 2: Calling AI API at https://jsonplaceholder.typicode.com/todos/2...
[1678886402.51] Task 3: Calling AI API at https://jsonplaceholder.typicode.com/todos/3...
[1678886402.51] Task 4: Calling AI API at https://jsonplaceholder.typicode.com/todos/4...
[1678886402.51] Task 5: Calling AI API at https://jsonplaceholder.typicode.com/todos/5...
[1678886403.01] Task 1: AI API responded with ID 1.
[1678886403.01] Task 2: AI API responded with ID 2.
[1678886403.01] Task 3: AI API responded with ID 3.
[1678886403.01] Task 4: AI API responded with ID 4.
[1678886403.01] Task 5: AI API responded with ID 5.
--- 非同期処理の完了。所要時間: 0.51秒 ---
非同期処理の結果: ['Response for task 1: delectus aut autem', ..., 'Response for task 5: fugiat est facilis corporis id cupiditate']

見ての通り、同期処理が約2.5秒かかっているのに対し、非同期処理は約0.5秒で完了しています。これは、AI API呼び出しがI/Oバウンドな処理であり、非同期処理がその待機時間を有効活用できたためです。

注意点・ハマりどころ

非同期処理による並列化は強力ですが、闇雲に適用するとかえって問題を引き起こすことがあります。特にAI API連携では、以下のような点に注意が必要です。

1. AI API側の制約

  • レートリミット (Rate Limit): ほとんどのAI APIには、一定時間内に呼び出せる回数に制限（例: 1分あたり60回）が設けられています。非同期処理で大量にリクエストを送信すると、容易にこの制限に達し、APIからエラー応答（HTTP 429 Too Many Requests）が返されます。
    • 対策:
      • リトライ機構の実装: 指数バックオフ（Exponential Backoff）アルゴリズムを使って、エラー発生時に徐々に待機時間を延ばしながらリトライする。
      • トークンバケットやリーキーバケットなどのレートリミット制御アルゴリズムを自システムに導入する。
      • APIプロバイダに上限緩和の申請をする。
  • コスト: AI APIの多くは従量課金制です。並列処理で大量のリクエストを送信すると、意図しない高額な費用が発生する可能性があります。
    • 対策:
      • テスト環境と本番環境で異なるAPIキーを使用し、テストでの大量実行が本番の予算に影響しないようにする。
      • API利用状況を監視し、異常な呼び出し数を検知するアラートを設定する。
      • 予算アラートを設定する。
  • レスポンスタイムのばらつき: AIモデルの負荷状況や処理内容によって、APIの応答時間は大きく変動します。タイムアウト設定は必須ですが、短すぎると正当なリクエストも失敗し、長すぎるとリソースを無駄に占有します。

2. システムリソースの限界

  • CPU / メモリ: 並列処理は多くのコルーチンを同時に管理するため、CPUやメモリの使用量が増加します。特にPythonのGIL (Global Interpreter Lock) の特性上、CPUバウンドな処理（例: 大規模なデータの前処理・後処理）を純粋なasyncioだけで並列化しようとすると、期待通りの性能向上は見込めません。I/Oバウンドな処理に限定して活用しましょう。
  • ネットワーク帯域: 大量のAPI呼び出しは、ネットワーク帯域を消費します。システムが稼働しているサーバーのネットワークI/O性能がボトルネックになる可能性があります。
  • ファイルディスクリプタ: HTTPコネクションはファイルディスクリプタを消費します。多数のコネクションを同時に開こうとすると、OSが設定しているファイルディスクリプタの上限に達し、Too many open filesエラーが発生することがあります。
    • 対策: httpx.AsyncClientのようなクライアントセッションを使い回し、コネクションプールを効率的に利用する。OSのファイルディスクリプタ上限を適切に調整する。

3. 非同期処理の複雑性

  • エラーハンドリング: asyncio.gatherは、デフォルトではいずれかのタスクが例外を発生させると、他のタスクの完了を待たずに即座に例外を伝播させます。全てのタスクの実行結果を得たい場合は、return_exceptions=Trueを指定する必要があります。 
    responses = await asyncio.gather(*tasks, return_exceptions=True)
for res in responses:
    if isinstance(res, Exception):
        print(f"Task failed: {res}")
    else:
        print(f"Task succeeded: {res}")
  • タイムアウト: 各API呼び出しにタイムアウトを設定するのはもちろん、asyncio.wait_forを使って全体の処理にもタイムアウトを設定することで、特定のタスクがハングアップした場合に全体のプロセスが停止するのを防げます。
  • キャンセル処理: asyncio.create_task()で生成したタスクは、task.cancel()でキャンセルできます。しかし、既にI/O処理に入っているタスクをキャンセルしても、APIサーバー側では処理が進行している可能性があるため注意が必要です。
  • デバッグの困難さ: 非同期処理は同期処理に比べてコードの流れが複雑になりがちで、デバッグが難しくなることがあります。ログを適切に出力し、実行の流れを追えるようにすることが重要です。

4. スケーラビリティの考慮

  • コネクションプーリング: httpx.AsyncClientやaiohttp.ClientSessionを使い回すことで、TCPコネクションの確立・切断コストを削減し、API呼び出しの効率を高められます。使い捨てのクライアントは避けるべきです。
  • メッセージキューの導入: 処理するAPI呼び出しが非常に多い場合や、システムが一時的に過負荷になった場合に、リクエストをキューに入れて非同期に処理することで、システムの安定性を高めることができます（例: RabbitMQ, Kafka, AWS SQS）。これにより、AI APIのレートリミットに合わせた流量制御も容易になります。
  • バックオフとリトライ: API側のエラー（レートリミット、サーバーエラーなど）に対しては、単にリトライするだけでなく、バックオフ戦略（指数バックオフなど）を取り入れることで、APIサーバーへの負荷をさらに増やすことを避け、復旧を促すことができます。

5. セキュリティ（インフラ/NW視点も含む）

  • APIキーの安全な管理: AI APIキーは機密情報であり、コードに直書きしたり、Gitリポジトリに含めたりしてはいけません。
    • 対策: 環境変数、Docker Secrets、Kubernetes Secrets、AWS Secrets Manager、Azure Key Vault、HashiCorp Vaultなどのシークレット管理サービスを利用して安全に管理し、最小限の権限でアクセスするようにしましょう。
  • 通信の暗号化: ほとんどのAI APIはHTTPSで提供されていますが、念のため通信経路が暗号化されていることを確認し、中間者攻撃のリスクを低減しましょう。
  • ネットワーク分離: 必要に応じて、AI APIを呼び出すシステムをプライベートなサブネットに配置し、外部からの不正アクセスを防ぐネットワーク構成を検討します。

6. 運用と監視（ソフトスキル視点も含む）

  • 監視体制: API呼び出しの成功/失敗回数、レスポンスタイム、API側のレートリミット到達回数などを常時監視し、異常を検知できるアラートを設定することが重要です。
    • 対策: Prometheus, Grafana, Datadogなどの監視ツールと連携し、ダッシュボードで可視化する。
  • ログの活用: 適切な粒度でログを出力し、問題発生時の原因究明に役立てましょう。特にリクエストIDやタスクIDを含めることで、分散したログの中から特定の処理を追跡しやすくなります。
  • チームへの共有: 非同期処理を導入する際は、その設計思想、メリット、注意点をチーム内でしっかり共有し、共通認識を持つことが重要です。特に実務初学者が躓きやすいポイントでもあるため、設計レビューやペアプログラミングを通じて知識を移転していく工夫が求められます。

7. 独学者が陥りがちな落とし穴

  • 闇雲な並列化: 全てのI/O処理を並列化すれば速くなるというわけではありません。CPUバウンドな処理を無理に非同期化したり、そもそもボトルネックではない部分を並列化しても効果は薄く、かえってコードの複雑性だけが増します。
  • ボトルネックの見誤り: パフォーマンスの問題を解決するためには、まずどこがボトルネックになっているのかをプロファイリングツールなどを使って正確に特定することが重要です。I/O待機なのか、CPU処理なのか、DBアクセスなのか、ネットワーク遅延なのかを見極めましょう。
  • asyncio.run()の誤用: asyncio.run()はイベントループを起動・停止する関数であり、アプリケーションのメインエントリーポイントで一度だけ呼び出すのが基本です。既存の同期コードから安易にasyncio.run()を何度も呼び出すと、イベントループのオーバーヘッドや予期せぬ挙動につながることがあります。既存の同期システムに非同期処理を導入する場合は、asyncio.to_thread()やThreadPoolExecutorの活用も視野に入れると良いでしょう。

まとめ

本記事では、AI APIのようなI/Oバウンドな外部サービス連携において、Pythonのasyncioと非同期HTTPクライアントライブラリhttpx（またはaiohttp）を活用することで、処理速度を大幅に向上させる方法を解説しました。同期処理と非同期処理のコード例を通じて、そのパフォーマンスの差を体感いただけたかと思います。

非同期処理は、現代の高性能なアプリケーション開発において欠かせない技術です。特にAIサービスを組み込むシステムでは、複数のAIモデルを呼び出したり、大量のデータを一度に処理したりする場面が多いため、このような並列化技術が極めて重要になります。

しかし、その強力さゆえに、レートリミット、リソース枯渇、複雑なエラーハンドリング、そしてセキュリティといった多くの注意点が存在します。これらの「ハマりどころ」を理解し、適切な対策を講じることが、安定してスケーラブルなAIシステムを構築する鍵となります。

独学で学習を進める皆さんには、まず今回の基本的な実装パターンを理解し、実際にコードを動かして「なぜ速くなるのか」を体感してみてください。そして、今回提示した「注意点・ハマりどころ」を意識しながら、ご自身のプロジェクトに非同期処理を導入する際の設計段階で、AI API側の制約、システムリソース、セキュリティ、運用といった多角的な視点から検討を加える習慣をつけることを強くお勧めします。

この知識と経験は、皆さんが現場でAIエンジニアとして活躍していく上で、必ず役立つ実践的なスキルとなるでしょう。

エンジニアのスキルシェアプラットフォーム「DokuPro」

教えたい人と学びたい人を繋ぐDokuProでは、新規登録（先生・生徒）を募集中です。
詳細はこちら: https://dokupro.dev/

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?