引言 路由是Agent系统的"大脑"——它决定了每个请求由哪个模型处理、调用哪些工具、使用什么记忆策略。一个优秀的路由架构可以在不增加硬件资源的前提下，将Agent系统的整体性能提升30-50%，同时显著降低运营成本。
2026年，Agent路由已从简单的规则匹配进化到基于语义理解和强化学习的智能路由，成为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) } 路由监控指标 指标 说明 目标 路由准确率 路由决策被后续验证为正确的比例 >90% 模型利用率 各模型的负载均衡度 偏差<20% 平均路由延迟 路由决策耗时 <50ms 级联触发率 需要升级模型的比例 <15% 成本节省率 相比全用最强模型的节省比例 >40% 总结 Agent路由架构是系统性能和成本的杠杆点。语义路由实现了基于意图的精准分发，学习型路由通过历史数据持续优化决策，多模型调度在能力、负载和成本之间找到最优平衡，级联路由则提供了"先省后花"的成本优化思路。
...
