很多开发者第一次接入大模型 API 的体验都特别“丝滑”：几行代码跑通，页面上能看到回答，Demo 跑起来了，感觉可以交付了。

但一旦推到线上，麻烦就来了：

  • 用户盯着白屏等了 10 秒，以为程序崩了，直接关闭页面重新请求；

  • 后端返回了半截 JSON，前端解析报错，日志里只剩下一串残缺的数据碎片；

  • 供应商突然甩过来一个 429 限流错误，你的服务开始疯狂重试，结果越重试越被拦；

  • 用户取消了请求，浏览器断开了连接，但后端还在傻傻地消耗 Token 计费；

  • 同一个订单因为重试被处理了两遍，数据库里多了一条重复记录，用户收到了两条通知……

这些问题，我在实际项目中全都遇到过。今天这篇文章，就是要告诉你：如何把大模型 API 从“能跑的 Demo”变成“扛得住压力的生产系统”。

一、为什么你的 AI 应用总是“不太稳”？

真正的难点不在于“怎么发一个 HTTP 请求给模型”，而在于 如何把大模型 API 当成一个不稳定、昂贵、受配额约束的外部依赖来治理。

生产级调用链路的 8 个关键阶段

一次完整的大模型调用，远比你想象的复杂：

  1. 业务请求校验 – 用户身份、租户权限、套餐限制、请求参数合法性，全都要验。

  2. 上下文组装 – 系统提示词、用户输入、历史对话、RAG 检索结果、工具定义、输出格式约束，一个都不能少。

  3. Token 预算预估 – 提前估算输入 Token，预留输出空间，决定是否裁剪历史、压缩上下文或切换小模型。

  4. 网关路由决策 – 选择模型供应商、部署区域、超时参数、重试策略、限流规则。

  5. 供应商 API 调用 – 同步等待完整结果，或者流式接收增量数据。

  6. 响应解析处理 – 处理增量片段、结束原因、工具调用、Token 用量、拒答场景、结构化 JSON、异常中断。

  7. 状态持久化 – 保存完整回答、增量记录、Token 消耗、调用成本、失败原因、业务状态。

  8. 可观测性记录 – 记录链路追踪 ID、供应商请求 ID、首 Token 延迟、总耗时、重试次数、限流次数、解析失败率。

很多团队最大的误区是：把模型网关当成透明代理。它不是简单的 HTTP 转发，而是 AI 应用的稳定性控制中心。

二、流式输出真的能让响应“更快”吗？

流式输出的真相

很多人以为开启 streaming 就能让模型“跑得更快”，其实这是个误解。

流式输出的本质是把等待过程拆解成可感知的进度条。它不会让模型少算 Token，也不会天然省钱，但能让用户更早看到第一个字，体感上觉得系统“活着”。

维度同步返回流式返回
首字延迟要等完整结果生成完毕收到第一个片段就能展示
端到端耗时取决于完整生成时间总耗时通常差不多
用户体验像提交表单等待结果像 ChatGPT 逐字打印
后端复杂度简单，拿完整字符串处理复杂，要处理增量、取消、断流
结构化解析完整 JSON 一次解析需要缓存或增量解析器

经验法则：面向用户的长文本默认用流式，后台批处理和强结构化任务默认用同步。

SSE 协议的“致命陷阱”

很多开发者在本地测试流式输出时一切正常，但一上生产就发现“流式变批量”，用户要等很久才能看到第一批文字。

问题往往出在事件边界上。SSE 协议用 \n\n（连续两个换行符）作为事件分隔符。如果模型输出的内容里包含真实的换行（比如列表、代码块、多段文字），客户端就会把它解析成多个事件，导致边界错位。

Java 后端实战方案
/**
 * SSE流式输出处理器
 * 核心：转义换行符，避免事件边界错乱
 */
@RestController
@RequestMapping("/api/ai")
public class AIStreamController {
​
    @Autowired
    private LLMService llmService;
​
    /**
     * 流式聊天接口
     * 使用Spring的SseEmitter实现服务端推送
     */
    @GetMapping(value = "/chat/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public SseEmitter streamChat(@RequestParam String question) {
        SseEmitter emitter = new SseEmitter(60_000L); // 60秒超时
​
        // 异步处理，避免阻塞主线程
        CompletableFuture.runAsync(() -> {
            try {
                llmService.chatStream(question, chunk -> {
                    try {
                        // 关键：转义换行符，防止SSE事件边界混乱
                        String escapedChunk = escapeForSSE(chunk);
​
                        SseEmitter.SseEventBuilder event = SseEmitter.event()
                                .id(UUID.randomUUID().toString())
                                .name("message")
                                .data(escapedChunk);
​
                        emitter.send(event);
                    } catch (IOException e) {
                        emitter.completeWithError(e);
                    }
                });
​
                // 流式结束
                emitter.complete();
​
            } catch (Exception e) {
                emitter.completeWithError(e);
            }
        });
​
        return emitter;
    }
​
    /**
     * SSE内容转义：防止换行符破坏事件边界
     */
    private String escapeForSSE(String content) {
        if (content == null) return "";
        return content
                .replace("\r\n", "\\n")
                .replace("\n", "\\n")
                .replace("\r", "\\r");
    }
}

Nginx 网关配置别忘了关缓冲：

location /api/ai/ {
    proxy_pass http://ai-backend;
    proxy_buffering off;           # 关键：关闭缓冲
    proxy_cache off;               # 禁用缓存
    proxy_read_timeout 300s;       # 长生成任务需要更长超时
    proxy_set_header Connection "";
    add_header Cache-Control no-cache;
}

流式场景的四大“鬼门关”

  1. 用户取消 – 前端停止展示了，但后端还在生成，Token 账单照样跑。必须在后端同步取消供应商请求。

  2. 超时分层 – 连接超时、首 Token 超时、总时长超时要分开记录，否则排查问题时根本不知道卡在哪一步。

  3. 断流处理 – 没收到正常结束标记时，不要把半截内容当成功，要标记为“已中断”并提示用户可重试。

  4. 重连陷阱 – SSE 虽然支持自动重连，但大模型输出不是普通新闻推送，重连后无法从 Token 级别续传。更稳的做法是缓存已发送片段，重连时先补发缓存，失效就提示重新生成。

三、重试策略：不是所有错误都能重试

大模型 API 的重试有两个特殊之处：

  1. 失败请求也可能消耗配额 – 甚至已经消耗了部分 Token。

  2. 输出非确定性 – 同样的 Prompt，第二次返回可能完全不同。

错误分类与重试策略

错误类型典型场景能否重试处理方式
网络瞬断连接重置、DNS抖动、读超时✅ 可以指数退避+抖动，限制最大3次
供应商5xx500、502、503、504✅ 可以短暂重试，多次失败切换模型
429限流RPM/TPM/RPD超限⚠️ 谨慎优先看 Retry-After 头，排队或降级
流式中断未收到结束事件⚠️ 视场景用户任务不自动重试，后台任务可幂等重试
400参数错误Schema不合法、上下文超限❌ 不可以修正请求，不要重试同一 payload
401/403鉴权错误API Key无效、权限不足❌ 不可以告警并停用 Key
安全拒答内容策略拒绝❌ 不可以进入业务拒答流程

重试 + 幂等的 Java 实现

/**
 * 大模型调用重试器
 * 核心：指数退避 + 抖动 + 幂等保障
 */
@Service
public class LLMRetryService {
​
    @Autowired
    private LLMClient llmClient;
​
    @Autowired
    private IdempotencyRepository idempotencyRepo;
​
    private static final int MAX_RETRIES = 3;
    private static final long BASE_DELAY_MS = 1000L;
    private static final long MAX_DELAY_MS = 10000L;
​
    /**
     * 带重试和幂等保障的调用
     */
    public LLMResponse callWithRetry(LLMRequest request) {
        String idempotencyKey = generateIdempotencyKey(request);
​
        // 幂等检查：如果已成功过，直接返回历史结果
        Optional<IdempotencyRecord> existing = idempotencyRepo.findByKey(idempotencyKey);
        if (existing.isPresent() && existing.get().isSuccess()) {
            log.info("Idempotency hit, returning cached response: {}", idempotencyKey);
            return existing.get().toResponse();
        }
​
        Exception lastException = null;
​
        for (int attempt = 0; attempt < MAX_RETRIES; attempt++) {
            String attemptId = idempotencyKey + ":attempt:" + attempt;
            long startTime = System.currentTimeMillis();
​
            try {
                // 标记执行中
                idempotencyRepo.markRunning(idempotencyKey, attemptId);
​
                // 调用大模型
                LLMResponse response = llmClient.call(request);
​
                // 记录成功
                idempotencyRepo.markSuccess(
                    idempotencyKey,
                    attemptId,
                    response,
                    System.currentTimeMillis() - startTime
                );
​
                log.info("LLM call succeeded at attempt {}: {}", attempt, attemptId);
                return response;
​
            } catch (RateLimitException e) {
                // 429限流：读取Retry-After头
                lastException = e;
                long retryAfter = e.getRetryAfterSeconds() * 1000L;
                log.warn("Rate limited at attempt {}, retry after {}ms", attempt, retryAfter);
​
                if (attempt < MAX_RETRIES - 1) {
                    sleep(retryAfter);
                }
​
            } catch (RetryableException e) {
                // 可重试错误：指数退避+抖动
                lastException = e;
                long delay = calculateDelayWithJitter(attempt);
                log.warn("Retryable error at attempt {}, retry after {}ms: {}",
                         attempt, delay, e.getMessage());
​
                if (attempt < MAX_RETRIES - 1) {
                    sleep(delay);
                }
​
            } catch (NonRetryableException e) {
                // 不可重试错误：立即失败
                log.error("Non-retryable error at attempt {}: {}", attempt, e.getMessage());
                idempotencyRepo.markFailed(idempotencyKey, attemptId, e);
                throw e;
            }
        }
​
        // 所有重试都失败
        idempotencyRepo.markFailed(idempotencyKey, "all-attempts", lastException);
        throw new LLMException("LLM call failed after " + MAX_RETRIES + " attempts", lastException);
    }
​
    /**
     * 指数退避 + 随机抖动
     * 公式：min(maxDelay, baseDelay * 2^attempt) + random(0, 1000)
     */
    private long calculateDelayWithJitter(int attempt) {
        long exponentialDelay = BASE_DELAY_MS * (1L << attempt); // 2^attempt
        long delayWithCap = Math.min(exponentialDelay, MAX_DELAY_MS);
        long jitter = ThreadLocalRandom.current().nextLong(0, 1000);
        return delayWithCap + jitter;
    }
​
    /**
     * 生成幂等Key：租户+用户+会话+消息
     */
    private String generateIdempotencyKey(LLMRequest request) {
        return String.format("%s:%s:%s:%s",
            request.getTenantId(),
            request.getUserId(),
            request.getConversationId(),
            request.getMessageId()
        );
    }
​
    private void sleep(long millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new LLMException("Retry sleep interrupted", e);
        }
    }
}

为什么必须做幂等？ 因为用户重复点击、网关超时、重试机制都会导致同一个业务请求被执行多次。没有幂等 Key，就可能重复落库、重复扣费、重复发通知。

四、限流：QPS 远远不够

传统 API 限流按 QPS（每秒请求数），但大模型 API 只按 QPS 限流是严重不足的。

为什么 Token 比请求数更重要？

看这两个请求：

  • 请求 A：输入 500 Token，输出 100 Token

  • 请求 B：输入 80K Token，输出 8K Token

它们都算 1 次请求，但对模型推理、供应商配额、账单成本的压力完全不在一个量级。

四层限流架构

/**
 * 四层限流管理器
 * 用户层 → 租户层 → 模型层 → 供应商层
 */
@Component
public class MultiLevelRateLimiter {
​
    @Autowired
    private RedisTemplate<String, String> redis;
​
    /**
     * 分层限流检查
     * @return 是否允许通过
     */
    public boolean tryAcquire(LLMRequest request, int estimatedTokens) {
        String userId = request.getUserId();
        String tenantId = request.getTenantId();
        String model = request.getModel();
        String provider = request.getProvider();
​
        // 第一层：用户级限流（防止单用户滥用）
        if (!checkUserLimit(userId, estimatedTokens)) {
            log.warn("User rate limit exceeded: {}", userId);
            throw new RateLimitException("用户请求过于频繁，请稍后再试");
        }
​
        // 第二层：租户级限流（控制套餐成本）
        if (!checkTenantLimit(tenantId, estimatedTokens)) {
            log.warn("Tenant rate limit exceeded: {}", tenantId);
            throw new RateLimitException("租户配额不足，请联系管理员");
        }
​
        // 第三层：模型级限流（避免热门模型打满）
        if (!checkModelLimit(model, estimatedTokens)) {
            log.warn("Model rate limit exceeded: {}", model);
            throw new RateLimitException("当前模型繁忙，已自动切换备用模型");
        }
​
        // 第四层：供应商级限流（保护外部依赖）
        if (!checkProviderLimit(provider, estimatedTokens)) {
            log.warn("Provider rate limit exceeded: {}", provider);
            throw new RateLimitException("系统繁忙，请稍后重试");
        }
​
        return true;
    }
​
    /**
     * 用户级限流：每分钟请求数 + 每日Token上限
     */
    private boolean checkUserLimit(String userId, int tokens) {
        String requestKey = "rate:user:requests:" + userId;
        String tokenKey = "rate:user:tokens:" + userId;
​
        // 每分钟最多10次请求
        Long requestCount = redis.opsForValue().increment(requestKey);
        if (requestCount == 1) {
            redis.expire(requestKey, 1, TimeUnit.MINUTES);
        }
        if (requestCount > 10) {
            return false;
        }
​
        // 每天最多100K Token
        Long tokenCount = redis.opsForValue().increment(tokenKey, tokens);
        if (tokenCount == tokens) {
            redis.expire(tokenKey, 1, TimeUnit.DAYS);
        }
        if (tokenCount > 100_000) {
            return false;
        }
​
        return true;
    }
​
    /**
     * 租户级限流：月度配额 + 并发上限
     */
    private boolean checkTenantLimit(String tenantId, int tokens) {
        String budgetKey = "rate:tenant:budget:" + tenantId;
        String concurrencyKey = "rate:tenant:concurrency:" + tenantId;
​
        // 月度配额检查
        Long usedBudget = redis.opsForValue().increment(budgetKey, tokens);
        if (usedBudget == tokens) {
            // 设置为当月结束时过期
            LocalDateTime monthEnd = LocalDateTime.now()
                .with(TemporalAdjusters.lastDayOfMonth())
                .withHour(23).withMinute(59).withSecond(59);
            long ttl = Duration.between(LocalDateTime.now(), monthEnd).getSeconds();
            redis.expire(budgetKey, ttl, TimeUnit.SECONDS);
        }
        if (usedBudget > 10_000_000) { // 1000万Token/月
            return false;
        }
​
        // 并发限制：最多10个同时执行的请求
        Long concurrency = redis.opsForValue().increment(concurrencyKey);
        if (concurrency > 10) {
            redis.opsForValue().decrement(concurrencyKey);
            return false;
        }
​
        return true;
    }
​
    /**
     * 模型级限流：令牌桶算法
     */
    private boolean checkModelLimit(String model, int tokens) {
        String key = "rate:model:" + model;
        // 使用Redis的令牌桶脚本（这里简化展示）
        return executeTokenBucketScript(key, tokens, 100_000, 10_000);
    }
​
    /**
     * 供应商级限流：全局RPM/TPM + 熔断器
     */
    private boolean checkProviderLimit(String provider, int tokens) {
        String rpmKey = "rate:provider:rpm:" + provider;
        String tpmKey = "rate:provider:tpm:" + provider;
​
        // 每分钟最多500次请求
        Long rpm = redis.opsForValue().increment(rpmKey);
        if (rpm == 1) {
            redis.expire(rpmKey, 1, TimeUnit.MINUTES);
        }
        if (rpm > 500) {
            return false;
        }
​
        // 每分钟最多1M Token
        Long tpm = redis.opsForValue().increment(tpmKey, tokens);
        if (tpm == tokens) {
            redis.expire(tpmKey, 1, TimeUnit.MINUTES);
        }
        if (tpm > 1_000_000) {
            return false;
        }
​
        return true;
    }
​
    /**
     * 执行令牌桶Lua脚本（伪代码）
     */
    private boolean executeTokenBucketScript(String key, int tokens,
                                             int capacity, int refillRate) {
        // 实际实现需要使用Redis Lua脚本
        // 这里简化展示逻辑
        return true;
    }
​
    /**
     * 释放并发锁（请求完成时调用）
     */
    public void releaseConcurrency(String tenantId) {
        String key = "rate:tenant:concurrency:" + tenantId;
        redis.opsForValue().decrement(key);
    }
}

关键经验：

  • 用户级：防滥用，按请求数 + 日 Token 限制

  • 租户级：控成本，按月度预算 + 并发数

  • 模型级：防打满，按令牌桶 + 降级备选

  • 供应商级：保稳定，按 RPM/TPM + 熔断

五、结构化输出：让模型输出变成“数据契约”

很多人这样写 Prompt：

请分析用户问题，输出 JSON，字段包括 intent、confidence、answer。

然后后端直接 JSON.parse()，Demo 阶段看起来没问题，但生产环境会遇到：

  • 模型在 JSON 前加了“好的，以下是结果：”

  • 字段缺失或类型错误

  • 枚举值乱写

  • 流式返回时只拿到半个对象

  • 安全拒答时压根不是业务 Schema

Java 结构化输出最佳实践

/**
 * 结构化输出处理器
 * 四级兜底策略：本地校验 → 轻量修复 → 降级Schema → 人工兜底
 */
@Service
public class StructuredOutputHandler {
​
    @Autowired
    private LLMClient llmClient;
​
    @Autowired
    private ObjectMapper objectMapper;
​
    @Autowired
    private Validator validator;
​
    /**
     * 带结构化校验的调用
     */
    public <T> T callWithStructuredOutput(LLMRequest request, Class<T> outputClass) {
        // 第一级：本地JSON Schema校验
        String rawOutput = llmClient.call(request).getContent();
​
        try {
            T result = parseAndValidate(rawOutput, outputClass);
            log.info("Structured output validation passed");
            return result;
​
        } catch (JsonProcessingException e) {
            log.warn("JSON parsing failed, trying lightweight fix: {}", e.getMessage());
​
            // 第二级：轻量修复（去除多余文本）
            try {
                String cleaned = extractJsonBlock(rawOutput);
                T result = parseAndValidate(cleaned, outputClass);
                log.info("Lightweight fix succeeded");
                return result;
​
            } catch (Exception e2) {
                log.warn("Lightweight fix failed, trying schema downgrade: {}", e2.getMessage());
​
                // 第三级：降级Schema（拆成多个小对象）
                try {
                    T result = callWithDowngradedSchema(request, outputClass);
                    log.info("Downgraded schema succeeded");
                    return result;
​
                } catch (Exception e3) {
                    log.error("All automatic recovery failed, requiring manual intervention", e3);
​
                    // 第四级：人工兜底
                    throw new StructuredOutputException(
                        "Structured output validation failed after all recovery attempts",
                        rawOutput, e3
                    );
                }
            }
        }
    }
​
    /**
     * 解析并校验结构化输出
     */
    private <T> T parseAndValidate(String json, Class<T> clazz)
            throws JsonProcessingException, ValidationException {
​
        // JSON反序列化
        T object = objectMapper.readValue(json, clazz);
​
        // Bean Validation校验
        Set<ConstraintViolation<T>> violations = validator.validate(object);
        if (!violations.isEmpty()) {
            String errors = violations.stream()
                .map(v -> v.getPropertyPath() + ": " + v.getMessage())
                .collect(Collectors.joining("; "));
            throw new ValidationException("Validation failed: " + errors);
        }
​
        return object;
    }
​
    /**
     * 提取JSON块：去除模型添加的解释文本
     */
    private String extractJsonBlock(String rawOutput) {
        // 匹配 { ... } 或 [ ... ]
        Pattern pattern = Pattern.compile("(?s)(\\{.*\\}|\\[.*\\])");
        Matcher matcher = pattern.matcher(rawOutput);
​
        if (matcher.find()) {
            return matcher.group(1);
        }
​
        throw new IllegalArgumentException("No JSON block found in output");
    }
​
    /**
     * 降级Schema：从复杂对象改为分步抽取
     */
    private <T> T callWithDowngradedSchema(LLMRequest request, Class<T> clazz) {
        // 示例：如果是意图识别，先分类再抽取实体
        if (clazz == IntentRecognitionResult.class) {
            // 第一步：只要求输出意图类型（更简单）
            String intentPrompt = "只需要输出意图类型，不要其他字段：" + request.getUserPrompt();
            LLMRequest simpleRequest = request.withUserPrompt(intentPrompt);
            String intent = llmClient.call(simpleRequest).getContent().trim();
​
            // 第二步：基于意图类型抽取实体
            String entityPrompt = "基于意图" + intent + "，抽取相关实体：" + request.getUserPrompt();
            LLMRequest entityRequest = request.withUserPrompt(entityPrompt);
            String entities = llmClient.call(entityRequest).getContent();
​
            // 手动组装结果
            return (T) IntentRecognitionResult.builder()
                .intent(intent)
                .entities(objectMapper.readValue(entities, Map.class))
                .confidence(0.85) // 降级模式设置默认置信度
                .build();
        }
​
        throw new UnsupportedOperationException("No downgrade strategy for " + clazz);
    }
}
​
/**
 * 意图识别结果（使用Bean Validation）
 */
@Data
@Builder
public class IntentRecognitionResult {
​
    @NotBlank(message = "意图类型不能为空")
    @Pattern(regexp = "refund_request|complaint|inquiry|praise",
             message = "意图类型必须是预定义枚举值")
    private String intent;
​
    @DecimalMin(value = "0.0", message = "置信度最小为0")
    @DecimalMax(value = "1.0", message = "置信度最大为1")
    private Double confidence;
​
    @NotNull(message = "实体字段不能为null")
    private Map<String, Object> entities;
​
    private Boolean needHumanReview;
}

四级兜底的设计哲学：

  1. 本地校验 – JSON Schema + Bean Validation，快速拦截明显错误

  2. 轻量修复 – 只修格式，不重新生成业务内容，节省 Token

  3. 降级 Schema – 复杂对象拆小，分步抽取，降低失败率

  4. 人工兜底 – 高价值场景（订单、金融、医疗、法务）必须有人工复核

六、可观测性：没有指标就没有稳定性

很多团队只记录“调用成功/失败”，但这远远不够。线上出问题时，你需要知道：

  • 是哪个供应商、哪个模型、哪次重试出的问题？

  • 用户等了多久才看到第一个字？

  • Token 消耗是否异常？

  • 是断流、超时还是主动取消？

必须记录的关键指标

/**
 * LLM调用观测记录
 */
@Data
@Builder
public class LLMObservationLog {
    // 链路追踪
    private String traceId;
    private String spanId;
    private String providerRequestId;
​
    // 业务标识
    private String tenantId;
    private String userId;
    private String conversationId;
    private String messageId;
    private String attemptId;
​
    // 模型信息
    private String model;
    private String provider;
    private String promptVersion;
​
    // 性能指标
    private Long ttftMs;              // Time To First Token（首字延迟）
    private Long totalLatencyMs;      // 端到端总耗时
    private Integer inputTokens;      // 输入Token数
    private Integer outputTokens;     // 输出Token数
    private Integer retryCount;       // 重试次数
​
    // 结果状态
    private String finishReason;      // stop/length/tool_calls/cancelled/error
    private String errorType;         // rate_limit/timeout/parse_error/etc
    private Boolean parseSuccess;     // 结构化解析是否成功
    private Boolean userCancelled;    // 用户是否主动取消
​
    // 成本计算
    private Double costUsd;           // 本次调用成本（美元）
​
    // 时间戳
    private Instant requestTime;
    private Instant firstTokenTime;
    private Instant completeTime;
}
​
/**
 * 观测服务
 */
@Service
public class LLMObservationService {
​
    @Autowired
    private MetricRegistry metricRegistry;
​
    @Autowired
    private ObservationLogRepository logRepository;
​
    /**
     * 记录成功调用
     */
    public void recordSuccess(LLMObservationLog log) {
        // 写入日志
        logRepository.save(log);
​
        // 更新Metrics
        metricRegistry.counter("llm.calls.success",
            "model", log.getModel(),
            "provider", log.getProvider()
        ).increment();
​
        metricRegistry.timer("llm.ttft",
            "model", log.getModel()
        ).record(log.getTtftMs(), TimeUnit.MILLISECONDS);
​
        metricRegistry.timer("llm.latency",
            "model", log.getModel()
        ).record(log.getTotalLatencyMs(), TimeUnit.MILLISECONDS);
​
        metricRegistry.counter("llm.tokens.input").increment(log.getInputTokens());
        metricRegistry.counter("llm.tokens.output").increment(log.getOutputTokens());
        metricRegistry.counter("llm.cost.usd").increment(log.getCostUsd());
    }
​
    /**
     * 记录失败调用
     */
    public void recordFailure(LLMObservationLog log) {
        logRepository.save(log);
​
        metricRegistry.counter("llm.calls.failure",
            "model", log.getModel(),
            "provider", log.getProvider(),
            "error_type", log.getErrorType()
        ).increment();
​
        // 特别关注429限流
        if ("rate_limit".equals(log.getErrorType())) {
            metricRegistry.counter("llm.rate_limit",
                "provider", log.getProvider()
            ).increment();
        }
​
        // 特别关注结构化解析失败
        if (Boolean.FALSE.equals(log.getParseSuccess())) {
            metricRegistry.counter("llm.parse_failure",
                "model", log.getModel(),
                "prompt_version", log.getPromptVersion()
            ).increment();
        }
    }
​
    /**
     * 记录用户取消
     */
    public void recordCancellation(LLMObservationLog log) {
        log.setUserCancelled(true);
        log.setFinishReason("cancelled");
        logRepository.save(log);
​
        metricRegistry.counter("llm.calls.cancelled",
            "model", log.getModel()
        ).increment();
​
        // 如果取消发生在首Token之前，说明等待时间过长
        if (log.getFirstTokenTime() == null) {
            metricRegistry.counter("llm.cancelled_before_first_token").increment();
        }
    }
​
    /**
     * 生成可观测性报告（用于排查问题）
     */
    public ObservationReport generateReport(String traceId) {
        List<LLMObservationLog> logs = logRepository.findByTraceId(traceId);
​
        return ObservationReport.builder()
            .traceId(traceId)
            .totalAttempts(logs.size())
            .successCount(logs.stream().filter(l -> "stop".equals(l.getFinishReason())).count())
            .averageTtft(logs.stream().mapToLong(LLMObservationLog::getTtftMs).average().orElse(0))
            .totalTokens(logs.stream().mapToInt(l -> l.getInputTokens() + l.getOutputTokens()).sum())
            .totalCost(logs.stream().mapToDouble(LLMObservationLog::getCostUsd).sum())
            .timeline(logs)
            .build();
    }
}

排查问题时的黄金问题清单：

  1. 用户说“AI 没返回” → 查 ttftMs 是否为 null、finishReason 是 cancelled 还是 timeout

  2. 用户说“回答不对” → 查 promptVersion、inputTokens 是否异常、model 是否被降级

  3. 账单暴涨 → 查 outputTokens 分布、retryCount 是否过高、是否有用户在刷接口

  4. 响应变慢 → 查 ttftMs 和 totalLatencyMs 的 P50/P99、是否集中在某个 provider

  5. 解析失败率上升 → 查 parseSuccess、对应的 promptVersion、是否某个模型不适配

七、生产级架构总结

把前面所有模块串起来，一个完整的 LLM 网关应该长这样：

/**
 * 生产级LLM网关
 * 集成：限流 + 重试 + 幂等 + 结构化 + 观测
 */
@Service
public class ProductionLLMGateway {
​
    @Autowired
    private MultiLevelRateLimiter rateLimiter;
​
    @Autowired
    private LLMRetryService retryService;
​
    @Autowired
    private StructuredOutputHandler structuredHandler;
​
    @Autowired
    private LLMObservationService observationService;
​
    @Autowired
    private TokenEstimator tokenEstimator;
​
    /**
     * 统一调用入口
     */
    public <T> T call(BusinessRequest businessRequest, Class<T> outputClass) {
        Instant requestTime = Instant.now();
        String traceId = MDC.get("traceId");
​
        // 1. 构建LLM请求
        LLMRequest llmRequest = buildLLMRequest(businessRequest);
​
        // 2. Token预估
        int estimatedTokens = tokenEstimator.estimate(llmRequest);
​
        // 3. 四层限流检查
        try {
            rateLimiter.tryAcquire(llmRequest, estimatedTokens);
        } catch (RateLimitException e) {
            recordFailureAndThrow(traceId, llmRequest, requestTime, "rate_limit", e);
        }
​
        // 4. 带重试+幂等的调用
        LLMResponse response;
        int retryCount = 0;
        Instant firstTokenTime = null;
​
        try {
            response = retryService.callWithRetry(llmRequest);
            firstTokenTime = response.getFirstTokenTime();
            retryCount = response.getRetryCount();
​
        } catch (Exception e) {
            recordFailureAndThrow(traceId, llmRequest, requestTime,
                                 classifyErrorType(e), e);
            throw e; // unreachable
        } finally {
            // 5. 释放并发锁
            rateLimiter.releaseConcurrency(llmRequest.getTenantId());
        }
​
        // 6. 结构化输出校验
        T result;
        boolean parseSuccess = true;
​
        try {
            result = structuredHandler.callWithStructuredOutput(llmRequest, outputClass);
        } catch (Exception e) {
            parseSuccess = false;
            recordFailureAndThrow(traceId, llmRequest, requestTime, "parse_error", e);
            throw e; // unreachable
        }
​
        // 7. 记录成功观测
        recordSuccess(traceId, llmRequest, response, requestTime,
                     firstTokenTime, retryCount, parseSuccess);
​
        return result;
    }
​
    /**
     * 记录成功并返回
     */
    private void recordSuccess(String traceId, LLMRequest request, LLMResponse response,
                              Instant requestTime, Instant firstTokenTime,
                              int retryCount, boolean parseSuccess) {
​
        Instant completeTime = Instant.now();
​
        LLMObservationLog log = LLMObservationLog.builder()
            .traceId(traceId)
            .providerRequestId(response.getProviderRequestId())
            .tenantId(request.getTenantId())
            .userId(request.getUserId())
            .conversationId(request.getConversationId())
            .messageId(request.getMessageId())
            .model(request.getModel())
            .provider(request.getProvider())
            .promptVersion(request.getPromptVersion())
            .ttftMs(Duration.between(requestTime, firstTokenTime).toMillis())
            .totalLatencyMs(Duration.between(requestTime, completeTime).toMillis())
            .inputTokens(response.getUsage().getInputTokens())
            .outputTokens(response.getUsage().getOutputTokens())
            .retryCount(retryCount)
            .finishReason("stop")
            .parseSuccess(parseSuccess)
            .costUsd(calculateCost(response))
            .requestTime(requestTime)
            .firstTokenTime(firstTokenTime)
            .completeTime(completeTime)
            .build();
​
        observationService.recordSuccess(log);
    }
​
    /**
     * 记录失败并抛出异常
     */
    private void recordFailureAndThrow(String traceId, LLMRequest request,
                                      Instant requestTime, String errorType,
                                      Exception e) {
​
        LLMObservationLog log = LLMObservationLog.builder()
            .traceId(traceId)
            .tenantId(request.getTenantId())
            .userId(request.getUserId())
            .messageId(request.getMessageId())
            .model(request.getModel())
            .provider(request.getProvider())
            .errorType(errorType)
            .finishReason("error")
            .requestTime(requestTime)
            .build();
​
        observationService.recordFailure(log);
​
        throw new LLMGatewayException("LLM gateway call failed: " + errorType, e);
    }
​
    /**
     * 错误类型分类
     */
    private String classifyErrorType(Exception e) {
        if (e instanceof RateLimitException) return "rate_limit";
        if (e instanceof TimeoutException) return "timeout";
        if (e instanceof AuthException) return "auth_error";
        if (e instanceof NetworkException) return "network_error";
        if (e instanceof NonRetryableException) return "non_retryable";
        return "unknown_error";
    }
​
    /**
     * 成本计算（根据供应商定价）
     */
    private Double calculateCost(LLMResponse response) {
        // 示例：GPT-4o定价
        // 输入：$2.5 / 1M tokens
        // 输出：$10 / 1M tokens
        double inputCost = response.getUsage().getInputTokens() * 2.5 / 1_000_000;
        double outputCost = response.getUsage().getOutputTokens() * 10.0 / 1_000_000;
        return inputCost + outputCost;
    }
​
    private LLMRequest buildLLMRequest(BusinessRequest businessRequest) {
        // 组装System Prompt、用户输入、历史消息、RAG证据等
        return LLMRequest.builder()
            .tenantId(businessRequest.getTenantId())
            .userId(businessRequest.getUserId())
            .conversationId(businessRequest.getConversationId())
            .messageId(businessRequest.getMessageId())
            .model(businessRequest.getModel())
            .provider(businessRequest.getProvider())
            .systemPrompt(businessRequest.getSystemPrompt())
            .userPrompt(businessRequest.getUserPrompt())
            .context(businessRequest.getContext())
            .promptVersion("v1.2.3")
            .build();
    }
}

八、写在最后

从 Demo 到生产，差的不是“能不能跑通”，而是：

  • 能不能在流量高峰时稳定运行

  • 能不能在供应商抖动时快速降级

  • 能不能在出问题时快速定位根因

  • 能不能在成本失控前及时刹车

这篇文章分享的所有技术方案，都是我在实际项目中踩坑后总结出来的。希望能帮你少走弯路，让你的 AI 应用真正做到“生产级可用”。

记住这几个核心原则：

  1. 模型网关是稳定性入口 – 路由、限流、重试、幂等、观测都在这里收口

  2. 流式输出改善的是首字延迟 – 不是总成本，要处理好取消、超时、断流

  3. 重试必须和幂等绑定 – 否则用户狂点就会产生重复订单

  4. 限流不能只按 QPS – Token 才是成本和压力的核心指标

  5. 结构化输出是数据契约 – 四级兜底策略缺一不可

  6. 没有观测就没有稳定性 – TTFT、usage、retryCount 这些指标必须记录

大模型 API 调用，本质上是接入一个聪明但昂贵、偶尔排队、会被限流、输出还需要校验的外部系统。把这套工程治理做到位，AI 应用才算真正从 Demo 走向生产。

# java # 后端 # 面试
Logo
亚马逊云科技技术品牌专区

更多推荐

cover

基于大数据+Hadoop+深度学习的淘宝用户购物可视化与行为预测系统设计与开发

avatar 亚马逊云科技技术品牌专区
cover

基于大数据爬虫+Hadoop+深度学习的旅游推荐系统设计与开发

avatar 亚马逊云科技技术品牌专区

学Linux选学校：一线城市还是老家？

而二线城市很多机构的老师，自己都没进过大厂，技术停留在五年前的水平，甚至对着PPT照本宣科。而二线机构缺乏企业合作资源，项目案例永远是“搭建博客系统”“配置个LAMP环境”——这些在现在的云计算岗位面试中，连初筛都过不了。在一线城市学习，你接触的同学、老师、企业内推资源，都是二线城市没法比的。而老家机构的学员，最好的结果也就是去本地小公司做网管，月薪五六千，天花板肉眼可见。在IT培训这行干了快十年

avatar 亚马逊云科技技术品牌专区