人工智能实战多模型服务如何统一管理从硬编码模型地址到 Model Gateway 的工程化架构一、问题场景模型越来越多系统越来越乱在大模型系统演进过程中最开始通常只有一个模型Qwen接口里直接写死MODEL_URLhttp://127.0.0.1:8001但随着业务发展很快会出现1. 一个模型负责通用问答 2. 一个模型负责代码生成 3. 一个模型负责结构化抽取 4. 一个小模型负责快速分类 5. 一个备用模型负责降级 6. 一个外部 API 作为兜底于是代码里开始出现iftaskchat:urlhttp://qwen:8000eliftaskcode:urlhttp://deepseek-code:8000eliftaskextract:urlhttp://small-model:8000一开始还能忍后面就会变成灾难。典型问题1. 模型地址散落在各个服务 2. 切换模型要改代码 3. 灰度发布困难 4. 模型失败没有统一降级 5. 成本和延迟无法统一统计 6. 多业务重复接入模型这篇文章解决的问题是如何设计一个 Model Gateway统一管理多模型路由、降级、监控和调用。二、为什么需要 Model GatewayModel Gateway 的本质不是代理转发而是模型调用控制平面。它负责1. 模型注册 2. 任务路由 3. 参数校验 4. 降级兜底 5. 统一日志 6. 成本统计 7. 灰度发布架构从业务系统 → 模型服务升级为业务系统 → Model Gateway → 模型服务三、目标架构Client ↓ Business API ↓ Model Gateway ├── Chat Model ├── Code Model ├── Extract Model └── Fallback Model核心收益业务不关心模型地址 业务不关心降级策略 业务不关心模型切换四、模型注册表先定义模型配置。MODEL_REGISTRY{chat-primary:{url:http://qwen:8000/v1/chat/completions,model:qwen-chat,type:chat,timeout:60},chat-backup:{url:http://backup:8000/v1/chat/completions,model:backup-chat,type:chat,timeout:60},extract-small:{url:http://extractor:8000/v1/chat/completions,model:extract-small,type:extract,timeout:30}}任务路由TASK_ROUTE{general_chat:[chat-primary,chat-backup],info_extract:[extract-small,chat-primary]}注意这里用数组第一个是主模型 后面是备用模型五、项目结构model-gateway-demo/ ├── app.py ├── registry.py ├── router.py ├── client.py ├── schemas.py └── metrics.py安装依赖pipinstallfastapi uvicorn httpx pydantic六、请求 SchemafrompydanticimportBaseModel,FieldfromtypingimportLiteral,OptionalclassModelRequest(BaseModel):task:Literal[general_chat,info_extract]prompt:strField(...,min_length1,max_length4000)max_tokens:intField(default256,ge1,le1024)temperature:floatField(default0.7,ge0,le2)classModelResponse(BaseModel):answer:strmodel_used:strfallback_used:boolcost_ms:int七、模型客户端 client.pyimporttimeimporthttpxasyncdefcall_openai_compatible(model_config:dict,prompt:str,max_tokens:int,temperature:float):starttime.time()payload{model:model_config[model],messages:[{role:user,content:prompt}],max_tokens:max_tokens,temperature:temperature}asyncwithhttpx.AsyncClient(timeoutmodel_config[timeout])asclient:respawaitclient.post(model_config[url],jsonpayload)resp.raise_for_status()dataresp.json()answerdata[choices][0][message][content]return{answer:answer,cost_ms:int((time.time()-start)*1000)}八、路由与降级 router.pyfromregistryimportMODEL_REGISTRY,TASK_ROUTEfromclientimportcall_openai_compatibleasyncdefroute_and_call(task:str,prompt:str,max_tokens:int,temperature:float):candidatesTASK_ROUTE[task]last_errorNoneforidx,model_nameinenumerate(candidates):configMODEL_REGISTRY[model_name]try:resultawaitcall_openai_compatible(config,prompt,max_tokens,temperature)return{answer:result[answer],model_used:model_name,fallback_used:idx0,cost_ms:result[cost_ms]}exceptExceptionase:last_errorstr(e)continueraiseRuntimeError(fall models failed:{last_error})这个逻辑实现了主模型失败 → 自动尝试备用模型九、FastAPI 网关 app.pyfromfastapiimportFastAPI,HTTPExceptionfromschemasimportModelRequest,ModelResponsefromrouterimportroute_and_call appFastAPI(titleModel Gateway)app.post(/generate,response_modelModelResponse)asyncdefgenerate(req:ModelRequest):try:resultawaitroute_and_call(taskreq.task,promptreq.prompt,max_tokensreq.max_tokens,temperaturereq.temperature)returnModelResponse(**result)exceptExceptionase:raiseHTTPException(500,str(e))app.get(/health)defhealth():return{status:ok}启动uvicorn app:app--port8000十、测试请求curl-XPOSThttp://127.0.0.1:8000/generate\-HContent-Type: application/json\-d{ task: general_chat, prompt: 解释一下什么是模型网关, max_tokens: 128, temperature: 0.7 }返回{answer:...,model_used:chat-primary,fallback_used:false,cost_ms:1200}如果主模型失败则可能返回{model_used:chat-backup,fallback_used:true}十一、灰度发布设计假设要把 10% 流量切到新模型importrandom TASK_ROUTE_GRAY{general_chat:{stable:chat-primary,gray:chat-new,gray_ratio:0.1,backup:chat-backup}}defchoose_model_for_gray(task:str):configTASK_ROUTE_GRAY[task]ifrandom.random()config[gray_ratio]:return[config[gray],config[backup]]return[config[stable],config[backup]]这样可以做到不改业务代码 只改网关配置十二、监控指标Model Gateway 必须记录1. 每个模型调用次数 2. 每个模型错误率 3. 每个模型平均耗时 4. fallback 次数 5. 不同 task 的 token 成本示例日志{task:general_chat,model_used:chat-backup,fallback_used:True,cost_ms:1800,status:success}如果 fallback 频率突然升高说明主模型可能出问题。十三、踩坑记录坑 1业务代码直接写模型地址后期模型一多维护成本极高。坑 2没有 fallback主模型失败时整个业务失败。Model Gateway 应该内置降级策略。坑 3所有任务用一个模型不同任务需要不同模型。例如分类任务用小模型 复杂推理用大模型 结构化抽取用稳定模型坑 4没有记录 model_used线上问题排查时你必须知道到底哪个模型回答的。坑 5灰度发布靠改代码模型灰度应该是配置问题不应该改业务代码。十四、适合收藏的 Model Gateway Checklist模型注册 [ ] 是否统一管理模型地址 [ ] 是否有模型类型 [ ] 是否配置超时 [ ] 是否支持备用模型 路由 [ ] 是否按 task 路由 [ ] 是否支持 fallback [ ] 是否支持灰度 [ ] 是否支持权重 调用 [ ] 是否统一参数校验 [ ] 是否统一错误处理 [ ] 是否统一日志 [ ] 是否记录 model_used 监控 [ ] 是否统计模型耗时 [ ] 是否统计错误率 [ ] 是否统计 fallback 次数 [ ] 是否统计成本 运维 [ ] 是否支持配置热更新 [ ] 是否支持模型下线 [ ] 是否支持回滚十五、经验总结当系统里只有一个模型时直接调用模型服务没问题。但当模型变成多个后必须引入 Model Gateway。它的核心价值是把模型调用从业务代码中解耦出来。一句话总结Model Gateway 不是转发层而是大模型系统的流量控制中心。十六、优化建议后续可以继续做1. 配置中心管理模型路由 2. Prometheus 统计模型级指标 3. 支持按用户灰度 4. 支持按成本选择模型 5. 支持语义缓存 6. 支持多供应商兜底 7. 支持限流和熔断 8. 支持模型 A/B 测试最后一句经验模型越多越不能让业务直接感知模型。