第1772篇：大模型调用的智能路由——根据任务类型选择最优模型与最低成本

上周一个在做SaaS产品的朋友跟我说，他们把所有AI功能都用的GPT-4o，一个月账单下来快两万美元了，老板让他想办法降成本。我问他："你们所有任务都需要GPT-4o的能力吗？"他沉默了一下说："好像不是。"

这就是问题所在。

大部分团队在早期选模型时图省事，直接上最强的，效果确实好，但钱也烧得厉害。等业务量上来，才发现成本根本撑不住。

其实70%以上的AI任务，用能力弱但便宜得多的模型就完全够用。智能路由干的就是这件事：在保证质量的前提下，自动把任务路由到最合适（通常也是最便宜）的模型。

这篇文章，我来聊聊智能路由的设计思路和落地实现。

---

模型成本差异有多大

先建立一个直觉。以我写这篇文章时的主流模型为例：

模型	输入价格/1K token	输出价格/1K token	能力定位
GPT-4o	$0.005	$0.015	旗舰，强推理
GPT-4o-mini	$0.00015	$0.0006	轻量，快速响应
Claude 3.5 Sonnet	$0.003	$0.015	旗舰，长文本强
Claude 3 Haiku	$0.00025	$0.00125	轻量，极速
Gemini 1.5 Flash	$0.000075	$0.0003	超轻量，高吞吐

GPT-4o 和 GPT-4o-mini，输入价格相差33倍，输出相差25倍。如果一个分类任务用 gpt-4o-mini 就能做好，用 GPT-4o 就是纯粹在浪费钱。

---

智能路由的三种策略

路由策略从简单到复杂，分三种：

规则路由是最简单的，适合任务类型明确的场景。"文本分类用 mini，代码生成用旗舰模型"，直接写死逻辑。

成本路由稍微复杂，先定义每类任务的质量下限，然后从便宜的模型开始尝试，满足质量要求就停，不满足再升级到更贵的模型。

质量感知路由是最复杂也最智能的，它能动态感知任务难度，结合历史数据预测每个模型在当前任务上的成功概率，做出最优决策。

实际项目里，我建议从规则路由开始，积累数据后再升级到成本路由，大规模后再考虑质量感知路由。别一上来就搞最复杂的。

---

规则路由的实现

核心是建立任务类型到模型的映射关系，加上一些特征提取逻辑。

@Component
public class RuleBasedRouter {
    
    // 任务类型枚举
    public enum TaskType {
        TEXT_CLASSIFICATION,     // 文本分类
        SENTIMENT_ANALYSIS,      // 情感分析
        SIMPLE_QA,               // 简单问答
        TEXT_SUMMARIZATION,      // 文本摘要
        CODE_GENERATION,         // 代码生成
        CODE_REVIEW,             // 代码审查
        COMPLEX_REASONING,       // 复杂推理
        CREATIVE_WRITING,        // 创意写作
        DATA_EXTRACTION,         // 数据抽取
        TRANSLATION              // 翻译
    }
    
    // 任务类型 -> 推荐模型
    private static final Map<TaskType, String> TASK_MODEL_MAP = Map.of(
        TaskType.TEXT_CLASSIFICATION, "gpt-4o-mini",
        TaskType.SENTIMENT_ANALYSIS,  "gpt-4o-mini",
        TaskType.SIMPLE_QA,           "gpt-4o-mini",
        TaskType.TEXT_SUMMARIZATION,  "claude-3-haiku-20240307",
        TaskType.TRANSLATION,         "gpt-4o-mini",
        TaskType.DATA_EXTRACTION,     "gpt-4o-mini",
        TaskType.CODE_GENERATION,     "claude-3-5-sonnet-20241022",
        TaskType.CODE_REVIEW,         "gpt-4o",
        TaskType.COMPLEX_REASONING,   "gpt-4o",
        TaskType.CREATIVE_WRITING,    "claude-3-5-sonnet-20241022"
    );
    
    /**
     * 根据任务类型返回推荐模型
     */
    public String route(TaskType taskType) {
        return TASK_MODEL_MAP.getOrDefault(taskType, "gpt-4o-mini");
    }
    
    /**
     * 从请求特征推断任务类型
     */
    public TaskType inferTaskType(RoutingRequest request) {
        // 1. 业务方显式标注的类型优先级最高
        if (request.getExplicitTaskType() != null) {
            return request.getExplicitTaskType();
        }
        
        // 2. 根据功能代码映射
        TaskType byFeature = inferByFeatureCode(request.getFeatureCode());
        if (byFeature != null) {
            return byFeature;
        }
        
        // 3. 根据Prompt特征推断
        return inferByPromptFeatures(request.getSystemPrompt(), request.getUserMessage());
    }
    
    private TaskType inferByFeatureCode(String featureCode) {
        if (featureCode == null) return null;
        if (featureCode.contains("classify")) return TaskType.TEXT_CLASSIFICATION;
        if (featureCode.contains("sentiment")) return TaskType.SENTIMENT_ANALYSIS;
        if (featureCode.contains("code")) return TaskType.CODE_GENERATION;
        if (featureCode.contains("summary")) return TaskType.TEXT_SUMMARIZATION;
        if (featureCode.contains("translate")) return TaskType.TRANSLATION;
        return null;
    }
    
    private TaskType inferByPromptFeatures(String systemPrompt, String userMessage) {
        String combined = (systemPrompt + " " + userMessage).toLowerCase();
        
        if (combined.contains("分类") || combined.contains("classify") || combined.contains("categorize")) {
            return TaskType.TEXT_CLASSIFICATION;
        }
        if (combined.contains("代码") || combined.contains("code") || combined.contains("function")) {
            return TaskType.CODE_GENERATION;
        }
        if (combined.contains("摘要") || combined.contains("summary") || combined.contains("summarize")) {
            return TaskType.TEXT_SUMMARIZATION;
        }
        if (combined.contains("翻译") || combined.contains("translate")) {
            return TaskType.TRANSLATION;
        }
        
        // 默认轻量模型
        return TaskType.SIMPLE_QA;
    }
}

---

基于质量下限的成本路由

这个策略更实用。核心逻辑是：先尝试最便宜的模型，检验结果是否达到质量标准，不达标再升级。

@Service
@Slf4j
public class CostOptimizedRouter {
    
    @Autowired
    private ModelRegistry modelRegistry;
    
    @Autowired
    private QualityEvaluator qualityEvaluator;
    
    @Autowired
    private AIClientFactory clientFactory;
    
    /**
     * 路由链：从便宜到贵，直到满足质量要求
     */
    public RouterResult routeWithFallback(RoutingRequest request) {
        List<String> modelCandidates = buildModelChain(request);
        
        for (int i = 0; i < modelCandidates.size(); i++) {
            String modelName = modelCandidates.get(i);
            
            try {
                // 调用当前模型
                AIResponse response = callModel(modelName, request);
                
                // 评估结果质量
                QualityScore score = qualityEvaluator.evaluate(request, response);
                
                if (score.meetsThreshold(request.getMinQualityScore())) {
                    log.info("路由成功: task={}, model={}, quality={}, attempt={}", 
                        request.getTaskType(), modelName, score.getScore(), i + 1);
                    
                    return RouterResult.success(response, modelName, i + 1, score);
                } else {
                    log.info("质量不达标，升级模型: task={}, model={}, quality={}/{}", 
                        request.getTaskType(), modelName, 
                        score.getScore(), request.getMinQualityScore());
                }
                
            } catch (Exception e) {
                log.error("模型调用失败: model={}, error={}", modelName, e.getMessage());
                // 调用失败也升级到下一个模型
            }
        }
        
        // 所有模型都不满足，返回最后一次的结果
        return RouterResult.fallback("all models below threshold");
    }
    
    /**
     * 构建模型候选链（从便宜到贵）
     */
    private List<String> buildModelChain(RoutingRequest request) {
        // 可以根据任务类型、请求特征定制候选链
        if (request.getTaskType() == TaskType.CODE_GENERATION) {
            // 代码任务：直接用强模型，mini模型代码质量差
            return List.of("claude-3-5-sonnet-20241022", "gpt-4o");
        }
        
        if (request.getInputTokenCount() > 50000) {
            // 长文本：Haiku和mini处理长文档性价比高
            return List.of("claude-3-haiku-20240307", "claude-3-5-sonnet-20241022");
        }
        
        // 通用任务：从mini到旗舰
        return List.of("gpt-4o-mini", "gpt-4o");
    }
}

质量评估器

质量评估是这套方案的核心难点。不同任务类型的质量标准完全不同。

@Component
public class QualityEvaluator {
    
    /**
     * 根据任务类型选择合适的评估方式
     */
    public QualityScore evaluate(RoutingRequest request, AIResponse response) {
        return switch (request.getTaskType()) {
            case TEXT_CLASSIFICATION -> evaluateClassification(request, response);
            case DATA_EXTRACTION -> evaluateExtraction(request, response);
            case SIMPLE_QA -> evaluateCompleteness(response);
            default -> evaluateGeneral(response);
        };
    }
    
    /**
     * 分类任务：检查输出是否是预期分类之一
     */
    private QualityScore evaluateClassification(RoutingRequest request, AIResponse response) {
        Set<String> validCategories = request.getExpectedCategories();
        String output = response.getContent().trim().toLowerCase();
        
        boolean isValid = validCategories.stream()
            .anyMatch(cat -> output.contains(cat.toLowerCase()));
        
        // 还要检查是否有幻觉（输出了不在列表里的类别）
        long invalidCount = Arrays.stream(output.split("[,，\n]"))
            .map(String::trim)
            .filter(s -> !s.isEmpty())
            .filter(s -> validCategories.stream().noneMatch(cat -> s.contains(cat.toLowerCase())))
            .count();
        
        double score = isValid ? (invalidCount == 0 ? 1.0 : 0.7) : 0.0;
        return QualityScore.of(score);
    }
    
    /**
     * 数据抽取：检查JSON格式是否合法，必填字段是否存在
     */
    private QualityScore evaluateExtraction(RoutingRequest request, AIResponse response) {
        try {
            JSONObject result = JSON.parseObject(response.getContent());
            
            // 检查必填字段
            List<String> requiredFields = request.getRequiredFields();
            long missingCount = requiredFields.stream()
                .filter(field -> !result.containsKey(field) || result.get(field) == null)
                .count();
            
            double score = 1.0 - (double) missingCount / requiredFields.size();
            return QualityScore.of(score);
            
        } catch (Exception e) {
            // JSON解析失败，质量为0
            return QualityScore.of(0.0);
        }
    }
    
    /**
     * 通用评估：基于响应完整性和长度合理性
     */
    private QualityScore evaluateGeneral(AIResponse response) {
        String content = response.getContent();
        
        if (content == null || content.trim().isEmpty()) {
            return QualityScore.of(0.0);
        }
        
        // 检查是否有明显的拒绝/错误回复
        String lower = content.toLowerCase();
        if (lower.contains("i cannot") || lower.contains("i'm unable") 
                || lower.contains("我无法") || lower.contains("抱歉，我不能")) {
            return QualityScore.of(0.2);
        }
        
        // 基础完整性得分（可以根据业务需要扩展）
        return QualityScore.of(0.8);
    }
    
    private QualityScore evaluateCompleteness(AIResponse response) {
        return evaluateGeneral(response);
    }
}

---

基于历史数据的智能路由

积累了一段时间数据之后，可以用历史成功率来预测路由决策，这是最智能也最实用的方式。

@Service
public class HistoryBasedRouter {
    
    @Autowired
    private RoutingHistoryRepository historyRepo;
    
    @Autowired
    private ModelPriceConfig priceConfig;
    
    /**
     * 基于历史胜率选择最佳模型
     * 目标：在满足质量要求的前提下最小化期望成本
     */
    public String selectOptimalModel(RoutingRequest request) {
        List<String> candidates = getCandidateModels(request.getTaskType());
        
        String bestModel = null;
        double bestScore = Double.MAX_VALUE;
        
        for (String modelName : candidates) {
            // 历史成功率
            double successRate = historyRepo.getSuccessRate(
                modelName, request.getTaskType(), 30  // 最近30天
            );
            
            // 如果成功率低于最低阈值，跳过
            if (successRate < 0.7) continue;
            
            // 预期成本 = 单次成本 / 成功率（失败需要重试）
            double estimatedCostPerSuccess = estimateCost(modelName, request) / successRate;
            
            if (estimatedCostPerSuccess < bestScore) {
                bestScore = estimatedCostPerSuccess;
                bestModel = modelName;
            }
        }
        
        return bestModel != null ? bestModel : getDefaultModel(request.getTaskType());
    }
    
    private double estimateCost(String modelName, RoutingRequest request) {
        ModelPrice price = priceConfig.getPrice(modelName);
        if (price == null) return Double.MAX_VALUE;
        
        // 估算输出token（输出通常是输入的0.3-1倍，根据任务类型调整）
        double outputRatio = getOutputRatio(request.getTaskType());
        int estimatedOutputTokens = (int)(request.getInputTokenCount() * outputRatio);
        
        return price.getInputPricePerK().doubleValue() * request.getInputTokenCount() / 1000
             + price.getOutputPricePerK().doubleValue() * estimatedOutputTokens / 1000;
    }
    
    private double getOutputRatio(TaskType taskType) {
        return switch (taskType) {
            case TEXT_CLASSIFICATION -> 0.05;   // 分类输出极短
            case SENTIMENT_ANALYSIS -> 0.05;
            case DATA_EXTRACTION -> 0.3;
            case TEXT_SUMMARIZATION -> 0.2;
            case CODE_GENERATION -> 2.0;         // 代码输出可能很长
            case CREATIVE_WRITING -> 1.5;
            default -> 0.5;
        };
    }
}

---

路由中间件：把所有策略串起来

实际系统里，不是用一种策略，而是多种策略组合。用责任链模式来组织很合适。

@Component
public class IntelligentRoutingGateway {
    
    @Autowired
    private RuleBasedRouter ruleRouter;
    
    @Autowired
    private HistoryBasedRouter historyRouter;
    
    @Autowired
    private CostOptimizedRouter costRouter;
    
    @Autowired
    private TokenCostRecorder costRecorder;
    
    /**
     * 统一入口：智能路由后调用模型
     */
    public AIResponse route(RoutingRequest request) {
        // 1. 确定路由策略
        RoutingStrategy strategy = determineStrategy(request);
        
        // 2. 选择模型
        String selectedModel = selectModel(request, strategy);
        
        // 3. 记录路由决策
        log.info("路由决策: taskType={}, strategy={}, model={}", 
            request.getTaskType(), strategy, selectedModel);
        
        // 4. 调用模型
        request.setSelectedModel(selectedModel);
        AIResponse response = callModel(request);
        
        // 5. 记录成本和结果
        recordRoutingResult(request, response, selectedModel);
        
        return response;
    }
    
    private RoutingStrategy determineStrategy(RoutingRequest request) {
        // 紧急任务：直接用规则路由，不做质量检验（响应速度优先）
        if (request.isUrgent()) return RoutingStrategy.RULE_BASED;
        
        // 有足够历史数据：用历史路由
        long historyCount = historyRouter.getHistoryCount(request.getTaskType());
        if (historyCount > 500) return RoutingStrategy.HISTORY_BASED;
        
        // 有明确质量要求：用成本路由（带质量验证）
        if (request.getMinQualityScore() > 0) return RoutingStrategy.COST_OPTIMIZED;
        
        // 默认规则路由
        return RoutingStrategy.RULE_BASED;
    }
    
    private String selectModel(RoutingRequest request, RoutingStrategy strategy) {
        return switch (strategy) {
            case RULE_BASED -> {
                TaskType taskType = ruleRouter.inferTaskType(request);
                yield ruleRouter.route(taskType);
            }
            case HISTORY_BASED -> historyRouter.selectOptimalModel(request);
            case COST_OPTIMIZED -> costRouter.routeWithFallback(request).getModelName();
        };
    }
}

---

路由效果监控

路由系统上线后，需要持续监控效果，防止路由决策退化。

监控指标：

@Component
public class RoutingMetricsCollector {
    
    private final MeterRegistry meterRegistry;
    
    // 路由分发到各模型的比例
    public void recordModelSelection(String taskType, String modelName) {
        meterRegistry.counter("routing.model.selected",
            "task_type", taskType,
            "model", modelName
        ).increment();
    }
    
    // 路由后的质量得分分布
    public void recordQualityScore(String modelName, String taskType, double score) {
        meterRegistry.summary("routing.quality.score",
            "model", modelName,
            "task_type", taskType
        ).record(score);
    }
    
    // 升级路由（从便宜模型升级到贵模型）的比例
    public void recordModelUpgrade(String fromModel, String toModel, String taskType) {
        meterRegistry.counter("routing.model.upgrade",
            "from", fromModel,
            "to", toModel,
            "task_type", taskType
        ).increment();
    }
    
    // 每类任务的平均成本
    public void recordTaskCost(String taskType, String modelName, double costUsd) {
        meterRegistry.summary("routing.task.cost_usd",
            "task_type", taskType,
            "model", modelName
        ).record(costUsd);
    }
}

---

踩过的坑

做这套系统过程中踩了一些坑，记录下来给大家参考。

坑一：质量评估本身引入了成本。 如果要验证一个分类结果对不对，最简单的方式是再调用一次模型来评估。但这个评估本身也要花钱。我们后来改成了基于规则的轻量评估（格式检查、范围检查），加上抽样人工审核，而不是每次都调用模型评估。

坑二：冷启动问题。历史路由需要历史数据，但新功能上线时没有历史数据。我们的做法是新功能前100次强制用"安全"模型（能力强一档的那个），积累数据后再启用智能路由。

坑三：模型能力变化导致历史数据失效。模型提供商会悄悄更新模型版本，比如 gpt-4o-2024-05-13 和 gpt-4o-2024-08-06 能力不同。历史数据记录的是旧版本的成功率，新版本上线后数据就不准了。需要在路由历史里记录模型版本，并有清除过期数据的机制。

坑四：并发场景下的路由一致性。同一个用户的多次请求，如果路由到不同模型，结果可能不一致，用户会觉得系统不稳定。对于需要上下文连续的对话场景，要在路由时考虑"会话粘性"，同一个会话内的请求路由到同一个模型。

---

一套好的智能路由系统，能在不降低用户体验的前提下把AI成本降低40%-60%。这不是小数字，对于每月AI账单过万美元的团队，相当于每年省出几十万的开销。

更重要的是，这套系统会随着数据积累越来越聪明，而不是一次性优化完就结束了。