FastAPI + litellm 统一代理大模型 API：优雅实现成本监控与 Fallback 策略

1. 多模型接入的混乱现场为什么需要一个统一代理层如果你正在做 AI 应用大概率遇到过这种局面产品要接 OpenAI 做主力DeepSeek 做性价比兜底偶尔还要试试通义千问的效果。每个厂商一套 SDK、一套鉴权、一套返回格式代码里到处是 if-else 判断走哪家。更麻烦的是某家 API 突然超时或限流整个请求就挂了用户看到的是 500 错误。我试过最原始的做法——在每个业务函数里写 try-except 逐个切换模型结果代码膨胀到没法维护成本统计更是无从下手。后来把代理层单独抽出来用 FastAPI 做网关、litellm 做格式统一才把这件事理顺。这套方案解决三个核心问题。第一是接口统一上游应用只发一种请求格式代理层负责翻译成各家 API 的格式。第二是故障降级主模型失败时自动切到备用模型用户无感知。第三是成本可观测每次调用的 token 消耗和费用都记录下来月底对账不再抓瞎。适合谁看如果你正在搭建 AI 中台、做多模型路由、或者单纯想给自己的 side project 加一层成本监控这篇可以直接照着搭。技术栈是 Python FastAPI litellm不需要额外的基础设施本地跑起来就能验证。整个代理层的核心思路可以用一句话概括对外暴露一个/chat接口内部按优先级依次尝试模型列表成功即返回同时记录成本。听起来简单但要做到优雅、可扩展、好排障有几个关键点需要处理好。下面从环境准备开始一步步把可运行的代码搭出来。2. TaoToken 统一 Key 通道一个 Key 打通多模型调用在写代码之前先解决一个前置问题多模型意味着多套 API Key管理起来很烦。OpenAI 一个 Key、DeepSeek 一个 Key、通义千问又一个 Key环境变量越堆越多团队协作时还要同步密钥。TaoToken 提供的是统一 Key 通道你只需要一个 API Key就能通过兼容 OpenAI 的接口调用多家模型。这对代理层来说非常友好——litellm 本身就支持自定义api_base把请求指向 TaoToken 的 API 地址模型名称按规范传入即可。具体来说TaoToken 的 API 地址是https://taotoken.net/api兼容 OpenAI 的/v1/chat/completions路径。你在代码里配置api_base和api_keylitellm 就会把请求发到这里由 TaoToken 路由到对应的底层模型。这样代理层不需要为每家厂商单独配置密钥一个 Key 搞定。对于成本监控来说统一通道还有个额外好处所有调用的计费口径一致不用去各家后台分别拉账单。你可以在代理层统一记录 token 用量结合价格表算出费用数据来源单一对账清晰。如果你还没有 Key可以去 TaoToken 控制台创建一个。拿到 Key 之后先别急着写完整代理用最简单的 curl 验证一下通道是否通畅curl https://taotoken.net/api/v1/chat/completions \ -H Content-Type: application/json \ -H Authorization: Bearer sk-你的Key \ -d { model: gpt-4o-mini, messages: [{role: user, content: 说一句你好}] }如果返回正常的 JSON 结构说明 Key 和通道都没问题。这一步很重要因为后面代理层排障时需要先排除是通道问题还是代码问题。确认通道可用后再进入 FastAPI litellm 的搭建环节。3. 可复制的 FastAPI litellm 代理配置路由、Fallback 与成本中间件这一节是全文的核心给出可以直接复制运行的完整配置。我会把代码拆成几个文件方便你按模块理解。整体结构是main.py放 FastAPI 应用和路由config.py放模型列表和价格表cost.py放成本记录服务。先安装依赖pip install fastapi uvicorn litellm pydantic然后创建config.py定义模型优先级和价格表。价格表用每千 token 的美元单价你可以根据实际采购价格调整# config.py MODEL_PRIORITY [ gpt-4o-mini, deepseek-chat, qwen-turbo, ] PRICE_TABLE { gpt-4o-mini: {input: 0.00015, output: 0.0006}, deepseek-chat: {input: 0.00014, output: 0.00028}, qwen-turbo: {input: 0.00005, output: 0.0001}, } TAOTOKEN_API_BASE https://taotoken.net/api TAOTOKEN_API_KEY sk-你的Key接着写cost.py用一个简单的内存列表记录成本。生产环境可以换成 Redis 或数据库但接口设计保持一致# cost.py import time from config import PRICE_TABLE class CostService: def __init__(self): self.records [] def add(self, model: str, prompt_tokens: int, completion_tokens: int): prices PRICE_TABLE.get(model, {input: 0, output: 0}) cost (prompt_tokens / 1000) * prices[input] \ (completion_tokens / 1000) * prices[output] self.records.append({ model: model, prompt_tokens: prompt_tokens, completion_tokens: completion_tokens, cost: cost, timestamp: time.time(), }) return cost def stats(self): if not self.records: return {total_cost: 0, count: 0, by_model: {}} total sum(r[cost] for r in self.records) by_model {} for r in self.records: by_model.setdefault(r[model], {cost: 0, count: 0}) by_model[r[model]][cost] r[cost] by_model[r[model]][count] 1 return {total_cost: round(total, 6), count: len(self.records), by_model: by_model}核心的main.py来了。这里用 litellm 的acompletion做异步调用配合 FastAPI 的异步路由。Fallback 逻辑是遍历模型列表捕获异常后继续尝试下一个# main.py import litellm from fastapi import FastAPI, HTTPException, Depends from pydantic import BaseModel from typing import List, Optional from config import MODEL_PRIORITY, TAOTOKEN_API_BASE, TAOTOKEN_API_KEY from cost import CostService app FastAPI(titleLLM Proxy) cost_service CostService() class ChatRequest(BaseModel): prompt: str models: Optional[List[str]] None temperature: Optional[float] 0.7 class ChatResponse(BaseModel): text: str model_used: str cost: float async def call_with_fallback(models: List[str], prompt: str, temperature: float): last_error None for model in models: try: response await litellm.acompletion( modelfopenai/{model}, messages[{role: user, content: prompt}], temperaturetemperature, api_baseTAOTOKEN_API_BASE, api_keyTAOTOKEN_API_KEY, ) content response.choices[0].message.content usage response.usage cost cost_service.add( model, usage.prompt_tokens, usage.completion_tokens, ) return content, model, cost except Exception as e: last_error e continue raise HTTPException(status_code503, detailfAll models failed: {last_error}) app.post(/chat, response_modelChatResponse) async def chat_endpoint(request: ChatRequest): models request.models or MODEL_PRIORITY content, model_used, cost await call_with_fallback( models, request.prompt, request.temperature ) return ChatResponse(textcontent, model_usedmodel_used, costcost) app.get(/cost/stats) async def get_cost_stats(): return cost_service.stats()注意model参数传的是openai/{model}格式这是 litellm 的约定——告诉它用 OpenAI 兼容协议发送请求实际地址由api_base决定。这样所有模型都走 TaoToken 通道不需要为每家单独配置。启动服务uvicorn main:app --reload --port 8000到这里一个带 Fallback 和成本监控的代理层就跑起来了。下一节验证实际请求效果。4. 验证请求与成功结果从单次调用到成本统计服务启动后先用一个简单请求验证主流程。打开另一个终端发一个 POST 请求curl -X POST http://localhost:8000/chat \ -H Content-Type: application/json \ -d { prompt: 用一句话解释什么是向量数据库, temperature: 0.7 }预期返回类似这样的 JSON{ text: 向量数据库是一种专门存储和检索高维向量数据的数据库常用于相似度搜索和推荐系统。, model_used: gpt-4o-mini, cost: 0.000042 }model_used显示实际命中的模型cost是这次调用的估算费用。如果主模型正常应该命中MODEL_PRIORITY里的第一个。接下来验证 Fallback。手动把第一个模型改成一个不存在的名称比如gpt-4o-mini-typo再发请求curl -X POST http://localhost:8000/chat \ -H Content-Type: application/json \ -d { prompt: 测试降级, models: [gpt-4o-mini-typo, deepseek-chat] }这次第一个模型会报错代理层自动切到deepseek-chat返回的model_used应该是deepseek-chat。这说明 Fallback 生效了。最后查看成本统计curl http://localhost:8000/cost/stats返回结果会按模型聚合类似{ total_cost: 0.000078, count: 2, by_model: { gpt-4o-mini: {cost: 0.000042, count: 1}, deepseek-chat: {cost: 0.000036, count: 1} } }到这里统一代理、故障降级、成本监控三个目标都验证通过了。你可以把这个/cost/stats接口接到 Grafana 或前端面板实时看调用量和费用趋势。有个细节值得注意litellm 返回的usage字段在不同模型下可能略有差异但走 TaoToken 统一通道后格式是一致的所以成本计算逻辑不需要为每个模型写分支。这也是统一通道带来的便利。5. 常见报错排查401、local proxy failed、reading choices 与 OAuth实际跑起来后大概率会遇到几个典型报错。这一节按报错信息对照排查都是真实踩过的坑。401 Authentication Error最常见的原因是 Key 没传对。检查TAOTOKEN_API_KEY是否以sk-开头有没有多余空格。另外注意 litellm 的api_key参数优先级高于环境变量如果你同时设置了OPENAI_API_KEY环境变量可能会被覆盖。建议在代码里显式传参避免混淆。local proxy failed / Connection error这个报错通常出现在api_base配置错误时。确认地址是https://taotoken.net/api不要漏掉/api路径也不要多加/v1——litellm 会自动拼接/v1/chat/completions。如果你在本地开了其他代理工具可能会干扰请求先关掉再试。reading choices / KeyError choices说明返回的 JSON 结构不符合预期。可能是模型名称写错了TaoToken 返回了错误信息而不是正常的 completion 结构。打印完整的response对象看看通常能看到具体的错误原因。另外确认model参数传的是openai/{model}格式少了openai/前缀 litellm 可能走错协议。OAuth / token expired如果你用的是某些需要 OAuth 的模型可能会遇到 token 过期。走 TaoToken 统一通道的话鉴权由通道处理你只需要保证自己的 API Key 有效。如果 Key 被禁用或额度耗尽也会返回类似鉴权失败的提示去控制台检查一下 Key 状态和余额。Fallback 不生效检查异常捕获的范围。litellm 抛出的异常类型比较多用except Exception能兜住大部分情况。另外确认模型列表里至少有一个可用模型如果全部失败会返回 503。排障时有个通用技巧先把litellm.acompletion单独拿出来在 Python 交互环境里跑一次确认通道和参数没问题再放回 FastAPI 里。这样能快速定位是通道问题还是框架问题。6. 从代理层到生产接入文档与 Coding Plan 的衔接代理层跑通之后下一步通常是接入实际业务。如果你用的是 Claude Code 做编码辅助或者想把代理层接到 Cline、Codex 这类工具里需要配置三件套Base URL、API Key、Model ID。Base URL 填https://taotoken.net/apiAPI Key 用你在控制台创建的那个Model ID 按工具要求填对应模型名称。以 Claude Code 为例在配置文件里指定ANTHROPIC_BASE_URL和ANTHROPIC_API_KEY就能把请求导向统一通道。Cline 的 MCP 配置类似在 settings 里填好这三项即可。如果你需要长期跑编码任务或 Agent 工作流可以了解一下 Coding Plan它针对高频调用场景做了额度优化。接入文档里有各工具的详细配置步骤包括 Claude Code、Cline、Codex 的 auth.json 写法照着填就行。验证模型是否正常响应可以用模型对话页面快速测一下不用写代码就能确认通道通畅。API Keys 管理页面可以创建和轮换 Key建议给不同环境分配不同的 Key方便追踪用量。整套方案的核心价值在于用一层薄薄的代理把多模型接入的复杂度收敛到一个地方。业务代码只关心 prompt模型切换、故障降级、成本统计都在代理层完成。后续要加新模型只需要在MODEL_PRIORITY和PRICE_TABLE里加一行配置不用改业务逻辑。这种架构在模型快速迭代的当下能省下不少重构时间。

FastAPI + litellm 统一代理大模型 API：优雅实现成本监控与 Fallback 策略｜TaoToken 统一 Key 通道实践

相关推荐

3分钟看懂MCP协议：TaoToken如何让AI的“万能插头”真正通电

AtomCode Token 消耗与成本控制实测：CodingPlan 免费额度够不够用，TaoToken 统一 Key 通道怎么配

参与OpenCloudOS社区：CubeSandbox实操教程与TaoToken接入实践

更多资讯

Claude Code 里的 MCP / Skills / Hooks / Commands：把 settings 改到 TaoToken 的完整配置清单

502个中文AI工具清单：分类逻辑、筛选决策树与工程化维护实践

后端转AI必会：如何用数据证明大模型系统有效？评估体系全解

模块化用法

DeepBot Web服务端部署教程：Docker构建、JWT认证与WebSocket架构实战

前端面试题：让 AI 生成组件，怎么保证不重复造轮子？

今日更新

Agent Skills 完全指南：原理、写法、安装与实战避坑

Agent Skills 实战：从 Genkit 定义到 GKE 部署与排查

Agent Skills 实战：从设计到调试的完整指南

本周精选

CMOS设计三大物理效应：体效应、沟道长度调制与亚阈值导电

微信小程序在线阅读毕设全解析：Java后端+MySQL+避坑实战

Android音频问题五层分析法：从App到Codec的系统化诊断

本月热门

我发现了一个新思路：用 Remotion + Claude Code 像写代码一样自动化生成短视频

Windows下 Codex 中 Chrome 和 Computer Use 插件不可用问题排查及解决参考方式：TaoToken 统一 Key 配置与验证

2026 大模型集体涨价：用 Python 做企业 Token 成本测算与选型避坑（附配置）

读完文章，想聊聊您的网站？