CASTER上下文感知模型路由 — 任务步骤级强弱LLM动态调度以破解性能-成本铁三角
Skill-CASTER-Context-Aware-Model-Routing · 10-MAS
1. 解决的问题
用同一个昂贵LLM处理所有Agent任务导致成本失控——CASTER步骤级神经路由将75%低复杂度步骤路由到弱模型，在保持质量的同时降低23-54%成本（2026 arXiv:2601.19793）
2. 核心算法逻辑
反直觉洞察：Denis Rothman书中默认用同一个模型（GPT5.1/5.4）处理所有Agent任务。但跨境电商MAS中，"帮我写一个商品标题"（简单）和"分析这个产品的法规合规风险"（复杂）不应该用同一个昂贵模型。CASTER的关键发现：MAS工作流中75%的子任务可以用弱模型（GPT4omini）完成，节省60%+成本，而不影响整体质量。关键是"动态决策哪步用强模型"，而非静态分配。
3. 业务应用场景
- 业务问题：某母婴品牌的选品MAS每月处理500次完整分析，每次包含20个步骤（数据收集、竞品对比、合规检查、财务建模、报告生成）。全部用GPT-4o，月成本$390；全部用GPT-4o-mini，质量下降40%不可接受 - CASTER路由策略： - 数据格式化/标准化步骤 → GPT-4o-mini（规则性强） - 竞品搜索/摘要步骤 → GPT-4o-mini（信息提取，低复杂度） - 法规合规分析步骤 → GPT-4o（需要专业推理） - 财务建模/预测步骤 → GPT-4o（定量推理） - 最终策略建议步骤 → GPT-4o（关键决策） - 结果：约65%的步骤用mini模型，月成
- **业务问题**：大促期间MAS需要实时响应（<2秒），但GPT-4o延迟高（平均3-5秒） - **CASTER方案**：识别时间敏感步骤（实时监控/告警）→ 强制路由到GPT-4o-mini（<1秒响应），分析性步骤允许等待强模型。平均延迟从3.2秒降至1.4秒，满足实时要求
4. 输入数据要求
请查看原始代码模板获取输入规格。
5. 输出结果
请查看原始代码模板获取输出规格。
6. 业务价值 / ROI
- ROI 预估：月调用500次完整MAS分析，平均每次20步骤，CASTER节省40%模型成本，年化节省约$2000-5000；同时关键步骤（合规/财务）用强模型保证质量；系统成本$3万，ROI≈600-1500%
- 实施难度：⭐⭐⭐☆☆（规则基础版本容易实现；学习路由器需要额外训练数据；关键是设计好每步骤的复杂度评估特征）
- 优先级：⭐⭐⭐⭐⭐（成本是MAS生产部署的最大障碍，CASTER直接攻克这个问题，2026年最新结果，论文已开源）
- 适用规模：有10+步骤的工作流，且包含不同复杂度的混合任务（既有格式化又有推理）
- 数据依赖：需要历史步骤质量记录来训练路由策略；冷启动用规则基础版，积累数据后升级为学习版
7. 代码模板
代码块数量：3 · 路径：未检测到
"""
CASTER上下文感知模型路由系统
功能：步骤级模型路由 + 上下文状态追踪 + 成本-质量优化
基于 arXiv:2601.19793 (2026)
"""
import numpy as np
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple, Any
from enum import Enum
import warnings
warnings.filterwarnings('ignore')
class ModelTier(Enum):
STRONG = "strong" # GPT-4o, Claude-3-Opus
WEAK = "weak" # GPT-4o-mini, Claude-3-Haiku
@dataclass
class ModelConfig:
"""模型配置"""
name: str
tier: ModelTier
cost_per_1k_tokens: float # USD/1K tokens
avg_latency_ms: float # 平均延迟
reasoning_score: float # 推理能力评分 0-1
@dataclass
class WorkflowStep:
"""MAS工作流步骤"""
step_id: str
description: str
agent_role: str
# 复杂度特征
requires_reasoning: bool = False # 是否需要深度推理
is_format_task: bool = False # 是否是格式化/结构化任务
has_error_risk: bool = False # 是否有错误风险（前步失败可能影响此步）
is_time_sensitive: bool = False # 是否对延迟敏感
complexity_score: float = 0.5 # 综合复杂度 0-1
@dataclass
class ContextState:
"""上下文状态（在步骤间传递）"""
accumulated_errors: int = 0
task_progress: float = 0.0 # 0-1
remaining_budget_usd: float = 1.0
quality_signals: List[float] = field(default_factory=list)
latency_budget_ms: Optional[float] = None
@property
def error_rate(self) -> float:
total_steps = max(len(self.quality_signals), 1)
return self.accumulated_errors / total_steps
@property
def avg_quality(self) -> float:
return np.mean(self.quality_signals) if self.quality_signals else 0.8
8. 论文来源
- 2601.19793