第4章 FastAPI 和 Pydantic¶

学习目标¶

使用 FastAPI 快速搭建符合 RESTful 风格的 API 服务
运用 Pydantic 进行数据校验、序列化与模型层设计
理解并实现依赖注入（Dependency Injection）解耦业务逻辑
编写中间件处理跨切关注点（日志、CORS、鉴权、错误处理）
利用自动文档与 OpenAPI 提升团队协作效率

问题导入¶

AI Agent 要对外提供服务，就需要一个"门面"：接收用户请求、校验参数、调用 Agent 核心、返回结构化结果。传统 Web 框架（如 Django、Flask）要么体量过重，要么缺少内建的异步与数据校验支持。FastAPI 基于 Starlette 与 Pydantic 构建，原生支持 asyncio，用类型提示自动完成参数校验、序列化、文档生成，被誉为"现代 Python Web 的标杆"。Pydantic v2 用 Rust 重写核心，性能比 v1 提升 5~50 倍，已成为 Python 数据校验的事实标准。

本章将带你从零搭建一个 AI Agent API 服务：定义 Pydantic 模型、编写路由、注入依赖、挂载中间件、自动生成文档。掌握这套组合，你就能在几十行代码内交付一个生产可用的 Agent HTTP 服务。

核心概念¶

FastAPI 路由与异步¶

FastAPI 用 @app.get("/path")、@app.post(...) 等装饰器声明路由。路由处理函数既可以是同步 def（会被丢到线程池执行），也推荐写成 async def（直接在事件循环中运行，与 httpx、数据库异步驱动配合最佳）。路径参数、查询参数、请求体的类型由函数签名中的类型提示决定，FastAPI 据此自动校验与转换。

Pydantic 模型与校验¶

Pydantic 模型继承自 BaseModel，通过字段类型注解声明结构。它提供： - 运行时校验：传入数据类型不符或缺失必填字段时抛出 ValidationError，含详细错误定位。 - 序列化：model_dump() 转 dict，model_dump_json() 转 JSON。 - 字段约束：Field(min_length=..., max_length=..., ge=..., pattern=...)。 - 嵌套模型：模型可嵌套，自动递归校验。 - 自定义校验器：@field_validator、@model_validator 实现复杂业务规则。

依赖注入¶

FastAPI 的依赖注入通过 Depends(callable) 实现。一个依赖可以是普通函数或协程，其返回值会被注入到路由参数中；依赖本身也可依赖其他依赖，形成依赖树。常见用途：解析并校验 Token、获取数据库会话、注入配置、复用查询参数逻辑。依赖的作用域可设为整个应用（app.dependency_overrides，便于测试替换）。

中间件¶

中间件包裹整个请求-响应周期，在路由处理前后插入逻辑。FastAPI（基于 Starlette）支持 @app.middleware("http") 装饰器或 app.add_middleware(CORSMiddleware, ...)。典型用途：CORS 跨域、请求日志与耗时统计、统一异常处理、压缩响应、鉴权前置。中间件按注册逆序执行（洋葱模型）。

自动文档与 OpenAPI¶

FastAPI 根据路由签名、Pydantic 模型、responses、tags、description 等元数据，自动生成 OpenAPI 3.1 schema，并在 /docs 提供 Swagger UI、/redoc 提供 ReDoc。这不仅省去手写文档的成本，更使前后端、QA、第三方集成方获得可交互的 API 契约。

原理图¶

flowchart LR
    Client[客户端请求] --> MW1[中间件1: CORS]
    MW1 --> MW2[中间件2: 日志/鉴权]
    MW2 --> Router[路由分发]
    Router --> Dep[依赖注入解析]
    Dep --> Val[Pydantic 请求体校验]
    Val --> Handler[路由处理函数 async def]
    Handler --> Core[Agent 核心业务]
    Core --> Resp[Pydantic 响应模型序列化]
    Resp --> MW2
    MW2 --> MW1
    MW1 --> Client
    Handler -.-> Docs[/docs 自动文档 OpenAPI]
    Val -.-> Docs
    Resp -.-> Docs

图：FastAPI 请求生命周期——中间件洋葱模型包裹路由，Pydantic 完成校验与序列化，元数据驱动自动文档

示例代码¶

定义 Pydantic 模型¶

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

class ChatRequest(BaseModel):
    """Agent 对话请求模型"""
    session_id: str = Field(..., min_length=8, description="会话 ID")
    message: str = Field(..., min_length=1, max_length=4000, description="用户消息")
    model: Literal["gpt-4o-mini", "gpt-4o"] = "gpt-4o-mini"
    temperature: float = Field(0.7, ge=0.0, le=2.0)

    @field_validator("message")
    @classmethod
    def strip_message(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("消息不能为空白")
        return v.strip()

class ChatResponse(BaseModel):
    """Agent 对话响应模型"""
    session_id: str
    reply: str
    tool_calls: list[str] = Field(default_factory=list)
    usage_tokens: int = Field(0, ge=0)

# 校验示例
req = ChatRequest(session_id="sess1234", message="  你好  ", model="gpt-4o")
print(req.model_dump())
# {'session_id': 'sess1234', 'message': '你好', 'model': 'gpt-4o', 'temperature': 0.7}

完整 FastAPI 服务¶

from fastapi import FastAPI, Depends, HTTPException, Header
from fastapi.middleware.cors import CORSMiddleware
from typing import Annotated
import time

app = FastAPI(
    title="AI Agent API",
    version="1.0.0",
    description="AI Agent 对话服务示例",
)

# 中间件：CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

# 中间件：请求耗时日志
@app.middleware("http")
async def timing_middleware(request, call_next):
    start = time.perf_counter()
    response = await call_next(request)
    elapsed = time.perf_counter() - start
    response.headers["X-Process-Time"] = f"{elapsed:.4f}"
    return response

# 依赖：鉴权
async def verify_token(authorization: Annotated[str, Header()]) -> str:
    if not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="无效的认证头")
    token = authorization.removeprefix("Bearer ")
    if token != "secret-token":
        raise HTTPException(status_code=403, detail="Token 无效")
    return token

# 依赖：根据 token 获取用户
async def get_current_user(token: Annotated[str, Depends(verify_token)]) -> dict:
    return {"user": "alice", "token": token}

@app.get("/health")
async def health() -> dict:
    return {"status": "ok"}

@app.post("/chat", response_model=ChatResponse, tags=["Agent"])
async def chat(
    body: ChatRequest,
    user: Annotated[dict, Depends(get_current_user)],
) -> ChatResponse:
    """Agent 对话接口"""
    # 这里接入真实 Agent 核心
    reply = f"[{user['user']}] 你说: {body.message}"
    return ChatResponse(
        session_id=body.session_id,
        reply=reply,
        tool_calls=[],
        usage_tokens=len(body.message),
    )

# 启动：uvicorn main:app --reload

流式响应（SSE）¶

from fastapi.responses import StreamingResponse
import asyncio
import json

async def stream_agent(message: str):
    for word in message.split():
        await asyncio.sleep(0.1)
        chunk = {"token": word}
        yield f"data: {json.dumps(chunk)}\n\n"
    yield "data: [DONE]\n\n"

@app.post("/chat/stream", tags=["Agent"])
async def chat_stream(body: ChatRequest):
    return StreamingResponse(
        stream_agent(body.message),
        media_type="text/event-stream",
    )

启动与访问¶

# 安装：pip install fastapi uvicorn[standard]
# 启动：uvicorn main:app --reload --port 8000
# 访问文档：http://localhost:8000/docs
# 健康检查：curl http://localhost:8000/health

企业案例¶

案例 1：AI Agent API 服务¶

某团队用 FastAPI 将 Agent 能力封装为 HTTP 服务：/chat 同步返回完整回答，/chat/stream 用 SSE 流式返回 token，/tools/execute 执行工具调用。所有请求/响应用 Pydantic 模型定义，OpenAPI 文档自动同步给前端与 QA。鉴权用 Depends 注入 JWT 校验，不同角色注入不同权限依赖。上线两周即完成 SDK 自动生成（基于 OpenAPI）。

案例 2：多租户配置与依赖覆盖¶

一个 SaaS Agent 平台需支持多租户配置。团队用 Depends 注入 get_tenant_config(request)，根据域名/Token 动态加载租户限流、模型配额、可用工具白名单。测试时通过 app.dependency_overrides[get_tenant_config] = lambda: test_config 注入假配置，无需改动业务代码即可隔离测试。

案例 3：统一异常处理与错误码¶

平台早期各路由自行 raise HTTPException，错误码不统一。后改为自定义业务异常 AgentError(code, message)，并用 @app.exception_handler(AgentError) 统一捕获，转换为标准 JSON 响应 {"code": ..., "message": ..., "request_id": ...}。前端按 code 字段统一处理，联调效率显著提升。

常见问题¶

Q1：FastAPI 路由该用 def 还是 async def？ 若调用异步库（httpx、异步数据库驱动）必须用 async def；若必须用同步阻塞库（如 requests），用普通 def，FastAPI 会自动放线程池执行避免阻塞事件循环。切勿在 async def 中直接调用阻塞 I/O。

Q2：Pydantic v2 与 v1 的主要区别？ v2 核心用 Rust 编写，性能大幅提升；API 调整：dict() → model_dump()、parse_obj() → model_validate()；校验器改为 @field_validator / @model_validator；Config 内部类改为 model_config = ConfigDict(...)。迁移可借助官方 bump-pydantic 工具。

Q3：依赖注入和直接在路由里写逻辑有什么好处？ 依赖可复用（多个路由共享鉴权/配置逻辑）、可测试（通过 dependency_overrides 替换为桩）、可组合（依赖可依赖其他依赖）。它让路由函数聚焦业务，横切逻辑集中管理，符合单一职责原则。

Q4：如何让自动文档更友好？ 为路由与模型添加 description、summary、tags、response_model、responses（声明各状态码响应）、examples。Pydantic 字段用 Field(..., description=..., examples=[...])。良好的元数据让 /docs 成为可交互的契约文档。

面试题¶

题1：FastAPI 为什么能自动生成 OpenAPI 文档？ 参考答案：FastAPI 在启动时遍历所有路由，从函数签名的类型提示推断参数（路径/查询/请求体）、从 Pydantic 模型生成 JSON Schema、结合装饰器元数据（tags、responses、description）组装成 OpenAPI 3.1 文档对象，挂载在 /openapi.json，并由 Swagger UI / ReDoc 渲染。

题2：解释 Pydantic 中 @field_validator 与 @model_validator 的区别。 参考答案：@field_validator 针对单个字段，在该字段解析后调用，常用于格式化或单字段规则；@model_validator(mode="after") 在整个模型校验完成后调用，可访问所有字段，用于跨字段联合校验。mode="before" 则在类型转换前调用，常用于处理原始输入。

题3：中间件的执行顺序是怎样的？ 参考答案：中间件以"洋葱模型"运行：注册顺序中先注册的在外层，请求进入时按外→内执行"前置"逻辑，路由处理完毕后按内→外执行"后置"逻辑。即 app.add_middleware(A) 后 app.add_middleware(B)，请求顺序为 A前 → B前 → 路由 → B后 → A后。

题4：如何在 FastAPI 中实现请求限流？ 参考答案：可用 Depends 注入限流依赖，基于 Redis 计数（如滑动窗口或令牌桶），超限 raise HTTPException(429)。也可挂载 slowapi 等第三方中间件。注意限流键应结合用户 ID 与 IP，避免被单一维度绕过。

练习题¶

上机：基于本章代码，扩展一个 POST /tools/search 接口，请求体含 query 与 top_k（默认 5，1~20），返回 list[SearchResult]（含 title、url、snippet）。用 Depends 注入鉴权，并启动服务在 /docs 验证。
思考：若 Agent 调用 LLM 耗时较长，如何避免 HTTP 客户端超时？请从同步接口返回任务 ID + 轮询、SSE 流式、WebSocket 三种方案比较优劣。
上机：为 /chat 接口编写一个 @app.exception_handler，捕获自定义 AgentError，统一返回 {"code": err.code, "message": err.message}，并测试一个触发错误的请求。

小结¶

FastAPI + Pydantic 是现代 Python Web 服务的黄金组合：类型提示驱动的校验与文档、原生异步、轻量依赖注入、灵活中间件，让 AI Agent 服务的搭建变得高效而规范。理解请求生命周期、善用依赖注入与异常处理，是交付生产级 API 的关键。下一章将深入 HTTP 客户端与流式交互，与本章服务端形成完整闭环。延伸阅读：FastAPI 官方文档、Pydantic v2 文档。