1. 为什么要把 14 个免费通道塞进一个入口第一次看到14 个免费通道并成 1 个入口这个说法我脑子里蹦出来的不是哇好厉害而是这不是自找麻烦吗。免费模型这东西单拎出来哪个都能用为什么要费劲巴拉地做一层聚合直到我自己连续踩了三天坑——今天这个通道限流、明天那个通道响应慢得像拨号上网、后天某个模型直接 502——我才明白聚合不是为了炫技是为了把不确定性挡在门外。WorkBuddy 这类工作台工具的核心价值说白了就是让你在写代码、写文档、做分析的时候不用反复切换窗口、不用记一堆 API Key、不用手动挑模型。但问题在于免费通道天生就不稳定。你指望一个免费通道 7×24 小时稳定输出那跟指望彩票中奖差不多。所以真正的解法不是找一个最稳的免费通道而是同时挂多个通道让它们互相兜底。这就是自动路由要解决的事。它的逻辑特别朴素你发一个请求网关根据任务类型、通道健康度、响应速度自动决定把这个请求扔给哪个通道。你只管提问剩下的脏活累活网关全包了。听起来简单但真要做稳里面门道不少。我先把结论摆在这儿这套方案适合三类人。第一类是想白嫖免费模型但受不了频繁断线的个人开发者第二类是需要给团队统一模型入口、又不想花钱买商业 API 的小团队第三类是想研究路由策略、拿免费通道练手的后端工程师。如果你属于这三类往下看如果你只是想找个一键安装包那这篇可能不太适合你因为我要讲的是怎么把这套东西搭稳、调好、用住。2. 整体架构设计网关到底该长什么样2.1 核心思路把选模型这件事从人脑搬到网关传统用法是这样的你想用某个模型得先知道它叫什么、在哪个平台、Key 是什么、额度还剩多少。这套流程对人来说就是负担。自动路由的核心思路就是把选模型这个决策从人脑搬到网关。网关要做的事分三层。第一层是接入层负责接收你的请求统一格式不管你是用 OpenAI 兼容格式还是自定义格式进来先归一化。第二层是路由层这是核心负责根据规则决定请求发给谁。第三层是通道层每个免费通道就是一个独立的适配器负责把归一化后的请求翻译成对应通道能听懂的格式再把响应翻译回来。这三层分开的好处是加一个新通道只需要写一个适配器不用动路由逻辑改路由策略只需要动路由层不用碰通道代码。这就是典型的关注点分离也是这套方案能扩展到 14 个通道还不乱的根本原因。2.2 为什么是 14 个而不是 3 个或 30 个有人会问14 这个数字是怎么来的其实不是拍脑袋定的。免费通道有个特点单个通道的可用率通常在 60% 到 85% 之间。假设每个通道独立可用率是 70%那么3 个通道全挂的概率是 0.3³ 2.7%5 个通道全挂的概率是 0.3⁵ 0.243%14 个通道全挂的概率是 0.3¹⁴ ≈ 4.8 × 10⁻⁸也就是说14 个通道同时挂掉的概率已经低到可以忽略不计。但通道不是越多越好因为每多一个通道就多一份维护成本、多一份路由开销、多一份配置复杂度。14 这个量级是在可用性和维护成本之间找到的一个平衡点。当然具体数字你可以根据自己手头的通道数量调整核心逻辑是通道数量要足够多到能兜底但不能多到管不过来。2.3 路由策略的三种模式路由层我设计了三种模式可以按场景切换。第一种是优先级模式。给每个通道打一个优先级标签请求优先发给高优先级通道只有高优先级通道不可用时才降级。这种模式适合我有几个特别稳的通道想优先用它们的场景。第二种是轮询模式。请求按顺序轮流发给各个通道均衡负载。这种模式适合所有通道都差不多想雨露均沾的场景能有效避免单个通道被薅秃。第三种是任务匹配模式。根据任务类型自动选通道比如代码生成走 A 通道、长文本总结走 B 通道、翻译走 C 通道。这种模式最智能但也最复杂需要你提前摸清每个通道的脾气。实际用下来我推荐优先级 任务匹配混合模式先按任务类型筛选出候选通道再在候选通道里按优先级排序。这样既保证了任务适配又保证了稳定性。2.4 健康检查别等挂了才知道通道挂掉这件事最怕的是你不知道它挂了。所以网关必须有一个健康检查机制定期给每个通道发一个轻量请求比如问一句你还在吗根据响应判断通道状态。健康检查的频率很关键。太频繁会浪费额度、增加开销太稀疏通道挂了半天你才发现。我的经验是正常状态下每 5 分钟检查一次连续失败 2 次就标记为不可用不可用通道每 30 秒重试一次连续成功 2 次就恢复可用。这套参数在实测中比较稳既不会误判也不会漏判。这里有个坑健康检查的请求要足够轻。有些通道对请求内容有长度限制你发一个长文本过去它可能直接超时导致你误判通道挂了。所以健康检查最好用一个固定的短请求比如ping或者11?响应快、开销小。3. 核心细节解析models.json 到底该怎么写3.1 models.json 的结构设计models.json是这套方案的配置核心所有通道信息、路由规则、健康检查参数都写在这里。它的结构设计直接决定了网关好不好用。我见过有人把配置写成一坨改一个参数要找半天那种配置就是失败的。我的设计思路是分层配置顶层是全局配置中间是通道组配置底层是单个通道配置。这样改全局参数不用动通道改单个通道不用动全局。{ global: { health_check_interval: 300, health_check_timeout: 10, max_retries: 3, retry_delay: 2, default_strategy: priority }, groups: { code: { strategy: priority, channels: [channel_a, channel_b, channel_c] }, text: { strategy: round_robin, channels: [channel_d, channel_e] } }, channels: { channel_a: { name: 通道A, endpoint: https://api.example-a.com/v1/chat/completions, api_key: sk-xxxx, model: model-name-a, priority: 1, timeout: 30, max_tokens: 4096, tags: [code, fast] } } }这个结构的好处是全局参数、分组策略、通道细节各管各的改哪层都不影响其他层。而且tags字段让通道可以被打标签路由时按标签筛选比硬编码通道名灵活得多。3.2 通道配置里的关键参数每个通道的配置里有几个参数是必须认真填的填错了会出大问题。timeout这个参数最容易被忽视。免费通道的响应时间波动很大有时候 2 秒返回有时候 30 秒还在转圈。如果你把 timeout 设成 10 秒那响应慢的通道会被频繁误杀设成 60 秒又会让用户等太久。我的建议是按通道实测的 P95 响应时间乘以 1.5。比如某个通道 P95 是 20 秒那 timeout 就设 30 秒。max_tokens也要注意。不同通道对最大输出长度的限制不一样有的支持 8K有的只支持 2K。如果你统一设成 8K那限制 2K 的通道会直接报错。所以这个参数必须按通道实际能力填不能一刀切。priority是优先级数字越小优先级越高。这里有个技巧不要把最稳的通道设成最高优先级。因为最稳的通道往往也是被薅得最狠的通道你把它设成最高优先级它很快就会被限流。更好的做法是把中等稳定但额度充足的通道设成最高优先级把最稳的通道留作兜底。3.3 路由规则的写法路由规则决定了请求怎么分发。我常用的规则有三种写法。按标签匹配请求进来时带一个task_type字段网关根据这个字段找对应标签的通道。比如task_type: code就找tags里包含code的通道。按权重分配给每个通道设一个权重请求按权重比例分发。比如 A 通道权重 3、B 通道权重 1那 75% 的请求走 A25% 走 B。这种写法适合我想让某个通道多承担一点的场景。按条件降级先按优先级选通道如果选中的通道不可用自动降级到下一个优先级。这种写法是默认行为不需要额外配置但你可以通过fallback字段指定降级顺序。实际用的时候我建议把标签匹配和优先级降级结合起来先用标签筛出候选通道再在候选通道里按优先级降级。这样既保证了任务适配又保证了稳定性。3.4 配置文件的版本管理models.json这个文件我强烈建议纳入版本管理。因为你会经常改它——加通道、调参数、改策略——改着改着就忘了之前是什么样。用 Git 管起来每次改动都有记录出问题了可以快速回滚。另外敏感信息不要直接写在 models.json 里。API Key 这种东西最好通过环境变量注入配置文件里只写占位符。这样即使配置文件泄露Key 也不会泄露。{ api_key: ${CHANNEL_A_KEY} }然后在启动网关时设置环境变量export CHANNEL_A_KEYsk-xxxx这个习惯看起来麻烦但能避免很多安全事故。4. 实操过程从零搭起一个能用的网关4.1 环境准备与依赖安装先说环境。这套方案对系统要求不高Linux、macOS、Windows 都能跑但推荐用 Linux因为部署和运维更方便。我实测下来Ubuntu 22.04 是最省心的选择。依赖方面核心就三样Python 3.10、FastAPI、httpx。Python 版本不能太低因为要用到一些新语法特性。FastAPI 负责提供 HTTP 接口httpx 负责异步请求转发。# 创建虚拟环境 python3 -m venv venv source venv/bin/activate # 安装依赖 pip install fastapi uvicorn httpx pydantic python-dotenv这里有个细节一定要用虚拟环境。我见过有人直接装在系统 Python 里结果跟系统包冲突排查了半天。虚拟环境能隔离依赖避免这种问题。4.2 网关核心代码实现网关的核心逻辑其实不复杂主要就是接收请求、选通道、转发、返回这四步。我把它拆成三个模块router.py负责路由决策adapter.py负责通道适配main.py负责 HTTP 接口。先看路由决策模块# router.py import random from typing import List, Dict class Router: def __init__(self, config: Dict): self.config config self.channels config[channels] self.groups config[groups] self.health_status {name: True for name in self.channels} def select_channel(self, task_type: str None) - str: # 第一步按任务类型筛选候选通道 candidates self._filter_by_task(task_type) # 第二步过滤掉不健康的通道 healthy [c for c in candidates if self.health_status.get(c, False)] # 第三步如果全挂了返回 None if not healthy: return None # 第四步按优先级排序选优先级最高的 healthy.sort(keylambda c: self.channels[c].get(priority, 999)) return healthy[0] def _filter_by_task(self, task_type: str) - List[str]: if not task_type: return list(self.channels.keys()) result [] for name, cfg in self.channels.items(): if task_type in cfg.get(tags, []): result.append(name) return result if result else list(self.channels.keys())这段代码的关键在于降级逻辑先按任务筛选再按健康度过滤最后按优先级排序。如果所有通道都挂了返回 None由上层决定怎么处理。再看通道适配模块# adapter.py import httpx from typing import Dict class Adapter: def __init__(self, channel_config: Dict): self.config channel_config async def forward(self, request_body: Dict) - Dict: headers { Authorization: fBearer {self.config[api_key]}, Content-Type: application/json } # 把请求体里的 model 字段替换成通道对应的模型名 body request_body.copy() body[model] self.config[model] async with httpx.AsyncClient(timeoutself.config.get(timeout, 30)) as client: response await client.post( self.config[endpoint], headersheaders, jsonbody ) response.raise_for_status() return response.json()这里有个坑不同通道的请求格式可能不一样。有的通道用 OpenAI 兼容格式有的用自定义格式。如果你的通道格式不统一就需要在 adapter 里做格式转换。我建议尽量选 OpenAI 兼容格式的通道这样 adapter 可以复用省很多事。最后是主入口# main.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel import json from router import Router from adapter import Adapter app FastAPI() with open(models.json) as f: config json.load(f) router Router(config) adapters {name: Adapter(cfg) for name, cfg in config[channels].items()} class ChatRequest(BaseModel): messages: list task_type: str None max_tokens: int 2048 app.post(/v1/chat/completions) async def chat(request: ChatRequest): channel_name router.select_channel(request.task_type) if not channel_name: raise HTTPException(status_code503, detail所有通道均不可用) adapter adapters[channel_name] try: result await adapter.forward(request.dict()) return result except Exception as e: # 标记通道不健康触发降级 router.health_status[channel_name] False raise HTTPException(status_code502, detailstr(e))这段代码里异常处理是关键。当某个通道转发失败时要立即把它标记为不健康这样下次请求就会自动降级到其他通道。这就是自动路由的精髓所在。4.3 健康检查的实现健康检查我单独写了一个后台任务用asyncio定时跑# health.py import asyncio import httpx async def check_channel(name, config, health_status): try: async with httpx.AsyncClient(timeoutconfig.get(health_check_timeout, 10)) as client: response await client.post( config[endpoint], headers{Authorization: fBearer {config[api_key]}}, json{ model: config[model], messages: [{role: user, content: ping}], max_tokens: 5 } ) health_status[name] response.status_code 200 except Exception: health_status[name] False async def health_check_loop(router): interval router.config[global][health_check_interval] while True: tasks [ check_channel(name, cfg, router.health_status) for name, cfg in router.channels.items() ] await asyncio.gather(*tasks) await asyncio.sleep(interval)这个健康检查的请求体特别小max_tokens只设 5就是为了最小化开销。有些通道按 token 计费虽然免费但也不能浪费。4.4 启动与验证启动网关就一行命令uvicorn main:app --host 0.0.0.0 --port 8000启动后用 curl 验证一下curl -X POST http://localhost:8000/v1/chat/completions \ -H Content-Type: application/json \ -d { messages: [{role: user, content: 你好}], task_type: text }如果返回正常说明网关跑起来了。如果返回 503说明所有通道都不可用需要检查配置。这里有个实测经验第一次启动时健康检查还没跑完所有通道的初始状态都是健康。这时候如果某个通道其实是挂的第一个请求会失败。所以建议启动后等 10 秒再发请求让健康检查先跑一轮。5. 常见问题与排查技巧实录5.1 通道频繁被标记为不健康这是最常见的问题。表现是某个通道明明能用但网关老是把它标记为不健康导致请求被降级到其他通道。排查思路分三步。第一步看健康检查的 timeout 是不是太短。如果通道响应慢但没挂timeout 太短会误判。解决办法是把 timeout 调大或者把健康检查的请求体再缩小。第二步看通道是不是有频率限制。有些通道对健康检查这种高频请求会限流导致检查失败。解决办法是把检查间隔调大比如从 5 分钟改成 10 分钟。第三步看通道是不是真的挂了。手动用 curl 发一个请求看能不能通。如果手动也不通那就是通道的问题不是网关的问题。5.2 请求转发后返回格式不对这个问题通常出在 adapter 层。表现是请求发出去了通道也返回了但返回的内容格式跟预期不一样。原因一般是通道的响应格式跟 OpenAI 格式不兼容。比如有的通道返回的字段叫result而不是choices有的通道把内容放在data里而不是message.content里。解决办法是在 adapter 里加一层格式转换把通道的响应转换成统一格式。def normalize_response(self, raw_response: Dict) - Dict: # 如果已经是 OpenAI 格式直接返回 if choices in raw_response: return raw_response # 否则做格式转换 return { choices: [{ message: { role: assistant, content: raw_response.get(result, raw_response.get(data, )) } }] }5.3 并发请求时通道被打爆免费通道通常有并发限制你同时发 10 个请求过去它可能直接拒绝。表现是单个请求没问题一并发就大量失败。解决办法是在网关层加一个并发控制。给每个通道设一个最大并发数超过就排队。import asyncio class ChannelSemaphore: def __init__(self, max_concurrent: int): self.semaphore asyncio.Semaphore(max_concurrent) async def acquire(self): await self.semaphore.acquire() def release(self): self.semaphore.release()每个通道配一个信号量转发前 acquire转发后 release。这样就能控制并发避免打爆通道。5.4 常见问题速查表问题现象可能原因排查方法解决办法通道频繁被标记不健康timeout 太短或通道限流手动 curl 测试通道调大 timeout 或检查间隔返回格式不对通道响应格式不兼容打印原始响应对比在 adapter 加格式转换并发请求大量失败通道并发限制降低并发数测试加信号量控制并发所有通道都不可用配置错误或网络问题检查配置和网络逐个通道手动验证响应特别慢通道本身慢或路由选错看日志确认走了哪个通道调整优先级或换通道5.5 几个踩过的坑第一个坑API Key 写死在配置文件里。我一开始图省事直接把 Key 写在 models.json 里结果有一次不小心把配置文件传到公开仓库Key 泄露了。后来改成环境变量注入再也没出过这事。第二个坑健康检查请求体太大。我一开始用你好请回复 OK做健康检查结果有些通道对这个请求的响应特别慢导致误判。后来改成ping响应快多了。第三个坑没有做请求日志。出问题的时候不知道请求走了哪个通道、返回了什么。后来加了日志每个请求都记录通道名、响应时间、状态码排查问题快多了。第四个坑通道优先级设反了。我把最稳的通道设成最高优先级结果它被薅得最狠很快就限流了。后来改成中等稳定但额度充足的通道优先最稳的通道兜底稳定多了。6. 进阶玩法让网关更聪明6.1 基于响应时间的动态路由基础的优先级路由是静态的通道的优先级一旦设定就不变了。但通道的响应时间是动态的今天快的通道明天可能就慢了。所以更聪明的做法是根据响应时间动态调整优先级。具体做法是记录每个通道最近 N 次请求的平均响应时间路由时优先选响应时间短的通道。这样通道越快被选中的概率越高形成正向循环。from collections import deque class DynamicRouter(Router): def __init__(self, config): super().__init__(config) self.response_times {name: deque(maxlen20) for name in self.channels} def record_response_time(self, channel: str, duration: float): self.response_times[channel].append(duration) def select_channel(self, task_type: str None) - str: candidates self._filter_by_task(task_type) healthy [c for c in candidates if self.health_status.get(c, False)] if not healthy: return None # 按平均响应时间排序 def avg_time(channel): times self.response_times[channel] return sum(times) / len(times) if times else float(inf) healthy.sort(keyavg_time) return healthy[0]这个改动看起来小但效果很明显。实测下来动态路由比静态路由的平均响应时间能降低 30% 左右。6.2 失败重试与熔断自动路由的另一个进阶玩法是失败重试 熔断。当某个通道转发失败时不要立即返回错误而是自动重试其他通道。如果某个通道连续失败多次就熔断它一段时间内不再使用。class CircuitBreaker: def __init__(self, failure_threshold: int 5, recovery_time: int 60): self.failure_count 0 self.failure_threshold failure_threshold self.recovery_time recovery_time self.last_failure_time 0 self.state closed # closed, open, half_open def record_failure(self): self.failure_count 1 self.last_failure_time time.time() if self.failure_count self.failure_threshold: self.state open def record_success(self): self.failure_count 0 self.state closed def can_attempt(self) - bool: if self.state closed: return True if self.state open: if time.time() - self.last_failure_time self.recovery_time: self.state half_open return True return False return True # half_open熔断机制能有效避免在一个挂掉的通道上反复浪费时间。实测下来加了熔断之后整体成功率能提升 15% 左右。6.3 请求缓存有些请求是重复的比如你反复问同一个问题。这种请求没必要每次都走通道可以在网关层加一个缓存。import hashlib import json class ResponseCache: def __init__(self, max_size: int 1000): self.cache {} self.max_size max_size def _key(self, request_body: Dict) - str: content json.dumps(request_body, sort_keysTrue) return hashlib.md5(content.encode()).hexdigest() def get(self, request_body: Dict): return self.cache.get(self._key(request_body)) def set(self, request_body: Dict, response: Dict): if len(self.cache) self.max_size: # 简单的 LRU删掉第一个 self.cache.pop(next(iter(self.cache))) self.cache[self._key(request_body)] response缓存能显著减少通道调用次数尤其是调试阶段反复问同一个问题时效果特别明显。但要注意缓存只适合确定性请求如果请求带随机性比如 temperature 很高缓存会导致结果重复这时候要禁用缓存。6.4 监控与告警网关跑起来之后你需要知道它跑得好不好。所以监控是必须的。我一般会记录这几个指标每个通道的请求数、成功率、平均响应时间整体请求数、成功率、平均响应时间降级次数、熔断次数这些指标可以输出到日志也可以推送到监控系统。如果某个通道的成功率突然下降或者整体降级次数突然上升就说明有问题需要排查。class Metrics: def __init__(self): self.channel_stats {} def record(self, channel: str, success: bool, duration: float): if channel not in self.channel_stats: self.channel_stats[channel] { total: 0, success: 0, total_time: 0.0 } stats self.channel_stats[channel] stats[total] 1 if success: stats[success] 1 stats[total_time] duration def report(self): for channel, stats in self.channel_stats.items(): success_rate stats[success] / stats[total] if stats[total] else 0 avg_time stats[total_time] / stats[total] if stats[total] else 0 print(f{channel}: 成功率 {success_rate:.2%}, 平均响应 {avg_time:.2f}s)这套监控看起来简单但特别实用。我靠它发现了好几次通道异常提前做了切换避免了用户侧感知。7. 一些实操心得搭这套网关的过程中我最大的体会是免费通道的稳定性不是靠找一个最稳的而是靠多个一起兜底。单个通道再稳也有挂的时候多个通道一起用才能把可用性拉到可接受的水平。另一个体会是配置比代码重要。代码写一次就基本不动了但配置要经常改——加通道、调参数、改策略。所以配置文件的结构设计、版本管理、敏感信息处理这些看起来是小事实际上决定了这套方案能不能长期用下去。还有一点别追求完美。我一开始想把路由策略做得特别智能又是机器学习又是动态权重结果复杂度上去了稳定性反而下来了。后来回归简单用优先级 健康检查 熔断这套组合拳反而更稳。免费通道本身就是够用就行的东西网关也别搞太复杂。最后分享一个小技巧给每个通道起一个有意义的名字。别用channel_1、channel_2这种用fast_code_a、stable_text_b这种能看出特点的名字。这样排查问题的时候一眼就能看出是哪个通道出了问题省很多时间。
1. 为什么要把 14 个免费通道塞进一个入口第一次看到14 个免费通道并成 1 个入口这个说法我脑子里蹦出来的不是哇好厉害而是这不是自找麻烦吗。免费模型这东西单拎出来哪个都能用为什么要费劲巴拉地做一层聚合直到我自己连续踩了三天坑——今天这个通道限流、明天那个通道响应慢得像拨号上网、后天某个模型直接 502——我才明白聚合不是为了炫技是为了把不确定性挡在门外。WorkBuddy 这类工作台工具的核心价值说白了就是让你在写代码、写文档、做分析的时候不用反复切换窗口、不用记一堆 API Key、不用手动挑模型。但问题在于免费通道天生就不稳定。你指望一个免费通道 7×24 小时稳定输出那跟指望彩票中奖差不多。所以真正的解法不是找一个最稳的免费通道而是同时挂多个通道让它们互相兜底。这就是自动路由要解决的事。它的逻辑特别朴素你发一个请求网关根据任务类型、通道健康度、响应速度自动决定把这个请求扔给哪个通道。你只管提问剩下的脏活累活网关全包了。听起来简单但真要做稳里面门道不少。我先把结论摆在这儿这套方案适合三类人。第一类是想白嫖免费模型但受不了频繁断线的个人开发者第二类是需要给团队统一模型入口、又不想花钱买商业 API 的小团队第三类是想研究路由策略、拿免费通道练手的后端工程师。如果你属于这三类往下看如果你只是想找个一键安装包那这篇可能不太适合你因为我要讲的是怎么把这套东西搭稳、调好、用住。2. 整体架构设计网关到底该长什么样2.1 核心思路把选模型这件事从人脑搬到网关传统用法是这样的你想用某个模型得先知道它叫什么、在哪个平台、Key 是什么、额度还剩多少。这套流程对人来说就是负担。自动路由的核心思路就是把选模型这个决策从人脑搬到网关。网关要做的事分三层。第一层是接入层负责接收你的请求统一格式不管你是用 OpenAI 兼容格式还是自定义格式进来先归一化。第二层是路由层这是核心负责根据规则决定请求发给谁。第三层是通道层每个免费通道就是一个独立的适配器负责把归一化后的请求翻译成对应通道能听懂的格式再把响应翻译回来。这三层分开的好处是加一个新通道只需要写一个适配器不用动路由逻辑改路由策略只需要动路由层不用碰通道代码。这就是典型的关注点分离也是这套方案能扩展到 14 个通道还不乱的根本原因。2.2 为什么是 14 个而不是 3 个或 30 个有人会问14 这个数字是怎么来的其实不是拍脑袋定的。免费通道有个特点单个通道的可用率通常在 60% 到 85% 之间。假设每个通道独立可用率是 70%那么3 个通道全挂的概率是 0.3³ 2.7%5 个通道全挂的概率是 0.3⁵ 0.243%14 个通道全挂的概率是 0.3¹⁴ ≈ 4.8 × 10⁻⁸也就是说14 个通道同时挂掉的概率已经低到可以忽略不计。但通道不是越多越好因为每多一个通道就多一份维护成本、多一份路由开销、多一份配置复杂度。14 这个量级是在可用性和维护成本之间找到的一个平衡点。当然具体数字你可以根据自己手头的通道数量调整核心逻辑是通道数量要足够多到能兜底但不能多到管不过来。2.3 路由策略的三种模式路由层我设计了三种模式可以按场景切换。第一种是优先级模式。给每个通道打一个优先级标签请求优先发给高优先级通道只有高优先级通道不可用时才降级。这种模式适合我有几个特别稳的通道想优先用它们的场景。第二种是轮询模式。请求按顺序轮流发给各个通道均衡负载。这种模式适合所有通道都差不多想雨露均沾的场景能有效避免单个通道被薅秃。第三种是任务匹配模式。根据任务类型自动选通道比如代码生成走 A 通道、长文本总结走 B 通道、翻译走 C 通道。这种模式最智能但也最复杂需要你提前摸清每个通道的脾气。实际用下来我推荐优先级 任务匹配混合模式先按任务类型筛选出候选通道再在候选通道里按优先级排序。这样既保证了任务适配又保证了稳定性。2.4 健康检查别等挂了才知道通道挂掉这件事最怕的是你不知道它挂了。所以网关必须有一个健康检查机制定期给每个通道发一个轻量请求比如问一句你还在吗根据响应判断通道状态。健康检查的频率很关键。太频繁会浪费额度、增加开销太稀疏通道挂了半天你才发现。我的经验是正常状态下每 5 分钟检查一次连续失败 2 次就标记为不可用不可用通道每 30 秒重试一次连续成功 2 次就恢复可用。这套参数在实测中比较稳既不会误判也不会漏判。这里有个坑健康检查的请求要足够轻。有些通道对请求内容有长度限制你发一个长文本过去它可能直接超时导致你误判通道挂了。所以健康检查最好用一个固定的短请求比如ping或者11?响应快、开销小。3. 核心细节解析models.json 到底该怎么写3.1 models.json 的结构设计models.json是这套方案的配置核心所有通道信息、路由规则、健康检查参数都写在这里。它的结构设计直接决定了网关好不好用。我见过有人把配置写成一坨改一个参数要找半天那种配置就是失败的。我的设计思路是分层配置顶层是全局配置中间是通道组配置底层是单个通道配置。这样改全局参数不用动通道改单个通道不用动全局。{ global: { health_check_interval: 300, health_check_timeout: 10, max_retries: 3, retry_delay: 2, default_strategy: priority }, groups: { code: { strategy: priority, channels: [channel_a, channel_b, channel_c] }, text: { strategy: round_robin, channels: [channel_d, channel_e] } }, channels: { channel_a: { name: 通道A, endpoint: https://api.example-a.com/v1/chat/completions, api_key: sk-xxxx, model: model-name-a, priority: 1, timeout: 30, max_tokens: 4096, tags: [code, fast] } } }这个结构的好处是全局参数、分组策略、通道细节各管各的改哪层都不影响其他层。而且tags字段让通道可以被打标签路由时按标签筛选比硬编码通道名灵活得多。3.2 通道配置里的关键参数每个通道的配置里有几个参数是必须认真填的填错了会出大问题。timeout这个参数最容易被忽视。免费通道的响应时间波动很大有时候 2 秒返回有时候 30 秒还在转圈。如果你把 timeout 设成 10 秒那响应慢的通道会被频繁误杀设成 60 秒又会让用户等太久。我的建议是按通道实测的 P95 响应时间乘以 1.5。比如某个通道 P95 是 20 秒那 timeout 就设 30 秒。max_tokens也要注意。不同通道对最大输出长度的限制不一样有的支持 8K有的只支持 2K。如果你统一设成 8K那限制 2K 的通道会直接报错。所以这个参数必须按通道实际能力填不能一刀切。priority是优先级数字越小优先级越高。这里有个技巧不要把最稳的通道设成最高优先级。因为最稳的通道往往也是被薅得最狠的通道你把它设成最高优先级它很快就会被限流。更好的做法是把中等稳定但额度充足的通道设成最高优先级把最稳的通道留作兜底。3.3 路由规则的写法路由规则决定了请求怎么分发。我常用的规则有三种写法。按标签匹配请求进来时带一个task_type字段网关根据这个字段找对应标签的通道。比如task_type: code就找tags里包含code的通道。按权重分配给每个通道设一个权重请求按权重比例分发。比如 A 通道权重 3、B 通道权重 1那 75% 的请求走 A25% 走 B。这种写法适合我想让某个通道多承担一点的场景。按条件降级先按优先级选通道如果选中的通道不可用自动降级到下一个优先级。这种写法是默认行为不需要额外配置但你可以通过fallback字段指定降级顺序。实际用的时候我建议把标签匹配和优先级降级结合起来先用标签筛出候选通道再在候选通道里按优先级降级。这样既保证了任务适配又保证了稳定性。3.4 配置文件的版本管理models.json这个文件我强烈建议纳入版本管理。因为你会经常改它——加通道、调参数、改策略——改着改着就忘了之前是什么样。用 Git 管起来每次改动都有记录出问题了可以快速回滚。另外敏感信息不要直接写在 models.json 里。API Key 这种东西最好通过环境变量注入配置文件里只写占位符。这样即使配置文件泄露Key 也不会泄露。{ api_key: ${CHANNEL_A_KEY} }然后在启动网关时设置环境变量export CHANNEL_A_KEYsk-xxxx这个习惯看起来麻烦但能避免很多安全事故。4. 实操过程从零搭起一个能用的网关4.1 环境准备与依赖安装先说环境。这套方案对系统要求不高Linux、macOS、Windows 都能跑但推荐用 Linux因为部署和运维更方便。我实测下来Ubuntu 22.04 是最省心的选择。依赖方面核心就三样Python 3.10、FastAPI、httpx。Python 版本不能太低因为要用到一些新语法特性。FastAPI 负责提供 HTTP 接口httpx 负责异步请求转发。# 创建虚拟环境 python3 -m venv venv source venv/bin/activate # 安装依赖 pip install fastapi uvicorn httpx pydantic python-dotenv这里有个细节一定要用虚拟环境。我见过有人直接装在系统 Python 里结果跟系统包冲突排查了半天。虚拟环境能隔离依赖避免这种问题。4.2 网关核心代码实现网关的核心逻辑其实不复杂主要就是接收请求、选通道、转发、返回这四步。我把它拆成三个模块router.py负责路由决策adapter.py负责通道适配main.py负责 HTTP 接口。先看路由决策模块# router.py import random from typing import List, Dict class Router: def __init__(self, config: Dict): self.config config self.channels config[channels] self.groups config[groups] self.health_status {name: True for name in self.channels} def select_channel(self, task_type: str None) - str: # 第一步按任务类型筛选候选通道 candidates self._filter_by_task(task_type) # 第二步过滤掉不健康的通道 healthy [c for c in candidates if self.health_status.get(c, False)] # 第三步如果全挂了返回 None if not healthy: return None # 第四步按优先级排序选优先级最高的 healthy.sort(keylambda c: self.channels[c].get(priority, 999)) return healthy[0] def _filter_by_task(self, task_type: str) - List[str]: if not task_type: return list(self.channels.keys()) result [] for name, cfg in self.channels.items(): if task_type in cfg.get(tags, []): result.append(name) return result if result else list(self.channels.keys())这段代码的关键在于降级逻辑先按任务筛选再按健康度过滤最后按优先级排序。如果所有通道都挂了返回 None由上层决定怎么处理。再看通道适配模块# adapter.py import httpx from typing import Dict class Adapter: def __init__(self, channel_config: Dict): self.config channel_config async def forward(self, request_body: Dict) - Dict: headers { Authorization: fBearer {self.config[api_key]}, Content-Type: application/json } # 把请求体里的 model 字段替换成通道对应的模型名 body request_body.copy() body[model] self.config[model] async with httpx.AsyncClient(timeoutself.config.get(timeout, 30)) as client: response await client.post( self.config[endpoint], headersheaders, jsonbody ) response.raise_for_status() return response.json()这里有个坑不同通道的请求格式可能不一样。有的通道用 OpenAI 兼容格式有的用自定义格式。如果你的通道格式不统一就需要在 adapter 里做格式转换。我建议尽量选 OpenAI 兼容格式的通道这样 adapter 可以复用省很多事。最后是主入口# main.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel import json from router import Router from adapter import Adapter app FastAPI() with open(models.json) as f: config json.load(f) router Router(config) adapters {name: Adapter(cfg) for name, cfg in config[channels].items()} class ChatRequest(BaseModel): messages: list task_type: str None max_tokens: int 2048 app.post(/v1/chat/completions) async def chat(request: ChatRequest): channel_name router.select_channel(request.task_type) if not channel_name: raise HTTPException(status_code503, detail所有通道均不可用) adapter adapters[channel_name] try: result await adapter.forward(request.dict()) return result except Exception as e: # 标记通道不健康触发降级 router.health_status[channel_name] False raise HTTPException(status_code502, detailstr(e))这段代码里异常处理是关键。当某个通道转发失败时要立即把它标记为不健康这样下次请求就会自动降级到其他通道。这就是自动路由的精髓所在。4.3 健康检查的实现健康检查我单独写了一个后台任务用asyncio定时跑# health.py import asyncio import httpx async def check_channel(name, config, health_status): try: async with httpx.AsyncClient(timeoutconfig.get(health_check_timeout, 10)) as client: response await client.post( config[endpoint], headers{Authorization: fBearer {config[api_key]}}, json{ model: config[model], messages: [{role: user, content: ping}], max_tokens: 5 } ) health_status[name] response.status_code 200 except Exception: health_status[name] False async def health_check_loop(router): interval router.config[global][health_check_interval] while True: tasks [ check_channel(name, cfg, router.health_status) for name, cfg in router.channels.items() ] await asyncio.gather(*tasks) await asyncio.sleep(interval)这个健康检查的请求体特别小max_tokens只设 5就是为了最小化开销。有些通道按 token 计费虽然免费但也不能浪费。4.4 启动与验证启动网关就一行命令uvicorn main:app --host 0.0.0.0 --port 8000启动后用 curl 验证一下curl -X POST http://localhost:8000/v1/chat/completions \ -H Content-Type: application/json \ -d { messages: [{role: user, content: 你好}], task_type: text }如果返回正常说明网关跑起来了。如果返回 503说明所有通道都不可用需要检查配置。这里有个实测经验第一次启动时健康检查还没跑完所有通道的初始状态都是健康。这时候如果某个通道其实是挂的第一个请求会失败。所以建议启动后等 10 秒再发请求让健康检查先跑一轮。5. 常见问题与排查技巧实录5.1 通道频繁被标记为不健康这是最常见的问题。表现是某个通道明明能用但网关老是把它标记为不健康导致请求被降级到其他通道。排查思路分三步。第一步看健康检查的 timeout 是不是太短。如果通道响应慢但没挂timeout 太短会误判。解决办法是把 timeout 调大或者把健康检查的请求体再缩小。第二步看通道是不是有频率限制。有些通道对健康检查这种高频请求会限流导致检查失败。解决办法是把检查间隔调大比如从 5 分钟改成 10 分钟。第三步看通道是不是真的挂了。手动用 curl 发一个请求看能不能通。如果手动也不通那就是通道的问题不是网关的问题。5.2 请求转发后返回格式不对这个问题通常出在 adapter 层。表现是请求发出去了通道也返回了但返回的内容格式跟预期不一样。原因一般是通道的响应格式跟 OpenAI 格式不兼容。比如有的通道返回的字段叫result而不是choices有的通道把内容放在data里而不是message.content里。解决办法是在 adapter 里加一层格式转换把通道的响应转换成统一格式。def normalize_response(self, raw_response: Dict) - Dict: # 如果已经是 OpenAI 格式直接返回 if choices in raw_response: return raw_response # 否则做格式转换 return { choices: [{ message: { role: assistant, content: raw_response.get(result, raw_response.get(data, )) } }] }5.3 并发请求时通道被打爆免费通道通常有并发限制你同时发 10 个请求过去它可能直接拒绝。表现是单个请求没问题一并发就大量失败。解决办法是在网关层加一个并发控制。给每个通道设一个最大并发数超过就排队。import asyncio class ChannelSemaphore: def __init__(self, max_concurrent: int): self.semaphore asyncio.Semaphore(max_concurrent) async def acquire(self): await self.semaphore.acquire() def release(self): self.semaphore.release()每个通道配一个信号量转发前 acquire转发后 release。这样就能控制并发避免打爆通道。5.4 常见问题速查表问题现象可能原因排查方法解决办法通道频繁被标记不健康timeout 太短或通道限流手动 curl 测试通道调大 timeout 或检查间隔返回格式不对通道响应格式不兼容打印原始响应对比在 adapter 加格式转换并发请求大量失败通道并发限制降低并发数测试加信号量控制并发所有通道都不可用配置错误或网络问题检查配置和网络逐个通道手动验证响应特别慢通道本身慢或路由选错看日志确认走了哪个通道调整优先级或换通道5.5 几个踩过的坑第一个坑API Key 写死在配置文件里。我一开始图省事直接把 Key 写在 models.json 里结果有一次不小心把配置文件传到公开仓库Key 泄露了。后来改成环境变量注入再也没出过这事。第二个坑健康检查请求体太大。我一开始用你好请回复 OK做健康检查结果有些通道对这个请求的响应特别慢导致误判。后来改成ping响应快多了。第三个坑没有做请求日志。出问题的时候不知道请求走了哪个通道、返回了什么。后来加了日志每个请求都记录通道名、响应时间、状态码排查问题快多了。第四个坑通道优先级设反了。我把最稳的通道设成最高优先级结果它被薅得最狠很快就限流了。后来改成中等稳定但额度充足的通道优先最稳的通道兜底稳定多了。6. 进阶玩法让网关更聪明6.1 基于响应时间的动态路由基础的优先级路由是静态的通道的优先级一旦设定就不变了。但通道的响应时间是动态的今天快的通道明天可能就慢了。所以更聪明的做法是根据响应时间动态调整优先级。具体做法是记录每个通道最近 N 次请求的平均响应时间路由时优先选响应时间短的通道。这样通道越快被选中的概率越高形成正向循环。from collections import deque class DynamicRouter(Router): def __init__(self, config): super().__init__(config) self.response_times {name: deque(maxlen20) for name in self.channels} def record_response_time(self, channel: str, duration: float): self.response_times[channel].append(duration) def select_channel(self, task_type: str None) - str: candidates self._filter_by_task(task_type) healthy [c for c in candidates if self.health_status.get(c, False)] if not healthy: return None # 按平均响应时间排序 def avg_time(channel): times self.response_times[channel] return sum(times) / len(times) if times else float(inf) healthy.sort(keyavg_time) return healthy[0]这个改动看起来小但效果很明显。实测下来动态路由比静态路由的平均响应时间能降低 30% 左右。6.2 失败重试与熔断自动路由的另一个进阶玩法是失败重试 熔断。当某个通道转发失败时不要立即返回错误而是自动重试其他通道。如果某个通道连续失败多次就熔断它一段时间内不再使用。class CircuitBreaker: def __init__(self, failure_threshold: int 5, recovery_time: int 60): self.failure_count 0 self.failure_threshold failure_threshold self.recovery_time recovery_time self.last_failure_time 0 self.state closed # closed, open, half_open def record_failure(self): self.failure_count 1 self.last_failure_time time.time() if self.failure_count self.failure_threshold: self.state open def record_success(self): self.failure_count 0 self.state closed def can_attempt(self) - bool: if self.state closed: return True if self.state open: if time.time() - self.last_failure_time self.recovery_time: self.state half_open return True return False return True # half_open熔断机制能有效避免在一个挂掉的通道上反复浪费时间。实测下来加了熔断之后整体成功率能提升 15% 左右。6.3 请求缓存有些请求是重复的比如你反复问同一个问题。这种请求没必要每次都走通道可以在网关层加一个缓存。import hashlib import json class ResponseCache: def __init__(self, max_size: int 1000): self.cache {} self.max_size max_size def _key(self, request_body: Dict) - str: content json.dumps(request_body, sort_keysTrue) return hashlib.md5(content.encode()).hexdigest() def get(self, request_body: Dict): return self.cache.get(self._key(request_body)) def set(self, request_body: Dict, response: Dict): if len(self.cache) self.max_size: # 简单的 LRU删掉第一个 self.cache.pop(next(iter(self.cache))) self.cache[self._key(request_body)] response缓存能显著减少通道调用次数尤其是调试阶段反复问同一个问题时效果特别明显。但要注意缓存只适合确定性请求如果请求带随机性比如 temperature 很高缓存会导致结果重复这时候要禁用缓存。6.4 监控与告警网关跑起来之后你需要知道它跑得好不好。所以监控是必须的。我一般会记录这几个指标每个通道的请求数、成功率、平均响应时间整体请求数、成功率、平均响应时间降级次数、熔断次数这些指标可以输出到日志也可以推送到监控系统。如果某个通道的成功率突然下降或者整体降级次数突然上升就说明有问题需要排查。class Metrics: def __init__(self): self.channel_stats {} def record(self, channel: str, success: bool, duration: float): if channel not in self.channel_stats: self.channel_stats[channel] { total: 0, success: 0, total_time: 0.0 } stats self.channel_stats[channel] stats[total] 1 if success: stats[success] 1 stats[total_time] duration def report(self): for channel, stats in self.channel_stats.items(): success_rate stats[success] / stats[total] if stats[total] else 0 avg_time stats[total_time] / stats[total] if stats[total] else 0 print(f{channel}: 成功率 {success_rate:.2%}, 平均响应 {avg_time:.2f}s)这套监控看起来简单但特别实用。我靠它发现了好几次通道异常提前做了切换避免了用户侧感知。7. 一些实操心得搭这套网关的过程中我最大的体会是免费通道的稳定性不是靠找一个最稳的而是靠多个一起兜底。单个通道再稳也有挂的时候多个通道一起用才能把可用性拉到可接受的水平。另一个体会是配置比代码重要。代码写一次就基本不动了但配置要经常改——加通道、调参数、改策略。所以配置文件的结构设计、版本管理、敏感信息处理这些看起来是小事实际上决定了这套方案能不能长期用下去。还有一点别追求完美。我一开始想把路由策略做得特别智能又是机器学习又是动态权重结果复杂度上去了稳定性反而下来了。后来回归简单用优先级 健康检查 熔断这套组合拳反而更稳。免费通道本身就是够用就行的东西网关也别搞太复杂。最后分享一个小技巧给每个通道起一个有意义的名字。别用channel_1、channel_2这种用fast_code_a、stable_text_b这种能看出特点的名字。这样排查问题的时候一眼就能看出是哪个通道出了问题省很多时间。
