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