【2026年版】Claude API使い方の常識を疑え|公式推奨が最適とは限らない理由

import anthropic

client = anthropic.Anthropic(api_key="your-api-key")
message = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    messages=[{"role": "user", "content": "記事を書いて"}]
)

class ClaudeRouter:
    def __init__(self):
        # 2025年1月時点の料金（参考値、最新情報は公式サイトを確認）
        self.model_costs = {
            "claude-3-opus-20240229": {"input": 15, "output": 75},      # $/M tokens
            "claude-3-5-sonnet-20241022": {"input": 3, "output": 15},
            "claude-3-haiku-20240307": {"input": 0.25, "output": 1.25}
        }

    def select_model(self, task_type: str, estimated_tokens: int) -> str:
        """タスクタイプと予想トークン数からモデルを選択"""
        if task_type == "analysis" or estimated_tokens > 50000:
            return "claude-3-opus-20240229"
        elif task_type in ["generation", "rewrite"]:
            return "claude-3-5-sonnet-20241022"
        else:  # summary, classification, etc.
            return "claude-3-haiku-20240307"

    def estimate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
        """コストを事前計算（円換算、1ドル=150円）"""
        costs = self.model_costs[model]
        input_cost = (input_tokens / 1_000_000) * costs["input"] * 150
        output_cost = (output_tokens / 1_000_000) * costs["output"] * 150
        return input_cost + output_cost

class CacheStrategy:
    def __init__(self):
        self.cache_stats = {"hits": 0, "misses": 0}

    def should_use_cache(self, system_prompt: str, frequency: int) -> bool:
        """キャッシュを使うべきか判定"""
        # システムプロンプトが1024トークン以上、かつ5回以上再利用される場合のみキャッシュ
        token_count = self._count_tokens(system_prompt)
        return token_count >= 1024 and frequency >= 5

    def create_cached_message(self, system_prompt: str, user_message: str):
        """キャッシュ対応メッセージを作成"""
        return {
            "model": "claude-3-5-sonnet-20241022",
            "max_tokens": 2048,
            "system": [
                {
                    "type": "text",
                    "text": system_prompt,
                    "cache_control": {"type": "ephemeral"}
                }
            ],
            "messages": [{"role": "user", "content": user_message}]
        }

    def _count_tokens(self, text: str) -> int:
        """トークン数を推定（日本語は1文字あたり約0.65トークン）"""
        return int(len(text) * 0.65)

import asyncio
from typing import Optional

class ResilientClient:
    def __init__(self, api_key: str):
        self.client = anthropic.AsyncAnthropic(api_key=api_key)
        self.rate_limit_remaining = 50000  # Tier 1の例: 50K tokens/min
        self.rate_limit_reset = 0

    async def create_with_retry(self, **kwargs) -> Optional[dict]:
        """リトライ機能付きメッセージ作成"""
        max_retries = 3
        base_delay = 1

        for attempt in range(max_retries):
            try:
                # Rate Limit チェック
                await self._wait_for_rate_limit()

                response = await self.client.messages.create(**kwargs)
                return response

            except anthropic.RateLimitError as e:
                if attempt == max_retries - 1:
                    raise
                # 指数バックオフ
                delay = base_delay * (2 ** attempt)
                await asyncio.sleep(delay)

            except anthropic.APIError as e:
                if e.status_code >= 500:  # サーバーエラー
                    if attempt == max_retries - 1:
                        raise
                    await asyncio.sleep(base_delay)
                else:
                    raise  # クライアントエラーは即座に失敗

    async def _wait_for_rate_limit(self):
        """Rate Limit到達時は待機"""
        if self.rate_limit_remaining < 10000:  # 残り1万トークン未満
            wait_time = max(0, self.rate_limit_reset - time.time())
            if wait_time > 0:
                await asyncio.sleep(wait_time)
            self.rate_limit_remaining = 50000  # リセット

class CostController:
    def __init__(self, daily_budget_yen: int = 10000):
        self.daily_budget = daily_budget_yen
        self.daily_spent = 0
        self.reset_date = datetime.now().date()

    def check_budget(self, estimated_cost_yen: float) -> bool:
        """予算内かチェック"""
        if datetime.now().date() > self.reset_date:
            self.daily_spent = 0
            self.reset_date = datetime.now().date()

        if self.daily_spent + estimated_cost_yen > self.daily_budget:
            return False
        return True

    def pre_calculate_cost(self, prompt: str, max_output_tokens: int, model: str) -> float:
        """リクエスト前にコストを計算"""
        # 日本語は1文字あたり約0.65トークン
        input_tokens = int(len(prompt) * 0.65)

        router = ClaudeRouter()
        estimated_cost = router.estimate_cost(
            model=model,
            input_tokens=input_tokens,
            output_tokens=max_output_tokens
        )
        return estimated_cost

# 使用例
controller = CostController(daily_budget_yen=10000)
estimated = controller.pre_calculate_cost(
    prompt="長いプロンプト...",
    max_output_tokens=2048,
    model="claude-3-5-sonnet-20241022"
)

if not controller.check_budget(estimated):
    raise Exception("日次予算を超過します")

# 絶対にやってはいけない例
client = anthropic.Anthropic(api_key="sk-ant-api03-xxxxx")

import os
from dotenv import load_dotenv

load_dotenv()
api_key = os.getenv("ANTHROPIC_API_KEY")

if not api_key:
    raise ValueError("ANTHROPIC_API_KEY environment variable is not set")

client = anthropic.Anthropic(api_key=api_key)

def validate_context_window(system_prompt: str, user_prompt: str, max_tokens: int, model: str) -> bool:
    """コンテキストウィンドウ超過をチェック"""
    # モデルごとのコンテキスト制限
    CONTEXT_LIMITS = {
        "claude-3-opus-20240229": 200000,
        "claude-3-5-sonnet-20241022": 200000,
        "claude-3-haiku-20240307": 200000
    }

    CONTEXT_LIMIT = CONTEXT_LIMITS.get(model, 200000)
    SAFETY_MARGIN = 10000  # 安全マージン

    # 日本語は1文字あたり約0.65トークン
    system_tokens = int(len(system_prompt) * 0.65)
    user_tokens = int(len(user_prompt) * 0.65)
    total = system_tokens + user_tokens + max_tokens

    if total > CONTEXT_LIMIT - SAFETY_MARGIN:
        return False
    return True

def estimate_japanese_tokens(text: str) -> int:
    """日本語テキストのトークン数を推定"""
    char_count = len(text)
    # 日本語は1文字あたり約0.65トークン
    return int(char_count * 0.65)

# より正確な計算が必要な場合は、公式のcount_tokensメソッドを使用
def count_tokens_accurate(text: str, client: anthropic.Anthropic) -> int:
    """正確なトークン数をカウント（公式ライブラリ使用）"""
    # Anthropic公式のトークンカウント機能を使用
    # 注: 実際の実装では公式ドキュメントの最新メソッドを参照
    return client.count_tokens(text)

import logging
from datetime import datetime

# ロギング設定
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler(f'claude_api_{datetime.now().strftime("%Y%m%d")}.log'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

class LoggingClaudeClient:
    def __init__(self, api_key: str):
        self.client = anthropic.Anthropic(api_key=api_key)

    def create_message(self, **kwargs):
        """ログ記録付きメッセージ作成"""
        request_id = datetime.now().strftime("%Y%m%d%H%M%S%f")

        logger.info(f"[{request_id}] Request started: model={kwargs.get('model')}")

        try:
            response = self.client.messages.create(**kwargs)

            logger.info(
                f"[{request_id}] Request succeeded: "
                f"input_tokens={response.usage.input_tokens}, "
                f"output_tokens={response.usage.output_tokens}"
            )

            return response

        except Exception as e:
            logger.error(
                f"[{request_id}] Request failed: {type(e).__name__}: {str(e)}"
            )
            raise

import httpx

# タイムアウト設定付きクライアント
client = anthropic.Anthropic(
    api_key=api_key,
    timeout=httpx.Timeout(60.0, connect=5.0)  # 接続5秒、全体60秒
)

import os
import asyncio
import logging
import time
from typing import Optional, Dict, Any
from datetime import datetime
import anthropic
from anthropic import AsyncAnthropic, APIError, RateLimitError
import httpx

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class ProductionClaudeClient:
    """本番運用に耐えるClaude APIクライアント"""

    def __init__(
        self,
        api_key: Optional[str] = None,
        daily_budget_yen: int = 10000,
        default_model: str = "claude-3-5-sonnet-20241022"
    ):
        self.api_key = api_key or os.getenv("ANTHROPIC_API_KEY")
        if not self.api_key:
            raise ValueError("API key is required")

        self.client = AsyncAnthropic(
            api_key=self.api_key,
            timeout=httpx.Timeout(60.0, connect=5.0)
        )
        self.default_model = default_model
        self.daily_budget = daily_budget_yen
        self.daily_spent = 0
        self.reset_date = datetime.now().date()

        # モデル別料金（円換算、1ドル=150円、2025年1月時点の参考値）
        self.model_costs = {
            "claude-3-opus-20240229": {"input": 2250, "output": 11250},  # ¥/M tokens
            "claude-3-5-sonnet-20241022": {"input": 450, "output": 2250},
            "claude-3-haiku-20240307": {"input": 37.5, "output": 187.5}
        }

    async def create_message(
        self,
        prompt: str,
        system_prompt: Optional[str] = None,
        max_tokens: int = 2048,
        model: Optional[str] = None,
        temperature: float = 0.3,
        use_cache: bool = False
    ) -> Dict[str, Any]:
        """メッセージ作成（エラーハンドリング完備）"""

        model = model or self.default_model

        # 1. コスト事前チェック
        estimated_cost = self._estimate_cost(prompt, system_prompt, max_tokens, model)
        if not self._check_budget(estimated_cost):
            raise Exception(f"Daily budget exceeded. Estimated: ¥{estimated_cost:.2f}")

        # 2. コンテキスト長チェック
        if not self._validate_context(prompt, system_prompt, max_tokens, model):
            raise ValueError("Context window exceeded")

        # 3. リクエスト実行（リトライ付き）
        response = await self._execute_with_retry(
            prompt=prompt,
            system_prompt=system_prompt,
            max_tokens=max_tokens,
            model=model,
            temperature=temperature,
            use_cache=use_cache
        )

        # 4. コスト記録
        actual_cost = self._calculate_actual_cost(response, model)
        self.daily_spent += actual_cost

        logger.info(f"Request completed. Cost: ¥{actual_cost:.2f}, Daily total: ¥{self.daily_spent:.2f}")

        return response

    async def _execute_with_retry(
        self,
        prompt: str,
        system_prompt: Optional[str],
        max_tokens: int,
        model: str,
        temperature: float,
        use_cache: bool,
        max_retries: int = 3
    ) -> Dict[str, Any]:
        """リトライ機能付きリクエスト実行"""

        base_delay = 1

        for attempt in range(max_retries):
            try:
                # メッセージ構築
                messages = [{"role": "user", "content": prompt}]

                kwargs = {
                    "model": model,
                    "max_tokens": max_tokens,
                    "temperature": temperature,
                    "messages": messages
                }

                # システムプロンプト（キャッシュ対応）
                if system_prompt:
                    if use_cache:
                        kwargs["system"] = [
                            {
                                "type": "text",
                                "text": system_prompt,
                                "cache_control": {"type": "ephemeral"}
                            }
                        ]
                    else:
                        kwargs["system"] = system_prompt

                response = await self.client.messages.create(**kwargs)
                return response

            except RateLimitError as e:
                if attempt == max_retries - 1:
                    logger.error(f"Rate limit exceeded after {max_retries} retries")
                    raise

                delay = base_delay * (2 ** attempt)
                logger.warning(f"Rate limit hit. Retrying in {delay}s...")
                await asyncio.sleep(delay)

            except APIError as e:
                if e.status_code >= 500:  # サーバーエラー
                    if attempt == max_retries - 1:
                        logger.error(f"Server error after {max_retries} retries: {e}")
                        raise

                    logger.warning(f"Server error. Retrying... ({e})")
                    await asyncio.sleep(base_delay)
                else:
                    # クライアントエラーは即座に失敗
                    logger.error(f"Client error: {e}")
                    raise

            except Exception as e:
                logger.error(f"Unexpected error: {type(e).__name__}: {str(e)}")
                raise

    def _estimate_cost(
        self,
        prompt: str,
        system_prompt: Optional[str],
        max_tokens: int,
        model: str
    ) -> float:
        """コストを事前見積もり"""
        input_tokens = len(prompt) * 0.65  # 日本語想定
        if system_prompt:
            input_tokens += len(system_prompt) * 0.65

        costs = self.model_costs.get(model, self.model_costs["claude-3-5-sonnet-20241022"])
        input_cost = (input_tokens / 1_000_000) * costs["input"]
        output_cost = (max_tokens / 1_000_000) * costs["output"]

        return input_cost + output_cost

    def _calculate_actual_cost(self, response: Any, model: str) -> float:
        """実際のコストを計算"""
        costs = self.model_costs.get(model, self.model_costs["claude-3-5-sonnet-20241022"])

        input_tokens = response.usage.input_tokens
        output_tokens = response.usage.output_tokens

        input_cost = (input_tokens / 1_000_000) * costs["input"]
        output_cost = (output_tokens / 1_000_000) * costs["output"]

        return input_cost + output_cost

    def _check_budget(self, estimated_cost: float) -> bool:
        """予算チェック"""
        if datetime.now().date() > self.reset_date:
            self.daily_spent = 0
            self.reset_date = datetime.now().date()

        return (self.daily_spent + estimated_cost) <= self.daily_budget

    def _validate_context(
        self,
        prompt: str,
        system_prompt: Optional[str],
        max_tokens: int,
        model: str
    ) -> bool:
        """コンテキストウィンドウ検証"""
        CONTEXT_LIMITS = {
            "claude-3-opus-20240229": 200000,
            "claude-3-5-sonnet-20241022": 200000,
            "claude-3-haiku-20240307": 200000
        }

        limit = CONTEXT_LIMITS.get(model, 200000)
        safety_margin = 10000

        total_tokens = len(prompt) * 0.65
        if system_prompt:
            total_tokens += len(system_prompt) * 0.65
        total_tokens += max_tokens

        return total_tokens <= (limit - safety_margin)


# 使用例
async def main():
    client = ProductionClaudeClient(daily_budget_yen=10000)

    response = await client.create_message(
        prompt="AIについて500文字で説明してください",
        system_prompt="あなたは技術ライターです。わかりやすく説明してください。",
        max_tokens=1024,
        model="claude-3-5-sonnet-20241022",
        temperature=0.3,
        use_cache=True
    )

    print(response.content[0].text)

# 実行
if __name__ == "__main__":
    asyncio.run(main())

import Anthropic from '@anthropic-ai/sdk';

interface ClaudeClientConfig {
  apiKey?: string;
  dailyBudgetYen?: number;
  defaultModel?: string;
}

class ProductionClaudeClient {
  private client: Anthropic;
  private dailyBudget: number;
  private dailySpent: number = 0;
  private resetDate: Date;
  private defaultModel: string;

  private modelCosts = {
    'claude-3-opus-20240229': { input: 2250, output: 11250 },
    'claude-3-5-sonnet-20241022': { input: 450, output: 2250 },
    'claude-3-haiku-20240307': { input: 37.5, output: 187.5 }
  };

  constructor(config: ClaudeClientConfig = {}) {
    const apiKey = config.apiKey || process.env.ANTHROPIC_API_KEY;
    if (!apiKey) {
      throw new Error('API key is required');
    }

    this.client = new Anthropic({
      apiKey,
      timeout: 60000,
    });

    this.dailyBudget = config.dailyBudgetYen || 10000;
    this.defaultModel = config.defaultModel || 'claude-3-5-sonnet-20241022';
    this.resetDate = new Date();
  }

  async createMessage(params: {
    prompt: string;
    systemPrompt?: string;
    maxTokens?: number;
    model?: string;
    temperature?: number;
  }) {
    const {
      prompt,
      systemPrompt,
      maxTokens = 2048,
      model = this.defaultModel,
      temperature = 0.3
    } = params;

    // コスト事前チェック
    const estimatedCost = this.estimateCost(prompt, systemPrompt, maxTokens, model);
    if (!this.checkBudget(estimatedCost)) {
      throw new Error(`Daily budget exceeded. Estimated: ¥${estimatedCost.toFixed(2)}`);
    }

    // リクエスト実行（リトライ付き）
    const response = await this.executeWithRetry({
      prompt,
      systemPrompt,
      maxTokens,
      model,
      temperature
    });

    // コスト記録
    const actualCost = this.calculateActualCost(response, model);
    this.dailySpent += actualCost;

    console.log(`Request completed. Cost: ¥${actualCost.toFixed(2)}, Daily total: ¥${this.dailySpent.toFixed(2)}`);

    return response;
  }

  private async executeWithRetry(params: any, maxRetries: number = 3): Promise<any> {
    let lastError;

    for (let attempt = 0; attempt < maxRetries; attempt++) {
      try {
        const messages = [{ role: 'user' as const, content: params.prompt }];

        const requestParams: any = {
          model: params.model,
          max_tokens: params.maxTokens,
          temperature: params.temperature,
          messages
        };

        if (params.systemPrompt) {
          requestParams.system = params.systemPrompt;
        }

        return await this.client.messages.create(requestParams);

      } catch (error: any) {
        lastError = error;

        if (error.status === 429) {
          const delay = 1000 * Math.pow(2, attempt);
          console.warn(`Rate limit hit. Retrying in ${delay}ms...`);
          await this.sleep(delay);
        } else if (error.status >= 500) {
          console.warn(`Server error. Retrying...`);
          await this.sleep(1000);
        } else {
          throw error;
        }
      }
    }

    throw lastError;
  }

  private estimateCost(
    prompt: string,
    systemPrompt: string | undefined,
    maxTokens: number,
    model: string
  ): number {
    let inputTokens = prompt.length * 0.65;
    if (systemPrompt) {
      inputTokens += systemPrompt.length * 0.65;
    }

    const costs = this.modelCosts[model as keyof typeof this.modelCosts];
    const inputCost = (inputTokens / 1_000_000) * costs.input;
    const outputCost = (maxTokens / 1_000_000) * costs.output;

    return inputCost + outputCost;
  }

  private calculateActualCost(response: any, model: string): number {
    const costs = this.modelCosts[model as keyof typeof this.modelCosts];
    const inputCost = (response.usage.input_tokens / 1_000_000) * costs.input;
    const outputCost = (response.usage.output_tokens / 1_000_000) * costs.output;
    return inputCost + outputCost;
  }

  private checkBudget(estimatedCost: number): boolean {
    const today = new Date();
    if (today.toDateString() !== this.resetDate.toDateString()) {
      this.dailySpent = 0;
      this.resetDate = today;
    }

    return (this.dailySpent + estimatedCost) <= this.dailyBudget;
  }

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

// 使用例
async function main() {
  const client = new ProductionClaudeClient({ dailyBudgetYen: 10000 });

  const response = await client.createMessage({
    prompt: 'AIについて500文字で説明してください',
    systemPrompt: 'あなたは技術ライターです。わかりやすく説明してください。',
    maxTokens: 1024,
    model: 'claude-3-5-sonnet-20241022',
    temperature: 0.3
  });

  console.log(response.content[0].text);
}

main().catch(console.error);