1. 为什么我会动自己搭网关的念头去年年底到今年年初这段时间我手头同时跑着三个方向的项目一个内部知识库问答、一个代码辅助工具、还有一个给运营团队用的文案生成小工具。这三个项目分别用了不同的模型服务有的走 OpenAI 兼容接口有的走国内某家的 MaaS 平台还有的干脆是本地部署的小模型。一开始我觉得没什么无非就是多配几个 key、多写几行 client 初始化代码的事。但真正跑起来之后问题一个接一个冒出来。最直接的痛点是429。你可能也遇到过那种情况某个模型服务在高峰期疯狂返回exceeded retry limit, last status: 429 too many requests代码里写的重试逻辑根本扛不住整个请求链路就卡死在那里。更麻烦的是不同服务商的限流策略完全不一样有的按分钟算有的按小时算有的甚至按天算配额。我试过在业务代码里硬编码重试和降级逻辑结果就是每个项目都要维护一套几乎相同的代码改一处要同步改三处时间一长自己都记不清哪个项目用的是哪个版本。另一个让我下定决心的是base_url 配置的混乱。OpenAI 的官方 SDK 支持自定义base_url这本来是个好事意味着任何兼容 OpenAI 接口的服务都能用同一套 SDK 接入。但实际操作中每个平台的 base_url 格式、路径拼接规则、鉴权 header 的写法都有细微差别。我遇到过最典型的一个报错就是api error: 400 配置错误: claude provider 缺少 base_url 配置排查了半天才发现是某个 provider 的配置文件里少写了一行。这种问题在单项目里还能忍多项目并行的时候简直是灾难。所以我就想能不能在中间加一层网关把所有模型服务的接入细节统一收口业务代码只跟网关打交道。这样换模型、加模型、调限流策略都只改一个地方。想法很美好但接下来面临一个选择是自己从零搭一个还是用现成的方案我决定两种都试各跑一段时间用真实数据说话。这一跑就是一个月中间踩的坑、省下的时间、多出来的运维成本我都记了下来。2. 两种方案的真实对比自建网关 vs 现成聚合服务2.1 方案 A基于 Flask 自建轻量网关我第一个动手的是自建方案。选 Flask 不是因为它是性能最好的而是因为我的需求本身不复杂请求转发、key 管理、简单的限流和重试、日志记录。Flask 的生态成熟调试方便出问题容易定位。整个网关的核心逻辑其实就是一个反向代理加一层策略层。具体来说我在网关里做了这么几件事。第一统一入口。所有业务请求都发到网关的/v1/chat/completions网关根据请求体里的model字段或者自定义 header 来决定转发到哪个上游服务。第二key 池管理。每个上游服务可能配了多个 key网关负责轮询和故障切换。第三限流与重试。针对 429 错误网关层做指数退避重试并且在不同上游之间做降级。第四日志与监控。每次请求的耗时、token 消耗、上游返回状态都记录下来方便后续分析。代码结构大概是这样# gateway.py 核心转发逻辑示意 import time import requests from flask import Flask, request, jsonify app Flask(__name__) UPSTREAMS { gpt-4o: { base_url: https://api.openai.com/v1, keys: [sk-xxx1, sk-xxx2], weight: 1 }, claude-sonnet: { base_url: https://api.anthropic.com/v1, keys: [sk-ant-xxx], weight: 1 } } def call_upstream(model, payload, retry3): upstream UPSTREAMS.get(model) if not upstream: return {error: model not found}, 404 for attempt in range(retry): key upstream[keys][attempt % len(upstream[keys])] headers { Authorization: fBearer {key}, Content-Type: application/json } try: resp requests.post( f{upstream[base_url]}/chat/completions, jsonpayload, headersheaders, timeout60 ) if resp.status_code 429: time.sleep(2 ** attempt) continue return resp.json(), resp.status_code except Exception as e: if attempt retry - 1: return {error: str(e)}, 500 time.sleep(1) return {error: exceeded retry limit}, 429这段代码看起来简单但实际跑起来要考虑的细节远不止这些。比如流式响应怎么处理OpenAI 的流式返回是 SSE 格式Flask 默认的 response 需要特殊处理才能正确透传。再比如不同上游的请求体格式差异虽然都号称兼容 OpenAI但有些字段名就是不一样得做一层字段映射。还有超时设置上游响应慢的时候不能让网关线程被占满。2.2 方案 B接入现成的多模型聚合平台自建方案跑了两周之后我开始试第二种方案直接用现成的多模型聚合服务。这类平台的核心卖点就是一个 key 调所有模型它们已经帮你处理好了各家 API 的差异、限流、计费等问题。我选了一家支持 OpenAI 兼容接口的平台把业务代码里的base_url一改就能用。接入过程确实简单很多。原来每个项目里要维护的多个 client 初始化代码现在统一成一个from openai import OpenAI client OpenAI( base_urlhttps://聚合平台的地址/v1, api_key平台分配的key ) response client.chat.completions.create( modelgpt-4o, messages[{role: user, content: 你好}] )就这么几行模型切换只需要改model参数。平台侧会自动路由到对应的上游服务429 的问题也由平台统一处理我这边基本感知不到。计费也是统一的不用再分别去各家充值、对账。但用了几天之后我也发现了这类方案的局限。首先是可控性。平台的限流策略、重试逻辑、路由规则都是黑盒我只能看到最终结果中间发生了什么不清楚。有一次某个模型响应特别慢我完全没法判断是上游的问题还是平台转发的问题。其次是数据经过第三方。虽然平台都声称不存储请求内容但对于一些敏感业务来说这个链路本身就是个合规风险。最后是成本透明度。平台通常会在上游价格基础上加一定的服务费具体加多少、怎么算账单上不一定看得清楚。2.3 一个月下来的关键指标对比为了客观比较我记录了一个月内两种方案在几个核心维度上的表现。需要说明的是这些数据来自我自己的使用场景不代表通用结论但能反映一些真实差异。对比维度自建网关现成聚合平台首次搭建耗时约 3 天含调试约 2 小时日常维护成本每周约 2-3 小时几乎为零429 错误率约 1.2%重试后约 0.3%平均响应延迟增加 80-150ms增加 200-400ms模型覆盖数量自己接入的 5 个平台支持的 30 个月度总成本纯上游费用上游费用 约 15% 服务费数据可控性完全自主经过第三方故障排查难度低日志全在自己手里高依赖平台支持这个表格里最让我意外的是延迟。我原本以为自建网关因为少了一跳延迟会更低但实际上自建网关的延迟增加主要来自我自己的重试逻辑和日志写入。而聚合平台虽然多了一跳但它们的转发链路优化得比我好整体延迟反而没有想象中那么高。当然这个数据受网络环境影响很大仅供参考。另一个值得说的是429 错误率。自建方案下我虽然写了重试逻辑但重试本身会消耗时间而且如果多个 key 同时被限流重试也无济于事。聚合平台因为接入了大量用户它们跟上游的议价能力和配额池更大所以 429 的概率确实更低。这一点是我之前没预料到的。3. 自建网关里那些文档不会告诉你的坑3.1 base_url 拼接的隐藏规则如果你决定自建第一个要面对的就是 base_url 的拼接问题。OpenAI 官方 SDK 的行为是你给的base_url会直接作为前缀SDK 会在后面拼上/chat/completions这样的路径。但不同平台的路径规则不一样。有的平台 base_url 要写到/v1有的要写到/api/v3有的甚至不需要版本号。我踩过最典型的一个坑是某平台的 base_url 是https://ark.cn-beijing.volces.com/api/v3我一开始照着 OpenAI 的习惯写成了https://ark.cn-beijing.volces.com/api/v3/多了一个斜杠结果请求路径变成了//chat/completions直接 404。这种问题在文档里通常不会写只能自己试出来。还有一个更隐蔽的问题有些平台的鉴权 header 不是标准的Authorization: Bearer xxx而是自定义的 header 名。如果你用 OpenAI SDK 直接连SDK 只会发标准 header平台就不认。这时候要么改 SDK 的默认 header要么在网关层做一层转换。我的做法是在网关里统一收口业务代码永远只发标准格式网关负责转换成各平台需要的格式。3.2 429 重试策略的陷阱关于 429我一开始的想法很简单遇到就重试指数退避最多重试三次。但实际跑下来发现这个策略有几个陷阱。第一个陷阱是重试放大。假设你的网关同时收到 10 个请求都遇到了 429然后都开始重试。如果重试间隔设置不当这 10 个请求会在同一时间再次打到上游造成第二波 429。正确的做法是加入随机抖动jitter让重试时间分散开。第二个陷阱是不同上游的 429 含义不同。有的上游 429 是你太快了慢点等几秒就好有的上游 429 是你今天配额用完了重试再多次也没用。如果不区分这两种情况代码会陷入无意义的重试循环。我的做法是在网关里维护一个上游状态表如果某个上游连续返回 429 超过一定次数就暂时把它标记为不可用请求直接降级到备用上游。第三个陷阱是流式请求的重试。流式响应一旦开始返回如果中途遇到 429已经吐出去的内容没法收回。这时候重试会导致客户端收到重复内容。我的处理方式是流式请求只在建立连接阶段重试一旦开始接收数据就不再重试而是把错误传递给客户端由客户端决定是否重新发起。3.3 日志与可观测性的最小必要集自建网关最大的优势是可控但前提是你得把日志做好。我一开始只记录了请求的模型、耗时、状态码后来发现这些远远不够。真正有用的日志至少应该包含请求 ID用于串联上下游、上游服务名、使用的 key 标识脱敏后、请求 token 数、响应 token 数、重试次数、最终状态。这些字段看起来多但真出问题的时候少一个都要花额外的时间去猜。比如有一次某个模型的响应突然变慢我通过日志发现是某个 key 对应的上游节点出了问题切换到另一个 key 就恢复了。如果没有 key 级别的日志我可能得排查半天。另外日志的存储也要考虑。如果请求量大日志写入本身会成为瓶颈。我的做法是异步写日志用一个队列缓冲后台线程批量落盘。这样对主请求链路的影响最小。4. 现成聚合平台没告诉你的另一面4.1 模型名称的映射问题用聚合平台的时候你以为model参数写什么就是什么但实际上平台内部有一层映射。比如你写gpt-4o平台可能路由到某个特定版本你写claude-sonnet平台可能对应的是某个具体日期快照。这层映射通常不透明导致同一个模型名在不同时间可能表现不一样。我遇到过的情况是同一个model参数月初和月末的响应风格有细微差异。后来问平台才知道他们上游的模型版本更新了但模型名没变。对于要求输出稳定性的场景这是个隐患。我的应对方式是在业务侧记录每次请求的实际响应特征如果发现异常波动及时跟平台确认。4.2 计费口径的差异聚合平台的计费通常按 token 数算但不同平台对 token 的计数方式可能不一样。有的按上游返回的 usage 字段算有的自己重新分词计算。这会导致你在平台上看到的消耗和上游实际消耗有出入。更需要注意的是缓存命中的情况。有些平台会对相同请求做缓存缓存命中的请求可能不计费或者少计费。但缓存策略是不透明的你没法预测哪些请求会被缓存。对于需要精确控制成本的场景这个不确定性比较麻烦。4.3 故障时的责任边界用聚合平台最让人头疼的是故障排查。当请求失败时你很难判断是平台的问题还是上游的问题。平台通常会提供一个状态页但状态页的更新往往滞后。我遇到过平台状态页显示正常但实际请求大量超时的情况。这时候你能做的很有限提工单、等回复。而自建网关虽然也要自己排查但至少日志全在自己手里能快速定位到是网络问题、上游问题还是自己的代码问题。这个差异在故障发生时体感特别明显。5. 我最终的选择和混合架构思路跑完一个月之后我没有完全倒向任何一边而是用了一个混合架构。核心思路是自建网关做统一入口和策略层聚合平台作为上游之一。具体来说业务代码只跟我的自建网关打交道。网关内部维护一个上游列表这个列表里既有我直接接入的模型服务也有聚合平台的接口。对于延迟敏感、数据敏感的场景网关直接路由到直连的上游对于需要快速接入新模型、或者直连上游不稳定的场景网关路由到聚合平台。这样做的好处是兼顾了可控性和灵活性。我既能享受自建网关的日志、限流、降级能力又能在需要的时候快速用上聚合平台的模型覆盖。而且当某个直连上游出问题时网关可以自动降级到聚合平台业务侧无感知。实现上关键是在网关里加一层路由策略。最简单的策略是按模型名路由复杂一点的可以按请求特征、当前上游健康状态、成本预算等维度动态决策。我目前用的是基于健康状态的加权轮询每个上游有一个健康分数请求优先打到健康分数高的上游失败则扣分成功则恢复。# 简化的健康度路由示意 class Upstream: def __init__(self, name, base_url, weight1): self.name name self.base_url base_url self.weight weight self.health 100 self.last_failure 0 def score(self): # 健康分随时间恢复 elapsed time.time() - self.last_failure recovery min(elapsed / 60, 50) return self.health recovery def pick_upstream(upstreams): total sum(u.score() for u in upstreams) r random.uniform(0, total) upto 0 for u in upstreams: upto u.score() if upto r: return u return upstreams[-1]这个策略不复杂但实际效果不错。当某个上游开始不稳定时它的健康分下降请求会自动流向其他上游等它恢复后再慢慢把流量分回来。6. 给正在纠结要不要自建的你几个判断依据如果你也在纠结这个问题我根据自己的经验整理了几个判断维度你可以对照自己的情况看看。适合自建的情况你对数据链路有合规要求请求不能经过第三方你需要深度定制限流、重试、降级策略你有一定的运维能力能接受每周花几个小时维护你的模型使用相对固定不需要频繁接入新模型。适合用聚合平台的情况你想快速验证想法不想在基础设施上花时间你需要覆盖大量模型自己一个个接入成本太高你的请求量不大自建的固定成本摊不平你对故障排查的实时性要求不高能接受提工单等回复。适合混合架构的情况你既有稳定的核心业务又有探索性的新项目你希望核心链路可控同时保留快速扩展的能力你愿意投入一次性的搭建成本换取长期的灵活性。还有一个很实际的考量是成本。自建网关的显性成本是服务器费用隐性成本是你的时间。如果你每小时的时间成本是 X搭建加维护花了 Y 小时那自建的总成本就是服务器费用加上 X 乘以 Y。聚合平台的服务费通常是上游费用的 10% 到 20%你可以算一下哪个更划算。对我来说因为同时跑着多个项目自建网关摊薄到每个项目上的时间成本可以接受所以混合架构是更优解。最后说一个我踩过的坑不要一上来就追求大而全的网关。我一开始想做一个支持所有模型、所有策略、带管理后台的完整系统结果两周过去了还在改架构。后来我砍掉了所有非核心功能只保留转发、重试、日志三件事两天就跑通了。先跑起来再迭代这个顺序很重要。