说明

对AI应用开发所涉及到的流程、工具、技能进行系列介绍，全部文章收录于《AI应用开发》专栏。关注+收藏，不错过后续精彩。

前置文章

AI应用开发01-环境准备
AI应用开发02-从零构建AI聊天机器人
AI应用开发03-RAG增强知识问答
AI应用开发04-对话+RAG管理桌面端GUI
AI应用开发05-MCP集成
AI应用开发06-SKILL的诞生
AI应用开发07-深入认识MCP

在AI应用开发的前置学习过程中，为了快速验证效果，并未考虑工程化实践。那么接下来，将进入工程化落地阶段，基于python的服务端开发是必学掌握的内容。

概述

FastAPI 是一个基于 Python 的现代 Web 框架，专为构建高性能 API 而设计。本指南聚焦于核心功能，提供选型对比、代码示例及用途说明，不涉及 Docker 或 AI 应用最佳实践。

前置知识：Python 装饰器、HTTP 协议基础、JSON 格式、RESTful API 设计规范。

一、框架选型对比

在 Python 生态中，FastAPI、Flask 和 Django 是最常用的三个 Web 框架。以下是关键维度的对比：

维度 FastAPI Flask Django
性能 高（基于 ASGI，原生异步） 中（同步 WSGI） 中（同步为主）
自动文档 ✅ 内置 Swagger UI / ReDoc ❌ 需自行配置 ❌ 需 django-rest-swagger
数据校验 Pydantic（高性能，自动校验） 需手动或借助 Marshmallow DRF Serializer
依赖注入 ✅ 原生 Depends ❌ 无 ❌ 无
适用场景 API 服务、微服务、AI 服务 小型应用、原型 大型后台、CMS、全栈

选型结论：对于纯粹的后端 API 开发，FastAPI 在开发效率、运行性能和文档体验上均优于 Flask 和 Django。

二、环境搭建与本地启动

2.1 安装

pip install fastapi[standard]

fastapi[standard] 会一并安装 uvicorn（ASGI 服务器）及常用依赖。

2.2 基础示例

创建 main.py：

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def root():
    return {"message": "Hello World"}

2.3 本地启动方式

在项目目录下，使用以下任一命令启动开发服务器：

# 方式一：fastapi 命令行（推荐，支持自动重载）
fastapi dev main.py

# 方式二：直接调用 uvicorn
uvicorn main:app --reload --host 127.0.0.1 --port 8000

启动后访问：

  • http://127.0.0.1:8000 返回 JSON 响应
  • http://127.0.0.1:8000/docs 查看 Swagger UI 交互文档
  • http://127.0.0.1:8000/redoc 查看 ReDoc 风格文档

--reload 参数使代码修改后自动重启服务，适用于开发环境。

三、核心功能详解

3.1 路径操作装饰器

@app.get("/items/{item_id}")
@app.post("/items")
@app.put("/items/{item_id}")
@app.delete("/items/{item_id}")
@app.patch("/items/{item_id}")

用途：将函数与特定的 HTTP 方法和 URL 路径绑定。支持所有标准 HTTP 方法。

3.2 路径参数与查询参数

@app.get("/users/{user_id}")
async def get_user(user_id: int, active: bool = True, page: int = 1):
    return {"user_id": user_id, "active": active, "page": page}
  • 路径参数（user_id）：URL 路径的一部分（如 /users/123），必须提供。类型注解 int 会自动转换并校验。
  • 查询参数（active, page）：URL 中 ? 后的键值对（如 /users/123?active=false&page=2），有默认值时可选，无默认值时则为必填。

用途：自动解析、类型转换和校验，减少手动处理代码。

3.3 请求体与 Pydantic 模型

from pydantic import BaseModel, Field
from typing import Optional

class Item(BaseModel):
    name: str
    price: float = Field(gt=0, description="价格必须大于0")
    tags: list[str] = []
    description: Optional[str] = None

@app.post("/items/")
async def create_item(item: Item):
    return {"item_name": item.name, "price": item.price}

用途：

  • BaseModel 定义数据结构。
  • Field(gt=0) 添加校验约束（如 gt：大于，ge：大于等于，max_length：最大长度）。
  • FastAPI 自动从请求体读取 JSON，按模型校验，失败时返回 422 错误并附带详细说明。

3.4 响应模型

class ItemResponse(BaseModel):
    name: str
    price: float
    created_at: str

@app.post("/items/", response_model=ItemResponse)
async def create_item(item: Item):
    return {"name": item.name, "price": item.price, "created_at": "2025-01-15"}

用途：response_model 用于过滤输出字段、校验响应数据、自动生成 API 文档。建议将请求模型和响应模型分开定义，避免暴露敏感字段。

3.5 依赖注入（Depends）

from fastapi import Depends, HTTPException, Header
from typing import Annotated

def pagination(skip: int = 0, limit: int = 10) -> dict:
    return {"skip": skip, "limit": limit}

def get_current_user(token: str = Header(...)):
    if token != "secret":
        raise HTTPException(status_code=401, detail="Invalid token")
    return {"user_id": 123, "name": "Alice"}

@app.get("/items/")
async def list_items(
    pagination_params: Annotated[dict, Depends(pagination)],
    current_user: Annotated[dict, Depends(get_current_user)]
):
    return {"user": current_user, "pagination": pagination_params}

用途：

  • Depends 声明依赖项，FastAPI 自动执行依赖函数并注入返回值。
  • 依赖函数可使用 yield 实现请求结束后的清理逻辑（如关闭数据库连接）。
  • Annotated 写法（Python 3.9+）保留类型提示，提高代码可读性。
  • 依赖可嵌套，框架自动解析顺序。

3.6 异步与并发

import httpx
import asyncio

@app.get("/proxy")
async def proxy_multiple():
    async with httpx.AsyncClient() as client:
        res1, res2 = await asyncio.gather(
            client.get("https://api1.example.com/data"),
            client.get("https://api2.example.com/data")
        )
    return {"data1": res1.json(), "data2": res2.json()}

用途：在 async def 函数中，I/O 等待时事件循环可处理其他请求，提升并发能力。asyncio.gather 并发执行多个协程。注意：CPU 密集型任务不应放在异步函数中，应使用 BackgroundTasks 或独立进程。

3.7 后台任务（BackgroundTasks）

from fastapi import BackgroundTasks

def send_email(email: str, message: str):
    # 模拟耗时操作
    print(f"Sending to {email}: {message}")

@app.post("/register/")
async def register(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(send_email, email, "Welcome!")
    return {"message": "User registered, email will be sent later"}

用途：在返回响应之后执行任务，不阻塞请求。适用于发送邮件、写日志、触发 webhook 等非关键路径操作。

3.8 流式响应（SSE）

from fastapi.responses import StreamingResponse
import asyncio
import json

async def event_stream():
    for i in range(5):
        yield f"data: {json.dumps({'count': i})}\n\n"
        await asyncio.sleep(1)
    yield "data: [DONE]\n\n"

@app.get("/stream")
async def stream():
    return StreamingResponse(event_stream(), media_type="text/event-stream")

用途：StreamingResponse 接收异步生成器，每次 yield 立即将数据发送给客户端。SSE 协议格式为 data: {json}\n\n，前端可通过 EventSource 接收。适用于 LLM 流式输出、实时日志等单向推送场景。

3.9 WebSocket

from fastapi import WebSocket, WebSocketDisconnect

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Echo: {data}")
    except WebSocketDisconnect:
        print("Client disconnected")

用途：双向实时通信，适用于聊天室、协同编辑、实时游戏等场景。连接建立后，双方均可主动发送消息。

3.10 APIRouter（模块化路由）

# routers/users.py
from fastapi import APIRouter

router = APIRouter(prefix="/users", tags=["用户"])

@router.get("/{user_id}")
async def get_user(user_id: int):
    return {"user_id": user_id}

# main.py
from fastapi import FastAPI
from routers import users

app = FastAPI()
app.include_router(users.router)

用途：

  • prefix 为路由组统一添加路径前缀。
  • tags 用于在自动文档中对接口分组。
  • 支持路由器嵌套，便于大型项目组织代码。

3.11 CORS 中间件

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000", "https://example.com"],  # 生产环境限定具体域名
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

用途：添加跨域资源共享（CORS）头，允许浏览器跨域请求。开发时可临时使用 ["*"]，生产环境必须指定明确域名。

四、推荐的项目结构

project/
├── app/
│   ├── __init__.py
│   ├── main.py                 # 应用入口，创建 app 实例，注册路由
│   ├── core/                   # 配置、依赖、异常处理
│   │   ├── config.py
│   │   ├── dependencies.py
│   │   └── exceptions.py
│   ├── routers/                # 按业务模块拆分路由
│   │   ├── users.py
│   │   └── items.py
│   ├── models/                 # Pydantic 模型（请求/响应）
│   ├── services/               # 业务逻辑层（可选）
│   └── utils/                  # 工具函数
├── tests/
├── .env
└── requirements.txt

该结构遵循关注点分离原则，路由层负责请求响应，服务层实现业务逻辑，模型层定义数据结构。

五、总结

FastAPI 的核心优势在于：

  • 自动文档：基于 OpenAPI 标准，零配置生成交互式文档。
  • 数据校验：Pydantic 提供声明式、高性能的校验。
  • 依赖注入：简化代码复用和测试。
  • 异步支持：原生 async/await 提升 I/O 密集型任务的并发能力。

掌握上述核心功能后，即可构建生产级别的 API 服务。

# python # fastapi # 开发语言 # 人工智能
Logo
AtomGit开源社区

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。

更多推荐

cover

企业级智能家居系统管理系统源码｜SpringBoot+Vue+MyBatis架构+MySQL数据库【完整版】

avatar AtomGit开源社区
cover

SpringBoot+Vue 疫情打卡健康评测系统平台完整项目源码+SQL脚本+接口文档【Java Web毕设】

avatar AtomGit开源社区
cover

前后端分离毕业设计系统系统｜SpringBoot+Vue+MyBatis+MySQL完整源码+部署教程

avatar AtomGit开源社区