💰 Cost-Aware Routing（成本感知路由）

任务输入
  ↓
[复杂度分类器] → 简单 | 中等 | 复杂
  ↓
[预算检查] → 足够 | 不足
  ↓
[路由决策] → 选择模型
  ↓
[执行 + 成本记录]
  ↓
[更新预算状态]

// OpenClaw 配置：定义模型池
{
  "models": {
    "cheap": {
      "provider": "openai",
      "model": "gpt-3.5-turbo",
      "costPer1MTokens": 1.5,
      "maxComplexity": "simple"
    },
    "mid": {
      "provider": "openai",
      "model": "gpt-4o-mini",
      "costPer1MTokens": 0.75,
      "maxComplexity": "medium"
    },
    "premium": {
      "provider": "anthropic",
      "model": "claude-3-opus",
      "costPer1MTokens": 90,
      "maxComplexity": "complex"
    },
    "local": {
      "provider": "ollama",
      "model": "llama3",
      "costPer1MTokens": 0,
      "maxComplexity": "medium"
    }
  }
}

// SKILL.md 中定义路由规则
## Model Routing Policy
- name: cost-aware-routing
  description: 根据任务复杂度和预算选择模型
  
  rules:
    - if: task.type == "simple_qa" && budget.remaining > 0
      then: use_model("cheap")
    
    - if: task.type == "code_generation"
      then: use_model("mid")
    
    - if: task.type == "complex_reasoning" && budget.remaining > 50
      then: use_model("premium")
    
    - if: budget.remaining < 10
      then: use_model("local")  # 降级到本地模型
    
    - else:
      then: use_model("mid")  # 默认

# Python示例：预算追踪器
class BudgetTracker:
    def __init__(self, daily_limit=100.0):
        self.daily_limit = daily_limit
        self.today_spent = 0.0
        self.history = []
    
    def estimate_cost(self, model, input_tokens, output_tokens):
        rate = MODEL_COSTS[model]
        return (input_tokens * rate['input'] + 
                output_tokens * rate['output']) / 1_000_000
    
    def can_afford(self, estimated_cost):
        return self.today_spent + estimated_cost <= self.daily_limit
    
    def record_usage(self, model, input_tokens, output_tokens):
        cost = self.estimate_cost(model, input_tokens, output_tokens)
        self.today_spent += cost
        self.history.append({
            'model': model,
            'cost': cost,
            'timestamp': time.time()
        })
        return cost

# 使用示例
tracker = BudgetTracker(daily_limit=50.0)  # 每天最多花50刀

if tracker.can_afford(estimated_cost=0.05):
    result = call_llm(model="gpt-4o-mini", prompt="...")
    actual_cost = tracker.record_usage("gpt-4o-mini", 500, 200)
    print(f"本次花费: ${actual_cost:.4f}，今日已花: ${tracker.today_spent:.2f}")
else:
    print("⚠️ 预算不足，切换到本地模型")
    result = call_llm(model="llama3", prompt="...")

// OpenClaw 中的含虾率+成本策略
{
  "routingStrategy": "cost-aware-shrimp",
  "metrics": {
    "shrimpRate": {  // 含虾率 = 正确完成率
      "weight": 0.7,
      "minThreshold": 0.85
    },
    "cost": {
      "weight": 0.3,
      "maxPerTask": 0.50  // 单任务最多花50美分
    }
  },
  "fallback": {
    "onLowShrimp": "upgrade_model",  // 含虾率低就升级模型
    "onBudgetExceeded": "use_local"   // 超预算用本地
  }
}

# Python完整示例：Cost-Aware Router
import re
from enum import Enum

class TaskComplexity(Enum):
    SIMPLE = "simple"
    MEDIUM = "medium"
    COMPLEX = "complex"

class CostAwareRouter:
    def __init__(self, budget_daily=100.0):
        self.budget = budget_daily
        self.spent = 0.0
        
        self.models = {
            "gpt-3.5": {"cost": 1.5, "max_complexity": TaskComplexity.SIMPLE},
            "gpt-4o-mini": {"cost": 0.75, "max_complexity": TaskComplexity.MEDIUM},
            "gpt-4-turbo": {"cost": 40.0, "max_complexity": TaskComplexity.COMPLEX},
            "local-llama3": {"cost": 0.0, "max_complexity": TaskComplexity.MEDIUM}
        }
    
    def classify_task(self, prompt: str) -> TaskComplexity:
        """简单规则分类任务复杂度"""
        if len(prompt) < 50 and "?" in prompt:
            return TaskComplexity.SIMPLE
        elif any(kw in prompt.lower() for kw in ["write", "generate", "code", "analyze"]):
            return TaskComplexity.COMPLEX
        else:
            return TaskComplexity.MEDIUM
    
    def select_model(self, prompt: str) -> str:
        """根据任务选择最合适的模型"""
        complexity = self.classify_task(prompt)
        
        # 按成本排序
        candidates = sorted(
            [(name, info) for name, info in self.models.items()],
            key=lambda x: x[1]['cost']
        )
        
        for name, info in candidates:
            if self._can_handle(info, complexity) and self._can_afford(info):
                return name
        
        # 都不行，用本地模型
        return "local-llama3"
    
    def _can_handle(self, model_info, complexity):
        return self._complexity_level(model_info['max_complexity']) >= self._complexity_level(complexity)
    
    def _can_afford(self, model_info):
        return self.spent + model_info['cost'] < self.budget
    
    def _complexity_level(self, c):
        return {"simple": 1, "medium": 2, "complex": 3}[c.value]

# 使用示例
router = CostAwareRouter(budget_daily=20.0)

prompts = [
    "今天天气怎么样？",
    "帮我写个Python爬虫，抓取新闻网站",
    "分析这份财报，给出投资建议（5000字）"
]

for prompt in prompts:
    model = router.select_model(prompt)
    print(f"任务: {prompt[:30]}...")
    print(f"  → 选择模型: {model}")
    print(f"  → 预估成本: ${router.models[model]['cost']:.2f}")
    print()