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

Olercs-tuisr

401人浏览 · 2026-03-18 19:00:00

Olercs-tuisr · 2026-03-18 19:00:00 发布

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

1. 为什么需要统一代理？

现在的大模型 API 百花齐放。

OpenAI、DeepSeek、通义千问、文心一言……每个提供商都有自己的 SDK 和请求格式。如果你的应用需要调用多家模型，代码会变得杂乱无章，难以维护。

统一代理的核心思想是：用一个中间层屏蔽底层差异，让上游应用用一套标准接口访问所有模型。

它的好处很明显：

接口统一：应用层只需关心 prompt，不用关心是哪个模型。
故障转移：某个模型超时或报错，自动切换到备用模型。
成本可观测：集中统计 token 消耗，方便核算费用。

下面介绍如何使用 FastAPI 和 litellm 快速搭建这样一个代理服务，并实现优雅的 fallback 和成本监控。

2. litellm 是什么？

简单说，litellm 是一个 Python 库，它把几十家大模型 API 的格式统一成了 OpenAI 的格式。

它的基本思想是：你只需要记住一个函数 completion()，传入 model 名称和 messages，它就能自动调用对应的 API。

例如：

import litellm

response = litellm.completion(
    model="openai/gpt-4",
    messages=[{"role": "user", "content": "Hello"}]
)

如果换成 DeepSeek，只需改 model 为 deepseek/deepseek-chat 并配置相应的环境变量。litellm 会自动处理鉴权和请求格式。

3. 整体架构设计

我们构建一个 FastAPI 应用，对外提供 /chat 接口。

内部流程如下：

接收请求（包含 prompt 和可选的模型列表）。
按优先级依次尝试模型，直到某个模型成功返回。
记录每次调用的 token 用量和成本。
返回最终结果给客户端。

我们可以这样理解：这是一个带有“断路器”和“计价器”的智能路由器。

4. 代码实现

4.1 环境准备

安装依赖：

pip install fastapi uvicorn litellm pydantic

设置环境变量（以 OpenAI 和 DeepSeek 为例）：

export OPENAI_API_KEY="sk-xxx"
export DEEPSEEK_API_KEY="sk-yyy"

4.2 定义请求和响应模型

from pydantic import BaseModel
from typing import List, Optional

class ChatRequest(BaseModel):
    prompt: str
    models: List[str] = ["openai/gpt-4", "deepseek/deepseek-chat"]  # 按优先级排序
    temperature: Optional[float] = 0.7

class ChatResponse(BaseModel):
    text: str
    model_used: str
    total_cost: float

4.3 初始化成本跟踪

我们用一个全局列表来记录每次调用的成本（生产环境建议用数据库）。

import time
cost_records = []

def record_cost(model: str, tokens: int, cost: float):
    cost_records.append({
        "model": model,
        "tokens": tokens,
        "cost": cost,
        "timestamp": time.time()
    })

4.4 核心 Fallback 逻辑

这里是最关键的部分。我们写一个函数，依次尝试模型列表，直到成功。

import litellm
from fastapi import HTTPException

def call_with_fallback(models: List[str], prompt: str, temperature: float):
    last_exception = None
    for model in models:
        try:
            # 调用 litellm
            response = litellm.completion(
                model=model,
                messages=[{"role": "user", "content": prompt}],
                temperature=temperature
            )
            # 解析结果
            content = response.choices[0].message.content
            usage = response.usage
            prompt_tokens = usage.prompt_tokens
            completion_tokens = usage.completion_tokens
            total_tokens = usage.total_tokens

            # 计算成本（这里使用硬编码价格，实际应动态获取）
            # OpenAI gpt-4 约 $0.03/1k 输入，$0.06/1k 输出，这里简化处理
            cost = (prompt_tokens / 1000) * 0.03 + (completion_tokens / 1000) * 0.06

            # 记录成本
            record_cost(model, total_tokens, cost)

            return content, model, cost

        except Exception as e:
            last_exception = e
            continue  # 失败就尝试下一个模型

    # 所有模型都失败
    raise HTTPException(status_code=503, detail=f"All models failed: {last_exception}")

4.5 FastAPI 路由

from fastapi import FastAPI

app = FastAPI()

@app.post("/chat", response_model=ChatResponse)
async def chat_endpoint(request: ChatRequest):
    content, model_used, cost = call_with_fallback(
        models=request.models,
        prompt=request.prompt,
        temperature=request.temperature
    )
    return ChatResponse(
        text=content,
        model_used=model_used,
        total_cost=cost
    )

4.6 成本监控接口

我们再加一个接口，方便查看累计成本。

@app.get("/cost/stats")
async def get_cost_stats():
    if not cost_records:
        return {"total_cost": 0, "count": 0}
    total = sum(r["cost"] for r in cost_records)
    return {"total_cost": total, "count": len(cost_records)}

5. 更优雅的改进点

上面的代码已经可以工作，但还有几个可以“更优雅”的地方。

5.1 使用依赖注入共享记录

我们可以把 cost_records 封装成一个服务类，通过 FastAPI 的 Depends 注入，方便测试。

from fastapi import Depends

class CostService:
    def __init__(self):
        self.records = []

    def add(self, model, tokens, cost):
        self.records.append(...)

    def stats(self):
        ...

@app.post("/chat")
async def chat(request: ChatRequest, cost_service: CostService = Depends()):
    ...

5.2 动态获取价格

上面的代码硬编码了价格，不够灵活。我们可以从配置文件或数据库读取价格表。

PRICE_TABLE = {
    "openai/gpt-4": {"input": 0.03, "output": 0.06},
    "deepseek/deepseek-chat": {"input": 0.001, "output": 0.002},  # 假设数字
}

def calculate_cost(model, prompt_tokens, completion_tokens):
    prices = PRICE_TABLE.get(model, {"input": 0, "output": 0})
    return (prompt_tokens / 1000) * prices["input"] + (completion_tokens / 1000) * prices["output"]

5.3 异步调用提高并发

litellm 支持异步调用，使用 litellm.acompletion()。我们可以把 fallback 函数改为异步，提高并发能力。

async def call_with_fallback_async(models, prompt, temperature):
    for model in models:
        try:
            response = await litellm.acompletion(
                model=model,
                messages=[{"role": "user", "content": prompt}],
                temperature=temperature
            )
            # ... 处理
        except Exception:
            continue