多 Agent Token 優化系列 pt.4：模型分層實戰 — 智能路由打造高效低成本系統

 1from enum import Enum
 2from dataclasses import dataclass
 3from typing import Optional
 4
 5class TaskComplexity(Enum):
 6    """任務複雜度等級"""
 7    TRIVIAL = "trivial"      # 極簡單：格式化、提取
 8    SIMPLE = "simple"        # 簡單：分類、摘要、翻譯
 9    MODERATE = "moderate"    # 中等：程式碼生成、分析
10    COMPLEX = "complex"      # 複雜：架構設計、深度推理
11    EXPERT = "expert"        # 專家級：研究、創新、關鍵決策
12
13class TaskCategory(Enum):
14    """任務類別"""
15    EXTRACTION = "extraction"        # 資訊提取
16    CLASSIFICATION = "classification" # 分類
17    SUMMARIZATION = "summarization"  # 摘要
18    TRANSLATION = "translation"      # 翻譯
19    GENERATION = "generation"        # 內容生成
20    CODE = "code"                    # 程式碼相關
21    ANALYSIS = "analysis"            # 分析
22    REASONING = "reasoning"          # 推理
23    CREATIVE = "creative"            # 創意
24    CONVERSATION = "conversation"    # 對話
25
26@dataclass
27class ModelSpec:
28    """模型規格"""
29    name: str
30    model_id: str
31    input_cost_per_1m: float
32    output_cost_per_1m: float
33    max_tokens: int
34    strengths: list[str]
35    weaknesses: list[str]
36
37# 定義可用模型
38MODELS = {
39    "opus": ModelSpec(
40        name="Claude Opus 4",
41        model_id="claude-opus-4-20250514",
42        input_cost_per_1m=15.0,
43        output_cost_per_1m=75.0,
44        max_tokens=4096,
45        strengths=["複雜推理", "創意寫作", "深度分析", "專業知識"],
46        weaknesses=["成本高", "速度較慢"]
47    ),
48    "sonnet": ModelSpec(
49        name="Claude Sonnet 4",
50        model_id="claude-sonnet-4-20250514",
51        input_cost_per_1m=3.0,
52        output_cost_per_1m=15.0,
53        max_tokens=4096,
54        strengths=["程式碼生成", "一般分析", "平衡性能"],
55        weaknesses=["複雜推理略遜 Opus"]
56    ),
57    "haiku": ModelSpec(
58        name="Claude Haiku 3.5",
59        model_id="claude-3-5-haiku-20241022",
60        input_cost_per_1m=0.80,
61        output_cost_per_1m=4.0,
62        max_tokens=4096,
63        strengths=["速度快", "成本低", "簡單任務"],
64        weaknesses=["複雜任務能力有限"]
65    )
66}
67
68# 任務類型到複雜度的預設映射
69DEFAULT_COMPLEXITY_MAP = {
70    TaskCategory.EXTRACTION: TaskComplexity.TRIVIAL,
71    TaskCategory.CLASSIFICATION: TaskComplexity.SIMPLE,
72    TaskCategory.SUMMARIZATION: TaskComplexity.SIMPLE,
73    TaskCategory.TRANSLATION: TaskComplexity.SIMPLE,
74    TaskCategory.GENERATION: TaskComplexity.MODERATE,
75    TaskCategory.CODE: TaskComplexity.MODERATE,
76    TaskCategory.ANALYSIS: TaskComplexity.MODERATE,
77    TaskCategory.REASONING: TaskComplexity.COMPLEX,
78    TaskCategory.CREATIVE: TaskComplexity.COMPLEX,
79    TaskCategory.CONVERSATION: TaskComplexity.SIMPLE,
80}
81
82# 複雜度到模型的映射
83COMPLEXITY_MODEL_MAP = {
84    TaskComplexity.TRIVIAL: "haiku",
85    TaskComplexity.SIMPLE: "haiku",
86    TaskComplexity.MODERATE: "sonnet",
87    TaskComplexity.COMPLEX: "sonnet",
88    TaskComplexity.EXPERT: "opus",
89}

  1import anthropic
  2import re
  3from dataclasses import dataclass
  4from typing import Optional, Callable
  5
  6client = anthropic.Anthropic()
  7
  8@dataclass
  9class RoutingRule:
 10    """路由規則"""
 11    name: str
 12    condition: Callable[[str], bool]  # 判斷函數
 13    model: str                         # 目標模型
 14    priority: int = 0                  # 優先級（越高越優先）
 15
 16class RuleBasedRouter:
 17    """
 18    規則基礎模型路由器
 19
 20    優點：
 21    - 可預測、可控
 22    - 無額外 API 成本
 23    - 執行速度快
 24
 25    缺點：
 26    - 規則維護成本
 27    - 邊界情況處理困難
 28    - 難以適應新場景
 29    """
 30
 31    def __init__(self):
 32        self.rules: list[RoutingRule] = []
 33        self.default_model = "sonnet"
 34        self._stats = {model: 0 for model in MODELS}
 35
 36    def add_rule(self, rule: RoutingRule):
 37        """添加路由規則"""
 38        self.rules.append(rule)
 39        # 按優先級排序
 40        self.rules.sort(key=lambda r: r.priority, reverse=True)
 41
 42    def route(self, task: str) -> str:
 43        """根據規則選擇模型"""
 44        for rule in self.rules:
 45            if rule.condition(task):
 46                self._stats[rule.model] += 1
 47                return rule.model
 48
 49        self._stats[self.default_model] += 1
 50        return self.default_model
 51
 52    def call(self, task: str, system: Optional[str] = None, **kwargs) -> str:
 53        """路由並呼叫模型"""
 54        model_key = self.route(task)
 55        model_spec = MODELS[model_key]
 56
 57        response = client.messages.create(
 58            model=model_spec.model_id,
 59            max_tokens=kwargs.get("max_tokens", 2048),
 60            system=system or "You are a helpful assistant.",
 61            messages=[{"role": "user", "content": task}]
 62        )
 63
 64        return response.content[0].text
 65
 66    def get_stats(self) -> dict:
 67        total = sum(self._stats.values())
 68        return {
 69            "routing_stats": self._stats,
 70            "total_requests": total,
 71            "distribution": {
 72                k: f"{v/total*100:.1f}%" if total > 0 else "0%"
 73                for k, v in self._stats.items()
 74            }
 75        }
 76
 77
 78# 建立常用規則
 79def create_standard_rules() -> list[RoutingRule]:
 80    """建立標準路由規則集"""
 81    return [
 82        # 高優先級：強制使用 Opus 的情況
 83        RoutingRule(
 84            name="security_analysis",
 85            condition=lambda t: any(kw in t.lower() for kw in [
 86                "安全漏洞", "security vulnerability", "滲透測試",
 87                "資安審計", "安全審查"
 88            ]),
 89            model="opus",
 90            priority=100
 91        ),
 92        RoutingRule(
 93            name="architecture_design",
 94            condition=lambda t: any(kw in t.lower() for kw in [
 95                "架構設計", "system design", "設計模式",
 96                "技術選型", "架構決策"
 97            ]),
 98            model="opus",
 99            priority=100
100        ),
101
102        # 中優先級：使用 Sonnet
103        RoutingRule(
104            name="code_generation",
105            condition=lambda t: any(kw in t.lower() for kw in [
106                "寫一個", "實作", "implement", "寫程式",
107                "function", "class", "def ", "async "
108            ]),
109            model="sonnet",
110            priority=50
111        ),
112        RoutingRule(
113            name="code_review",
114            condition=lambda t: any(kw in t.lower() for kw in [
115                "審查", "review", "檢查這段", "優化這個"
116            ]),
117            model="sonnet",
118            priority=50
119        ),
120        RoutingRule(
121            name="analysis",
122            condition=lambda t: any(kw in t.lower() for kw in [
123                "分析", "analyze", "比較", "compare",
124                "評估", "evaluate"
125            ]),
126            model="sonnet",
127            priority=50
128        ),
129
130        # 低優先級：使用 Haiku
131        RoutingRule(
132            name="simple_question",
133            condition=lambda t: len(t) < 100 and "?" in t,
134            model="haiku",
135            priority=20
136        ),
137        RoutingRule(
138            name="translation",
139            condition=lambda t: any(kw in t.lower() for kw in [
140                "翻譯", "translate", "轉換成"
141            ]),
142            model="haiku",
143            priority=20
144        ),
145        RoutingRule(
146            name="summarization",
147            condition=lambda t: any(kw in t.lower() for kw in [
148                "摘要", "summarize", "總結", "概述"
149            ]),
150            model="haiku",
151            priority=20
152        ),
153        RoutingRule(
154            name="formatting",
155            condition=lambda t: any(kw in t.lower() for kw in [
156                "格式化", "format", "轉成 json", "轉成 markdown"
157            ]),
158            model="haiku",
159            priority=20
160        ),
161        RoutingRule(
162            name="extraction",
163            condition=lambda t: any(kw in t.lower() for kw in [
164                "提取", "extract", "找出", "列出"
165            ]),
166            model="haiku",
167            priority=20
168        ),
169    ]
170
171
172# 使用範例
173if __name__ == "__main__":
174    router = RuleBasedRouter()
175
176    # 載入標準規則
177    for rule in create_standard_rules():
178        router.add_rule(rule)
179
180    # 測試不同類型的任務
181    test_tasks = [
182        "請幫我翻譯這段文字成英文",                    # → Haiku
183        "總結以下文章的重點",                          # → Haiku
184        "寫一個 Python 函數來計算費氏數列",            # → Sonnet
185        "分析這段程式碼的效能問題",                    # → Sonnet
186        "設計一個高可用的微服務架構",                  # → Opus
187        "審查這段程式碼的安全漏洞",                    # → Opus
188        "今天天氣如何？",                             # → Haiku (短問題)
189    ]
190
191    print("模型路由測試：\n")
192    for task in test_tasks:
193        model = router.route(task)
194        print(f"任務：{task[:30]}...")
195        print(f"路由至：{MODELS[model].name}\n")
196
197    print("路由統計：")
198    print(router.get_stats())

  1import anthropic
  2import json
  3from dataclasses import dataclass
  4from typing import Optional
  5from enum import Enum
  6
  7client = anthropic.Anthropic()
  8
  9@dataclass
 10class ClassificationResult:
 11    """分類結果"""
 12    complexity: TaskComplexity
 13    category: TaskCategory
 14    confidence: float
 15    reasoning: str
 16    suggested_model: str
 17
 18class LLMClassifierRouter:
 19    """
 20    LLM 輔助分類路由器
 21
 22    使用 Haiku 進行任務分類，基於分類結果選擇目標模型
 23
 24    優點：
 25    - 更準確的任務理解
 26    - 可處理複雜的邊界情況
 27    - 自動適應新類型任務
 28
 29    缺點：
 30    - 額外的分類 API 呼叫成本
 31    - 增加一次網路往返延遲
 32    """
 33
 34    CLASSIFICATION_PROMPT = """分析以下任務，判斷其複雜度和類別。
 35
 36任務：
 37{task}
 38
 39請以 JSON 格式回答：
 40{{
 41    "complexity": "trivial|simple|moderate|complex|expert",
 42    "category": "extraction|classification|summarization|translation|generation|code|analysis|reasoning|creative|conversation",
 43    "confidence": 0.0-1.0,
 44    "reasoning": "簡短說明判斷依據"
 45}}
 46
 47複雜度判斷標準：
 48- trivial: 簡單格式化、直接提取
 49- simple: 基本分類、摘要、翻譯
 50- moderate: 程式碼生成、一般分析
 51- complex: 深度分析、多步推理
 52- expert: 創新設計、專家級決策
 53
 54只返回 JSON，不要其他說明。"""
 55
 56    def __init__(
 57        self,
 58        classifier_model: str = "claude-3-5-haiku-20241022",
 59        enable_caching: bool = True
 60    ):
 61        self.classifier_model = classifier_model
 62        self.enable_caching = enable_caching
 63        self._classification_cache: dict[str, ClassificationResult] = {}
 64
 65        self._stats = {
 66            "classifications": 0,
 67            "cache_hits": 0,
 68            "model_usage": {model: 0 for model in MODELS}
 69        }
 70
 71    def _get_cache_key(self, task: str) -> str:
 72        """計算快取鍵"""
 73        # 使用任務的前 500 字元作為鍵（避免過長）
 74        import hashlib
 75        return hashlib.md5(task[:500].encode()).hexdigest()
 76
 77    def classify(self, task: str) -> ClassificationResult:
 78        """分類任務"""
 79        # 檢查快取
 80        if self.enable_caching:
 81            cache_key = self._get_cache_key(task)
 82            if cache_key in self._classification_cache:
 83                self._stats["cache_hits"] += 1
 84                return self._classification_cache[cache_key]
 85
 86        # 呼叫分類模型
 87        response = client.messages.create(
 88            model=self.classifier_model,
 89            max_tokens=300,
 90            messages=[{
 91                "role": "user",
 92                "content": self.CLASSIFICATION_PROMPT.format(task=task[:2000])
 93            }]
 94        )
 95
 96        self._stats["classifications"] += 1
 97
 98        # 解析結果
 99        try:
100            text = response.content[0].text
101            if "```" in text:
102                text = text.split("```")[1].split("```")[0]
103                if text.startswith("json"):
104                    text = text[4:]
105
106            data = json.loads(text.strip())
107
108            result = ClassificationResult(
109                complexity=TaskComplexity(data["complexity"]),
110                category=TaskCategory(data["category"]),
111                confidence=data.get("confidence", 0.8),
112                reasoning=data.get("reasoning", ""),
113                suggested_model=self._map_to_model(
114                    TaskComplexity(data["complexity"]),
115                    TaskCategory(data["category"])
116                )
117            )
118
119            # 快取結果
120            if self.enable_caching:
121                self._classification_cache[cache_key] = result
122
123            return result
124
125        except (json.JSONDecodeError, KeyError, ValueError) as e:
126            # 預設返回中等複雜度
127            return ClassificationResult(
128                complexity=TaskComplexity.MODERATE,
129                category=TaskCategory.GENERATION,
130                confidence=0.5,
131                reasoning=f"分類失敗，使用預設: {e}",
132                suggested_model="sonnet"
133            )
134
135    def _map_to_model(self, complexity: TaskComplexity, category: TaskCategory) -> str:
136        """根據複雜度和類別映射到模型"""
137        # 特殊類別覆寫
138        if category in [TaskCategory.REASONING, TaskCategory.CREATIVE]:
139            if complexity in [TaskComplexity.COMPLEX, TaskComplexity.EXPERT]:
140                return "opus"
141
142        if category == TaskCategory.CODE:
143            if complexity == TaskComplexity.EXPERT:
144                return "opus"
145            return "sonnet"
146
147        # 預設按複雜度
148        return COMPLEXITY_MODEL_MAP.get(complexity, "sonnet")
149
150    def route_and_call(
151        self,
152        task: str,
153        system: Optional[str] = None,
154        **kwargs
155    ) -> tuple[str, ClassificationResult]:
156        """
157        分類、路由並呼叫
158
159        Returns:
160            (模型回應, 分類結果)
161        """
162        # Step 1: 分類
163        classification = self.classify(task)
164
165        # Step 2: 取得目標模型
166        model_key = classification.suggested_model
167        model_spec = MODELS[model_key]
168        self._stats["model_usage"][model_key] += 1
169
170        # Step 3: 呼叫目標模型
171        response = client.messages.create(
172            model=model_spec.model_id,
173            max_tokens=kwargs.get("max_tokens", 2048),
174            system=system or "You are a helpful assistant.",
175            messages=[{"role": "user", "content": task}]
176        )
177
178        return response.content[0].text, classification
179
180    def get_stats(self) -> dict:
181        total_requests = sum(self._stats["model_usage"].values())
182        return {
183            "total_classifications": self._stats["classifications"],
184            "cache_hits": self._stats["cache_hits"],
185            "cache_hit_rate": f"{self._stats['cache_hits'] / max(self._stats['classifications'] + self._stats['cache_hits'], 1) * 100:.1f}%",
186            "model_usage": self._stats["model_usage"],
187            "model_distribution": {
188                k: f"{v/max(total_requests, 1)*100:.1f}%"
189                for k, v in self._stats["model_usage"].items()
190            }
191        }
192
193
194# 使用範例
195if __name__ == "__main__":
196    router = LLMClassifierRouter()
197
198    test_tasks = [
199        "把 'Hello World' 翻譯成中文",
200        "寫一個 Python 快速排序演算法",
201        "設計一個支援百萬用戶的即時通訊系統架構",
202        "這段程式碼有什麼問題？\ndef add(a, b): return a + b",
203        "分析 React 和 Vue 的優缺點",
204    ]
205
206    print("LLM 輔助分類路由測試：\n")
207    for task in test_tasks:
208        response, classification = router.route_and_call(task)
209        print(f"任務：{task[:40]}...")
210        print(f"分類：{classification.category.value} / {classification.complexity.value}")
211        print(f"信心度：{classification.confidence}")
212        print(f"路由至：{MODELS[classification.suggested_model].name}")
213        print(f"回應：{response[:100]}...\n")
214
215    print("統計：")
216    print(router.get_stats())

  1import anthropic
  2import re
  3from dataclasses import dataclass
  4from typing import Optional, Callable
  5from enum import Enum
  6
  7client = anthropic.Anthropic()
  8
  9class QualityLevel(Enum):
 10    """品質等級"""
 11    HIGH = "high"
 12    ACCEPTABLE = "acceptable"
 13    LOW = "low"
 14    FAILED = "failed"
 15
 16@dataclass
 17class QualityAssessment:
 18    """品質評估結果"""
 19    level: QualityLevel
 20    score: float  # 0-1
 21    issues: list[str]
 22    should_upgrade: bool
 23
 24@dataclass
 25class EscalationResult:
 26    """升級結果"""
 27    response: str
 28    final_model: str
 29    attempts: list[dict]  # 每次嘗試的記錄
 30    total_cost: float
 31
 32class QualityChecker:
 33    """品質檢查器"""
 34
 35    # 品質問題指標
 36    UNCERTAINTY_MARKERS = [
 37        "我不確定", "我無法", "不太清楚", "可能不準確",
 38        "I'm not sure", "I cannot", "I don't know",
 39        "抱歉，我無法", "這超出了我的能力"
 40    ]
 41
 42    INCOMPLETE_MARKERS = [
 43        "...", "等等", "以此類推", "更多內容",
 44        "（未完待續）", "TODO", "待補充"
 45    ]
 46
 47    def __init__(
 48        self,
 49        min_response_length: int = 50,
 50        max_uncertainty_ratio: float = 0.1,
 51        custom_validators: Optional[list[Callable[[str], bool]]] = None
 52    ):
 53        self.min_response_length = min_response_length
 54        self.max_uncertainty_ratio = max_uncertainty_ratio
 55        self.custom_validators = custom_validators or []
 56
 57    def assess(self, response: str, task: str) -> QualityAssessment:
 58        """評估回應品質"""
 59        issues = []
 60        score = 1.0
 61
 62        # 檢查長度
 63        if len(response) < self.min_response_length:
 64            issues.append("回應過短")
 65            score -= 0.3
 66
 67        # 檢查不確定性標記
 68        uncertainty_count = sum(
 69            1 for marker in self.UNCERTAINTY_MARKERS
 70            if marker.lower() in response.lower()
 71        )
 72        if uncertainty_count > 0:
 73            issues.append(f"包含 {uncertainty_count} 個不確定性標記")
 74            score -= 0.2 * uncertainty_count
 75
 76        # 檢查不完整標記
 77        incomplete_count = sum(
 78            1 for marker in self.INCOMPLETE_MARKERS
 79            if marker in response
 80        )
 81        if incomplete_count > 0:
 82            issues.append("回應可能不完整")
 83            score -= 0.2
 84
 85        # 檢查是否拒絕回答
 86        refusal_patterns = [
 87            r"我無法提供", r"我不能幫助", r"這違反",
 88            r"I cannot", r"I'm unable to"
 89        ]
 90        for pattern in refusal_patterns:
 91            if re.search(pattern, response, re.IGNORECASE):
 92                issues.append("模型拒絕回答")
 93                score -= 0.5
 94                break
 95
 96        # 執行自訂驗證器
 97        for validator in self.custom_validators:
 98            if not validator(response):
 99                issues.append("未通過自訂驗證")
100                score -= 0.2
101
102        # 計算最終評級
103        score = max(0, score)
104        if score >= 0.8:
105            level = QualityLevel.HIGH
106        elif score >= 0.6:
107            level = QualityLevel.ACCEPTABLE
108        elif score >= 0.3:
109            level = QualityLevel.LOW
110        else:
111            level = QualityLevel.FAILED
112
113        return QualityAssessment(
114            level=level,
115            score=score,
116            issues=issues,
117            should_upgrade=level in [QualityLevel.LOW, QualityLevel.FAILED]
118        )
119
120
121class DynamicEscalationRouter:
122    """
123    動態升級路由器
124
125    從便宜模型開始嘗試，品質不達標時升級到更強模型
126    """
127
128    # 模型升級順序
129    MODEL_LADDER = ["haiku", "sonnet", "opus"]
130
131    def __init__(
132        self,
133        quality_checker: Optional[QualityChecker] = None,
134        start_model: str = "haiku",
135        max_attempts: int = 3
136    ):
137        self.quality_checker = quality_checker or QualityChecker()
138        self.start_model = start_model
139        self.max_attempts = max_attempts
140
141        self._stats = {
142            "total_requests": 0,
143            "escalations": 0,
144            "model_attempts": {model: 0 for model in MODELS},
145            "model_finals": {model: 0 for model in MODELS},
146            "total_cost": 0.0
147        }
148
149    def _call_model(self, model_key: str, task: str, system: Optional[str] = None) -> tuple[str, float]:
150        """呼叫模型並計算成本"""
151        model_spec = MODELS[model_key]
152
153        response = client.messages.create(
154            model=model_spec.model_id,
155            max_tokens=2048,
156            system=system or "You are a helpful assistant.",
157            messages=[{"role": "user", "content": task}]
158        )
159
160        # 計算成本
161        input_tokens = response.usage.input_tokens
162        output_tokens = response.usage.output_tokens
163        cost = (
164            input_tokens * model_spec.input_cost_per_1m / 1_000_000 +
165            output_tokens * model_spec.output_cost_per_1m / 1_000_000
166        )
167
168        return response.content[0].text, cost
169
170    def _get_next_model(self, current_model: str) -> Optional[str]:
171        """取得下一個升級模型"""
172        try:
173            current_idx = self.MODEL_LADDER.index(current_model)
174            if current_idx < len(self.MODEL_LADDER) - 1:
175                return self.MODEL_LADDER[current_idx + 1]
176        except ValueError:
177            pass
178        return None
179
180    def route_and_call(
181        self,
182        task: str,
183        system: Optional[str] = None,
184        quality_threshold: QualityLevel = QualityLevel.ACCEPTABLE
185    ) -> EscalationResult:
186        """
187        動態路由並呼叫
188
189        Args:
190            task: 任務
191            system: System prompt
192            quality_threshold: 最低可接受品質等級
193
194        Returns:
195            EscalationResult
196        """
197        self._stats["total_requests"] += 1
198
199        attempts = []
200        current_model = self.start_model
201        total_cost = 0.0
202        final_response = None
203        final_model = None
204
205        acceptable_levels = {
206            QualityLevel.HIGH: [QualityLevel.HIGH],
207            QualityLevel.ACCEPTABLE: [QualityLevel.HIGH, QualityLevel.ACCEPTABLE],
208            QualityLevel.LOW: [QualityLevel.HIGH, QualityLevel.ACCEPTABLE, QualityLevel.LOW],
209        }
210        acceptable = acceptable_levels.get(quality_threshold, [QualityLevel.HIGH, QualityLevel.ACCEPTABLE])
211
212        for attempt in range(self.max_attempts):
213            if current_model is None:
214                break
215
216            # 嘗試當前模型
217            self._stats["model_attempts"][current_model] += 1
218            response, cost = self._call_model(current_model, task, system)
219            total_cost += cost
220
221            # 評估品質
222            assessment = self.quality_checker.assess(response, task)
223
224            attempts.append({
225                "model": current_model,
226                "quality_level": assessment.level.value,
227                "quality_score": assessment.score,
228                "issues": assessment.issues,
229                "cost": cost
230            })
231
232            # 檢查是否達標
233            if assessment.level in acceptable:
234                final_response = response
235                final_model = current_model
236                break
237
238            # 升級到下一個模型
239            self._stats["escalations"] += 1
240            current_model = self._get_next_model(current_model)
241
242            print(f"⬆️ 品質不達標 ({assessment.level.value})，升級模型...")
243
244        # 如果所有嘗試都失敗，使用最後一次的結果
245        if final_response is None and attempts:
246            final_response = response
247            final_model = attempts[-1]["model"]
248
249        self._stats["model_finals"][final_model] += 1
250        self._stats["total_cost"] += total_cost
251
252        return EscalationResult(
253            response=final_response,
254            final_model=final_model,
255            attempts=attempts,
256            total_cost=total_cost
257        )
258
259    def get_stats(self) -> dict:
260        total = self._stats["total_requests"]
261        return {
262            **self._stats,
263            "escalation_rate": f"{self._stats['escalations'] / max(total, 1) * 100:.1f}%",
264            "average_cost": f"${self._stats['total_cost'] / max(total, 1):.4f}",
265            "final_model_distribution": {
266                k: f"{v/max(total, 1)*100:.1f}%"
267                for k, v in self._stats["model_finals"].items()
268            }
269        }
270
271
272# 使用範例
273if __name__ == "__main__":
274    router = DynamicEscalationRouter(
275        start_model="haiku",
276        max_attempts=3
277    )
278
279    test_tasks = [
280        "1 + 1 等於多少？",  # 簡單，Haiku 可處理
281        "寫一個 Python 二分搜尋演算法",  # 中等，可能需要 Sonnet
282        "設計一個分散式鎖的實現方案，考慮網路分區容錯",  # 複雜
283    ]
284
285    print("動態升級路由測試：\n")
286    for task in test_tasks:
287        print(f"任務：{task[:50]}...")
288        result = router.route_and_call(task)
289
290        print(f"最終模型：{MODELS[result.final_model].name}")
291        print(f"嘗試次數：{len(result.attempts)}")
292        print(f"總成本：${result.total_cost:.4f}")
293
294        for i, attempt in enumerate(result.attempts, 1):
295            print(f"  嘗試 {i}: {attempt['model']} - {attempt['quality_level']} (${attempt['cost']:.4f})")
296
297        print(f"回應：{result.response[:100]}...\n")
298
299    print("統計：")
300    for k, v in router.get_stats().items():
301        print(f"  {k}: {v}")

  1import anthropic
  2import json
  3import time
  4from dataclasses import dataclass, field
  5from typing import Optional, Callable, Any
  6from enum import Enum
  7from collections import defaultdict
  8
  9client = anthropic.Anthropic()
 10
 11@dataclass
 12class RoutingDecision:
 13    """路由決策記錄"""
 14    task_hash: str
 15    selected_model: str
 16    selection_method: str  # "rule", "llm_classify", "escalation"
 17    confidence: float
 18    reasoning: str
 19    timestamp: float = field(default_factory=time.time)
 20
 21@dataclass
 22class ExecutionResult:
 23    """執行結果"""
 24    response: str
 25    model_used: str
 26    routing_decision: RoutingDecision
 27    quality_score: float
 28    total_cost: float
 29    latency_ms: float
 30    escalated: bool = False
 31
 32class HybridIntelligentRouter:
 33    """
 34    混合智能路由系統
 35
 36    結合規則、LLM 分類、動態升級的完整路由方案
 37    """
 38
 39    def __init__(
 40        self,
 41        enable_llm_classification: bool = True,
 42        enable_quality_escalation: bool = True,
 43        enable_feedback_learning: bool = True
 44    ):
 45        self.enable_llm_classification = enable_llm_classification
 46        self.enable_quality_escalation = enable_quality_escalation
 47        self.enable_feedback_learning = enable_feedback_learning
 48
 49        # 組件
 50        self.rule_router = RuleBasedRouter()
 51        self.llm_classifier = LLMClassifierRouter() if enable_llm_classification else None
 52        self.quality_checker = QualityChecker()
 53
 54        # 載入標準規則
 55        for rule in create_standard_rules():
 56            self.rule_router.add_rule(rule)
 57
 58        # 學習記錄
 59        self._routing_history: list[dict] = []
 60        self._model_performance: dict[str, list[float]] = defaultdict(list)
 61
 62        # 統計
 63        self._stats = {
 64            "total_requests": 0,
 65            "rule_matches": 0,
 66            "llm_classifications": 0,
 67            "escalations": 0,
 68            "total_cost": 0.0,
 69            "total_latency_ms": 0.0,
 70            "model_usage": defaultdict(int)
 71        }
 72
 73    def _compute_task_hash(self, task: str) -> str:
 74        """計算任務雜湊"""
 75        import hashlib
 76        return hashlib.md5(task[:500].encode()).hexdigest()[:12]
 77
 78    def _call_model(
 79        self,
 80        model_key: str,
 81        task: str,
 82        system: Optional[str] = None
 83    ) -> tuple[str, float, float]:
 84        """呼叫模型，返回 (回應, 成本, 延遲ms)"""
 85        model_spec = MODELS[model_key]
 86
 87        start_time = time.time()
 88        response = client.messages.create(
 89            model=model_spec.model_id,
 90            max_tokens=2048,
 91            system=system or "You are a helpful assistant.",
 92            messages=[{"role": "user", "content": task}]
 93        )
 94        latency_ms = (time.time() - start_time) * 1000
 95
 96        # 計算成本
 97        cost = (
 98            response.usage.input_tokens * model_spec.input_cost_per_1m / 1_000_000 +
 99            response.usage.output_tokens * model_spec.output_cost_per_1m / 1_000_000
100        )
101
102        return response.content[0].text, cost, latency_ms
103
104    def _try_rule_routing(self, task: str) -> Optional[RoutingDecision]:
105        """嘗試規則路由"""
106        for rule in self.rule_router.rules:
107            if rule.condition(task):
108                return RoutingDecision(
109                    task_hash=self._compute_task_hash(task),
110                    selected_model=rule.model,
111                    selection_method="rule",
112                    confidence=0.9,
113                    reasoning=f"匹配規則: {rule.name}"
114                )
115        return None
116
117    def _try_llm_classification(self, task: str) -> RoutingDecision:
118        """LLM 分類路由"""
119        classification = self.llm_classifier.classify(task)
120        return RoutingDecision(
121            task_hash=self._compute_task_hash(task),
122            selected_model=classification.suggested_model,
123            selection_method="llm_classify",
124            confidence=classification.confidence,
125            reasoning=f"LLM 分類: {classification.category.value}/{classification.complexity.value}"
126        )
127
128    def _try_escalation(
129        self,
130        task: str,
131        current_model: str,
132        current_response: str,
133        system: Optional[str]
134    ) -> Optional[tuple[str, str, float, float]]:
135        """嘗試升級到更強模型"""
136        # 評估當前品質
137        assessment = self.quality_checker.assess(current_response, task)
138
139        if not assessment.should_upgrade:
140            return None
141
142        # 取得下一個模型
143        model_ladder = ["haiku", "sonnet", "opus"]
144        try:
145            current_idx = model_ladder.index(current_model)
146            if current_idx >= len(model_ladder) - 1:
147                return None
148            next_model = model_ladder[current_idx + 1]
149        except ValueError:
150            return None
151
152        # 呼叫更強模型
153        response, cost, latency = self._call_model(next_model, task, system)
154        return response, next_model, cost, latency
155
156    def _record_feedback(
157        self,
158        decision: RoutingDecision,
159        result: ExecutionResult
160    ):
161        """記錄回饋用於學習"""
162        if not self.enable_feedback_learning:
163            return
164
165        record = {
166            "task_hash": decision.task_hash,
167            "model": result.model_used,
168            "method": decision.selection_method,
169            "quality": result.quality_score,
170            "cost": result.total_cost,
171            "latency": result.latency_ms,
172            "escalated": result.escalated,
173            "timestamp": time.time()
174        }
175
176        self._routing_history.append(record)
177        self._model_performance[result.model_used].append(result.quality_score)
178
179        # 保留最近 1000 條記錄
180        if len(self._routing_history) > 1000:
181            self._routing_history = self._routing_history[-1000:]
182
183    def route_and_execute(
184        self,
185        task: str,
186        system: Optional[str] = None
187    ) -> ExecutionResult:
188        """
189        智能路由並執行任務
190
191        流程：
192        1. 嘗試規則匹配
193        2. 若無匹配，使用 LLM 分類
194        3. 執行選定模型
195        4. 評估品質，必要時升級
196        5. 記錄回饋
197        """
198        self._stats["total_requests"] += 1
199        total_cost = 0.0
200        total_latency = 0.0
201
202        # 階段 1: 規則路由
203        decision = self._try_rule_routing(task)
204        if decision:
205            self._stats["rule_matches"] += 1
206        elif self.enable_llm_classification:
207            # 階段 2: LLM 分類
208            decision = self._try_llm_classification(task)
209            self._stats["llm_classifications"] += 1
210            # 分類成本
211            total_cost += 0.001  # 估算 Haiku 分類成本
212        else:
213            # 預設使用 Sonnet
214            decision = RoutingDecision(
215                task_hash=self._compute_task_hash(task),
216                selected_model="sonnet",
217                selection_method="default",
218                confidence=0.5,
219                reasoning="預設路由"
220            )
221
222        # 階段 3: 執行
223        selected_model = decision.selected_model
224        response, cost, latency = self._call_model(selected_model, task, system)
225        total_cost += cost
226        total_latency += latency
227        self._stats["model_usage"][selected_model] += 1
228
229        # 階段 4: 品質評估和可能的升級
230        escalated = False
231        if self.enable_quality_escalation:
232            escalation_result = self._try_escalation(task, selected_model, response, system)
233            if escalation_result:
234                response, selected_model, esc_cost, esc_latency = escalation_result
235                total_cost += esc_cost
236                total_latency += esc_latency
237                escalated = True
238                self._stats["escalations"] += 1
239                self._stats["model_usage"][selected_model] += 1
240
241        # 最終品質評估
242        final_assessment = self.quality_checker.assess(response, task)
243
244        # 更新統計
245        self._stats["total_cost"] += total_cost
246        self._stats["total_latency_ms"] += total_latency
247
248        # 構建結果
249        result = ExecutionResult(
250            response=response,
251            model_used=selected_model,
252            routing_decision=decision,
253            quality_score=final_assessment.score,
254            total_cost=total_cost,
255            latency_ms=total_latency,
256            escalated=escalated
257        )
258
259        # 階段 5: 記錄回饋
260        self._record_feedback(decision, result)
261
262        return result
263
264    def get_stats(self) -> dict:
265        """取得統計資訊"""
266        total = self._stats["total_requests"]
267        return {
268            "total_requests": total,
269            "rule_match_rate": f"{self._stats['rule_matches'] / max(total, 1) * 100:.1f}%",
270            "llm_classification_rate": f"{self._stats['llm_classifications'] / max(total, 1) * 100:.1f}%",
271            "escalation_rate": f"{self._stats['escalations'] / max(total, 1) * 100:.1f}%",
272            "total_cost": f"${self._stats['total_cost']:.4f}",
273            "average_cost": f"${self._stats['total_cost'] / max(total, 1):.4f}",
274            "average_latency_ms": f"{self._stats['total_latency_ms'] / max(total, 1):.0f}",
275            "model_usage": dict(self._stats["model_usage"]),
276            "model_distribution": {
277                k: f"{v/max(total, 1)*100:.1f}%"
278                for k, v in self._stats["model_usage"].items()
279            }
280        }
281
282    def get_model_performance_summary(self) -> dict:
283        """取得各模型效能摘要"""
284        summary = {}
285        for model, scores in self._model_performance.items():
286            if scores:
287                summary[model] = {
288                    "sample_count": len(scores),
289                    "avg_quality": f"{sum(scores)/len(scores):.2f}",
290                    "min_quality": f"{min(scores):.2f}",
291                    "max_quality": f"{max(scores):.2f}"
292                }
293        return summary
294
295
296# 使用範例與效能測試
297if __name__ == "__main__":
298    router = HybridIntelligentRouter(
299        enable_llm_classification=True,
300        enable_quality_escalation=True,
301        enable_feedback_learning=True
302    )
303
304    # 綜合測試任務
305    test_tasks = [
306        # 簡單任務（應該用 Haiku）
307        "把 'Hello' 翻譯成中文",
308        "列出 1 到 10 的數字",
309        "總結：今天天氣很好。",
310
311        # 中等任務（應該用 Sonnet）
312        "寫一個 Python 函數計算陣列的平均值",
313        "解釋 REST API 和 GraphQL 的差異",
314        "分析這段程式碼的時間複雜度",
315
316        # 複雜任務（可能需要 Opus）
317        "設計一個支援百萬用戶的即時通訊系統，考慮高可用和水平擴展",
318        "審查這段程式碼的安全漏洞並提供修復建議",
319    ]
320
321    print("="*70)
322    print("混合智能路由系統測試")
323    print("="*70)
324
325    for task in test_tasks:
326        print(f"\n任務：{task[:50]}...")
327        result = router.route_and_execute(task)
328
329        print(f"  路由方法：{result.routing_decision.selection_method}")
330        print(f"  初始模型：{result.routing_decision.selected_model}")
331        print(f"  最終模型：{result.model_used}")
332        print(f"  是否升級：{'是' if result.escalated else '否'}")
333        print(f"  品質分數：{result.quality_score:.2f}")
334        print(f"  成本：${result.total_cost:.4f}")
335        print(f"  延遲：{result.latency_ms:.0f}ms")
336
337    print("\n" + "="*70)
338    print("統計摘要")
339    print("="*70)
340    for k, v in router.get_stats().items():
341        print(f"  {k}: {v}")
342
343    print("\n模型效能摘要：")
344    for model, perf in router.get_model_performance_summary().items():
345        print(f"  {model}: {perf}")