トークン使用量の精算とレート制限戦略

トークン計算方法

DeepSeekの課金は、実際に処理されたトークン数に基づきます。1トークンは約0.75英単語または0.5中国語文字に相当します：

from openai import OpenAI

client = OpenAI(
    api_key='your-key',
    base_url='https://api.deepseek.com'
)

# 各リクエストはusage情報を返します
response = client.chat.completions.create(
    model='deepseek-v4-flash',
    messages=[{"role": "user", "content": "Hello"}]
)

print(f"Prompt tokens: {response.usage.prompt_tokens}")      # 入力消費
print(f"Completion tokens: {response.usage.completion_tokens}")  # 出力消費
print(f"Total tokens: {response.usage.total_tokens}")

# 思考モードではreasoning_tokensもあります
if hasattr(response.usage, 'reasoning_tokens'):
    print(f"Reasoning tokens: {response.usage.reasoning_tokens}")

リアルタイムコスト見積もり

# V4 Flashの価格（人民元、100万トークンあたり）
FLASH_INPUT_CACHE_HIT = 0.02
FLASH_INPUT_CACHE_MISS = 1.00
FLASH_OUTPUT = 2.00

def estimate_cost(prompt_tokens, completion_tokens, cache_hit=True):
    input_price = FLASH_INPUT_CACHE_HIT if cache_hit else FLASH_INPUT_CACHE_MISS
    input_cost = (prompt_tokens / 1_000_000) * input_price
    output_cost = (completion_tokens / 1_000_000) * FLASH_OUTPUT
    return input_cost + output_cost

# 例：3000トークン入力 + 500トークン出力
cost = estimate_cost(3000, 500, cache_hit=True)
print(f"今回の費用：¥{cost:.4f}")
# 出力：今回の費用：¥0.0011

レート制限

レート制限を超えると429ステータスコードが返され、バックオフリトライを実装する必要があります。

インテリジェントリトライ機構

import time
import random

def call_with_retry(func, max_retries=3):
    """指数バックオフ付きAPIリトライ"""
    for attempt in range(max_retries):
        try:
            return func()
        except Exception as e:
            if '429' in str(e) or 'rate_limit' in str(e).lower():
                wait = (2 ** attempt) + random.uniform(0, 1)
                print(f"レート制限、{wait:.1f}秒待って再試行...")
                time.sleep(wait)
            elif attempt == max_retries - 1:
                raise
            else:
                time.sleep(1)

# 使用
result = call_with_retry(lambda: client.chat.completions.create(
    model='deepseek-v4-flash',
    messages=[{"role": "user", "content": "Hello"}]
))

並行リクエストキュー

import asyncio
from asyncio import Semaphore

class DeepSeekClient:
    def __init__(self, max_concurrent=10):
        self.semaphore = Semaphore(max_concurrent)
        self.client = OpenAI(
            api_key=os.environ['DEEPSEEK_API_KEY'],
            base_url='https://api.deepseek.com'
        )

    async def chat_async(self, messages, model='deepseek-v4-flash'):
        async with self.semaphore:
            loop = asyncio.get_event_loop()
            return await loop.run_in_executor(
                None,
                lambda: self.client.chat.completions.create(
                    model=model,
                    messages=messages
                )
            )

# 100リクエストを並行処理
client = DeepSeekClient(max_concurrent=10)
async def process_batch(queries):
    tasks = [client.chat_async([{"role": "user", "content": q}]) for q in queries]
    results = await asyncio.gather(*tasks)
    return results

コスト監視ダッシュボードのアイデア

本番環境では、以下のメトリクスを記録することをお勧めします：

• 1日の総トークン消費量（モデル別）
• キャッシュヒット率（高いほどコスト削減）
• 平均応答遅延
• 429レート制限の発生回数
• ユーザー/APIキー別のコスト配分