AI APIを本番環境に導入する際、想定外のコスト爆増は頭を悩ませる重大な問題です。私は以前、深夜にAPIコストが10倍に跳ね上がり翌日朝の請求額を見て凍りついた経験があります。本稿では、HolySheep AIを活用した成本異常告警システムの設計パターンと具体的な実装コードを解説します。
HolySheep vs 公式API vs 他のリレーサービスの比較
| 項目 | HolySheep AI | 公式OpenAI API | 他リレーサービス（平均） |
|---|---|---|---|
| 為替レート | ¥1=$1（85%節約） | ¥7.3=$1 | ¥3.5-5.5=$1 |
| GPT-4.1出力成本 | $8.00/MTok | $15.00/MTok | $10-12/MTok |
| Claude Sonnet 4.5出力成本 | $15.00/MTok | $18.00/MTok | $14-16/MTok |
| DeepSeek V3.2出力成本 | $0.42/MTok（最安） | $0.42/MTok | $0.45-0.55/MTok |
| レイテンシ | <50ms | 100-300ms | 80-200ms |
| 決済方法 | WeChat Pay / Alipay対応 | クレジットカードのみ | 限定的 |
| 免费クレジット | 登録時付与 | $5様試用（期限あり） | ほとんどなし |
| コスト異常告警 | カスタマイズ可能 | .basic告警のみ | 限定的 |
コスト異常告警が必要な理由
AI APIのコスト異常は様々な要因で発生します：
- ループリクエスト：再帰処理のバグで無限リクエスト
- コンテキスト爆増：長文プロンプトの誤送信
- モデル誤選択：高コストモデルの誤った呼び出し
- スケーリング問題：トラフィック急増への対応漏れ
HolySheep AIでは<50msのレイテンシと¥1=$1の為替レートで、大量リクエストも低コストで処理可能です。しかし、どんなに優れたサービスでも監視がなければコスト管理は不可能です。
成本異常告警アーキテクチャ設計
システム構成
+------------------+ +-------------------+ +------------------+
| HolyShehep API |---->| API Gateway/Proxy |---->| Alert Service |
| (api.holysheep | | (リクエスト仲介) | | (コスト監視) |
| .ai/v1) | +-------------------+ +------------------+
+------------------+ | |
v v
+------------------+ +------------------+
| 使用量DB/ログ |---->| 通知先(Slack/ |
| (CloudWatch等) | | Email/PagerDuty)|
+------------------+ +------------------+
コスト異常告警ルールの実装
Pythonによるリアルタイムコスト監視システム
#!/usr/bin/env python3
"""
AI API コスト異常告警システム
対応サービス: HolySheep AI (https://api.holysheep.ai/v1)
"""
import time
import json
import hashlib
import hmac
from datetime import datetime, timedelta
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Optional
import threading
@dataclass
class CostAlertRule:
"""コスト告警ルール定義"""
name: str
threshold_yen: float # 閾値（円）
window_minutes: int # 監視ウィンドウ（分）
model_weights: dict = field(default_factory=lambda: {
"gpt-4.1": 8.0, # $8.00/MTok -> ¥8
"claude-sonnet-4-5": 15.0, # $15.00/MTok -> ¥15
"gemini-2.5-flash": 2.5, # $2.50/MTok -> ¥2.5
"deepseek-v3.2": 0.42, # $0.42/MTok -> ¥0.42
})
@dataclass
class APIRequest:
"""APIリクエスト記録"""
timestamp: datetime
model: str
input_tokens: int
output_tokens: int
cost_yen: float
class HolySheepCostMonitor:
"""
HolySheep AI コスト監視クラス
私はこのクラスを使用して月間¥50,000の予算超過を検出し、
深夜のコスト爆増をリアルタイムで通知できるようにしました。
"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.request_log: list[APIRequest] = []
self.alert_rules: list[CostAlertRule] = []
self.lock = threading.Lock()
self.notifications: list[dict] = []
def add_alert_rule(self, rule: CostAlertRule):
"""告警ルールを追加"""
self.alert_rules.append(rule)
print(f"[INFO] 告警ルール追加: {rule.name} (閾値: ¥{rule.threshold_yen}/{rule.window_minutes}分)")
def record_request(self, model: str, input_tokens: int, output_tokens: int):
"""
リクエストを記録し、コストを計算
HolySheep AIの為替レート: ¥1 = $1
"""
cost_yen = self._calculate_cost(model, input_tokens, output_tokens)
request = APIRequest(
timestamp=datetime.now(),
model=model,
input_tokens=input_tokens,
output_tokens=output_tokens,
cost_yen=cost_yen
)
with self.lock:
self.request_log.append(request)
self._cleanup_old_requests()
# コストチェック
self._check_cost_alerts()
return cost_yen
def _calculate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
"""コスト計算（HolySheep AI汇率 ¥1=$1）"""
model_lower = model.lower()
# 入力コスト（通常は出力の1/10）
input_rate = 0.0
output_rate = 0.0
for key, rate in self.alert_rules[0].model_weights.items() if self.alert_rules else [("gpt-4.1", 8.0)]:
if key.replace("-", "") in model_lower.replace("-", ""):
output_rate = rate
input_rate = rate / 10
break
else:
# デフォルト: GPT-4.1
input_rate = 0.8 # $0.8/MTok
output_rate = 8.0 # $8.00/MTok
# 成本 = (入力トークン + 出力トークン) * 汇率
return (input_tokens * input_rate + output_tokens * output_rate) / 1_000_000
def _cleanup_old_requests(self):
"""1時間以上前のリクエストを削除"""
cutoff = datetime.now() - timedelta(hours=1)
self.request_log = [r for r in self.request_log if r.timestamp > cutoff]
def _check_cost_alerts(self):
"""全告警ルールをチェック"""
now = datetime.now()
for rule in self.alert_rules:
window_start = now - timedelta(minutes=rule.window_minutes)
with self.lock:
recent_requests = [
r for r in self.request_log
if r.timestamp >= window_start
]
total_cost = sum(r.cost_yen for r in recent_requests)
if total_cost >= rule.threshold_yen:
self._trigger_alert(rule, total_cost, len(recent_requests))
def _trigger_alert(self, rule: CostAlertRule, total_cost: float, request_count: int):
"""告警トリガー（重複防止付き）"""
alert_id = hashlib.md5(
f"{rule.name}{datetime.now().strftime('%Y%m%d%H%M')}".encode()
).hexdigest()[:8]
# 重複チェック（5分間隔）
for notif in self.notifications[-10:]:
if (notif.get("rule_name") == rule.name and
notif.get("minute") == datetime.now().strftime('%Y%m%d%H%M')):
return
alert = {
"alert_id": alert_id,
"rule_name": rule.name,
"total_cost_yen": round(total_cost, 2),
"threshold_yen": rule.threshold_yen,
"request_count": request_count,
"timestamp": now.isoformat(),
"minute": datetime.now().strftime('%Y%m%d%H%M'),
"severity": "CRITICAL" if total_cost > rule.threshold_yen * 1.5 else "WARNING"
}
self.notifications.append(alert)
self._send_notification(alert)
def _send_notification(self, alert: dict):
"""通知送信（Slack/Email/Webhook対応）"""
severity_emoji = "🔴" if alert["severity"] == "CRITICAL" else "🟡"
message = f"""
{severity_emoji} AI API コスト異常告警
ルール: {alert['rule_name']}
コスト: ¥{alert['total_cost_yen']:.2f} / 閾値 ¥{alert['threshold_yen']}
リクエスト数: {alert['request_count']}
時刻: {alert['timestamp']}
告警ID: {alert['alert_id']}
"""
print(f"\n{'='*50}\n{ message }\n{'='*50}\n")
# 実際の通知連携はここに実装
# self._send_slack(message)
# self._send_email(message)
def get_cost_summary(self, minutes: int = 60) -> dict:
"""コストサマリー取得"""
cutoff = datetime.now() - timedelta(minutes=minutes)
with self.lock:
recent = [r for r in self.request_log if r.timestamp >= cutoff]
if not recent:
return {"total_cost": 0, "request_count": 0, "models": {}}
model_costs = defaultdict(lambda: {"cost": 0, "count": 0})
for r in recent:
model_costs[r.model]["cost"] += r.cost_yen
model_costs[r.model]["count"] += 1
return {
"total_cost": round(sum(r.cost_yen for r in recent), 2),
"request_count": len(recent),
"models": {k: {"cost": round(v["cost"], 2), "count": v["count"]}
for k, v in model_costs.items()},
"period_minutes": minutes
}
===== 使用例 =====
if __name__ == "__main__":
# HolySheep AI API キー
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
# コストモニター初期化
monitor = HolySheepCostMonitor(API_KEY)
# 告警ルール設定
monitor.add_alert_rule(CostAlertRule(
name="5分窓高コスト",
threshold_yen=100.0, # 5分間で¥100超え
window_minutes=5
))
monitor.add_alert_rule(CostAlertRule(
name="1時間窓予算超過",
threshold_yen=1000.0, # 1時間で¥1,000超え
window_minutes=60
))
monitor.add_alert_rule(CostAlertRule(
name="Claude高コスト警告",
threshold_yen=50.0, # Claude系で5分¥50超え
window_minutes=5,
model_weights={"claude-sonnet-4-5": 15.0}
))
# テストリクエスト模擬
print("[TEST] コスト監視テスト開始\n")
# GPT-4.1 リクエスト（入力: 1000トークン, 出力: 500トークン）
cost1 = monitor.record_request("gpt-4.1", 1000, 500)
print(f"GPT-4.1 リクエスト -> コスト: ¥{cost1:.4f}")
# DeepSeek V3.2 リクエスト
cost2 = monitor.record_request("deepseek-v3.2", 5000, 2000)
print(f"DeepSeek V3.2 リクエスト -> コスト: ¥{cost2:.4f}")
# Gemini 2.5 Flash リクエスト
cost3 = monitor.record_request("gemini-2.5-flash", 2000, 1000)
print(f"Gemini 2.5 Flash リクエスト -> コスト: ¥{cost3:.4f}")
# コストサマリー表示
summary = monitor.get_cost_summary(minutes=60)
print(f"\n[サマリー] 総コスト: ¥{summary['total_cost']:.2f}, "
f"リクエスト数: {summary['request_count']}")
Proxy層でのコスト制御と割込み
#!/usr/bin/env python3
"""
HolySheep AI API Proxy with Cost Control
ベースURL: https://api.holysheep.ai/v1
"""
import json
import time
import asyncio
import httpx
from typing import Optional, Any
from datetime import datetime, timedelta
from dataclasses import dataclass
@dataclass
class RateLimitConfig:
"""レート制限設定"""
max_requests_per_minute: int = 60
max_cost_per_hour_yen: float = 500.0
max_cost_per_day_yen: float = 5000.0
burst_allowance: int = 10
@dataclass
class CostBudget:
"""コスト予算"""
hourly_spent: float = 0.0
daily_spent: float = 0.0
hourly_reset: datetime = None
daily_reset: datetime = None
def __post_init__(self):
self.hourly_reset = datetime.now() + timedelta(hours=1)
self.daily_reset = datetime.now() + timedelta(days=1)
class HolySheepProxy:
"""
HolySheep AI API プロキシー
私はこのプロキシを使用して、API呼び出しごとに成本計算と
予算チェックを行い、超過前にリクエストをブロックします。
メリット:
- ¥1=$1汇率で85%コスト削減
- <50msレイテンシで高速応答
- WeChat Pay/Alipay対応
"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str, budget: Optional[CostBudget] = None,
rate_limit: Optional[RateLimitConfig] = None):
self.api_key = api_key
self.budget = budget or CostBudget()
self.rate_limit = rate_limit or RateLimitConfig()
self.request_count = 0
self.last_minute_requests = []
# HolySheep AI 対応モデルと汇率
self.model_costs = {
"gpt-4.1": {"input_per_1m": 0.8, "output_per_1m": 8.0},
"claude-sonnet-4-5": {"input_per_1m": 1.5, "output_per_1m": 15.0},
"gemini-2.5-flash": {"input_per_1m": 0.25, "output_per_1m": 2.5},
"deepseek-v3.2": {"input_per_1m": 0.042, "output_per_1m": 0.42},
}
def _estimate_cost(self, model: str, messages: list) -> float:
"""コスト見積もり"""
# 簡易計算: メッセージ長さをトークン数として概算
total_chars = sum(len(str(m.get("content", ""))) for m in messages)
estimated_input_tokens = total_chars // 4 # 1トークン≈4文字
estimated_output_tokens = estimated_input_tokens // 5
costs = self.model_costs.get(model, self.model_costs["gpt-4.1"])
cost = (estimated_input_tokens * costs["input_per_1m"] +
estimated_output_tokens * costs["output_per_1m"]) / 1_000_000
return cost
def _check_budget(self, estimated_cost: float) -> tuple[bool, str]:
"""予算チェック"""
now = datetime.now()
# 毎時リセットチェック
if now >= self.budget.hourly_reset:
self.budget.hourly_spent = 0
self.budget.hourly_reset = now + timedelta(hours=1)
# 日次リセットチェック
if now >= self.budget.daily_reset:
self.budget.daily_spent = 0
self.budget.daily_reset = now + timedelta(days=1)
# コスト超過チェック
if self.budget.hourly_spent + estimated_cost > self.rate_limit.max_cost_per_hour_yen:
return False, f"毎時予算超過: ¥{self.budget.hourly_spent:.2f} + ¥{estimated_cost:.2f} > ¥{self.rate_limit.max_cost_per_hour_yen}"
if self.budget.daily_spent + estimated_cost > self.rate_limit.max_cost_per_day_yen:
return False, f"日次予算超過: ¥{self.budget.daily_spent:.2f} + ¥{estimated_cost:.2f} > ¥{self.rate_limit.max_cost_per_day_yen}"
return True, "OK"
def _check_rate_limit(self) -> tuple[bool, str]:
"""レート制限チェック"""
now = datetime.now()
one_minute_ago = now - timedelta(minutes=1)
# 1分以内のリクエストをフィルタ
self.last_minute_requests = [
t for t in self.last_minute_requests if t > one_minute_ago
]
if len(self.last_minute_requests) >= self.rate_limit.max_requests_per_minute:
return False, f"レート制限超過: {len(self.last_minute_requests)}/min"
self.last_minute_requests.append(now)
return True, "OK"
async def chat_completions(self, model: str, messages: list,
max_cost_yen: float = 10.0) -> dict:
"""
Chat Completions API呼び出し（成本管理付き）
Args:
model: モデル名 (gpt-4.1, claude-sonnet-4-5, etc.)
messages: メッセージリスト
max_cost_yen: このリクエストの最大許容コスト
Returns:
API応答またはエラー情報
"""
# コスト見積もり
estimated_cost = self._estimate_cost(model, messages)
if estimated_cost > max_cost_yen:
return {
"error": True,
"type": "cost_limit_exceeded",
"message": f"推定コスト ¥{estimated_cost:.2f} > 最大 ¥{max_cost_yen}",
"estimated_cost": estimated_cost
}
# 予算チェック
budget_ok, budget_msg = self._check_budget(estimated_cost)
if not budget_ok:
return {
"error": True,
"type": "budget_exceeded",
"message": budget_msg,
"hourly_spent": self.budget.hourly_spent,
"daily_spent": self.budget.daily_spent
}
# レート制限チェック
rate_ok, rate_msg = self._check_rate_limit()
if not rate_ok:
return {
"error": True,
"type": "rate_limit_exceeded",
"message": rate_msg
}
# HolySheep AI API呼び出し
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"max_tokens": 2048
}
start_time = time.time()
try:
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
f"{self.BASE_URL}/chat/completions",
headers=headers,
json=payload
)
response.raise_for_status()
result = response.json()
elapsed_ms = (time.time() - start_time) * 1000
# 実際のコスト更新
actual_cost = estimated_cost # 概算値を使用
self.budget.hourly_spent += actual_cost
self.budget.daily_spent += actual_cost
# 応答にコスト情報追加
result["_cost_info"] = {
"estimated_cost_yen": round(actual_cost, 4),
"hourly_spent_yen": round(self.budget.hourly_spent, 2),
"daily_spent_yen": round(self.budget.daily_spent, 2),
"latency_ms": round(elapsed_ms, 2)
}
# 高コスト時に警告ログ
if actual_cost > max_cost_yen * 0.8:
print(f"[警告] コスト注意: ¥{actual_cost:.4f} / 最大¥{max_cost_yen}")
return result
except httpx.HTTPStatusError as e:
return {
"error": True,
"type": "api_error",
"status_code": e.response.status_code,
"message": str(e)
}
except Exception as e:
return {
"error": True,
"type": "unknown_error",
"message": str(e)
}
===== 使用例 =====
async def main():
"""使用例"""
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
# プロキシ初期化（毎時¥500、日次¥5000制限）
proxy = HolySheepProxy(
api_key=API_KEY,
rate_limit=RateLimitConfig(
max_requests_per_minute=30,
max_cost_per_hour_yen=500.0,
max_cost_per_day_yen=5000.0
)
)
# DeepSeek V3.2で低成本テスト（$0.42/MTok）
result = await proxy.chat_completions(
model="deepseek-v3.2",
messages=[
{"role": "user", "content": "日本の四季について教えてください"}
],
max_cost_yen=1.0 # 最大¥1まで許可
)
if result.get("error"):
print(f"[エラー] {result['message']}")
else:
print(f"[成功] コスト: ¥{result['_cost_info']['estimated_cost_yen']:.4f}")
print(f" レイテンシ: {result['_cost_info']['latency_ms']:.2f}ms")
print(f" 日次累計: ¥{result['_cost_info']['daily_spent_yen']:.2f}")
if __name__ == "__main__":
asyncio.run(main())
ダッシュボード設定（CloudWatchアラーム統合）
AWS CloudWatchとHolySheep AIを統合した監視ダッシュボードの設定例：
# CloudFormation テンプレート断片 - 成本監視アラーム
AWSTemplateFormatVersion: '2010-09-09'
Resources:
# 日次コスト合計アラーム
DailyCostAlarm:
Type: AWS::CloudWatch::Alarm
Properties:
AlarmName: holy-sheep-daily-cost-exceeded
AlarmDescription: HolySheep AI 日次コストが¥5000超え
MetricName: EstimatedCharges
Namespace: AWS/Billing
Statistic: Maximum
Period: 86400
EvaluationPeriods: 1
Threshold: 5000
ComparisonOperator: GreaterThanThreshold
TreatMissingData: notBreaching
# カスタムメトリクス用 IAM Role
CostMonitorRole:
Type: AWS::IAM::Role
Properties:
AssumeRolePolicyDocument:
Version: '2012-10-17'
Statement:
- Effect: Allow
Principal:
Service: lambda.amazonaws.com
Action: sts:AssumeRole
Policies:
- PolicyName: cost-monitor-policy
PolicyDocument:
Version: '2012-10-17'
Statement:
- Effect: Allow
Action:
- cloudwatch:PutMetricData
- logs:CreateLogGroup
- logs:CreateLogStream
- logs:PutLogEvents
Resource: '*'
HolySheep AIの推荐監視設定
| 監視項目 | 警告閾値 | 重大閾値 | 対応アクション |
|---|---|---|---|
| 毎時コスト | ¥300 | ¥500 | Slack通知 → リクエスト制限 |
| 日次コスト | ¥3,000 | ¥5,000 | Email通知 → 全リクエスト停止 |
| 単一リクエストコスト | ¥5 | ¥10 | Slack通知 → モデル降格 |
| リクエスト数/分 | 50 | 80 | レートリミット適用 |
| レイテンシ | >200ms | >500ms | 健全性チェック → 代替サービス切替 |
よくあるエラーと対処法
エラー1: コスト計算の不一致
現象: 実際の請求額と計算値が大きく異なる
# 問題のあるコード
def calculate_cost_broken(model, tokens):
# 単純なトークン数 × 基本汇率で計算
return tokens * 0.0001 # 精度不足
修正後のコード
def calculate_cost_fixed(model, input_tokens, output_tokens):
"""
HolySheep AI 汇率: ¥1 = $1
各モデルの正確な汇率を適用
"""
model_rates = {
"gpt-4.1": {"input": 0.8, "output": 8.0}, # $0.8/$8 per 1M
"claude-sonnet-4-5": {"input": 1.5, "output": 15.0}, # $1.5/$15 per 1M
"gemini-2.5-flash": {"input": 0.25, "output": 2.5}, # $0.25/$2.5 per 1M
"deepseek-v3.2": {"input": 0.042, "output": 0.42}, # $0.042/$0.42 per 1M
}
rates = model_rates.get(model, model_rates["gpt-4.1"])
# コスト = (トークン数 / 1,000,000) × 汇率
input_cost = (input_tokens / 1_000_000) * rates["input"]
output_cost = (output_tokens / 1_000_000) * rates["output"]
return input_cost + output_cost # ¥単位で返す
検証
cost = calculate_cost_fixed("deepseek-v3.2", 100000, 50000)
print(f"DeepSeek V3.2 コスト: ¥{cost:.4f}") # ¥0.063
エラー2: 無限リクエストループによるコスト爆増
現象: APIコストが数分で数万円に跳ね上がる
# 問題のあるコード
def process_user_request(user_input):
while True:
response = call_api(user_input)
if response.is_complete():
return response
# 完了判定のバグで無限ループ！
修正後のコード（リクエスト上限付き）
def process_user_request_fixed(user_input, max_retries=3, max_cost_yen=5.0):
"""
リトライ回数とコスト上限をを設定
"""
total_cost = 0.0
for attempt in range(max_retries):
# コスト予測
estimated_cost = estimate_cost(user_input)
if total_cost + estimated_cost > max_cost_yen:
raise CostLimitExceeded(
f"コスト上限超過: ¥{total_cost:.2f} + ¥{estimated_cost:.2f} > ¥{max_cost_yen}"
)
response = call_api(user_input)
actual_cost = response.get("cost", estimated_cost)
total_cost += actual_cost
if response.is_complete():
return response
# 次ループ前に必ず待機（レート制限対策）
time.sleep(1)
raise MaxRetriesExceeded(f"{max_retries}回リトライしても完了せず")
追加: circuit breakerパターン
class CircuitBreaker:
def __init__(self, failure_threshold=5, timeout_seconds=60):
self.failure_count = 0
self.failure_threshold = failure_threshold
self.timeout = timeout_seconds
self.last_failure_time = None
self.state = "closed" # closed, open, half-open
def call(self, func):
if self.state == "open":
if time.time() - self.last_failure_time > self.timeout:
self.state = "half-open"
else:
raise CircuitOpenError("サーキットブレーカーが開いています")
try:
result = func()
self.on_success()
return result
except Exception as e:
self.on_failure()
raise
def on_success(self):
self.failure_count = 0
self.state = "closed"
def on_failure(self):
self.failure_count += 1
self.last_failure_time = time.time()
if self.failure_count >= self.failure_threshold:
self.state = "open"
エラー3: モデル選択ミスによる高コスト
現象: 簡単なクエリに高コストモデルを使用してしまう
# 問題のあるコード
def handle_request(message):
# 常にGPT-4.1を使用（高コスト）
return call_api("gpt-4.1", message)
修正後のコード（タスク別モデル選択）
MODEL_SELECTION_RULES = {
"simple_qa": {
"models": ["deepseek-v3.2", "gemini-2.5-flash"],
"max_cost_yen": 0.5,
"description": "単純質問"
},
"code_generation": {
"models": ["deepseek-v3.2", "gpt-4.1"],
"max_cost_yen": 2.0,
"description": "コード生成"
},
"complex_reasoning": {
"models": ["gpt-4.1", "claude-sonnet-4-5"],
"max_cost_yen": 10.0,
"description": "複雑な推論"
}
}
def classify_task(message: str) -> str:
"""タスク分類"""
message_lower = message.lower()
if any(kw in message_lower for kw in ["なぜ", "どうして", "複雑な"]):
return "complex_reasoning"
elif any(kw in message_lower for kw in ["コード", "関数", "python"]):
return "code_generation"
else:
return "simple_qa"
def handle_request_fixed(message: str, budget_yen: float = 1.0):
"""
タスクに応じて適切なモデルを選択
HolySheep AI: ¥1=$1汇率で低成本運用
"""
task_type = classify_task(message)
rule = MODEL_SELECTION_RULES[task_type]
# 予算内で使用可能なモデルを選択
available_models = [
m for m in rule["models"]
if MODEL_COSTS[m] <= budget_yen
]
if not available_models:
# 予算内で最も安いモデルを選択
model = min(rule["models"], key=lambda m: MODEL_COSTS[m])
print(f"[警告] 予算超出: {task_type} -> {model}")
else:
model = available_models[0]
# DeepSeek V3.2 ($0.42/MTok) を優先的に使用
if "deepseek-v