← 代码课堂

第 08 课:FastAPI 入门 —— 从零到接口

难度:★★☆(需要会 Python 函数)
教材:worry_debate_game/app.py(230 行)
预计时间:讲解 60 分钟 + 练习 40 分钟
学完你能:用 FastAPI 写出接口、看懂自动文档,并理解框架比手写服务器省了哪些活


主线实验:保持卡片规则不变,只替换 HTTP 入口

前置:02 的方法/路径/JSON、04 的存储函数、Python 的函数和类基础。先完成一个不调用模型的接口,再对照原来的辩论应用;API Key、LangChain 和 SSE 都不属于本课首次成功运行的必要条件。

下载包 assets/主线实验/requirements.txt 固定了本轮验证的 FastAPI/Pydantic/Uvicorn 版本。按该目录 README 建虚拟环境安装,然后运行:

.venv/Scripts/python -m uvicorn api:app --host 127.0.0.1 --port 8031

macOS/Linux 改用 .venv/bin/python。api:app 表示从 api.py 模块取出名为 app 的应用;终端要位于主线实验目录。打开 http://127.0.0.1:8031/docs。

先看一条最短调用链

POST /api/cards
  → Pydantic 按 CardInput 解析和校验
  → add_card 接收 body 对象
  → 复用 card_store.create_card
  → 返回字典,框架序列化,状态 201

与标准库版本对照:你省掉了读取 Content-Length、解码 JSON、手工写响应头的代码,但标题规则、数据库事务仍由业务代码承担。框架没有替你决定什么数据在业务上合理。

在文档页做四次输入实验

请求体预期原因
{"title":"学习接口"}201,返回 id/title模型和业务规则均通过
{}422模型必填字段缺失,函数尚未执行
{"title":""}422min_length 约束未通过
{"title":" "}400长度检查通过,但业务去空白后为空

再 GET /api/cards,只有第一项新增了记录。区分“框架校验”和“业务校验”,能帮助你决定错误应该在哪里处理。

类型标注不等于严格拒绝所有类型变化

Pydantic 默认允许部分合理转换。例如 int 字段通常可接受数字字符串 "3";"abc" 无法转换才会失败,严格模式或 Field(strict=True) 有不同契约。讲义不能承诺“只要传字符串就必然 422”。

同样,topic: str 只说明类型,不自动排除空白主题;需要长度和去空白后的业务检查。测试应根据实际约定编写,而不是凭类型名称猜测行为。

本次不需要 CORS 的原因

用 /docs 测本机 API 可以先把接口跑通。前端和 API 来源不同时才配置允许来源;带凭据时应列出明确来源,不能把 * 与携带凭据当成万能组合。CORS 管浏览器跨来源读取,不替代登录和对象权限。

排错:ModuleNotFoundError 先检查用的是虚拟环境解释器;找不到 api 模块先检查目录;连接失败先看 Uvicorn 监听地址;422 要读 detail 中的字段位置;500 看服务端日志,面向用户不要直接返回可能含敏感细节的底层堆栈。

验收:四项输入实验和一次 GET 均符合表格,且能在源码指出“由框架完成”和“仍由自己完成”的边界。原课会话、静态页面与模型调用在此基础上选读。


课前须知:FastAPI 是什么

第 02 课 「查资料」里用标准库手写了一个 HTTP 服务器:if 判断路径、自己拼 JSON、
自己解析请求体。那样的写法能让你理解原理,但项目一大就很啰嗦。

FastAPI 是一个现代 Web 框架,帮你把这些活全包了:

「天使与恶魔」整个后端只用了一个 230 行的 app.py,就是因为框架承担了大量杂活。


第一步:跑起来,先看文档

cd worry_debate_game
py -3 -m pip install fastapi uvicorn langchain-openai
py -3 run.py

启动后打开浏览器:

  1. 前端页面:http://127.0.0.1:8000/
  2. 自动接口文档:http://127.0.0.1:8000/docs ← 重点
  3. 备选文档:http://127.0.0.1:8000/redoc

在 /docs 页面里,你能看到所有接口,点开任意一个还能直接在网页上测试调用。
注意看每个接口下面的"请求体示例"和"响应示例"——这些不是你写的,
是 FastAPI 根据你的代码自动生成的。这是 FastAPI 最出名的特性。


第二步:模块地图

worry_debate_game/           (Python 包)
├── app.py       FastAPI 应用:路由、请求模型、会话存储、SSE(本课主角)
├── state.py     游戏状态的数据结构定义(TypedDict)
├── nodes.py     调用大模型生成台词(第 09、10 课)
├── graph.py     用 LangGraph 编排辩论流程(第 10 课)
├── main.py      终端版入口
└── static/      前端页面(HTML/CSS/JS)

app.py 内部又分为五块(先有地图,再读细节):

app.py
├── 应用配置      第 13-32 行   创建 app、跨域设置、挂载静态文件
├── 请求模型      第 49-59 行   Pydantic 类,定义前端该传什么
├── 普通接口      第 64-135 行  /api/health /api/set-key /api/start /api/next
├── SSE 工具      第 138-140 行 拼装流式事件格式(第 09 课细讲)
└── 流式接口      第 143-230 行 /api/start-stream /api/next-stream

第三步:逐块精讲

块 1:创建应用 + 路由装饰器(第 13 行、35 行、64 行)

app = FastAPI(title="天使与恶魔 - 烦恼辩论")

@app.get("/")
async def serve_index():
    ...

@app.get("/api/health")
def health():
    return {"status": "ok"}

@app.post("/api/set-key")
def set_key(req: SetKeyRequest):
    ...

告诉框架「以后有人 GET /api/health,就调用下面这个函数」

还记得第 02 课吗? 「查资料」里是这么找处理函数的:

if path == "/api/health":
    return self._json({...})

一堆 if 链 vs 一行装饰器——装饰器就是更优雅的路由表。
框架内部帮你维护了"路径 → 函数"的映射,你只管挂装饰器。

块 2:Pydantic 请求模型 —— 自动校验(第 49-59 行)

from pydantic import BaseModel

class StartRequest(BaseModel):
    topic: str
    max_rounds: int = 3

class SetKeyRequest(BaseModel):
    api_key: str
@app.post("/api/set-key")
def set_key(req: SetKeyRequest):     # ← 参数类型直接写成模型类
    if not req.api_key or len(req.api_key.strip()) < 8:
        raise HTTPException(400, detail="API Key 格式不正确")
    set_api_key(req.api_key.strip())

这一段是 FastAPI 的核心魔法:

  1. 函数参数写了 req: SetKeyRequest,框架就知道"请求体是个 JSON,应该按这个结构解析"
  2. 前端发来的数据会自动被解析进 req 对象,用 req.api_key 取值
  3. 例如 StartRequest 缺 topic,或 max_rounds 传入无法转换的 "abc" 时,会返回 422;数字字符串 "3" 默认可能转换成功,严格模式另行约定,

并告诉你哪个字段不对——模型约束减少重复校验,但业务规则仍需显式编写

  1. max_rounds: int = 3 表示"不传就默认 3"

对比第 02 课:那边你要自己读 Content-Length、读字节、json.loads、再检查字段;
这里框架全包了。这就是"框架省掉的活"。

顺带看 state.py(第 4-12 行)——游戏状态也用了类似的"类型描述":

class GameState(TypedDict):
    topic: str
    worry: str
    angel_history: List[str]
    demon_history: List[str]
    round: int
    max_rounds: int
    game_active: bool
    tendency: int   # -100 ~ +100,负数倾向恶魔,正数倾向天使

TypedDict 的意思是"这个字典应该有这些键、每个键是这个类型"——
给人和工具看的说明,运行时不强制,但能帮你少犯错。

块 3:错误处理(第 73 行、98-99 行)

if not req.api_key or len(req.api_key.strip()) < 8:
    raise HTTPException(400, detail="API Key 格式不正确")

...
except ValueError as e:
    raise HTTPException(400, detail=str(e))
except Exception as e:
    raise HTTPException(500, detail=f"LLM调用失败: {str(e)}")

框架会把它变成规范的 JSON 错误响应 + 正确的状态码

预期输入错误可以显示给用户;意外异常应写服务端日志,公开响应使用稳定提示,避免原样暴露路径、密钥或其他内部细节

复习第 02 课:自己写时是 self._json({"error": "not found"}, 404);
FastAPI 是 raise HTTPException(404, ...)。同样的事,更清晰的表达。

块 4:跨域(CORS)中间件(第 16-22 行)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

为什么需要这个? 浏览器的安全规则规定:一个网页默认只能请求"同一个来源"的接口。
你的前端如果从别的地址(或用 file:// 直接打开)来调这个后端,浏览器会拦截。
CORS 响应头决定浏览器是否允许页面读取跨来源响应。带凭据时应列出具体来源,不能把下面历史示例中的通配来源与凭据组合当成通用配置;CORS 也不等于鉴权。

就叫中间件(middleware)

块 5:静态文件与首页(第 25-41 行)

static_dir = os.path.join(os.path.dirname(__file__), "static")
if os.path.isdir(static_dir):
    app.mount("/static", StaticFiles(directory=static_dir), name="static")

@app.get("/")
async def serve_index():
    index_path = os.path.join(static_dir, "index.html")
    if os.path.isfile(index_path):
        return FileResponse(index_path)
    raise HTTPException(404, "index.html not found")

以后 /static/style.css 自动对应到磁盘上的文件——一行顶第 02 课几十行 _send_file

这样无论从哪个目录启动程序都不会找不到文件(很实用的习惯)

块 6:会话存储(第 45 行、101 行)

sessions: dict[str, GameState] = {}

@app.post("/api/start")
def start_game(req: StartRequest):
    session_id = str(uuid.uuid4())[:8]     # 生成一个短 id
    state: GameState = { ... }
    ...
    sessions[session_id] = state           # 存进内存字典
    return {"session_id": session_id, ...}

把每局游戏的状态存在一个内存字典里:sessions["abc123"] = state。
后续的 /api/next 用 session_id 取回状态,接着辩论。

和第 04 课对比:这里没上数据库,直接用字典。
好处是简单、快;坏处是——

所以这是"学习/单机原型"的合理选择,不是生产方案。
(生产会用 Redis 或数据库。能说清这个取舍,面试很加分。)

块 7:一次请求的完整旅程

前端点击「开始辩论」时发生了什么:

前端 POST /api/start,body = {"topic": "熬夜", "max_rounds": 3}
   ↓
FastAPI 解析 body → 按 StartRequest 校验(错了自动 422)
   ↓
start_game() 生成 session_id 和初始 state
   ↓
调用 generate_worry_node / angel_node / demon_node(第 09、10 课的 nodes.py)
   ↓
把 state 存进 sessions
   ↓
返回 JSON(FastAPI 自动序列化)
   ↓
前端拿到 session_id 和台词,渲染画面

和第 02 课「查资料」的流程是同一条线:
请求进来 → 找到处理函数 → 干活 → 返回。
差别只在"框架帮你做了多少杂活"。


第四步:对照开源,别人怎么写接口

对照 FastAPI 官方教程

去 FastAPI 官网中文文档看「第一步」,你会发现套路一模一样:

from fastapi import FastAPI
app = FastAPI()

@app.get("/")
def read_root():
    return {"Hello": "World"}

@app.get("/items/{item_id}")
def read_item(item_id: int, q: str | None = None):
    return {"item_id": item_id, "q": q}

你已经掌握了官方教程的核心内容,可以放心去看官方文档进阶部分。

对照 Flask / Django

你的 FastAPIFlaskDjango
路由@app.get(...)@app.route(...)urls.py 配置
参数校验Pydantic 自动手动表单/序列化器
自动文档内置需要插件需要插件
异步支持原生有限较新版本支持
适合API、AI 服务小网站、快速原型大而全的网站

AI 类项目特别喜欢 FastAPI,因为它异步 + 流式响应做得好——
这正是项目里 /api/start-stream 用到的(第 09 课细讲)。

对照:内存会话 vs 真正的存储

方案例子特点
内存字典(教材的)sessions = {}简单、快、重启丢
文件/SQLite「查资料」的 knowledge.db持久、单机够用
Redis大厂常用快 + 持久 + 多进程共享

同一份需求,规模不同,选型不同。能讲清"为什么现在选简单的"也是能力。


动手练习(做完才算过关)

练习 1(热身):用 /docs 测试接口
打开 http://127.0.0.1:8000/docs,找到 /api/health,
点「Try it out」→「Execute」,看看返回结果。
再试试 /api/start,随便填一个 topic,观察它真的调用大模型生成台词了
(没配 API Key 时会报错,这正好让你看 500/400 错误长什么样)。

练习 2(必做):加一个自己的接口

@app.get("/api/version")
def version():
    return {"name": "天使与恶魔", "version": "1.0"}

加完重启服务,刷新 /docs——新接口会自动出现在文档里。
体会一下"改代码 → 文档自动更新"的舒服。

练习 3(必做):给请求模型加约束
把 StartRequest 改成:

from pydantic import BaseModel, Field

class StartRequest(BaseModel):
    topic: str = Field(min_length=1, max_length=50)
    max_rounds: int = Field(default=3, ge=1, le=10)

然后去 /docs 里:把 max_rounds 填成 99 调用,观察返回的 422 错误内容。
说说看:这些校验逻辑,如果手写要写多少行?现在写了几行?

练习 4(必做):观察缺字段的报错
用 /docs 调用 /api/start,但不填 topic,看返回的错误。
把错误信息里的字段名找出来——框架是不是精确告诉了你缺什么?

练习 5(思考题):
如果把 sessions = {} 换成写进 SQLite(像第 04 课那样),
程序重启后会发生什么变化?会带来什么新的麻烦?

看答案

从"持久化"和"生命周期/过期清理"两个角度想。

练习 6(选做,练对照阅读):
去 FastAPI 官方文档找一个"路径参数"的例子(比如 /items/{item_id}),
仿照它给项目加一个 GET /api/session/{session_id},返回该局的状态。
做完你就掌握了"路径参数 + 统一错误处理"这两个常用技能。


本课小结

术语表

术语人话解释
FastAPI现代 Python Web 框架,主打 API 和异步
装饰器(@)在函数上方写的一行,用来"登记"函数(路由、测试等)
端点(endpoint)一个具体的接口地址 + 方法
Pydantic做数据校验的库,FastAPI 的请求模型就是它
BaseModel / TypedDict描述数据结构的方式(前者会校验,后者只是说明)
HTTPException主动抛出 HTTP 错误响应的异常类
中间件请求到达处理函数之前/之后统一做事的组件
CORS浏览器的跨来源访问规则;用中间件放行
静态文件不变的资源(HTML/CSS/JS/图片)
会话 / session_id用 id 区分"不同人/不同局"的状态
422参数校验失败的标准状态码

下节预告

第 09 课:流式输出:AI 为什么一个字一个字蹦。
聚焦 app.py 里那几行看起来奇怪的 sse() 和 StreamingResponse,
以及 nodes.py 里的 stream_angel / stream_demon。
你会搞懂 SSE(服务器推送事件)的原理,以及"打字机效果"是怎么实现的。