构建高可用AI Token中转站：解决大模型API调用的并发瓶颈

共计 3130 个字符，预计需要花费 8 分钟才能阅读完成。

背景痛点

直接调用大模型 API 时，开发者常遇到几个典型问题：

• 突发流量导致 429 错误 ：当多个服务同时发起请求时，容易触发 API 的 rate limit，导致服务不可用。
• 多项目 Token 配额冲突 ：不同项目共享同一个 API Key 时，Token 配额管理混乱，容易互相影响。
• 长文本场景下的 Token 计算误差 ：大模型 API 的 Token 计算与实际消耗可能存在偏差，导致预算超支。

这些问题不仅影响服务的稳定性，还可能增加不必要的成本。

架构设计

替代方案对比

• 反向代理 ：简单易用，但缺乏动态调度能力。
• 消息队列 ：可以缓冲请求，但增加了系统复杂性。

三层架构设计

1. 接入层 ：负责请求的鉴权、限流和日志记录。
2. 调度层 ：实现 Token 的动态分配和请求合并。
3. 持久层 ：存储 Token 使用记录和监控数据。

Token 池的动态扩缩容算法

Token 池的核心是动态调整配额，算法流程如下：

1. 初始化时，根据历史数据设置初始配额。
2. 实时监控请求成功率，动态调整配额。
3. 当请求失败率超过阈值时，自动缩减配额。
4. 当请求成功率稳定时，逐步增加配额。

sequenceDiagram
    participant Client
    participant AccessLayer
    participant Scheduler
    participant Persistence
    Client->>AccessLayer: 发起请求
    AccessLayer->>Scheduler: 申请 Token
    Scheduler->>Persistence: 查询剩余 Token
    Persistence-->>Scheduler: 返回 Token 数量
    Scheduler->>AccessLayer: 分配 Token
    AccessLayer->>Client: 返回结果 

核心代码

限流中间件实现

以下是一个基于滑动窗口的限流中间件示例：

package middleware

import (
    "sync"
    "time"
)

type RateLimiter struct {
    windowSize time.Duration
    maxRequests int
    requests map[int64]int
    mu sync.Mutex
}

func NewRateLimiter(windowSize time.Duration, maxRequests int) *RateLimiter {
    return &RateLimiter{
        windowSize: windowSize,
        maxRequests: maxRequests,
        requests: make(map[int64]int),
    }
}

func (r *RateLimiter) Allow() bool {r.mu.Lock()
    defer r.mu.Unlock()

    now := time.Now().Unix()
    cutoff := now - int64(r.windowSize.Seconds())

    // 清理过期的请求记录
    for timestamp := range r.requests {
        if timestamp < cutoff {delete(r.requests, timestamp)
        }
    }

    // 统计当前窗口内的请求数量
    var count int
    for _, v := range r.requests {count += v}

    if count >= r.maxRequests {return false}

    r.requests[now]++
    return true
}

BloomFilter 防重复请求

BloomFilter 可以高效判断请求是否已处理：

package utils

import (
    "github.com/bits-and-blooms/bloom"
    "sync"
)

type RequestDeduplicator struct {
    filter *bloom.BloomFilter
    mu sync.Mutex
}

func NewRequestDeduplicator(capacity uint, falsePositiveRate float64) *RequestDeduplicator {
    return &RequestDeduplicator{filter: bloom.NewWithEstimates(capacity, falsePositiveRate),
    }
}

func (d *RequestDeduplicator) Seen(data []byte) bool {d.mu.Lock()
    defer d.mu.Unlock()

    if d.filter.Test(data) {return true}

    d.filter.Add(data)
    return false
}

Prometheus 监控埋点

监控关键指标示例：

package metrics

import (
    "github.com/prometheus/client_golang/prometheus"
    "github.com/prometheus/client_golang/prometheus/promauto"
)

var (
    RequestsTotal = promauto.NewCounterVec(prometheus.CounterOpts{
        Name: "api_requests_total",
        Help: "Total number of API requests",
    }, []string{"endpoint", "status"})

    TokenUsage = promauto.NewGaugeVec(prometheus.GaugeOpts{
        Name: "token_usage",
        Help: "Current token usage",
    }, []string{"project"})
)

生产实践

压测报告

我们对比了有无中转站时的性能表现：

冷启动预热方案

为避免冷启动时的性能问题，我们实现了以下预热流程：

1. 服务启动时，预先加载 Token 配额。
2. 逐步增加请求量，监控系统表现。
3. 达到稳定状态后，开放全部流量。

JWT 鉴权实践

JWT 鉴权的安全最佳实践：

• 使用强密钥（至少 256 位）。
• 设置合理的过期时间（建议不超过 1 小时）。
• 实现 Token 刷新机制。

避坑指南

避免 Token 双花问题

在分布式环境中，确保 Token 不会被重复使用：

1. 使用分布式锁（如 Redis）保护关键操作。
2. 实现乐观锁机制检查 Token 状态。

处理 rate-limit 响应头

正确处理第三方 API 的限流响应：

func parseRateLimitHeaders(headers http.Header) (time.Time, error) {resetTimeStr := headers.Get("X-RateLimit-Reset")
    if resetTimeStr == "" {return time.Time{}, errors.New("no rate limit header")
    }

    resetUnix, err := strconv.ParseInt(resetTimeStr, 10, 64)
    if err != nil {return time.Time{}, err
    }

    return time.Unix(resetUnix, 0), nil
}

日志脱敏方案

保护敏感信息的日志记录：

func sanitizeLog(input string) string {if len(input) > 8 {return input[:2] + "****" + input[len(input)-2:]
    }
    return "****"
}

动手挑战

尝试扩展中转站以支持 Llama3 的 streaming 调用，考虑以下几点：

1. 如何高效管理 streaming 连接的生命周期。
2. 实现流式数据的 Token 计数。
3. 处理 streaming 过程中的错误和重试。

欢迎在评论区分享你的实现方案！