AI 写代码有一个很常见的问题：

代码能跑，但一到多人访问就变慢。

尤其在 Python Web 项目中，即使使用 FastAPI，函数也写成了 async def，数据库查询、HTTP 请求、大模型调用等操作仍然可能同步阻塞。

问题的关键在于：

async def 只是异步函数的入口，并不会自动把里面的同步代码变成异步代码。

对于 Web 服务、知识库和智能体平台这类大量依赖外部 I/O 的系统，建立统一的异步与并发架构，是提升系统吞吐能力的重要基础。

📌 技术名片

异步编程 / Asynchronous Programming

程序在等待数据库、网络、大模型等外部操作返回时，不占着执行线程干等，而是把执行机会让给其他任务。

与之相关的是：

并发 / Concurrency

在同一时间段内推进多个任务，而不是严格地一个做完再做下一个。

需要注意：

  • Concurrency：并发——多个任务交替推进；
  • Parallelism：并行——多个任务真正同时执行。

Web 后端首先关注的通常是并发。

💡 一个简单的比喻

把服务器想象成餐厅服务员。

同步模式下，服务员把 A 的菜单送进厨房后，就站在那里等菜做好：

A 点菜
  ↓
等待厨房……
  ↓
菜做好
  ↓
再服务 B

异步模式则是：

A 点菜 → 厨房处理
            ↓
服务员 → 服务 B
       → 服务 C
       → 服务 D
            ↓
A 的菜好了 → 回来处理 A

服务员没有变多，只是：

等待的时候不再傻站着。

这就是异步编程的核心思想。

一、AI 最容易犯的错误：表面异步，实际阻塞

例如 AI 很容易生成：

async def handle_request():
    result = requests.get(url)
    return result.text

看起来使用了：

async def

但 requests.get() 仍然是同步阻塞调用。

类似的问题还有：

time.sleep(5)

出现在异步函数中。

应该使用支持异步的实现，例如：

async with httpx.AsyncClient() as client:
    response = await client.get(url)

以及：

await asyncio.sleep(5)

所以判断代码是否真正异步，不能只看有没有 async def，还要看：

整个调用链有没有混入阻塞操作。

二、架构规范

1. 异步必须贯穿调用链

真正可靠的异步架构应该是：

flowchart TD A[FastAPI API] --> B[Async Service] B --> C[Async Repository] C --> D[AsyncSession] D --> E[Async Database Driver]

涉及 I/O 的层级尽量保持异步。

所谓 I/O（Input/Output，输入/输出）操作，主要包括：

  • 数据库访问；
  • HTTP 请求；
  • 大模型调用；
  • Redis 访问；
  • 文件读写；
  • Web Search；
  • 向量数据库访问。

这些操作最大的成本往往不是 CPU 计算，而是等待外部结果。

因此原则很简单：

I/O 密集型操作优先采用异步接口。

2. 能并发的任务，不要串行等待

异步并不自动等于并发。

例如：

user = await get_user()
kb = await search_kb()
web = await search_web()

虽然三个函数都是异步的，但仍然按照顺序执行。

如果三个任务互不依赖，可以：

user, kb, web = await asyncio.gather(
    get_user(),
    search_kb(),
    search_web(),
)

假设三个任务分别需要：

用户信息      0.4 秒
知识库检索    1.2 秒
Web Search   1.5 秒

串行执行理论上约为：

0.4 + 1.2 + 1.5 = 3.1 秒

并发执行则更接近最慢任务：

≈ 1.5 秒

但并发也不能滥用。

例如：

创建订单
 ↓
扣减库存
 ↓
生成付款记录

后一步依赖前一步，就不能简单放进 asyncio.gather()。

因此可以记住一句话：

无依赖任务考虑并发，有依赖任务保持顺序。

3. CPU 密集任务不要阻塞 Event Loop

异步主要解决的是 I/O 等待问题。

对于：

  • OCR；
  • 图片处理；
  • 视频转码；
  • PDF 重计算；
  • 大规模数据计算；
  • 本地模型推理；

这类任务通常属于：

CPU-bound Task：CPU 密集型任务

它们不是在“等待别人”，而是真的持续占用 CPU。

如果直接写：

async def endpoint():
    result = heavy_cpu_task()

即使函数是 async，仍然可能堵塞：

Event Loop：事件循环

因此通常应该把重计算移到：

Thread Pool
线程池

Process Pool
进程池

Background Worker
后台工作进程

可以简单概括为：

I/O 密集
   ↓
async / await

CPU 密集
   ↓
Thread / Process / Worker

3. 异步资源应该统一管理

数据库 Engine、SessionFactory、HTTP Client 等资源，不应该在每个业务函数里重复创建。

更加合理的架构是：

Application Startup
        ↓
ServiceContainer
        ↓
Async Engine / Client
        ↓
Repository
        ↓
Service

统一管理这些资源，可以减少重复创建连接，也让初始化、关闭和异常处理更加清晰。

三、异步架构到底带来了什么？

最重要的并不是让单个数据库查询突然变快。

假设数据库查询本身需要 500ms：

同步：500ms
异步：可能仍然是 500ms

区别在于等待这 500ms 的时候服务器能不能继续处理其他请求。

同步：

Request A
████████████

Request B
            ████████████

异步：

Request A
████────████

Request B
  ███────████

Request C
    ███────████

因此异步架构重点提升的是：

Throughput：吞吐量，即单位时间内系统能够处理的请求数量。

这对于 AI 系统尤其重要，因为一次 Agent 请求可能涉及：

数据库
 ↓
LLM
 ↓
知识库
 ↓
Embedding
 ↓
Reranker
 ↓
Web Search
 ↓
工具调用
 ↓
再次调用 LLM

其中大量时间都花在网络和外部服务等待上。

所以 AI 应用天然非常适合异步架构。

四、提示词落地：直接约束 AI

与其每次人工检查，不如把异步规范写进 Project Rules（项目级规则）：

## 异步与并发规则

本项目对 I/O 密集型操作采用异步优先架构。

1. 执行 I/O 操作的 FastAPI 路由应使用 async def。
2. 数据库访问必须使用 SQLAlchemy AsyncSession。
3. 存储库数据库方法必须是异步的。
4. 切勿在异步函数内部直接调用阻塞式 I/O。

避免使用：
- requests
- time.sleep
- 同步数据库会话

推荐使用：
- httpx.AsyncClient
- asyncio.sleep
- AsyncSession

5. 独立的 I/O 任务可以使用 asyncio.gather。
6. 不要并行处理具有数据依赖性或事务顺序要求的操作或对 CPU 密集型操作有阻塞。
7. CPU 密集型操作不得阻塞事件循环。在适当情况下，请使用线程池、进程池或后台工作线程。
8. 保持异步调用链：API→ 服务→ 仓库→ 异步驱动程序
9. 生成代码之前，检查所有被调用的库是否为同步库，以及是否可能阻塞事件循环。

这比简单告诉 AI：

“帮我优化一下性能。”

要明确得多。

五、正面产出：miniagent 的异步架构

miniagent 是一个基于 FastAPI、SQLAlchemy、知识库检索和 Agent Runtime 构建的智能体平台。

它的数据库访问并不是只在 API 层使用几个 async def，而是形成了完整的异步数据访问链路。

1. 统一创建 Async Engine

在 miniagent 的 ServiceContainer 中：

from sqlalchemy.ext.asyncio import (
    create_async_engine,
    async_sessionmaker,
    AsyncSession,
)

database_url = f"sqlite+aiosqlite:///{db_path}"

self.engine = create_async_engine(
    database_url,
    echo=False,
    future=True,
)

self.session_factory = async_sessionmaker(
    bind=self.engine,
    class_=AsyncSession,
    autoflush=False,
    autocommit=False,
    expire_on_commit=False,
)

这里形成的是：

FastAPI
   ↓
SQLAlchemy Async Engine
   ↓
AsyncSession
   ↓
aiosqlite
   ↓
SQLite

而不是“FastAPI 是异步的，底层数据库却仍然同步访问”。

2. Repository 统一异步化

miniagent 的数据访问层包含：

async_agent.py
async_chat.py
async_chunk.py
async_document.py
async_embedding.py
async_knowledge_base.py
async_llm.py
...

也就是说，异步被作为 Repository（仓储/数据访问层）的统一设计方式，而不是某个接口的局部优化。

3. Session 生命周期也是异步的

miniagent 定义了统一的 AsyncBaseDatabase：

@asynccontextmanager
async def get_session(self):
    async with self.AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except SQLAlchemyError:
            await session.rollback()
            raise
        finally:
            await session.close()

数据库 Session 的：

创建
提交
回滚
关闭

全部被统一纳入异步生命周期管理。

4. 数据库操作真正使用 await

例如 miniagent 的聊天数据访问：

async def get_user_session(
    self,
    session_id: int,
    user_id: int,
):
    async with self.get_session() as session:
        stmt = (
            select(ChatSession)
            .where(
                ChatSession.id == session_id,
                ChatSession.user_id == user_id,
            )
        )

        return (
            await session.execute(stmt)
        ).scalar_one_or_none()

真正重要的不是函数前面的：

async def

而是数据库操作本身：

await session.execute(stmt)

这才让数据库等待期间，事件循环有机会继续处理其他任务。

5. 最终形成完整的异步链路

miniagent 的主要异步链路如下图：

flowchart TD A[HTTP Request] --> B[FastAPI] B --> C[Async API] C --> D[Service] D --> E[Async Repository] E --> F[AsyncBaseDatabase] F --> G[SQLAlchemy AsyncSession] G --> H[aiosqlite] H --> I[SQLite] G -. Waiting .-> J[Event Loop] J --> K[Other Requests]

当一个请求等待数据库时，执行权可以返回 Event Loop（事件循环），服务器继续推进其他请求。

这才是异步架构提高系统吞吐能力的核心。

下图展示了更多异步链路细节：

flowchart TD U["User / Client"] --> API["FastAPI
Async API"] API --> EL["Event Loop
事件循环"] EL --> SVC["Service Layer
异步服务层"] SVC --> AGENT["Agent Runtime"] SVC --> REPO["Async Repository"] SVC --> KB["Knowledge Base"] SVC --> WEB["Web Search / HTTP"] SVC --> LLM["LLM"] REPO --> BASE["AsyncBaseDatabase"] BASE --> SESSION["SQLAlchemy AsyncSession"] SESSION --> DRIVER["aiosqlite"] DRIVER --> DB[("SQLite")] AGENT -. "await" .-> EL SESSION -. "await" .-> EL KB -. "await" .-> EL WEB -. "await" .-> EL LLM -. "await" .-> EL EL --> OTHER["Other Requests
继续处理其他请求"] SVC --> CPU["CPU-heavy Tasks
OCR / PDF / Heavy Compute"] CPU --> WORKER["Thread / Process / Worker
移出 Event Loop"]

六、给 AI 一份简单的检查清单

让 AI 完成后端代码后，可以要求它自行检查：

Async Architecture Checklist

□ I/O 操作是否使用异步 API？
□ 数据库是否使用 AsyncSession？
□ async 函数内部是否调用了同步阻塞库？
□ 独立 I/O 任务是否可以并发？
□ 有依赖关系的任务是否错误并发？
□ CPU 重任务是否阻塞 Event Loop？
□ 是否保持：

API
→ Service
→ Repository
→ Async Driver

完整异步链路？

总结

AI 很容易写出：

调用 A
 ↓
等待
 ↓
调用 B
 ↓
等待
 ↓
调用 C

这种代码对于 Demo 没什么问题，但进入真实并发环境后，很容易成为系统瓶颈。

因此应该提前建立几条明确规则：

I/O 默认异步
      ↓
异步贯穿调用链
      ↓
独立任务考虑并发
      ↓
CPU 重任务移出 Event Loop
      ↓
异步资源统一管理

像 miniagent 这样的项目，通过 AsyncSession、异步 Repository、统一 Session 管理以及 FastAPI 异步调用链，把异步能力放进基础架构，而不是散落在几个接口里。

对于 AI 编程，我们最终要约束的，不只是：

“这段代码能不能运行？”

还要多问一句：

“它会不会在等待的时候，把整个服务器一起堵住？”

这正是异步与并发架构存在的价值。

开源代码

🪐祝您好运🪐