返回文章列表 · 文章归档

从 asyncio 到 FastAPI:重新理解 Python 后端的“异步”

从 asyncio 的协程、Task、取消、超时与背压出发,逐步理解 ASGI、Uvicorn、FastAPI 中间件、依赖注入和请求生命周期,并结合 Agent 系统总结异步后端工程化的关键问题。

  • Python
  • FastAPI

从 asyncio 到 FastAPI:重新理解 Python 后端的“异步”

> 学习随记 · Python 后端工程化

最近重新系统梳理了一遍 Python 的异步编程,并顺着 asyncio 一路学习到了 ASGI、Uvicorn 和 FastAPI。

以前写 Python 项目时,我对“异步”的理解其实比较简单:

```python

async def xxx():

await something()

```

看到 `async` 和 `await`,就觉得这已经是异步编程了。

但这次真正把 asyncio、事件循环、Task、取消、超时,以及 FastAPI 的请求生命周期串起来之后,我才发现:

> **异步并不是把函数前面加一个 `async`,而是一套完整的并发执行模型。**

而 FastAPI 也不只是“一个很好用的 Python Web 框架”,它背后实际上连接了 **ASGI、事件循环、Middleware、依赖注入以及请求生命周期**。

这也是我这阶段学习最大的收获。

---

一、重新认识 asyncio

1. `async/await` 只是入口

Python 的异步编程核心并不是 `async` 和 `await` 本身,而是:

```text

Coroutine

Task

Event Loop

调度多个协程

```

一个最简单的例子:

```python

import asyncio

async def work(name: str):

print(f"{name} start")

await asyncio.sleep(1)

print(f"{name} done")

async def main():

await asyncio.gather(

work("A"),

work("B"),

work("C"),

)

asyncio.run(main())

```

这里最重要的并不是“用了三个协程”。

而是:

> 当一个协程执行到 `await`,并且等待的是一个真正异步的操作时,事件循环可以去执行其他 Task。

这也是为什么异步特别适合:

  • 网络请求
  • 数据库 IO
  • 文件 IO
  • RPC
  • Agent Tool 调用
  • LLM API 请求

这些场景的共同特点都是:

> **CPU 不一定很忙,但程序经常需要等待外部资源。**

---

二、Task 和 Coroutine 并不是一回事

这是这次学习过程中我觉得比较重要的一个概念。

Coroutine 更像是:

> “我有一个可以被异步执行的工作。”

而 Task 更像是:

> “把这个工作交给事件循环调度。”

例如:

```python

coro = work("A")

task = asyncio.create_task(coro)

```

创建 Task 以后,事件循环才真正可以独立调度它。

所以在设计异步程序时,不能只看到:

```python

async def

```

就认为它已经“并发执行”了。

---

三、异步系统真正麻烦的地方:取消和超时

如果只是:

```python

await asyncio.gather(...)

```

异步其实并没有想象中复杂。

真正让我意识到异步工程化难度的,是:

> **如果任务执行到一半,怎么办?**

例如一个 Agent 正在执行:

```text

用户请求

生成 Plan

调用 Tool

执行命令

等待结果

生成最终回答

```

如果用户中途取消请求,或者系统触发超时,那么整个任务应该怎么办?

不能简单粗暴地:

```python

task.cancel()

```

然后认为事情结束了。

因为一个真正的业务任务可能已经:

  • 创建数据库记录
  • 调用外部 API
  • 执行工具
  • 修改系统状态
  • 写入审计日志

所以取消实际上是一个**业务语义问题**。

---

四、Cancellation:取消不是“强制杀死”

这让我开始意识到:

> **任务取消和任务失败不是一回事。**

例如:

```python

async def run_task():

try:

await do_something()

except asyncio.CancelledError:

清理资源

raise

finally:

无论成功、失败还是取消,都需要执行

await write_audit()

```

这里的 `finally` 就非常重要。

因为对于 Agent / 运维系统来说:

```text

成功

失败

超时

取消

异常

```

都应该尽可能留下完整的状态记录。

尤其是我的 Agent 项目中存在:

```text

Plan

Guard

Preflight

Approval

Execute

Audit

```

那么“取消”就不能只是 Python 层面的一个异常。

它应该成为整个任务状态机的一部分。

---

五、Timeout 也不是简单的“超过时间就报错”

同样的道理:

```python

async with asyncio.timeout(10):

await run_task()

```

看起来很简单。

但真正需要考虑的是:

```text

Timeout

当前 Task 被取消

业务任务进入什么状态?

正在执行的 Tool 怎么办?

是否需要回滚?

Audit 怎么记录?

用户最终看到什么?

```

所以我现在对 Timeout 的理解已经从:

> “给函数设置一个最大执行时间”

变成了:

> **Timeout 是整个任务生命周期中的一种异常状态。**

这也是为什么我觉得 Agent 系统里的异步编程,比普通 CRUD 应用复杂很多。

---

六、Semaphore、Queue 和 Backpressure

另一个比较重要的认识是:

> 异步 ≠ 无限并发。

例如:

```python

sem = asyncio.Semaphore(10)

async def call_tool():

async with sem:

return await real_call()

```

这里限制的不是“异步”,而是:

> **同时最多允许多少任务访问某个资源。**

这在 Agent 系统里非常常见。

假设一个 Agent 同时产生 100 个 Tool Call:

```text

Agent

├── Tool 1

├── Tool 2

├── Tool 3

├── ...

└── Tool 100

```

如果毫无限制地全部发出去,很容易把:

  • 数据库
  • API
  • RabbitMQ
  • 下游服务
  • CPU
  • 内存

直接打爆。

所以异步系统除了要考虑:

> “怎么并发?”

还必须考虑:

> **“并发到什么程度?”**

这就是 Backpressure 开始变得重要的地方。

---

七、从 asyncio 走到 FastAPI

把 asyncio 的这些概念串起来以后,再看 FastAPI,感觉完全不一样了。

以前我看到的是:

```text

FastAPI

写 API

```

现在更倾向于把它理解成:

```text

Client

Nginx

Uvicorn

ASGI

FastAPI

Middleware

Dependency Injection

Route Handler

Business Logic

```

FastAPI 只是整个链路中的一部分。

---

八、ASGI:FastAPI 为什么能异步

这次学习让我比较明显地感受到:

> **如果不了解 ASGI,就很难真正理解 FastAPI 的异步。**

传统 WSGI 可以简单理解成:

```text

Request

Application

Response

```

而 ASGI 面向的是异步世界:

```text

HTTP / WebSocket

ASGI

Application

async / await

```

因此:

```python

async def endpoint():

result = await call_llm()

return result

```

并不是 FastAPI 自己凭空实现的。

它背后依赖的是整个 ASGI 异步生态。

---

九、Uvicorn 也不是“一个启动 FastAPI 的命令”

以前:

```bash

uvicorn main:app

```

对我来说就是:

> 启动 FastAPI。

现在更准确的理解是:

```text

┌──────────────┐

HTTP Request ──────→│ Uvicorn │

│ ASGI Server │

└──────┬───────┘

ASGI App

FastAPI

Middleware / DI

Endpoint

```

Uvicorn 负责的是 ASGI Server 这一层。

FastAPI 则是 ASGI Application。

这两个概念分开以后,整个 Python Web 服务的结构清晰了很多。

---

十、Middleware:请求真正进入业务代码之前

FastAPI 的 Middleware 也让我重新理解了:

> 一个 HTTP 请求并不是直接进入 Endpoint。

例如:

```python

@app.middleware("http")

async def middleware(request, call_next):

start = time.time()

response = await call_next(request)

cost = time.time() - start

print(f"request cost: {cost}")

return response

```

可以抽象成:

```text

Request

Middleware

Authentication

Business Logic

Response

Middleware

```

这就非常适合放:

  • 日志
  • Trace ID
  • 请求耗时
  • 全局上下文
  • 安全检查
  • 统一响应处理

当然,不是什么东西都应该塞进 Middleware。

这也是后面继续学习 FastAPI 时需要进一步区分的问题。

---

十一、Depends:FastAPI 的依赖注入

FastAPI 的:

```python

Depends()

```

之前看起来只是一个“方便写认证代码”的东西。

现在再看,它实际上解决的是:

> **如何把一个请求所依赖的公共能力抽离出来,并由框架负责组装。**

例如:

```python

async def get_current_user():

...

@app.get("/profile")

async def profile(

user=Depends(get_current_user)

):

return user

```

于是业务函数可以更加关注业务本身:

```text

Request

Dependency

Authentication

Authorization

Endpoint

```

这对于我之后做 Agent API、RBAC、权限控制会比较重要。

---

十二、从“会写 API”到“理解请求生命周期”

目前这阶段最大的变化,其实不是学会了多少 API。

而是我开始尝试从:

> **一个函数是怎么执行的**

转变成:

> **一个请求从进入服务器到离开服务器,完整经历了什么。**

现在我会更倾向于这样理解一个 FastAPI 请求:

```mermaid

flowchart LR

A[Client] --> B[Nginx]

B --> C[Uvicorn]

C --> D[ASGI]

D --> E[Middleware]

E --> F[Dependency Injection]

F --> G[Authentication]

G --> H[Authorization]

H --> I[Endpoint]

I --> J[Business Logic]

J --> K[Response]

K --> E

E --> B

B --> A

```

这张图对我来说,比单独记住几个 FastAPI 装饰器有意义得多。

---

十三、这一阶段真正学到的东西

如果把这一阶段压缩成几个关键词:

```text

asyncio

Event Loop

Coroutine / Task

Cancellation

Timeout

Semaphore / Queue

Backpressure

ASGI

Uvicorn

FastAPI

Middleware

Depends

Authentication / Authorization

```

它们并不是一堆孤立的知识点。

而是一条逐渐向上的抽象链:

```text

Python 异步执行模型

异步任务管理

Web Server

ASGI

Web Framework

请求生命周期

业务系统

```

---

十四、和 Agent 开发联系起来

这次学习还有一个比较明显的感受:

以前我学习 Agent 时,很容易把注意力放在:

  • Prompt
  • Tool Calling
  • RAG
  • Multi-Agent
  • LLM

这些“看起来很 AI”的东西上。

但真正做复杂 Agent 系统之后,会越来越发现:

> **Agent 最终还是一个软件系统。**

一个真实 Agent 可能同时涉及:

```text

LLM

Agent Loop

Tool

Async Task

Queue

Database

Message Broker

Audit

API

Authentication

```

所以很多 Agent 系统真正难的地方,并不是“让模型多聪明一点”。

而是:

> **当模型开始执行不确定的动作以后,整个 Runtime 怎么保证系统仍然可控。**

这也是我这段时间重新学习 Python 后端工程化的一个重要原因。

---

十五、下一阶段

目前 asyncio 这一部分基本已经形成了比较完整的知识体系。

FastAPI / ASGI 则还在继续。

接下来准备继续往下面几个方向推进:

1. FastAPI 异常处理

重点理解:

  • `HTTPException`
  • 全局异常处理器
  • 自定义 Exception
  • 统一错误响应
  • 异常和 Middleware 的关系

2. FastAPI Lifespan

进一步理解:

```text

Application Start

初始化资源

Application Running

处理请求

Application Shutdown

释放资源

```

以及数据库连接池、HTTP Client、模型、Agent Runtime 等资源到底应该在哪里初始化和释放。

3. BackgroundTask vs asyncio.create_task()

进一步搞清楚:

> Web 请求结束以后,任务到底应该交给谁管理?

这是一个看起来简单,但实际上非常容易踩坑的问题。

4. 高并发下的 FastAPI

最终希望把:

```text

asyncio

+

FastAPI

+

ASGI

+

Queue

+

Worker

+

Timeout

+

Cancellation

+

Backpressure

```

真正串成一个完整的后端系统模型。

---

最后

这阶段最大的收获不是:

> “我学会了 FastAPI。”

而是开始意识到:

> **写后端 API 和理解后端系统,是两件完全不同的事情。**

以前我可能会关注:

```python

@app.post("/xxx")

async def xxx():

...

```

现在会开始继续往下想:

```text

这个请求谁接收?

谁负责调度?

它运行在哪个 Event Loop?

Middleware 在哪里执行?

依赖什么时候解析?

任务取消怎么办?

超时怎么办?

资源什么时候初始化?

请求结束后任务还存在吗?

服务关闭时怎么清理?

```

感觉这才是从“会用框架”向“理解工程系统”迈过去的一步。

**下一阶段:继续深入 FastAPI 生命周期。**