Agent路由架构：从简单路由到智能路由

阶段1：固定路由              阶段2：规则路由              阶段3：语义路由
┌──────────┐               ┌──────────┐               ┌──────────┐
│ 所有请求  │               │ 规则匹配   │               │ 语义理解   │
│   ↓      │               │   ↓      │               │   ↓      │
│ 同一模型  │               │ 多模型分发 │               │ 意图分类   │
│ 同一工具  │               │ 工具选择   │               │ 智能选模型  │
└──────────┘               └──────────┘               └──────────┘

                           阶段4：学习型路由
                           ┌──────────┐
                           │ 历史数据   │
                           │   ↓      │
                           │ RL优化    │
                           │   ↓      │
                           │ 动态路由   │
                           └──────────┘

class RuleBasedRouter:
    """基于规则的路由器"""
    
    def __init__(self):
        self.rules = []
    
    def add_rule(self, condition: callable, target: dict, priority: int = 0):
        """添加路由规则"""
        self.rules.append({
            "condition": condition,
            "target": target,
            "priority": priority
        })
        self.rules.sort(key=lambda r: r["priority"], reverse=True)
    
    async def route(self, request: dict) -> dict:
        """路由请求"""
        for rule in self.rules:
            if rule["condition"](request):
                return rule["target"]
        
        # 默认路由
        return {"model": "gpt-4o-mini", "tools": ["search"]}
    
    def setup_default_rules(self):
        """设置默认规则集"""
        
        # 代码相关 → 代码模型
        self.add_rule(
            condition=lambda r: "code" in r["input"].lower(),
            target={"model": "deepseek-coder", "tools": ["code_exec"]},
            priority=10
        )
        
        # 数学相关 → 数学增强模型
        self.add_rule(
            condition=lambda r: any(w in r["input"].lower() 
                                   for w in ["计算", "求解", "公式"]),
            target={"model": "gpt-4o", "tools": ["calculator"]},
            priority=8
        )
        
        # 简单问候 → 小模型
        self.add_rule(
            condition=lambda r: len(r["input"]) < 20,
            target={"model": "gpt-4o-mini", "tools": []},
            priority=5
        )

class SemanticRouter:
    """基于语义理解的路由器"""
    
    def __init__(self, embedding_model, vector_store):
        self.embedder = embedding_model
        self.vectors = vector_store
        self.routes = {}
    
    async def register_route(
        self,
        name: str,
        description: str,
        examples: list,
        target: dict
    ):
        """注册语义路由"""
        # 为路由描述和示例生成嵌入
        texts = [description] + examples
        embeddings = await self.embedder.embed_batch(texts)
        
        self.routes[name] = {
            "description": description,
            "examples": examples,
            "embeddings": embeddings,
            "target": target
        }
    
    async def route(self, user_input: str) -> dict:
        """语义路由"""
        # 生成输入嵌入
        input_embedding = await self.embedder.embed(user_input)
        
        # 计算与每个路由的相似度
        scores = {}
        for name, route in self.routes.items():
            # 使用所有嵌入的最大相似度
            similarities = [
                self._cosine_similarity(input_embedding, emb)
                for emb in route["embeddings"]
            ]
            scores[name] = max(similarities)
        
        # 选择最匹配的路由
        best_route = max(scores, key=scores.get)
        best_score = scores[best_route]
        
        # 如果置信度不足，使用默认路由
        if best_score < 0.7:
            logger.info(
                f"Low confidence route: {best_route} ({best_score:.2f}), "
                f"using default"
            )
            return self.routes.get("default", {}).get("target", {})
        
        return self.routes[best_route]["target"]
    
    @staticmethod
    def _cosine_similarity(a, b) -> float:
        return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

class LearningRouter:
    """基于历史数据的学习型路由器"""
    
    def __init__(self, feature_extractor, model_registry):
        self.features = feature_extractor
        self.models = model_registry
        self.routing_history = []
        self.performance_predictor = None  # 训练好的ML模型
    
    async def route(self, request: dict) -> dict:
        """智能路由——预测最佳模型和工具组合"""
        
        # 提取请求特征
        features = self._extract_features(request)
        
        # 预测每个候选模型的性能
        candidates = await self._get_candidates(features)
        
        predictions = []
        for candidate in candidates:
            perf = await self._predict_performance(features, candidate)
            predictions.append({
                "candidate": candidate,
                "predicted_quality": perf["quality"],
                "predicted_latency": perf["latency"],
                "predicted_cost": perf["cost"],
                "score": self._compute_score(perf)
            })
        
        # 选择综合得分最高的
        best = max(predictions, key=lambda p: p["score"])
        
        return best["candidate"]
    
    def _extract_features(self, request: dict) -> dict:
        """提取请求特征"""
        return {
            "input_length": len(request["input"]),
            "language": self._detect_language(request["input"]),
            "complexity_score": self._estimate_complexity(request["input"]),
            "has_code": "```" in request["input"],
            "has_math": any(c in request["input"] for c in ["∑", "∫", "√"]),
            "requires_search": self._needs_search(request["input"]),
            "conversation_depth": request.get("turn_count", 0),
            "user_tier": request.get("user_tier", "free"),
        }
    
    def _compute_score(self, perf: dict) -> float:
        """综合评分"""
        # 质量权重0.5，延迟权重0.3，成本权重0.2
        return (
            perf["predicted_quality"] * 0.5 +
            (1 - min(perf["predicted_latency"] / 5000, 1)) * 0.3 +
            (1 - min(perf["predicted_cost"] / 0.1, 1)) * 0.2
        )
    
    async def record_outcome(
        self,
        request: dict,
        route_decision: dict,
        outcome: dict
    ):
        """记录路由结果，用于持续学习"""
        self.routing_history.append({
            "features": self._extract_features(request),
            "decision": route_decision,
            "outcome": outcome,  # quality_score, latency, cost
            "timestamp": datetime.now()
        })
        
        # 定期重训练
        if len(self.routing_history) % 1000 == 0:
            await self._retrain()

class MultiModelScheduler:
    """多模型调度器"""
    
    def __init__(self):
        self.models = {
            "gpt-4o": {
                "strength": ["reasoning", "code", "creative"],
                "cost_per_1k": 0.015,
                "avg_latency_ms": 1500,
                "context_window": 128000,
                "max_concurrent": 10,
            },
            "gpt-4o-mini": {
                "strength": ["simple_qa", "summarization"],
                "cost_per_1k": 0.0003,
                "avg_latency_ms": 500,
                "context_window": 128000,
                "max_concurrent": 50,
            },
            "claude-3.5-sonnet": {
                "strength": ["analysis", "writing", "long_context"],
                "cost_per_1k": 0.008,
                "avg_latency_ms": 1200,
                "context_window": 200000,
                "max_concurrent": 20,
            },
            "deepseek-coder-v2": {
                "strength": ["code", "debug"],
                "cost_per_1k": 0.0014,
                "avg_latency_ms": 800,
                "context_window": 128000,
                "max_concurrent": 30,
            },
        }
        self.model_loads = {m: 0 for m in self.models}
    
    async def select_model(self, request: dict) -> str:
        """选择最优模型"""
        scores = {}
        
        for model_name, config in self.models.items():
            # 能力匹配分
            capability_score = self._capability_match(
                request, config["strength"]
            )
            
            # 负载分（负载越低越好）
            load_ratio = self.model_loads[model_name] / config["max_concurrent"]
            load_score = 1 - load_ratio
            
            # 成本分
            estimated_tokens = len(request["input"]) // 4
            cost = estimated_tokens * config["cost_per_1k"]
            cost_score = 1 - min(cost / 0.5, 1)
            
            # 综合得分
            scores[model_name] = (
                capability_score * 0.5 +
                load_score * 0.3 +
                cost_score * 0.2
            )
        
        return max(scores, key=scores.get)

class CascadingRouter:
    """级联路由——先试便宜模型，不够再升级"""
    
    CASCADE = [
        {"model": "gpt-4o-mini", "confidence_threshold": 0.85},
        {"model": "gpt-4o", "confidence_threshold": 0.75},
        {"model": "gpt-4o", "confidence_threshold": 0.0},  # 最终兜底
    ]
    
    async def route_with_cascade(self, request: dict) -> dict:
        """级联路由"""
        total_cost = 0
        
        for stage in self.CASCADE:
            model = stage["model"]
            threshold = stage["confidence_threshold"]
            
            response = await self._call_model(model, request)
            total_cost += response["cost"]
            
            if response["confidence"] >= threshold:
                return {
                    **response,
                    "model_used": model,
                    "total_cost": total_cost,
                    "cascade_stages": 1
                }
            
            # 将低置信度响应作为上下文传给下一级
            request["previous_attempt"] = response
        
        return {
            **response,
            "model_used": model,
            "total_cost": total_cost,
            "cascade_stages": len(self.CASCADE)
        }