AI网关实战：从零搭建到502错误排查

AI 网关（AI Gateway）现在几乎是 AI 应用开发里绕不开的一层。很多人在本地调试 Codex、Cursor、PyCharm AI 插件这类工具时，经常遇到502 Bad Gateway，问题通常不在模型本身，而在网关地址、上游端口、令牌头这几个环节。项目标题里“We removed ALL fees from our AI gateway”这句描述，如果翻译过来是“我们把 AI 网关的全部费用移除了”，更值得展开的其实不是“免费”这个结果，而是网关在真实工程里承担的职责：统一转发、路由、限流、密钥管理、成本计量和异常隔离。下面围绕一条主线展开：为什么 AI 应用需要网关，如何搭一个本地可运行的最小网关，关键参数怎么配，502 怎么查，以及生产环境怎么落地。完成后你会得到一份能继续扩展的网关骨架，以及一套针对本地 AI 工具连接报错的排查方法。

1. 先理解 AI 网关在应用里的位置，再决定要不要引入

1.1 直连模型 API 的工程问题

早期很多 AI 功能是这么做的：在业务代码里直接引入 OpenAI 或其他模型提供方的 SDK，配置一个api_key和base_url，然后到处调用chat.completions.create。这种方式的优点是上手快，缺点是模型接入和业务逻辑完全耦合。

直连方式在项目变大之后会暴露几类问题：

• 每个服务各自保存一份模型 API Key，Key 泄露后无法从一处快速吊销。
• 所有请求直接打向上游，没有统一限流，某个接口流量抖动会拖垮整条链路。
• 想换模型供应商，需要改所有调用方的代码和配置。
• 请求日志分散在各服务里，没法统计每个用户、每个部门、每个功能消耗了多少 token。
• 上游模型接口不稳定时，每个服务都要自己实现重试、降级和超时控制，重复劳动还容易写错。

这些问题不是模型本身的问题，而是“缺少一层统一边界”的问题。AI 网关做的就是把这层边界补上：客户端只面向网关，网关再面向一个或多个模型供应商。

1.2 网关收敛的职责，不只是转发请求

AI 网关和普通 API 网关在转发层面很像，但多了模型场景特有的能力。一个完整 AI 网关至少承担这些职责：

• 统一入口：客户端只调/v1/chat/completions或/v1/responses，不用关心真正的模型服务在哪个地址。
• 模型路由：同一个请求里的model字段，可以映射到不同供应商、不同区域的模型实例。
• Key 管理：上游模型的真实 API Key 放在网关侧，客户端只使用网关签发的 Token。
• 认证授权：校验请求是否带网关 Token，是否来自合法的用户或应用。
• 限流配额：按用户、按应用、按模型限制每分钟请求数、每分钟 token 数、每日消耗预算。
• 缓存：对重复的 prompt 前缀或完全相同的请求做缓存，降低 token 消耗和延迟。
• 超时与重试：统一设置连接超时、读超时、总超时，并实现安全的失败重试。
• 流式转发：支持 SSE 流式响应，不把整个响应缓冲到内存。
• 可观测性：记录 request_id、模型名称、token 用量、响应耗时、上游状态码。
• 成本计量：把每个请求的usage字段落库，按维度核算 credits 或费用。

如果只是转发请求，用 Nginx 的反向代理也能做到一部分。AI 网关的价值在于它理解模型请求的结构，能解析messages、model、stream、usage这些字段，所以能在业务语义层面做路由、限流和计费。

1.3 “移除费用”背后真正要解决的是成本透明

标题里的 fees 可以有两层含义。第一层是网关产品自身是否对调用量额外收费，这属于商业策略；第二层更贴近工程：模型 API 本身按 token 计费，不同平台还会用 credits 作为计量单位，调用方经常不清楚一笔请求花掉了多少额度。

在 AI 平台语境里，credits 是一种配额计量单位，通常和 token 消耗量、模型单价、请求次数绑定。1 个 credits 具体对应多少 token，不同平台规则不同。网关的价值不在于让 credits 消失，而在于把消耗变成可见数据：每个请求消耗了哪些模型、多少 prompt token、多少 completion token、对应多少 credits。没有这层数据，成本优化无从谈起。

所以“移除所有费用”更准确的工程解释是：让成本变得透明、可控、不被重复收取，而不是让模型调用真的零成本。后面第 6 章会专门讲如何在网关层做成本治理。

2. 用 FastAPI 写一个最小 AI 网关，本地 Ollama 做上游

2.1 环境准备

要跑通本文的最小网关，不需要注册任何云厂商 API，用本地模型服务作为上游就可以。这样既方便学习，也不会产生真实费用。

建议准备以下环境：

安装命令：

pip install fastapi uvicorn httpx

Ollama 安装完成后，先拉一个可用的小模型：

ollama pull qwen2.5:7b ollama serve

Ollama 默认监听在http://127.0.0.1:11434，它提供了 OpenAI 兼容端点/v1/chat/completions。可以用下面的命令先确认上游可用：

curl http://127.0.0.1:11434/v1/chat/completions \ -H "Content-Type: application/json" \ -d '{ "model": "qwen2.5:7b", "messages": [{"role": "user", "content": "你好，请回复OK"}], "stream": false }'

如果这个请求能返回 JSON，说明上游正常。如果在本地确实没有可用的 OpenAI 兼容服务，也可以把UPSTREAM_BASE指向其他兼容服务地址，但要注意不同供应商的鉴权头可能不同。

2.2 项目结构

最小网关项目结构如下：

ai-gateway-demo/ ├── app/ │ ├── __init__.py │ └── main.py ├── requirements.txt └── README.md

main.py是网关的全部核心逻辑，包括认证、转发、日志和流式响应。为了让代码尽量短，这里先不拆路由和服务层，实际生产项目建议按模块拆分。

2.3 网关核心代码

下面的代码实现了一个最小可运行的 AI 网关，支持普通响应和 SSE 流式响应。它会把客户端请求转发到UPSTREAM_BASE指向的上游服务，并把上游的usage字段记录到日志里。

import os import time import uuid from fastapi import FastAPI, Request, HTTPException from fastapi.responses import StreamingResponse import httpx app = FastAPI(title="minimal-ai-gateway") GATEWAY_API_KEY = os.getenv("GATEWAY_API_KEY", "gw-local-123") UPSTREAM_BASE = os.getenv("UPSTREAM_BASE", "http://127.0.0.1:11434") UPSTREAM_API_KEY = os.getenv("UPSTREAM_API_KEY", "ollama") def build_upstream_headers(): return { "Authorization": f"Bearer {UPSTREAM_API_KEY}", "Content-Type": "application/json", } @app.middleware("http") async def add_request_id(request: Request, call_next): request_id = request.headers.get("X-Request-ID", uuid.uuid4().hex[:12]) request.state.request_id = request_id response = await call_next(request) response.headers["X-Request-ID"] = request_id return response @app.get("/healthz") async def healthz(): return {"status": "ok", "service": "ai-gateway"} @app.post("/v1/chat/completions") async def chat_completions(request: Request): auth = request.headers.get("Authorization", "") if auth != f"Bearer {GATEWAY_API_KEY}": raise HTTPException(status_code=401, detail="gateway token missing") payload = await request.json() stream = payload.get("stream", False) request_id = request.state.request_id start = time.monotonic() upstream_headers = build_upstream_headers() if stream: async def stream_proxy(): async with httpx.AsyncClient(timeout=httpx.Timeout(300.0, connect=10.0)) as client: async with client.stream( "POST", f"{UPSTREAM_BASE}/v1/chat/completions", json=payload, headers=upstream_headers, ) as upstream_resp: if upstream_resp.status_code >= 400: body = await upstream_resp.aread() raise HTTPException( status_code=upstream_resp.status_code, detail=body.decode(), ) async for chunk in upstream_resp.aiter_bytes(): yield chunk duration_ms = int((time.monotonic() - start) * 1000) print( f"request_id={request_id} " f"status={upstream_resp.status_code} " f"duration_ms={duration_ms} stream=true" ) return StreamingResponse(stream_proxy(), media_type="text/event-stream") async with httpx.AsyncClient(timeout=httpx.Timeout(120.0, connect=10.0)) as client: resp = await client.post( f"{UPSTREAM_BASE}/v1/chat/completions", json=payload, headers=upstream_headers, ) duration_ms = int((time.monotonic() - start) * 1000) if resp.status_code >= 400: print( f"request_id={request_id} " f"status={resp.status_code} " f"duration_ms={duration_ms}" ) raise HTTPException(status_code=resp.status_code, detail=resp.text[:2000]) data = resp.json() usage = data.get("usage", {}) print( f"request_id={request_id} " f"method=POST path=/v1/chat/completions " f"upstream={UPSTREAM_BASE} " f"status={resp.status_code} " f"duration_ms={duration_ms} " f"prompt_tokens={usage.get('prompt_tokens', 'N/A')} " f"completion_tokens={usage.get('completion_tokens', 'N/A')}" ) return data

代码有几点需要解释：

• 网关自己的 Token 是GATEWAY_API_KEY，客户端调用时必须携带，否则返回 401。这个配置从环境变量读取，避免把密钥写死在代码里。
• UPSTREAM_BASE是真正的模型服务地址，这里是 Ollama。
• 普通请求用httpx.AsyncClient一次性转发，超时设置为 120 秒。流式请求单独走client.stream，按块把上游数据转发回客户端。
• 响应里的usage是模型服务返回的 token 计费数据，网关打印到日志，后续可以做成本统计。
• 打印日志用的是print，只适合演示，生产环境应该换成标准logging。

2.4 启动网关并完成第一次转发

启动网关：

uvicorn app.main:app --host 127.0.0.1 --port 8080

健康检查：

curl http://127.0.0.1:8080/healthz

调用一次非流式对话：

curl http://127.0.0.1:8080/v1/chat/completions \ -H "Authorization: Bearer gw-local-123" \ -H "Content-Type: application/json" \ -d '{ "model": "qwen2.5:7b", "messages": [{"role": "user", "content": "你好，请用一句话介绍什么是 AI 网关"}], "stream": false }'

如果一切正常，会返回类似下面的 JSON：

{ "id": "chatcmpl-xxx", "object": "chat.completion", "model": "qwen2.5:7b", "choices": [ { "index": 0, "message": { "role": "assistant", "content": "AI 网关是统一的模型访问入口，负责路由、鉴权、限流和成本管理。" }, "finish_reason": "stop" } ], "usage": { "prompt_tokens": 32, "completion_tokens": 23, "total_tokens": 55 } }

这一步跑通，说明“客户端 -> 网关 -> 上游模型”整条链路是通顺的。后面再改配置、接线、排查 502，都是在验证这个链路里的某个环节。

3. 网关配置里最容易出问题的参数：路由、限流、超时、重试

3.1 模型路由与上游分组

真实场景里，一个网关往往对接多个模型供应商。比如：普通文本问答走 A 供应商，代码生成走 B 供应商，图片理解走 C 供应商。客户端不必知道这些映射关系，它只需要传模型名，网关负责路由。

可以设计一个简单的路由配置：

routes: - model: "qwen2.5:7b" upstreams: - url: "http://127.0.0.1:11434" priority: 1 - url: "http://127.0.0.1:11435" priority: 2 fallback_model: "llama3.2:1b"

参数含义：

路由决策不能只在代码里写死。生产环境建议把路由表放到配置文件或配置中心，让运维可以不改代码就调整模型供应商。学习环境可以先在内存字典里维护：

MODEL_ROUTES = { "qwen2.5:7b": ["http://127.0.0.1:11434"], "gpt-oss": ["http://127.0.0.1:11434"], }

路由在这里的作用是隔离上游变化：上游地址变了，只改网关配置，客户端无感知。

3.2 限流与配额

限流是网关最容易漏掉但生产必需要的功能。没有限流，一个异常流量或一个恶意调用方可以把整月的模型预算几分钟内耗尽。

学习环境可以用一个简单的内存滑动窗口实现：

import time from collections import defaultdict class SlidingWindowLimiter: def __init__(self, max_requests: int, window_seconds: int = 60): self.max_requests = max_requests self.window_seconds = window_seconds self.requests = defaultdict(list) def check(self, key: str) -> bool: now = time.monotonic() while self.requests[key] and self.requests[key][0] <= now - self.window_seconds: self.requests[key].pop(0) if len(self.requests[key]) >= self.max_requests: return False self.requests[key].append(now) return True

这个实现适合单实例学习验证。生产环境不能用单机内存做限流，因为网关多半会多副本部署，请求会分散到多个进程，限流计数必须用 Redis 这类共享存储。常见方案是 Redis + Lua 脚本实现原子计数，或直接用成熟网关组件。

限流维度至少要考虑：

• 按调用方：每个 API Key 每分钟请求数。
• 按模型：某个高成本模型每分钟 token 数。
• 按用户或团队：每日累计预算，达到阈值直接拒绝或降级。
• 按并发：同一时刻在途的流式请求数量，防止大量长连接打满资源。

配置示例：

rate_limit: request_per_minute: 60 token_per_minute: 100000 daily_credits_limit: 1000

超过限流时，网关应返回429 Too Many Requests，并且在响应头带上Retry-After，而不是把请求继续转发到上游。

3.3 超时、重试与流式转发

超时配置是 502 的高发来源。网关需要区分几种超时：

流式请求尤其要注意：模型生成慢时，两个事件之间可能间隔很久。如果读超时设置成 10 秒，模型思考超过 10 秒没有输出，网关就会断开连接，客户端看到的就是连接中断或 502。

重试策略要谨慎。模型接口的 POST 请求往往不是安全的，盲目重试可能导致重复扣费、重复创建内容。安全做法是：

• 只对连接超时、上游 5xx 这类可恢复错误重试。
• 重试次数限制在 1-2 次。
• 使用幂等键，让上游可以识别重复请求。
• 对 4xx 错误不要重试，因为那是调用参数或权限问题，重试没有意义。

网关内部记录重试次数，客户端可以在响应头看到类似X-Retry-Count: 1的信息，方便排查时判断这次响应是否经历过多路尝试。

3.4 密钥管理与令牌传递

密钥泄露是 AI 网关最直接的资损风险。最小网关里已经演示了用环境变量管理GATEWAY_API_KEY和UPSTREAM_API_KEY，生产环境还要更进一步：

• 网关签发自己的 Token，客户端只持有网关 Token，不知道真实上游 Key。
• 上游 Key 用环境变量、密钥管理服务或挂载文件注入，不进代码仓库。
• 不同供应商的 Key 分开存储，防止一个 Key 泄露导致全部模型可用。
• Token 支持过期轮换，轮换时要考虑客户端连接中的请求，避免旧 Token 被立即吊销导致大面积失败。
• 日志里不要打印完整 Authorization 头，只记录 Token 的前几位或哈希值。

很多本地 AI 工具报错unauthorized: gateway token missing，本质就是客户端请求里没有带网关 Token，或者 Token 从配置文件里读取失败。这类问题在第 5 章会再展开。

4. 从健康检查到流式响应，验证网关不是“能启动”就行

4.1 健康检查

网关服务能启动，不等于它能正常转发。健康检查至少应该确认两件事：进程在监听，上游可连通。

简单版本只返回进程状态：

curl http://127.0.0.1:8080/healthz

返回：

{"status": "ok", "service": "ai-gateway"}

生产版本建议把上游连通状态也放进去，例如返回：

{ "status": "ok", "upstream": "http://127.0.0.1:11434", "upstream_status": "reachable", "latency_ms": 8 }

这样负载均衡器发现上游不可达时，可以直接把网关实例摘掉，避免用户请求打到网关后才发现上游挂了。

4.2 非流式请求验证

非流式验证用第一节的 curl 命令就可以。需要关注几个检查点：

• 状态码必须是 200。
• 返回 JSON 中有choices[0].message.content。
• usage里有prompt_tokens、completion_tokens、total_tokens。
• 响应头里有X-Request-ID，且和网关日志中的 request_id 一致。

如果请求返回 401，先看Authorization头有没有写成Bearer gw-local-123。很多新手漏了Bearer前缀，导致网关认不出令牌。

4.3 流式请求验证

流式请求验证用-N参数禁用 curl 缓冲：

curl -N http://127.0.0.1:8080/v1/chat/completions \ -H "Authorization: Bearer gw-local-123" \ -H "Content-Type: application/json" \ -d '{ "model": "qwen2.5:7b", "messages": [{"role": "user", "content": "从 1 数到 5，每次停顿一下"}], "stream": true }'

预期输出是 SSE 格式，每行以data:开头，最后一行是data: [DONE]：

data: {"id":"chatcmpl-xxx","object":"chat.completion.chunk","choices":[{"delta":{"content":"1"}}]} data: {"id":"chatcmpl-xxx","object":"chat.completion.chunk","choices":[{"delta":{"content":"2"}}]} data: [DONE]

如果 curl 一直等不到输出，说明流式转发没有把上游数据及时返回。常见原因是上游不支持流式，或网关把流式请求当成普通请求缓冲后再返回。

4.4 限流与异常分支验证

限流验证可以快速写一个循环：

for i in $(seq 1 70); do curl -s -o /dev/null -w "%{http_code}\n" \ http://127.0.0.1:8080/v1/chat/completions \ -H "Authorization: Bearer gw-local-123" \ -H "Content-Type: application/json" \ -d '{"model":"qwen2.5:7b","messages":[{"role":"user","content":"hi"}],"stream":false}' done

如果配置了每分钟 60 次限制，前 60 次应返回 200，后面开始返回 429。这一步能验证限流配置真的生效，而不是只在代码里写了没接线。

异常分支可以用来验证网关对上游错误的处理。比如故意把UPSTREAM_BASE指向一个不存在的端口：

UPSTREAM_BASE=http://127.0.0.1:19999 uvicorn app.main:app --host 127.0.0.1 --port 8080

此时调用网关，预期返回 502 或 504，日志里会记录连接失败。这个实验和第 5 章的排查链路是配套的。

4.5 日志验证

上面例子中，网关日志会打印类似下面的一行：

request_id=8f2a3c9b1e2a method=POST path=/v1/chat/completions upstream=http://127.0.0.1:11434 status=200 duration_ms=832 prompt_tokens=32 completion_tokens=23

验证时要把这段日志和客户端请求对应起来。只有带上 request_id，才能在“客户端报了 502”和“上游发生了什么”之间建立关联。没有 request_id，排查只能靠猜。

5. AI 网关 502 Bad Gateway 排查链路与典型日志

5.1 502 到底是谁返回的

502 Bad Gateway的语义是：代理或网关作为中间层，访问上游时收到了无效响应。出现这个错误，说明请求已经到达了中间层，但中间层没能从上游拿到可用结果。

在 AI 编程工具和本地代理场景里，502 经常来自这几类组件：

• 本地网关进程，比如配置在http://127.0.0.1:1572的代理服务。
• 开发工具自带的本地代理，比如 Codex、Cursor 等连接到本地网关时。
• WebSocket 类网关，比如地址写成ws://127.0.0.1:18789的本地服务。
• 云上反向代理或 Ingress，比如 Nginx、Kong、云负载均衡。
• 上游模型服务本身返回了 500，被中间层包装成 502 返回给客户端。

所以第一步不是急着改配置，而是确认 502 是在哪一层产生的。可以用请求路径和端口号判断，也可以用响应头里的 Server 字段辅助判断。

注意：不要只验证网关进程是否启动，还要验证客户端实际访问的端口、地址、协议和认证头是否与网关配置完全一致。很多本地工具报 502，原因是配置里写的是 1572，但真正监听的进程在 8080。

5.2 按这条顺序排查

推荐按以下顺序，每步都先确认现象再改配置：

1. 网关进程是否在监听目标端口。

ss -lntp | grep 1572 lsof -i :1572

如果没有任何进程监听，说明网关或本地代理没有启动。Windows 下可以用：

netstat -ano | findstr 1572

2. 客户端配置的 URL 是否可访问。

curl -v http://127.0.0.1:1572/v1/models

curl 能通，说明网关在监听；curl 直接拒绝连接，说明端口写错或服务未启动。

3. 端口号是否拼接错误。

127.0.0.1:1572、ws://127.0.0.1:18789这类地址不是通用标准，通常是某个工具或本地代理在配置中写入的默认端口。换一台机器、换一个版本，端口可能完全不同。必须以实际监听端口为准，不能照抄网上的配置。

4. 认证令牌是否缺失或错误。

unauthorized: gateway token missing表示调用请求没有带网关 Token，或者网关不认识这个 Token。去配置文件、环境变量或启动脚本里确认 Token 是否注入，再确认客户端把 Token 放到了Authorization头而不是请求体。

5. 上游服务是否正常。

如果网关能启动但转发失败，要单独调用上游：

curl -v http://127.0.0.1:11434/v1/chat/completions \ -H "Content-Type: application/json" \ -d '{"model":"qwen2.5:7b","messages":[{"role":"user","content":"hi"}],"stream":false}'

如果上游直接返回 500，说明问题在上游模型服务，网关把它包装成 502 是正常的。

6. 检查超时配置。

如果上游正常，但流式请求在长时间等待后断掉，优先检查 read timeout 和 stream idle timeout。模型思考时间超过超时阈值时，连接会被中间层断开。

5.3 典型报错与处理对照表

下面这些报错对应不同根因，排查时可以快速对照：

5.4 本地开发最常见的五个坑

第一个坑：端口地址照抄社区配置。不同工具、不同版本、不同插件的本地监听端口都不一样，别人的http://127.0.0.1:1572到你的环境可能就是空的。正确做法是看启动日志，确认你的网关实际监听在哪个端口。

第二个坑：只改了显示配置，没改实际生效的配置文件。很多本地工具提供 UI 配置项，但底层真正生效的是某个 JSON、YAML 或环境变量文件。UI 显示已经改成了新地址，进程读取的却是旧地址，表现就是改完仍然 502。

第三个坑：网关进程没启动，但客户端还在连接。这种问题最简单也最容易忽略。本地代理、AI 插件、CI 任务在机器重启后不会自动拉起，客户端重试时就报 502。把本地网关做成系统服务或容器，设置开机自启。

第四个坑：流式超时设置太短。模型生成长文本时，两个 SSE 事件之间的间隔可能超过几十秒。如果网关把读超时设为 10 秒，响应稍慢就被切断。区分 read timeout 和 total timeout，流式请求单独配置更宽松的超时。

第五个坑：上游 500 被误判成网关故障。模型服务不是永远可靠，它也会因为上下文过长、模型未加载、资源不足而返回 500。先直接调用上游，确认是模型问题还是网关问题，不要把责任都推给网关。

6. 生产环境网关落地：成本治理、检查清单与扩展方向

6.1 从本地到生产，网关还要补什么

本地最小网关能跑通链路，不代表能直接上生产。生产环境和本地环境之间的差距主要在四个方面：

• 高可用：网关多副本部署，前面加负载均衡，避免单点。本地单进程挂掉可以接受，生产挂掉就是故障。
• 可观测性：print 日志要换成结构化日志，加上请求追踪、指标监控和告警。每次请求要能查 request_id、模型、token、耗时、上游状态。
• 配置管理：路由表、限流阈值、上游地址、密钥都要外置化，改配置不重启服务，或至少通过配置中心统一发布。
• 安全合规：使用 mTLS 或内部网络限制访问网关，记录完整的审计日志，对 Key 做定期轮换，敏感字段脱敏。

本地验证可以接受手动重启，生产必须考虑变更发布、回滚和灰度。网关作为流量的统一入口，一次错误发布会影响所有模型调用，所以配置变更最好能先在小流量实例上验证。

6.2 在网关层控制 token 成本和 credits 消耗

网关不能消除模型调用成本，但它可以把“看不见的成本”变成“可按维度核算的数据”，同时通过技术手段降低浪费。常见做法如下：

• 记录 usage 并落库：每个请求的prompt_tokens、completion_tokens、model、request_id、调用方标识都要保存。没有这些数据，月底账单来了都说不清钱花在哪。
• 模型降级：高成本模型超时或不可用时，切换到低成本模型，前提是业务允许降级。例如普通文本问答可以从大模型降级到小模型。
• 缓存常见请求：相同的系统提示词加用户问题，如果结果允许复用，可以在网关层做语义缓存或精确缓存。
• 配额和熔断：按用户、部门、团队设定每日 credits 上限，超过后拒绝新请求或走审批流程。预算快用完时自动告警，而不是等到账单爆炸。
• 避免失败重试放大成本：重试是成本放大点。一次超时如果重试三次，实际扣费可能是正常请求的几倍。重试要限次数、加退避，并对重复写类请求使用幂等键。

在 credits 语境下，网关的重要职责是把 token 消耗折算成 credits 消耗，并按调用方维度汇总。这样每个团队看到自己的用量和预算，而不是月底笼统看一个总账单。

6.3 上线前检查清单

AI 网关上线前，建议按下面的清单逐项确认：