本文记录从零搭建一个支持GPT、Claude、Gemini、DeepSeek、Qwen的多模型聊天后端的全过程。包含项目结构设计、SDK选型、流式处理、错误重试、成本控制的完整代码。一、项目背景团队需要一个内部AI助手要求支持多个模型自由切换统一的API接口前端不用关心后端用的是哪个模型流式输出用户体验好成本可控能按模型分别统计用量技术栈Python FastAPI OpenAI SDK兼容协议二、项目结构multi-model-chat/ ├── app/ │ ├── main.py # FastAPI 入口 │ ├── config.py # 配置管理 │ ├── models.py # 模型路由配置 │ ├── router.py # 请求路由与转发 │ ├── billing.py # 用量统计 │ └── utils.py # 工具函数 ├── requirements.txt └── .env三、第1步配置管理不同模型的API地址和Key分开管理。这里我们通过环境变量加载支持多个中转站和直连方式# app/config.pyimportosfromdataclassesimportdataclassfromtypingimportDictdataclassclassProviderConfig:name:strbase_url:strapi_key:strmodels:list# 从环境变量加载配置# 支持多种接入方式# 1. 直连官方API如 OpenAI 官方# 2. 通过中转站接入如 OpenRouter、硅基流动、魔芋AI 等# 3. 自建中转层如 one-api、new-apidefload_providers()-Dict[str,ProviderConfig]:providers{}# OpenAI 直连ifos.getenv(OPENAI_API_KEY):providers[openai]ProviderConfig(nameOpenAI 直连,base_urlhttps://api.openai.com/v1,api_keyos.getenv(OPENAI_API_KEY),models[gpt-4o,gpt-4o-mini,gpt-4-turbo])# OpenRouter 中转国际主流中转站# 官网https://openrouter.aiifos.getenv(OPENROUTER_API_KEY):providers[openrouter]ProviderConfig(nameOpenRouter,base_urlhttps://openrouter.ai/api/v1,api_keyos.getenv(OPENROUTER_API_KEY),models[gpt-4o,claude-3.5-sonnet,gemini-2.0-flash])# 硅基流动国内中转站开源模型为主# 官网https://siliconflow.cnifos.getenv(SILICONFLOW_API_KEY):providers[siliconflow]ProviderConfig(name硅基流动,base_urlhttps://api.siliconflow.cn/v1,api_keyos.getenv(SILICONFLOW_API_KEY),models[deepseek-v3,qwen2.5-72b,yi-34b])# 魔芋AI国内中转站性价比方案# 官网https://www.moyu.info/register?affCRB8ifos.getenv(MOYU_API_KEY):providers[moyu]ProviderConfig(name魔芋AI,base_urlhttps://api.moyu.info/v1,api_keyos.getenv(MOYU_API_KEY),models[gpt-4o,claude-3.5-sonnet,deepseek-chat])returnproviders配置说明OpenAI直连适合对延迟要求极高的场景但国内访问需要代理OpenRouteropenrouter.ai国际主流中转站模型覆盖最全支持200模型硅基流动siliconflow.cn国内中转站开源模型为主价格便宜魔芋AImoyu.info国内中转站支持主流闭源模型新用户有免费额度自建方案用 one-api 或 new-api 开源项目自建中转层适合对数据隐私要求高的场景四、第2步模型路由根据用户请求的模型名称自动路由到对应的中转站# app/router.pyfromopenaiimportAsyncOpenAIfrom.configimportProviderConfigfromtypingimportOptionalimportlogging loggerlogging.getLogger(__name__)classModelRouter:def__init__(self,providers:dict):self.providersproviders# 建立模型到Provider的反向索引self.model_index{}forprovider_name,configinproviders.items():formodelinconfig.models:self.model_index[model]provider_name logger.info(f已加载{len(self.model_index)}个模型路由)defget_client(self,model:str)-Optional[tuple]:根据模型名获取对应的AsyncOpenAI客户端provider_nameself.model_index.get(model)ifnotprovider_name:logger.error(f模型{model}未找到可用的Provider)returnNoneconfigself.providers[provider_name]clientAsyncOpenAI(api_keyconfig.api_key,base_urlconfig.base_url)returnclient,provider_namedeflist_models(self)-list:列出所有可用模型returnlist(self.model_index.keys())核心思路所有中转站都兼容OpenAI协议所以用同一个AsyncOpenAI客户端只是base_url不同。路由表负责把模型名映射到对应的中转站。五、第3步流式聊天接口实现一个统一的流式聊天接口前端通过SSE接收# app/main.pyfromfastapiimportFastAPI,HTTPExceptionfromfastapi.responsesimportStreamingResponsefrompydanticimportBaseModelfrom.routerimportModelRouterfrom.configimportload_providersfrom.billingimportUsageTrackerimportjsonimportlogging logging.basicConfig(levellogging.INFO)appFastAPI(title多模型AI聊天后端)# 初始化providersload_providers()routerModelRouter(providers)trackerUsageTracker()classChatRequest(BaseModel):model:strmessages:listtemperature:float0.7max_tokens:int2000app.post(/chat/stream)asyncdefchat_stream(req:ChatRequest):流式聊天接口resultrouter.get_client(req.model)ifnotresult:raiseHTTPException(404,f模型{req.model}不可用)client,provider_nameresultasyncdefgenerate():total_tokens0try:streamawaitclient.chat.completions.create(modelreq.model,messagesreq.messages,temperaturereq.temperature,max_tokensreq.max_tokens,streamTrue)asyncforchunkinstream:ifchunk.choicesandchunk.choices[0].delta.content:contentchunk.choices[0].delta.content total_tokens1# 粗略统计# SSE 格式返回yieldfdata:{json.dumps({content:content,model:req.model})}\n\n# 流结束发送统计信息yieldfdata:{json.dumps({done:True,tokens:total_tokens,provider:provider_name})}\n\nexceptExceptionase:logger.error(f聊天出错:{e})yieldfdata:{json.dumps({error:str(e)})}\n\nreturnStreamingResponse(generate(),media_typetext/event-stream)app.get(/models)asyncdeflist_models():列出所有可用模型return{models:router.list_models()}六、第4步错误处理与重试网络请求难免失败需要实现自动重试和Provider降级# app/utils.pyimportasyncioimportloggingfromfunctoolsimportwraps loggerlogging.getLogger(__name__)defretry_with_backoff(max_retries3,base_delay1.0):指数退避重试装饰器defdecorator(func):wraps(func)asyncdefwrapper(*args,**kwargs):last_errorNoneforattemptinrange(max_retries):try:returnawaitfunc(*args,**kwargs)exceptExceptionase:last_errore delaybase_delay*(2**attempt)logger.warning(f第{attempt1}/{max_retries}次重试f等待{delay}s错误:{e})awaitasyncio.sleep(delay)raiselast_errorreturnwrapperreturndecoratorclassFallbackChain:Provider降级链主Provider失败时自动切换到备用def__init__(self,router):self.routerrouterasyncdefchat_with_fallback(self,model,messages,**kwargs):带降级的聊天# 获取主Providerresultself.router.get_client(model)ifnotresult:raiseValueError(f模型{model}不可用)primary_client,primary_providerresulttry:# 尝试主Providerresponseawaitprimary_client.chat.completions.create(modelmodel,messagesmessages,**kwargs)returnresponse,primary_providerexceptExceptionase:logger.warning(f主Provider{primary_provider}失败:{e})# 尝试备用Provider同模型在其他中转站forprovider_name,configinself.router.providers.items():ifprovider_nameprimary_provider:continueifmodelinconfig.models:try:backup_clientAsyncOpenAI(api_keyconfig.api_key,base_urlconfig.base_url)responseawaitbackup_client.chat.completions.create(modelmodel,messagesmessages,**kwargs)logger.info(f降级到{provider_name}成功)returnresponse,provider_nameexceptExceptionase2:logger.warning(f备用{provider_name}也失败:{e2})continueraiseException(f所有Provider均失败最后错误:{e})七、第5步用量统计按模型和Provider分别统计Token用量方便成本分析# app/billing.pyimporttimefromcollectionsimportdefaultdictimportlogging loggerlogging.getLogger(__name__)classUsageTracker:def__init__(self):# {model: {provider: {calls: int, tokens: int}}}self.statsdefaultdict(lambda:defaultdict(lambda:{calls:0,tokens:0}))defrecord(self,model:str,provider:str,tokens:int):记录一次调用self.stats[model][provider][calls]1self.stats[model][provider][tokens]tokens logger.info(f用量记录: model{model}, provider{provider}, ftokens{tokens}, total_calls{self.stats[model][provider][calls]})defget_summary(self)-dict:获取用量汇总summary{}formodel,providersinself.stats.items():summary[model]{p:{calls:d[calls],tokens:d[tokens]}forp,dinproviders.items()}returnsummarydefestimate_cost(self,model:str,tokens:int)-float:估算单次调用成本美元# 简化版价格表实际应从配置读取price_table{gpt-4o:0.0025,# $2.5/1Mgpt-4o-mini:0.00015,# $0.15/1Mclaude-3.5-sonnet:0.003,# $3/1Mdeepseek-chat:0.00027,# $0.27/1Mgemini-2.0-flash:0.0001,# $0.1/1M}rateprice_table.get(model,0.001)# 默认 $1/1Mreturntokens*rate/1000八、第6步启动与测试# 启动服务# uvicorn app.main:app --host 0.0.0.0 --port 8000# 测试请求importrequests# 1. 查看可用模型resprequests.get(http://localhost:8000/models)print(resp.json())# 2. 流式聊天importhttpxasyncwithhttpx.AsyncClient()asclient:asyncwithclient.stream(POST,http://localhost:8000/chat/stream,json{model:gpt-4o-mini,messages:[{role:user,content:你好}]})asresp:asyncforlineinresp.aiter_lines():ifline.startswith(data: ):importjson datajson.loads(line[6:])ifcontentindata:print(data[content],end,flushTrue)elifdoneindata:print(f\n\n完成:{data})九、踩坑记录坑1不同中转站的SSE格式略有差异大部分中转站的SSE格式和OpenAI官方一致但少数中转站会在每个chunk后多加一个空行或者finish_reason的值不同。建议在generate()函数里做兼容处理# 兼容不同中转站的SSE格式ifchunk.choices:choicechunk.choices[0]ifhasattr(choice,delta)andchoice.delta.content:contentchoice.delta.contentelifhasattr(choice,message)andchoice.message.content:contentchoice.message.contentelse:continue坑2Anthropic模型的system消息处理Claude的system消息是独立字段不是放在messages里。好的中转站会自动处理这个转换但部分中转站不做转换导致Claude收不到system提示。接入前用相同prompt测试一下。坑3Token统计不一致不同中转站用的Tokenizer可能不同同一个请求在不同中转站计费Token数有1-3%偏差。建议以模型官方的Tokenizer为准中转站的统计仅作参考。十、总结搭建多模型AI应用的核心是统一接口 路由分发。因为OpenAI兼容协议已经成为事实标准所有主流中转站都支持所以我们只需要用同一个SDKOpenAI Python SDK不同的base_url指向不同中转站路由表负责模型名到中转站的映射这种架构的好处是扩展性极强——新增一个模型只需要在配置里加一行路由表自动生效。如果你刚开始搭建可以参考文中提到的几个中转站OpenRouter、硅基流动、魔芋AI等选择适合自己场景的组合。完整代码已开源有问题欢迎评论区交流。