难度:★★☆(需要会 Python 函数)
教材:worry_debate_game/app.py(230 行)
预计时间:讲解 60 分钟 + 练习 40 分钟
学完你能:用 FastAPI 写出接口、看懂自动文档,并理解框架比手写服务器省了哪些活
前置: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":""} | 422 | min_length 约束未通过 |
| {"title":" "} | 400 | 长度检查通过,但业务去空白后为空 |
再 GET /api/cards,只有第一项新增了记录。区分“框架校验”和“业务校验”,能帮助你决定错误应该在哪里处理。
Pydantic 默认允许部分合理转换。例如 int 字段通常可接受数字字符串 "3";"abc" 无法转换才会失败,严格模式或 Field(strict=True) 有不同契约。讲义不能承诺“只要传字符串就必然 422”。
同样,topic: str 只说明类型,不自动排除空白主题;需要长度和去空白后的业务检查。测试应根据实际约定编写,而不是凭类型名称猜测行为。
用 /docs 测本机 API 可以先把接口跑通。前端和 API 来源不同时才配置允许来源;带凭据时应列出明确来源,不能把 * 与携带凭据当成万能组合。CORS 管浏览器跨来源读取,不替代登录和对象权限。
排错:ModuleNotFoundError 先检查用的是虚拟环境解释器;找不到 api 模块先检查目录;连接失败先看 Uvicorn 监听地址;422 要读 detail 中的字段位置;500 看服务端日志,面向用户不要直接返回可能含敏感细节的底层堆栈。
验收:四项输入实验和一次 GET 均符合表格,且能在源码指出“由框架完成”和“仍由自己完成”的边界。原课会话、静态页面与模型调用在此基础上选读。
第 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
启动后打开浏览器:
http://127.0.0.1:8000/http://127.0.0.1:8000/docs ← 重点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
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):
...
FastAPI(title=...) 创建应用,title 会显示在 /docs 顶部@app.get("/api/health") 叫装饰器:它的作用是"登记"——告诉框架「以后有人 GET /api/health,就调用下面这个函数」
json.dumps)还记得第 02 课吗? 「查资料」里是这么找处理函数的:
if path == "/api/health":
return self._json({...})
一堆 if 链 vs 一行装饰器——装饰器就是更优雅的路由表。
框架内部帮你维护了"路径 → 函数"的映射,你只管挂装饰器。
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 的核心魔法:
req: SetKeyRequest,框架就知道"请求体是个 JSON,应该按这个结构解析"req 对象,用 req.api_key 取值"abc" 时,会返回 422;数字字符串 "3" 默认可能转换成功,严格模式另行约定,并告诉你哪个字段不对——模型约束减少重复校验,但业务规则仍需显式编写
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 的意思是"这个字典应该有这些键、每个键是这个类型"——
给人和工具看的说明,运行时不强制,但能帮你少犯错。
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)}")
raise HTTPException(状态码, detail=说明):主动报错的标准姿势。框架会把它变成规范的 JSON 错误响应 + 正确的状态码
except 捕获底层错误,包成带 detail 的 HTTP 错误,预期输入错误可以显示给用户;意外异常应写服务端日志,公开响应使用稳定提示,避免原样暴露路径、密钥或其他内部细节
复习第 02 课:自己写时是 self._json({"error": "not found"}, 404);
FastAPI 是 raise HTTPException(404, ...)。同样的事,更清晰的表达。
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
为什么需要这个? 浏览器的安全规则规定:一个网页默认只能请求"同一个来源"的接口。
你的前端如果从别的地址(或用 file:// 直接打开)来调这个后端,浏览器会拦截。
CORS 响应头决定浏览器是否允许页面读取跨来源响应。带凭据时应列出具体来源,不能把下面历史示例中的通配来源与凭据组合当成通用配置;CORS 也不等于鉴权。
allow_origins=["*"] = 允许任何来源(开发阶段方便,上线要收紧)就叫中间件(middleware)
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")
app.mount("/static", StaticFiles(...)):把整个 static/ 文件夹变成可访问的网址。以后 /static/style.css 自动对应到磁盘上的文件——一行顶第 02 课几十行 _send_file
FileResponse:专门用来返回文件(会自动处理类型、长度等响应头)os.path.dirname(__file__) 是"这个 py 文件所在目录",用来拼出绝对路径,这样无论从哪个目录启动程序都不会找不到文件(很实用的习惯)
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 或数据库。能说清这个取舍,面试很加分。)
前端点击「开始辩论」时发生了什么:
前端 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 官网中文文档看「第一步」,你会发现套路一模一样:
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}
{item_id} 会自动传进函数参数(你在第 02 课是手动裁字符串)item_id: int 会自动把字符串转成整数、类型错就报错StartRequest 用法的"函数参数版"你已经掌握了官方教程的核心内容,可以放心去看官方文档进阶部分。
| 你的 FastAPI | Flask | Django | |
|---|---|---|---|
| 路由 | @app.get(...) | @app.route(...) | urls.py 配置 |
| 参数校验 | Pydantic 自动 | 手动 | 表单/序列化器 |
| 自动文档 | 内置 | 需要插件 | 需要插件 |
| 异步支持 | 原生 | 有限 | 较新版本支持 |
| 适合 | API、AI 服务 | 小网站、快速原型 | 大而全的网站 |
AI 类项目特别喜欢 FastAPI,因为它异步 + 流式响应做得好——
这正是项目里 /api/start-stream 用到的(第 09 课细讲)。
| 方案 | 例子 | 特点 |
|---|---|---|
| 内存字典(教材的) | 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},返回该局的状态。
做完你就掌握了"路径参数 + 统一错误处理"这两个常用技能。
@app.get 顶手写服务器一串 ifHTTPException 是主动报错的标准方式,400 怪请求方、500 怪服务端StaticFiles + FileResponse 一行搞定静态文件服务(对比第 02 课手写 _send_file)/docs 自动文档是 FastAPI 的招牌——接口写完就自带说明书| 术语 | 人话解释 |
|---|---|
| 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(服务器推送事件)的原理,以及"打字机效果"是怎么实现的。