Python非同期AI API呼び出し：asyncioとaiohttp実践ガイド

AI APIをPythonから呼び出す際、一件ずつ順番に処理すると時間がかかってしまうことがあります。非同期処理を使えば、同時に複数のリクエストを送信でき、大幅に時間を短縮できます。この記事では、HolySheep AIを例に、Pythonのasyncioとaiohttpを使った非同期API呼び出しの基本をゼロから解説します。

前提知識と環境準備

まず、この記事で必要になる知識を説明します。既にPythonがインストール済みの方は読み飛ばして構いません。

必要なもの

Python 3.7以上がインストールされていること
APIキー（HolySheep AIに登録すると無料で取得できます）
コマンドライン（ターミナル）の基本操作

必要なライブラリのインストール

以下のコマンドを実行して、必要なライブラリをインストールします。

pip install aiohttp python-dotenv

インストール確認のテストコード

# インストール確認用コード
import aiohttp
import asyncio

print(f"aiohttpバージョン: {aiohttp.__version__}")
print("インストール成功！")

非同期処理とは？同步処理との違い

традиционная синхронная обработка（同步処理）では、一つのリクエストが完了するまで次のリクエストを待つ必要があります。 например:

リクエスト1 → [1秒待つ] → 完了
リクエスト2 → [1秒待つ] → 完了
リクエスト3 → [1秒待つ] → 完了
─────────────────────────
合計: 3秒

非同期処理では、同時に複数のリクエストを送信できます：

リクエスト1 → [1秒待つ] → 完了
リクエスト2 ──→ [1秒待つ] → 完了
リクエスト3 ──→ [1秒待つ] → 完了
─────────────────────────
合計: 約1秒

HolySheep AIは<50msの低レイテンシを提供しているので、非同期処理の効果が особенно大きくなります。

シンプルな非同期API呼び出し

まずは一番シンプルな非同期API呼び出しのコードを見てみましょう。

import asyncio
import aiohttp
import os
from dotenv import load_dotenv

環境変数の読み込み
load_dotenv()

APIキーの設定（.envファイルに HOLYSHEEP_API_KEY=あなたのキー を記述）
API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
BASE_URL = "https://api.holysheep.ai/v1"

async def call_ai(prompt: str) -> dict:
    """HolySheep AIに単一の非同期リクエストを送信"""
    
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    }
    
    payload = {
        "model": "gpt-4.1",
        "messages": [
            {"role": "user", "content": prompt}
        ],
        "max_tokens": 500
    }
    
    async with aiohttp.ClientSession() as session:
        async with session.post(
            f"{BASE_URL}/chat/completions",
            headers=headers,
            json=payload
        ) as response:
            result = await response.json()
            return result

async def main():
    """メインの非同期処理"""
    print("AIに質問を送信中...")
    
    response = await call_ai("Pythonのasync/awaitについて1文で説明して")
    
    # 応答の確認
    if "choices" in response:
        answer = response["choices"][0]["message"]["content"]
        print(f"AIの回答: {answer}")
    else:
        print(f"エラー: {response}")

実行
asyncio.run(main())

コードのポイント解説

async def：非同期関数を定義するキーワードです
await：非同期処理が完了するのを待つ命令です
async with：リソースを効率的に管理するための非同期版です
asyncio.run()：非同期プログラムを実行開始します

複数のAIリクエストを同時に送信する

ここが非同期処理の本領発揮場所です。複数のプロンプトを同時に送信してみましょう。

import asyncio
import aiohttp
import os
import time
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
BASE_URL = "https://api.holysheep.ai/v1"

async def call_ai(session: aiohttp.ClientSession, prompt: str, model: str = "gpt-4.1") -> dict:
    """HolySheep AIに非同期リクエストを送信"""
    
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    }
    
    payload = {
        "model": model,
        "messages": [{"role": "user", "content": prompt}],
        "max_tokens": 200
    }
    
    async with session.post(
        f"{BASE_URL}/chat/completions",
        headers=headers,
        json=payload
    ) as response:
        result = await response.json()
        return {"prompt": prompt[:30] + "...", "response": result}

async def main():
    """複数のリクエストを同時に送信"""
    
    # 送信する質問リスト
    prompts = [
        "Pythonのリストとは何か簡潔に説明して",
        "非同期処理の利点を3つ教えて",
        "aiohttpの基本的な使い方を教えて",
        "API呼び出しのエラーハンドリングの重要性を教えて",
        "HolySheep AIの特徴を1つ教えて"
    ]
    
    print(f"📤 {len(prompts)}件のリクエストを同時に送信...\n")
    
    start_time = time.time()
    
    # 単一のセッションで複数のリクエストを同時に送信
    async with aiohttp.ClientSession() as session:
        # asyncio.gather()で複数のタスクを同時に実行
        tasks = [call_ai(session, prompt) for prompt in prompts]
        results = await asyncio.gather(*tasks)
    
    elapsed = time.time() - start_time
    
    # 結果の表示
    print(f"⏱️ 完了！実行時間: {elapsed:.2f}秒\n")
    print("=" * 50)
    
    for result in results:
        if "choices" in result["response"]:
            answer = result["response"]["choices"][0]["message"]["content"]
            print(f"📝 {result['prompt']}")
            print(f"   → {answer}\n")
        else:
            print(f"❌ {result['prompt']}: エラー - {result['response']}\n")

asyncio.run(main())

実行結果の例

📤 5件のリクエストを同時に送信...

⏱️ 完了！実行時間: 1.23秒

==================================================
📝 Pythonのリストとは何か簡潔に説明し...
   → Pythonのリストは、複数の要素を一つの変数で管理できるデータ構造です。

📝 非同期処理の利点を3つ教えて...
   → 1. 同時処理による高速化、2. ネットワーク待ち時間の有効活用、3. リソース効率の向上

📝 aiohttpの基本的な使い方を教えて...
   → aiohttpはPythonの非同期HTTPクライアントライブラリで、async/await構文を使って効率的なHTTPリクエストを行えます。

📝 API呼び出しのエラーハンドリングの重要性を教えて...
   → エラーハンドリングにより、APIの障害や一時的な問題があってもプログラムが中断することを防げます。

📝 HolySheep AIの特徴を1つ教えて...
   → HolySheep AIは<50msの低レイテンシを実現し、リアルタイムアプリケーションに最適なAPIです。

※HolySheep AIは<50msのレイテンシを実現しているため、5件のリクエストでも約1秒で完了します。

異なるモデルを同時に呼び出す

非同期処理を使えば、異なるAIモデルを同時に試すこともできます。

import asyncio
import aiohttp
import os
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
BASE_URL = "https://api.holysheep.ai/v1"

async def compare_models(prompt: str) -> dict:
    """複数のモデルで同じ質問を実行し、結果を比較"""
    
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    }
    
    # 比較するモデルのリスト（HolySheep AIの価格表より）
    models = {
        "gpt-4.1": {"price": 8.00, "currency": "USD"},
        "gemini-2.5-flash": {"price": 2.50, "currency": "USD"},
        "deepseek-v3.2": {"price": 0.42, "currency": "USD"}
    }
    
    async with aiohttp.ClientSession() as session:
        tasks = []
        
        for model_name in models.keys():
            payload = {
                "model": model_name,
                "messages": [{"role": "user", "content": prompt}],
                "max_tokens": 100
            }
            
            task = session.post(
                f"{BASE_URL}/chat/completions",
                headers=headers,
                json=payload
            )
            tasks.append((model_name, models[model_name], task))
        
        # すべてのリクエストを同時に送信
        responses = await asyncio.gather(*[t[2] for t in tasks])
        
        results = {}
        for i, (model_name, info, _) in enumerate(tasks):
            resp_json = await responses[i].json()
            results[model_name] = {
                "price_per_mtok": info["price"],
                "response": resp_json
            }
        
        return results

async def main():
    """メイン処理"""
    prompt = "AIについて一句话で教えてください"
    
    print(f"📊 モデル比較: 「{prompt}」\n")
    print("=" * 60)
    
    results = await compare_models(prompt)
    
    for model, data in results.items():
        if "choices" in data["response"]:
            answer = data["response"]["choices"][0]["message"]["content"]
            print(f"\n🤖 モデル: {model}")
            print(f"   価格: ${data['price_per_mtok']}/MTok")
            print(f"   回答: {answer}")
        else:
            print(f"\n🤖 モデル: {model}")
            print(f"   エラー: {data['response']}")

asyncio.run(main())

HolySheep AIでは、DeepSeek V3.2が$0.42/MTokと非常に安価で、GPT-4.1の$8/MTokと比較して95%以上のコスト削減が可能です。

エラーハンドリングの実装

実際の運用では、ネットワークエラーやAPI制限など、さまざまな問題が発生します。堅牢なエラーハンドリングを実装しましょう。

import asyncio
import aiohttp
import os
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
BASE_URL = "https://api.holysheep.ai/v1"

class AIAPIError(Exception):
    """AI APIエラー用のカスタム例外"""
    def __init__(self, message, status_code=None):
        super().__init__(message)
        self.status_code = status_code

async def call_ai_with_retry(
    prompt: str,
    max_retries: int = 3,
    retry_delay: float = 1.0
) -> dict:
    """リトライ機能付きのAI API呼び出し"""
    
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    }
    
    payload = {
        "model": "gpt-4.1",
        "messages": [{"role": "user", "content": prompt}],
        "max_tokens": 500
    }
    
    last_error = None
    
    for attempt in range(max_retries):
        try:
            async with aiohttp.ClientSession() as session:
                async with session.post(
                    f"{BASE_URL}/chat/completions",
                    headers=headers,
                    json=payload,
                    timeout=aiohttp.ClientTimeout(total=30)
                ) as response:
                    
                    # HTTPステータスの確認
                    if response.status == 429:
                        raise AIAPIError("レート制限に達しました（429）", 429)
                    elif response.status == 401:
                        raise AIAPIError("APIキーが無効です（401）", 401)
                    elif response.status >= 500:
                        raise AIAPIError(f"サーバーエラー（{response.status}）", response.status)
                    
                    result = await response.json()
                    
                    # API側のエラーレスポンスを確認
                    if "error" in result:
                        raise AIAPIError(f"APIエラー: {result['error']}")
                    
                    return result
                    
        except aiohttp.ClientError as e:
            last_error = f"ネットワークエラー: {str(e)}"
            print(f"⚠️ 試行 {attempt + 1}/{max_retries}: {last_error}")
            
        except AIAPIError as e:
            # リトライ不值得エラー（認証エラーなど）
            if e.status_code in [401, 403]:
                raise
            last_error = str(e)
            print(f"⚠️ 試行 {attempt + 1}/{max_retries}: {last_error}")
        
        # リトライ前的待機
        if attempt < max_retries - 1:
            await asyncio.sleep(retry_delay * (attempt + 1))
    
    raise AIAPIError(f"最大リトライ回数を超過しました: {last_error}")

async def main():
    """エラーハンドリングのテスト"""
    
    # 正常系のテスト
    print("🧪 正常系のテスト...")
    try:
        result = await call_ai_with_retry("こんにちは！")
        if "choices" in result:
            print("✅ 成功！")
            print(f"   回答: {result['choices'][0]['message']['content']}")
    except AIAPIError as e:
        print(f"❌ 失敗: {e}")

asyncio.run(main())

応用：バッチ処理の実装

最後に、大量のプロンプトを効率的に処理するバッチ処理の例を示します。

import asyncio
import aiohttp
import os
import time
from typing import List
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
BASE_URL = "https://api.holysheep.ai/v1"

class BatchProcessor:
    """大量プロンプトのバッチ処理クラス"""
    
    def __init__(self, api_key: str, max_concurrent: int = 10):
        self.api_key = api_key
        self.max_concurrent = max_concurrent  # 同時実行数の上限
        self.results = []
        self.errors = []
    
    async def process_single(
        self,
        session: aiohttp.ClientSession,
        prompt: str,
        index: int
    ) -> dict:
        """单个プロンプトの処理"""
        
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "model": "deepseek-v3.2",  # 安価なモデルを使用
            "messages": [{"role": "user", "content": prompt}],
            "max_tokens": 100
        }
        
        try:
            async with session.post(
                f"{BASE_URL}/chat/completions",
                headers=headers,
                json=payload,
                timeout=aiohttp.ClientTimeout(total=60)
            ) as response:
                result = await response.json()
                
                if "choices" in result:
                    return {
                        "index": index,
                        "success": True,
                        "answer": result["choices"][0]["message"]["content"]
                    }
                else:
                    return {
                        "index": index,
                        "success": False,
                        "error": result.get("error", "不明なエラー")
                    }
                    
        except Exception as e:
            return {
                "index": index,
                "success": False,
                "error": str(e)
            }
    
    async def process_batch(self, prompts: List[str]) -> dict:
        """バッチ処理の実行"""
        
        print(f"📦 {len(prompts)}件のプロンプトを処理開始")
        print(f"   同時実行数上限: {self.max_concurrent}")
        
        start_time = time.time()
        
        # セマフォ用于控制同時実行数
        semaphore = asyncio.Semaphore(self.max_concurrent)
        
        async def bounded_process(session, prompt, index):
            async with semaphore:
                return await self.process_single(session, prompt, index)
        
        async with aiohttp.ClientSession() as session:
            tasks = [
                bounded_process(session, prompt, i)
                for i, prompt in enumerate(prompts)
            ]
            
            results = await asyncio.gather(*tasks)
        
        elapsed = time.time() - start_time
        
        # 結果の集計
        success_count = sum(1 for r in results if r["success"])
        error_count = len(results) - success_count
        
        return {
            "total": len(prompts),
            "success": success_count,
            "errors": error_count,
            "elapsed_time": elapsed,
            "results": results
        }

async def main():
    """バッチ処理のデモ"""
    
    # テスト用プロンプト
    test_prompts = [
        f"数字{i}について教えてください" for i in range(1, 21)
    ]
    
    processor = BatchProcessor(
        api_key=API_KEY,
        max_concurrent=5  # 5件同時実行
    )
    
    result = await processor.process_batch(test_prompts)
    
    print("\n" + "=" * 50)
    print("📊 バッチ処理結果:")
    print(f"   合計: {result['total']}件")
    print(f"   成功: {result['success']}件")
    print(f"   失敗: {result['errors']}件")
    print(f"   実行時間: {result['elapsed_time']:.2f}秒")
    print(f"   平均処理時間: {result['elapsed_time']/result['total']:.2f}秒/件")

asyncio.run(main())

よくあるエラーと対処法

エラー1：AttributeError: module 'asyncio' has no attribute 'run'

原因：Python 3.6以前ではasyncio.run()が利用できません。

# Python 3.6以前の場合の代替コード
import asyncio

代わりにこの方法を使用
def run_async(coro):
    loop = asyncio.get_event_loop()
    return loop.run_until_complete(coro)

使用例
run_async(main())

解決：Python 3.7以降にアップグレードするか、上記の代替コードを使用してください。

エラー2：aiohttp.client_exceptions.ClientConnectorError: Cannot connect to host

原因：APIキーが無効、またはネットワーク接続の問題です。

# 接続確認用のテストコード
import asyncio
import aiohttp

async def test_connection():
    try:
        async with aiohttp.ClientSession() as session:
            # APIのベースURLにGETリクエストを送信
            async with session.get(
                "https://api.holysheep.ai/v1/models",
                headers={"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY"}
            ) as response:
                if response.status == 200:
                    print("✅ 接続成功！")
                    return True
                elif response.status == 401:
                    print("❌ APIキーが無効です")
                    return False
                else:
                    print(f"❌ エラー: {response.status}")
                    return False
    except Exception as e:
        print(f"❌ 接続エラー: {e}")
        return False

asyncio.run(test_connection())

解決：.envファイル内のAPIキーが正しく設定されているか確認し、HolySheep AIのダッシュボードでキーを再確認してください。

エラー3：429 Too Many Requests（レート制限エラー）

原因：短時間にリクエストが多すぎます。HolySheep AIは経済的な価格（¥1=$1）を提供していますが、それでもレート制限はあります。

import asyncio
import aiohttp

async def call_with_rate_limit_handling(prompt: str, max_retries: int = 5):
    """レート制限を適切に処理するリクエスト"""
    
    headers = {
        "Authorization": "Bearer YOUR_HOLYSHEEP_API_KEY",
        "Content-Type": "application/json"
    }
    
    payload = {
        "model": "gpt-4.1",
        "messages": [{"role": "user", "content": prompt}]
    }
    
    for attempt in range(max_retries):
        async with aiohttp.ClientSession() as session:
            async with session.post(
                "https://api.holysheep.ai/v1/chat/completions",
                headers=headers,
                json=payload
            ) as response:
                
                if response.status == 200:
                    return await response.json()
                
                elif response.status == 429:
                    # Retry-Afterヘッダーを確認（秒単位）
                    retry_after = response.headers.get("Retry-After", "5")
                    wait_time = int(retry_after)
                    print(f"⏳ レート制限待機中... {wait_time}秒")
                    await asyncio.sleep(wait_time)
                    continue
                
                else:
                    return {"error": f"HTTP {response.status}"}
    
    return {"error": "最大リトライ回数を超過"}

使用例
async def main():
    result = await call_with_rate_limit_handling("テスト")
    print(result)

asyncio.run(main())

解決：Semaphoreを使用して同時実行数を制限し、429エラー時は指数バックオフで待機してください。HolySheep AIの<50msレイテンシを活かせば、より少ないリクエストで高いスループットを達成できます。

エラー4：json.decoder.JSONDecodeError

原因：レスポンスがJSON形式でない場合（HTMLエラー画面など）に発生します。

import asyncio
import aiohttp

async def safe_json_request(session, url, headers, json_data):
    """JSONパースを安全に処理するリクエスト"""
    
    try:
        async with session.post(url, headers=headers, json=json_data) as response:
            # レスポンスのContent-Typeを確認
            content_type = response.headers.get("Content-Type", "")
            
            if "application/json" not in content_type:
                # レスポンスボディをテキストとして取得
                text = await response.text()
                raise ValueError(
                    f"予期しないContent-Type: {content_type}\n"
                    f"レスポンス: {text[:500]}"
                )
            
            return await response.json()
            
    except aiohttp.ContentTypeError as e:
        print(f"⚠️ JSONではないレスポンス: {e}")
        raise
    except Exception as e:
        print(f"⚠️ リクエストエラー: {e}")
        raise

使用例
async def main():
    headers = {"Authorization": "Bearer YOUR_API_KEY"}
    
    try:
        async with aiohttp.ClientSession() as session:
            result = await safe_json_request(
                session,
                "https://api.holysheep.ai/v1/chat/completions",
                headers,
                {"model": "gpt-4.1", "messages": [{"role": "user", "content": "hello"}]}
            )
            print(result)
    except Exception as e:
        print(f"リクエスト失敗: {e}")

asyncio.run(main())

解決：レスポンスのContent-Typeを確認してからJSONパースを行うエラーハンドリングを実装してください。

まとめ

この記事では、Pythonのasyncioとaiohttpを使った非同期AI API呼び出しの基本を学びました。

主要なポイント

async/awaitを使って非同期関数を定義し、awaitで処理完了を待つ
asyncio.gather()で複数のタスクを同時に実行
Semaphoreで同時実行数を制御し、レート制限に対応
エラーハンドリングを実装して、堅牢なアプリケーションを作る

HolySheep AIでは、¥1=$1という経済的な料金体系と<50msの低レイテンシを組み合わせることで、非同期処理の効果を最大化できます。新規登録で無料クレジットも獲得できますので、ぜひ試してみてください。

👉 HolySheep AI に登録して無料クレジットを獲得

Python非同期AI API呼び出し：asyncioとaiohttp実践ガイド

前提知識と環境準備

必要なもの

必要なライブラリのインストール

インストール確認のテストコード

非同期処理とは？同步処理との違い

シンプルな非同期API呼び出し

環境変数の読み込み

APIキーの設定（.envファイルに HOLYSHEEP_API_KEY=あなたのキーを記述）

実行

コードのポイント解説

複数のAIリクエストを同時に送信する

実行結果の例

異なるモデルを同時に呼び出す

エラーハンドリングの実装

応用：バッチ処理の実装

よくあるエラーと対処法

エラー1：AttributeError: module 'asyncio' has no attribute 'run'

代わりにこの方法を使用

使用例

`run_async(main())`

エラー2：aiohttp.client_exceptions.ClientConnectorError: Cannot connect to host

エラー3：429 Too Many Requests（レート制限エラー）

使用例

エラー4：json.decoder.JSONDecodeError

使用例

まとめ

主要なポイント

関連リソース

関連記事

前提知識と環境準備

必要なもの

必要なライブラリのインストール

インストール確認のテストコード

非同期処理とは？同步処理との違い

シンプルな非同期API呼び出し

環境変数の読み込み

APIキーの設定（.envファイルに HOLYSHEEP_API_KEY=あなたのキー を記述）

実行

コードのポイント解説

複数のAIリクエストを同時に送信する

実行結果の例

異なるモデルを同時に呼び出す

エラーハンドリングの実装

応用：バッチ処理の実装

よくあるエラーと対処法

エラー1：AttributeError: module 'asyncio' has no attribute 'run'

代わりにこの方法を使用

使用例

run_async(main())

エラー2：aiohttp.client_exceptions.ClientConnectorError: Cannot connect to host

エラー3：429 Too Many Requests（レート制限エラー）

使用例

エラー4：json.decoder.JSONDecodeError

使用例

まとめ

主要なポイント

関連リソース

関連記事

🔥 HolySheep AIを使ってみる

APIキーの設定（.envファイルに HOLYSHEEP_API_KEY=あなたのキーを記述）

`run_async(main())`