AI推理服務閘道實戰：LLM API閘道、模型路由與限流降級
技术架构
摘要
- AI推理閘道是LLM服務的「前門」：路由、限流、降級、可觀測4大核心能力缺一不可
- 模型路由的3種策略：成本優先、延遲優先、品質優先，按業務場景選擇
- 限流不止防刷：Token Rate Limiting比請求頻率限制更精準，避免單使用者佔滿上下文視窗
- Fallback機制是SLA的底線：主模型逾時→備模型接管→快取兜底→優雅降級
- 本文提供從閘道架構到Go實作的完整方案，含K8s部署與Prometheus監控
目錄
- 為什麼LLM服務需要專用閘道
- AI推理閘道架構設計
- 模型路由：3種策略與實作
- Token Rate Limiting：精準限流
- Fallback與降級：SLA的底線
- Go閘道實作與K8s部署
- 總結與延伸閱讀
為什麼LLM服務需要專用閘道
傳統API閘道的3大不足
| 不足 | 說明 | 影響 |
|---|---|---|
| 無Token感知 | 按請求頻率限流，不感知Token消耗 | 單使用者長上下文請求可佔滿GPU |
| 無模型路由 | 無法根據請求特徵路由到不同模型 | 小問題用大模型，浪費成本 |
| 無串流適配 | SSE串流響應的限流/降級邏輯不同 | 串流請求逾時後無法優雅降級 |
AI推理閘道 vs 傳統API閘道
| 維度 | 傳統API閘道 | AI推理閘道 |
|---|---|---|
| 限流維度 | 請求頻率 | Token消耗 + 請求頻率 |
| 路由策略 | URL路徑 | 模型能力 + 成本 + 延遲 |
| 響應模式 | 請求-響應 | SSE串流 + 請求-響應 |
| 降級策略 | 回傳錯誤 | Fallback到備模型 |
| 可觀測性 | QPS/延遲 | Token吞吐/首Token延遲/上下文長度 |
AI推理閘道架構設計
┌──────────────────────────────────────────────────────────────┐
│ AI推理閘道架構 │
│ │
│ ┌──────────┐ │
│ │ 客戶端 │ │
│ └────┬─────┘ │
│ │ │
│ ┌────▼──────────────────────────────────────────────────┐ │
│ │ AI Inference Gateway │ │
│ │ │ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌─────────┐ │ │
│ │ │ 認證 │ │ 限流 │ │ 路由 │ │ 降級 │ │ │
│ │ │ API Key │ │ Token RL │ │ 模型路由 │ │Fallback │ │ │
│ │ │ OAuth2.0 │ │ 優先佇列 │ │ 負載均衡 │ │ 快取 │ │ │
│ │ └──────────┘ └──────────┘ └──────────┘ └─────────┘ │ │
│ │ │ │
│ │ ┌──────────────────────────────────────────────────┐ │ │
│ │ │ 可觀測性 │ │ │
│ │ │ Prometheus + OpenTelemetry + 結構化日誌 │ │ │
│ │ └──────────────────────────────────────────────────┘ │ │
│ └──────────────────────────────────────────────────────┘ │
│ │ │ │ │
│ ┌────▼─────┐ ┌───▼──────┐ ┌─▼────────┐ │
│ │ vLLM │ │ TGI │ │ SGLang │ │
│ │ Qwen-7B │ │ Llama-70B│ │ Qwen-7B │ │
│ └──────────┘ └──────────┘ └──────────┘ │
└──────────────────────────────────────────────────────────────┘
模型路由：3種策略與實作
路由策略對比
| 策略 | 路由邏輯 | 成本 | 延遲 | 品質 | 適用場景 |
|---|---|---|---|---|---|
| 成本優先 | 優先路由到最小模型 | 最低 | 低 | 中 | 內部工具 |
| 延遲優先 | 優先路由到最快模型 | 中 | 最低 | 中 | 即時對話 |
| 品質優先 | 優先路由到最強模型 | 最高 | 高 | 最高 | 專業場景 |
Go路由實作
package gateway
import (
"context"
"math"
"sync"
"time"
)
type ModelEndpoint struct {
Name string
URL string
ModelID string
MaxTokens int
CostPerToken float64
AvgLatency time.Duration
CurrentLoad float64
Capabilities []string
}
type RoutingStrategy string
const (
CostFirst RoutingStrategy = "cost_first"
LatencyFirst RoutingStrategy = "latency_first"
QualityFirst RoutingStrategy = "quality_first"
)
type ModelRouter struct {
endpoints []*ModelEndpoint
strategy RoutingStrategy
mu sync.RWMutex
}
func NewModelRouter(strategy RoutingStrategy) *ModelRouter {
return &ModelRouter{strategy: strategy}
}
func (r *ModelRouter) Register(endpoint *ModelEndpoint) {
r.mu.Lock()
defer r.mu.Unlock()
r.endpoints = append(r.endpoints, endpoint)
}
func (r *ModelRouter) Route(ctx context.Context, req *InferenceRequest) (*ModelEndpoint, error) {
r.mu.RLock()
defer r.mu.RUnlock()
candidates := r.filterByCapability(req)
if len(candidates) == 0 {
return nil, ErrNoAvailableModel
}
switch r.strategy {
case CostFirst:
return r.routeByCost(candidates), nil
case LatencyFirst:
return r.routeByLatency(candidates), nil
case QualityFirst:
return r.routeByQuality(candidates), nil
default:
return r.routeByLatency(candidates), nil
}
}
func (r *ModelRouter) routeByCost(candidates []*ModelEndpoint) *ModelEndpoint {
best := candidates[0]
for _, ep := range candidates[1:] {
if ep.CostPerToken < best.CostPerToken && ep.CurrentLoad < 0.9 {
best = ep
}
}
return best
}
func (r *ModelRouter) routeByLatency(candidates []*ModelEndpoint) *ModelEndpoint {
best := candidates[0]
for _, ep := range candidates[1:] {
effectiveLatency := float64(ep.AvgLatency) / (1.0 - ep.CurrentLoad + 0.01)
bestLatency := float64(best.AvgLatency) / (1.0 - best.CurrentLoad + 0.01)
if effectiveLatency < bestLatency {
best = ep
}
}
return best
}
func (r *ModelRouter) routeByQuality(candidates []*ModelEndpoint) *ModelEndpoint {
best := candidates[0]
for _, ep := range candidates[1:] {
if ep.MaxTokens > best.MaxTokens && ep.CurrentLoad < 0.9 {
best = ep
}
}
return best
}
func (r *ModelRouter) filterByCapability(req *InferenceRequest) []*ModelEndpoint {
var filtered []*ModelEndpoint
for _, ep := range r.endpoints {
if ep.CurrentLoad >= 0.95 {
continue
}
if req.MaxTokens > 0 && ep.MaxTokens < req.MaxTokens {
continue
}
filtered = append(filtered, ep)
}
return filtered
}
Token Rate Limiting：精準限流
Token限流 vs 請求限流
| 限流維度 | 優點 | 缺點 | 適用場景 |
|---|---|---|---|
| 請求頻率 | 實作簡單 | 不感知Token消耗 | 低精度限流 |
| Token/分鐘 | 精準控制GPU消耗 | 需預估Token數 | 生產推薦 |
| 並行請求數 | 控制GPU並行 | 不感知上下文長度 | 簡單場景 |
Go Token限流實作
package gateway
import (
"context"
"sync"
"time"
)
type TokenBucket struct {
mu sync.Mutex
tokens float64
maxTokens float64
refillRate float64
lastRefill time.Time
}
func NewTokenBucket(maxTokens, refillRate float64) *TokenBucket {
return &TokenBucket{
tokens: maxTokens,
maxTokens: maxTokens,
refillRate: refillRate,
lastRefill: time.Now(),
}
}
func (tb *TokenBucket) Allow(estimatedTokens int) bool {
tb.mu.Lock()
defer tb.mu.Unlock()
now := time.Now()
elapsed := now.Sub(tb.lastRefill).Seconds()
tb.tokens = math.Min(tb.maxTokens, tb.tokens+elapsed*tb.refillRate)
tb.lastRefill = now
if tb.tokens >= float64(estimatedTokens) {
tb.tokens -= float64(estimatedTokens)
return true
}
return false
}
type TokenRateLimiter struct {
buckets map[string]*TokenBucket
mu sync.RWMutex
}
func NewTokenRateLimiter() *TokenRateLimiter {
return &TokenRateLimiter{buckets: make(map[string]*TokenBucket)}
}
func (rl *TokenRateLimiter) RegisterUser(userID string, maxTokens, refillRate float64) {
rl.mu.Lock()
defer rl.mu.Unlock()
rl.buckets[userID] = NewTokenBucket(maxTokens, refillRate)
}
func (rl *TokenRateLimiter) Allow(userID string, estimatedTokens int) bool {
rl.mu.RLock()
bucket, ok := rl.buckets[userID]
rl.mu.RUnlock()
if !ok {
return false
}
return bucket.Allow(estimatedTokens)
}
Fallback與降級：SLA的底線
4級降級策略
┌──────────────────────────────────────────────────────────┐
│ 4級降級策略 │
│ │
│ Level 0: 正常服務 │
│ ┌──────────────────────────────────────────┐ │
│ │ 主模型(Qwen-72B) → 延遲50ms, 品質最高 │ │
│ └──────────────────────────────────────────┘ │
│ ↓ 逾時/過載 │
│ Level 1: 備模型接管 │
│ ┌──────────────────────────────────────────┐ │
│ │ 備模型(Qwen-7B) → 延遲20ms, 品質降低 │ │
│ └──────────────────────────────────────────┘ │
│ ↓ 備模型也過載 │
│ Level 2: 快取兜底 │
│ ┌──────────────────────────────────────────┐ │
│ │ 語意快取命中 → 延遲5ms, 品質取決於快取 │ │
│ └──────────────────────────────────────────┘ │
│ ↓ 快取未命中 │
│ Level 3: 優雅降級 │
│ ┌──────────────────────────────────────────┐ │
│ │ 回傳預設回覆 + 重試提示 │ │
│ └──────────────────────────────────────────┘ │
└──────────────────────────────────────────────────────────┘
Go Fallback實作
type FallbackChain struct {
primary *ModelEndpoint
secondary *ModelEndpoint
cache SemanticCache
timeout time.Duration
}
func (fc *FallbackChain) Generate(ctx context.Context, req *InferenceRequest) (string, error) {
ctx, cancel := context.WithTimeout(ctx, fc.timeout)
defer cancel()
result, err := fc.callModel(ctx, fc.primary, req)
if err == nil {
return result, nil
}
log.Warn("primary model failed, falling back", "error", err)
if fc.secondary != nil {
result, err = fc.callModel(ctx, fc.secondary, req)
if err == nil {
return result, nil
}
log.Warn("secondary model failed", "error", err)
}
if fc.cache != nil {
cached, ok := fc.cache.Get(req.Prompt)
if ok {
log.Info("cache hit on fallback")
return cached, nil
}
}
return "服務暫時無法使用，請稍後重試。", ErrServiceUnavailable
}
Go閘道實作與K8s部署
K8s Deployment
apiVersion: apps/v1
kind: Deployment
metadata:
name: ai-inference-gateway
namespace: ai-inference
spec:
replicas: 3
selector:
matchLabels:
app: ai-inference-gateway
template:
spec:
containers:
- name: gateway
image: myregistry/ai-inference-gateway:v1.0
ports:
- containerPort: 8080
resources:
requests:
cpu: "2"
memory: 2Gi
limits:
cpu: "4"
memory: 4Gi
env:
- name: PRIMARY_MODEL_URL
value: "http://vllm-qwen72b:8000/v1"
- name: SECONDARY_MODEL_URL
value: "http://vllm-qwen7b:8000/v1"
- name: REDIS_URL
value: "redis://redis:6379"
- name: ROUTING_STRATEGY
value: "latency_first"
- name: TOKEN_RATE_LIMIT
value: "100000"
---
apiVersion: v1
kind: Service
metadata:
name: ai-gateway-svc
namespace: ai-inference
spec:
selector:
app: ai-inference-gateway
ports:
- port: 8080
targetPort: 8080
type: ClusterIP
Prometheus監控
apiVersion: monitoring.coreos.com/v1
kind: PrometheusRule
metadata:
name: ai-gateway-alerts
namespace: ai-inference
spec:
groups:
- name: ai-gateway
rules:
- alert: HighFallbackRate
expr: rate(gateway_fallback_total[5m]) / rate(gateway_requests_total[5m]) > 0.1
for: 5m
labels:
severity: warning
annotations:
summary: "AI閘道Fallback率過高"
- alert: TokenRateLimitExceeded
expr: rate(gateway_rate_limit_exceeded_total[5m]) > 10
for: 2m
labels:
severity: warning
annotations:
summary: "Token限流觸發頻繁"
總結與延伸閱讀
AI推理閘道是LLM服務的「前門」，4大核心能力（路由、限流、降級、可觀測）缺一不可。Token Rate Limiting比請求頻率限流更精準，4級Fallback保障SLA底線。
設計要點回顧：
- AI推理閘道必須Token感知，傳統API閘道不夠
- 3種路由策略：成本優先/延遲優先/品質優先，按場景選擇
- Token Rate Limiting是生產限流的標配
- 4級Fallback：主模型→備模型→快取→優雅降級
- Go實作+K8s部署+Prometheus監控是生產標配
相關閱讀：
- 雲原生AI部署全攻略 — AI推理服務的K8s部署
- 大模型推理加速基準測試 — 閘道後端推理引擎選型
- K8s Gateway API遷移完全指南 — K8s流量管理演進
權威參考：
本站提供瀏覽器本地工具，免註冊即可試用 →
#AI推理网关#LLM API网关#模型路由策略#推理服务治理#AI网关限流#2026