FastAPI与Pydantic服务开发#

1. ASGI vs WSGI，uvicorn 的 worker 模型#

ASGI 把一次连接拆成事件：scope 描述连接（http / websocket / lifespan），receive() 拿到请求体分块或客户端消息，send() 发响应头、响应体分块。流式输出就是多次 send 响应体。

worker 模型：uvicorn 一个进程 = 一个事件循环。uvicorn app:app --workers 4 或 gunicorn 管理 uvicorn worker，本质是多进程，每个进程各自一份内存。所以：

• 进程内的 dict 缓存、限流计数、「当前谁在跑」这类状态，多 worker / 多副本下各算各的，需要放 Redis 或数据库；
• worker 数通常按 CPU 核数起步，AI 应用瓶颈在下游 API，单进程靠协程就能扛很多并发，worker 多了反而多占内存（每个进程各加载一次模型客户端、tokenizer）。

2. async def 和 def 路由的区别#

依赖函数同理：同步依赖也会进线程池。判断规则很简单：全链路有异步驱动就写 async def，否则老实写 def，最差的是 async def 里调 requests / 同步 DB 驱动。临时混用时用 await asyncio.to_thread(...)，原理见 Python异步编程与asyncio事件循环。

3. 路由、依赖注入、中间件、生命周期#

from contextlib import asynccontextmanager
from fastapi import FastAPI, Depends, HTTPException, Request
import httpx

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动：建连接池、做启动校验；失败直接抛异常，服务起不来
    app.state.http = httpx.AsyncClient(timeout=30)
    app.state.db = await create_pool(DSN)
    yield
    # 关闭：优雅释放资源
    await app.state.http.aclose()
    await app.state.db.close()

app = FastAPI(lifespan=lifespan)          # 替代已弃用的 @app.on_event("startup")

async def get_session(request: Request):  # yield 依赖：默认 scope="request"，yield 之后的清理在响应发出后执行
    async with request.app.state.db.acquire() as conn:
        yield conn

async def current_user(token: str = Depends(oauth2_scheme)) -> User:
    user = decode_jwt(token)
    if not user:
        raise HTTPException(status_code=401, detail="invalid token")
    return user

@app.post("/api/chat", response_model=ChatResp)
async def chat(req: ChatReq, user: User = Depends(current_user),
               conn = Depends(get_session)):
    ...

• Depends：FastAPI 在启动时解析函数签名，构建依赖树；同一请求内同一依赖默认只算一次（use_cache=True）；yield 依赖用来管 DB 连接、事务，yield 之后的代码做清理。测试时用 app.dependency_overrides[current_user] = fake_user 替换，不用 mock 框架。
• yield 依赖的清理时机（官方文档的版本变迁）：
  • 0.106.0：清理从「响应发出后」改到「路由函数返回后、响应发出前」，目的是不在网络传输期间占着资源；副作用是 StreamingResponse 在生成器里再用这个 DB 会话时，会话已经被关了。
  • 0.110.0：yield 依赖里 except 捕获异常后如果不重新 raise，异常不再被自动转发给异常处理器，和普通 Python 语义一致。
  • 0.118.0：清理改回响应发出之后执行，流式响应可以放心使用 yield 依赖给的资源。
  • 0.121.0：新增 Depends(fn, scope="function")，清理在路由函数结束后、响应发出前执行；默认 scope="request" 在响应发出后执行。scope="request" 的依赖，它的子依赖也必须是 "request"。
• 中间件：@app.middleware("http") 包在所有路由外层，适合打 trace_id、记耗时、统一异常。它基于 BaseHTTPMiddleware，对流式响应和 contextvars 有已知限制，流式场景更稳的做法是写纯 ASGI 中间件（直接包 receive/send）。CORS、GZip 用内置中间件。
• lifespan：进程级的启动/关闭钩子。每个 worker 进程都会各跑一遍，所以「只执行一次」的事（数据库迁移）不能直接放这里，要加分布式锁或单独跑。另外 httpx 的 ASGITransport 测试客户端默认不触发 lifespan，测试里依赖 lifespan 初始化的资源要自己准备。
• 异常处理：HTTPException 返回指定状态码；@app.exception_handler(MyError) 把业务异常转成统一 JSON；请求体校验失败自动返回 422 和逐字段错误。

4. Pydantic v2：校验与序列化#

v2 的核心校验逻辑用 Rust 重写（pydantic-core），API 也改了名：

from typing import Literal, Annotated
from pydantic import BaseModel, Field, ConfigDict, field_validator

class ChatReq(BaseModel):
    model_config = ConfigDict(extra="forbid")          # 多传未知字段直接 422
    thread_id: str | None = None
    query: Annotated[str, Field(min_length=1, max_length=4000)]
    temperature: float = Field(0.3, ge=0, le=2)

    @field_validator("query", mode="before")          # before：先 strip 再做 min_length 校验，全空格会被拒
    @classmethod
    def strip(cls, v):
        return v.strip() if isinstance(v, str) else v

• v1 的旧方法名在 v2 里大多还能调用，但会发 DeprecationWarning；parse_raw / parse_file 已弃用，用 model_validate_json 代替。
• 默认是宽松模式："1" 会被转成 int 1；strict=True 才要求类型精确。接 LLM 输出时宽松模式反而有用，能容忍模型把数字写成字符串。
• response_model 会按模型过滤返回值，内部字段（如 password_hash）不会漏出去——这是它和直接返回 dict 的区别。
• 不是 BaseModel 的类型（list[Item]、联合类型）用 TypeAdapter(list[Item]).validate_python(data)。

5. 用 Pydantic 约束 LLM 结构化输出#

同一个模型类，既生成给 LLM 的 schema，又校验 LLM 的返回：

class SearchCall(BaseModel):
    action: Literal["search"]
    platform: Literal["jd", "taobao"]
    keyword: str

class AskUser(BaseModel):
    action: Literal["ask_user"]
    question: str

class Decision(BaseModel):
    step: Annotated[SearchCall | AskUser, Field(discriminator="action")]  # 判别联合

schema = Decision.model_json_schema()   # 放进 tool 定义 / response_format

async def decide(messages, max_retry=2) -> Decision:
    for _ in range(max_retry + 1):
        raw = await llm.complete(messages, json_schema=schema)
        try:
            return Decision.model_validate_json(raw)
        except ValidationError as e:
            # 把逐字段错误回灌给模型，让它自己改，而不是盲目重试
            messages.append({"role": "user",
                             "content": f"上次输出不合法：{e.errors()}，请只输出合法 JSON"})
    raise LLMOutputError("结构化输出多次校验失败")

• 判别联合（discriminator）让校验器按 action 字段直接选分支，报错也更清楚。
• 服务端校验不能省：即使用了厂商的 JSON mode / strict schema，字段语义（枚举是否在业务白名单、ID 是否存在）还得自己校验。约束解码层面的原理见 结构化输出与约束解码。

6. 流式端点：StreamingResponse / SSE / WebSocket#

from fastapi.responses import StreamingResponse

@app.post("/api/chat/stream")
async def chat_stream(req: ChatReq, request: Request):
    async def gen():
        try:
            async for delta in llm.stream(req.query):
                if await request.is_disconnected():   # 客户端走了就别再烧 token
                    break
                yield f"data: {json.dumps({'delta': delta}, ensure_ascii=False)}\n\n"
            yield "event: done\ndata: {}\n\n"
        except asyncio.CancelledError:
            # 连接断开时生成器可能被取消：记录后重新抛
            raise
    return StreamingResponse(gen(), media_type="text/event-stream",
                             headers={"Cache-Control": "no-cache",
                                      "X-Accel-Buffering": "no"})  # 让 Nginx 别缓冲

@app.websocket("/ws/{thread_id}")
async def ws(websocket: WebSocket, thread_id: str):
    await websocket.accept()
    try:
        while True:
            msg = await websocket.receive_json()
            await websocket.send_json({"type": "ack", "id": msg.get("id")})
    except WebSocketDisconnect:
        cleanup(thread_id)

• SSE 格式：每条 data: ... 以空行结束，可带 event:、id:；id 配合重连时的 Last-Event-ID 做断点续传。
• FastAPI 0.135.0 起内置 SSE：from fastapi.sse import EventSourceResponse, ServerSentEvent，路由写 response_class=EventSourceResponse 并直接 yield，每个 item 按 JSON 编码进 data:；声明返回类型 AsyncIterable[Item] 时用 Pydantic 校验和序列化。要设 event / id / retry 就 yield ServerSentEvent(...)。它会自动在空闲时每 15 秒发一条 ping 注释、设置 Cache-Control: no-cache 和 X-Accel-Buffering: no，也支持 POST。老版本常用第三方 sse-starlette 或上面的 StreamingResponse 手写。
• SSE 单向、走普通 HTTP，适合纯输出；WebSocket 双向，适合中途要回传（取消、澄清回复）。选型和前端渲染见 流式输出与实时交互。
• BackgroundTasks 在响应发出后于同进程里执行，不适合跑 Agent 长任务（进程重启就丢、没有重试），长任务走队列，见 Agent长任务队列与服务化。

7. 和 Spring Boot 的对照#

最大的思维差异：Spring 的 Bean 是进程内长生命周期单例，FastAPI 的 Depends 默认是每请求求值；想要单例就放 app.state 或模块级对象，在 lifespan 里初始化。

FastAPI 是基于 ASGI 的 Python Web 框架，底层 Starlette 负责路由、中间件和 WebSocket，Pydantic 负责按类型注解做校验和序列化。ASGI 和 WSGI 的区别在于它是异步接口，一个请求是一个协程，等 IO 时能让出，天然支持流式和长连接，所以很适合 LLM 应用。部署用 uvicorn，一个进程一个事件循环，多 worker 是多进程，进程内状态不共享。开发时有几个关键点：第一，async def 路由直接跑在事件循环上，里面不能调同步阻塞库，def 路由会被丢进线程池，写错了会卡死整个进程；第二，Depends 做依赖注入，yield 依赖管连接和事务，测试用 dependency_overrides 替换；第三，用 lifespan 管启动和关闭，比如建 httpx 客户端和连接池，但每个 worker 都会跑一遍。Pydantic v2 用 model_validate 和 model_dump，extra forbid 拒绝未知字段，response_model 过滤敏感字段。我也用它约束 LLM 结构化输出：同一个模型导出 JSON Schema 给模型，返回后用 model_validate_json 校验，失败就把字段错误回灌让模型改，限定重试次数。流式用 StreamingResponse 返回 SSE，新版本也可以用内置的 EventSourceResponse，注意检测客户端断开和关 Nginx 缓冲。长任务不要放 BackgroundTasks，API 进程只负责入队，Agent 在独立 worker 里跑。结合项目时可以讲：lifespan 里做了哪些启动校验（比如依赖的数据库、队列不可用就拒绝启动）、进程怎么拆分，以及怎么验证多 worker 下状态没有各算各的。

追问方向：

• “async def 里调了 requests 会怎样？” → 事件循环被阻塞，这个 worker 进程里所有请求都等它返回，表现为延迟整体飙升但 CPU 不高；改用 httpx.AsyncClient，或者把路由改成 def 让它进线程池。
• “def 路由为什么在高并发下吞吐上不去？” → 同步路由由 anyio 线程池执行，默认上限 40 个线程，超出的请求排队；可以调大 limiter，但根本办法是换异步驱动。
• “yield 依赖的清理代码什么时候执行？流式响应里能用它给的 DB 会话吗？” → 取决于版本：0.106.0 到 0.117.x 是在响应发出前清理，流式生成器里再用会话会报已关闭；0.118.0 起默认改回响应发出后清理，可以用；0.121.0 起还能用 Depends(fn, scope="function") 显式要求在路由函数返回后就清理。
• “Depends 的依赖在一个请求里会被调用几次？” → 默认同一请求内缓存结果，只调一次（use_cache=True）；要每次重新求值就写 Depends(fn, use_cache=False)。
• “lifespan 和 on_event 的区别？多 worker 下有什么坑？” → lifespan 是一个 asynccontextmanager，启动和关闭写在同一个函数里，能共享局部变量，on_event 已弃用；每个 worker 进程各执行一次，迁移、定时任务这类全局只做一次的事要加锁或拆出去。
• “Pydantic v2 和 v1 有哪些不兼容？” → 方法改名（parse_obj → model_validate，dict() → model_dump()），validator 改成 field_validator / model_validator，class Config 改成 model_config = ConfigDict(...)；v2 核心在 Rust，校验更快。
• “response_model 有什么用，直接返回 dict 不行吗？” → response_model 会按模型过滤和校验输出，内部字段不会漏，文档里也有正确的响应 schema；直接返回 dict 就原样序列化，容易把 password_hash 之类的字段带出去。
• “LLM 返回的 JSON 校验失败怎么处理？” → 捕获 ValidationError，把 e.errors() 里的字段路径和原因回灌给模型重试，设最大重试次数；超过后走降级（返回澄清问题或报错），不要无限重试。
• “SSE 客户端断开了，服务端怎么知道？” → 在生成器里轮询 await request.is_disconnected()，或者依赖断开时生成器被取消抛 CancelledError；两者都要停止继续调用 LLM，否则还在花钱。
• “为什么 Agent 长任务不用 BackgroundTasks？” → BackgroundTasks 在同一进程里、响应之后执行，进程重启就丢，没有重试和状态查询，也不能单独扩容；长任务要进消息队列由 worker 执行。

易错：

• ❌ “FastAPI 是异步的，所以随便写都快” → async def 里混同步阻塞调用比全写 def 还差。
• ❌ “开 --workers 8 后内存里的限流计数还准” → 多进程各算一份，全局状态要放 Redis。
• ❌ “用了 JSON mode 就不用 Pydantic 校验” → JSON mode 只保证语法合法，字段语义和业务白名单还是要自己校验。