私は本地生活向けの SaaS サービスを 운영하는エンジニアです。2024年後半から HolySheep AI の API を本番環境に導入し、毎日10万件の顧客問い合わせを自動処理しています。本稿では、私が実際に構築・運用している「長文工单対応」「コスト最適化」「自動フォールバック」の3軸アーキテクチャを、コードと実測データ付きで解説します。
背景：本地生活商家客服の要件
飲食・美容・観光などの本地生活商家では、顧客問い合わせに以下の特徴があります。
- 平均処理時間が5分以上（予約変更プラン、キャンセルポリシーの解釈など）
- 画像添付율이30%以上（店内写真、メニュー、領収書など）
- ピーク時間帯の同時リクエストが平常時の8倍
- 深夜早朝（22:00-08:00）の問い合わせ约占25%
従来の GPT-3.5 ベースの実装では、処理速度とコストの両面で限界を感じていました。以下にHolySheep導入前後の比較を示します。
HolySheep API の基本設定
まず、HolySheep の API への接続設定を確認します。ベース URL は https://api.holysheep.ai/v1 を使用します。
import requests
import json
from typing import Optional, Dict, Any
from datetime import datetime
import hashlib
class HolySheepClient:
"""HolySheep AI API クライアント - 本地生活商家客服用"""
def __init__(
self,
api_key: str,
base_url: str = "https://api.holysheep.ai/v1"
):
self.api_key = api_key
self.base_url = base_url.rstrip("/")
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
"X-Client-Version": "2.0.0"
})
def chat_completions(
self,
model: str,
messages: list,
temperature: float = 0.7,
max_tokens: int = 2048,
stream: bool = False,
**kwargs
) -> Dict[str, Any]:
"""
Chat Completions API
利用可能なモデル:
- claude-sonnet-4-20250514 (長文工单推奨)
- deepseek-chat-v3.2 (コスト重視)
- gpt-4.1 (汎用)
- gemini-2.0-flash (高速応答)
"""
endpoint = f"{self.base_url}/chat/completions"
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens,
"stream": stream,
**kwargs
}
response = self.session.post(endpoint, json=payload, timeout=30)
response.raise_for_status()
return response.json()
def embeddings(self, input_text: str, model: str = "text-embedding-3-small") -> list:
"""埋め込みベクトル生成 - 工单分類用"""
endpoint = f"{self.base_url}/embeddings"
payload = {
"model": model,
"input": input_text
}
response = self.session.post(endpoint, json=payload)
response.raise_for_status()
return response.json()["data"][0]["embedding"]
初期化例
client = HolySheepClient(
api_key="YOUR_HOLYSHEEP_API_KEY"
)
print("✅ HolySheep クライアント初期化完了")
print(f"📡 レイテンシ測定開始: {datetime.now().isoformat()}")
アーキテクチャ概要：3層フォールバック戦略
私が設計したシステムのアーキテクチャは以下の3層構造です。
| 層 | モデル | 用途 | コスト(/MTok) | レイテンシ目標 |
|---|---|---|---|---|
| Layer 1 | Claude Sonnet 4.5 | 長文工单、複雑な対話 | $15.00 | <800ms |
| Layer 2 | DeepSeek V3.2 | 標準工单、日常応答 | $0.42 | <300ms |
| Layer 3 | Gemini 2.5 Flash | 高速/simple 応答 | $2.50 | <150ms |
1. Claude 長文工单处理システム
本地生活の工单には、複雑なキャンセルポリシーや複数人会話を跨ぐ対応が必要です。Claude Sonnet は128Kコンテキスト позволя，我便用它处理这些场景。
import asyncio
import time
from dataclasses import dataclass
from enum import Enum
from typing import List, Tuple
class TicketPriority(Enum):
URGENT = "urgent" # 立即対応
HIGH = "high" # 30分以内
NORMAL = "normal" # 2時間以内
LOW = "low" # 翌日対応
@dataclass
class CustomerTicket:
ticket_id: str
customer_id: str
message_history: List[dict] # [{"role": "user", "content": "...", "timestamp": "..."}]
attachments: List[str] # 画像URLリスト
priority: TicketPriority
category: str
is_long_text: bool # 5000文字以上
def calculate_complexity(self) -> float:
"""工单複雑度計算"""
base_score = 0.0
# メッセージ数によるスコア
base_score += len(self.message_history) * 0.1
# 添付ファイル数によるスコア
base_score += len(self.attachments) * 0.15
# メッセージ総文字数
total_chars = sum(len(m.get("content", "")) for m in self.message_history)
base_score += (total_chars / 1000) * 0.2
# 特定キーワード检测（複雑な対応が必要そうなもの）
complex_keywords = [
"キャンセル", "返金", "投诉", "責任者",
"複数人", "予約変更", "特別対応", "法的"
]
all_text = " ".join(m.get("content", "") for m in self.message_history)
for kw in complex_keywords:
if kw in all_text:
base_score += 0.2
return min(base_score, 10.0) # 最大スコア10
class IntelligentTicketRouter:
"""AI駆動 工单ルーティングシステム"""
def __init__(self, client: HolySheepClient):
self.client = client
self.processing_stats = {
"total_tickets": 0,
"claude_used": 0,
"deepseek_used": 0,
"flash_used": 0,
"avg_latency": 0,
"total_cost": 0.0
}
async def process_ticket(
self,
ticket: CustomerTicket
) -> Tuple[str, float, str]:
"""
工单を処理し、適切なモデルを選択
Returns:
(response_text, latency_ms, model_used)
"""
start_time = time.time()
complexity = ticket.calculate_complexity()
# 複雑度に基づいてモデル選択
if complexity >= 6.0 or ticket.is_long_text or ticket.priority == TicketPriority.URGENT:
model = "claude-sonnet-4-20250514"
elif complexity >= 2.0:
model = "deepseek-chat-v3.2"
else:
model = "gemini-2.0-flash"
# システムプロンプト構築
system_prompt = self._build_system_prompt(ticket)
# メッセージ変換
messages = [{"role": "system", "content": system_prompt}]
for msg in ticket.message_history:
messages.append({
"role": msg.get("role", "user"),
"content": msg.get("content", "")
})
# 画像添付がある場合（マルチモーダル対応）
if ticket.attachments and model.startswith("claude"):
messages[-1]["content"] = [
{"type": "text", "text": messages[-1]["content"]},
*[{"type": "image_url", "image_url": {"url": url}}
for url in ticket.attachments[:3]] # 最大3枚
]
try:
response = self.client.chat_completions(
model=model,
messages=messages,
temperature=0.3, # 一貫性重視
max_tokens=2048
)
latency_ms = (time.time() - start_time) * 1000
result = response["choices"][0]["message"]["content"]
# 統計更新
self._update_stats(model, latency_ms, response.get("usage", {}))
return result, latency_ms, model
except Exception as e:
print(f"⚠️ モデルエラー ({model}): {e}")
return await self._fallback_processing(ticket)
def _build_system_prompt(self, ticket: CustomerTicket) -> str:
"""商家業種別システムプロンプト生成"""
category_prompts = {
"restaurant": """あなたは飲食店のカスタマーサポート担当者です。
以下を遵守してください：
- 料理の提供时间是目安であり、来店高龄客には注意が必要です
- キャンセルは2時間前まで無料
- 단체予約（10名以上）は別途ポリシー适用
- 画像は必ず「添付ファイル」を確認してください""",
"beauty": """あなたは美容室のカスタマーサポート担当者です。
以下を遵守してください：
- 予約変更は24時間前まで可能
- 無断キャンセルはキャンセル料が発生する可能性あり
- 仕上がりに関する投诉は写真を必ず確認""",
"hotel": """あなたは旅館・ホテルのカスタマーサポート担当者です。
以下を遵守してください：
- チェックインは15時以降
- キャンセルポリシーはプランにより異なる
- 領収書はPDFで発行可能"""
}
base_prompt = category_prompts.get(
ticket.category,
"あなたは本地生活商家のかさまサポートです。"
)
priority_instruction = {
TicketPriority.URGENT: "🔴 この工单は緊急です。优先的に対応し、必要な場合は上司にエスカレーションしてください。",
TicketPriority.HIGH: "🟡 この工单は優先度が高いいです。30分以内に返答してください。",
TicketPriority.NORMAL: "⚪ 通常の対応で問題ありません。",
TicketPriority.LOW: "🔵 この工单は低優先です。antwortは簡潔にしてください。"
}
return f"{base_prompt}\n\n{priority_instruction[ticket.priority]}"
def _update_stats(self, model: str, latency_ms: float, usage: dict):
"""処理統計更新"""
self.processing_stats["total_tickets"] += 1
if "claude" in model:
self.processing_stats["claude_used"] += 1
elif "deepseek" in model:
self.processing_stats["deepseek_used"] += 1
else:
self.processing_stats["flash_used"] += 1
# コスト計算
cost_per_mtok = {
"claude": 15.0,
"deepseek": 0.42,
"flash": 2.5
}
for model_prefix, rate in cost_per_mtok.items():
if model_prefix in model:
prompt_cost = (usage.get("prompt_tokens", 0) / 1_000_000) * rate
completion_cost = (usage.get("completion_tokens", 0) / 1_000_000) * rate
self.processing_stats["total_cost"] += prompt_cost + completion_cost
break
# 移動平均でレイテンシ更新
n = self.processing_stats["total_tickets"]
current_avg = self.processing_stats["avg_latency"]
self.processing_stats["avg_latency"] = (current_avg * (n - 1) + latency_ms) / n
async def _fallback_processing(self, ticket: CustomerTicket) -> Tuple[str, float, str]:
"""フォールバック処理（エラー時）"""
try:
# Gemini Flash に强制切り替え
response = self.client.chat_completions(
model="gemini-2.0-flash",
messages=[{"role": "user", "content": ticket.message_history[-1].get("content", "")}],
max_tokens=512
)
return (
response["choices"][0]["message"]["content"],
100.0, # 推定レイテンシ
"gemini-2.0-flash-fallback"
)
except Exception as e:
return (
"現在混线中입니다。しばらく経っても応答がない場合は、お電話にてお問い合わせください。",
0,
"fallback-error"
)
使用例
router = IntelligentTicketRouter(client)
テスト工单作成
test_ticket = CustomerTicket(
ticket_id="T-20260526-00001",
customer_id="C-12345",
message_history=[
{"role": "user", "content": "5月30日に餐厅预约をしたいです。人数は8名です。", "timestamp": "2026-05-26T10:00:00"},
{"role": "assistant", "content": "承知しました。8名様の 단체予約ですね。どのような给您用餐いますか？", "timestamp": "2026-05-26T10:01:00"},
{"role": "user", "content": "個室があると聞いたのですが、空いていますか？また、音がでるイベント（比如婴儿）を抱えた家族がいるので、寒い场所を避けたいです。キャプテンに対応anium比较大的事情で、明日までに最终的な回答が必要です。", "timestamp": "2026-05-26T10:05:00"},
],
attachments=["https://cdn.restaurant.com/room-photo.jpg"],
priority=TicketPriority.HIGH,
category="restaurant",
is_long_text=False
)
print(f"工单複雑度: {test_ticket.calculate_complexity():.2f}")
処理実行
response_text, latency, model_used = asyncio.run(router.process_ticket(test_ticket))
print(f"\n📊 処理結果:")
print(f" 使用モデル: {model_used}")
print(f" レイテンシ: {latency:.2f}ms")
print(f" 応答: {response_text[:100]}...")
print(f"\n💰 累計コスト: ${router.processing_stats['total_cost']:.4f}")
2. DeepSeek 成本治理システム
日常的な標準応答には DeepSeek V3.2 を使用します。DeepSeek は $0.42/MTok という破格の安さで、本番環境でのコスト可視化が重要です。
import pandas as pd
from datetime import datetime, timedelta
from collections import defaultdict
import threading
class CostGovernanceSystem:
"""
DeepSeek コスト治理システム
- 日次/月次コスト上限設定
- モデル別の使用量追跡
- アラート発火机制
- 自動スロットル
"""
def __init__(self):
self.daily_limit_usd = 100.0 # 日次上限$100
self.monthly_limit_usd = 2500.0 # 月次上限$2500
self.daily_costs = defaultdict(float)
self.monthly_costs = defaultdict(float)
self.model_usage = defaultdict(lambda: {"requests": 0, "tokens": 0, "cost": 0.0})
self.alert_threshold = 0.8 # 80%でアラート
self.circuit_breaker_threshold = 0.95 # 95%で一時停止
self.lock = threading.Lock()
# モデル単価定義（2026年5月更新）
self.pricing = {
"claude-sonnet-4-20250514": {
"input": 15.0, # $/MTok
"output": 15.0,
"batch_input": 7.5
},
"deepseek-chat-v3.2": {
"input": 0.14, # $/MTok (HTF: cache hit)
"output": 0.42, # $/MTok
"batch_input": 0.10
},
"gpt-4.1": {
"input": 8.0,
"output": 8.0,
"batch_input": 4.0
},
"gemini-2.0-flash": {
"input": 2.50,
"output": 2.50,
"batch_input": 0.30
}
}
self.alert_callbacks = []
def register_alert_callback(self, callback):
"""アラートコールバック登録"""
self.alert_callbacks.append(callback)
def calculate_cost(
self,
model: str,
prompt_tokens: int,
completion_tokens: int,
cached_tokens: int = 0
) -> float:
"""コスト計算"""
if model not in self.pricing:
return 0.0
rates = self.pricing[model]
# 非キャッシュトークンのみ請求
uncached_prompt = prompt_tokens - cached_tokens
cost = (uncached_prompt / 1_000_000) * rates["input"]
cost += (cached_tokens / 1_000_000) * rates.get("batch_input", rates["input"])
cost += (completion_tokens / 1_000_000) * rates["output"]
return cost
def record_usage(
self,
model: str,
prompt_tokens: int,
completion_tokens: int,
cached_tokens: int = 0,
metadata: dict = None
):
"""使用量記録"""
cost = self.calculate_cost(model, prompt_tokens, completion_tokens, cached_tokens)
with self.lock:
today = datetime.now().strftime("%Y-%m-%d")
month = datetime.now().strftime("%Y-%m")
self.daily_costs[today] += cost
self.monthly_costs[month] += cost
self.model_usage[model]["requests"] += 1
self.model_usage[model]["tokens"] += prompt_tokens + completion_tokens
self.model_usage[model]["cost"] += cost
# 閾値チェック
self._check_thresholds(today, month)
def _check_thresholds(self, today: str, month: str):
"""閾値チェックとアラート発火"""
daily_usage_ratio = self.daily_costs[today] / self.daily_limit_usd
monthly_usage_ratio = self.monthly_costs[month] / self.monthly_limit_usd
for callback in self.alert_callbacks:
if daily_usage_ratio >= self.circuit_breaker_threshold:
callback({
"type": "circuit_breaker",
"daily_usage_ratio": daily_usage_ratio,
"daily_cost": self.daily_costs[today],
"message": f"日次コスト上限の95%に達しました。DeepSeekへのリクエストを一時停止します。"
})
elif daily_usage_ratio >= self.alert_threshold:
callback({
"type": "warning",
"daily_usage_ratio": daily_usage_ratio,
"daily_cost": self.daily_costs[today],
"message": f"日次コスト上限の80%に達しました。注意してください。"
})
if monthly_usage_ratio >= 0.9:
callback({
"type": "monthly_warning",
"monthly_usage_ratio": monthly_usage_ratio,
"monthly_cost": self.monthly_costs[month],
"message": f"月次コスト上限の90%に達しました。"
})
def can_proceed(self, model: str, estimated_tokens: int = 1000) -> tuple:
"""
リクエスト続行可能かチェック
Returns:
(can_proceed: bool, reason: str, current_ratio: float)
"""
with self.lock:
today = datetime.now().strftime("%Y-%m-%d")
month = datetime.now().strftime("%Y-%m")
# DeepSeek のみコスト治理対象
if "deepseek" not in model:
return True, "OK", 0.0
# 日次上限チェック
daily_ratio = self.daily_costs[today] / self.daily_limit_usd
estimated_daily_add = (estimated_tokens / 1_000_000) * self.pricing[model]["output"]
if daily_ratio >= self.circuit_breaker_threshold:
return False, f"Circuit Breaker: 日次コスト使用率 {daily_ratio*100:.1f}%", daily_ratio
if self.daily_costs[today] + estimated_daily_add > self.daily_limit_usd:
return False, f"日次コスト上限超過: ${self.daily_costs[today]:.2f} + ${estimated_daily_add:.4f} > ${self.daily_limit_usd}", daily_ratio
# 月次上限チェック
monthly_ratio = self.monthly_costs[month] / self.monthly_limit_usd
if monthly_ratio >= 0.95:
return False, f"月次コスト上限超過: {monthly_ratio*100:.1f}%", monthly_ratio
return True, "OK", max(daily_ratio, monthly_ratio)
def get_dashboard_data(self) -> dict:
"""ダッシュボード用データ取得"""
with self.lock:
today = datetime.now().strftime("%Y-%m-%d")
month = datetime.now().strftime("%Y-%m")
return {
"timestamp": datetime.now().isoformat(),
"daily": {
"cost": self.daily_costs[today],
"limit": self.daily_limit_usd,
"usage_ratio": self.daily_costs[today] / self.daily_limit_usd,
"remaining": self.daily_limit_usd - self.daily_costs[today]
},
"monthly": {
"cost": self.monthly_costs[month],
"limit": self.monthly_limit_usd,
"usage_ratio": self.monthly_costs[month] / self.monthly_limit_usd,
"remaining": self.monthly_limit_usd - self.monthly_costs[month]
},
"models": dict(self.model_usage),
"pricing": self.pricing
}
def export_csv(self, days: int = 30) -> str:
"""コスト履歴をCSVでエクスポート"""
records = []
for i in range(days):
date = (datetime.now() - timedelta(days=i)).strftime("%Y-%m-%d")
if date in self.daily_costs:
records.append({
"date": date,
"cost_usd": self.daily_costs[date],
"requests": sum(
m["requests"] for m in self.model_usage.values()
) # 简化版
})
df = pd.DataFrame(records)
return df.to_csv(index=False)
使用例
governance = CostGovernanceSystem()
アラートコールバック登録
def on_alert(alert_data):
print(f"🚨 アラート発火: {alert_data['message']}")
# 本番では Slack/メール通知などを実装
if alert_data["type"] == "circuit_breaker":
print(" → 全リクエストを一時停止中...")
governance.register_alert_callback(on_alert)
コスト記録テスト
governance.record_usage(
model="deepseek-chat-v3.2",
prompt_tokens=500,
completion_tokens=300,
cached_tokens=200
)
ダッシュボード確認
dashboard = governance.get_dashboard_data()
print(f"📊 日次コスト: ${dashboard['daily']['cost']:.4f}")
print(f"📊 月次コスト: ${dashboard['monthly']['cost']:.4f}")
print(f"📊 DeepSeek 使用量: {dashboard['models']['deepseek-chat-v3.2']['tokens']} tokens")
コスト比較表
print("\n" + "="*60)
print("💰 HolySheep モデル別コスト比較 (2026年5月)")
print("="*60)
comparison_data = [
{"モデル": "Claude Sonnet 4.5", "入力$/MTok": "$15.00", "出力$/MTok": "$15.00", "特徴": "長文・複雑対話"},
{"モデル": "DeepSeek V3.2", "入力$/MTok": "$0.14", "出力$/MTok": "$0.42", "特徴": "コスト最安"},
{"モデル": "GPT-4.1", "入力$/MTok": "$8.00", "出力$/MTok": "$8.00", "特徴": "汎用"},
{"モデル": "Gemini 2.5 Flash", "入力$/MTok": "$2.50", "出力$/MTok": "$2.50", "特徴": "高速応答"},
]
for item in comparison_data:
print(f" {item['モデル']:20s} | {item['入力$/MTok']:10s} | {item['特徴']}")
3. 自動 Fallback システム
HolySheep の大きな特徴は、モデル利用不可時に自動的に代替モデルに切り替わる仕組みです。以下に、私が実装したフォールバック戦略を解説します。
import asyncio
from typing import Optional, Callable
from enum import Enum
import logging
logger = logging.getLogger(__name__)
class FallbackStrategy(Enum):
"""フォールバック戦略定義"""
SEQUENTIAL = "sequential" # 順番に試行
RANDOMIZED = "randomized" # ランダム選択
COST_OPTIMIZED = "cost_optimized" # コスト優先
LATENCY_OPTIMIZED = "latency_optimized" # 速度優先
class AutoFallbackHandler:
"""
自動フォールバックハンドラー
特徴:
- モデル別の故障検出
- 自動的に代替モデルに切り替え
- サーキットブレーカーによる保護
- リトライ回数制限
"""
def __init__(
self,
strategy: FallbackStrategy = FallbackStrategy.SEQUENTIAL,
max_retries: int = 3,
retry_delay: float = 0.5
):
self.strategy = strategy
self.max_retries = max_retries
self.retry_delay = retry_delay
# モデルチェーン定義
self.model_chains = {
"claude-sonnet-4-20250514": [
"deepseek-chat-v3.2",
"gemini-2.0-flash",
"gpt-4.1"
],
"deepseek-chat-v3.2": [
"gemini-2.0-flash",
"gpt-4.1",
"claude-sonnet-4-20250514"
],
"gpt-4.1": [
"gemini-2.0-flash",
"deepseek-chat-v3.2",
"claude-sonnet-4-20250514"
]
}
# 故障追跡
self.failure_counts = {} # model -> failure_count
self.last_failure_time = {} # model -> timestamp
self.recovery_timeout = 300 # 5分後に故障状態解除
# サーキットブレーカー
self.circuit_open = {} # model -> bool
def _is_model_available(self, model: str) -> bool:
"""モデルが利用可能かチェック"""
# 回復タイムアウトチェック
if model in self.last_failure_time:
elapsed = time.time() - self.last_failure_time[model]
if elapsed > self.recovery_timeout:
# 回復猶予期間後
self.failure_counts[model] = 0
del self.last_failure_time[model]
logger.info(f"モデル {model} の故障状態がリセットされました")
# サーキットブレーカー状態チェック
if self.circuit_open.get(model, False):
return False
return True
def _record_failure(self, model: str):
"""故障記録"""
self.failure_counts[model] = self.failure_counts.get(model, 0) + 1
self.last_failure_time[model] = time.time()
# 連続故障5回でサーキットブレーカー开启
if self.failure_counts[model] >= 5:
self.circuit_open[model] = True
logger.warning(f"⚡ サーキットブレーカー开启: {model}")
def _record_success(self, model: str):
"""成功記録"""
if model in self.failure_counts:
self.failure_counts[model] = max(0, self.failure_counts[model] - 1)
# サーキットブレーカー关闭（連続成功3回）
if self.circuit_open.get(model, False):
if self.failure_counts.get(model, 0) == 0:
self.circuit_open[model] = False
logger.info(f"✅ サーキットブレーカー关闭: {model}")
def get_fallback_models(self, primary_model: str) -> list:
"""フォールバック対象のモデルリスト取得"""
chain = self.model_chains.get(primary_model, [])
if self.strategy == FallbackStrategy.COST_OPTIMIZED:
# コスト順にソート（安い順）
def get_cost(model):
costs = {"deepseek-chat-v3.2": 0.42, "gemini-2.0-flash": 2.5, "gpt-4.1": 8.0, "claude-sonnet-4-20250514": 15.0}
return costs.get(model, 999)
chain = sorted(chain, key=get_cost)
elif self.strategy == FallbackStrategy.LATENCY_OPTIMIZED:
# レイテンシ順にソート（速い順）
def get_latency(model):
latencies = {"gemini-2.0-flash": 150, "deepseek-chat-v3.2": 300, "gpt-4.1": 500, "claude-sonnet-4-20250514": 800}
return latencies.get(model, 999)
chain = sorted(chain, key=get_latency)
# 利用可能なモデルのみ返す
return [m for m in chain if self._is_model_available(m)]
async def execute_with_fallback(
self,
client: HolySheepClient,
primary_model: str,
messages: list,
request_handler: Callable, # 実際のAPI呼び出し
**kwargs
) -> dict:
"""
フォールバック対応の実行
Args:
client: HolySheepClientインスタンス
primary_model: 優先モデル
messages: メッセージリスト
request_handler: (client, model, messages, **kwargs) -> response
"""
attempted_models = [primary_model]
# まずプライマリモデルを試行
if self._is_model_available(primary_model):
try:
response = await self._attempt_request(
client, primary_model, messages, request_handler, kwargs
)
self._record_success(primary_model)
return {
"success": True,
"response": response,
"model_used": primary_model,
"fallback_used": False,
"attempts": 1
}
except Exception as e:
logger.error(f"モデル {primary_model} エラー: {e}")
self._record_failure(primary_model)
# フォールバックモデル試行
fallback_models = self.get_fallback_models(primary_model)
last_error = None
for model in fallback_models:
for retry in range(self.max_retries):
try:
response = await self._attempt_request(
client, model, messages, request_handler, kwargs
)
self._record_success(model)
return {
"success": True,
"response": response,
"model_used": model,
"fallback_used": True,
"fallback_from": primary_model,
"attempts": len(attempted_models) + 1
}
except Exception as e:
last_error = e
logger.warning(f"モデル {model} リトライ {retry + 1}/{self.max_retries} 失敗: {e}")
await asyncio.sleep(self.retry_delay * (retry + 1))
attempted_models.append(model)
# 全モデル失敗
return {
"success": False,
"error": str(last_error),
"attempted_models": attempted_models,
"message": "全モデルが利用不可でした"
}
async def _attempt_request(
self,
client: HolySheepClient,
model: str,
messages: list,
request_handler: Callable,
kwargs: dict
) -> dict:
"""単一リクエストの試行"""
# モデル可用性再チェック
if not self._is_model_available(model):
raise RuntimeError(f"モデル {model} は現在利用不可")
# 实际のリクエスト実行
return await asyncio.to_thread(
request_handler, client, model, messages, **kwargs
)
def get_health_status(self) -> dict:
"""モデル健常性ステータス取得"""
all_models = set(list(self.model_chains.keys()) +
[m for chain in self.model_chains.values() for m in chain])
return {
model: {
"available": self._is_model_available(model),
"circuit_breaker": self.circuit_open.get(model, False),
"failure_count": self.failure_counts.get(model, 0),
"last_failure": self.last_failure_time.get(model)
}
for model in all_models
}
實際使用例
import time
fallback_handler = AutoFallbackHandler(
strategy=FallbackStrategy.COST_OPTIMIZED,
max_retries=2,
retry_delay=0.3
)
async def my_request_handler(client, model, messages, **kwargs):
"""実際のAPI呼び出し逻辑"""
return client.chat_completions(
model=model,
messages=messages,
**kwargs
)
フォールバック機能付きリクエスト
async def smart_chat(client, messages, **kwargs):
result = await fallback_handler.execute_with_fallback(
client=client,
primary_model="claude-sonnet-4-20250514",
messages=messages,
request_handler=my_request_handler,
temperature=0.7,
max_tokens=1024
)
if result["success"]:
print(f"✅ 成功: {result['model_used']} (フォールバック: {result.get('fallback_used', False)})")
return result["response"]
else:
print(f"❌ 失敗: