AIコストを78%削減する5つの実践手法

import anthropic

client = anthropic.Anthropic(
    api_key="sk-brievio-...",
    base_url="https://api.brievio.com",
)

# プロンプトの長く静的な部分をキャッシュ可能として印を付ける。同じプレフィックスを
# 持つ次の呼び出しは Anthropic のキャッシュにヒットする — コストは 10%。
resp = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": "<<your 20K-token system prompt — schema, rules, examples>>",
            "cache_control": {"type": "ephemeral"},   # キャッシュ可能として印を付ける
        },
    ],
    messages=[{"role": "user", "content": "Latest question…"}],
)

def pick_model(task_difficulty: int) -> str:
    """
    十分な品質を保てる最も安いモデルが勝つ。
    まず体感の難易度で選び、次に出力品質で検証する。
    """
    if task_difficulty <= 2:
        return "claude-haiku-4-5"     # Opus より約 10 倍安い
    if task_difficulty <= 4:
        return "gemini-2.5-flash"       # 非常に強力で、出力はさらに安い
    if task_difficulty <= 7:
        return "claude-sonnet-4-6"
    return "claude-opus-4-7"          # 本当に必要なときだけ

# 二層の上限: 呼び出しごとの max_tokens に加え、キーに財布レベルの厳格な上限を。
resp = client.chat.completions.create(
    model="claude-sonnet-4-6",
    messages=[...],
    max_tokens=800,       # 呼び出しごとに出力を制限
    stop=["</answer>"],   # 既知の終端で打ち切る
)

# Brievio のダッシュボードで: /app/keys → 各キーに「Spending limit」を設定。
# 本番キー: $50/日。実験キー: $5/日。CI キー: $1/日。

# N 個のタスクが独立しているなら、逐次的な推論にお金を払う必要はない。
import asyncio
from openai import AsyncOpenAI

client = AsyncOpenAI(
    api_key=os.environ["BRIEVIO_API_KEY"],
    base_url="https://api.brievio.com/v1",
)

async def classify_one(text: str) -> str:
    resp = await client.chat.completions.create(
        model="claude-haiku-4-5",     # 並列 + 安い = 勝ち
        messages=[{"role": "user", "content": f"Classify: {text}"}],
        max_tokens=20,
    )
    return resp.choices[0].message.content

results = await asyncio.gather(*(classify_one(t) for t in batch))

from openai import APIError, RateLimitError
import time, random

def call_with_backoff(client, **kwargs):
    for attempt in range(5):
        try:
            return client.chat.completions.create(**kwargs)
        except RateLimitError as e:
            # Retry-After があれば尊重する。
            ra = getattr(e, "retry_after", None) or (2 ** attempt + random.random())
            time.sleep(min(ra, 30))
        except APIError as e:
            if 500 <= e.status_code < 600:
                time.sleep(2 ** attempt + random.random())
                continue
            raise           # 4xx — 一時的なものではなく、あなたのバグ
    raise RuntimeError("exhausted retries")