前言

如果说前一阶段补的是 Python 服务基础，比如配置、类型标注、日志、异常、装饰器、异步这些“底层能力”，那么到了这一部分，就真正开始进入“写后端接口”的阶段了。

这部分需要真正理解：

• 路由层到底该放什么
• 请求模型和响应模型为什么重要
• 中间件和依赖注入分别适合解决什么问题
• 为什么文件上传、流式输出、任务接口在 AI/Agent 服务里特别常见
• 一个接口从请求进来到返回结果，中间到底发生了什么

所以这篇文章从更接近真实项目的角度，把我对 FastAPI 核心能力的理解完整讲一遍。

---

一、FastAPI 这一部分到底在学什么

从服务开发的角度看，FastAPI 这一部分真正要学的，其实是下面这些问题：

1. 一个后端服务的接口入口层应该怎么组织
2. 请求参数和响应结构应该怎么定义
3. 什么逻辑适合放路由里，什么逻辑应该放到业务层
4. 什么是依赖注入，它为什么能让接口更整洁
5. 文件上传和流式输出为什么是很典型的服务能力
6. 为什么有些接口适合同步返回结果，有些接口更适合提交任务后再查询状态

---

二、先建立一个整体画面：FastAPI 在后端服务里负责什么

一个最常见的 HTTP 服务，通常会经历这样的链路：

1. 应用启动
2. 路由系统初始化
3. 中间件挂载
4. 请求进入
5. 路由匹配
6. 参数解析和校验
7. 前置依赖执行，比如鉴权
8. 业务函数执行
9. 返回结果
10. 框架把结果转成 HTTP 响应

如果再细一点，从 FastAPI 的角度来看，它主要承担的是：

• 接收 HTTP 请求
• 解析请求参数
• 校验请求数据
• 执行依赖逻辑
• 调用业务层
• 返回规范响应
• 生成接口文档

所以fastapi不是业务本身，而是业务的 HTTP 入口层。

---

三、路由组织：为什么不是所有接口都堆在一个文件里

刚开始写小项目时，很多人都会把接口直接堆在一个文件里。例如：

from fastapi import FastAPI

app = FastAPI()

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

@app.post("/chat")
async def chat():
    ...

@app.post("/files/upload")
async def upload():
    ...

小 demo 这样当然能跑。但一旦接口开始变多，很快就会出现这些问题：

• 文件越来越长
• 健康检查、聊天、文件上传、任务接口都混在一起
• 后续维护时不好找
• 团队协作时也容易相互影响

所以更推荐的做法是：按领域拆路由。

一个典型示意写法如下：

from fastapi import APIRouter

api_router = APIRouter()
api_router.include_router(health_router, tags=["health"])
api_router.include_router(chat_router, prefix="/chat", tags=["chat"])
api_router.include_router(file_router, prefix="/files", tags=["files"])
api_router.include_router(task_router, prefix="/tasks", tags=["tasks"])

这段代码想表达的核心思想是：

• 总路由器负责聚合
• 每个业务领域独立维护自己的接口
• 路径前缀和文档标签也更清晰

---

四、请求模型和响应模型：它们不是装饰，而是在定义接口契约

很多人一开始知道要写 BaseModel，但并没有真的意识到它的重要性。实际上，请求模型和响应模型做的是一件很重要的事情：定义接口契约。

1. 请求模型

from typing import Literal
from pydantic import BaseModel, Field

class ModelSettings(BaseModel):
    temperature: float = Field(default=0.2, ge=0.0, le=1.0)
    max_tokens: int = Field(default=400, ge=64, le=2000)

class ChatRequest(BaseModel):
    user_id: str = Field(min_length=2, max_length=64)
    message: str = Field(min_length=1, max_length=2000)
    stream_mode: Literal["sentence", "word"] = "sentence"
    response_format: Literal["text", "structured"] = "text"
    use_tools: bool = False
    enable_memory: bool = True
    model_settings: ModelSettings = Field(default_factory=ModelSettings)

这段代码其实是在明确告诉外部调用方：

• 这个接口要收哪些字段
• 每个字段是什么类型
• 哪些字段有长度限制
• 哪些字段有默认值
• 哪些字段只能取固定几个值

比如：

• user_id 必须是字符串，而且长度有限制
• message 不能为空
• stream_mode 只能是 sentence 或 word
• response_format 只能是 text 或 structured
• model_settings 如果不传，会用一套默认配置

所以请求模型是在替接口回答：我到底接收什么输入。

为什么这比手写 if 判断更好？如果不写模型，你也可以手动判断：

if not user_id:
    ...

if len(message) > 2000:
    ...

if stream_mode not in ["sentence", "word"]:
    ...

但这样写会有几个问题：

• 校验逻辑容易散
• 不同接口风格容易不一致
• 重复代码很多
• 文档信息很难统一沉淀

而模型的好处是：

• 结构集中
• 校验规则清楚
• 框架自动处理很多错误场景
• 自动参与接口文档生成

2. 响应模型

from typing import Any
from pydantic import BaseModel, Field

class UsageMetrics(BaseModel):
    input_tokens: int
    output_tokens: int
    total_tokens: int
    latency_ms: int

class ChatResponse(BaseModel):
    reply: str
    trace_id: str
    provider: str
    model: str
    structured_data: dict[str, Any] | None = None
    usage: UsageMetrics | None = None
    memory_keys: list[str] = Field(default_factory=list)

这段代码的重点是：接口返回什么，也要是明确可描述的。

后端服务最怕的一种情况就是：

• 今天返回这个字段
• 明天字段换名字
• 某些情况下又缺字段
• 前端和调用方完全靠猜

响应模型的意义就是：把输出结构稳定下来。

3. 接口契约

最简单的话来总结这一块：请求模型定义“你可以给我什么”，响应模型定义“我会还给你什么”。这就是接口契约。

---

五、路由函数为什么通常都很短

一个很典型的聊天接口，可能会像这样：

from fastapi import APIRouter

router = APIRouter()

@router.post("", response_model=ChatResponse)
async def chat(payload: ChatRequest, container: ContainerDep) -> ChatResponse:
    return await container.agent_service.chat(payload)

一个好的路由函数往往就是应该短。因为它的职责本来就不应该是“把所有事都做完”，而应该是：

1. 接住 HTTP 请求
2. 接收已经校验好的参数
3. 拿到依赖对象
4. 调用业务层
5. 返回结果

如果把很多业务都写在路由里，很容易变成这样：

• 参数校验逻辑写一堆

• 鉴权写一堆

• 调数据库写一堆

• 拼响应写一堆

• 捕获异常写一堆
  结果就是路由层越来越长、越来越乱。

  把职责分开后：

• 路由层负责接入和协调

• service 层负责业务处理

• schema 层负责结构定义

---

六、依赖注入：它在 FastAPI 里到底解决了什么问题

1. 最直观的理解：帮你自动拿对象

from typing import Annotated
from fastapi import Depends, Request

def get_container(request: Request) -> ServiceContainer:
    return request.app.state.container

ContainerDep = Annotated[ServiceContainer, Depends(get_container)]

然后你在路由里可以这样写：

@router.post("")
async def chat(payload: ChatRequest, container: ContainerDep) -> ChatResponse:
    return await container.agent_service.chat(payload)

依赖注入的本质：路由函数并没有自己创建 container，也没有自己去某个全局变量里找它，而是通过依赖让 FastAPI 自动提供这个对象。所以依赖注入在这里最直观的价值是：让路由函数不必手动准备它需要的资源。

2. 好处

这种写法的好处其实很实际：

• 路由更短
• 获取对象的方式统一
• 更容易复用
• 更容易测试和替换实现

你现在不用把依赖注入想得特别“架构化”，先把它理解成一句话就够了：依赖注入是在帮你自动准备路由执行所需的前置对象和逻辑。

---

七、鉴权：为什么它特别适合做成依赖

再往前一步，依赖注入不只是能“拿对象”，它还特别适合做“接口前置逻辑”。鉴权就是最典型的例子之一。

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

async def verify_api_token(
    authorization: Annotated[str | None, Header(alias="Authorization")] = None,
    x_api_key: Annotated[str | None, Header(alias="X-API-Key")] = None,
) -> None:

    expected = "dev-token"
    bearer = f"Bearer {expected}"

    if authorization == bearer or x_api_key == expected:
        return

    raise HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Invalid or missing API token",
    )

这段逻辑的意思：

• 从请求头拿 token
• 验证是否符合预期
• 符合就放行
• 不符合就返回 401

然后你可以把它挂在整个 router 上：

router = APIRouter(dependencies=[Depends(verify_api_token)])

这个 router 下的所有接口，在真正执行业务逻辑之前，都会先跑一遍鉴权。

为什么鉴权适合做成依赖？

因为它具备几个非常典型的特征：

• 是接口执行前的前置条件
• 很多接口都共享
• 但不一定要对全局所有请求都生效

如果把这种逻辑写在每个接口里，不仅重复，而且容易漏。

---

八、文件上传： FastAPI 里一个很有代表性的能力点

很多人学 Web 接口时，脑子里默认的接口形态就是：

• 收 JSON
• 回 JSON

但真实服务显然不止这一种情况，文件上传就是一个非常典型的例子。

from fastapi import APIRouter, File, UploadFile, status

@router.post("/upload", response_model=FileUploadResponse, status_code=status.HTTP_201_CREATED)
async def upload_file(
    container: ContainerDep,
    file: UploadFile = File(...),
) -> FileUploadResponse:
    return await container.file_service.save_upload(file)

1. 这段代码里最关键的点是什么

1. UploadFile。

它表示这个参数不是普通文本字段，而是上传文件对象。

1. File(...)。

它告诉框架：这个参数来自文件上传，而不是普通 JSON 请求体。

1. 201 Created。

这个状态码在表达：这次请求成功创建了一份新的资源，而不只是普通“调用成功”。

2. 为什么文件上传值得重视

在 AI/Agent 应用里，文件上传场景特别常见。

例如：

• 上传知识库文档
• 上传待解析文件
• 上传上下文资料
• 上传图片、表格或文本数据
  所以文件上传不是边缘功能，而是很多 AI 应用服务的真实需求。

3. 建立的思维

接口并不只有“收 JSON 回 JSON”一种形态。一个成熟一点的服务，往往还要能处理：

• 文件
• 流
• 任务

---

九、流式输出：为什么它在 AI/LLM 场景里这么常见

如果说文件上传是在表达“接口不只是收 JSON”，那流式输出则是在表达：接口也不一定非得一次性返回完整结果。

from fastapi.responses import StreamingResponse

@router.post("/stream")
async def stream_chat(payload: ChatRequest, container: ContainerDep) -> StreamingResponse:
    stream = container.agent_service.stream_chat(payload)
    return StreamingResponse(stream, media_type="text/event-stream")

1. 这段代码在干什么

• service 层提供一个可以持续产出内容的异步流
• 路由层把这个流包装成 HTTP 流式响应
• text/event-stream 表示这是一种事件流输出格式

2. 为什么这特别适合大模型场景

因为大模型生成文本时，本来就不是“一次性瞬间产生全部结果”的。更真实的情况通常是：

• 先生成一点
• 再继续生成一点
• 边生成边返回给用户

这带来的体验会明显更好：

• 首字反馈更快
• 用户不会一直干等
• 对话感更强

所以在 AI/Agent 服务里，流式输出是非常真实的产品需求。

---

十、任务接口：为什么不是所有接口都应该同步把事情做完

一个典型任务提交接口：

from fastapi import status

@router.post("/chat", response_model=TaskSubmitResponse, status_code=status.HTTP_202_ACCEPTED)
async def submit_chat_task(
    payload: ChatRequest,
    container: ContainerDep,
) -> TaskSubmitResponse:
    task = await container.task_service.submit_chat_task(payload)
    return TaskSubmitResponse(task_id=task.task_id, status=task.status, trace_id=task.trace_id)

然后再看查询接口：

@router.get("/{task_id}", response_model=TaskDetailResponse)
async def get_task(task_id: str, container: ContainerDep) -> TaskDetailResponse:
    return await container.task_service.get_task(task_id)

1. 这里最值得注意的是 202 Accepted

它的意思不是“结果已经有了”，而是：请求我收到了，但真正处理还在继续。这和普通同步接口的思路完全不同。

普通同步接口更像：

• 请求来了
• 服务立即处理
• 处理完后直接返回最终结果

任务型接口更像：

• 请求来了
• 服务先接单
• 返回一个 task id
• 后续客户端再根据 task id 查询处理状态和结果

2. 为什么这在 AI/Agent 服务里很常见

因为很多任务并不适合一直挂在主请求里同步执行。

例如：

• 长时间文本生成
• 文件解析
• 知识库构建
• 检索索引更新
• 多步骤工具调用

如果这些都同步做，接口会面临很多问题：

• 响应时间太长
• 超时风险高
• 用户体验差
• 失败和重试难管理

3. 建立了什么系统感

它帮你开始区分两种接口思维：

• 立即返回最终结果的接口
• 先提交任务、后查询状态的接口

而这正是后面继续学 Redis、任务队列、Celery、Agent 长流程执行时非常重要的基础认知。

---

十一、健康检查接口：虽然简单，但很有代表性

一个最小健康检查通常像这样：

class HealthResponse(BaseModel):
    status: str
    app: str
    task_backend: str

@router.get("/health", response_model=HealthResponse)
async def health_check(request: Request) -> HealthResponse:
    return HealthResponse(
        status="ok",
        app="Agent Backend Playground",
        task_backend=request.app.state.container.task_backend_name,
    )

它虽然简单，但非常有代表性，因为它体现了三件事：

1. 服务即使是最简单的接口，也应该有明确结构
2. 运行时状态信息是可以对外暴露一部分的
3. 健康检查是服务可观测性的最小入口

---

十二、把 FastAPI 这部分重新串起来：一条请求到底经历了什么

如果把前面的内容重新串成一条完整的请求链路，大概可以理解成这样：

1. FastAPI 应用启动
2. 总路由器挂载各个领域路由
3. 请求进入后先匹配到具体接口
4. 路由在执行前先跑依赖逻辑，比如鉴权
5. 框架自动解析和校验请求模型
6. 路由函数拿到已经校验好的参数
7. 路由函数调用 service 层
8. service 层返回业务结果
9. 路由层把结果交给响应模型组织输出
10. 如果是文件上传、流式响应、任务型接口，则走对应特殊模式

这条链路里，FastAPI 做的其实不是“处理所有业务”，而是：把 HTTP 请求组织、校验、分发、包装成一套稳定入口。

这个视角一旦建立起来，后面再写接口时就不会只想着“这段代码能不能跑”，而会更自然地去思考：

• 这段逻辑属于路由层还是业务层
• 这个校验应该放在 schema 里还是 service 里
• 这个前置条件适合依赖还是中间件
• 这个接口适合同步返回还是任务化处理

---

十三、这一部分对后面学习 Agent 服务有什么帮助

现在学的不是一个独立的 Web 框架，而是在为后面的 AI/Agent 应用服务能力打基础。

比如：

1. 对聊天服务有帮助

普通聊天接口和流式聊天接口，都是 AI 应用里非常高频的能力。

2. 对文件型应用有帮助

知识库问答、文档解析、上传资料辅助生成，这些都离不开文件上传接口。

3. 对任务型应用有帮助

很多 Agent 能力不是瞬时完成的，所以任务提交与状态查询思维非常重要。

4. 对接口设计能力有帮助

开始学会从“接口契约”而不是“临时能跑”来思考 API。

---

十四、总结：

这一部分最大的收获是：开始知道一个后端服务的接口层应该怎么组织了。

理解得更清楚了：

• 路由层负责接和转发，不应该堆满业务逻辑
• 请求模型和响应模型是在定义接口契约
• 依赖注入是在帮接口自动准备资源和前置逻辑
• 鉴权这种接口前置条件很适合做成依赖
• 文件上传和流式输出不是特殊技巧，而是真实服务常见能力
• 任务接口和同步接口的设计思路本来就不一样

FastAPI 核心能力学的不是“怎么把接口写出来”，而是“怎么把接口写成一个真正像样的服务入口层”。

---

下一篇预告

下一篇会继续复盘存储与任务系统这一层：为什么 MySQL、Redis 和异步任务会成为 Agent 服务里非常重要的一部分，以及它们到底应该怎么分工。