一文搞懂 AI 网关：多模型动态路由、failover 与配额治理

问题背景：为什么不能让业务直连各家大模型

当系统只接一个大模型时，在业务代码里写死 endpoint 和 API Key 似乎最简单；但一旦进入多模型、多团队、多租户阶段，问题会迅速放大：

成本不可控：高级模型被用于低价值任务，token 消耗月底才暴露。
稳定性脆弱：单一供应商限流或宕机时，调用方没有统一 failover 能力。
路由规则分散：A 团队按任务类型选模型，B 团队按租户等级选模型，逻辑重复且难审计。
配额治理滞后：Key 级限额无法表达“某租户每天 10 万 token、每分钟 200 次请求”。

AI 网关正是位于业务服务与模型供应商之间的基础设施。它不替代模型能力，而是把协议适配、模型路由、failover、配额、鉴权、观测和成本核算收敛成统一控制面。

方案对比：SDK 直连、轻量代理与 AI 网关

常见方案大致分三类：

业务 SDK 直连：接入最快，但路由、重试、配额散落在各服务中。
轻量反向代理：适合统一域名、密钥托管和日志审计，但缺乏租户级治理。
独立 AI 网关：在代理基础上增加动态路由、故障转移、配额计费、熔断和指标体系。

如果只有一个内部应用，轻量代理足够；如果要支持多业务线、多模型供应商和成本分摊，独立网关的长期收益更明显。

核心设计：路由、failover 与配额如何协同

建议把一次请求拆成四层：

接入层：统一使用类 OpenAI Chat Completions 协议，降低业务改造成本。
决策层：根据任务类型、租户等级、上下文长度、灰度标签和实时健康度选择模型。
执行层：按候选链调用，遇到可重试错误执行 failover，并记录每次尝试。
治理层：请求前做 QPS 与 token 预算检查，响应后按实际用量结算。

注意，配额不能只依赖供应商响应头。供应商限额是“你与供应商之间的约束”，而平台配额是“你与内部租户之间的约束”，两者必须分开。

Go 实战：实现一个可运行的最小 AI 网关

下面示例使用 Go 标准库加内存配额实现，便于直接运行。生产环境可将 MemoryQuota 替换为 Redis Lua 脚本，将路由配置放入 etcd、Consul 或配置中心。

第一步：定义模型与路由配置

package main

import (
	"context"
	"errors"
	"fmt"
	"sync"
	"time"
)

// ModelEndpoint 描述一个具体模型供应商端点
type ModelEndpoint struct {
	Name     string
	Vendor   string
	Priority int // 数值越小优先级越高
	Healthy  bool
}

// RouteRule 表示一条动态路由规则
type RouteRule struct {
	TaskType string // chat、summary、code 等
	Tier     string // free、pro、enterprise
	Models   []string
}

type ChatRequest struct {
	TenantID string
	Tier     string
	TaskType string
	Message  string
}

type ChatResponse struct {
	Model       string
	Content     string
	InputToken  int
	OutputToken int
}

第二步：实现动态路由与 failover

路由不只按静态配置选择，还要过滤不健康节点，并按优先级形成候选链。真实系统中还可以引入权重、延迟、区域和碳成本等因子。

type Router struct {
	mu        sync.RWMutex
	endpoints map[string]*ModelEndpoint
	rules     []RouteRule
}

func NewRouter() *Router {
	return &Router{endpoints: map[string]*ModelEndpoint{}}
}

func (r *Router) UpdateEndpoint(e ModelEndpoint) {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.endpoints[e.Name] = &e
}

func (r *Router) Candidates(taskType, tier string) []*ModelEndpoint {
	r.mu.RLock()
	defer r.mu.RUnlock()

	var modelNames []string
	for _, rule := range r.rules {
		if rule.TaskType == taskType && rule.Tier == tier {
			modelNames = rule.Models
			break
		}
	}

	var candidates []*ModelEndpoint
	for _, name := range modelNames {
		if ep, ok := r.endpoints[name]; ok && ep.Healthy {
			candidates = append(candidates, ep)
		}
	}
	return candidates
}

var ErrNoHealthyModel = errors.New("no healthy model available")

// CallWithFailover 按候选链依次尝试，避免首个供应商故障导致整体失败
func (r *Router) CallWithFailover(ctx context.Context, req ChatRequest, call func(context.Context, *ModelEndpoint) (*ChatResponse, error)) (*ChatResponse, error) {
	candidates := r.Candidates(req.TaskType, req.Tier)
	if len(candidates) == 0 {
		return nil, ErrNoHealthyModel
	}

	var lastErr error
	for _, ep := range candidates {
		select {
		case <-ctx.Done():
			return nil, ctx.Err()
		default:
		}

		resp, err := call(ctx, ep)
		if err == nil {
			return resp, nil
		}
		lastErr = err
		fmt.Printf("model %s failed: %v, try next\n", ep.Name, err)
	}
	return nil, fmt.Errorf("all models failed: %w", lastErr)
}

第三步：实现租户用量配额

配额至少包含两类：短周期请求数限流和周期 token 预算。请求前预占、请求后结算，是防止超额调用的关键。

type Quota struct {
	DailyToken int64 // 每日 token 上限
	UsedToken  int64
	ResetAt    time.Time
}

type MemoryQuota struct {
	mu sync.Mutex
	m  map[string]*Quota
}

func NewMemoryQuota() *MemoryQuota {
	return &MemoryQuota{m: map[string]*Quota{}}
}

// Reserve 在调用模型前预占额度；生产环境建议使用 Redis Lua 保证原子性
func (q *MemoryQuota) Reserve(tenant string, estimate, dailyLimit int64) error {
	q.mu.Lock()
	defer q.mu.Unlock()

	item, ok := q.m[tenant]
	now := time.Now()
	if !ok || now.After(item.ResetAt) {
		item = &Quota{DailyToken: dailyLimit, ResetAt: now.Add(24 * time.Hour)}
		q.m[tenant] = item
	}

	if item.UsedToken+estimate > item.DailyToken {
		return fmt.Errorf("tenant %s quota exceeded: used=%d limit=%d", tenant, item.UsedToken, item.DailyToken)
	}
	item.UsedToken += estimate
	return nil
}

// Settle 根据供应商返回的真实 token 修正预估值
func (q *MemoryQuota) Settle(tenant string, estimate, actual int64) {
	q.mu.Lock()
	defer q.mu.Unlock()

	if item, ok := q.m[tenant]; ok {
		item.UsedToken += actual - estimate
		if item.UsedToken < 0 {
			item.UsedToken = 0
		}
	}
}

第四步：组装网关处理链路

type Gateway struct {
	router *Router
	quota  *MemoryQuota
}

func (g *Gateway) Chat(ctx context.Context, req ChatRequest) (*ChatResponse, error) {
	// 简单按消息字符数估算输入 token，实际可用分词器精确计算
	estimate := int64(len(req.Message)) + 256
	if err := g.quota.Reserve(req.TenantID, estimate, 100000); err != nil {
		return nil, err
	}

	resp, err := g.router.CallWithFailover(ctx, req, func(ctx context.Context, ep *ModelEndpoint) (*ChatResponse, error) {
		// 这里替换为真实 HTTP 调用，并设置超时、追踪和请求体大小限制
		return &ChatResponse{
			Model:       ep.Name,
			Content:     "来自 " + ep.Name + " 的回答",
			InputToken:  int(estimate),
			OutputToken: 128,
		}, nil
	})
	if err != nil {
		g.quota.Settle(req.TenantID, estimate, 0)
		return nil, err
	}

	actual := int64(resp.InputToken + resp.OutputToken)
	g.quota.Settle(req.TenantID, estimate, actual)
	return resp, nil
}

func main() {
	router := NewRouter()
	router.rules = []RouteRule{
		{TaskType: "chat", Tier: "pro", Models: []string{"model-premium", "model-standard", "model-lite"}},
	}
	router.UpdateEndpoint(ModelEndpoint{Name: "model-premium", Vendor: "vendor-a", Priority: 1, Healthy: true})
	router.UpdateEndpoint(ModelEndpoint{Name: "model-standard", Vendor: "vendor-b", Priority: 2, Healthy: true})
	router.UpdateEndpoint(ModelEndpoint{Name: "model-lite", Vendor: "vendor-c", Priority: 3, Healthy: true})

	gw := &Gateway{router: router, quota: NewMemoryQuota()}
	resp, err := gw.Chat(context.Background(), ChatRequest{
		TenantID: "tenant-001",
		Tier:     "pro",
		TaskType: "chat",
		Message:  "解释一下什么是 AI 网关",
	})
	if err != nil {
		panic(err)
	}
	fmt.Println(resp.Model, resp.Content)
}

生产化建议：从示例到高可用系统

示例只展示核心骨架，上线前还需要补齐以下能力：

配置热更新：路由规则变更不应重启进程，建议配合版本号和灰度发布。
分布式配额：多实例部署时，内存计数会重复放行，应使用 Redis Lua 或独立限流服务完成原子扣减。
熔断与半开探测：某模型连续超时后短暂摘除，随后用少量流量试探恢复。
幂等与重试边界：非流式请求可用 request_id 做幂等；流式响应一旦已经向客户端输出，就不能透明重试。
指标维度：至少记录 tenant、task_type、model、vendor、status、latency、input_token、output_token、fallback_from。

避坑总结：多模型治理中的高频问题

1. 把所有错误都 failover，导致流量雪崩

常见现象：主模型因为上下文超长稳定返回 400，网关仍逐个重试所有候选模型，最终把低级模型也打满。修正方法是只对 429、5xx、网络超时、连接重置等可恢复错误转移；400、401、内容审核拒绝等确定性错误应直接返回。

2. 只在响应后扣配额，突发流量瞬间超额

常见现象：压测时大量请求同时进入，调用结束后才发现 token 已远超预算。修正方法是请求前基于 tokenizer 或保守上界预占，响应后按供应商 usage 结算差额；对超长输出还要设置 max_tokens。

3. 路由规则只看模型名，不看能力与约束

常见现象：把需要 128K 上下文的请求路由到 32K 模型，或者把函数调用请求发给不支持 tools 的模型。修正方法是在端点元数据中维护上下文长度、模态、工具调用、价格和延迟等级，路由时做能力匹配。

4. 重试引发重复扣费和重复副作用

常见现象：供应商实际已处理成功，但网关因读超时重试，造成重复生成甚至重复外部动作。修正方法是设置总超时和单模型超时，非幂等场景限制重试次数，并透传业务幂等键；涉及工具执行的请求默认不自动重试。

5. 配额维度过粗，无法定位成本责任

常见现象：平台总量超支，但只能看到某个 API Key 的消耗，无法定位团队和功能。修正方法是从入口解析租户、应用、环境、任务类型，并把这些维度写入日志、指标和费用报表。

结语

AI 网关的价值不是“多转发一次请求”，而是在模型供应快速变化时，为业务提供稳定、可控、可审计的访问边界。先用统一协议和候选链解决可用性，再用预占式配额和多维观测解决成本治理，最后通过热更新、熔断和灰度能力支撑规模化运营，这是多模型接入场景下更稳妥的落地路径。

免责声明

本工具与文章内容仅用于学习测试场景，优化建议仅供参考，请勿直接用于生产环境，由此产生的风险自行承担。

一文搞懂 AI 网关：多模型动态路由、failover 与配额治理 ​

问题背景：为什么不能让业务直连各家大模型 ​

方案对比：SDK 直连、轻量代理与 AI 网关 ​

核心设计：路由、failover 与配额如何协同 ​

Go 实战：实现一个可运行的最小 AI 网关 ​

第一步：定义模型与路由配置 ​

第二步：实现动态路由与 failover ​

第三步：实现租户用量配额 ​

第四步：组装网关处理链路 ​

生产化建议：从示例到高可用系统 ​

避坑总结：多模型治理中的高频问题 ​

1. 把所有错误都 failover，导致流量雪崩 ​

2. 只在响应后扣配额，突发流量瞬间超额 ​

3. 路由规则只看模型名，不看能力与约束 ​

4. 重试引发重复扣费和重复副作用 ​

5. 配额维度过粗，无法定位成本责任 ​

结语 ​