FastAPI 服务器开发

之前的课程学习了 RAG 流程和向量数据库。今天进入Web 服务器开发领域——学习 FastAPI 框架，掌握接口定义、参数接收、子路由嵌套、流式输出等核心技能，将 LLM 能力封装为可外部调用的 API 接口。

一、服务器基础概念

1. 什么是服务器

服务器在我们的生活中无处不在：

• 如果需要下载客户端再使用 →C/S 模式（客户端/服务器模式）
• 如果不需要下载客户端就可以使用 →B/S 模式（浏览器/服务器模式）

我们做的Web 应用开发基于B/S 模式实现——通过构造网页来实现项目中的内容和功能展示。

服务器开发的重点：

1. 只有通过服务器我们才可以操作数据库中的内容
2. 注意服务器开发时的分包思想（结构化组织代码）
3. 注意服务器开发中的术语——一个内容可以有不同称呼（如接口、端点、路由）

2. Python 服务器开发框架选型

FastAPI 官网：https://fastapi.tiangolo.com/zh/

3. 接口术语

• 接口：服务器中提供给客户端实现某个功能访问的函数。和普通 Python 函数定义没有区别，只是多了一个请求路径配置（告诉客户端通过什么请求地址才能访问到这个函数）
• 接口函数不能像普通函数一样直接调用，需要通过请求地址去访问
• 常用接口测试工具：Apipost、Postman
• FastAPI 内置集成了Swagger UI，直接通过xxx/docs地址即可在浏览器中测试所有接口

二、FastAPI 环境搭建

1. 安装

激活虚拟环境后执行：

pipinstallfastapi"uvicorn[standard]"-ihttps://repo.huaweicloud.com/repository/pypi/simple/

项目类型选择上，非普通 Python 项目，需创建为 FastAPI 项目类型。

2. 启动方式

方式一：通过 IDE，调整内置参数后点击运行按钮直接启动

方式二：通过命令行启动

if __name__ == "__main__": import uvicorn as uv uv.run( app="main:app", host="localhost", port=8000, reload=False, )

三、接口定义

1. 接口定义格式

接口定义分为两种情况：

• 第一种：直接在main.py中定义接口（项目不使用，仅用于演示）
• 第二种：在其他文件中定义接口（项目中使用的方案）

2. 请求方式

3. 返回值

• 接口返回值统一以 JSON 格式返回
• 不需要手动调用json.dumps()转换，把返回值设为字典即可自动完成 JSON 序列化

4. 命名规范

5. main.py 中直接定义接口（演示）

""" 定义一个 GET 请求的接口：假设需要返回 msg: hello 给客户端 1、先直接定义一个函数 2、通过装饰器配置访问路径和请求方式 3、设置函数内容：逻辑处理、返回值等 """fromfastapiimportFastAPI app=FastAPI()@app.get("/say")defsay():print("say 接口函数执行了")# 设置返回值 --- 字典，自动转 JSONreturn{"msg":"hello"}@app.post("/say2")defsay2():print("say2 函数执行了")return{"msg":"hello2"}

关键点：

• @app.get()/@app.post()通过装饰器将普通函数变为接口函数
• app是 FastAPI 实例，get/post设置请求方式
• 请求路径（如/say）需要拼接在服务器地址（http://localhost:8000）后面
• 返回值直接写字典即可，FastAPI 自动转为 JSON

四、MVC 分包思想【重要】

按照不同的项目模块和功能代码进行分包处理，降低代码的耦合度，使得代码分层清晰、便于测试和维护。

1. 标准分包结构

项目根目录 ├── users（用户模块） │ ├── controller/ --- 定义接口，接收和响应客户端请求 │ ├── service/ --- 业务逻辑处理，供 controller 调用 │ ├── dao/ --- 数据库操作层，只操作数据库不做逻辑处理 │ ├── utils/ --- 当前模块的工具函数 │ └── entity/ --- 实体类（数据验证、接收 JSON） ├── chat（对话模块） │ ├── controller/ │ ├── service/ │ ├── dao/ │ ├── utils/ │ └── entity/ ├── common（公共模块） --- 多个模块共用的工具代码 └── ai（AI 模块） --- 大模型相关封装

核心原则：

• 同一模块下，不同业务创建不同文件来实现，不需要创建类，直接定义函数接口
• 比如 chat 模块既有聊天业务、也有加载历史对话记录业务 → 创建两套文件分别处理

2. 各层职责

五、父子路由嵌套【重点】

因为采用分包分模块思想，接口不在main.py中定义，而在各模块的controller包下。但 controller 包中没有 FastAPI 对象，无法直接定义接口。

解决方案：将 controller 中的接口定义为子路由，然后在main.py中注册子路由。

1. 子路由定义（controller 层）

# users/controller/TestController_1.pyfromfastapiimportAPIRouter# 创建子路由对象users_router=APIRouter()# 定义子路由接口 --- 配置的路径并非最终接口访问路径@users_router.get("/sayHello")defsay_hello():return{"msg":"hello"}

关键点：

• APIRouter()创建子路由对象，代替 FastAPI 实例
• 装饰器使用@users_router.get()而非@app.get()
• 子路由中配置的路径不是最终路径，需要通过main.py注册后才完整

2. 子路由注册（main.py）

# main.pyfromfastapiimportFastAPI# 导入子路由fromusers.controller.TestController_1importusers_router app=FastAPI()# 注册子路由 --- 访问路径为：/users/sayHelloapp.include_router(users_router,prefix="/users"， tags=["users"])

关键点：

• app.include_router(子路由对象, prefix="/模块名")注册子路由
• prefix="/users"设置路由前缀，最终接口路径 =prefix + 子路由中配置的路径
• 子路由注册后，在 controller 中定义的接口才能被外部访问

六、接口接收客户端请求参数【核心】

参数传递方式分为三种，取决于请求方式和数据格式：

方式一：GET 请求 + key=value 表单格式

参数通过k=v格式拼接在请求地址后面，接口直接用形参接收（形参名必须和 key 一致）。

""" Way 1: GET + key=value URL: localhost:8001/users/getParams?username=admin&password=111 形参名必须和 key 相同，否则接收不到数据 """@users_router.get("/getParams")defget_params(username:str=None,password:str=None):print(f"接收到的数据为：username={username}, password={password}")return{"code":200,"msg":"success","data":{"username":username,"password":password}}

关键点：

• 直接用函数形参接收，形参名必须与 URL 中的 key 一致
• 设置默认值= None使参数可选，避免客户端不传时报错
• 客户端访问示例：/getParams?username=admin&password=111

方式二：GET 请求 + 参数在请求路径中

参数直接写在请求路径里（没有 key），需要在路径中定义{变量}占位符。

""" Way 2: GET + URL 路径参数 URL: localhost:8001/users/getParamsTwo/admin/111 顺序匹配路径中的占位符 常用于查询、删除操作 """@users_router.get("/getParamsTwo/{username}/{password}")defget_params_two(username:str,password:str):print(f"接收到数据为：username={username}, password={password}")return{"code":200,"msg":"success","data":{"username":username,"password":password}}

关键点：

• 路径中使用{变量名}占位，客户端按顺序传入值
• ⚠️ 形参名必须和路径中的占位符名字一致，否则返回 422 错误
• 有顺序问题：/getParamsTwo/admin/111按路径顺序匹配 username=admin, password=111

方式三：POST 请求 + JSON 格式数据

参数在请求体中传输（Content-Type: application/json），需要定义一个数据类来接收。

""" Way 3: POST + JSON 需要定义类来接收，类属性名必须和 JSON 中的 key 一致 客户端： curl -X POST "localhost:8001/users/postParams" \ -H "Content-Type: application/json" \ -d '{"username":"admin","password":"111"}' """frompydanticimportBaseModel,Field# 定义接收数据的类 --- 直接继承 BaseModelclassTestClass(BaseModel):# Field(..., title="用户名") 表示这是一个必填字段username:str=Field(...,title="用户名")password:str=Field(...,title="密码")@users_router.post("/postParams")defpost_params(testClass:TestClass):print(f"接收到的数据为：{testClass}")print(testClass.username,testClass.password)return{"code":200,"msg":"success","data":testClass}

关键点：

• 继承BaseModel（Pydantic）定义数据类，属性名必须和 JSON 的 key 一致
• Field(..., title="用户名")中的...表示该字段为必填；改为默认值则为可选
• 接口形参直接用类类型接收，FastAPI 自动解析 JSON 并验证数据
• 访问示例：POST /postParams，Body 为{"username":"admin","password":"111"}

方式四：POST 请求 + 文件上传

文件类型的参数必须用 POST 请求，使用UploadFile类型接收文件，其他额外参数用Form接收。

""" Way 4: POST + file file 类型数据必须用 POST 请求 文件用 UploadFile 接收，额外参数用 Form 接收 """fromfastapiimportUploadFile,File,Form@users_router.post("/postFile")defpost_file(file:UploadFile=File(...),username:str=Form(...)):print(f"接收到的数据为：file={file}, \n username={username}")# 重新定义文件名字 --- 时间戳唯一标识文件filename=str(int(time.time()))+"."+file.filename.split(".")[-1]# 文件存储地址 + 文件名字save_path=r"D\stu_fastapi\static\upload\\"+filename# 存储文件，wb是w(写入)b(二进制形式)withopen(save_path,"wb")asf:# file.file.read() 读取文件内容f.write(file.file.read())return{"code":200,"msg":"success","data":""}

关键点：

• UploadFile：FastAPI 提供的文件类型，自动处理上传文件
• File(...)表示这是一个文件类型的必填参数
• Form(...)接收文件外的普通表单字段
• 文件名用时间戳重命名防止冲突：str(int(time.time())) + "." + 扩展名
• file.file.read()读取上传文件的内容，file.filename获取原始文件名

四种传参方式对比

核心原则：无论选择什么方式传递数据给服务器，一定要满足key 对得上——客户端和服务器通过 key:value 交互数据，只能通过 key 找 value。

七、流式输出 — StreamingResponse

在 FastAPI 中通过StreamingResponse实现流式输出，核心是返回一个生成器（迭代器）对象。

fromstarlette.responsesimportStreamingResponseimporttimeimportjson

方案一：基础流式输出（fetch 请求）

客户端使用 fetch 请求接收，服务器直接返回结果，服务端代码简单，客户端代码较难写。

@users_router.get("/testStream")deftest_stream():"""基础流式输出：假设模型返回 0-9 十个数字"""defgenerator():foriinrange(9):yieldf"{i}"time.sleep(0.1)# 模拟模型逐 token 生成returnStreamingResponse(content=generator(),# 迭代器对象media_type="text/event-stream",# 媒体类型)

方案二：SSE 流式输出（标准方案）

客户端使用SSE（Server-Sent Events）请求，服务器必须将数据包装成data: 内容\n\n格式，推荐方案。

@users_router.get("/testStreamSSE")deftest_stream_sse():"""SSE 流式输出：标准 data: 格式，便于客户端处理"""result="你好！👋 很高兴见到你。有什么我可以帮你的吗？"defgenerator():foriinresult:# 包装为 SSE 标准格式，内容转为 JSON 便于客户端解析yieldf"data:{json.dumps({'content':i})}\n\n"time.sleep(0.1)# 模拟耗时# 发送结束标记yieldf"data:{json.dumps({'content':'[DONE]'})}\n\n"returnStreamingResponse(content=generator(),# 迭代器对象media_type="text/event-stream",# SSE 媒体类型)

关键点：

• StreamingResponse的content参数接收一个生成器/迭代器对象，函数中使用yield逐次返回数据
• media_type="text/event-stream"指定 SSE 媒体类型，告知客户端以流式事件接收
• 方案二（SSE）数据必须包装成data: 内容\n\n字符串格式，否则客户端报错
• 通常将数据转为 JSON 格式返回，便于客户端处理
• 需要告诉客户端流式输出何时结束，发送一个约定的结束标识符（如[DONE]）

八、综合实战：LLM 流式回复接口

将前面所学知识点串联——结合 LLM 模型调用，实现一个完整的流式对话 API。

1. 封装 LLM 加载工具（ai/TestLLM.py）

# ai/TestLLM.pyimportosfromlangchain_openaiimportChatOpenAIdefLLM_Model(question:str):"""封装 LLM 加载和调用，返回流式生成器"""chatLLM=ChatOpenAI(api_key=os.getenv("DASHSCOPE_API_KEY"),base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",model="qwen3.7-max-preview",streaming=True,)messages=[{"role":"user","content":question}]# stream() 返回生成器，逐 token 产出forchunkinchatLLM.stream(messages):yieldchunk.content

2. 实现流式对话接口（TestController_1.py）

# users/controller/TestController_1.pyimportjsonimporttimefromfastapiimportAPIRouterfromstarlette.responsesimportStreamingResponsefromai.TestLLMimportLLM_Model users_router=APIRouter()@users_router.get("/StreamSSE")deftest_stream_sse(question:str="你好"):""" 用户输入问题 → LLM 生成回复 → 流式 SSE 输出给客户端 客户端访问：/users/StreamSSE?question=你好 """print(f"接收到的数据为：question={question}")# 调用 LLM 获取流式生成器result=LLM_Model(question=question)# 生成器 --- 包装为 SSE 格式输出defgenerator():foriinresult:yieldf"data:{json.dumps({'content':i})}\n\n"time.sleep(0.1)# 模拟网络传输延迟# 数据结束标记yieldf"data:{json.dumps({'content':'[DONE]'})}\n\n"returnStreamingResponse(content=generator(),media_type="text/event-stream",)

关键点：

• LLM_Model()返回生成器，通过yield逐 token 产出的回复内容
• 接口使用 GET 请求 + k=v 参数（方式一）接收用户问题
• 将 LLM 的流式输出包装为 SSE 标准格式，逐 token 推送给客户端
• 客户端接收完所有数据后通过[DONE]标记判断流是否结束

九、完整开发流程总结

FastAPI 接口开发完整流程：

① 环境搭建 pip install fastapi uvicorn[standard] ② 创建项目、分包 users/ controller/ ← 定义接口 service/ ← 业务逻辑 dao/ ← 数据库操作 entity/ ← 数据模型 ③ 在 controller 中定义子路由 router = APIRouter() @router.get("/path") → 接口函数 ④ 在 main.py 注册子路由 app.include_router(router, prefix="/users") ⑤ 启动服务器 uvicorn main:app --reload --host 0.0.0.0 --port 8001 ⑥ 访问 Swagger UI 测试 http://localhost:8001/docs

接口定义规范速查：

核心要点：

• 接口 = 普通函数 + 请求路径配置，不能直接调用，必须通过 HTTP 请求访问
• MVC 分包降低耦合度，controller → service → dao 三层职责分明
• 子路由是项目开发的标配方案，APIRouter()+app.include_router()
• 客户端和服务器通过key:value交互，形参名必须和 key 一致
• 接口返回值统一字典格式，FastAPI 自动转 JSON
• 流式输出使用StreamingResponse+ 生成器（yield），SSE 格式需要data: 内容\n\n包装
• Swagger UI（/docs）是 FastAPI 最强大的特性之一，无需第三方测试工具

FastAPI常见错误码

1. 查路径对不对？→404（路径错了） /405（路径对了但 Method 错了）。
2. 查请求体格式错没错？→ JSON 结构坏了给400，字段类型错了给422。
3. 查登录没？→ 没 Token 给401，有 Token 但没权限给403。

最终思考：从"RAG 知识库搭建"到"FastAPI 服务器开发"的全栈链路学习，也学会了将 LLM 对话功能封装为 API 接口，通过浏览器或客户端调用，实现完整的 AI 应用服务。