025、AI应用后端开发:FastAPI框架与RESTful API设计
现在Python后端框架选择不少,Flask轻量但生态散,Django重但自带全家桶。FastAPI站在中间那个微妙的位置——它不像Flask那样需要自己拼装各种插件,又比Django更适配现代异步编程。最关键的是,它天生为AI应用设计:自动生成OpenAPI文档、内置数据验证、原生支持async/await。你部署个模型服务,总不能每次改接口都手动更新API文档吧?app = FastAPI(t
昨天深夜调试一个模型服务接口,客户端传过来的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时考虑支持异步任务、进度查询、结果缓存这些特性,别等到业务方提需求再返工。
更多推荐
所有评论(0)