书架 · Python 学习系列 · 07 · Web 服务开发（FastAPI）下一章：08 实战项目与资源 →

07 · Web 服务开发（FastAPI）

本章目标：用 FastAPI 写生产可用的 AI Web 服务：参数校验、依赖注入、异步、SSE 流式透传 LLM 输出、数据库、部署。学完能独立交付第 08 章的 RAG 问答 API 项目。

前置：03（asyncio）、04（流式输出）。

📌 本章 FastAPI SSE 写法核对自官方文档（访问日期 2026-09-05，见文末）。

1. 为什么 AI 服务首选 FastAPI

框架	定位	AI 场景适配
FastAPI	现代 API 框架	原生 async、类型驱动校验、SSE/WebSocket 流式、自动 OpenAPI 文档
Flask	轻量老牌	同步为主，流式/异步需额外拼装
Django	全家桶（ORM/Admin/模板）	内容型网站强；纯 API 服务偏重

☕ Spring Boot 老兵的映射：FastAPI ≈ "Spring Boot Web + Bean Validation + springdoc-openapi"三件套，但约定大于配置到极致——路由装饰器声明、类型提示即校验规则、注解（Depends）即注入。没有 IoC 容器：依赖就是普通函数。

uv add "fastapi[standard]"      # 含 uvicorn 服务器等官方推荐全家桶

2. 十分钟后端起步：路由与自动文档

# main.py
from fastapi import FastAPI

app = FastAPI(title="AI 服务", version="0.1.0")

@app.get("/hello")
def hello() -> dict:
    return {"msg": "world"}          # dict/list/Pydantic 模型直接返回，自动序列化 JSON

uv run fastapi dev          # 开发模式：保存即热重载（≈ spring-boot-devtools）
# 访问 http://127.0.0.1:8000/docs  ← 自动生成的 Swagger UI ⭐
# 访问 http://127.0.0.1:8000/redoc ← ReDoc 风格文档

路径/查询参数：

from fastapi import FastAPI

app = FastAPI()                             # 接 2.1 的应用实例

@app.get("/users/{user_id}")                # 路径参数
def get_user(user_id: int):                  # 类型提示 int → 自动转换 + 422 校验
    ...

@app.get("/search")                          # 查询参数：?q=py&limit=10
def search(q: str, limit: int = 10):         # 无默认值 = 必填，有默认值 = 可选
    ...

☕ @app.get ≈ @GetMapping；路径参数 ≈ @PathVariable、查询参数 ≈ @RequestParam，但 FastAPI 按位置和类型自动区分，不用注解逐个标。

3. 请求校验：Pydantic

from pydantic import BaseModel, Field, EmailStr

class ChatRequest(BaseModel):               # ≈ DTO + Bean Validation 注解合一
    prompt: str = Field(min_length=1, max_length=8000, description="用户提问")
    model: str = "deepseek-v4-pro"
    temperature: float = Field(default=0.7, ge=0, le=2)
    stream: bool = False
    history: list[dict] = []                # 嵌套结构直接声明

@app.post("/chat")
def chat(req: ChatRequest):                 # 请求体：类型即模型
    return {"echo": req.prompt, "model": req.model}

# 非法请求体（超长/缺字段/类型错）→ FastAPI 自动返回 422 + 逐字段错误明细
# 校验通过的才进函数 —— 相当于 @Valid + 全局 ExceptionHandler 白送

响应模型（同时约束出参 + 文档）：

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()          # 接 2.1
text = "模型回复内容"     # 业务代码里生成（如调用 LLM 的结果）

class ChatResponse(BaseModel):
    answer: str
    total_tokens: int

@app.post("/chat", response_model=ChatResponse)
def chat(req) -> ChatResponse:
    ...
    return ChatResponse(answer=text, total_tokens=42)   # 多余字段自动剔除

☕ Pydantic v2 核心用 Rust 写的校验内核，性能好；Field(ge=0, le=2) ≈ @DecimalMin/@DecimalMax。LLM 结构化输出解析也用它（第 04 章第 5 节）——一份模型定义同时管：请求校验、响应文档、LLM 输出解析。

4. 依赖注入

from fastapi import Depends, Header, HTTPException

# 依赖 = 普通函数（没有 @Component/@Bean）
async def get_llm_client() -> AsyncOpenAI:
    return AsyncOpenAI(api_key=..., base_url=...)

# 鉴权依赖：抛 HTTPException 即中断（≈ Filter/Interceptor）
async def verify_token(authorization: str = Header()):      # 自动读请求头
    if not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="未认证")
    token = authorization.removeprefix("Bearer ")
    if not check(token):
        raise HTTPException(status_code=403, detail="token 无效")
    return token

@app.post("/chat")
async def chat(
    req: ChatRequest,
    client: AsyncOpenAI = Depends(get_llm_client),   # 注入依赖
    token: str = Depends(verify_token),              # 依赖可叠加、可缓存
):
    ...

# 依赖还能嵌套（verify_token 里再 Depends()）——够用，没有 Spring 那套容器魔法

☕ Depends ≈ 构造器注入 + AOP 前置校验的合体。同请求内默认缓存（use_cache=True），别拿它存会话状态。

5. 异步路由：AI 服务的正确姿势

# ⭐ 调 LLM / IO 的路由必须 def 前加 async，用 AsyncOpenAI + await
@app.post("/chat")
async def chat(req: ChatRequest, client: AsyncOpenAI = Depends(get_llm_client)):
    resp = await client.chat.completions.create(
        model=req.model,
        messages=[{"role": "user", "content": req.prompt}],
    )
    return {"answer": resp.choices[0].message.content}

# ❌ 反面教材：async 路由里用同步 OpenAI 客户端
#    → 一个慢请求卡住整个事件循环，所有用户一起转圈（第 03 章铁律 3）

# CPU 密集（本地模型推理后处理等）：丢线程池，别阻塞
result = await asyncio.to_thread(heavy_fn, data)

同步 def 路由（连数据库的旧同步库）FastAPI 会自动丢线程池执行，不会卡事件循环——但 AI 服务统一 async 风格最省心。

6. SSE 流式接口：把 LLM 输出透传给前端

AI 服务的核心场景：前端打字机效果 = 服务端 SSE 流。两种官方写法（出处：FastAPI 官方 SSE 教程，见文末）：

方式一：StreamingResponse（通用写法，所有版本可用）

import json
from collections.abc import AsyncIterator
from fastapi.responses import StreamingResponse

@app.post("/chat/stream")
async def chat_stream(req: ChatRequest) -> StreamingResponse:
    async def gen() -> AsyncIterator[str]:
        stream = await aclient.chat.completions.create(   # 第 04 章的异步生成器
            model=req.model,
            messages=[{"role": "user", "content": req.prompt}],
            stream=True,
        )
        async for chunk in stream:
            delta = chunk.choices[0].delta.content
            if delta:
                yield f"data: {json.dumps({'v': delta}, ensure_ascii=False)}\n\n"  # SSE 格式
        yield "data: [DONE]\n\n"

    return StreamingResponse(gen(), media_type="text/event-stream")

SSE 线格式（前端 EventSource/fetch 流式读取消费的就是这个）：

data: {"v": "你"}\n\n
data: {"v": "好"}\n\n
data: [DONE]\n\n

方式二：EventSourceResponse（新版 FastAPI 内置，自动拼 SSE 帧）

from fastapi.sse import EventSourceResponse, ServerSentEvent

@app.post("/chat/stream", response_class=EventSourceResponse)
async def chat_stream(req: ChatRequest):
    async def gen():
        async for delta in stream_from_llm(req):
            yield ServerSentEvent(data={"v": delta}, event="token")   # 自动 JSON 编码
        yield ServerSentEvent(raw_data="[DONE]", event="done")
    return EventSourceResponse(gen())

依据：FastAPI 官方文档新增了 Server-Sent Events 教程（fastapi.sse.EventSourceResponse / ServerSentEvent；data 自动 JSON 序列化、raw_data 原样发送）。旧项目里常见的第三方 sse-starlette 即被此内置能力替代。

☕ 对 Java 老兵：SSE ≈ Spring 的 SseEmitter/Flux<ServerSentEvent>，但这里返回值是一个异步生成器（第 02 章知识闭环）——框架边收 LLM chunk 边往 socket 推，内存占用恒定。

7. 中间件、CORS、异常处理

from fastapi import Request
from fastapi.middleware.cors import CORSMiddleware
import time, logging

logger = logging.getLogger("uvicorn.error")

@app.middleware("http")                        # ≈ Filter/HandlerInterceptor
async def log_requests(request: Request, call_next):
    start = time.perf_counter()
    response = await call_next(request)        # 放行到路由
    cost = (time.perf_counter() - start) * 1000
    logger.info(f"{request.method} {request.url.path} -> {response.status_code} {cost:.0f}ms")
    return response

app.add_middleware(                            # CORS：前后端分离必备
    CORSMiddleware,
    allow_origins=["http://localhost:5173"],   # 前端 dev 地址，生产收紧
    allow_methods=["*"], allow_headers=["*"],
)

from fastapi.responses import JSONResponse

class BusinessError(Exception): ...

@app.exception_handler(BusinessError)          # ≈ @ControllerAdvice
async def biz_handler(request: Request, exc: BusinessError):
    return JSONResponse(status_code=400, content={"code": "BIZ_ERROR", "msg": str(exc)})

8. 数据库：SQLAlchemy 2.x

uv add sqlalchemy

from sqlalchemy import create_engine, String, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session

engine = create_engine("sqlite:///app.db", echo=False)    # 换 MySQL: "mysql+pymysql://..."

class Base(DeclarativeBase): ...                          # 2.x 新式声明基类

class Message(Base):
    __tablename__ = "message"
    id: Mapped[int] = mapped_column(primary_key=True)     # 类型注解即列类型（2.x 风格）
    session_id: Mapped[str] = mapped_column(String(64), index=True)
    role: Mapped[str]
    content: Mapped[str]

Base.metadata.create_all(engine)                          # 原型阶段建表；生产用 Alembic 迁移

def save_msg(session_id: str, role: str, content: str) -> None:
    with Session(engine) as session:                      # 上下文管理器自动提交/回滚
        session.add(Message(session_id=session_id, role=role, content=content))
        session.commit()

def load_history(session_id: str) -> list[Message]:
    with Session(engine) as session:
        stmt = select(Message).where(Message.session_id == session_id).order_by(Message.id)
        return list(session.scalars(stmt))                # 2.x 统一用 select() 风格

☕ SQLAlchemy ≈ Hibernate/JPA：DeclarativeBase≈@Entity、Session≈EntityManager、2.x 的 select()≈JPA Criteria/JPQL。AI 服务里它最常干的活就是第 05 章说的：存对话历史。

9. 部署

# 生产：uvicorn 多进程（Windows 上 workers 支持有限，容器内跑 Linux 是正解）
uv run fastapi run --port 8000 --workers 4

# 或直接 uvicorn
uv run uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

Dockerfile（AI 服务标准形态）：

FROM python:3.12-slim
COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv   # uv 官方镜像用法
WORKDIR /app
COPY pyproject.toml uv.lock ./
RUN uv sync --frozen --no-dev          # 按锁文件精确还原（≈ mvn -o）
COPY . .
CMD ["uv", "run", "uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

生产清单：密钥走环境变量注入（严禁进镜像）；健康检查端点 /health；超时与重试（LLM 上游）；反向代理（Nginx）终结 SSE 时关闭缓冲（proxy_buffering off）。

自测清单与练习

[ ] FastAPI 拿什么生成文档？两个文档路径是？
[ ] 路径参数和查询参数在代码里怎么区分？
[ ] Pydantic Field(ge=0, le=2) 相当于 Bean Validation 什么注解？
[ ] Depends 的鉴权依赖里抛什么异常？async def 依赖返回值怎么进路由参数？
[ ] async 路由里用同步 SDK 客户端会发生什么？
[ ] SSE 的线格式长什么样？StreamingResponse 与 EventSourceResponse 各自怎么写？
[ ] SQLAlchemy 2.x 查询用什么函数构建语句？Session 上下文管理器保证什么？
[ ] 生产部署 SSE 需要反向代理注意什么？

练习 1：把第 04 章多轮对话壳改成 /chat + /chat/stream 两个接口，带 ChatRequest 校验与鉴权依赖。 练习 2：加 GET /history/{session_id}，从 SQLite 返回历史（用第 8 节的模型）。 练习 3：写 pytest（httpx + FastAPI 的 TestClient 或 ASGITransport）测：非法请求 422、无 token 401、mock LLM 后正常返回。 练习 4：用 curl -N http://127.0.0.1:8000/chat/stream 直接肉眼观察 SSE 流。

10. 参考与出处

以下均为官方一手来源（访问日期：2026-09-05）：

主题	出处
FastAPI 教程全貌	fastapi.tiangolo.com/tutorial（路由/校验/依赖/中间件均为官方教程页）
SSE / EventSourceResponse / ServerSentEvent	Server-Sent Events 教程、SSE API 参考
Pydantic	docs.pydantic.dev（Field 校验/序列化）
SQLAlchemy 2.x	docs.sqlalchemy.org（ORM Tutorial）（DeclarativeBase / Mapped / select 2.0 风格）
uvicorn / 部署	FastAPI 部署文档、uvicorn.org
uv 官方 Docker 用法	docs.astral.sh/uv/guides/integration/docker

⬅️ 返回目录｜ ➡️ 下一章：08-实战项目与资源