多模型路由系统：任务难度、风险等级与成本预算如何决策

┌──────────────┐
│   任务输入    │
│  (Issue/命令) │
└──────┬───────┘
       │
┌──────▼───────────────────────────────────────────┐
│              任务分类器 (Task Classifier)          │
│  ┌─────────┐  ┌──────────┐  ┌──────────────────┐ │
│  │ 难度评估 │  │ 风险评估  │  │ 上下文大小估算   │ │
│  └────┬────┘  └────┬─────┘  └────────┬─────────┘ │
│       └─────────────┼────────────────┘            │
│                     │                             │
│              ┌──────▼──────┐                      │
│              │  路由决策引擎 │                      │
│              │  (Rule-based │                      │
│              │   + ML)      │                      │
│              └──────┬──────┘                      │
└─────────────────────┼────────────────────────────┘
                      │
         ┌────────────┼────────────┐
         │            │            │
    ┌────▼────┐  ┌────▼────┐  ┌───▼─────┐
    │ Haiku   │  │ Sonnet  │  │ Opus    │
    │ (简单)  │  │ (中等)  │  │ (复杂)  │
    └────┬────┘  └────┬────┘  └───┬─────┘
         │            │            │
         └────────────┼────────────┘
                      │
               ┌──────▼──────┐
               │  质量守卫    │
               │  (失败升级)  │
               └─────────────┘

# model-routing.yaml
version: "2.0"

# 模型定义
models:
  haiku:
    id: "claude-haiku-4-5-20251001"
    cost_per_1k_tokens: 0.001
    speed: "fast"          # 响应速度
    max_context: 200000
    strengths: ["formatting", "simple_edits", "test_generation"]
    
  sonnet:
    id: "claude-sonnet-4-6"
    cost_per_1k_tokens: 0.003
    speed: "medium"
    max_context: 200000
    strengths: ["bugfix", "code_review", "refactoring"]
    
  opus:
    id: "claude-opus-4-8"
    cost_per_1k_tokens: 0.015
    speed: "slow"
    max_context: 200000
    strengths: ["architecture", "complex_debugging", "large_refactor"]

# 路由规则（按优先级排序，匹配第一条即停止）
routing_rules:
  # 强制规则（优先级最高）
  - name: "production_deploy"
    condition: "task.risk_level == 'CRITICAL'"
    model: "opus"
    reason: "生产部署相关任务强制使用最强模型"
  
  - name: "security_fix"
    condition: "task.labels contains 'security'"
    model: "opus"
    reason: "安全修复必须用最强模型"
  
  - name: "budget_guard"
    condition: "daily_budget_remaining < daily_budget * 0.2"
    model: "haiku"
    reason: "预算低于 20% 时降级到最便宜模型"
  
  # 基于难度的规则
  - name: "trivial_edit"
    condition: |
      task.type in ['formatting', 'lint_fix', 'rename']
      AND task.files_changed <= 1
      AND task.estimated_lines <= 5
    model: "haiku"
    reason: "单文件少量修改，简单任务"
  
  - name: "standard_bugfix"
    condition: |
      task.type == 'bugfix'
      AND task.complexity == 'low'
      AND task.files_changed <= 3
    model: "sonnet"
    reason: "标准 Bug 修复"
  
  - name: "complex_bugfix"
    condition: |
      task.type == 'bugfix'
      AND (task.complexity == 'high' OR task.files_changed > 5)
    model: "opus"
    reason: "复杂 Bug 修复需要更强的推理能力"
  
  - name: "code_review"
    condition: "task.type == 'code_review'"
    model: "sonnet"
    reason: "代码审查 Sonnet 够用"
  
  - name: "test_generation"
    condition: "task.type == 'test_generation'"
    model: "sonnet"
    reason: "测试生成 Sonnet 质量好且性价比高"
  
  - name: "large_refactor"
    condition: |
      task.type == 'refactor'
      AND task.files_changed > 10
    model: "opus"
    reason: "大规模重构需要全局理解能力"
  
  # 默认规则
  - name: "default"
    condition: "true"
    model: "sonnet"
    reason: "默认使用 Sonnet（性价比最优）"

# Fallback 策略
fallback:
  on_error: "upgrade"       # 出错时升级到更强的模型
  max_upgrade: "opus"       # 最多升级到 Opus
  max_retries: 2            # 最多重试 2 次
  on_budget_exceeded: "downgrade"  # 超预算时降级

# app/routing/router.py
"""
多模型路由决策引擎。
根据任务属性和路由规则选择最合适的模型。
"""
import yaml
from dataclasses import dataclass
from typing import Optional

@dataclass
class TaskAttributes:
    """任务属性（路由决策的输入）"""
    task_id: str
    task_type: str              # bugfix / review / refactor / test / format / ...
    risk_level: str             # LOW / MEDIUM / HIGH / CRITICAL
    complexity: str             # low / medium / high
    labels: list[str]
    files_changed: int
    estimated_lines: int
    context_tokens: int
    approval_required: bool

@dataclass
class RoutingDecision:
    """路由决策结果"""
    model: str
    model_id: str
    reason: str
    matched_rule: str
    fallback_model: Optional[str] = None
    estimated_cost: float = 0.0

class ModelRouter:
    def __init__(self, config_path: str):
        with open(config_path) as f:
            self.config = yaml.safe_load(f)
        self.models = {m["id"]: m for m in self.config["models"].values()}
        self.rules = self.config["routing_rules"]
    
    def route(self, task: TaskAttributes) -> RoutingDecision:
        """为任务选择模型"""
        
        # 按优先级遍历规则
        for rule in self.rules:
            if self._evaluate_condition(rule["condition"], task):
                model_name = rule["model"]
                model_config = self.config["models"][model_name]
                
                # 检查预算
                if not self._check_budget(model_config, task):
                    return self._apply_fallback(task, "budget_exceeded")
                
                estimated_cost = self._estimate_cost(model_config, task)
                
                return RoutingDecision(
                    model=model_name,
                    model_id=model_config["id"],
                    reason=rule["reason"],
                    matched_rule=rule["name"],
                    fallback_model=self._get_fallback(model_name),
                    estimated_cost=estimated_cost,
                )
        
        # 不应该到达这里（default 规则兜底）
        return self._apply_fallback(task, "no_rule_matched")
    
    def _evaluate_condition(self, condition: str, task: TaskAttributes) -> bool:
        """评估规则条件（简化版，实际用 CEL 或 Python eval）"""
        # 安全评估：只允许访问 task 的属性
        context = {"task": task}
        try:
            # 实际生产环境应该用安全的表达式引擎（如 CEL）
            # 这里用简单的字符串匹配演示
            if "contains" in condition:
                # task.labels contains 'security'
                parts = condition.split("contains")
                labels = getattr(task, parts[0].strip().split(".")[-1])
                value = parts[1].strip().strip("'\"")
                return value in labels
            
            if "==" in condition:
                left, right = condition.split("==")
                left = left.strip()
                right = right.strip().strip("'\"")
                if left.startswith("task."):
                    attr = left.split(".")[-1]
                    return str(getattr(task, attr)) == right
                return False
            
            if condition == "true":
                return True
            
            return False
        except Exception:
            return False
    
    def _check_budget(self, model_config: dict, task: TaskAttributes) -> bool:
        """检查预算是否充足"""
        estimated = model_config["cost_per_1k_tokens"] * task.context_tokens / 1000
        remaining = get_daily_budget_remaining()
        return estimated <= remaining
    
    def _estimate_cost(self, model_config: dict, task: TaskAttributes) -> float:
        return model_config["cost_per_1k_tokens"] * task.context_tokens / 1000
    
    def _get_fallback(self, model_name: str) -> Optional[str]:
        fallback_chain = {"haiku": None, "sonnet": "haiku", "opus": "sonnet"}
        return fallback_chain.get(model_name)
    
    def _apply_fallback(self, task: TaskAttributes, reason: str) -> RoutingDecision:
        return RoutingDecision(
            model="haiku",
            model_id=self.config["models"]["haiku"]["id"],
            reason=f"Fallback: {reason}",
            matched_rule="fallback",
            estimated_cost=0,
        )

# app/routing/quality_guard.py
"""
质量守卫：如果低模型失败，自动升级到更强的模型。
"""

async def execute_with_guard(task: TaskAttributes, router: ModelRouter):
    """带质量守卫的执行流程"""
    
    decision = router.route(task)
    current_model = decision.model
    retry_count = 0
    max_retries = 2
    
    while retry_count <= max_retries:
        # 执行任务
        result = await execute_with_model(task, current_model)
        
        # 质量检查
        if result.success and result.quality_score >= 0.7:
            log_routing_decision(task, current_model, decision, "success")
            return result
        
        # 失败处理
        retry_count += 1
        
        if retry_count > max_retries:
            log_routing_decision(task, current_model, decision, "max_retries_exceeded")
            return result  # 返回最后的结果，标记为需要人工介入
        
        # 升级模型
        next_model = router._get_fallback(current_model)
        if next_model:
            log.info(f"Task {task.task_id}: {current_model} failed, upgrading to {next_model}")
            current_model = next_model
        else:
            log.warning(f"Task {task.task_id}: already at strongest model, retrying")
    
    return result

# app/routing/budget.py
"""
成本预算守卫：防止单日/单任务超出预算。
"""
from datetime import date, timedelta

class BudgetGuard:
    def __init__(self, config: dict):
        self.daily_limit = config["daily_limit"]       # 每日预算
        self.per_task_limit = config["per_task_limit"]  # 单任务预算上限
        self.alert_threshold = config["alert_threshold"]  # 告警阈值（0.8 = 80%）
    
    def check(self, estimated_cost: float, task_id: str) -> tuple[bool, str]:
        """检查是否允许执行"""
        
        # 1. 单任务预算检查
        if estimated_cost > self.per_task_limit:
            return False, f"Estimated cost ${estimated_cost:.2f} exceeds per-task limit ${self.per_task_limit:.2f}"
        
        # 2. 每日预算检查
        spent_today = get_daily_spent(date.today())
        remaining = self.daily_limit - spent_today
        
        if estimated_cost > remaining:
            return False, f"Estimated cost ${estimated_cost:.2f} exceeds daily remaining ${remaining:.2f}"
        
        # 3. 告警检查
        usage_ratio = spent_today / self.daily_limit
        if usage_ratio >= self.alert_threshold:
            send_alert(
                f"⚠️ 每日预算已使用 {usage_ratio:.0%}",
                details=f"已花费 ${spent_today:.2f} / ${self.daily_limit:.2f}",
                task_id=task_id,
            )
        
        return True, "OK"
    
    def record_cost(self, task_id: str, model: str, tokens: int, cost: float):
        """记录实际成本"""
        record_event({
            "task_id": task_id,
            "model": model,
            "input_tokens": tokens.get("input", 0),
            "output_tokens": tokens.get("output", 0),
            "cost": cost,
            "timestamp": datetime.now().isoformat(),
        })

// routing-log-entry.json
{
  "task_id": "TASK-2024-001",
  "task_type": "bugfix",
  "risk_level": "MEDIUM",
  "routing_decision": {
    "model": "sonnet",
    "matched_rule": "standard_bugfix",
    "reason": "标准 Bug 修复",
    "estimated_cost": 0.12,
    "actual_cost": 0.09,
    "fallback_triggered": false,
    "upgrade_chain": []
  },
  "quality": {
    "tests_passed": true,
    "quality_score": 0.85,
    "human_approved": false
  },
  "cost_saving_vs_opus": 0.41
}