LLM APIコスト90%削減の実践最適化戦略

シナリオ: 小規模B2B SaaS、デイリーアクティブユーザー5,000人

使用パターン:
- ユーザー1人 × 1日10回の会話
- 会話1回 = 入力200トークン + 出力300トークン

1日のトークン使用量:
5,000人 × 10回 × 500トークン = 25,000,000トークン/日

月換算:
25,000,000 × 30 = 750,000,000トークン/月

コスト比較（月額）:
GPT-4o ($2.50/1M入力 + $10/1M出力):
  → 入力$37,500 + 出力$45,000 = 月額$82,500

GPT-4o-mini ($0.15/1M入力 + $0.60/1M出力):
  → 入力$2,250 + 出力$2,700 = 月額$4,950

自社ホスティングLlama: 月額~$500-2,000

import anthropic

client = anthropic.Anthropic()

# すべてのリクエストで送信される長いシステムプロンプト
COMPANY_KNOWLEDGE_BASE = """
[数千トークンの会社文書、製品情報、ポリシーなど...]
...これをキャッシュなしで毎回送信するとコストが爆発する。
"""

def chat_with_caching(user_message: str) -> dict:
    response = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1024,
        system=[
            {
                "type": "text",
                "text": COMPANY_KNOWLEDGE_BASE,
                "cache_control": {"type": "ephemeral"}  # このブロックをキャッシュ！
            },
            {
                "type": "text",
                "text": "あなたはカスタマーサポートの専門家です。上記の会社情報に基づいて回答してください。"
                # こちらはキャッシュしない（短く、変更可能）
            }
        ],
        messages=[{"role": "user", "content": user_message}]
    )

    usage = response.usage
    print(f"キャッシュ書き込み: {usage.cache_creation_input_tokens} tokens (1.25倍コスト)")
    print(f"キャッシュ読み取り: {usage.cache_read_input_tokens} tokens (0.1倍コスト！90%割引)")
    print(f"通常入力: {usage.input_tokens} tokens (1倍コスト)")

    return {
        "content": response.content[0].text,
        "cache_hit": usage.cache_read_input_tokens > 0
    }

# 最初の呼び出し：キャッシュ作成（1.25倍コスト）
result1 = chat_with_caching("返金ポリシーはどうなっていますか？")

# 2回目以降：キャッシュヒット（0.1倍コスト！90%節約）
result2 = chat_with_caching("配送期間はどのくらいかかりますか？")
result3 = chat_with_caching("製品保証期間は？")

from openai import OpenAI
client = OpenAI()

# OpenAIは1,024トークン以上のプロンプトを自動的にキャッシュ
# 追加設定不要で自動的に50%割引が適用される
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {
            "role": "system",
            # 1,024トークン以上なら自動キャッシュ（50%割引）
            "content": LONG_SYSTEM_PROMPT  # 2,000トークン以上を推奨
        },
        {"role": "user", "content": "質問です"}
    ]
)

# キャッシュヒットの確認
usage = response.usage
if hasattr(usage, 'prompt_tokens_details'):
    cached = usage.prompt_tokens_details.cached_tokens
    total_input = usage.prompt_tokens
    savings_pct = (cached / total_input * 50) if total_input > 0 else 0
    print(f"キャッシュ済み: {cached}/{total_input} tokens ({savings_pct:.1f}%コスト削減)")

from openai import OpenAI
import re

client = OpenAI()

class ModelRouter:
    """リクエストの複雑度に応じて最適なモデルにルーティング"""

    SIMPLE_PATTERNS = [
        r"^(what is|what are|define|who is|when was|where is)",
        r"^(translate|how do you say|what does .* mean)",
        r"(yes or no|true or false)",
    ]

    COMPLEX_PATTERNS = [
        r"(analyze|compare|design|architect|evaluate)",
        r"(step.by.step|detailed|comprehensive|in.depth)",
        r"(implement|code|build|create a system|write a program)",
        r"(explain why|what causes|pros and cons|trade.?offs)",
        r"(分析|比較|設計|評価|詳細|実装|コード|理由|メリット|デメリット)",
    ]

    def classify(self, query: str) -> str:
        lower = query.lower()

        if any(re.search(p, lower) for p in self.COMPLEX_PATTERNS):
            return "complex"

        if (len(query.split()) < 20 and
                any(re.search(p, lower) for p in self.SIMPLE_PATTERNS)):
            return "simple"

        if len(query) > 500:
            return "complex"

        return "medium"

    def complete(self, query: str, system: str = "") -> dict:
        complexity = self.classify(query)

        messages = []
        if system:
            messages.append({"role": "system", "content": system})
        messages.append({"role": "user", "content": query})

        if complexity in ("simple", "medium"):
            # GPT-4o-miniはGPT-4oの約60分の1のコスト
            response = client.chat.completions.create(
                model="gpt-4o-mini",
                messages=messages,
                max_tokens=500 if complexity == "medium" else 200
            )
            model = "gpt-4o-mini"
        else:
            response = client.chat.completions.create(
                model="gpt-4o",
                messages=messages
            )
            model = "gpt-4o"

        return {
            "answer": response.choices[0].message.content,
            "model": model,
            "complexity": complexity
        }


# 実際のコスト影響:
# トラフィック分布: 60%シンプル、25%中程度、15%複雑と仮定
# すべてGPT-4oの場合:  100% × $12.50/1M avg = $12.50
# ルーティング適用後: 85% × $0.375 + 15% × $12.50 = $0.32 + $1.88 = $2.20
# → 約82%コスト削減！

router = ModelRouter()
examples = [
    "Transformerアーキテクチャとは何ですか？",    # シンプル → mini
    "分散レートリミットシステムを設計してください",  # 複雑 → GPT-4o
    "HNSWとはどういう意味ですか？",              # シンプル → mini
]

for q in examples:
    result = router.complete(q)
    print(f"[{result['complexity'].upper()}] → {result['model']}")

import hashlib
import numpy as np
from openai import OpenAI
from datetime import datetime, timedelta

client = OpenAI()

class SemanticCache:
    """
    意味的に類似したクエリをキャッシュ
    「RAGとは？」と「RAGについて説明してください」は同じキャッシュ結果を返す
    """

    def __init__(self, similarity_threshold: float = 0.92, ttl_hours: int = 24):
        self.cache: dict = {}
        self.threshold = similarity_threshold
        self.ttl = timedelta(hours=ttl_hours)
        self.stats = {"hits": 0, "misses": 0}

    def _embed(self, text: str) -> list:
        return client.embeddings.create(
            input=text,
            model="text-embedding-3-small",
            dimensions=256  # キャッシュ検索の高速化のために小さい次元を使用
        ).data[0].embedding

    def _similarity(self, a: list, b: list) -> float:
        a_np, b_np = np.array(a), np.array(b)
        return float(np.dot(a_np, b_np) / (np.linalg.norm(a_np) * np.linalg.norm(b_np)))

    def lookup(self, query: str) -> tuple:
        query_emb = self._embed(query)
        best_score, best_response = 0.0, None

        for entry in self.cache.values():
            if datetime.now() - entry["ts"] > self.ttl:
                continue
            score = self._similarity(query_emb, entry["emb"])
            if score > best_score:
                best_score, best_response = score, entry["response"]

        if best_score >= self.threshold:
            self.stats["hits"] += 1
            return best_response, best_score

        self.stats["misses"] += 1
        return None, best_score

    def store(self, query: str, response: str) -> None:
        key = hashlib.md5(query.encode()).hexdigest()
        self.cache[key] = {
            "emb": self._embed(query),
            "response": response,
            "ts": datetime.now()
        }

    def ask(self, query: str, model: str = "gpt-4o-mini") -> dict:
        cached, score = self.lookup(query)

        if cached:
            return {"response": cached, "source": "cache", "similarity": score}

        response = client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": query}]
        ).choices[0].message.content

        self.store(query, response)
        return {"response": response, "source": "api", "similarity": 0}

    @property
    def hit_rate(self) -> float:
        total = self.stats["hits"] + self.stats["misses"]
        return self.stats["hits"] / total if total else 0


# FAQのキャッシュを事前ウォームアップ
cache = SemanticCache(similarity_threshold=0.92)
faqs = [
    "返金ポリシーはどうなっていますか？",
    "配送にはどのくらいかかりますか？",
    "パスワードをリセットするにはどうすればいいですか？",
]
for q in faqs:
    cache.ask(q)  # 初回：API呼び出し後にキャッシュ

# 類似した質問はキャッシュからヒット
r1 = cache.ask("返品できますか？")          # 「返金ポリシー」に類似 → キャッシュ
r2 = cache.ask("到着まで何日かかりますか？") # 「配送」に類似 → キャッシュ
print(f"キャッシュヒット率: {cache.hit_rate:.1%}")  # FAQ重点の場合~70%

from openai import OpenAI
import json
import tempfile

client = OpenAI()

def submit_batch_job(texts: list, task_description: str) -> str:
    """
    大量のテキスト処理を50%割引で実行
    24時間以内に処理
    適したタスク：感情分析、分類、要約、翻訳
    """

    requests = [
        {
            "custom_id": f"item-{i}",
            "method": "POST",
            "url": "/v1/chat/completions",
            "body": {
                "model": "gpt-4o-mini",
                "messages": [
                    {"role": "system", "content": task_description},
                    {"role": "user", "content": text}
                ],
                "max_tokens": 100,
                "response_format": {"type": "json_object"}
            }
        }
        for i, text in enumerate(texts)
    ]

    with tempfile.NamedTemporaryFile(mode='w', suffix='.jsonl', delete=False) as f:
        for req in requests:
            f.write(json.dumps(req, ensure_ascii=False) + '\n')
        tmp_path = f.name

    with open(tmp_path, 'rb') as f:
        batch_file = client.files.create(file=f, purpose="batch")

    batch = client.batches.create(
        input_file_id=batch_file.id,
        endpoint="/v1/chat/completions",
        completion_window="24h"
    )

    print(f"バッチID: {batch.id}")
    print(f"コスト: リアルタイムAPIより50%削減")
    return batch.id


def get_batch_results(batch_id: str) -> list:
    batch = client.batches.retrieve(batch_id)

    if batch.status != "completed":
        print(f"処理中: {batch.status}")
        return []

    content = client.files.content(batch.output_file_id)
    results = []
    for line in content.text.strip().split('\n'):
        item = json.loads(line)
        body = item["response"]["body"]
        answer = body["choices"][0]["message"]["content"]
        results.append({
            "id": item["custom_id"],
            "result": json.loads(answer) if answer.startswith('{') else answer
        })

    return results


# 使用例：10,000件の顧客レビューを夜間に感情分析
reviews = ["本当に良い製品です！", "配送が遅すぎました", "品質に満足しています"] * 3334

batch_id = submit_batch_job(
    texts=reviews,
    task_description='感情を分類してください。JSONで回答: {"sentiment": "positive/negative/neutral", "confidence": 0.0-1.0}'
)
# 翌日実行
results = get_batch_results(batch_id)

from openai import OpenAI
from pydantic import BaseModel
from typing import Literal
import json

client = OpenAI()

# 悪い例：冗長な散文形式の応答
def analyze_verbose(text: str) -> str:
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": f"次のテキストの感情を分析してください: {text}"}]
    )
    # 例: 「このテキストは非常にポジティブな感情を示しています。著者が製品に満足していることが...」
    # 平均80〜200トークンの出力
    return response.choices[0].message.content


# 良い例：構造化された簡潔な出力
class SentimentResult(BaseModel):
    sentiment: Literal["positive", "negative", "neutral"]
    confidence: float
    key_phrase: str   # 最大5単語

def analyze_structured(text: str) -> SentimentResult:
    response = client.beta.chat.completions.parse(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "感情を分類してください。key_phraseは最大5単語。"},
            {"role": "user", "content": text}
        ],
        response_format=SentimentResult
    )
    # 例: {"sentiment": "positive", "confidence": 0.94, "key_phrase": "本当に良い製品"}
    # 平均20〜30トークンの出力 → 85%削減！
    return response.choices[0].message.parsed


# max_tokensによる出力長の明示的制限
def summarize_concisely(text: str, max_words: int = 50) -> str:
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": f"要約は必ず{max_words}単語以内で作成してください。前置きなし。"
            },
            {"role": "user", "content": f"要約してください: {text}"}
        ],
        max_tokens=max_words * 3
    )
    return response.choices[0].message.content

from collections import defaultdict
from datetime import date
from openai import OpenAI

client = OpenAI()

class CostTracker:
    """LLM APIコストのリアルタイム追跡"""

    PRICING = {
        "gpt-4o":           {"input": 2.50,  "output": 10.00},
        "gpt-4o-mini":      {"input": 0.15,  "output": 0.60},
        "claude-3-5-sonnet-20241022": {"input": 3.00, "output": 15.00},
        "claude-3-haiku-20240307":    {"input": 0.25, "output": 1.25},
    }

    def __init__(self, daily_budget: float = 100.0):
        self.daily_budget = daily_budget
        self.by_model = defaultdict(lambda: {"in": 0, "out": 0, "cost": 0.0})
        self.by_day = defaultdict(float)

    def record(self, model: str, input_tokens: int, output_tokens: int) -> float:
        p = self.PRICING.get(model, {"input": 0, "output": 0})
        cost = (input_tokens * p["input"] + output_tokens * p["output"]) / 1_000_000
        today = date.today().isoformat()

        self.by_model[model]["in"] += input_tokens
        self.by_model[model]["out"] += output_tokens
        self.by_model[model]["cost"] += cost
        self.by_day[today] += cost

        daily_so_far = self.by_day[today]
        if daily_so_far > self.daily_budget * 0.8:
            print(f"警告: 1日の予算の80%を使用済み (${daily_so_far:.2f} / ${self.daily_budget:.2f})")
        if daily_so_far > self.daily_budget:
            raise RuntimeError(f"1日の予算超過！${daily_so_far:.2f} > ${self.daily_budget:.2f}")

        return cost

    def report(self) -> None:
        total = sum(d["cost"] for d in self.by_model.values())
        today = date.today().isoformat()
        print(f"\n本日の支出: ${self.by_day[today]:.4f} / ${self.daily_budget:.2f} 予算")
        for model, d in sorted(self.by_model.items(), key=lambda x: -x[1]["cost"]):
            pct = d["cost"] / total * 100 if total else 0
            print(f"  {model}: ${d['cost']:.4f} ({pct:.1f}%)")


tracker = CostTracker(daily_budget=50.0)

def tracked_call(model: str, messages: list, **kwargs) -> str:
    response = client.chat.completions.create(model=model, messages=messages, **kwargs)
    tracker.record(model, response.usage.prompt_tokens, response.usage.completion_tokens)
    return response.choices[0].message.content