DeepSeek API リトライ机制：幂等性と例外处理

DeepSeek API を本番環境に導入する際、ネットワーク障害やサーバー輻輳による一時的な失敗にどう 대응するかは、システム安定性の要となります。本稿では、HTTPS リクエストの冪等性（べきとうせい）を確保しつつ、HolySheep AI を网关とした効果的なリトライ机制の実装方法を解説します。

HolySheep AI vs 公式API vs 他のリレーサービス：比較表

今すぐ登録して、85%のコスト削減と<50msの低レイテンシを体験してください。

幂等性とは？なぜ重要か

幂等性とは、同じリクエストを何度実行しても結果が同じになる性質のことです。DeepSeek API を使用する際、以下のシナリオで問題が発生します：

• リトライ時の重複実行：ネットワークタイムアウト後、自动リトライ导致同一プロンプトが複数回実行
• 金融取引のリスク： pagamentos等の副作用を伴う操作の重複
• コスト增加：リトライの度にAPI呼び出し费用が発生

HolySheep AI は X-Idempotency-Key ヘッダーをサポートしており、同じキーを指定すれば応答がキャッシュされ、重複実行を防ぎます。

Python での実装例：包括的なリトライクラス

import time
import uuid
import logging
from typing import Optional, Dict, Any
from datetime import datetime, timedelta
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

logger = logging.getLogger(__name__)

class DeepSeekRetryClient:
    """
    HolySheep AI 网关を使用した DeepSeek API 用リトライクライアント
    特徴：
    - 指数バックオフによる段階的リトライ
    - 幂等性キーの自動生成与管理
    - ネットワーク障害の自动検出
    - コスト最適化（429 Too Many Requests対応）
    """
    
    def __init__(
        self,
        api_key: str,
        base_url: str = "https://api.holysheep.ai/v1",
        max_retries: int = 5,
        base_delay: float = 1.0,
        max_delay: float = 60.0,
        timeout: int = 120
    ):
        self.api_key = api_key
        self.base_url = base_url.rstrip('/')
        self.max_retries = max_retries
        self.base_delay = base_delay
        self.max_delay = max_delay
        self.timeout = timeout
        
        # 幂等性キー存储器（本番环境ではRedis等を使用）
        self._idempotency_cache: Dict[str, Dict[str, Any]] = {}
        
        # HTTPセッションのセットアップ
        self.session = self._create_session()
    
    def _create_session(self) -> requests.Session:
        """再接続可能なHTTPセッションを作成"""
        session = requests.Session()
        
        # 指数バックオフ策略でリトライを設定
        retry_strategy = Retry(
            total=3,  # urllib3レベルのリトライ（幂等性无关な一時的エラー）
            backoff_factor=0.5,
            status_forcelist=[429, 500, 502, 503, 504],
            allowed_methods=["POST", "GET"]
        )
        
        adapter = HTTPAdapter(max_retries=retry_strategy)
        session.mount("https://", adapter)
        
        return session
    
    def _generate_idempotency_key(self, prompt: str, model: str) -> str:
        """リクエスト内容に基づく幂等性キーを生成"""
        # 簡易実装：実際はハッシュを使用
        content_hash = str(hash(prompt + model + str(datetime.now().date())))
        return f"deepseek-{content_hash}"
    
    def _is_retryable_error(self, status_code: int, error_response: Optional[Dict]) -> bool:
        """リトライすべきエラーかどうか判定"""
        # 429: Rate Limit - バックオフ後にリトライ
        if status_code == 429:
            return True
        
        # 500-504: サーバー侧エラー - リトライ価値あり
        if 500 <= status_code <= 504:
            return True
        
        # ネットワークレベルエラー
        if error_response and error_response.get("error", {}).get("type") == "server_error":
            return True
        
        return False
    
    def _calculate_backoff(self, attempt: int, retry_after: Optional[int] = None) -> float:
        """指数バックオフ延迟を計算"""
        if retry_after:
            # Retry-After ヘッダーがある場合はそれを使用
            return min(retry_after, self.max_delay)
        
        # 指数バックオフ: base_delay * (2 ** attempt) + ランダム抖动
        import random
        delay = self.base_delay * (2 ** attempt)
        jitter = random.uniform(0, 0.5)  # 0-0.5秒の抖动
        return min(delay + jitter, self.max_delay)
    
    def chat_completions(
        self,
        messages: list,
        model: str = "deepseek-chat",
        temperature: float = 0.7,
        max_tokens: int = 2048,
        stream: bool = False
    ) -> Dict[str, Any]:
        """
        DeepSeek API へのリトライ機能付きリクエスト
        
        Args:
            messages: OpenAI互換フォーマットのメッセージリスト
            model: モデル名（deepseek-chat / deepseek-coder）
            temperature: 生成の多样性を制御
            max_tokens: 最大出力トークン数
            stream: ストリーミングモード
            
        Returns:
            APIレスポンス（dict形式）
        """
        url = f"{self.base_url}/chat/completions"
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        # 幂等性キーを生成
        prompt_text = "".join([m.get("content", "") for m in messages])
        idempotency_key = self._generate_idempotency_key(prompt_text, model)
        headers["X-Idempotency-Key"] = idempotency_key
        
        payload = {
            "model": model,
            "messages": messages,
            "temperature": temperature,
            "max_tokens": max_tokens,
            "stream": stream
        }
        
        last_exception = None
        
        for attempt in range(self.max_retries):
            try:
                response = self.session.post(
                    url,
                    headers=headers,
                    json=payload,
                    timeout=self.timeout
                )
                
                # 成功時
                if response.status_code == 200:
                    result = response.json()
                    logger.info(
                        f"API呼び出し成功: model={model}, "
                        f"tokens={result.get('usage', {}).get('total_tokens', 'N/A')}"
                    )
                    return result
                
                # 幂等性キーが原因で409 Conflictが返った場合
                if response.status_code == 409:
                    cached = self._idempotency_cache.get(idempotency_key)
                    if cached:
                        logger.info(f"キャッシュから幂等性结果を返却: {idempotency_key}")
                        return cached
                
                # リトライ対象外のエラー
                if not self._is_retryable_error(response.status_code, None):
                    error_detail = response.json() if response.text else {}
                    raise APIError(
                        f"APIエラー (リトライ不可): {response.status_code}",
                        status_code=response.status_code,
                        response=error_detail
                    )
                
                # Retry-After ヘッダーの解析
                retry_after = None
                if "Retry-After" in response.headers:
                    retry_after = int(response.headers["Retry-After"])
                
                # Rate Limit の場合はより長いバックオフ
                if response.status_code == 429:
                    logger.warning(
                        f"Rate Limit 到達 (Attempt {attempt + 1}/{self.max_retries})"
                    )
                
                delay = self._calculate_backoff(attempt, retry_after)
                logger.warning(
                    f"リトライ実行 (Attempt {attempt + 1}/{self.max_retries}): "
                    f"status={response.status_code}, delay={delay:.2f}秒"
                )
                time.sleep(delay)
                
            except requests.exceptions.Timeout:
                logger.warning(f"タイムアウト (Attempt {attempt + 1}/{self.max_retries})")
                delay = self._calculate_backoff(attempt)
                time.sleep(delay)
                
            except requests.exceptions.ConnectionError as e:
                logger.warning(f"接続エラー (Attempt {attempt + 1}/{self.max_retries}): {e}")
                delay = self._calculate_backoff(attempt)
                time.sleep(delay)
                last_exception = e
                
            except requests.exceptions.RequestException as e:
                logger.error(f"リクエスト例外: {e}")
                raise APIError(f"リクエスト失敗: {e}") from e
        
        # 全リトライ失敗
        raise APIError(
            f"最大リトライ回数 ({self.max_retries}) を超過",
            is_retriable=True,
            last_exception=last_exception
        )


class APIError(Exception):
    """API関連エラーのカスタム例外"""
    def __init__(
        self,
        message: str,
        status_code: int = None,
        response: dict = None,
        is_retriable: bool = False,
        last_exception: Exception = None
    ):
        super().__init__(message)
        self.status_code = status_code
        self.response = response
        self.is_retriable = is_retriable
        self.last_exception = last_exception


使用例
if __name__ == "__main__":
    # HolySheep AI 用のAPIクライアントを初期化
    client = DeepSeekRetryClient(
        api_key="YOUR_HOLYSHEEP_API_KEY",
        max_retries=5,
        base_delay=1.0,
        timeout=120
    )
    
    messages = [
        {"role": "system", "content": "あなたは有用なアシスタントです。"},
        {"role": "user", "content": "Pythonでの指数バックオフの実装例を説明してください。"}
    ]
    
    try:
        response = client.chat_completions(
            messages=messages,
            model="deepseek-chat",
            temperature=0.7,
            max_tokens=2048
        )
        print(f"生成結果: {response['choices'][0]['message']['content']}")
        
        # 使用量とコストを表示
        usage = response.get('usage', {})
        print(f"入力トークン: {usage.get('prompt_tokens', 0)}")
        print(f"出力トークン: {usage.get('completion_tokens', 0)}")
        print(f"合計トークン: {usage.get('total_tokens', 0)}")
        
    except APIError as e:
        print(f"APIエラー: {e}")
        if e.is_retriable:
            print("このエラーはリトライ対象です。数秒後に再試行してください。")

TypeScript/JavaScript での実装例

/**
 * HolySheep AI DeepSeek API 用リトライクライアント (Node.js/TypeScript)
 * 
 * 特徴:
 * - async/await 対応
 * - AbortController によるタイムアウト制御
 * - 指数バックオフ＋抖动
 * - 幂等性キーの自動管理
 */

interface RetryOptions {
  maxRetries?: number;
  baseDelay?: number;
  maxDelay?: number;
  timeout?: number;
}

interface DeepSeekMessage {
  role: 'system' | 'user' | 'assistant';
  content: string;
}

interface DeepSeekResponse {
  id: string;
  choices: Array<{
    message: DeepSeekMessage;
    finish_reason: string;
    index: number;
  }>;
  usage: {
    prompt_tokens: number;
    completion_tokens: number;
    total_tokens: number;
  };
  created: number;
  model: string;
}

class DeepSeekRetryError extends Error {
  constructor(
    message: string,
    public readonly statusCode?: number,
    public readonly isRetriable: boolean = false,
    public readonly response?: any
  ) {
    super(message);
    this.name = 'DeepSeekRetryError';
  }
}

class DeepSeekRetryClient {
  private readonly baseUrl = 'https://api.holysheep.ai/v1';
  private readonly maxRetries: number;
  private readonly baseDelay: number;
  private readonly maxDelay: number;
  private readonly timeout: number;
  private readonly idempotencyCache: Map = new Map();

  constructor(
    private readonly apiKey: string,
    options: RetryOptions = {}
  ) {
    this.maxRetries = options.maxRetries ?? 5;
    this.baseDelay = options.baseDelay ?? 1000;
    this.maxDelay = options.maxDelay ?? 60000;
    this.timeout = options.timeout ?? 120000;
  }

  private generateIdempotencyKey(prompt: string, model: string): string {
    const timestamp = new Date().toISOString().split('T')[0];
    const hash = this.simpleHash(prompt + model + timestamp);
    return deepseek-${hash};
  }

  private simpleHash(str: string): string {
    let hash = 0;
    for (let i = 0; i < str.length; i++) {
      const char = str.charCodeAt(i);
      hash = ((hash << 5) - hash) + char;
      hash = hash & hash;
    }
    return Math.abs(hash).toString(36);
  }

  private async sleep(ms: number): Promise {
    return new Promise(resolve => setTimeout(resolve, ms));
  }

  private calculateBackoff(attempt: number, retryAfter?: number): number {
    if (retryAfter) {
      return Math.min(retryAfter * 1000, this.maxDelay);
    }
    
    // 指数バックオフ: baseDelay * 2^attempt + 抖动(0-500ms)
    const exponentialDelay = this.baseDelay * Math.pow(2, attempt);
    const jitter = Math.random() * 500;
    return Math.min(exponentialDelay + jitter, this.maxDelay);
  }

  private isRetriableStatus(statusCode: number): boolean {
    // 429: Rate Limit
    if (statusCode === 429) return true;
    
    // 500-504: サーバーエラー
    if (statusCode >= 500 && statusCode <= 504) return true;
    
    return false;
  }

  async chatCompletions(
    messages: DeepSeekMessage[],
    model: string = 'deepseek-chat',
    options: {
      temperature?: number;
      maxTokens?: number;
      stream?: boolean;
    } = {}
  ): Promise {
    const { temperature = 0.7, maxTokens = 2048, stream = false } = options;
    
    const url = ${this.baseUrl}/chat/completions;
    const promptText = messages.map(m => m.content).join('');
    const idempotencyKey = this.generateIdempotencyKey(promptText, model);
    
    // キャッシュチェック
    const cached = this.idempotencyCache.get(idempotencyKey);
    if (cached) {
      console.log([DeepSeekRetry] キャッシュから返却: ${idempotencyKey});
      return cached;
    }
    
    const payload = {
      model,
      messages,
      temperature,
      max_tokens: maxTokens,
      stream,
    };

    let lastError: Error | null = null;

    for (let attempt = 0; attempt < this.maxRetries; attempt++) {
      try {
        const controller = new AbortController();
        const timeoutId = setTimeout(() => controller.abort(), this.timeout);

        const response = await fetch(url, {
          method: 'POST',
          headers: {
            'Authorization': Bearer ${this.apiKey},
            'Content-Type': 'application/json',
            'X-Idempotency-Key': idempotencyKey,
          },
          body: JSON.stringify(payload),
          signal: controller.signal,
        });

        clearTimeout(timeoutId);

        if (response.ok) {
          const result: DeepSeekResponse = await response.json();
          
          // 結果をキャッシュ
          this.idempotencyCache.set(idempotencyKey, result);
          
          console.log(
            [DeepSeekRetry] 成功: model=${model},  +
            tokens=${result.usage?.total_tokens ?? 'N/A'}
          );
          
          return result;
        }

        // 409 Conflict: 幂等性キーの重複
        if (response.status === 409) {
          const cachedResult = this.idempotencyCache.get(idempotencyKey);
          if (cachedResult) {
            return cachedResult;
          }
        }

        // リトライ不可のエラー
        if (!this.isRetriableStatus(response.status)) {
          const errorBody = await response.text();
          throw new DeepSeekRetryError(
            APIエラー (${response.status}): ${errorBody},
            response.status,
            false
          );
        }

        // Retry-After ヘッダーの取得
        const retryAfter = response.headers.get('Retry-After');
        const delay = this.calculateBackoff(attempt, retryAfter ? parseInt(retryAfter) : undefined);

        console.warn(
          [DeepSeekRetry] リトライ (${attempt + 1}/${this.maxRetries}):  +
          status=${response.status}, delay=${delay.toFixed(0)}ms
        );

        await this.sleep(delay);

      } catch (error) {
        if (error instanceof DeepSeekRetryError) {
          throw error;
        }

        if (error instanceof DOMException && error.name === 'AbortError') {
          console.warn([DeepSeekRetry] タイムアウト (Attempt ${attempt + 1}));
          lastError = new Error('リクエストタイムアウト');
        } else if (error instanceof TypeError) {
          // ネットワークエラー
          console.warn([DeepSeekRetry] 接続エラー: ${error.message});
          lastError = error as Error;
        } else {
          lastError = error as Error;
        }

        if (attempt < this.maxRetries - 1) {
          const delay = this.calculateBackoff(attempt);
          await this.sleep(delay);
        }
      }
    }

    throw new DeepSeekRetryError(
      最大リトライ回数 (${this.maxRetries}) を超過: ${lastError?.message},
      undefined,
      true
    );
  }
}

// 使用例
async function main() {
  const client = new DeepSeekRetryClient(
    process.env.YOUR_HOLYSHEEP_API_KEY || 'YOUR_HOLYSHEEP_API_KEY',
    {
      maxRetries: 5,
      baseDelay: 1000,
      timeout: 120000,
    }
  );

  const messages: DeepSeekMessage[] = [
    { role: 'system', content: 'あなたは有用なアシスタントです。' },
    { role: 'user', content: 'JavaScriptでの例外处理のベストプラクティスを教えてください。' },
  ];

  try {
    const response = await client.chatCompletions(
      messages,
      'deepseek-chat',
      {
        temperature: 0.7,
        maxTokens: 2048,
      }
    );

    console.log('\n=== API レスポンス ===');
    console.log(生成結果:\n${response.choices[0].message.content});
    console.log(\n入力トークン: ${response.usage.prompt_tokens});
    console.log(出力トークン: ${response.usage.completion_tokens});
    console.log(合計トークン: ${response.usage.total_tokens});

    // コスト計算 (DeepSeek V3.2: $0.42/MTok出力)
    const outputCost = (response.usage.completion_tokens / 1_000_000) * 0.42;
    console.log(`\
関連リソース
📚 AI API 記事一覧
💰 料金を見る
📖 開発者ドキュメント
🚀 無料登録
関連記事
GPT-4o Realtime API 音声交互：WebSocket 流式传输实战 — HolySheep 移行プレイブ
K12 教育 AI アプリケーション移行プレイブック：课后辅导と知识点強化 приложений к HolySheep
Claude API 呼び出し降格戦略：Opus から Haiku への自動切り替え実装ガイド