1、限流架构图和实现流程图

如下展示了从Controller到Redis Lua脚本的完整限流+熔断+降级流程。核心就是Redis Lua脚本的原子限流 + Resilience4j的熔断降级 + 多模型容灾。

┌─────────────────────────────────────────────────────────────────────┐
│                          客户端请求                                  │
│                     POST /api/llm/chat                             │
└─────────────────────────────────────────────────────────────────────┘
                                    │
                                    ▼
┌─────────────────────────────────────────────────────────────────────┐
│                      ChatController                                │
│                   接收请求，调用Gateway                             │
└─────────────────────────────────────────────────────────────────────┘
                                    │
                                    ▼
┌─────────────────────────────────────────────────────────────────────┐
│                    TokenCounter.countInputTokens()                  │
│              计算输入Token数（使用jtokkit）                         │
└─────────────────────────────────────────────────────────────────────┘
                                    │
                                    ▼
┌─────────────────────────────────────────────────────────────────────┐
│              TokenBucketRateLimiter.acquire()                      │
│        调用Redis Lua脚本进行原子限流检查                            │
│  ┌─────────────────────────────────────────────────────────────┐   │
│  │          Redis执行Lua脚本（原子操作）                        │   │
│  │  1. ZREMRANGEBYSCORE 清理过期数据                          │   │
│  │  2. ZCARD 统计请求数                                      │   │
│  │  3. ZRANGE 统计Token总数                                  │   │
│  │  4. 判断是否超限                                         │   │
│  │  5. ZADD 记录新请求（通过时）                            │   │
│  │  6. 返回 {1,0} 或 {-1,waitTime}                         │   │
│  └─────────────────────────────────────────────────────────────┘   │
└─────────────────────────────────────────────────────────────────────┘
                                    │
                    ┌───────────────┴───────────────┐
                    │                               │
                 通过(1)                         拒绝(-1)
                    │                               │
                    ▼                               ▼
┌───────────────────────────┐          ┌───────────────────────────┐
│   executeWithCircuitBreaker │          │   返回429限流错误          │
│   带熔断保护的调用          │          └───────────────────────────┘
└───────────────────────────┘
                    │
                    ▼
┌─────────────────────────────────────────────────────────────────────┐
│              createModelCall() 实际调用LLM API                     │
│          使用OpenAI Java SDK发起请求                               │
└─────────────────────────────────────────────────────────────────────┘
                    │
        ┌───────────┼───────────┐
        │           │           │
      成功        失败        熔断打开
        │           │           │
        ▼           ▼           ▼
┌───────────┐ ┌───────────┐ ┌─────────────────────────────────────┐
│ 提取响应   │ │ fallbackChain │ │ 降级链：                         │
│ Token使用  │ │ 多级降级      │ │ gpt-4 → gpt-3.5-turbo → claude │
│ 构建返回   │ └───────────┘ └─────────────────────────────────────┘
└───────────┘
        │
        ▼
┌─────────────────────────────────────────────────────────────────────┐
│                   返回ChatResponse                                 │
│           包含content、Token使用、耗时、费用                        │
└─────────────────────────────────────────────────────────────────────┘

2、完整项目结构

llm-gateway/
├── pom.xml
├── src/main/java/com/example/llmgateway/
│   ├── LLMGatewayApplication.java
│   ├── config/
│   │   ├── RedisConfig.java
│   │   └── Resilience4jConfig.java
│   ├── controller/
│   │   └── ChatController.java
│   ├── service/
│   │   ├── LLMGatewayService.java
│   │   ├── TokenBucketRateLimiter.java
│   │   └── TokenCounter.java
│   ├── model/
│   │   ├── ChatRequest.java
│   │   └── ChatResponse.java
│   └── exception/
│       └── RateLimitException.java
└── src/main/resources/
    └── application.yml

3、pom.xml 完整依赖和配置文件 application.yml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 
         http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.example</groupId>
    <artifactId>llm-gateway</artifactId>
    <version>1.0.0</version>
    <packaging>jar</packaging>

    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>3.1.5</version>
    </parent>

    <properties>
        <java.version>17</java.version>
        <resilience4j.version>2.2.0</resilience4j.version>
        <jtokkit.version>1.0.1</jtokkit.version>
    </properties>

    <dependencies>
        <!-- Spring Boot WebFlux -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-webflux</artifactId>
        </dependency>

        <!-- Spring Boot Redis Reactive -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis-reactive</artifactId>
        </dependency>

        <!-- Resilience4j 熔断器 -->
        <dependency>
            <groupId>io.github.resilience4j</groupId>
            <artifactId>resilience4j-spring-boot3</artifactId>
            <version>${resilience4j.version}</version>
        </dependency>
        <dependency>
            <groupId>io.github.resilience4j</groupId>
            <artifactId>resilience4j-reactor</artifactId>
            <version>${resilience4j.version}</version>
        </dependency>

        <!-- JTokkit (Java版Tiktoken) -->
        <dependency>
            <groupId>com.knuddels</groupId>
            <artifactId>jtokkit</artifactId>
            <version>${jtokkit.version}</version>
        </dependency>

        <!-- OpenAI Java SDK -->
        <dependency>
            <groupId>com.openai</groupId>
            <artifactId>openai-java</artifactId>
            <version>4.0.0</version>
        </dependency>

        <!-- Jackson 用于JSON处理 -->
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
        </dependency>

        <!-- Lombok 简化代码 -->
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>

        <!-- 测试依赖 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>io.projectreactor</groupId>
            <artifactId>reactor-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>
</project>

spring:
  application:
    name: llm-gateway
  
  redis:
    host: localhost
    port: 6379
    password: 
    timeout: 5000ms
    lettuce:
      pool:
        max-active: 50
        max-idle: 20
        min-idle: 5

# LLM配置
llm:
  rate-limit:
    tpm: 100000  # 每分钟Token限制
    rpm: 100     # 每分钟请求限制
  models:
    gpt-4:
      api-key: ${OPENAI_GPT4_KEY:sk-dummy-key}
      max-tokens: 2000
    gpt-3.5-turbo:
      api-key: ${OPENAI_GPT35_KEY:sk-dummy-key}
      max-tokens: 1500
    claude-3-opus:
      api-key: ${ANTHROPIC_KEY:sk-ant-dummy-key}
      max-tokens: 2000

# Resilience4j配置
resilience4j:
  circuitbreaker:
    configs:
      default:
        sliding-window-size: 10
        failure-rate-threshold: 50
        wait-duration-in-open-state: 60s
        permitted-number-of-calls-in-half-open-state: 3
        automatic-transition-from-open-to-half-open-enabled: true
    instances:
      gpt-4:
        base-config: default
      gpt-3.5-turbo:
        base-config: default
      claude-3-opus:
        base-config: default
  
  retry:
    configs:
      default:
        max-attempts: 3
        wait-duration: 1s
    instances:
      llm-retry:
        base-config: default

logging:
  level:
    com.example: DEBUG
    io.lettuce.core: INFO

4、Redis配置类

package com.example.llmgateway.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.core.ReactiveRedisTemplate;
import org.springframework.data.redis.serializer.RedisSerializationContext;
import org.springframework.data.redis.serializer.StringRedisSerializer;

@Configuration
public class RedisConfig {

    @Bean
    public ReactiveRedisTemplate<String, String> reactiveRedisTemplate(
            ReactiveRedisConnectionFactory connectionFactory) {
        StringRedisSerializer serializer = new StringRedisSerializer();
        RedisSerializationContext<String, String> context = 
            RedisSerializationContext.<String, String>newSerializationContext(serializer)
                .key(serializer)
                .value(serializer)
                .hashKey(serializer)
                .hashValue(serializer)
                .build();
        return new ReactiveRedisTemplate<>(connectionFactory, context);
    }
}

5、Resilience4j配置类

package com.example.llmgateway.config;

import io.github.resilience4j.circuitbreaker.CircuitBreakerConfig;
import io.github.resilience4j.circuitbreaker.CircuitBreakerRegistry;
import io.github.resilience4j.retry.Retry;
import io.github.resilience4j.retry.RetryConfig;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;

@Configuration
public class Resilience4jConfig {

    @Bean
    public CircuitBreakerRegistry circuitBreakerRegistry() {
        CircuitBreakerConfig config = CircuitBreakerConfig.custom()
            .failureRateThreshold(50)
            .slidingWindowSize(10)
            .slidingWindowType(CircuitBreakerConfig.SlidingWindowType.COUNT_BASED)
            .waitDurationInOpenState(Duration.ofSeconds(60))
            .permittedNumberOfCallsInHalfOpenState(3)
            .automaticTransitionFromOpenToHalfOpenEnabled(true)
            .recordException(throwable -> true)
            .build();
        
        return CircuitBreakerRegistry.of(config);
    }

    @Bean
    public Retry retry() {
        RetryConfig config = RetryConfig.custom()
            .maxAttempts(3)
            .waitDuration(Duration.ofSeconds(1))
            .retryExceptions(Exception.class)
            .build();
        
        return Retry.of("llm-retry", config);
    }
}

6、Token计数器

package com.example.llmgateway.service;

import com.knuddels.jtokkit.Encodings;
import com.knuddels.jtokkit.api.Encoding;
import com.knuddels.jtokkit.api.EncodingRegistry;
import com.knuddels.jtokkit.api.ModelType;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;

@Component
public class TokenCounter {
    
    private final EncodingRegistry registry;
    private final Encoding encoding;
    
    public TokenCounter() {
        this.registry = Encodings.newLazyEncodingRegistry();
        // 默认使用cl100k_base编码（GPT-4、GPT-3.5使用）
        this.encoding = registry.getEncodingForModel(ModelType.GPT_4).orElseThrow();
    }
    
    /**
     * 计算输入消息的Token数（精确计算）
     */
    public int countInputTokens(List<Map<String, String>> messages) {
        int totalTokens = 0;
        
        // 1. 计算所有消息内容的Token
        for (Map<String, String> msg : messages) {
            String role = msg.get("role");
            String content = msg.get("content");
            
            if (content != null) {
                totalTokens += encoding.countTokens(content);
            }
            
            // 每个角色名称也需要Token（如 "user", "assistant"）
            if (role != null) {
                totalTokens += encoding.countTokens(role);
            }
        }
        
        // 2. Chat Completion API的固定格式开销
        // 每条消息有3个Token的开销: <|im_start|>, role, <|im_sep|>
        totalTokens += messages.size() * 3;
        
        // 3. 回复的起始Token: <|im_start|>assistant<|im_sep|>
        totalTokens += 1;
        
        return totalTokens;
    }
    
    /**
     * 计算单个文本的Token数
     */
    public int countTokens(String text) {
        if (text == null || text.isEmpty()) {
            return 0;
        }
        return encoding.countTokens(text);
    }
}

7、核心：Redis Lua脚本限流器

package com.example.llmgateway.service;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.ReactiveRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Mono;
import reactor.util.retry.Retry;

import java.time.Duration;
import java.util.Arrays;
import java.util.List;

@Slf4j
@Component
public class TokenBucketRateLimiter {
    
    private final ReactiveRedisTemplate<String, String> redisTemplate;
    private final DefaultRedisScript<List<Long>> rateLimitScript;
    
    @Value("${llm.rate-limit.tpm:100000}")
    private int tpmLimit;
    
    @Value("${llm.rate-limit.rpm:100}")
    private int rpmLimit;
    
    // ===== 完整的Lua脚本 =====
    private static final String LUA_SCRIPT = """
        -- KEYS[1]: Redis键名，如 "llm:rate:limit:bucket:{model}"
        -- ARGV[1]: 当前时间戳（秒）
        -- ARGV[2]: 时间窗口（秒），固定60秒
        -- ARGV[3]: TPM限制（每分钟Token数）
        -- ARGV[4]: RPM限制（每分钟请求数）
        -- ARGV[5]: 本次请求的输入Token数
        -- ARGV[6]: 本次请求的输出Token数（预估）
        
        local key = KEYS[1]
        local now = tonumber(ARGV[1])
        local window = tonumber(ARGV[2])
        local tpmLimit = tonumber(ARGV[3])
        local rpmLimit = tonumber(ARGV[4])
        local inputTokens = tonumber(ARGV[5])
        local outputTokens = tonumber(ARGV[6])
        local requiredTokens = inputTokens + outputTokens
        
        -- 1. 清理过期数据（删除60秒前的记录）
        redis.call('ZREMRANGEBYSCORE', key, 0, now - window)
        
        -- 2. 统计当前窗口内的请求数
        local requests = redis.call('ZCARD', key)
        
        -- 3. 统计当前窗口内的Token总数
        local totalTokens = 0
        local elements = redis.call('ZRANGE', key, 0, -1, 'WITHSCORES')
        for i = 1, #elements, 2 do
            local member = elements[i]
            local colonPos = string.find(member, ':')
            if colonPos then
                local tokens = tonumber(string.sub(member, colonPos + 1))
                if tokens then
                    totalTokens = totalTokens + tokens
                end
            end
        end
        
        -- 4. 检查是否超限
        if (requests + 1 > rpmLimit) or (totalTokens + requiredTokens > tpmLimit) then
            -- 被拒绝！计算建议等待时间
            local firstExpire = redis.call('ZRANGE', key, 0, 0, 'WITHSCORES')
            if #firstExpire > 0 then
                local firstTime = tonumber(firstExpire[2])
                if firstTime then
                    local waitTime = firstTime + window - now
                    if waitTime < 0 then waitTime = 0 end
                    return {-1, waitTime}
                end
            end
            return {-1, 0}
        end
        
        -- 5. 通过限流！记录本次请求
        local member = now .. ':' .. requiredTokens
        redis.call('ZADD', key, now, member)
        redis.call('EXPIRE', key, window)
        
        -- 6. 返回成功
        return {1, 0}
        """;
    
    public TokenBucketRateLimiter(ReactiveRedisTemplate<String, String> redisTemplate) {
        this.redisTemplate = redisTemplate;
        
        // 初始化Lua脚本
        this.rateLimitScript = new DefaultRedisScript<>();
        this.rateLimitScript.setScriptText(LUA_SCRIPT);
        this.rateLimitScript.setResultType(List.class);
        
        log.info("TokenBucketRateLimiter initialized with TPM={}, RPM={}", tpmLimit, rpmLimit);
    }
    
    /**
     * 获取令牌
     * @param inputTokens 输入Token数
     * @param outputTokens 预估输出Token数
     * @param timeout 等待超时时间
     * @return true=获取成功，false=超时失败
     */
    public Mono<Boolean> acquire(int inputTokens, int outputTokens, Duration timeout) {
        String key = "llm:rate:limit:bucket";
        long now = System.currentTimeMillis() / 1000;
        int window = 60;
        
        log.debug("尝试获取令牌: inputTokens={}, outputTokens={}, 总计={}", 
                  inputTokens, outputTokens, inputTokens + outputTokens);
        
        return Mono.defer(() -> 
            redisTemplate.execute(
                rateLimitScript,
                Arrays.asList(key),
                String.valueOf(now),
                String.valueOf(window),
                String.valueOf(tpmLimit),
                String.valueOf(rpmLimit),
                String.valueOf(inputTokens),
                String.valueOf(outputTokens)
            )
        )
        .flatMap(result -> {
            List<Long> values = (List<Long>) result;
            long status = values.get(0);
            long waitTime = values.get(1);
            
            if (status == 1) {
                log.debug("✅ 限流通过");
                return Mono.just(true);
            } else {
                // 被限流，需要等待
                if (waitTime > 0) {
                    log.warn("⏳ 限流被拒，建议等待 {} 秒", waitTime);
                    // 等待建议时间后重试（由外层retry处理）
                } else {
                    log.warn("❌ 限流被拒，无可用配额");
                }
                return Mono.just(false);
            }
        })
        // 如果返回false，重试（最多3次）
        .retryWhen(Retry.fixedDelay(3, Duration.ofSeconds(1))
            .doBeforeRetry(rs -> log.info("限流重试，剩余尝试: {}", 3 - rs.totalRetries())))
        .timeout(timeout, Mono.just(false))
        .onErrorResume(e -> {
            log.error("限流检查异常: {}", e.getMessage());
            return Mono.just(false);
        });
    }
    
    /**
     * 获取当前限流状态（监控用）
     */
    public Mono<RateLimitStatus> getStatus() {
        String key = "llm:rate:limit:bucket";
        long now = System.currentTimeMillis() / 1000;
        int window = 60;
        
        return redisTemplate.execute(
            script -> {
                // 清理过期数据
                redisTemplate.opsForZSet().removeRangeByScore(key, 0, now - window);
                // 获取请求数
                Long requests = redisTemplate.opsForZSet().size(key);
                // 获取Token总数（需要遍历）
                // 简化实现，返回null
                return null;
            },
            Arrays.asList(key)
        ).thenReturn(null);
    }
    
    @lombok.Data
    public static class RateLimitStatus {
        private int currentRequests;
        private int currentTokens;
        private int tpmLimit;
        private int rpmLimit;
        private double tpmUsagePercent;
        private double rpmUsagePercent;
    }
}

8、LLM网关服务

package com.example.llmgateway.service;

import com.example.llmgateway.model.ChatRequest;
import com.example.llmgateway.model.ChatResponse;
import com.openai.client.OpenAIClient;
import com.openai.client.okhttp.OpenAIOkHttpClient;
import com.openai.models.ChatCompletion;
import com.openai.models.ChatCompletionCreateParams;
import io.github.resilience4j.circuitbreaker.CircuitBreaker;
import io.github.resilience4j.circuitbreaker.CircuitBreakerRegistry;
import io.github.resilience4j.decorators.Decorators;
import io.github.resilience4j.retry.Retry;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

@Slf4j
@Service
public class LLMGatewayService {
    
    @Autowired
    private TokenBucketRateLimiter rateLimiter;
    
    @Autowired
    private TokenCounter tokenCounter;
    
    @Autowired
    private CircuitBreakerRegistry circuitBreakerRegistry;
    
    @Autowired
    private Retry retry;
    
    // 模型客户端映射
    private final Map<String, OpenAIClient> modelClients = new HashMap<>();
    
    // 模型降级链配置
    private final Map<String, List<String>> fallbackChains = new HashMap<>();
    
    // 熔断器缓存
    private final Map<String, CircuitBreaker> circuitBreakers = new HashMap<>();
    
    @Value("${llm.models.gpt-4.api-key}")
    private String gpt4ApiKey;
    
    @Value("${llm.models.gpt-3.5-turbo.api-key}")
    private String gpt35ApiKey;
    
    @Value("${llm.models.claude-3-opus.api-key}")
    private String claudeApiKey;
    
    public LLMGatewayService() {
        // 初始化将在@PostConstruct中完成
    }
    
    @javax.annotation.PostConstruct
    public void init() {
        // 初始化模型客户端
        if (gpt4ApiKey != null && !gpt4ApiKey.startsWith("sk-dummy")) {
            modelClients.put("gpt-4", OpenAIOkHttpClient.builder()
                .apiKey(gpt4ApiKey)
                .build());
        }
        if (gpt35ApiKey != null && !gpt35ApiKey.startsWith("sk-dummy")) {
            modelClients.put("gpt-3.5-turbo", OpenAIOkHttpClient.builder()
                .apiKey(gpt35ApiKey)
                .build());
        }
        if (claudeApiKey != null && !claudeApiKey.startsWith("sk-ant-dummy")) {
            // Claude使用OpenAI兼容接口
            modelClients.put("claude-3-opus", OpenAIOkHttpClient.builder()
                .apiKey(claudeApiKey)
                .baseUrl("https://api.anthropic.com/v1")
                .build());
        }
        
        // 配置降级链
        fallbackChains.put("gpt-4", Arrays.asList("gpt-3.5-turbo", "claude-3-opus"));
        fallbackChains.put("gpt-3.5-turbo", Arrays.asList("claude-3-opus"));
        fallbackChains.put("claude-3-opus", Arrays.asList("gpt-4"));
        
        // 为每个模型创建熔断器
        for (String model : modelClients.keySet()) {
            CircuitBreaker breaker = circuitBreakerRegistry.circuitBreaker(model);
            circuitBreakers.put(model, breaker);
            log.info("已注册模型: {}, 熔断器状态: {}", model, breaker.getState());
        }
    }
    
    /**
     * 统一的LLM调用入口
     */
    public Mono<ChatResponse> callLLM(ChatRequest request) {
        String model = request.getModel();
        List<Map<String, String>> messages = request.getMessages();
        int maxOutputTokens = request.getMaxTokens() != null ? request.getMaxTokens() : 2000;
        
        // ===== 步骤1：精确计算输入Token =====
        int inputTokens = tokenCounter.countInputTokens(messages);
        log.info("📊 输入Token数: {}, 模型: {}", inputTokens, model);
        
        // ===== 步骤2：限流检查（使用inputTokens + maxOutputTokens作为预算） =====
        return rateLimiter.acquire(inputTokens, maxOutputTokens, Duration.ofSeconds(30))
            .flatMap(acquired -> {
                if (!acquired) {
                    // 限流被拒绝
                    log.warn("🚫 限流拒绝: model={}, inputTokens={}, maxOutputTokens={}", 
                             model, inputTokens, maxOutputTokens);
                    return Mono.just(ChatResponse.error(
                        "Rate limit exceeded. Please try again later.",
                        429,
                        inputTokens,
                        0
                    ));
                }
                
                log.info("✅ 限流通过，开始调用模型: {}", model);
                
                // ===== 步骤3：执行实际的LLM调用（带熔断和降级） =====
                return executeWithCircuitBreaker(model, messages, maxOutputTokens)
                    .flatMap(response -> {
                        // ===== 步骤4：从响应中提取实际输出Token =====
                        int actualOutputTokens = extractOutputTokens(response);
                        log.info("📊 实际输出Token: {} (预估: {})", actualOutputTokens, maxOutputTokens);
                        
                        // ===== 步骤5：构建响应 =====
                        return Mono.just(ChatResponse.success(
                            response.get("content").toString(),
                            model,
                            inputTokens,
                            actualOutputTokens,
                            (long) response.get("elapsedMs")
                        ));
                    })
                    .onErrorResume(e -> {
                        log.error("❌ LLM调用失败: {}", e.getMessage());
                        return Mono.just(ChatResponse.error(
                            "LLM service error: " + e.getMessage(),
                            503,
                            inputTokens,
                            0
                        ));
                    });
            });
    }
    
    /**
     * 带熔断、重试的执行
     */
    private Mono<Map<String, Object>> executeWithCircuitBreaker(
            String model,
            List<Map<String, String>> messages,
            int maxOutputTokens) {
        
        CircuitBreaker breaker = circuitBreakers.get(model);
        if (breaker == null) {
            return Mono.error(new IllegalArgumentException("Unknown model: " + model));
        }
        
        // 检查熔断器状态
        if (breaker.getState() == CircuitBreaker.State.OPEN) {
            log.warn("🔓 熔断器打开，直接降级: {}", model);
            return fallbackChain(model, messages, maxOutputTokens);
        }
        
        // 创建主调用
        var primaryCall = createModelCall(model, messages, maxOutputTokens);
        
        // 使用Resilience4j装饰器
        var decoratedCall = Decorators
            .ofSupplier(() -> {
                try {
                    return primaryCall.get();
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }
            })
            .withCircuitBreaker(breaker)
            .withRetry(retry)
            .decorate();
        
        return Mono.fromCallable(decoratedCall::get)
            .subscribeOn(Schedulers.boundedElastic())
            .onErrorResume(e -> {
                log.error("主模型 {} 调用失败: {}", model, e.getMessage());
                return fallbackChain(model, messages, maxOutputTokens);
            });
    }
    
    /**
     * 多级降级链
     */
    private Mono<Map<String, Object>> fallbackChain(
            String failedModel,
            List<Map<String, String>> messages,
            int maxOutputTokens) {
        
        List<String> fallbacks = fallbackChains.getOrDefault(failedModel, Collections.emptyList());
        if (fallbacks.isEmpty()) {
            return Mono.error(new RuntimeException("No fallback available for: " + failedModel));
        }
        
        log.info("🔄 开始降级链: {} -> {}", failedModel, String.join(" -> ", fallbacks));
        return tryFallbackModels(fallbacks, messages, maxOutputTokens, 0);
    }
    
    private Mono<Map<String, Object>> tryFallbackModels(
            List<String> fallbacks,
            List<Map<String, String>> messages,
            int maxOutputTokens,
            int index) {
        
        if (index >= fallbacks.size()) {
            return Mono.error(new RuntimeException("All fallback models failed"));
        }
        
        String fallbackModel = fallbacks.get(index);
        CircuitBreaker fallbackBreaker = circuitBreakers.get(fallbackModel);
        
        if (fallbackBreaker == null || fallbackBreaker.getState() == CircuitBreaker.State.OPEN) {
            log.warn("⏭️ 跳过模型 {} (熔断器状态: {})", fallbackModel, 
                     fallbackBreaker != null ? fallbackBreaker.getState() : "不存在");
            return tryFallbackModels(fallbacks, messages, maxOutputTokens, index + 1);
        }
        
        log.info("⬇️ 尝试降级到: {}", fallbackModel);
        
        var fallbackCall = createModelCall(fallbackModel, messages, maxOutputTokens);
        var decoratedCall = Decorators
            .ofSupplier(() -> {
                try {
                    return fallbackCall.get();
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }
            })
            .withCircuitBreaker(fallbackBreaker)
            .decorate();
        
        return Mono.fromCallable(decoratedCall::get)
            .subscribeOn(Schedulers.boundedElastic())
            .doOnSuccess(result -> log.info("✅ 降级成功: {}", fallbackModel))
            .onErrorResume(e -> {
                log.warn("❌ 降级模型 {} 失败: {}", fallbackModel, e.getMessage());
                return tryFallbackModels(fallbacks, messages, maxOutputTokens, index + 1);
            });
    }
    
    /**
     * 创建具体模型的调用
     */
    private CompletableFuture<Map<String, Object>> createModelCall(
            String model,
            List<Map<String, String>> messages,
            int maxOutputTokens) {
        
        return CompletableFuture.supplyAsync(() -> {
            try {
                OpenAIClient client = modelClients.get(model);
                if (client == null) {
                    throw new IllegalArgumentException("Model client not found: " + model);
                }
                
                // 构建请求
                var paramsBuilder = ChatCompletionCreateParams.builder()
                    .model(model)
                    .maxCompletionTokens(maxOutputTokens);
                
                for (Map<String, String> msg : messages) {
                    String role = msg.get("role");
                    String content = msg.get("content");
                    if ("user".equals(role)) {
                        paramsBuilder.addUserMessage(content);
                    } else if ("assistant".equals(role)) {
                        paramsBuilder.addAssistantMessage(content);
                    } else if ("system".equals(role)) {
                        paramsBuilder.addSystemMessage(content);
                    }
                }
                
                long startTime = System.currentTimeMillis();
                ChatCompletion completion = client.chat().completions().create(paramsBuilder.build());
                long elapsed = System.currentTimeMillis() - startTime;
                
                String content = completion.choices().get(0).message().content().orElse("");
                
                Map<String, Object> result = new HashMap<>();
                result.put("content", content);
                result.put("model", model);
                result.put("status", 200);
                result.put("elapsedMs", elapsed);
                
                if (completion.usage() != null) {
                    Map<String, Object> usage = new HashMap<>();
                    usage.put("prompt_tokens", completion.usage().promptTokens().orElse(0));
                    usage.put("completion_tokens", completion.usage().completionTokens().orElse(0));
                    usage.put("total_tokens", completion.usage().totalTokens().orElse(0));
                    result.put("usage", usage);
                }
                
                log.info("✅ 模型调用成功: {} ({}ms)", model, elapsed);
                return result;
                
            } catch (Exception e) {
                throw new RuntimeException("Model [" + model + "] call failed: " + e.getMessage(), e);
            }
        }, CompletableFuture.delayedExecutor(0, TimeUnit.SECONDS));
    }
    
    /**
     * 从响应中提取输出Token
     */
    private int extractOutputTokens(Map<String, Object> response) {
        Object usage = response.get("usage");
        if (usage instanceof Map) {
            Map<String, Object> usageMap = (Map<String, Object>) usage;
            Object completionTokens = usageMap.get("completion_tokens");
            if (completionTokens instanceof Integer) {
                return (Integer) completionTokens;
            }
        }
        return 0;
    }
}

9、数据模型

package com.example.llmgateway.model;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

import java.util.List;
import java.util.Map;

@Data
@NoArgsConstructor
@AllArgsConstructor
public class ChatRequest {
    private String model;
    private List<Map<String, String>> messages;
    private Integer maxTokens;
    private Double temperature = 0.7;
}

package com.example.llmgateway.model;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

@Data
@NoArgsConstructor
@AllArgsConstructor
public class ChatResponse {
    private String content;
    private String model;
    private Integer status;
    private String error;
    private Integer inputTokens;
    private Integer outputTokens;
    private Integer totalTokens;
    private Long elapsedMs;
    private Double cost;
    
    public static ChatResponse success(String content, String model, 
                                       int inputTokens, int outputTokens, long elapsedMs) {
        ChatResponse response = new ChatResponse();
        response.setContent(content);
        response.setModel(model);
        response.setStatus(200);
        response.setInputTokens(inputTokens);
        response.setOutputTokens(outputTokens);
        response.setTotalTokens(inputTokens + outputTokens);
        response.setElapsedMs(elapsedMs);
        response.setCost(calculateCost(model, inputTokens, outputTokens));
        return response;
    }
    
    public static ChatResponse error(String error, int status, int inputTokens, int outputTokens) {
        ChatResponse response = new ChatResponse();
        response.setError(error);
        response.setStatus(status);
        response.setInputTokens(inputTokens);
        response.setOutputTokens(outputTokens);
        response.setTotalTokens(inputTokens + outputTokens);
        return response;
    }
    
    private static double calculateCost(String model, int input, int output) {
        // GPT-4定价: 输入$0.03/1K, 输出$0.06/1K
        return (input * 0.03 + output * 0.06) / 1000.0;
    }
}

10、Controller

package com.example.llmgateway.controller;

import com.example.llmgateway.model.ChatRequest;
import com.example.llmgateway.model.ChatResponse;
import com.example.llmgateway.service.LLMGatewayService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;
import reactor.core.publisher.Mono;

@Slf4j
@RestController
@RequestMapping("/api/llm")
public class ChatController {
    
    @Autowired
    private LLMGatewayService gatewayService;
    
    @PostMapping("/chat")
    public Mono<ChatResponse> chat(@RequestBody ChatRequest request) {
        log.info("收到请求: model={}, messages={}", request.getModel(), request.getMessages().size());
        return gatewayService.callLLM(request);
    }
    
    @GetMapping("/health")
    public Mono<Map<String, String>> health() {
        return Mono.just(Map.of("status", "OK", "service", "llm-gateway"));
    }
}

11、测试脚本

# 1. 启动Redis
docker run -d -p 6379:6379 redis:alpine

# 2. 启动应用
mvn spring-boot:run

# 3. 测试请求
curl -X POST http://localhost:8080/api/llm/chat \
  -H "Content-Type: application/json" \
  -d '{
    "model": "gpt-4",
    "messages": [
      {"role": "system", "content": "你是一个有用的助手"},
      {"role": "user", "content": "请讲一个关于编程的短故事"}
    ],
    "maxTokens": 500
  }'

# 4. 压力测试（并发50个请求）
for i in {1..50}; do
  curl -X POST http://localhost:8080/api/llm/chat \
    -H "Content-Type: application/json" \
    -d '{"model":"gpt-4","messages":[{"role":"user","content":"讲个故事"}]}' &
done
wait