Spring AI多模型路由与动态切换的完整实现_Java

前置知识

spring ai基础配置
了解不同llm模型的特点与价格
spring aop与route界定
负载均衡基本概念

核心概念

在生产环境中，单一模型往往无法满足所有场景需求。多模型路由允许根据请求特征（复杂度、类型、用户等级、延迟要求）选择最合适的模型，同时实现容灾、降级和成本控制。

多模型路由架构

请求 → 路由策略评估 → 模型选择 → api调用
                                    ↓
                            ← 失败 → 降级到备用模型

完整实现

1. 多模型chatmodel注册

@configuration
public class multimodelconfig {

    /**
     * 主力模型 - 通义千问plus
     */
    @bean("qwempluschatmodel")
    @primary
    public chatmodel qwempluschatmodel() {
        dashscopeapi api = new dashscopeapi(
            system.getenv("dashscope_api_key")
        );
        
        dashscopechatoptions defaultoptions = dashscopechatoptions.builder()
                .withmodel("qwen-plus")
                .withtemperature(0.7)
                .withmaxtokens(4096)
                .build();
        
        return new dashscopechatmodel(api, defaultoptions);
    }

    /**
     * 快速模型 - qwen turbo (低成本快速响应)
     */
    @bean("qwenturbochatmodel")
    public chatmodel qwenturbochatmodel() {
        dashscopeapi api = new dashscopeapi(
            system.getenv("dashscope_api_key")
        );
        
        dashscopechatoptions defaultoptions = dashscopechatoptions.builder()
                .withmodel("qwen-turbo")
                .withtemperature(0.7)
                .withmaxtokens(2048)
                .build();
        
        return new dashscopechatmodel(api, defaultoptions);
    }

    /**
     * 代码模型 - qwen coder
     */
    @bean("qwencoderchatmodel")
    public chatmodel qwencoderchatmodel() {
        dashscopeapi api = new dashscopeapi(
            system.getenv("dashscope_api_key")
        );
        
        dashscopechatoptions defaultoptions = dashscopechatoptions.builder()
                .withmodel("qwen-coder")
                .withtemperature(0.2)
                .withmaxtokens(4096)
                .build();
        
        return new dashscopechatmodel(api, defaultoptions);
    }

    /**
     * 备用模型 - deepseek
     */
    @bean("deepseekchatmodel")
    public chatmodel deepseekchatmodel() {
        openaiapi api = new openaiapi(
            "https://api.deepseek.com",
            system.getenv("deepseek_api_key")
        );
        
        openaichatoptions defaultoptions = openaichatoptions.builder()
                .withmodel("deepseek-chat")
                .withtemperature(0.7)
                .withmaxtokens(4096)
                .build();
        
        return new openaichatmodel(api, defaultoptions);
    }

    /**
     * 模型注册表
     */
    @bean
    public modelrouter modelrouter(
            @qualifier("qwempluschatmodel") chatmodel qwemplus,
            @qualifier("qwenturbochatmodel") chatmodel qwenturbo,
            @qualifier("qwencoderchatmodel") chatmodel qwencoder,
            @qualifier("deepseekchatmodel") chatmodel deepseek) {
        
        map<string, chatmodel> models = map.of(
            "qwen-plus", qwemplus,
            "qwen-turbo", qwenturbo,
            "qwen-coder", qwencoder,
            "deepseek-chat", deepseek
        );
        
        return new modelrouter(models);
    }
}

2. 模型路由核心逻辑

@slf4j
@component
public class modelrouter {

    private final map<string, chatmodel> models;
    private final atomicreference<string> currentprimarymodel;
    private final map<string, modelmetrics> metrics = new concurrenthashmap<>();

    public modelrouter(map<string, chatmodel> models) {
        this.models = models;
        this.currentprimarymodel = new atomicreference<>("qwen-plus");
        
        // 初始化指标
        models.keyset().foreach(name -> 
            metrics.put(name, new modelmetrics()));
    }

    /**
     * 路由策略: 根据请求内容选择模型
     */
    public chatmodel route(routecontext context) {
        // 1. 用户指定模型
        if (context.preferredmodel() != null 
            && models.containskey(context.preferredmodel())) {
            string preferred = context.preferredmodel();
            if (!metrics.get(preferred).iscircuitopen()) {
                return models.get(preferred);
            }
            log.warn("指定模型 {} 熔断中，尝试降级", preferred);
        }

        // 2. 根据内容类型路由
        if (context.iscoderequest()) {
            return routewithfallback("qwen-coder", "qwen-plus");
        }
        
        if (context.iscreativerequest()) {
            return routewithfallback("qwen-plus", "deepseek-chat");
        }
        
        if (context.isfactualrequest()) {
            return routewithfallback("qwen-plus", "deepseek-chat");
        }

        // 3. 根据复杂度路由
        if (context.complexity() < 3) {
            return routewithfallback("qwen-turbo", "qwen-plus");
        }

        // 4. 默认主力模型
        return routewithfallback(currentprimarymodel.get(), "deepseek-chat");
    }

    /**
     * 带故障转移的路由
     */
    private chatmodel routewithfallback(string primary, string fallback) {
        modelmetrics primarymetrics = metrics.get(primary);
        
        if (primarymetrics != null && primarymetrics.iscircuitopen()) {
            log.warn("模型 {} 已熔断，使用备用模型 {}", primary, fallback);
            
            modelmetrics fallbackmetrics = metrics.get(fallback);
            if (fallbackmetrics != null && fallbackmetrics.iscircuitopen()) {
                // 都熔断，选择失败率最低的
                return getleastfailingmodel();
            }
            return models.get(fallback);
        }
        
        chatmodel model = models.get(primary);
        if (model == null) {
            return models.get(fallback);
        }
        return model;
    }

    /**
     * 选择失败率最低的模型
     */
    private chatmodel getleastfailingmodel() {
        return metrics.entryset().stream()
                .min(comparator.comparingdouble(e -> e.getvalue().geterrorrate()))
                .map(e -> models.get(e.getkey()))
                .orelse(models.values().iterator().next());
    }

    /**
     * 动态修改当前主力模型
     */
    public void switchprimarymodel(string modelname) {
        if (models.containskey(modelname)) {
            string oldmodel = currentprimarymodel.getandset(modelname);
            log.info("模型切换: {} -> {}", oldmodel, modelname);
        }
    }

    /**
     * 记录成功调用
     */
    public void recordsuccess(string modelname, long durationms) {
        modelmetrics m = metrics.get(modelname);
        if (m != null) {
            m.recordsuccess(durationms);
        }
    }

    /**
     * 记录失败调用
     */
    public void recordfailure(string modelname, string errortype) {
        modelmetrics m = metrics.get(modelname);
        if (m != null) {
            m.recordfailure(errortype);
        }
    }

    /**
     * 获取模型健康状态
     */
    public map<string, modelhealth> gethealthstatus() {
        map<string, modelhealth> status = new hashmap<>();
        metrics.foreach((name, m) -> {
            status.put(name, new modelhealth(
                name,
                m.iscircuitopen() ? "circuit_open" : m.geterrorrate() > 0.1 
                    ? "degraded" : "healthy",
                m.gettotalcalls(),
                m.geterrorrate(),
                m.getavglatencyms()
            ));
        });
        return status;
    }
}

/**
 * 路由上下文
 */
public record routecontext(
    string usermessage,
    string userid,
    string conversationid,
    string preferredmodel,
    int complexity
) {
    public boolean iscoderequest() {
        if (usermessage == null) return false;
        return usermessage.matches("(?i).*(代码|code|function|class|算法|编程|bug|debug|实现).*")
            || usermessage.contains("```");
    }

    public boolean iscreativerequest() {
        if (usermessage == null) return false;
        return usermessage.matches("(?i).*(创意|故事|创作|写诗|小说|营销|广告|社交媒体|romantic).*");
    }

    public boolean isfactualrequest() {
        if (usermessage == null) return false;
        return usermessage.matches("(?i).*(是什么|解释|定义|原理|科学|研究|事实|真理).*");
    }
}

3. 模型指标与熔断器

/**
 * 模型调用指标 - 滑动窗口计数器
 */
@slf4j
public class modelmetrics {

    private final atomicinteger totalcalls = new atomicinteger(0);
    private final atomicinteger errorcalls = new atomicinteger(0);
    private final concurrentlinkeddeque<long> latentcalls = 
        new concurrentlinkeddeque<>();
    private final concurrentlinkeddeque<boolean> recentresults = 
        new concurrentlinkeddeque<>();
    
    private volatile boolean circuitopen = false;
    private volatile long circuitopentime = 0;
    
    private static final int window_size = 100;
    private static final double circuit_open_threshold = 0.5;
    private static final long circuit_open_duration_ms = 30_000; // 30秒后尝试半开

    public void recordsuccess(long durationms) {
        totalcalls.incrementandget();
        latentcalls.add(durationms);
        recentresults.add(true);
        maintainwindow();
        
        // 恢复熔断
        if (circuitopen && system.currenttimemillis() > circuitopentime 
                + circuit_open_duration_ms) {
            circuitopen = false;
            log.info("模型熔断恢复，进入半开状态");
        }
    }

    public void recordfailure(string errortype) {
        totalcalls.incrementandget();
        errorcalls.incrementandget();
        recentresults.add(false);
        maintainwindow();
        
        // 检查熔断阈值
        if (geterrorrate() >= circuit_open_threshold 
                && totalcalls.get() >= 10) {
            circuitopen = true;
            circuitopentime = system.currenttimemillis();
            log.warn("模型熔断开启，错误率: {:.2%}", geterrorrate());
        }
    }

    public boolean iscircuitopen() {
        if (!circuitopen) return false;
        
        // 熔断时间窗口过后进入半开状态，允许一个请求尝试
        if (system.currenttimemillis() > circuitopentime + circuit_open_duration_ms) {
            return false; // 半开状态
        }
        return true;
    }

    public double geterrorrate() {
        int total = totalcalls.get();
        if (total == 0) return 0.0;
        return (double) errorcalls.get() / total;
    }

    public double getavglatencyms() {
        if (latentcalls.isempty()) return 0;
        return latentcalls.stream()
                .maptolong(long::longvalue)
                .average()
                .orelse(0);
    }

    public int gettotalcalls() {
        return totalcalls.get();
    }

    private void maintainwindow() {
        if (recentresults.size() > window_size) {
            recentresults.pollfirst();
        }
        while (latentcalls.size() > window_size) {
            latentcalls.pollfirst();
        }
    }

    public double getrecenterrorrate() {
        if (recentresults.isempty()) return 0;
        long failcount = recentresults.stream().filter(r -> !r).count();
        return (double) failcount / recentresults.size();
    }
}

public record modelhealth(
    string modelname,
    string status,
    int totalcalls,
    double errorrate,
    double avglatencyms
) {}

4. 动态模型切换的aop实现

/**
 * 路由拦截器 - 自动为chatclient调用选择模型
 */
@aspect
@component
@slf4j
public class modelrouteaspect {

    private final modelrouter router;
    private final meterregistry meterregistry;

    public modelrouteaspect(modelrouter router, meterregistry meterregistry) {
        this.router = router;
        this.meterregistry = meterregistry;
    }

    @around("execution(* com.example.ai.service.*.*(..)) && " +
            "@annotation(routetomodel)")
    public object routemodel(proceedingjoinpoint pjp, 
                              routetomodel routetomodel) throws throwable {
        routecontext context = extractcontext(pjp.getargs());
        chatmodel selected = router.route(context);
        string modelname = getmodelname(selected);
        
        transaction transaction = cat.newtransaction("llm", modelname);
        long starttime = system.currenttimemillis();
        
        try {
            object result = pjp.proceed();
            long duration = system.currenttimemillis() - starttime;
            
            router.recordsuccess(modelname, duration);
            meterregistry.timer("llm.route.success")
                    .tag("model", modelname)
                    .record(duration, timeunit.milliseconds);
            transaction.setstatus(transaction.success);
            
            return result;
        } catch (exception e) {
            long duration = system.currenttimemillis() - starttime;
            router.recordfailure(modelname, e.getclass().getsimplename());
            meterregistry.counter("llm.route.failure")
                    .tag("model", modelname)
                    .tag("error", e.getclass().getsimplename())
                    .increment();
            transaction.setstatus(e);
            throw e;
        } finally {
            transaction.complete();
        }
    }

    private routecontext extractcontext(object[] args) {
        // 从方法参数构造路由上下文
        if (args.length > 0 && args[0] instanceof chatrequest request) {
            return new routecontext(
                request.message(),
                request.userid(),
                request.conversationid(),
                request.preferredmodel(),
                estimatecomplexity(request.message())
            );
        }
        return new routecontext("", "", null, null, 5);
    }

    private int estimatecomplexity(string message) {
        if (message == null) return 5;
        int score = 5;
        if (message.length() > 500) score += 2;
        if (message.contains("复杂") || message.contains("详细")) score += 2;
        if (message.contains("简洁") || message.contains("简短")) score -= 2;
        return math.max(1, math.min(10, score));
    }

    private string getmodelname(chatmodel model) {
        try {
            field field = model.getclass().getdeclaredfield("chatoptions");
            field.setaccessible(true);
            chatoptions options = (chatoptions) field.get(model);
            return options != null ? options.getmodel() : "unknown";
        } catch (exception e) {
            return "unknown";
        }
    }
}

@target(elementtype.method)
@retention(retentionpolicy.runtime)
public @interface routetomodel {
}

5. 基于使用量的模型调度

@service
public class quotabasedscheduler {

    private final modelrouter router;
    private final map<string, atomicinteger> dailyusage = new concurrenthashmap<>();
    private final map<string, integer> dailylimits = map.of(
        "qwen-turbo", 10000,
        "qwen-plus", 5000,
        "qwen-coder", 3000,
        "deepseek-chat", 2000
    );

    public quotabasedscheduler(modelrouter router) {
        this.router = router;
        
        // 每天重置配额
        scheduledexecutorservice scheduler = executors.newsinglethreadscheduledexecutor();
        scheduler.scheduleatfixedrate(
            dailyusage::clear,
            getsecondsuntilmidnight(), 
            timeunit.days.toseconds(1),
            timeunit.seconds
        );
    }

    /**
     * 配额感知的模型路由
     */
    public chatmodel routewithquota(routecontext context) {
        chatmodel preferred = router.route(context);
        string modelname = getmodelname(preferred);
        
        atomicinteger usage = dailyusage.computeifabsent(
            modelname, k -> new atomicinteger(0));
        int limit = dailylimits.getordefault(modelname, 1000);
        
        if (usage.incrementandget() > limit) {
            log.warn("模型 {} 配额已用尽 ({}), 尝试降级", modelname, limit);
            // 选择还有配额的模型
            return getnextavailablemodel(modelname);
        }
        
        return preferred;
    }

    private chatmodel getnextavailablemodel(string exhaustedmodel) {
        for (map.entry<string, integer> entry : dailylimits.entryset()) {
            if (entry.getkey().equals(exhaustedmodel)) continue;
            
            atomicinteger usage = dailyusage.computeifabsent(
                entry.getkey(), k -> new atomicinteger(0));
            
            if (usage.get() < entry.getvalue()) {
                return router.getmodel(entry.getkey());
            }
        }
        
        // 全部配额用尽
        log.error("所有模型配额均已用尽");
        return null;
    }

    private long getsecondsuntilmidnight() {
        localdatetime now = localdatetime.now();
        localdatetime midnight = now.tolocaldate().plusdays(1).atstartofday();
        return duration.between(now, midnight).getseconds();
    }

    /**
     * 获取当前配额使用情况
     */
    public map<string, quotastatus> getquotastatus() {
        map<string, quotastatus> status = new hashmap<>();
        dailylimits.foreach((model, limit) -> {
            int used = dailyusage.getordefault(model, new atomicinteger(0)).get();
            status.put(model, new quotastatus(model, used, limit, limit - used));
        });
        return status;
    }

    public record quotastatus(
        string model, int used, int limit, int remaining
    ) {}
}

6. 模型响应质量评估与自动切换

@service
public class qualityawarerouter {

    private final modelrouter router;
    private final map<string, double> qualityscores = new concurrenthashmap<>();

    public qualityawarerouter(modelrouter router) {
        this.router = router;
        qualityscores.put("qwen-turbo", 0.7);
        qualityscores.put("qwen-plus", 0.85);
        qualityscores.put("qwen-coder", 0.9);
        qualityscores.put("deepseek-chat", 0.8);
    }

    /**
     * 反馈驱动的质量评估
     */
    public void recordfeedback(string modelname, double score) {
        // 指数移动平均
        double current = qualityscores.getordefault(modelname, 0.5);
        double updated = current * 0.9 + score * 0.1;
        qualityscores.put(modelname, updated);
    }

    /**
     * 结合质量分数和当前路由策略
     */
    public chatmodel routewithquality(routecontext context) {
        chatmodel candidates = router.route(context);
        double score = qualityscores.getordefault(getmodelname(candidates), 0.5);
        
        // 如果质量分数低于阈值，尝试使用更高的质量模型
        if (score < 0.6) {
            log.info("当前模型质量分数 {} 偏低，切换高分模型", score);
            return gethighestqualitymodel();
        }
        
        return candidates;
    }

    private chatmodel gethighestqualitymodel() {
        return qualityscores.entryset().stream()
                .max(map.entry.comparingbyvalue())
                .map(e -> router.getmodel(e.getkey()))
                .orelse(null);
    }
}