AIアプリ開発で差をつける！Pythonの非同期処理（asyncio・aiohttp）入門【API呼び出しを爆速化する実践ガイド】
「複数のAPIを同時に呼び出したいのに、順番に待ってたら遅すぎる…」
AIアプリを作り始めると、必ずこの壁にぶつかりますよね。OpenAIのAPIを10回呼ぶだけで数十秒かかる、なんて経験はありませんか？
その悩み、Pythonの非同期処理（asyncio・aiohttp）で一気に解決できるかもしれません！
この記事では「非同期処理って難しそう…」という印象を「意外と使えそう！」に変えるべく、基礎の仕組みから実際のAI APIへの応用まで、順を追って解説していきます。
✅ 対象読者：Pythonの基礎は分かる・中級者を目指している方
✅ 難易度：★★★☆☆（中級者向け）
そもそも「非同期処理」って何？
まずはざっくりとしたイメージを掴んでおきましょう。
通常のPythonコード（同期処理）は、料理の注文を1つずつこなす「一人のウェイター」のようなものです。
- テーブルAの注文を取る → キッチンに伝える → 待つ → 料理が来たら運ぶ
- その間、テーブルBのお客さんはひたすら待ちぼうけ
一方、非同期処理は「テーブルAの注文を伝えたら、待ち時間中にテーブルBの注文も取りに行ける」スタイルです。
AIアプリでいうと、「APIのレスポンス待ち時間に他の処理を進める」ことができるわけです。これだけで体感速度が劇的に変わります 🚀
asyncioの基本構文を押さえよう
Pythonで非同期処理を書くときのキーワードは主に3つです。
- async def：非同期関数（コルーチン）の定義
- await：非同期処理の完了を待つ
- asyncio.run()：非同期処理のエントリーポイント
まずはシンプルな例から試してみましょう。
import asyncio
# async def で「非同期関数」を定義する
async def fetch_data(name, delay):
print(f"{name}: データ取得を開始します")
await asyncio.sleep(delay) # I/O待ちをシミュレート（awaitで他の処理に譲る）
print(f"{name}: データ取得が完了しました！")
return f"{name}のデータ"
async def main():
# asyncio.gather() で複数のコルーチンを同時実行
results = await asyncio.gather(
fetch_data("API_A", 2),
fetch_data("API_B", 1),
fetch_data("API_C", 3),
)
print(f"全結果: {results}")
# asyncio.run() でメインの非同期関数を実行
asyncio.run(main())
ポイントをまとめるとこんな感じです👇
await asyncio.sleep()は「I/O待ちの間、他の処理に制御を渡す」動作をシミュレートしていますasyncio.gather()に複数のコルーチンを渡すと、同時並行で実行されます- 合計6秒かかるはずの処理が、最も長い3秒で終わります（同期処理なら6秒！）
aiohttpで実際のHTTPリクエストを非同期化する
次は本番です。aiohttpというライブラリを使うと、実際のHTTPリクエストを非同期で投げられます。
まずインストールしておきましょう。
pip install aiohttp
試しに、複数のURLから同時にデータを取得してみます。
import asyncio
import aiohttp
import time
# 取得したいURLのリスト（ここでは公開JSONAPIを例示）
URLs = [
"https://jsonplaceholder.typicode.com/posts/1",
"https://jsonplaceholder.typicode.com/posts/2",
"https://jsonplaceholder.typicode.com/posts/3",
"https://jsonplaceholder.typicode.com/posts/4",
"https://jsonplaceholder.typicode.com/posts/5",
]
async def fetch(session, url):
"""1件のURLからデータを非同期で取得する"""
async with session.get(url) as response: # async with でセッションを管理
data = await response.json() # JSONレスポンスをawaitで受け取る
return data["title"]
async def main():
start = time.time()
# aiohttp.ClientSession() をセッションとして使い回す（効率的！）
async with aiohttp.ClientSession() as session:
tasks = [fetch(session, url) for url in URLs] # タスクリストを作成
results = await asyncio.gather(*tasks) # まとめて同時実行
elapsed = time.time() - start
for i, title in enumerate(results, 1):
print(f"[{i}] {title}")
print(f"\n⏱ 処理時間: {elapsed:.2f}秒")
asyncio.run(main())
ここが重要です👇
aiohttp.ClientSession()はセッションを使い回すことで接続コストを削減しますasync withは「非同期のwith文」。自動的にリソースを閉じてくれますasyncio.gather(*tasks)の*（アンパック）に注意！リストをそのまま渡すのではなく展開します
実践：OpenAI APIを非同期で複数同時呼び出しする
ここからが本番です 🔥 AIアプリ開発で最も効果を実感できる「複数プロンプトの同時並行処理」を実装してみましょう。
たとえば「商品レビューを10件まとめてAIに分析させたい」というケース。同期処理だと10回順番に待つことになりますが、非同期なら一気に投げられます。
import asyncio
import aiohttp
import os
import time
# 環境変数からAPIキーを取得（ハードコードは絶対NG！）
API_KEY = os.environ.get("OPENAI_API_KEY")
API_URL = "https://api.openai.com/v1/chat/completions"
# 分析したいレビューのリスト
REVIEWS = [
"この商品は最高です！品質も良くて大満足。",
"少し高いですが、それだけの価値はありました。",
"梱包が雑で残念。中身は良かったです。",
"期待外れでした。説明と違います。",
"リピート買いです。ずっと愛用しています！",
]
async def analyze_review(session, review, index):
"""1件のレビューをOpenAI APIで感情分析する"""
headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json",
}
payload = {
"model": "gpt-4o-mini",
"messages": [
{"role": "system", "content": "レビューをポジティブ・ネガティブ・中立の3段階で分類してください。1単語で回答。"},
{"role": "user", "content": review},
],
"max_tokens": 10,
}
async with session.post(API_URL, headers=headers, json=payload) as resp:
data = await resp.json()
sentiment = data["choices"][0]["message"]["content"]
return index, review[:20] + "...", sentiment
async def main():
start = time.time()
async with aiohttp.ClientSession() as session:
# 全レビューのタスクを同時に投げる
tasks = [
analyze_review(session, review, i)
for i, review in enumerate(REVIEWS, 1)
]
results = await asyncio.gather(*tasks)
elapsed = time.time() - start
print("=== 感情分析結果 ===")
for idx, text, sentiment in results:
print(f"[{idx}] {text} → {sentiment}")
print(f"\n⏱ 処理時間: {elapsed:.2f}秒（同期処理なら約{elapsed * len(REVIEWS):.1f}秒相当）")
asyncio.run(main())
5件のAPIコールを同時並行で処理できるので、レート制限に引っかからない程度の件数であれば、体感で3〜5倍の速度改善が期待できます。
エラーハンドリングと並列数の制御
実際の開発では「全部まとめて投げればOK！」とはいきません。APIにはレート制限（Rate Limit）があるので、同時リクエスト数を制御する必要があります。
asyncio.Semaphoreを使うと、同時実行数の上限を設定できます。
import asyncio
import aiohttp
# 同時実行の上限を3に設定（APIのレート制限に合わせて調整）
SEM_LIMIT = 3
async def fetch_with_limit(session, url, semaphore):
"""セマフォで同時実行数を制限しながらリクエストを送る"""
async with semaphore: # セマフォを取得（上限に達したら待機）
try:
async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
if resp.status == 200:
return await resp.json()
else:
# ステータスコードが200以外のエラー処理
print(f"⚠ HTTPエラー: {resp.status} ({url})")
return None
except asyncio.TimeoutError:
print(f"⚠ タイムアウト: {url}")
return None
except aiohttp.ClientError as e:
print(f"⚠ 通信エラー: {e}")
return None
async def main():
urls = [f"https://jsonplaceholder.typicode.com/posts/{i}" for i in range(1, 11)]
semaphore = asyncio.Semaphore(SEM_LIMIT) # セマフォを作成
async with aiohttp.ClientSession() as session:
tasks = [fetch_with_limit(session, url, semaphore) for url in urls]
results = await asyncio.gather(*tasks)
# Noneを除いて有効な結果だけ表示
valid = [r for r in results if r is not None]
print(f"✅ 取得成功: {len(valid)}/{len(urls)}件")
asyncio.run(main())
ポイントをまとめるとこんな感じです👇
asyncio.Semaphore(3)で「同時に3つまで」という制限を設けられますaiohttp.ClientTimeout(total=10)でタイムアウトを設定（必須！設定しないと無限待ちになることも）try/exceptで通信エラーとタイムアウトを個別にハンドリングするのがおすすめです
同期 vs 非同期：速度の差を数字で見てみよう
「実際どのくらい速くなるの？」という疑問に答えるために、同期処理と非同期処理を比較してみます。
import asyncio
import aiohttp
import requests # 同期版HTTPライブラリ
import time
URLs = [f"https://jsonplaceholder.typicode.com/posts/{i}" for i in range(1, 11)]
# === 同期処理（従来のやり方）===
def sync_fetch_all():
start = time.time()
results = []
for url in URLs:
resp = requests.get(url) # 1つ終わるまで次に進めない
results.append(resp.json()["id"])
elapsed = time.time() - start
print(f"[同期] {len(results)}件取得 → {elapsed:.2f}秒")
# === 非同期処理（asyncio + aiohttp）===
async def async_fetch_all():
start = time.time()
async with aiohttp.ClientSession() as session:
tasks = [session.get(url) for url in URLs]
responses = await asyncio.gather(*tasks)
results = []
for resp in responses:
data = await resp.json()
results.append(data["id"])
elapsed = time.time() - start
print(f"[非同期] {len(results)}件取得 → {elapsed:.2f}秒")
print("=== 速度比較 ===")
sync_fetch_all()
asyncio.run(async_fetch_all())
実際に実行すると、10件のリクエストで次のような結果になります👇
| 処理方式 | 10件の処理時間（目安） |
|---|---|
| 同期処理（requests） | 約3〜5秒 |
| 非同期処理（aiohttp） | 約0.5〜1秒 |
件数が増えるほど差が開いていきますよね。AIアプリでバッチ処理をするなら非同期は必須といっても過言ではありません。
よくある落とし穴と注意点 ⚠
非同期処理を使い始めると、つまずきやすいポイントがいくつかあります。確認しておきましょう。
① awaitを忘れるとコルーチンが実行されない
# ❌ NG: awaitがないとコルーチンオブジェクトが返るだけ
result = fetch_data()
# ✅ OK: awaitをつけて実際に実行する
result = await fetch_data()
② requestsライブラリは非同期に対応していない
requestsは同期ライブラリなので、async defの中で使っても非同期の恩恵を受けられません。HTTPリクエストにはaiohttpを使いましょう。
③ asyncio.run()はネストできない
Jupyter Notebookなど、すでにイベントループが動いている環境ではasyncio.run()がエラーになります。その場合はawait main()を直接呼び出すか、nest_asyncioライブラリを使いましょう。
# Jupyter Notebookでasyncioを使う場合
import nest_asyncio
nest_asyncio.apply() # これを実行してからasyncio.run()を使う
まとめ
今回は、AIアプリ開発で差をつけるPythonの非同期処理（asyncio・aiohttp）を解説しました。
- ✅ async/awaitでI/O待ちの間に他の処理を進められる
- ✅ asyncio.gather()で複数のAPIコールを同時並行実行
- ✅ aiohttpで実際のHTTPリクエストを非同期化
- ✅ Semaphoreでレート制限を守りながら並列実行
- ✅ エラーハンドリングとタイムアウト設定は必ず入れる
最初は「awaitってどこに書けばいいの？」と混乱するかもしれませんが、慣れてしまえばシンプルです。
AIアプリで複数のAPIを呼び出す場面は必ずやってきます。そのとき「知ってるかどうか」で開発スピードが変わってきますよね。
ぜひ今回のサンプルコードを実際に動かして、非同期処理の感覚を掴んでみてください 🎉
こちらも読まれています
📚 関連商品・おすすめ書籍
※本記事にはアフィリエイトリンクが含まれます。
PR
