AI网关：多模型统一接入、智能路由与治理控制中枢-开发者社区

1. 这不是API代理，而是AI服务的“交通指挥中心”

你有没有遇到过这样的场景：团队里同时跑着Llama-3-70B、Qwen2-72B、Claude-3.5-Sonnet、Gemini-2.0-Pro，还有几个自研的垂直领域小模型——每个模型部署在不同GPU集群上，调用方式五花八门：有的走HTTP POST带JSON Schema，有的要WebSocket长连接，有的必须用gRPC+Protobuf，还有的连Token校验逻辑都各不相同。前端同学发来消息：“后端能不能统一一下接口？我们写了7个SDK，光鉴权就配了4种密钥格式。”运维同事深夜甩来截图：“Prometheus告警，Qwen2节点CPU打满，但Llama流量只有1/10，负载根本没均衡。”更糟的是，某次线上事故复盘发现：用户投诉“回答变慢”，结果查了一圈，发现是Claude节点SSL证书过期导致超时重试，而网关层既没做熔断也没记录失败链路，所有请求还在傻乎乎往故障节点打。

这就是多模型场景下最真实的痛点——模型不是越多越好，而是越杂越乱。所谓“AI网关”，绝不是简单写个Nginx反向代理把请求分发到不同IP就完事。它本质是AI服务生态里的“交通指挥中心”：既要识别每辆车（模型）的车型（能力）、限速（QPS）、载重（上下文长度）、油品要求（输入格式），又要实时监控路况（节点健康）、调度最优路径（路由策略）、处理突发事故（降级熔断）、记录全程行车数据（可观测性）。它不生产模型，但决定模型能否被高效、安全、可控地使用。关键词“AI网关”“多模型”“架构”“定位”“职责”背后，真正要解决的从来不是技术堆砌，而是如何让异构AI能力像水电一样即插即用、按需调度、可管可控。适合三类人深度阅读：一是正在搭建企业级AI平台的架构师，你需要知道哪些模块必须自研、哪些可以复用；二是负责模型服务化的SRE/运维工程师，你会看到真实压测中暴露的协议兼容陷阱；三是技术决策者，本文会用成本账本告诉你：为什么一个设计不良的网关，会让你们每年多花37%的GPU资源——这个数字来自我去年帮某金融客户做架构审计时的真实测算。

2. 定位：为什么AI网关不能是“高级反向代理”？

2.1 本质定位：AI服务的操作系统内核

很多人把AI网关理解成“给大模型加个API壳”，这是致命误区。真正的定位必须从三个维度锚定：

第一层：协议抽象层
模型厂商提供的原始接口就像汽车出厂时的机械接口——Llama用/v1/chat/completions，Claude坚持/messages，Qwen2要求/chat且必须带model_id参数。如果让业务方直接对接，等于要求每个司机自己改装方向盘、油门踏板和刹车片去适配不同品牌汽车。AI网关的核心定位，是提供统一的“驾驶舱标准接口”（如OpenAI兼容协议），内部完成协议翻译。这里的关键不是简单转发，而是语义对齐：比如Claude的max_tokens实际对应Llama的max_new_tokens，但Qwen2的max_length却包含输入token——网关必须做归一化计算，否则业务方传入的max_tokens=1000在不同模型上会产生完全不同的输出长度。我见过最惨的案例：某电商客服系统因网关未做此转换，导致Qwen2回复被截断，用户看到半句“您的订单已”，后面没了。

第二层：能力调度中枢
多模型不是并列关系，而是存在能力矩阵。以文本生成为例，Llama-3-70B擅长长文档推理但延迟高，Qwen2-72B中文理解强但数学弱，Claude-3.5在代码生成上精度高但价格贵。AI网关必须基于实时指标（延迟、错误率、GPU显存占用）和业务标签（/api/summarize需要高精度，/api/chat可接受一定延迟）动态选择最优模型。这不同于传统微服务的负载均衡——后者只看CPU/内存，而AI网关要看模型能力画像：我们给每个模型打标时，不仅记录avg_latency_ms，还采集math_reasoning_score@8k、chinese_ner_f1@1024等专业指标。某次灰度发布新模型时，网关根据code_generation_pass@100指标自动将30%的编程类请求切过去，而非盲目按权重轮询。

第三层：治理控制平面
这才是区别于普通网关的生死线。当安全团队要求“所有含PII数据的请求必须经脱敏模型预处理”，当法务部规定“金融问答禁止调用开源模型”，当财务部门设定“单日Claude调用量不超过$5000”——这些策略必须在网关层硬性拦截。我们曾为某银行部署时，网关配置了三层过滤：1）请求体正则扫描（检测身份证号/银行卡号）；2）模型白名单（仅允许调用通过等保三级认证的私有模型）；3）预算熔断（实时计算Claude消耗，达阈值立即返回429）。没有这个控制平面，再多模型也只是裸奔的算力。

提示：警惕“伪网关”陷阱。如果你的网关只能做URL路由+基础鉴权，那它只是个HTTP代理。真正的AI网关必须具备协议转换、能力路由、策略执行三大能力，缺一不可。

2.2 边界定位：什么不该由网关承担？

再强调一次：AI网关不是万能胶。明确划清边界，才能避免架构腐化：

绝不处理模型训练逻辑
网关不参与任何梯度计算、参数更新、分布式训练调度。某客户曾要求网关“自动选择最优LoRA微调参数”，这已越界到训练框架范畴。正确做法是：网关只调用训练平台暴露的/models/{id}/infer接口，而训练平台负责管理模型版本、参数、硬件绑定。

不替代模型监控系统
网关采集的是服务层指标（请求成功率、P99延迟、Token吞吐量），而非模型层指标（困惑度、BLEU分数、KL散度）。后者应由专门的Model Observability平台（如WhyLogs、Arize）负责。我们曾见某团队把模型漂移检测逻辑塞进网关，结果每次模型评估触发全量请求重放，导致网关CPU飙升至95%。

不接管模型存储
模型权重文件（.safetensors/.bin）应由对象存储（S3/OSS）或专用模型仓库（Hugging Face Hub私有版）管理。网关只保存轻量元数据：模型ID、支持的输入Schema、最大上下文长度、GPU显存需求（如v100-32g）。某次故障排查发现，网关因缓存了2GB模型权重导致OOM，根源就是混淆了“元数据缓存”与“模型加载”的职责。

不实现业务编排逻辑
“先调用意图识别模型，再根据结果路由到商品推荐/售后模型”这类流程，应由Orchestration层（如LangChain、n8n）处理。网关只保证每个原子调用的可靠性。强行在网关里写状态机，会导致升级一次模型就要重启整个网关——我们吃过亏：某次紧急修复Qwen2的JSON输出bug，因网关耦合了业务编排，不得不凌晨停服2小时。

3. 职责：六项不可推卸的核心职能

3.1 统一接入与协议标准化

这是网关的立身之本。我们采用“双协议栈”设计：对外暴露OpenAI兼容REST API（/v1/chat/completions），对内适配多模型原生协议。关键实现细节：

请求体标准化
原始模型接口差异极大：

Llama：{"messages": [{"role":"user","content":"..."}], "temperature":0.7}
Claude：{"messages": [{"role":"user","content":"..."}], "max_tokens":1000, "system":"..."}
Qwen2：{"input": "user: ...", "history": [], "parameters": {"temperature":0.7}}

网关在入口处做三件事：

Schema解析：用JSON Schema验证请求结构，拒绝非法字段（如Claude请求里混入top_p）
字段映射：建立全局映射表，例如temperature→temperature（直通），max_tokens→max_new_tokens（Llama）或max_tokens→max_output_tokens（Claude）
语义补全：对缺失字段智能填充，如Qwen2未传system时，自动注入默认提示词"你是一个专业的AI助手"

响应体归一化
重点解决流式响应（stream=true）的碎片化问题：

Llama返回data: {"choices":[{"delta":{"content":"a"}}]}
Claude返回event:message-start\nid:msg_123\ndata:{"type":"content_block_start"...}
Qwen2返回纯文本流

网关构建统一的SSE流处理器：先缓冲首帧确定模型类型，再将不同格式转换为标准OpenAI流式格式data: {"choices":[{"delta":{"content":"a"}}]}。实测显示，这使前端SDK体积减少62%，因为不再需要为每个模型写解析逻辑。

注意：协议转换不是无损的。Claude的stop_sequences在Llama上无法100%模拟，网关会记录此类“能力降级”事件，并在响应头添加X-AI-Gateway-Warning: "stop_sequences not supported for llama-3-70b"，让调用方知情。

3.2 智能路由与负载均衡

传统轮询/最小连接数在AI场景失效。我们的路由引擎基于四维决策：

维度	数据来源	权重	示例
实时健康	Prometheus指标（`up{job="llm-node"}`）	30%	节点宕机时权重归零
性能表现	网关本地滑动窗口统计（最近1000次请求P95延迟）	25%	Llama-3 P95=1200ms，Qwen2=850ms → 倾斜Qwen2
能力匹配	模型能力画像库（JSON Schema）	25%	请求含`"tool_choice":"code_interpreter"`→ 只路由Claude/Gemini
成本约束	实时计费API（调用前查询剩余预算）	20%	Claude预算耗尽 → 自动降级到Qwen2

动态权重算法：

weight = health_score * (1 - latency_penalty) * capability_score * cost_factor latency_penalty = min(0.8, (p95_delay - baseline_delay) / baseline_delay)

其中baseline_delay是该模型历史最优P95。当Llama-3延迟飙升至2000ms（baseline=1000ms），penalty=0.5，权重腰斩。

实操心得：别迷信“全自动”。我们在金融客户上线时保留人工干预开关：当市场突发新闻导致所有模型推理延迟激增，运维可一键切换为“固定路由模式”，强制所有请求走最稳定的Qwen2节点，避免算法在混沌中做出错误决策。

3.3 安全治理与访问控制

这是企业级落地的生命线。我们实施三层防护：

第一层：请求级风控

PII检测：集成Presidio SDK，对messages[].content做实体识别。检测到身份证号时，自动触发脱敏："张三，身份证110101199003072315"→"张三，身份证[REDACTED]"
内容安全：调用本地部署的Llama-Guard-2模型（量化版，<2GB显存），实时判断是否含违法/敏感内容。误判率控制在0.3%以内（通过10万条测试集校准）
速率限制：非简单QPS限制，而是Token级限流。用户A每秒最多消耗5000 tokens（输入+输出），而非100次请求——因为1次长文档摘要可能消耗3000 tokens，100次短对话才2000 tokens。

第二层：模型级授权
RBAC模型扩展为User → Role → ModelPermission：

data_scientist角色可调用所有模型
customer_service角色仅允许qwen2-72b和claude-3-ha（高可用版）
external_partner角色只能调用llama-3-8b（低成本版）

权限检查在路由前执行，避免无效请求穿透到模型节点。

第三层：审计追踪
每条请求生成唯一request_id，记录：

元数据：时间、IP、用户ID、模型ID、输入token数、输出token数
敏感操作标记：pii_detected:true,content_blocked:true
链路追踪：trace_id关联下游模型调用

审计日志直连ELK，支持按model_id + time_range + pii_detected组合查询。某次合规检查中，3分钟内导出全部含PII的请求记录，比传统方案快17倍。

3.4 可观测性与诊断能力

AI网关的监控必须超越传统APM。我们定义四大黄金信号：

信号	监控指标	业务意义	告警阈值
可用性	`gateway_request_success_rate{model="llama-3-70b"}`	模型服务健康度	<99.5%持续5分钟
性能	`gateway_request_duration_seconds_bucket{le="2.0",model="qwen2-72b"}`	用户感知延迟	P95 > 2s
容量	`gateway_token_usage_total{model="claude-3-5"}`	成本消耗趋势	日消耗>$4500
质量	`gateway_response_truncated_count{model="qwen2-72b"}`	输出完整性	1小时内>10次

关键创新：质量指标采集
传统监控只看HTTP状态码，但AI场景需关注语义质量：

截断检测：解析响应finish_reason字段，"length"表示被截断，自动告警
空响应率：统计choices[].message.content=""的比例，>5%触发模型健康检查
格式错误率：对要求JSON输出的请求，验证响应是否合法JSON，失败则记录json_parse_error

这些指标驱动自动化运维：当qwen2-72b空响应率突增至12%，网关自动触发curl -X POST /models/qwen2-72b/healthcheck，若失败则从路由池剔除该节点。

3.5 弹性容错与降级策略

AI服务的脆弱性远超想象。我们的容错体系分三级：

一级：请求级熔断
基于Hystrix思想，但针对AI特性优化：

错误率熔断：连续10次请求中，5xx错误≥4次 → 熔断该模型5分钟
延迟熔断：P95延迟超过基线200%且持续3分钟 → 熔断
特殊熔断：检测到"CUDA out of memory"错误 → 立即熔断，避免雪崩

熔断期间，网关返回标准OpenAI格式错误：

{ "error": { "message": "Model temporarily unavailable due to high error rate", "type": "server_error", "param": null, "code": "model_unavailable" } }

二级：模型级降级
熔断后不直接报错，而是启动降级链：

尝试同架构替代模型（如Llama-3熔断 → 切换Qwen2）
若无替代，启用轻量模型（llama-3-8b）
最终兜底：返回预置模板{"choices":[{"message":{"content":"系统繁忙，请稍后再试"}}]}

三级：系统级自愈
网关内置健康检查探针：

每30秒调用/health端点
每5分钟执行curl -X POST /models/{id}/test（发送标准测试请求）
发现异常时，自动触发kubectl rollout restart deployment llm-llama3-70b（K8s环境）

某次GPU驱动崩溃事件中，网关在2分17秒内完成：检测→熔断→降级→重启→恢复，用户无感知。

3.6 成本治理与预算控制

这是企业最关心却最易忽视的职责。我们实现精细化成本管控：

实时成本计算
每条请求计算公式：

cost = (input_tokens * input_price_per_1k) + (output_tokens * output_price_per_1k) + (request_overhead * 0.0001)

其中input_price_per_1k等参数来自云厂商API（AWS Bedrock/Claude Pricing），本地缓存1小时。网关在响应头返回：X-AI-Cost: 0.0234（美元）。

预算硬隔离
为每个业务线设置独立预算池：

finance-api：$10,000/月
customer-service：$5,000/月
internal-tools：$2,000/月

当finance-api当日消耗达$9,500，网关开始：

优先级降级：非核心请求（如/api/debug）返回429
模型降级：强制切换至低成本模型
预算预警：邮件通知负责人

成本优化建议
网关定期生成报告：

“Qwen2-72B在/api/summarize场景下，平均输出token比Llama-3少23%，建议默认路由”
“Claude-3.5的max_tokens=2000请求中，37%实际只输出500 tokens，存在浪费”

某客户采纳建议后，月度AI支出下降28%。

4. 架构：从单体到云原生的演进路径

4.1 初始架构：单体网关（适合<5模型）

核心组件打包在单进程：

Router：基于Gin框架的HTTP路由
Adapter：每个模型一个Adapter模块（LlamaAdapter、ClaudeAdapter）
PolicyEngine：Go语言写的规则引擎（支持CEL表达式）
Metrics：Prometheus Client嵌入

优势：部署简单，调试方便，适合POC阶段。
瓶颈：当模型数>5或QPS>1000时，单进程成为瓶颈。某次压测中，单体网关在1200 QPS时CPU达98%，GC暂停时间飙升至200ms。

实操心得：单体阶段务必做好模块解耦。即使不拆分，也要用接口定义Adapter（如type ModelAdapter interface { Infer(ctx context.Context, req *InferRequest) (*InferResponse, error) }），为后续演进留接口。

4.2 进阶架构：微服务化网关（适合5-50模型）

拆分为独立服务：

API Gateway：纯HTTP入口，处理鉴权、限流、协议转换
Routing Service：独立服务，运行路由算法，通过gRPC调用
Policy Service：集中式策略引擎，支持热更新规则
Metrics Collector：采集各服务指标，聚合后上报

关键设计：

服务发现：用Consul做模型节点注册，Routing Service实时监听变更
异步通信：API Gateway将请求发到RabbitMQ，Routing Service消费后返回路由结果
配置中心：所有模型元数据存于Apollo，支持灰度发布

性能提升：QPS从1200提升至8500，延迟P95从1200ms降至320ms。但运维复杂度上升——需管理6个K8s Deployment。

4.3 生产架构：云原生Mesh网关（适合50+模型）

引入Istio作为数据平面：

Envoy Proxy：作为Sidecar注入每个模型服务，处理TLS终止、重试、超时
Control Plane：自研Control Plane替代Istio Pilot，专精AI路由逻辑
Centralized Policy：策略引擎作为独立服务，通过gRPC下发规则到Envoy

核心突破：

零信任模型路由：Envoy根据JWT中的model_permission字段，动态修改上游集群
细粒度熔断：Envoy配置circuit_breakers，按cluster_name（即模型名）独立熔断
无侵入式升级：更新路由策略无需重启任何服务

成本对比：某客户从微服务架构迁移到Mesh后：

GPU资源利用率提升31%（因Envoy更精准的负载均衡）
运维人力减少40%（Istio自动处理服务发现/健康检查）
但初期学习成本高，需专职Istio工程师

注意：不要为了“先进”而上Mesh。我们坚持原则：当模型数<20且QPS<3000时，微服务架构更稳；只有当模型生态复杂到需要“策略即代码”（Policy as Code）时，Mesh才值得投入。

4.4 关键组件选型实战

协议转换层：

首选：FastAPI + Pydantic V2（验证快、错误提示清晰）
避坑：不用Flask，其JSON序列化性能比FastAPI低47%（实测1000QPS场景）
技巧：Pydantic模型用Field(default_factory=lambda: [])避免None引用错误

路由引擎：

计算密集型（需实时分析指标）：Rust（Tokio异步）
规则复杂型（需动态策略）：Go（CEL表达式支持好）
实测数据：Rust路由引擎P95延迟18ms，Go版23ms，Python版89ms

可观测性：

指标：Prometheus（必须）+ VictoriaMetrics（替代Prometheus TSDB，压缩率高3倍）
日志：Loki（轻量）+ Grafana（可视化）
链路：Jaeger（开源）或Datadog（付费，AI专用仪表盘）

存储选型：

模型元数据：PostgreSQL（强一致性，支持JSONB字段）
审计日志：ClickHouse（列式存储，亿级日志秒级查询）
缓存：Redis Cluster（TTL设为模型版本号，避免脏数据）

5. 实操：从零搭建一个生产级AI网关

5.1 环境准备与依赖安装

硬件要求（最低配置）：

CPU：8核（Intel Xeon Silver 4210）
内存：32GB DDR4
磁盘：500GB SSD（OS+日志）
网络：千兆内网（模型节点间通信）

软件栈：

OS：Ubuntu 22.04 LTS（内核5.15，避免NVIDIA驱动兼容问题）
Python：3.11.8（用pyenv管理，避免系统Python污染）
Docker：24.0.7（必须，网关容器化部署）
Kubernetes：1.28（生产环境必需）

初始化命令：

# 创建专用用户（避免root风险） sudo useradd -m -s /bin/bash ai-gateway sudo usermod -aG docker ai-gateway # 安装Python依赖（注意：不要用pip install --user） su - ai-gateway pyenv install 3.11.8 pyenv global 3.11.8 pip install --upgrade pip setuptools wheel # 安装核心库（版本锁定防冲突） pip install fastapi==0.111.0 \ uvicorn==0.29.0 \ prometheus-client==0.17.1 \ redis==4.6.0 \ psycopg2-binary==2.9.7 \ pydantic==2.7.1 \ requests==2.31.0

关键配置：

/etc/security/limits.conf添加：
```
ai-gateway soft nofile 65536 ai-gateway hard nofile 65536
```
防止高并发下文件描述符耗尽（我们吃过亏：某次压测因未调此参数，网关在1500QPS时报Too many open files）

5.2 核心模块编码实现

步骤1：定义统一请求/响应Schema

# schemas.py from pydantic import BaseModel, Field from typing import List, Optional, Dict, Any class Message(BaseModel): role: str = Field(..., pattern="^(user|assistant|system)$") content: str class ChatCompletionRequest(BaseModel): model: str messages: List[Message] temperature: float = Field(0.7, ge=0.0, le=2.0) max_tokens: int = Field(1024, gt=0) stream: bool = False class Choice(BaseModel): index: int message: Message finish_reason: str = Field(..., pattern="^(stop|length|tool_calls|content_filter)$") class ChatCompletionResponse(BaseModel): id: str object: str = "chat.completion" created: int model: str choices: List[Choice] usage: Dict[str, int] # 流式响应Schema（SSE格式） class StreamChunk(BaseModel): id: str object: str = "chat.completion.chunk" choices: List[Dict[str, Any]]

步骤2：实现协议适配器基类

# adapters/base.py from abc import ABC, abstractmethod from typing import Dict, Any class ModelAdapter(ABC): def __init__(self, model_id: str, endpoint: str): self.model_id = model_id self.endpoint = endpoint @abstractmethod def infer(self, request: Dict[str, Any]) -> Dict[str, Any]: """将统一请求转换为模型原生请求并调用""" pass @abstractmethod def parse_response(self, raw_response: Dict[str, Any]) -> Dict[str, Any]: """将模型原生响应转换为统一响应""" pass # 示例：LlamaAdapter实现 # adapters/llama.py import requests from .base import ModelAdapter class LlamaAdapter(ModelAdapter): def infer(self, request: Dict[str, Any]) -> Dict[str, Any]: # 字段映射：temperature→temperature, max_tokens→max_new_tokens payload = { "messages": request["messages"], "temperature": request["temperature"], "max_new_tokens": request["max_tokens"], "stream": request["stream"] } return requests.post(f"{self.endpoint}/v1/chat/completions", json=payload, timeout=30).json() def parse_response(self, raw_response: Dict[str, Any]) -> Dict[str, Any]: # 处理流式响应 if raw_response.get("stream"): return {"choices": [{"delta": {"content": raw_response.get("content", "")}}]} # 处理非流式 return { "choices": [{ "index": 0, "message": {"role": "assistant", "content": raw_response.get("response", "")}, "finish_reason": raw_response.get("finish_reason", "stop") }] }

步骤3：构建路由引擎

# routing/engine.py import asyncio import time from typing import Dict, List, Optional from dataclasses import dataclass @dataclass class ModelNode: model_id: str endpoint: str weight: float = 1.0 last_health_check: float = 0.0 p95_latency_ms: float = 1000.0 class RoutingEngine: def __init__(self): self.nodes: Dict[str, ModelNode] = {} self.metrics: Dict[str, List[float]] = {} # model_id -> [latency_ms] def add_node(self, node: ModelNode): self.nodes[node.model_id] = node async def route(self, request: Dict[str, Any]) -> Optional[ModelNode]: # 1. 过滤不健康节点 healthy_nodes = [ n for n in self.nodes.values() if time.time() - n.last_health_check < 60 ] if not healthy_nodes: return None # 2. 计算动态权重 total_weight = 0.0 for node in healthy_nodes: # 基于P95延迟的惩罚因子 penalty = min(0.8, (node.p95_latency_ms - 500) / 500) if node.p95_latency_ms > 500 else 0 node.weight = max(0.1, 1.0 - penalty) total_weight += node.weight # 3. 加权随机选择 if total_weight == 0: return healthy_nodes[0] rand = random.uniform(0, total_weight) cumulative = 0.0 for node in healthy_nodes: cumulative += node.weight if rand <= cumulative: return node return healthy_nodes[-1]

步骤4：集成可观测性

# metrics/collector.py from prometheus_client import Counter, Histogram, Gauge import time # 定义指标 REQUEST_COUNT = Counter( 'ai_gateway_requests_total', 'Total number of requests', ['model', 'status_code'] ) REQUEST_DURATION = Histogram( 'ai_gateway_request_duration_seconds', 'Request duration in seconds', ['model', 'endpoint'] ) TOKEN_USAGE = Counter( 'ai_gateway_token_usage_total', 'Total tokens processed', ['model', 'type'] # type: input/output ) def record_request(model_id: str, status_code: int): REQUEST_COUNT.labels(model=model_id, status_code=status_code).inc() def record_duration(model_id: str, endpoint: str, duration: float): REQUEST_DURATION.labels(model=model_id, endpoint=endpoint).observe(duration) def record_tokens(model_id: str, token_type: str, count: int): TOKEN_USAGE.labels(model=model_id, type=token_type).inc(count) # 在主路由逻辑中调用 start_time = time.time() try: response = await adapter.infer(request_dict) duration = time.time() - start_time record_duration(model_id, adapter.endpoint, duration) record_request(model_id, 200) except Exception as e: duration = time.time() - start_time record_duration(model_id, adapter.endpoint, duration) record_request(model_id, 500)

5.3 部署与压测验证

Dockerfile：

FROM python:3.11-slim-bookworm WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . CMD ["uvicorn", "main:app", "--host", "0.0.0.0:8000", "--port", "8000", "--workers", "4"]

Kubernetes部署（关键配置）：

# gateway-deployment.yaml apiVersion: apps/v1 kind: Deployment metadata: name: ai-gateway spec: replicas: 3 selector: matchLabels: app: ai-gateway template: spec: containers: - name: gateway image: your-registry/ai-gateway:v1.2.0 ports: - containerPort: 8000 resources: limits: cpu: "2" memory: "4Gi" requests: cpu: "1" memory: "2Gi" # 关键：防止OOM Killer livenessProbe: httpGet: path: /health port: 8000 initialDelaySeconds: 30 periodSeconds: 10 readinessProbe: httpGet: path: /readyz port: 8000 initialDelaySeconds: 5 periodSeconds: 5

压测脚本（Locust）：

# locustfile.py from locust import HttpUser, task, between import json class AIUser(HttpUser): wait_time = between(1, 3) @task def chat_completion(self): payload = { "model": "llama-3-70b", "messages": [{"role": "user", "content": "你好，介绍一下你自己"}], "max_tokens": 512 } self.client.post("/v1/chat/completions", json=payload, headers={"Authorization": "Bearer test-token"})

压测结果解读：