最初の本番API呼び出し(コスト意識)

中級

What you'll learn

• 本番呼び出しをおもちゃの一行と分ける4つの規律に名前を付ける:シークレット、ストリーミング、コスト、エラー処理
• 一時的な失敗(429/5xx)をバックオフでリトライする回復力のあるストリーミング呼び出しを書く — そして400は決してリトライしない
• モデルIDを設定に保持し、モデル切り替えが検索置換ではなく1行の変更で済むようにする
• すべての呼び出しでトークンコストを監視し、意図的に上限を設ける

おもちゃのAPI呼び出しは1行です。本番呼び出しは、エラーを処理し、出力をストリームし、コストを監視し、シークレットを安全に保ちます。段階的に構築していきましょう。

ステップ1 — 設定からシークレットとモデル

export ANTHROPIC_API_KEY="sk-ant-..."   # never in source control

モデルIDを設定に保持し、リテラルを散在させないでください、そうすれば移行が容易になります(なぜ)。意図的に選ぶ — モデルの選び方。

ステップ2 — 回復力のあるストリーミング呼び出し

Python

import os, time, random, anthropic
client = anthropic.Anthropic()
MODEL = os.environ.get("CLAUDE_MODEL", "claude-sonnet-5")

def ask_stream(prompt, system=None, max_tokens=1024):
    for attempt in range(5):
        try:
            with client.messages.stream(
                model=MODEL, max_tokens=max_tokens,
                system=system or anthropic.NOT_GIVEN,
                messages=[{"role": "user", "content": prompt}],
            ) as stream:
                for text in stream.text_stream:
                    print(text, end="", flush=True)
                final = stream.get_final_message()
            print()
            usage = final.usage
            print(f"\n[tokens in/out: {usage.input_tokens}/{usage.output_tokens}]")
            return final
        except (anthropic.RateLimitError, anthropic.APIStatusError):
            if attempt == 4: raise
            time.sleep(min(2 ** attempt + random.random(), 30))

TypeScript

import Anthropic from "@anthropic-ai/sdk";
const client = new Anthropic();
const MODEL = process.env.CLAUDE_MODEL ?? "claude-sonnet-5";

export async function askStream(prompt: string, system?: string, maxTokens = 1024) {
  for (let attempt = 0; attempt < 5; attempt++) {
    try {
      const stream = client.messages.stream({ model: MODEL, max_tokens: maxTokens, system,
        messages: [{ role: "user", content: prompt }] });
      for await (const e of stream)
        if (e.type === "content_block_delta") process.stdout.write(e.delta.text ?? "");
      const final = await stream.finalMessage();
      console.error(`\n[tokens in/out: ${final.usage.input_tokens}/${final.usage.output_tokens}]`);
      return final;
    } catch (e: any) {
      if (attempt === 4 || ![429, 500, 529].includes(e?.status)) throw e;
      await new Promise(r => setTimeout(r, Math.min(2 ** attempt * 1000, 30000)));
    }
  }
}

何かに配線する前にスモークテストしてください — 1つの小さな呼び出しが、ストリーム、トークン行、そしてキーがすべて動くことを証明します:

最初のスモークテスト(Python REPL)

ask_stream("Say hello in one sentence.", max_tokens=64)

ステップ3 — コストに気をつける

• トークン使用量をログする(上記)、各呼び出しのコストが見えるように。
• max_tokensとモデルを適切なサイズに、集中的なプロンプトで入力を制限する。
• 繰り返される安定したプレフィックスには、プロンプトキャッシュを追加。
• トークンと価格とコストとレイテンシを参照。

ステップ4 — 不幸な経路を処理する

• 一時的なエラー(429/5xx)はバックオフでリトライ(上記);400はリトライしない。
• 拒否を優雅に処理する。
• エージェント的なものにはタイムアウトとコスト/イテレーション予算を設定する。

検証

各経路を強制して何が起きるか観察してください — 本番呼び出しはただ成功するだけでなく、うまく失敗することでその名に値します:

Guided walkthrough1 of 4

1. テキストが最後に一括ブロックではなく生成されるにつれ段階的に印字される。それがユーザーが感じるレイテンシの勝利。

理解度チェック

0/3

1. 1コード全体にリテラル文字列を散在させるのではなく、なぜモデルIDを設定に保持するのですか?
2. 2リトライループは一部の失敗をキャッチし、他は再スローします。どのエラーをリトライすべきですか?
3. 3各呼び出しが実際にいくらかかっているか見る最も安い方法は?

Key takeaways

• 本番呼び出しは一行に重ねられた4つの規律:シークレットをソースの外に保つ、出力をストリーム、コストを監視、エラーを処理
• 一時的な失敗(429/5xx)を指数バックオフとジッタでリトライ — そして400は決してリトライしない、それはあなたが修正しなければならないリクエストのバグ
• モデルIDを設定に保持し、モデル切り替えが1行、コードベース全体の検索置換ではないようにする
• すべての呼び出しでトークン使用量をログ:リクエストごとのコストメーターで、追加コストはゼロ
• ストリーミングは知覚されるレイテンシを改善;タイムアウトとコスト/イテレーション予算がエージェント的ループの暴走を防ぐ

次のステップ

• エラー、レート制限と信頼性
• 本番APIスニペット
• APIでのエージェント構築