AI后端工程化：FastAPI + Pydantic + JWT 鉴权实战，从零构建 AI 接口服务

作者：飞哥（一个喜欢讲故事的全栈开发者，擅长把复杂的代码翻译成“人话”）
难度：⭐⭐⭐
关键词：FastAPI, Async, Pydantic, JWT, 鉴权

大家好，我是飞哥！👋

在之前的课程里，我们写了很多 Python 脚本。脚本就像是在“自家厨房做饭”🍳，想吃什么随手拿，做完自己吃，很随意。

但是，我们要构建的 AI 应用（比如一个公司内部的知识库助手），是要给成百上千人用的。这时候，脚本就不够用了。我们需要开一家“正规餐厅”🏪（Web 服务），这里有：

服务员（API 接口）：负责接待客人。
菜单（文档）：告诉客人有什么菜。
保安（鉴权）：防止有人吃霸王餐。

FastAPI就是 Python 界开“餐厅”的神器。今天飞哥就带大家从零开始，把我们的 RAG 脚本包装成一个高并发、安全的 API 服务！🚀

📦 准备工作

在开始之前，我们需要安装几个“装修材料”：

pipinstallfastapi"uvicorn[standard]"pyjwt python-multipart

fastapi: 核心框架。
uvicorn: 高性能的服务器（ASGI）。
pyjwt: 生成和校验 Token 的工具。
python-multipart: 处理登录表单数据。

第一步：理解“异步” (Async/Await) ⚡️

1. 为什么要异步？（Anchor & Analogy）

大家去餐厅吃饭，如果你点完菜，服务员就一直站在桌边等你吃完收盘子，才去招呼下一桌，这餐厅肯定得倒闭。这叫同步 (Sync)，效率极低。😫

聪明的服务员是这样的：帮你点完菜，把单子给厨房（耗时操作），然后立刻去招呼下一桌客人。等厨房菜好了，他再回来给你上菜。这就叫异步 (Async)。🏃

本质：在等待 IO（数据库查询、大模型生成）的时候，不傻等，释放 CPU 去处理别的请求。

2. 代码实战

在 FastAPI 里，只要加个async关键字，你的接口就有了“超级服务员”的能力。

importasynciofromfastapiimportFastAPIimportuvicorn app=FastAPI()# 模拟一个耗时的 AI 生成任务asyncdeffake_ai_generation():awaitasyncio.sleep(2)# 模拟思考 2 秒 💤return"AI 思考完毕！"@app.get("/chat")asyncdefchat():print("开始接待客人...")result=awaitfake_ai_generation()# 关键：await 让出控制权print("客人接待完毕！")return{"message":result}if__name__=="__main__":uvicorn.run(app,host="127.0.0.1",port=8000)

一句话记住它：async是服务员的工牌，await是他在厨房窗口等待的那个动作。

第二步：严格的“保安” (Pydantic) 🛡️

1. 为什么要数据校验？

如果你的接口需要接收用户发来的 JSON 数据，比如：

{"query":"你好","temperature":0.7}

万一用户瞎发，把temperature发成了字符串 “hot”，或者忘了发query，你的程序可能直接崩掉。💥

我们需要一个“保安”，在门口检查数据格式。如果不合格，直接拦回去，告诉他“格式不对”。这个保安就是Pydantic。

2. 代码实战

这是一个完整的可运行示例，保存为pydantic_demo.py试试：

fromfastapiimportFastAPIfrompydanticimportBaseModelimportuvicorn app=FastAPI()# 定义数据的“入场券”格式classChatRequest(BaseModel):query:strtemperature:float=0.7# 默认值，可以是小数history:list[str]=[]# 必须是字符串列表@app.post("/api/v1/chat")asyncdefcreate_chat(request:ChatRequest):# 只要进到这里，request.query 一定是 string，temperature 一定是 float# 飞哥再也不用写 if type(x) != str 这种代码了！😂return{"reply":f"你问了：{request.query}, 温度是：{request.temperature}"}if__name__=="__main__":uvicorn.run(app,host="127.0.0.1",port=8000)

运行后，打开 Swagger UI (http://127.0.0.1:8000/docs)，找到 POST 接口，点击 Try it out，你可以故意传个错的数据（比如把 temperature 改成 “hot”），看看保安会不会拦你。🚫

第三步：只有 VIP 能进 (JWT 鉴权) 🎫

1. 什么是 JWT？

现在我们要给 API 加锁，只有登录用户才能用。🔐

HTTP 是“健忘”的，服务器记不住谁是谁。所以，用户登录成功后，我们发给他一张“电子手环” (JWT Token)。
以后他每次来请求，只要亮一下手环，我们就知道他是谁，不用反复查身份证（数据库）。

JWT (JSON Web Token)的本质就是一段加密的字符串，里面藏着用户的 ID 和过期时间。

2. 代码实战：完整可运行代码 (main.py)

请创建一个main.py文件，复制以下所有代码：

fromfastapiimportFastAPI,Depends,HTTPExceptionfromfastapi.securityimportOAuth2PasswordBearer,OAuth2PasswordRequestFormfrompydanticimportBaseModelimportjwtfromdatetimeimportdatetime,timedelta app=FastAPI()# --- 配置区 ---SECRET_KEY="feige_secret_key"# ⚠️ 真实的密钥要藏好ALGORITHM="HS256"# 定义鉴权方案：客户端要在 Header 里带 Authorization: Bearer <token>oauth2_scheme=OAuth2PasswordBearer(tokenUrl="token")# --- 1. 登录接口：发手环 ---@app.post("/token")asyncdeflogin(form_data:OAuth2PasswordRequestForm=Depends()):# 模拟查数据库ifform_data.username!="feige"orform_data.password!="123456":raiseHTTPException(status_code=400,detail="用户名或密码错误")# 生成 JWTtoken_data={"sub":form_data.username,"exp":datetime.utcnow()+timedelta(minutes=30)# 30分钟过期}token=jwt.encode(token_data,SECRET_KEY,algorithm=ALGORITHM)return{"access_token":token,"token_type":"bearer"}# --- 2. 验证函数：查手环 ---asyncdefget_current_user(token:str=Depends(oauth2_scheme)):try:# 解码 Tokenpayload=jwt.decode(token,SECRET_KEY,algorithms=[ALGORITHM])username=payload.get("sub")ifusernameisNone:raiseHTTPException(status_code=401)returnusernameexceptjwt.PyJWTError:# 捕获 pyjwt 的错误raiseHTTPException(status_code=401,detail="手环无效或过期")# --- 3. 受保护的接口 ---@app.get("/users/me")asyncdefread_users_me(username:str=Depends(get_current_user)):return{"username":username,"status":"VIP 贵宾"}# --- 启动方式 ---# 方式 1: 在终端运行 uvicorn main:app --reload# 方式 2: 直接运行此脚本 python main.pyif__name__=="__main__":importuvicorn uvicorn.run(app,host="127.0.0.1",port=8000)