LiteLLMのfallbacks・Router・Prompt CachingでLLM運用を最適化してみる

Python 3.13
litellm 1.83.14
langchain-litellm 0.6.4

"""LiteLLM の fallbacks: 主モデルが失敗したら自動的に別モデルへ切替。"""

from litellm import completion

# 主: GPT-5.5、フォールバック先: GPT-5.4 → Sonnet 4.6
response = completion(
    model="openai/gpt-5.5",
    messages=[{"role": "user", "content": "AIエージェントを1文で説明して"}],
    fallbacks=["openai/gpt-5.4", "anthropic/claude-sonnet-4-6"],
    num_retries=2,  # リトライ回数（プロバイダーがレートリミット等で失敗したとき）
)

print(f"使用モデル: {response.model}")
print(f"応答: {response.choices[0].message.content}")

$ python fallback.py
使用モデル: gpt-5.5-2026-04-23
応答: AIエージェントとは、目的に応じて状況を判断し、必要な情報収集やツール操作を自律的に行ってタスクを遂行するAIシステムです。

"""LiteLLM の completion_cost で累計コストを追跡する。

各呼び出しのトークン使用量から正確なコストを算出する。
"""

from litellm import completion, completion_cost

queries = [
    "東京の天気は？",
    "AIエージェントを1段落で説明して",
    "Pythonでリスト内包表記を3例示して",
]

models = ["openai/gpt-5-mini", "openai/gpt-5.5", "anthropic/claude-sonnet-4-6"]

total_costs: dict[str, float] = {m: 0.0 for m in models}

for q in queries:
    print(f"\n[Q] {q}")
    for m in models:
        resp = completion(model=m, messages=[{"role": "user", "content": q}])
        cost = completion_cost(completion_response=resp)
        total_costs[m] += cost
        print(f"  {m}: ${cost:.6f}")

print("\n=== 累計コスト比較 ===")
for m, c in total_costs.items():
    print(f"  {m}: ${c:.6f}")

$ python cost_tracking.py

[Q] 東京の天気は？
  openai/gpt-5-mini: $0.001875
  openai/gpt-5.5: $0.004110
  anthropic/claude-sonnet-4-6: $0.002790

[Q] AIエージェントを1段落で説明して
  openai/gpt-5-mini: $0.000696
  openai/gpt-5.5: $0.005070
  anthropic/claude-sonnet-4-6: $0.003510

[Q] Pythonでリスト内包表記を3例示して
  openai/gpt-5-mini: $0.002013
  openai/gpt-5.5: $0.007115
  anthropic/claude-sonnet-4-6: $0.008139

=== 累計コスト比較 ===
  openai/gpt-5-mini: $0.004584
  openai/gpt-5.5: $0.016295
  anthropic/claude-sonnet-4-6: $0.014439

"""LiteLLM Router: 複数モデルへのロードバランシング・コストルーティング。

Router を使うと「複数のデプロイ済みモデルに重み付けで振り分ける」「タスクの
複雑度に応じてモデルを自動選択する」といった戦略が組める。
"""

from litellm import Router

# モデルプール: 同じ役割でも複数のデプロイメントを束ねる
model_list = [
    {
        "model_name": "fast",  # 軽量タスク用エイリアス
        "litellm_params": {"model": "openai/gpt-5-mini"},
    },
    {
        "model_name": "fast",  # フォールバック先
        "litellm_params": {"model": "openai/gpt-5.1"},
    },
    {
        "model_name": "smart",  # 高品質タスク用エイリアス
        "litellm_params": {"model": "anthropic/claude-sonnet-4-6"},
    },
]

router = Router(
    model_list=model_list,
    routing_strategy="simple-shuffle",  # シャッフル / least-busy / latency-based 等
    fallbacks=[{"smart": ["fast"]}],  # smart 失敗時に fast へフォールバック
)

# fast エイリアスで呼ぶと、Haiku か GPT-mini のいずれかに自動ルーティング
for _ in range(3):
    resp = router.completion(
        model="fast",
        messages=[{"role": "user", "content": "東京の天気を1文で"}],
    )
    print(f"使用モデル: {resp.model}, 応答: {resp.choices[0].message.content[:60]}")

# 重要なタスクは smart に
resp = router.completion(
    model="smart",
    messages=[{"role": "user", "content": "AIエージェントの定義を3行で"}],
)
print(f"\nsmart: {resp.model}")
print(resp.choices[0].message.content)

$ python router_routing.py
使用モデル: gpt-5.1-2025-11-13, 応答: 申し訳ありませんが、現在の天気はリアルタイムでは取得できません。ただ、通常この時期の東京は...
使用モデル: gpt-5-mini-2025-08-07, 応答: いつの東京の天気を一文で知りたいですか？...
使用モデル: gpt-5.1-2025-11-13, 応答: 現在の東京の天気情報はリアルタイムでは取得できませんが、最新の状況は気象庁や天気予報サイト...

smart: claude-sonnet-4-6
## AIエージェントの定義

自律的に**目標を設定・判断・行動**できるAIシステム。
環境を認識し、ツールや外部サービスを使いながら**タスクを自ら遂行**する。
人間の逐次的な指示なしに、**複数ステップの問題を自動で解決**する能力を持つ。

"""Anthropic Prompt Caching を LiteLLM 経由で利用する。

長い system プロンプトや大きなコンテキストを cache_control マーカー付きで
渡すと、Anthropicが内部キャッシュし、2回目以降の同一プレフィックス利用時に
コストを最大90%削減できる。
"""

from litellm import completion

# 長い system プロンプトをキャッシュ対象として指定
LONG_SYSTEM_PROMPT = (
    """\
あなたは社内ナレッジマスターです。以下の社内ルールを完全に把握して回答してください。

【就業規則】
- 勤務時間: 9:00-18:00
- フレックス: コアタイム 11:00-15:00
- 休憩: 12:00-13:00 が標準
- 残業: 事前申請必須
- 有給休暇: 入社6ヶ月後に10日付与
- リモート: 週3日まで可能

【経費規程】
- 月末締め、翌月15日払い
- 領収書必須
- 5万円以上は事前承認

【その他】
- 健康診断: 年1回、会社負担
- 育児休業: 子が1歳まで取得可能（2歳まで延長可）

上記ルールに基づき、簡潔に回答してください。
"""
    * 10
)  # 実運用ではここに数千〜数万トークンの本物のドキュメントが入る

def chat_with_cache(question: str) -> dict:
    response = completion(
        model="anthropic/claude-sonnet-4-6",
        messages=[
            {
                "role": "system",
                "content": [
                    {
                        "type": "text",
                        "text": LONG_SYSTEM_PROMPT,
                        # 重要: cache_control を指定してキャッシュ対象にする
                        "cache_control": {"type": "ephemeral"},
                    }
                ],
            },
            {"role": "user", "content": question},
        ],
    )
    usage = response.usage
    return {
        "answer": response.choices[0].message.content,
        "input_tokens": usage.prompt_tokens,
        "cache_creation": getattr(usage, "cache_creation_input_tokens", 0),
        "cache_read": getattr(usage, "cache_read_input_tokens", 0),
    }

# 1回目: キャッシュを作成
print("=== 1回目（キャッシュ作成） ===")
r1 = chat_with_cache("リモートワークは何日できる？")
print(f"  cache_creation: {r1['cache_creation']} tokens")
print(f"  cache_read: {r1['cache_read']} tokens")
print(f"  answer: {r1['answer'][:80]}")

# 2回目: キャッシュヒット
print("\n=== 2回目（キャッシュヒット） ===")
r2 = chat_with_cache("有給休暇の付与日数は？")
print(f"  cache_creation: {r2['cache_creation']} tokens")
print(f"  cache_read: {r2['cache_read']} tokens")
print(f"  answer: {r2['answer'][:80]}")

$ python prompt_caching.py
=== 1回目（キャッシュ作成） ===
  cache_creation: 2672 tokens
  cache_read: 0 tokens
  answer: ## リモートワークの日数

**週3日まで**可能です。

> 残りの週2日以上は出社が必要です。ご不明な点があればお気軽にどうぞ！

=== 2回目（キャッシュヒット） ===
  cache_creation: 0 tokens
  cache_read: 2672 tokens
  answer: ## 有給休暇の付与について

**入社6ヶ月後に10日**付与されます。

> ※それ以降の付与日数（勤続年数に応じた増加分）については、就業規則の詳細をご確