FastAPI简介
概述
FastAPI,GitHub,一个现代的、轻量级、快速(高性能)的Web框架,专为构建基于Python的API服务而设计。提供异步编程(基于async/await)、强类型自动数据验证(基于Pydantic)、依赖注入、类型安全等特性,使得开发高性能、可扩展的API变得简单高效。非常适合处理高并发的大模型(如GPT、Qwen、DeepSeek等)推理任务。
特点:
- 构建REST API:适合开发需要高性能和类型安全的API服务;
- 微服务架构:作为微服务的一部分,提供高效的接口;
- 自动生成文档:Swagger & ReDoc;
- 机器学习推理服务:结合大模型构建推理API;
- 实时数据流:通过WebSocket或SSE实现实时数据推送。
对比
对比项 | FastAPI | Flask |
---|---|---|
性能 | 异步支持,超快(比Flask快3~5倍) | 同步阻塞,性能相对较低 |
类型检查 | Pydantic强类型验证,减少错误 | 手动解析请求数据,易出错 |
自动文档 | 内置Swagger和ReDoc | 需手动集成Swagger |
异步支持 | 天生支持async/await,高并发任务不阻塞 | 需要手动使用线程池或协程 |
学习成本 | 比Flask略高,但上手快 | 语法简单,适合初学者 |
适用场景 | 高性能API、微服务、异步应用 | 小型Web项目、简单API |
入门
环境安装:pip install fastapi uvicorn SQLAlchemy aiomysql redis gunicorn -i https://pypi.tuna.tsinghua.edu.cn/simple
解读:
- fastapi:核心库;
- uvicorn:ASGI服务器,用于运行应用;
- SQLAlchemy:ORM库,用于数据库交互;
- aiomysql:支持异步操作的MySQL客户端;
- redis:Redis客户端;
- gunicorn:一个高效的WSGI HTTP服务器,用于生产环境中运行应用。
典型的FastAPI项目结构,通过组织不同的功能模块和代码,使得项目更易于维护和扩展:
├── api
│ └── users.py # 用户相关API
├── forms
│ └── user_form.py
├── utils
│ ├── depends.py # 依赖注入工具
│ └── response.py # HTTP响应工具类
├── main.py
└── requirements.txt
解读:
api/
:存放路由文件,每个文件负责不同的业务模块;forms/
:存放Pydantic数据模型,用于数据验证和序列化;utils/
:存放通用的工具类和函数,如响应工具和依赖注入工具;main.py
:FastAPI应用的入口,通常在此处启动应用并包含路由;requirements.txt
:存放所有项目依赖包的列表。
在main.py
中,通过FastAPI创建应用实例,并将不同的API路由引入到应用中。
from fastapi import FastAPI, Request
from api import usersapp = FastAPI()
app.include_router(users.users_router) # 引用路由
users.py
中,定义与用户相关的API路由。FastAPI提供非常灵活的方式来定义路径参数、查询参数、请求体参数,并支持详细的验证:
- 使用
APIRouter()
来定义路由; - 定义多个具体的API路由,包括路径参数、查询参数、请求体参数和依赖注入。
from typing import Optional
from fastapi import APIRouter, Depends, Query, Body, Path
from forms.user_form import UserReq
from utils.depends import default_user
from utils.response import HttpResponseusers_router = APIRouter(prefix="/api/v1", # 路径前缀tags=["users"], # 文档分组responses={404: {"description": "Not found"}}, # 自定义某个状态码的响应内容
)# 路径参数及验证
@users_router.get("/users_v1/{user_id}")
async def get_users_v1(user_id: int = Path(..., ge=1,title="The ID of the user to get")
):return HttpResponse.success(data={"user_id": user_id})# 查询参数及验证
@users_router.get("/users_v2/")
async def get_users_v2(p: str = Query(..., # ... 代表必需填min_length=1,max_length=20,regex="^@", # 以@开头title="Query string",description="Query string to search",alias="params", # url查询参数的别名deprecated=True,),q: Optional[str] = Query(None) # 不是必填,默认为None
):result = p + (q if q else '')return HttpResponse.success(data={"result": result})# 请求体参数示例
@users_router.post("/users_v3/",tags=["users"],summary="Get an user", # 简短描述description="Get an user detail", # 详细描述response_model=UserReq, # 响应数据的模型类型response_model_exclude_unset=True, # 忽略默认参数response_model_include={"name", "mobile"}, # 包含(忽略其他的)response_description="The user item", # 对响应内容的描述deprecated=True # 已废弃
)
async def get_users_v3(user: UserReq = Body(...)
):return HttpResponse.success(data={"user": user})# 依赖注入示例
@users_router.get("/users_v4/")
async def get_users_v4(user: UserReq = Depends(default_user)):return HttpResponse.success(data={"user": user})
user_form.py
中,通过Pydantic创建UserReq类,用于对请求体数据进行验证和序列化:
- 继承自BaseModel,所有字段都会自动验证;
- 使用
Field(..., gt=0)
进行校验; - Optional标记字段可选。
from typing import Optional
from pydantic import BaseModel, Fieldclass UserReq(BaseModel):name: strmobile: strage: int = Field(..., gt=0, description="The age must be greater than zero")introduction: Optional[str] = Field(None, title="The introduction of the user", max_length=300, example="A very nice man")
depends.py
中,定义一个默认的用户对象,使用FastAPI的依赖注入系统返回一个 UserReq 对象。default_user
创建默认的用户对象,返回给调用者。它被用作依赖注入,可以在路由中通过Depends()
获取。
from forms.user_form import UserReqasync def default_user() -> UserReq:user = UserReq(name='default',mobile='123',age=18,introduction='default')return user
在response.py
中定义HttpResponse工具类,简化响应生成过程:
- 提供多种常见HTTP状态类型的响应方法:成功、未授权、参数错误和服务器错误;
response()
根据给定参数构建标准的响应格式。
from fastapi import status
from fastapi.encoders import jsonable_encoder
from fastapi.responses import JSONResponse, Response
from typing import Any, Dict, Optionalclass HttpResponse:ok = 200unautherror = 401paramserror = 400servererror = 500@classmethoddef response(cls, code: int, msg: str, data: Optional[Any], state: Optional[Any]) -> Response:result = {"code": code,"msg": msg,"data": data,"state": state}return JSONResponse(status_code=status.HTTP_200_OK, content=jsonable_encoder(result))@classmethoddef success(cls,msg: str = '操作成功',data: Optional[Dict] = None,state: Optional[Any] = None,) -> Response:return cls.response(cls.ok, msg, data, state)@classmethoddef unauth_error(cls,msg: str = '登录信息已过期',data: Optional[Dict] = None,state: Optional[Any] = None,) -> Response:return cls.response(cls.unautherror, msg, data, state)@classmethoddef params_error(cls,msg: str = '客户端参数错误',data: Optional[Dict] = None,state: Optional[Any] = None,) -> Response:return cls.response(cls.paramserror, msg, data, state)@classmethoddef server_error(cls,msg: str = '服务器内部错误',data: Optional[Dict] = None,state: Optional[Any] = None,) -> Response:return cls.response(cls.servererror, msg, data, state)
API测试
启动服务后,访问http://127.0.0.1:5000/docs
来查看自动生成的在线Swagger文档,访问http://127.0.0.1:5000/redoc
查看ReDoc文档。
使用curl或浏览器文档来测试API接口:
- 支持直接在文档中测试API,并可复制为curl命令;
- 使用curl命令发送请求,测试路径参数验证、查询参数验证和请求体参数。
# 启动服务
uvicorn main:app --reload --port 5000
# 路径参数验证
curl -X GET 127.0.0.1:5000/api/v1/users_v1/0
# 查询参数及验证
curl -X GET 127.0.0.1:5000/api/v1/users_v2/?params=@hello
# 请求体参数示例
curl -H "Content-Type: application/json" -X POST -d '{"name":"myname", "mobile":"123", "age": 18}' 127.0.0.1:5000/api/v1/users_v3/
# 依赖注入示例
curl -X GET 127.0.0.1:5000/api/v1/users_v4/
后台任务
FastAPI提供BackgroundTasks,适用于不影响主请求的任务,如发送邮件、日志记录:
from fastapi import FastAPI, BackgroundTasksapp = FastAPI()def write_log(msg: str):with open("log.txt", "a") as f:f.write(msg + "\n")@app.get("/task")
def start_task(background_tasks: BackgroundTasks):background_tasks.add_task(write_log, "异步任务执行!")return {"message": "任务已提交"}
其他
应用
使用FastAPI的框架或平台非常多,包括:
- Xinference推理框架
推理服务
深度学习模型从训练阶段进入生产环境的关键桥梁,其核心价值在于将训练好的模型转化为实际可用的应用能力。
- 训练阶段:模型在离线环境中,依赖大量计算资源进行参数优化,关注模型性能指标(如准确率、损失值)。
- 推理阶段:模型需在生产环境中实时响应用户请求,关注低延迟、高吞吐量、资源利用率,且需适应动态变化的输入数据。
选择FastAPI构建推理服务的核心原因:
- 极致性能:异步 + 高并发
- 异步非阻塞 IO:FastAPI基于Starlette和Pydantic,原生支持异步编程,能高效处理大量并发请求,避免线程阻塞。比如:推理服务需同时响应多个用户请求(如智能客服),异步特性可显著降低延迟;
- 性能对比:在基准测试中,FastAPI 的性能接近(甚至超过)Node.js 和 Go,远超传统同步框架,如Flask、Django。
- 开发效率:类型安全 + 自动文档
- 类型安全:FastAPI强制使用Python类型注解(Type Hints),减少因参数错误导致的运行时崩溃;
- 自动生成交互式文档:包括Swagger和ReDoc文档,便于前后端联调和测试。
- 生态兼容:深度学习框架无缝集成
- 与PyTorch/TensorFlow无缝对接:可直接调用PyTorch、TensorFlow等深度学习模型,适合推理场景。比如:使用Transformers加载模型,通过FastAPI提供推理接口;
- 支持GPU加速:可与CUDA无缝协作,充分利用GPU提升推理速度。
将本地部署模型封装为REST API,提供推理服务:
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from transformers import pipeline
# 初始化FastAPI应用
app = FastAPI(title="大模型推理服务", description="基于FastAPI和HuggingFace Transformers")
# 使用transformers加载预训练模型(例如文本生成模型),并将其封装为可复用的类或函数
model = pipeline("text-generation", model="DeepSeek-R1")
# 定义请求体模型
class TextInput(BaseModel):prompt: strmax_length: int = 50num_return_sequences: int = 1
# 定义推理接口
@app.post("/predict")
async def predict(input: TextInput):try:# 调用模型进行推理result = model(input.prompt, max_length=input.max_length, num_return_sequences=input.num_return_sequences)return {"output": result[0]['generated_text']}except Exception as e:raise HTTPException(status_code=500, detail=str(e))
FastAPI通过封装预训练大模型为可复用类,结合异步接口处理用户请求,利用 uvicorn启动高性能 ASGI 服务,实现快速构建大模型推理服务:uvicorn my_app:app --host 0.0.0.0 --port 8000 --reload
。--reload
启用自动重载(开发环境使用)。
uvicorn和gunicorn
上面简单介绍过:
- uvicorn:ASGI服务器;
- gunicorn:WSGI服务器。
Uvicorn,一个轻量级的 ASGI (Asynchronous Server Gateway Interface) 服务器。特点:
- 专为Python异步框架设计(如FastAPI,Starlette,Quart等);
- 基于uvloop和httptools构建,性能优异;
- 开发环境快速启动;
- 轻量级API服务;
- 原生支持HTTP/1.1和WebSockets;
- 支持HTTP/2(需要额外配置)
Gunicorn,一个成熟的 WSGI (Web Server Gateway Interface) 服务器,特点:
- 传统的Web服务器;
- 支持多种worker类型(同步和异步);
- 提供进程管理、负载均衡等功能;
- 稳定性高,适合生产环境
特性 | Uvicorn | Gunicorn |
---|---|---|
协议 | ASGI | WSGI,可通过worker支持ASGI |
性能 | 更高,尤其异步场景 | 良好,取决于worker类型 |
并发模型 | 原生异步 | 多进程+线程,可配异步worker |
适用框架 | FastAPI,Starlette等异步框架 | Django,Flask等传统框架 |
配置 | 较简单 | 较复杂(更多可调参数) |
生产环境成熟度 | 较新但快速增长 | 非常成熟稳定 |
HTTP2 | 支持 | 不支持 |
Uvicorn + Gunicorn组合
- 生产环境的最佳实践
- Gunicorn作为进程管理器,Uvicorn作为worker