阅读提示本文约3000字含完整Python代码示例建议收藏后配合实战操作。一、先说我踩过的坑你在调用多个AI模型时是不是也遇到过这些问题每个模型得单独申请接入凭证、单独看文档、单独对接SDK。今天用A写代码明天用B做翻译后天用C搞数据分析——三套接入方式、三种计费规则、三个后台来回切。我就干过这事儿。去年接了个项目要给用户提供多模型切换功能光接入层就写了小一千行。每次加新模型都得改核心逻辑测试时漏了一个异常处理半夜三点被报警叫起来。更坑的是成本。不同模型单价差好几倍没有统一的计量视图月底看账单才傻眼。这篇文章我把这半年摸索出来的聚合接入方案完整分享出来。二、为什么需要聚合接入简单来说就三件事统一入口不管你后端接了多少模型对外暴露一个接口统一计量所有调用走同一套Token计数和成本核算智能路由根据任务类型、预算、延迟要求自动选模型架构一览核心思路是中间加一层抽象。上层应用不关心下面接了谁下层模型变动不影响上层业务。关键设计决策三、实战从零搭建聚合接入层3.1 项目结构3.2 配置文件# config.yamlmodels:deepseek:provider:deepseekendpoint:https://api.deepseek.com/v1/chat/completionsprice_per_1k_input:0.001price_per_1k_output:0.002max_retries:3timeout:30qwen:provider:qwenendpoint:https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completionsprice_per_1k_input:0.0008price_per_1k_output:0.002glm:provider:glmendpoint:https://open.bigmodel.cn/api/paas/v4/chat/completionsprice_per_1k_input:0.001price_per_1k_output:0.001routing:default_model:deepseekfallback_chain:[deepseek,qwen,glm]cost_limit_per_call:0.053.3 适配器基类这是整个框架的基石定义统一接口# gateway/adapters/base.pyfromabcimportABC,abstractmethodfromdataclassesimportdataclass,fieldfromtypingimportOptional,Dict,Any,ListfromenumimportEnumclassModelRole(str,Enum):SYSTEMsystemUSERuserASSISTANTassistantdataclassclassMessage:统一消息格式role:ModelRole content:strdataclassclassModelRequest:统一请求体messages:List[Message]model:Optional[str]Nonetemperature:float0.7max_tokens:int4096dataclassclassTokenUsage:input_tokens:intoutput_tokens:inttotal_tokens:intdataclassclassModelResponse:统一响应体content:strmodel:strusage:TokenUsage cost:floatlatency_ms:floatclassBaseAdapter(ABC):模型适配器基类def__init__(self,config:Dict[str,Any],credentials:str):self.configconfig self.credentialscredentials self.endpointconfig[endpoint]self.input_priceconfig[price_per_1k_input]self.output_priceconfig[price_per_1k_output]abstractmethoddefbuild_request_payload(self,request:ModelRequest)-Dict[str,Any]:将统一请求转换为该模型的原生请求格式passabstractmethoddefparse_response(self,raw_response:Dict[str,Any])-ModelResponse:将模型原生响应转换为统一响应格式passdefcalculate_cost(self,input_tokens:int,output_tokens:int)-float:returnround(input_tokens*self.input_price/1000output_tokens*self.output_price/1000,6)3.4 具体适配器实现# gateway/adapters/deepseek.pyclassDeepSeekAdapter(BaseAdapter):defbuild_request_payload(self,request:ModelRequest)-Dict[str,Any]:messages[{role:m.role.value,content:m.content}forminrequest.messages]return{model:deepseek-chat,messages:messages,temperature:request.temperature,max_tokens:request.max_tokens,}defparse_response(self,raw_response:Dict[str,Any])-ModelResponse:choiceraw_response[choices][0]usage_dataraw_response[usage]usageTokenUsage(input_tokensusage_data[prompt_tokens],output_tokensusage_data[completion_tokens],total_tokensusage_data[total_tokens])returnModelResponse(contentchoice[message][content],modeldeepseek-chat,usageusage,costself.calculate_cost(usage.input_tokens,usage.output_tokens),latency_ms0.0)3.5 核心路由器# gateway/router.pyclassModelRouter:多模型路由器 - 聚合层的核心ADAPTER_MAP{deepseek:DeepSeekAdapter,qwen:QWenAdapter,glm:GLMAdapter,}def__init__(self,config:Dict,credentials:Dict[str,str]):self.routing_configconfig.get(routing,{})self.default_modelself.routing_config.get(default_model,deepseek)self.fallback_chainself.routing_config.get(fallback_chain,[])self.adapters{}forname,clsinself.ADAPTER_MAP.items():ifnameincredentialsandnameinconfig[models]:self.adapters[name]cls(config[models][name],credentials[name])self.clienthttpx.AsyncClient(timeouthttpx.Timeout(60.0),limitshttpx.Limits(max_connections20))asyncdefcall(self,request:ModelRequest)-ModelResponse:智能调用首选模型 → 失败按降级链路依次尝试preferredrequest.modelifrequest.modelinself.adapterselseself.default_model candidates[preferred][mforminself.fallback_chainifmnotin[preferred]andminself.adapters]formodel_nameincandidates:resultawaitself._call_single(model_name,request)ifresult:returnresultraiseRuntimeError(f所有模型调用失败:{candidates})3.6 使用示例# examples/demo.pyimportasynciofromgatewayimportModelGatewayasyncdefmain():gatewayModelGateway(config.yaml)# 自动路由responseawaitgateway.chat(用Python写一个斐波那契数列生成器)print(f模型:{response.model}| 成本: ¥{response.cost})# 指定模型responseawaitgateway.chat(翻译人工智能正在改变世界,modelqwen)print(f结果:{response.content})# 查看统计statsgateway.stats()print(f总成本: ¥{stats[total_cost]}| 预算剩余: ¥{stats[budget_remaining]})awaitgateway.close()asyncio.run(main())四、成本对比聚合接入到底省多少一个月真实数据对比场景知识库问答系统日均2000次调用。成本下降拆解语义缓存省了约35% — 知识库场景大量相似问题缓存命中直接返回智能路由省了约15% — 简单任务走低价模型复杂任务走深度推理成本计量省了约10% — 有了明细才发现某功能每天无意义重复调用200次五、避坑指南坑1各家Token计数规则不一致。 同一个文本在A模型算100 TokenB模型可能算120。在聚合层统一用tiktoken独立计数。坑2连接池管理。 总连接数20、保持连接10、读取超时60秒是最佳实践。坑3降级链路2-3个足够。 超过3个直接返回错误总超时设30秒。坑4接入凭证永远别硬编码。 用环境变量读取生产环境接密钥管理服务。坑5缓存不适合所有场景。 时间敏感“现在几点”、实时信息、需要随机性的场景加黑名单过滤。六、总结核心就三件事统一接口一套代码调所有模型新增模型只需50行适配器智能路由自动选模型、自动降级、自动重试成本可控精确到每次调用的成本明细完整代码已整理好需要资料包的朋友评论区留言或关注后私信获取。点赞收藏关注三连支持一下。我在实际项目中使用TokEase进行Token成本管理和多模型调度计量精度和路由灵活性让方案落地更顺畅。感兴趣可以去了解一下。本文所有代码基于Apache 2.0协议开源。