FastAPI 服务器开发

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

---

一、服务器基础概念

1. 什么是服务器

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

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

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

服务器开发的重点：

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

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

框架	说明
Django	重量级全栈框架，适合大型项目
Flask	轻量级微框架，适合中小项目
FastAPI 🔥	现代高性能框架，异步原生，自动生成 API 文档——本项目选型

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

3. 接口术语

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

---

二、FastAPI 环境搭建

1. 安装

激活虚拟环境后执行：

pip install fastapi "uvicorn[standard]" -i https://repo.huaweicloud.com/repository/pypi/simple/

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

2. 启动方式

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

启动参数	说明
main:app	main = 文件名（main.py），app = FastAPI 实例变量名
--reload	热重载模式，代码修改后自动重启（开发时启用）
--host	监听地址，0.0.0.0 允许所有 IP 访问
--port	监听端口，默认 8000

方式二： 通过命令行启动

if __name__ == "__main__":
    import uvicorn as uv

    uv.run(
        app="main:app",
        host="localhost",
        port=8000,
        reload=False,
    )

值	效果
reload=True	开发模式 — 代码文件一保存修改，服务器自动重启
reload=False	生产模式 — 改完代码必须手动停服务再重启才生效

---

三、接口定义

1. 接口定义格式

接口定义分为两种情况：

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

2. 请求方式

请求方式	使用场景	参数传递方式
GET	只查数据、下拉列表、详情页，浏览器地址栏直接访问，用于获取数据。它应该是安全的（只读）且幂等的（多次请求结果一致，不会改变服务器状态）。 没有请求体（Body）。	参数拼接在请求地址后面（表单格式 k=v）
POST	登录注册、上传文件、新增/修改/删除数据库记录，用于创建或提交数据。它既不安全也不幂等（多次提交可能会创建多个资源，比如重复下单）。 拥有请求体（Body）。	参数包装在请求体里面（JSON 格式）
PUT	更新操作	参数在请求体中（JSON 格式）
DELETE	删除操作	参数通常在 URL 路径中

3. 返回值

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

4. 命名规范

项目	规范
接口函数名	蛇形命名（snake_case）：say_hello
请求路径	小驼峰命名：/sayHello
形参	小驼峰命名：userName

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

"""
定义一个 GET 请求的接口：假设需要返回 msg: hello 给客户端
    1、先直接定义一个函数
    2、通过装饰器配置访问路径和请求方式
    3、设置函数内容：逻辑处理、返回值等
"""

from fastapi import FastAPI

app = FastAPI()


@app.get("/say")
def say():
    print("say 接口函数执行了")
    # 设置返回值 --- 字典，自动转 JSON
    return {
        "msg": "hello"
    }


@app.post("/say2")
def say2():
    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. 各层职责

层	职责	核心任务
controller	接口层	定义接口、接收客户端参数、调用 service、返回响应
service	业务层	实现具体业务逻辑处理，调用 dao 操作数据
dao	数据层	只负责数据库的增删改查，不做逻辑处理
entity	实体层	定义数据模型类，用于接收 JSON 参数和数据验证
utils	工具层	抽取冗余代码形成工具函数

---

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

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

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

1. 子路由定义（controller 层）

# users/controller/TestController_1.py
from fastapi import APIRouter

# 创建子路由对象
users_router = APIRouter()

# 定义子路由接口 --- 配置的路径并非最终接口访问路径
@users_router.get("/sayHello")
def say_hello():
    return {
        "msg": "hello"
    }

关键点：

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

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

# main.py
from fastapi import FastAPI
# 导入子路由
from users.controller.TestController_1 import users_router

app = FastAPI()

# 注册子路由 --- 访问路径为：/users/sayHello
app.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")
def get_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}")
def get_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"}'
"""
from pydantic import BaseModel, Field


# 定义接收数据的类 --- 直接继承 BaseModel
class TestClass(BaseModel):
    # Field(..., title="用户名") 表示这是一个必填字段
    username: str = Field(..., title="用户名")
    password: str = Field(..., title="密码")


@users_router.post("/postParams")
def post_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 接收
"""

from fastapi import UploadFile, File, Form


@users_router.post("/postFile")
def post_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(二进制形式)
    with open(save_path, "wb") as f:
        # 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 获取原始文件名

---

四种传参方式对比

方式	请求方式	数据格式	接收方式	适用场景
方式一	GET	表单（k=v）	直接形参接收	查询列表、简单参数传递
方式二	GET	URL 路径参数	路径占位符 + 形参	查询/删除单个资源
方式三	POST	JSON	Pydantic 数据类	新增/登录/复杂参数
方式四	POST	multipart/form-data	UploadFile + Form	文件上传

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

---

七、流式输出 — StreamingResponse

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

from starlette.responses import StreamingResponse
import time
import json

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

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

@users_router.get("/testStream")
def test_stream():
    """基础流式输出：假设模型返回 0-9 十个数字"""

    def generator():
        for i in range(9):
            yield f"{i}"
            time.sleep(0.1)  # 模拟模型逐 token 生成

    return StreamingResponse(
        content=generator(),  # 迭代器对象
        media_type="text/event-stream",  # 媒体类型
    )

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

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

@users_router.get("/testStreamSSE")
def test_stream_sse():
    """SSE 流式输出：标准 data: 格式，便于客户端处理"""

    result = "你好！👋 很高兴见到你。有什么我可以帮你的吗？"

    def generator():
        for i in result:
            # 包装为 SSE 标准格式，内容转为 JSON 便于客户端解析
            yield f"data: {json.dumps({'content': i})}\n\n"
            time.sleep(0.1)  # 模拟耗时
        # 发送结束标记
        yield f"data: {json.dumps({'content': '[DONE]'})}\n\n"

    return StreamingResponse(
        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.py
import os
from langchain_openai import ChatOpenAI


def LLM_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 产出
    for chunk in chatLLM.stream(messages):
        yield chunk.content

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

# users/controller/TestController_1.py
import json
import time
from fastapi import APIRouter
from starlette.responses import StreamingResponse
from ai.TestLLM import LLM_Model

users_router = APIRouter()


@users_router.get("/StreamSSE")
def test_stream_sse(question: str = "你好"):
    """
    用户输入问题 → LLM 生成回复 → 流式 SSE 输出给客户端
    
    客户端访问：/users/StreamSSE?question=你好
    """
    print(f"接收到的数据为：question={question}")

    # 调用 LLM 获取流式生成器
    result = LLM_Model(question=question)

    # 生成器 --- 包装为 SSE 格式输出
    def generator():
        for i in result:
            yield f"data: {json.dumps({'content': i})}\n\n"
            time.sleep(0.1)  # 模拟网络传输延迟
        # 数据结束标记
        yield f"data: {json.dumps({'content': '[DONE]'})}\n\n"

    return StreamingResponse(
        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

接口定义规范速查：

请求方式	参数传递	接收方式	示例
GET	URL 查询参数（k=v）	直接形参	/getParams?username=admin
GET	URL 路径参数	路径占位符	/getParamsTwo/{id}
POST	JSON 请求体	Pydantic 数据类	{"name": "张三"}
POST	文件上传	UploadFile + Form	multipart/form-data
GET/POST	流式输出	StreamingResponse + 生成器	SSE 格式逐 token 推送

核心要点：

• 接口 = 普通函数 + 请求路径配置，不能直接调用，必须通过 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 应用服务。