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 多模型协作通过模型路由、并行调用和结果聚合三个环节，将请求分配到最合适的模型并合并输出。路由器基于任务分类和成本预算选择模型组合，聚合器支持投票、级联和加权三种策略。关键权衡在于路由准确率与延迟、并行调用的成本倍增、结果聚合的语义对齐，以及模型可用性的级联故障。多模型协作的目标不是用更多模型，而是让每个请求以最低成本获得最高质量的结果。
AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。
更多推荐
8
0- 0
已为社区贡献99条内容
所有评论(0)