AI Agent 多模型协作：从模型路由到结果聚合的编排策略
AI Agent 多模型协作：从模型路由到结果聚合的编排策略
一、单模型的"能力天花板"：通用模型做不好专业任务
单一 LLM 在通用对话中表现优秀，但在专业任务上存在明显短板。某客服系统中，GPT-4 处理一般咨询的满意度达 92%，但处理退款计算时错误率 15%（数学推理弱），处理多语言工单时翻译质量不稳定。引入专业模型后：数学计算使用专用推理模型（错误率降至 2%），翻译使用翻译模型（BLEU 提升 12 分），通用对话仍用 GPT-4。整体满意度从 92% 提升到 97%，但总推理成本仅增加 8%——因为 80% 的请求仍由通用模型处理。
多模型协作的核心思想是：让每个请求被路由到最合适的模型，而非用最强模型处理所有请求。这既是性能优化，也是成本优化。
二、多模型协作的编排架构
flowchart TB subgraph 路由层["模型路由层"] R1[意图分类器<br/>轻量模型 / 规则引擎<br/>延迟 < 50ms] R2[复杂度评估器<br/>判断是否需要强模型] R3[成本预算器<br/>Token 预算分配] end subgraph 模型层["模型池"] M1[通用大模型<br/>GPT-4 / Claude<br/>复杂推理] M2[专业小模型<br/>Code / Math / Translation<br/>垂直任务] M3[快速模型<br/>GPT-3.5 / 本地模型<br/>简单任务] end subgraph 聚合层["结果聚合层"] A1[结果校验<br/>交叉验证 / 格式检查] A2[结果融合<br/>投票 / 加权 / 级联] A3[质量评估<br/>置信度打分] end R1 --> M1 & M2 & M3 R2 --> M1 & M3 R3 --> M1 & M2 & M3 M1 --> A1 & A2 M2 --> A1 & A2 M3 --> A1 A1 --> A3 A2 --> A3 style 路由层 fill:#eef,stroke:#333 style 模型层 fill:#fee,stroke:#333 style 聚合层 fill:#efe,stroke:#333三、多模型协作的代码实现
from dataclasses import dataclass, field from typing import List, Dict, Optional, Callable, Any, Tuple from enum import Enum from datetime import datetime import time class TaskType(Enum): GENERAL_QA = "general_qa" # 通用问答 CODE_GENERATION = "code_generation" # 代码生成 MATH_REASONING = "math_reasoning" # 数学推理 TRANSLATION = "translation" # 翻译 SUMMARIZATION = "summarization" # 摘要 CREATIVE_WRITING = "creative_writing" # 创意写作 class ModelTier(Enum): POWERFUL = "powerful" # 强模型（高成本、高质量） SPECIALIZED = "specialized" # 专业模型（中等成本、垂直高质量） FAST = "fast" # 快速模型（低成本、基础质量） @dataclass class ModelConfig: """模型配置""" model_id: str tier: ModelTier supported_tasks: List[TaskType] cost_per_1k_tokens: float # 美元 avg_latency_ms: float quality_scores: Dict[TaskType, float] # 各任务类型的质量评分 0-1 max_context_tokens: int @dataclass class RoutingDecision: """路由决策""" selected_models: List[str] task_type: TaskType confidence: float estimated_cost: float estimated_latency_ms: float routing_reason: str @dataclass class ModelResponse: """模型响应""" model_id: str content: str latency_ms: float token_count: int cost: float confidence: float = 0.0 class ModelRouter: """ 模型路由器：根据请求特征选择最优模型组合 """ def __init__(self): self._models: Dict[str, ModelConfig] = {} self._routing_rules: List[Dict] = [] self._cost_budget: float = 0.05 # 单次请求预算（美元） def register_model(self, config: ModelConfig): """注册模型""" self._models[config.model_id] = config def route(self, prompt: str, task_type: Optional[TaskType] = None) -> RoutingDecision: """路由请求到最优模型""" # Step 1: 确定任务类型 if not task_type: task_type = self._classify_task(prompt) # Step 2: 根据任务类型和成本预算选择模型 candidates = self._get_candidates(task_type) if not candidates: # 降级到通用模型 fallback = [ m for m in self._models.values() if TaskType.GENERAL_QA in m.supported_tasks ] candidates = fallback # Step 3: 选择最优模型 selected = self._select_optimal(candidates, task_type) return RoutingDecision( selected_models=[m.model_id for m in selected], task_type=task_type, confidence=self._estimate_confidence(selected, task_type), estimated_cost=sum(m.cost_per_1k_tokens * 0.5 for m in selected), estimated_latency_ms=max(m.avg_latency_ms for m in selected), routing_reason=self._explain_routing(selected, task_type), ) def _classify_task(self, prompt: str) -> TaskType: """基于规则的任务分类""" rules = { TaskType.CODE_GENERATION: [ "写代码", "实现函数", "编程", "code", "function", "class", "def ", "import", ], TaskType.MATH_REASONING: [ "计算", "求解", "方程", "积分", "概率", "证明", "数学", "calculate", ], TaskType.TRANSLATION: [ "翻译", "translate", "英文", "中文", "日文", ], TaskType.SUMMARIZATION: [ "总结", "摘要", "概括", "summarize", "summary", ], TaskType.CREATIVE_WRITING: [ "写一首", "创作", "故事", "小说", "poem", ], } prompt_lower = prompt.lower() for task_type, keywords in rules.items(): if any(kw in prompt_lower for kw in keywords): return task_type return TaskType.GENERAL_QA def _get_candidates(self, task_type: TaskType) -> List[ModelConfig]: """获取支持该任务类型的模型""" return [ m for m in self._models.values() if task_type in m.supported_tasks ] def _select_optimal(self, candidates: List[ModelConfig], task_type: TaskType) -> List[ModelConfig]: """选择最优模型组合""" if not candidates: return [] # 按任务类型质量评分排序 scored = [] for model in candidates: quality = model.quality_scores.get(task_type, 0.5) cost_efficiency = quality / max(model.cost_per_1k_tokens, 0.001) scored.append((model, quality, cost_efficiency)) scored.sort(key=lambda x: x[1], reverse=True) # 选择策略：高质量模型为主，低成本模型为备选 selected = [scored[0][0]] # 最优模型 # 如果最优模型成本高，添加快速模型作为降级选项 if scored[0][0].tier == ModelTier.POWERFUL: fast_models = [ m for m, _, _ in scored if m.tier == ModelTier.FAST ] if fast_models: selected.append(fast_models[0]) return selected def _estimate_confidence(self, models: List[ModelConfig], task_type: TaskType) -> float: """估算路由置信度""" if not models: return 0.0 return max(m.quality_scores.get(task_type, 0.5) for m in models) def _explain_routing(self, models: List[ModelConfig], task_type: TaskType) -> str: """生成路由解释""" if not models: return "无可选模型" primary = models[0] return ( f"任务类型 {task_type.value}，" f"选择 {primary.model_id}（质量 {primary.quality_scores.get(task_type, 0):.0%}）" ) # ============ 结果聚合器 ============ class ResultAggregator: """ 结果聚合器：合并多模型输出 """ @staticmethod def majority_vote(responses: List[ModelResponse]) -> ModelResponse: """多数投票：选择出现最多的答案""" if not responses: raise ValueError("无响应可聚合") if len(responses) == 1: return responses[0] # 精确匹配投票 vote_count: Dict[str, int] = {} for resp in responses: normalized = resp.content.strip().lower() vote_count[normalized] = vote_count.get(normalized, 0) + 1 # 选择票数最多的 winner_content = max(vote_count, key=vote_count.get) winner_resp = next( r for r in responses if r.content.strip().lower() == winner_content ) return ModelResponse( model_id="aggregated:vote", content=winner_resp.content, latency_ms=max(r.latency_ms for r in responses), token_count=sum(r.token_count for r in responses), cost=sum(r.cost for r in responses), confidence=vote_count[winner_content] / len(responses), ) @staticmethod def cascade(responses: List[ModelResponse], quality_threshold: float = 0.8) -> ModelResponse: """级联聚合：按质量排序，取第一个达标的""" sorted_responses = sorted( responses, key=lambda r: r.confidence, reverse=True ) for resp in sorted_responses: if resp.confidence >= quality_threshold: return resp # 都不达标，返回置信度最高的 return sorted_responses[0] @staticmethod def weighted_merge(responses: List[ModelResponse], weights: Dict[str, float]) -> ModelResponse: """加权融合：按模型权重合并输出""" if not responses: raise ValueError("无响应可聚合") if len(responses) == 1: return responses[0] # 按权重选择主导响应 weighted = [ (resp, weights.get(resp.model_id, 0.5)) for resp in responses ] weighted.sort(key=lambda x: x[1], reverse=True) primary = weighted[0][0] return ModelResponse( model_id=f"aggregated:weighted({','.join(r.model_id for r in responses)})", content=primary.content, latency_ms=max(r.latency_ms for r in responses), token_count=sum(r.token_count for r in responses), cost=sum(r.cost for r in responses), confidence=primary.confidence, ) # ============ 多模型编排引擎 ============ class MultiModelOrchestrator: """ 多模型编排引擎：整合路由、调用和聚合 """ def __init__(self, router: ModelRouter): self._router = router self._aggregator = ResultAggregator() self._model_callers: Dict[str, Callable] = {} self._stats: Dict[str, Dict] = {} def register_caller(self, model_id: str, caller: Callable): """注册模型调用函数""" self._model_callers[model_id] = caller async def execute(self, prompt: str, task_type: Optional[TaskType] = None) -> Dict: """执行多模型编排""" start_time = time.time() # Step 1: 路由 decision = self._router.route(prompt, task_type) # Step 2: 并行调用模型 responses = [] for model_id in decision.selected_models: caller = self._model_callers.get(model_id) if caller: try: resp_start = time.time() content = caller(prompt) resp_latency = (time.time() - resp_start) * 1000 responses.append(ModelResponse( model_id=model_id, content=content, latency_ms=resp_latency, token_count=len(content) // 4, cost=0.01, # 简化 confidence=0.85, )) except Exception as e: responses.append(ModelResponse( model_id=model_id, content=f"Error: {str(e)}", latency_ms=0, token_count=0, cost=0, confidence=0.0, )) # Step 3: 聚合结果 if len(responses) > 1: final = self._aggregator.cascade(responses) elif responses: final = responses[0] else: return {"status": "error", "message": "所有模型调用失败"} total_time = (time.time() - start_time) * 1000 # 记录统计 self._update_stats(decision, final, total_time) return { "status": "success", "content": final.content, "model_used": final.model_id, "task_type": decision.task_type.value, "total_latency_ms": total_time, "cost": final.cost, "confidence": final.confidence, "routing_reason": decision.routing_reason, } def _update_stats(self, decision: RoutingDecision, response: ModelResponse, total_time: float): """更新统计信息""" for model_id in decision.selected_models: if model_id not in self._stats: self._stats[model_id] = { "call_count": 0, "total_cost": 0.0, "total_latency": 0.0, } self._stats[model_id]["call_count"] += 1 self._stats[model_id]["total_cost"] += response.cost self._stats[model_id]["total_latency"] += total_time def get_stats(self) -> Dict: """获取编排统计""" return self._stats四、多模型协作的 Trade-offs
路由准确率与延迟的矛盾。意图分类器的准确率直接影响路由质量。轻量规则引擎延迟低（< 5ms）但准确率有限（约 85%），LLM 分类器准确率高（约 95%）但增加 200-500ms 延迟。建议用规则引擎做初筛，低置信度时再调用 LLM 分类器。
并行调用的成本倍增。多模型并行调用虽然降低了延迟，但成本是所有模型的总和。3 个模型并行意味着 3 倍成本。建议对非关键路径的模型使用异步调用（先返回主模型结果，后台补充验证），或仅在高价值请求上启用多模型。
结果聚合的语义对齐。不同模型对同一提示的输出格式和风格可能不同，直接投票或合并可能产生语义冲突。例如数学题的答案格式不一致（"3" vs "3.0" vs "三"），需要标准化后再聚合。
模型可用性的级联故障。依赖多个模型时，任一模型不可用都会影响整体。需要为每个模型设置超时和降级策略：主模型超时后切换到备选模型，而非等待所有模型响应。
五、总结
AI Agent 多模型协作通过模型路由、并行调用和结果聚合三个环节，将请求分配到最合适的模型并合并输出。路由器基于任务分类和成本预算选择模型组合，聚合器支持投票、级联和加权三种策略。关键权衡在于路由准确率与延迟、并行调用的成本倍增、结果聚合的语义对齐，以及模型可用性的级联故障。多模型协作的目标不是用更多模型，而是让每个请求以最低成本获得最高质量的结果。