这篇能做出什么
做完这套配置，你会得到一条从 Google Cloud 内部调用 OpenAI 模型的稳定通路，并且它不依赖单一供应商：
- 统一入口：所有调用都走一个自建的 OpenAI 兼容网关（部署在 Cloud Run），业务代码里只认一个
base_url，将来更换供应商时，不必修改业务代码。 - 密钥不出云：OpenAI API Key 存放在 Secret Manager，只授权给一个最小权限服务账号，代码里没有任何明文密钥。
- 配额三层防线：供应商侧用量上限、GCP 预算告警、网关内的租户级限流，任何一层都不会让费用失控。
- 自动降级：主供应商超时、429 或 5xx 时，网关自动切到 Vertex AI 上的备用模型，调用方拿到的是正常响应，只是响应头里多了一个“本次命中哪个供应商”的标记。
- 灰度切换：新版本或新路由策略先给 10% 流量，对比错误率和 P95 延迟后再放量，出问题一条命令回滚。
- 能接 IDE：Antigravity 这类智能 IDE 以及各类支持自定义 Base URL 的客户端，都能接到这个网关上。
前置条件清单
开始前确认这些东西齐了：
1. 一个已开通结算账号的 Google Cloud 项目，你的账号具备 Owner 或足够的管理员权限。
2. 一个 OpenAI 平台的 API Key，并且该账号下已设置好用量上限。
3. 本地装好 gcloud CLI 并完成 gcloud auth login，能正常访问项目。
4. 基础工具：curl、Python 3.10+（仅本地调试用，部署走 Cloud Build 即可）。
5. 想清楚两件事：主模型 ID 和备用模型 ID。这两个 ID 请从各自官方文档的当前模型列表里取，本文用 PRIMARY_MODEL / FALLBACK_MODEL 占位，不写死具体型号。
6. Vertex AI 上你打算用的模型，在你选择的区域（region）里可用——这个以官方文档的区域可用性表格为准。
步骤一：打通项目与 API
先把需要的服务全部启用。命令里的 YOUR_PROJECT_ID 换成你的项目 ID。
```bash
gcloud config set project YOUR_PROJECT_ID
gcloud services enable \
run.googleapis.com \
secretmanager.googleapis.com \
aiplatform.googleapis.com \
cloudbuild.googleapis.com \
artifactregistry.googleapis.com \
logging.googleapis.com \
monitoring.googleapis.com \
billingbudgets.googleapis.com
```
启用完成后，顺手确认 Vertex AI 能用：
```bash
gcloud ai models list --region=us-central1 --limit=5
```
能返回列表（哪怕是空的）就说明鉴权链路通了。报 403 通常是 API 还没生效，等一两分钟重试。
步骤二：把密钥放进 Secret Manager
不要把 Key 写进环境变量或代码仓库。用标准输入写入，避免 shell 历史记录里留下明文：
```bash
read -s -p "OpenAI API Key: " OPENAI_API_KEY; echo
printf '%s' "$OPENAI_API_KEY" | gcloud secrets create openai-api-key \
--replication-policy=automatic \
--data-file=-
```
注意用 printf 而不是 echo，echo 会多带一个换行符，服务端会把它当成 Key 的一部分，然后给你一个莫名其妙的 401。
接着创建一个专用服务账号，只给它两样东西：读这个 Secret 的权限、调用 Vertex AI 的权限。
```bash
gcloud iam service-accounts create llm-gateway \
--display-name="LLM Gateway"
gcloud secrets add-iam-policy-binding openai-api-key \
--member="serviceAccount:llm-gateway@YOUR_PROJECT_ID.iam.gserviceaccount.com" \
--role="roles/secretmanager.secretAccessor"
gcloud projects add-iam-policy-binding YOUR_PROJECT_ID \
--member="serviceAccount:llm-gateway@YOUR_PROJECT_ID.iam.gserviceaccount.com" \
--role="roles/aiplatform.user"
```
轮换密钥时不要覆盖旧版本，而是新增一个版本，网关读 latest 别名即可平滑切换：
```bash
printf '%s' "$NEW_KEY" | gcloud secrets versions add openai-api-key --data-file=-
```
步骤三：写一个 OpenAI 兼容网关
网关的职责很单纯：收 OpenAI 格式的请求，先打主供应商，失败就换备用供应商，同时做限流和埋点。下面是最小可用版本。
先写依赖文件：
```text
requirements.txt
fastapi
uvicorn[standard]
httpx
google-auth
google-cloud-secretmanager
```
主体代码：
```python
main.py
import os
import time
import httpx
import google.auth
import google.auth.transport.requests
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
from google.cloud import secretmanager
PROJECT_ID = os.environ["GOOGLE_CLOUD_PROJECT"]
REGION = os.environ.get("VERTEX_REGION", "us-central1")
OPENAI_BASE = os.environ.get("OPENAI_BASE_URL", "https://api.openai.com/v1")
PRIMARY_MODEL = os.environ["PRIMARY_MODEL"]
FALLBACK_MODEL = os.environ["FALLBACK_MODEL"]
app = FastAPI()
---------- 取密钥（带进程内缓存） ----------
_sm = secretmanager.SecretManagerServiceClient()
_key_cache = {"value": None, "ts": 0.0}
KEY_TTL = 300 # 秒，到期重新读取 latest，实现密钥轮换
def openai_key() -> str:
now = time.time()
if _key_cache["value"] is None or now - _key_cache["ts"] > KEY_TTL:
name = f"projects/{PROJECT_ID}/secrets/openai-api-key/versions/latest"
_key_cache["value"] = _sm.access_secret_version(name=name).payload.data.decode()
_key_cache["ts"] = now
return _key_cache["value"]
---------- 取 Vertex 访问令牌 ----------
_creds, _ = google.auth.default(scopes=["https://www.googleapis.com/auth/cloud-platform"])
_auth_req = google.auth.transport.requests.Request()
def vertex_token() -> str:
if not _creds.valid:
_creds.refresh(_auth_req)
return _creds.token
---------- 两个后端的调用 ----------
async def call_openai(payload: dict, timeout: float) -> dict:
async with httpx.AsyncClient(timeout=timeout) as client:
resp = await client.post(
f"{OPENAI_BASE}/chat/completions",
headers={"Authorization": f"Bearer {openai_key()}"},
json=payload,
)
resp.raise_for_status()
return resp.json()
async def call_vertex(payload: dict, timeout: float) -> dict:
Vertex AI 提供 OpenAI 兼容的 chat/completions 端点，具体路径以官方文档为准
url = (f"https://{REGION}-aiplatform.googleapis.com/v1/projects/{PROJECT_ID}"
f"/locations/{REGION}/endpoints/openapi/chat/completions")
async with httpx.AsyncClient(timeout=timeout) as client:
resp = await client.post(
url,
headers={"Authorization": f"Bearer {vertex_token()}"},
json=payload,
)
resp.raise_for_status()
return resp.json()
---------- 熔断器 ----------
class Breaker:
def __init__(self, threshold: int = 5, cooldown: float = 30.0):
self.fails = 0
self.threshold = threshold
self.cooldown = cooldown
self.open_until = 0.0
def allow(self) -> bool:
return time.time() >= self.open_until
def record(self, ok: bool) -> None:
if ok:
self.fails = 0
return
self.fails += 1
if self.fails >= self.threshold:
self.open_until = time.time() + self.cooldown
self.fails = 0
primary_breaker = Breaker()
---------- 租户级令牌桶 ----------
BUCKETS: dict[str, tuple[float, float]] = {}
def allow(tenant: str, rate: float = 2.0, burst: int = 10) -> bool:
now = time.time()
tokens, last = BUCKETS.get(tenant, (float(burst), now))
tokens = min(burst, tokens + (now - last) * rate)
if tokens < 1.0:
BUCKETS[tenant] = (tokens, now)
return False
BUCKETS[tenant] = (tokens - 1.0, now)
return True
---------- 主路由 ----------
RETRYABLE = {408, 409, 429, 500, 502, 503, 504}
def tagged(data: dict, provider: str, trace: list[str]) -> JSONResponse:
return JSONResponse(
content=data,
headers={
"x-llm-provider": provider,
"x-llm-trace": ";".join(trace)[:512],
},
)
@app.post("/v1/chat/completions")
async def chat(request: Request):
body = await request.json()
tenant = request.headers.get("x-tenant-id", "default")
if not allow(tenant):
raise HTTPException(status_code=429, detail="tenant rate limited")
trace: list[str] = []
if primary_breaker.allow():
try:
data = await call_openai(body, timeout=30.0)
primary_breaker.record(True)
return tagged(data, "primary", trace)
except httpx.HTTPStatusError as exc:
code = exc.response.status_code
primary_breaker.record(False)
trace.append(f"primary:{code}")
if code not in RETRYABLE:
4xx 里的鉴权/参数错误，换供应商也没用，直接抛给调用方
raise HTTPException(status_code=code, detail=exc.response.text[:500])
except Exception as exc:
primary_breaker.record(False)
trace.append(f"primary:err:{type(exc).__name__}")
else:
trace.append("primary:breaker-open")
fallback_body = dict(body)
fallback_body["model"] = FALLBACK_MODEL
data = await call_vertex(fallback_body, timeout=30.0)
return tagged(data, "fallback", trace)
```
Dockerfile：
```dockerfile
FROM python:3-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY main.py .
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8080"]
```
步骤四：部署并验证鉴权
部署时关掉公开访问，只允许带身份令牌的调用：
```bash
gcloud run deploy llm-gateway \
--source . \
--region=asia-east1 \
--service-account=llm-gateway@YOUR_PROJECT_ID.iam.gserviceaccount.com \
--set-env-vars=GOOGLE_CLOUD_PROJECT=YOUR_PROJECT_ID,VERTEX_REGION=us-central1,PRIMARY_MODEL=YOUR_PRIMARY_MODEL,FALLBACK_MODEL=YOUR_FALLBACK_MODEL \
--no-allow-unauthenticated \
--min-instances=1 \
--timeout=120 \
--memory=512Mi
```
--min-instances=1 是为了避开冷启动——首包超时是降级误触发的主要来源之一。
调用时用当前用户身份生成令牌：
```bash
TOKEN=$(gcloud auth print-identity-token)
URL=$(gcloud run services describe llm-gateway --region=asia-east1 --format='value(status.url)')
curl -s "$URL/v1/chat/completions" \
-H "Authorization: Bearer $TOKEN" \
-H "Content-Type: application/json" \
-H "x-tenant-id: team-a" \
-d '{"model":"'"$PRIMARY_MODEL"'","messages":[{"role":"user","content":"用一句话解释什么是降级路由"}]}' \
-D /tmp/headers.txt
grep -i x-llm /tmp/headers.txt
```
响应头里出现 x-llm-provider: primary 就说明主链路通了。想看降级是否真的生效，临时把 PRIMARY_MODEL 改成一个不存在的名字重部署一次——网关会收到 4xx，此时按上面的逻辑它会直接抛出，不会降级；想验证 429/5xx 分支，更稳妥的办法是在测试环境把 OPENAI_BASE_URL 指向一个必定超时的地址，观察 x-llm-provider 变成 fallback。
步骤五：把配额和预算配到位
这一层是防止账单失控的关键，三层都要配。
供应商侧：在 OpenAI 平台后台设置月度用量上限，在 Vertex AI 的配额页面确认你所在区域的请求配额，必要时提工单申请提升。
GCP 预算告警：
```bash
gcloud billing budgets create \
--billing-account=YOUR_BILLING_ACCOUNT \
--display-name="LLM 月度预算" \
--budget-amount=500USD \
--threshold-rule=percent=0.5 \
--threshold-rule=percent=0.9 \
--threshold-rule=percent=1.0
```
预算告警只发通知、不会自动停服，所以真正兜底的是网关内的限流。上面代码里的令牌桶是按租户维度的，生产环境建议换成 Redis 或 Memorystore 做全局限流，否则多实例部署时每个实例各算各的。
再给网关加一条 5xx 比例的监控告警：
```bash
gcloud monitoring policies create --policy-from-file=policy.json
```
策略文件可以从控制台的 Alerting 页面创建一个样例后导出改，比手写省事。
步骤六：灰度切换与回滚
新策略不要一次性全量。Cloud Run 支持给新版本打标签并只分配部分流量：
```bash
部署一个新版本，先不给流量
gcloud run deploy llm-gateway \
--source . \
--region=asia-east1 \
--tag=canary \
--no-traffic
分 10% 流量给 canary
gcloud run services update-traffic llm-gateway \
--region=asia-east1 \
--to-tags=canary=10
```
观察两三个关键指标再决定：x-llm-provider 中 fallback 的占比、P95 延迟、5xx 比例。如果 fallback 占比在灰度版本里明显升高，说明新版本触发降级的条件太敏感，先调阈值而不是直接放量。
全量：
```bash
gcloud run services update-traffic llm-gateway \
--region=asia-east1 \
--to-latest
```
回滚就是把流量切回上一个稳定版本：
```bash
gcloud run services update-traffic llm-gateway \
--region=asia-east1 \
--to-revisions=PREVIOUS_REVISION_NAME=100
```
如果只想按租户灰度，不改基础设施，那就在网关代码里按 x-tenant-id 做哈希分流，命中白名单的走新路由，其余走老路由。这种方式更灵活，代价是逻辑要自己维护。
步骤七：接上 Antigravity 与其他客户端
智能 IDE 类工具通常在设置里有模型选择器，可选的模型清单会随版本更新，具体以界面显示和官方文档为准。如果目标模型已经出现在列表里，直接选就行，鉴权和计费走的是工具自身的配置。
如果列表里没有你要的模型，就把它指向自建网关。麻烦点在于 Cloud Run 部署成了私有服务，多数客户端不会自动附带 Google 身份令牌。解决办法是用本地代理把身份补齐：
```bash
gcloud run services proxy llm-gateway --region=asia-east1 --port=8081
```
然后在客户端的自定义模型配置里填：
- Base URL：
http://localhost:8081/v1 - API Key：任意非空字符串（真正的鉴权由代理完成）
- 模型名：填
PRIMARY_MODEL的值
同理，本地脚本、CI 任务、同事的笔记本都可以这么接。网关对外只暴露一个 OpenAI 兼容协议，内部怎么切换供应商、怎么限流，调用方完全不用关心——这就是避免被单一云绑定的具体做法。
常见坑与排错
401 来自 OpenAI
先怀疑密钥里的空白字符。用 printf 写入、用 gcloud secrets versions access latest --secret=openai-api-key | xxd | tail 检查结尾有没有多余字节。其次是 Secret 版本没授权给服务账号，看 Cloud Run 日志里的 Permission denied on secret。
403 来自 Vertex AI
两种可能：服务账号缺 roles/aiplatform.user，或者 aiplatform.googleapis.com 没启用。还有一种是模型在你选的 region 不可用，换区域或换模型。
429 分不清来源
看 x-llm-trace。primary:429 是供应商限流，tenant rate limited 是网关自己的令牌桶拦下的。前者要退避重试或提配额，后者要调令牌桶参数，处理方式完全不同。
降级被误触发
最常见的三个原因：Cloud Run 冷启动、网关超时设得比模型生成时间还短、把 400/401 这类不该降的错误也纳入了重试集合。前两个靠 min-instances 和放大 timeout 解决，第三个靠白名单错误码解决。
流式响应不能中途切换
一旦开始往客户端推 token，就没法回头换供应商了。正确做法是只在“首 token 之前”允许降级：先等第一个 chunk 到达，确认成功后再开始转发。否则会出现半句话来自 A、半句话来自 B 的诡异输出。
成本对不上账
两家的 token 计量口径不完全一致，缓存命中、推理 token 的计费方式也有差别。做法是给每个请求带上 x-tenant-id，在网关里记录输入输出 token 数，按租户汇总，别指望直接用某一家的账单反推。
密钥轮换后旧连接还在用老 Key
网关里加了 KEY_TTL 缓存，轮换后最多等一个 TTL 周期生效。如果要立刻生效，重启一个 revision 即可。
下一步建议
跑通之后，按这个顺序加固：
1. 建评测集。准备 50～100 条真实业务问题，每次灰度前跑一遍，比较两个供应商的输出质量，别只看延迟和错误率。
2. 加成本看板。把网关的 token 统计打到 Cloud Logging，再用 BigQuery 做聚合，按租户、按模型出日报。
3. 上语义缓存。高频重复问题命中缓存后直接返回，能显著降低两侧的调用量。
4. 多区域部署。网关在两个 region 各部署一份，前面挂全局负载均衡，单区域故障时不影响服务。
5. 接入 OpenTelemetry。把 trace 打通到 Cloud Trace，排查“这一次为什么降级了”会快很多。
整套东西的核心思路只有一句：业务代码只依赖一个协议，不依赖任何一家供应商。做到这一点，将来再多接一家、换掉一家，都是改网关的几十行代码，而不是全公司改调用点。