AI 驱动的独立产品 API 网关架构设计大家好我是蔓蔓。在打磨自己独立产品的过程中随着接入的第三方服务越来越多API 管理的复杂度呈指数级增长。七个不同的 AI 模型提供商、三个支付服务商、两个云存储平台——每个都有不同的认证方式、限流规则和错误格式。这次我设计了一个 AI 驱动的 API 网关来统一管理这些调用。今天和大家分享完整架构。一、独立产品 API 管理失控当后端变成意大利面条独立产品在发展过程中后端服务通常从单体开始逐步接入各种第三方 API。这个过程的典型问题失控的调用链路// 常见的混乱状况散落在各处的 API 调用 // services/order.ts const payment await stripe.charges.create({ ... }); // 硬编码 key // services/ai.ts const completion await openai.chat.completions.create({ ... }); const translation await deepl.translate({ ... }); // services/storage.ts const file await s3.upload({ ... }).promise();问题清单认证凭证分散API Key 散落在各处轮换成本极高限流各自为战每个服务独立限流无法做到全局协调错误处理不一致Stripe 返回error.codeOpenAI 返回error.type监控盲区无法统一追踪跨服务的调用链路和成本降级逻辑缺失一个服务挂了整个产品宕机二、网关架构设计智能路由与统一抽象核心设计思想是所有外部 API 调用统一通过网关网关负责认证、限流、容错、转换和监控。flowchart TB subgraph 业务层 A1[订单服务] A2[AI 内容服务] A3[用户服务] end subgraph API 网关核心 B1[智能路由器] B2[认证管理器] B3[限流控制器] B4[响应适配器] B5[熔断器] B6[监控与计费] end subgraph 外部服务 C1[OpenAI] C2[Anthropic] C3[Stripe] C4[AWS S3] C5[DeepL] end A1 -- B1 A2 -- B1 A3 -- B1 B1 -- B2 -- B3 -- B5 B5 -- B4 -- C1 B5 -- B4 -- C2 B5 -- B4 -- C3 B5 -- B4 -- C4 B5 -- B4 -- C5 B2 -- B6 B3 -- B6 B5 -- B6统一请求/响应标准化// types/gateway.ts interface GatewayRequestT unknown { service: string; // openai | stripe | s3 action: string; // chat.completions | charges.create payload: T; options?: { timeout?: number; retries?: number; priority?: high | normal | low; }; } interface GatewayResponseT unknown { success: boolean; data?: T; error?: { code: string; message: string; original?: unknown; // 保留原始错误用于调试 }; metadata: { latency: number; provider: string; cost?: number; // AI 调用成本追踪 retries: number; }; }认证管理器// gateways/auth-manager.ts import { createCipheriv, createDecipheriv, randomBytes } from crypto; class AuthManager { private keyCache new Mapstring, { keys: string[]; index: number }(); constructor(private encryptionKey: Buffer) {} async getCredential(service: string): Promisestring { // 先从加密存储中获取原始凭证 const encrypted await this.loadFromVault(service); const decrypted this.decrypt(encrypted); // 多 Key 轮转策略均匀分布调用压力 if (Array.isArray(decrypted)) { return this.roundRobin(service, decrypted); } return decrypted; } private roundRobin(service: string, keys: string[]): string { if (!this.keyCache.has(service)) { this.keyCache.set(service, { keys, index: 0 }); } const cache this.keyCache.get(service)!; const key keys[cache.index % keys.length]; cache.index; return key; } private encrypt(data: string): Buffer { const iv randomBytes(16); const cipher createCipheriv(aes-256-gcm, this.encryptionKey, iv); const encrypted Buffer.concat([cipher.update(data), cipher.final()]); const tag cipher.getAuthTag(); return Buffer.concat([iv, tag, encrypted]); } private decrypt(buffer: Buffer): string { const iv buffer.subarray(0, 16); const tag buffer.subarray(16, 32); const encrypted buffer.subarray(32); const decipher createDecipheriv(aes-256-gcm, this.encryptionKey, iv); decipher.setAuthTag(tag); return decipher.update(encrypted) decipher.final(utf8); } }三、限流控制与熔断保护独立产品的流量特征是波动的需要实现多层级限流来保护后端预算。滑动窗口限流器// gateways/rate-limiter.ts class SlidingWindowLimiter { private windows new Mapstring, number[](); constructor( private limits: Mapstring, { maxRequests: number; windowMs: number } ) {} async acquire(service: string): Promiseboolean { const limit this.limits.get(service); if (!limit) return true; // 无限流配置则放行 const now Date.now(); const windowStart now - limit.windowMs; let timestamps this.windows.get(service) || []; timestamps timestamps.filter((t) t windowStart); if (timestamps.length limit.maxRequests) { return false; } timestamps.push(now); this.windows.set(service, timestamps); return true; } }熔断器实现// gateways/circuit-breaker.ts enum CircuitState { CLOSED, // 正常通行 OPEN, // 拒绝请求 HALF_OPEN, // 探测恢复 } class CircuitBreaker { private state CircuitState.CLOSED; private failureCount 0; private lastFailureTime 0; constructor( private threshold: number 5, // 失败阈值 private resetTimeout: number 30000, // 熔断恢复时间 ) {} async callT(fn: () PromiseT): PromiseT { if (this.state CircuitState.OPEN) { if (Date.now() - this.lastFailureTime this.resetTimeout) { this.state CircuitState.HALF_OPEN; } else { throw new Error(Circuit breaker is OPEN); } } try { const result await fn(); this.onSuccess(); return result; } catch (error) { this.onFailure(); throw error; } } private onSuccess(): void { this.failureCount 0; this.state CircuitState.CLOSED; } private onFailure(): void { this.failureCount; this.lastFailureTime Date.now(); if ( this.state CircuitState.HALF_OPEN || this.failureCount this.threshold ) { this.state CircuitState.OPEN; } } }AI 成本追踪// gateways/cost-tracker.ts interface CostRecord { service: string; model: string; tokensInput: number; tokensOutput: number; cost: number; timestamp: number; } class CostTracker { private records: CostRecord[] []; // 已知的模型定价USD/1K tokens private pricing new Mapstring, { input: number; output: number }([ [gpt-4o, { input: 0.005, output: 0.015 }], [gpt-4o-mini, { input: 0.00015, output: 0.0006 }], [claude-3.5-sonnet, { input: 0.003, output: 0.015 }], ]); track(service: string, model: string, usage: { input: number; output: number }): void { const pricing this.pricing.get(model); if (!pricing) return; const cost (usage.input / 1000) * pricing.input (usage.output / 1000) * pricing.output; this.records.push({ service, model, tokensInput: usage.input, tokensOutput: usage.output, cost, timestamp: Date.now(), }); // 成本告警 const dailyCost this.getDailyCost(); if (dailyCost 50) { // 每日预算阈值 console.warn(AI 调用日成本已超过 $50当前 $${dailyCost.toFixed(2)}); } } private getDailyCost(): number { const today new Date().setHours(0, 0, 0, 0); return this.records .filter((r) r.timestamp today) .reduce((sum, r) sum r.cost, 0); } }踩坑SlidingWindowLimiter 的内存膨胀与多实例一致性独立产品初期使用单进程部署时SlidingWindowLimiter的内存问题并不明显。但当服务扩展到多进程如 Node.js cluster 模式或 Kubernetes 多 Pod会出现两个问题第一各进程的限流窗口独立计数实际限流阈值变成了maxRequests × 进程数。一个配置了每分钟 60 次限流的接口在 4 个 Pod 环境下实际允许了 240 次/分钟。解决方案是将限流状态迁移到 Redis使用EVAL命令实现原子性的滑动窗口操作// Redis 滑动窗口限流原子操作多进程一致 async function acquireViaRedis(service: string, limit: RateLimit): Promiseboolean { const now Date.now(); const key rate_limit:${service}; const result await redis.eval( local windowStart ARGV[2] - ARGV[3] redis.call(ZREMRANGEBYSCORE, KEYS[1], 0, windowStart) local count redis.call(ZCARD, KEYS[1]) if count tonumber(ARGV[1]) then redis.call(ZADD, KEYS[1], ARGV[2], ARGV[2]) redis.call(EXPIRE, KEYS[1], ARGV[3] / 1000) return 1 end return 0, 1, key, limit.maxRequests, now, limit.windowMs ); return result 1; }第二timestamps 数组在高频服务下持续膨胀。一个每秒 100 次请求的服务windowsMap 中的数组每分钟增长 6000 个元素。虽然filter操作会清理过期数据但在极端峰值下内存峰值可能超出预期。使用 Redis 的 Sorted SetZADDZREMRANGEBYSCORE天然解决了这个问题。四、架构权衡与容量规划网关的性能代价API 网关在请求链路中引入了一个额外的网络跳平均增加 5-15ms 的延迟。对于延迟敏感的场景如 AI 流式响应需要特殊处理// 流式响应的代理模式不缓存、不过滤 async function proxyStream(req: GatewayRequest): PromiseReadableStream { const credential await authManager.getCredential(req.service); const upstream await fetch(getUpstreamUrl(req), { headers: { Authorization: Bearer ${credential} }, body: JSON.stringify(req.payload), }); // 直接透传流不做任何处理 return upstream.body!; }单点故障风险网关本身成为单点。对于独立产品轻量化的高可用方案是在多进程间共享状态通过 Redis而非引入完整的服务网格。五、总结AI 驱动的 API 网关是独立产品技术栈中的隐形守护者。核心设计原则有三条第一统一抽象是网关的立身之本。标准化请求/响应格式消除了业务代码中的大量适配逻辑让开发者只需关注业务本身。第二保护性能力比功能性更重要。限流、熔断、多 Key 轮转这些能力对于独立产品的稳定性至关重要尤其是当你的月账单直接与 API 调用量挂钩时。第三成本可见性是决策基础。AI 产品的成本结构与传统 SaaS 完全不同——模型调用是主要开销。成本追踪必须内建在网关中而非事后统计。推荐实施路径先接入调用量最高的 2-3 个服务验证统一抽象层的设计是否合理再逐步加入认证管理、限流和成本追踪最后完善熔断和降级策略。你的独立产品目前是如何管理第三方 API 调用的欢迎在评论区交流
