atmAPI智慧路由架构设计智能路由与熔断器实现当你的 AI 应用同时调用 GPT-4、Claude、GLM 等多个模型时如何保证高可用、低成本、低延迟本文分享一个基于 OpenAI 兼容协议的中转站架构核心是智能路由算法和熔断器机制。一、问题背景AI API 调用面临三个核心问题单点故障某个模型提供商宕机整个应用瘫痪成本不可控不同模型价格差异大简单调用无法优化成本延迟波动同一模型在不同时段响应速度差异可达 10 倍我们需要的不是一个简单的代理而是一个具备智能调度能力的中间层。二、架构设计2.1 整体架构┌─────────────┐ ┌──────────────────┐ ┌─────────────┐ │ 客户端应用 │────▶│ atmApi 中转站 │────▶│ 模型提供商 │ │(OpenAI SDK)│ │ │ │ │ ─────────────┘ │ ────────────┐ │ │ OpenAI │ │ │ 智能路由层 │ │ │ Anthropic │ │ ├────────────┤ │ │ 智谱 GLM │ │ │ 熔断器池 │ │ │ DeepSeek │ │ ├──────────── │ │... │ │ │ 缓存层 │ │ └─────────────┘ │ └──────────── │ └──────────────────┘2.2 核心模块智能路由层根据延迟、成功率、成本三维决策动态选择最优渠道熔断器池每个渠道独立熔断器故障时自动隔离恢复后自动重试缓存层Prompt Cache Response Cache 双层缓存命中时零成本返回三、智能路由算法实现3.1 路由决策模型路由决策基于三个维度的加权评分classSmartRouter:def__init__(self):self.channels{}# channel_id - ChannelStatsself.weights{latency:0.4,# 延迟权重success_rate:0.4,# 成功率权重cost:0.2# 成本权重}defcalculate_score(self,channel_id:str,model:str)-float:计算渠道综合评分statsself.channels[channel_id]# 延迟评分越低越好归一化到 0-1latency_score1.0/(1.0stats.avg_latency/1000)# 成功率评分success_scorestats.success_rate# 成本评分越低越好cost_score1.0/(1.0stats.cost_per_1k_tokens)# 加权综合评分total_score(self.weights[latency]*latency_scoreself.weights[success_rate]*success_scoreself.weights[cost]*cost_score)returntotal_scoredefselect_channel(self,model:str)-str:选择最优渠道scores{}forchannel_idinself.get_available_channels(model):scores[channel_id]self.calculate_score(channel_id,model)# 返回评分最高的渠道returnmax(scores,keyscores.get)3.2 渠道状态追踪fromdataclassesimportdataclass,fieldfromcollectionsimportdequeimporttimedataclassclassChannelStats:channel_id:strmodel:str# 滑动窗口统计最近 100 次请求latencies:dequefield(default_factorylambda:deque(maxlen100))successes:dequefield(default_factorylambda:deque(maxlen100))costs:dequefield(default_factorylambda:deque(maxlen100))# 熔断器状态circuit_state:strclosed# closed/open/half-openfailure_count:int0last_failure_time:float0propertydefavg_latency(self)-float:returnsum(self.latencies)/len(self.latencies)ifself.latencieselse1000propertydefsuccess_rate(self)-float:returnsum(self.successes)/len(self.successes)ifself.successeselse0.5propertydefcost_per_1k_tokens(self)-float:returnsum(self.costs)/len(self.costs)ifself.costselse0.01defrecord_success(self,latency_ms:float,cost:float):记录成功请求self.latencies.append(latency_ms)self.successes.append(1.0)self.costs.append(cost)self.failure_count0self.circuit_statecloseddefrecord_failure(self):记录失败请求self.successes.append(0.0)self.failure_count1self.last_failure_timetime.time()# 连续失败 5 次触发熔断ifself.failure_count5:self.circuit_stateopen四、熔断器机制实现4.1 三态熔断器熔断器有三种状态Closed关闭正常状态请求正常转发Open打开故障状态请求直接拒绝快速失败Half-Open半开恢复探测状态允许少量请求测试classCircuitBreaker:def__init__(self,failure_threshold5,recovery_timeout60):self.failure_thresholdfailure_threshold self.recovery_timeoutrecovery_timeout self.stateclosedself.failure_count0self.last_failure_time0defallow_request(self)-bool:判断是否允许请求通过ifself.stateclosed:returnTrueifself.stateopen:# 检查是否超过恢复超时iftime.time()-self.last_failure_timeself.recovery_timeout:self.statehalf-openreturnTruereturnFalseifself.statehalf-open:# 半开状态只允许一个探测请求returnTruereturnFalsedefrecord_success(self):记录成功重置熔断器self.failure_count0self.statecloseddefrecord_failure(self):记录失败可能触发熔断self.failure_count1self.last_failure_timetime.time()ifself.failure_countself.failure_threshold:self.stateopen4.2 熔断器与路由联动classRouterWithCircuitBreaker:def__init__(self):self.routerSmartRouter()self.breakers{}# channel_id - CircuitBreakerdefexecute_request(self,model:str,prompt:str)-dict:执行请求带熔断保护# 获取候选渠道按评分排序candidatesself.router.get_ranked_channels(model)forchannel_idincandidates:breakerself.breakers.get(channel_id)# 熔断器检查ifbreakerandnotbreaker.allow_request():continuetry:# 执行请求responseself.call_channel(channel_id,model,prompt)# 记录成功ifbreaker:breaker.record_success()self.router.record_success(channel_id,response)returnresponseexceptExceptionase:# 记录失败ifbreaker:breaker.record_failure()self.router.record_failure(channel_id)# 继续尝试下一个渠道continueraiseException(所有渠道均不可用)五、效果验证5.1 测试数据在 7 天连续运行中系统处理了 12,000 次 API 调用指标优化前优化后提升平均延迟2,300ms890ms61% ↓成功率87%99.2%12.2% ↑单次调用成本¥0.08¥0.0362% ↓故障恢复时间手动介入自动 60s-5.2 熔断器触发案例2026-07-15 14:23某渠道连续返回 500 错误14:23:01[WARN]Channel openai-gpt4: failure#114:23:02[WARN]Channel openai-gpt4: failure#214:23:03[WARN]Channel openai-gpt4: failure#314:23:04[WARN]Channel openai-gpt4: failure#414:23:05[WARN]Channel openai-gpt4: failure#514:23:05[INFO]Channel openai-gpt4: CIRCUIT OPEN - 自动隔离14:23:05[INFO]Router: 切换到备选渠道 anthropic-claude14:23:06[INFO]Request completed via anthropic-claude(latency:1,200ms)14:24:05[INFO]Channel openai-gpt4: HALF-OPEN - 探测请求14:24:06[INFO]Channel openai-gpt4: 恢复成功CIRCUIT CLOSED整个故障期间用户无感知请求自动路由到备选渠道。六、关键经验总结滑动窗口统计用 deque(maxlen100) 做滑动窗口避免历史数据影响当前决策熔断器三态设计Closed → Open → Half-Open → Closed自动恢复无需人工介入多渠道降级路由失败时自动尝试下一个渠道保证服务可用性成本感知路由算法中成本权重占 20%在延迟和成本之间取得平衡七、关于 AiToMoneyAiToMoney 是一支专注于 AI 技术产品化的团队atmApi 是我们打造的 AI API 中转服务支持多模型智能路由、成本看板、月卡计费。团队秉承用 AI 杠杆放大个人能力的理念产品化数字资产。本文代码已开源欢迎交流讨论。