昨天深夜调试一个模型服务接口,客户端传过来的JSON里有个字段叫input_data,服务端定义的Pydantic模型里字段名是inputData。就这一个大小写差异,前端死活收不到推理结果。用FastAPI的自动文档试了一下才发现,默认配置下它严格区分蛇形命名和驼峰命名。这个问题让我意识到,框架选得再新潮,细节配置不到位照样踩坑。

为什么是FastAPI?

现在Python后端框架选择不少,Flask轻量但生态散,Django重但自带全家桶。FastAPI站在中间那个微妙的位置——它不像Flask那样需要自己拼装各种插件,又比Django更适配现代异步编程。最关键的是,它天生为AI应用设计:自动生成OpenAPI文档、内置数据验证、原生支持async/await。你部署个模型服务,总不能每次改接口都手动更新API文档吧?

看看这个最简单的例子:

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI(title="模型服务")  # 这里title一定要写,文档里显示用

class InferenceRequest(BaseModel):
    text: str
    temperature: float = 0.7  # 默认值这么写,客户端不传就用这个

@app.post("/predict")
async def predict(request: InferenceRequest):
    # 不用写request.json(),FastAPI自动帮你反序列化
    processed_text = preprocess(request.text)
    # 假设这里调用模型
    return {"result": processed_text}

启动服务后访问/docs,完整的交互式文档已经在那儿了。前端同事直接对着文档调试,省了一半扯皮时间。

数据验证的坑

Pydantic是FastAPI的默认验证库,但它的行为有时候很微妙。比如刚才说的命名转换问题:

class UserRequest(BaseModel):
    user_name: str  # 蛇形命名
    userAge: int    # 驼峰命名
    
    class Config:
        alias_generator = None  # 关掉自动别名生成,避免混乱

默认情况下,FastAPI会尝试在蛇形和驼峰之间自动转换。如果你的前端团队坚持用驼峰命名,最好显式配置:

class Config:
    allow_population_by_field_name = True  # 允许用字段名或别名填充
    alias_generator = to_camel  # 需要自己实现转换函数

另一个常见坑是可选字段。Pydantic v2之后语法变了:

# 错误写法(v1语法)
class OldModel(BaseModel):
    optional_field: Optional[str] = None

# 正确写法(v2)
from typing import Union
class NewModel(BaseModel):
    optional_field: Union[str, None] = None
    # 或者更简洁的
    optional_field: str | None = None  # Python 3.10+

这里我建议团队统一Python版本,别在类型注解上折腾兼容性。

异步处理要小心

FastAPI支持异步端点,但异步不是银弹。特别是调用模型推理时:

@app.post("/predict")
async def predict(request: InferenceRequest):
    # 错误:在异步函数里调用阻塞操作
    result = heavy_model.predict(request.text)  # 这个函数是同步的!
    return {"result": result}

如果模型推理是CPU密集型同步操作,会阻塞整个事件循环。两种解决方案:

# 方案1:用线程池执行阻塞操作
from concurrent.futures import ThreadPoolExecutor
executor = ThreadPoolExecutor()

@app.post("/predict")
async def predict(request: InferenceRequest):
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(
        executor, 
        heavy_model.predict, 
        request.text
    )
    return {"result": result}

# 方案2:直接声明为同步端点
@app.post("/predict")
def predict(request: InferenceRequest):  # 去掉async
    return {"result": heavy_model.predict(request.text)}

简单经验:IO密集型用异步,CPU密集型用同步+工作进程。

依赖注入的实用技巧

FastAPI的依赖注入系统比想象中强大。比如处理API密钥验证:

from fastapi import Depends, HTTPException, Header

async def verify_token(authorization: str = Header(None)):
    if not authorization:
        raise HTTPException(status_code=403, detail="没传token")
    # 实际项目里这里要查数据库或缓存
    return {"user_id": "test_user"}

@app.post("/secure_predict")
async def secure_predict(
    request: InferenceRequest,
    user_info: dict = Depends(verify_token)  # 自动验证
):
    # 到这里时token已经验证过了
    return {"user": user_info["user_id"], "result": "predicted"}

依赖可以嵌套,可以带参数,还能复用。比如数据库连接池:

async def get_db():
    db = DatabaseSession()
    try:
        yield db  # 用yield实现请求后自动清理
    finally:
        db.close()

@app.post("/log_prediction")
async def log_prediction(
    request: InferenceRequest,
    db = Depends(get_db)
):
    db.insert_prediction(request.text)
    return {"status": "logged"}

错误处理要统一

AI服务常见的错误类型:模型加载失败、输入超出长度限制、GPU内存不足。这些都需要统一处理:

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()

class ModelUnavailableError(Exception):
    pass

@app.exception_handler(ModelUnavailableError)
async def model_unavailable_handler(request: Request, exc: ModelUnavailableError):
    return JSONResponse(
        status_code=503,
        content={"error": "模型暂时不可用", "retry_after": 30}
    )

# 业务代码里直接抛自定义异常
@app.post("/predict")
async def predict(request: InferenceRequest):
    if not model_loaded:
        raise ModelUnavailableError()
    # ...正常处理

别在每个端点里写一堆try-except,维护起来是噩梦。

部署时的配置细节

开发环境和生产环境配置不同。我习惯用环境变量:

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    model_path: str = "models/default"
    max_request_size: int = 1024 * 1024  # 1MB
    enable_docs: bool = False  # 生产环境关掉文档
    
    class Config:
        env_file = ".env"  # 从.env文件加载

settings = Settings()

app = FastAPI(
    title="模型服务",
    docs_url="/docs" if settings.enable_docs else None  # 条件启用文档
)

.env文件里写:

MODEL_PATH=/opt/models/bert
MAX_REQUEST_SIZE=5242880
ENABLE_DOCS=false

这样不同环境切换只需要改环境变量,代码不动。

个人经验建议

FastAPI的自动文档虽然方便,但生产环境一定要关掉,或者至少加权限控制。见过有团队把测试环境文档暴露在公网,接口参数全被爬走。

模型服务的内存管理要格外小心。特别是多模型加载时,容易内存泄漏。建议用@app.on_event("startup")@app.on_event("shutdown")显式管理模型生命周期。

输入验证别完全依赖Pydantic。它检查数据类型没问题,但业务逻辑验证还得自己写。比如文本长度限制、图像尺寸检查,这些写在Pydantic验证器里更合适。

最后,FastAPI的版本兼容性比较激进。Pydantic从v1到v2是不兼容升级,Starlette也经常变API。生产环境锁定版本,别追最新。

调试时多用FastAPI自带的/docs/redoc,但真正压测要用专业工具。我遇到过本地测试一切正常,上线后QPS上到50就开始报错的坑——后来发现是数据库连接池配置小了。

AI后端和传统业务后端不太一样,请求处理时间长,资源占用大。设计API时考虑支持异步任务、进度查询、结果缓存这些特性,别等到业务方提需求再返工。

Logo

腾讯云面向开发者汇聚海量精品云计算使用和开发经验,营造开放的云计算技术生态圈。

更多推荐