AI API 负载测试：Locust + k6 压测方案完全ガイド

Locust でのAI API負荷テスト実装

Locust は Python ベースの負荷テストツールで、シナリオ記述が容易な点が特徴です。<\/p>

# locust_ai_loadtest.py
from locust import HttpUser, task, between
import json
import random

class AIStressUser(HttpUser):
    # 各ユーザーの待機時間（1-5秒）
    wait_time = between(1, 5)
    
    def on_start(self):
        """テスト開始時にAPIキーを設定"""
        self.api_key = "YOUR_HOLYSHEEP_API_KEY"
        self.base_url = "https://api.holysheep.ai/v1"
    
    @task(3)
    def chat_completion_deepseek(self):
        """DeepSeek V3.2 でのチャット完了テスト（重み付け3）"""
        prompt_variations = [
            "AI APIの負荷テストについて150文字で説明してください",
            "Pythonでの非同期処理の利点を簡潔に教えてください",
            "KubernetesとDockerの違いを50文字で",
        ]
        
        payload = {
            "model": "deepseek-chat",
            "messages": [
                {"role": "user", "content": random.choice(prompt_variations)}
            ],
            "max_tokens": 500,
            "temperature": 0.7
        }
        
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        with self.client.post(
            f"{self.base_url}/chat/completions",
            json=payload,
            headers=headers,
            catch_response=True,
            name="/chat/completions - DeepSeek"
        ) as response:
            if response.elapsed.total_seconds() < 2.0:
                response.success()
            elif response.elapsed.total_seconds() < 5.0:
                response.success()  # 5秒以内は許容
            else:
                response.failure(f"Timeout: {response.elapsed.total_seconds()}s")
    
    @task(1)
    def chat_completion_gpt(self):
        """GPT-4.1 でのテスト（重み付け1）"""
        payload = {
            "model": "gpt-4.1",
            "messages": [
                {"role": "user", "content": "こんにちは、简要的に自己紹介してください"}
            ],
            "max_tokens": 300
        }
        
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        with self.client.post(
            f"{self.base_url}/chat/completions",
            json=payload,
            headers=headers,
            catch_response=True,
            name="/chat/completions - GPT-4.1"
        ) as response:
            if response.status_code == 200:
                response.success()
            elif response.status_code == 429:
                response.failure("Rate Limited")
            else:
                response.failure(f"HTTP {response.status_code}")
    
    @task(2)
    def embedding_test(self):
        """Embedding API のテスト（重み付け2）"""
        sample_text = "これは負荷テスト用のサンプルテキストです" * 10
        
        payload = {
            "model": "text-embedding-3-small",
            "input": sample_text
        }
        
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        self.client.post(
            f"{self.base_url}/embeddings",
            json=payload,
            headers=headers,
            name="/embeddings"
        )

実行コマンド：<\/p>

# 100ユーザー、spawn rate 10で実行
locust -f locust_ai_loadtest.py \
  --host=https://api.holysheep.ai \
  --users=100 \
  --spawn-rate=10 \
  --run-time=300s \
  --headless \
  --html=report.html

Web UI で確認する場合
locust -f locust_ai_loadtest.py --host=https://api.holysheep.ai

k6 でのAI API負荷テスト実装

k6 は Go 製の高性能負荷テストツールで、結果のビジュアル化が優秀です。<\/p>

// k6_ai_loadtest.js
import http from 'k6/http';
import { check, sleep } from 'k6';
import { Rate, Trend } from 'k6/metrics';

// カスタムメトリクス定義
const latency = new Trend('ai_latency');
const tokenCost = new Trend('token_cost');
const errorRate = new Rate('errors');

// テスト設定
export const options = {
  stages: [
    { duration: '30s', target: 20 },   // ウォームアップ
    { duration: '1m', target: 50 },    // 段階的に増加
    { duration: '2m', target: 100 },   // ピーク負荷
    { duration: '30s', target: 0 },    // クールダウン
  ],
  thresholds: {
    'ai_latency': ['p(95)<5000'],      // 95%ile 5秒以内
    'errors': ['rate<0.05'],           // エラー率5%未満
    'http_req_duration': ['p(99)<8000'],
  },
};

// テストシナリオ
export default function () {
  const apiKey = 'YOUR_HOLYSHEEP_API_KEY';
  const baseUrl = 'https://api.holysheep.ai/v1';
  
  // ランダムなモデル選択（実際のトラフィック分布を想定）
  const models = [
    { name: 'deepseek-chat', weight: 0.5, input: 800, output: 400 },
    { name: 'gpt-4.1', weight: 0.3, input: 500, output: 200 },
    { name: 'claude-sonnet-4-5', weight: 0.2, input: 600, output: 300 },
  ];
  
  const selectedModel = weightedRandom(models);
  
  // テストプロンプト（多様なパターンを生成）
  const prompts = [
    製品コード ${__VU}-${__ITER} のレビューを短く纏めてください。,
    次のデータ構造を最適化してください： ${generateDummyData()},
    APIの設計パターンを3つ説明してください。,
  ];
  
  const payload = JSON.stringify({
    model: selectedModel.name,
    messages: [
      { role: 'system', content: 'あなたは有用なアシスタントです。' },
      { role: 'user', content: prompts[Math.floor(Math.random() * prompts.length)] }
    ],
    max_tokens: selectedModel.output,
    temperature: 0.7,
  });
  
  const params = {
    headers: {
      'Authorization': Bearer ${apiKey},
      'Content-Type': 'application/json',
    },
    tags: { name: selectedModel.name },
  };
  
  // リクエスト実行
  const startTime = Date.now();
  const response = http.post(${baseUrl}/chat/completions, payload, params);
  const latencyMs = Date.now() - startTime;
  
  // メトリクス記録
  latency.add(latencyMs);
  
  // レスポンス検証
  const result = check(response, {
    'status is 200': (r) => r.status === 200,
    'has content': (r) => r.body.length > 0,
    'has usage data': (r) => {
      try {
        const data = JSON.parse(r.body);
        return data.usage && data.usage.total_tokens > 0;
      } catch (e) {
        return false;
      }
    },
    'no error field': (r) => {
      try {
        const data = JSON.parse(r.body);
        return !data.error;
      } catch (e) {
        return true;
      }
    },
  });
  
  if (!result) {
    errorRate.add(1);
    console.error(Error response: ${response.body});
  } else {
    errorRate.add(0);
    // トークンコスト計算（概算）
    try {
      const data = JSON.parse(response.body);
      if (data.usage) {
        const cost = calculateCost(selectedModel.name, data.usage);
        tokenCost.add(cost);
      }
    } catch (e) {}
  }
  
  sleep(Math.random() * 2 + 1);  // 1-3秒待機
}

// 加重ランダム選択
function weightedRandom(items) {
  const totalWeight = items.reduce((sum, item) => sum + item.weight, 0);
  let random = Math.random() * totalWeight;
  
  for (const item of items) {
    random -= item.weight;
    if (random <= 0) return item;
  }
  return items[items.length - 1];
}

// ダミーデータ生成
function generateDummyData() {
  const data = [];
  for (let i = 0; i < 50; i++) {
    data.push({ id: i, value: Math.random().toString(36) });
  }
  return JSON.stringify(data);
}

// コスト計算（2026年価格）
function calculateCost(model, usage) {
  const prices = {
    'deepseek-chat': { input: 0.27, output: 0.42 },    // $0.27/$0.42 per MTok
    'gpt-4.1': { input: 2, output: 8 },                  // $2/$8 per MTok
    'claude-sonnet-4-5': { input: 3, output: 15 },      // $3/$15 per MTok
  };
  
  const price = prices[model] || { input: 1, output: 5 };
  const inputCost = (usage.prompt_tokens / 1_000_000) * price.input;
  const outputCost = (usage.completion_tokens / 1_000_000) * price.output;
  
  return inputCost + outputCost;
}

実行コマンド：<\/p>

# 基本的な実行
k6 run k6_ai_loadtest.js

Cloud実行（ результат をクラウドに保存）
k6 run -o cloud k6_ai_loadtest.js

InfluxDB + Grafana で可視化
k6 run \
  --out influxdb=http://localhost:8086/k6 \
  k6_ai_loadtest.js

実行後にInfluxDBにクエリ
SELECT percentile(value, 95) FROM ai_latency WHERE time > now() - 1h

Locust vs k6：実践的な比較