FastAPI 并发实战：进程、线程、协程在一个框架里的分工

你的路径函数 / Agent 编排
          ▲
FastAPI   │ 依赖图、参数提取、OpenAPI、请求/响应模型
          ├──────────────> Pydantic Core（Rust：校验与序列化核心）
          ▲
Starlette │ 路由、中间件、Request/Response、WebSocket
          ▲
   ASGI   │ scope + receive + send 消息协议
          ▲
Uvicorn   │ 连接管理、协议适配、流控、创建请求 Task
          ├─ HTTP：httptools（C 解析器）/ h11（Python 回退）
          └─ Loop：uvloop（Cython + libuv）/ asyncio（标准回退）
                         ▲
                       libuv（C）
                         ▲
Linux epoll / macOS kqueue / 其他平台对应的内核机制
                         ▲
                   非阻塞 socket

python -m pip install "uvicorn[standard]"
uvicorn app:app --loop auto --http auto

“这 10000 个 socket 里，哪个能读/能写了再通知我。”

协程执行 await socket.read()
  -> 当前 Task 挂起
  -> socket fd 注册到事件循环
  -> 线程进入 kqueue/epoll 等待
  -> 内核发现 fd 可读
  -> 事件循环把对应 Task 放回 ready queue
  -> 协程从 await 后面继续

await client.get(url)

asyncio API -> uvloop -> libuv -> kqueue -> macOS 内核

POST /agents/run HTTP/1.1\r\n
Host: example.com\r\n
Content-Type: application/json\r\n
Content-Length: 37\r\n
\r\n
{"message":"帮我查一下 FastAPI"}

async def app(scope, receive, send):
    request = await receive()
    assert request["type"] == "http.request"

    await send({
        "type": "http.response.start",
        "status": 200,
        "headers": [(b"content-type", b"text/plain")],
    })
    await send({
        "type": "http.response.body",
        "body": b"hello",
    })

Uvicorn
  -> 错误处理中间件
    -> CORS / 鉴权 / 日志中间件
      -> Router
        -> Endpoint

1. 网卡收到数据，内核 TCP 栈把字节放进 socket 接收缓冲区
2. kqueue/epoll 标记这个 socket 可读，唤醒正在等待的事件循环
3. libuv/uvloop 触发对应 transport/protocol 回调
4. httptools 增量解析请求行、Header 和 Body 边界
5. Uvicorn 组装 ASGI scope 和 http.request 消息，创建请求 Task
6. Starlette 依次经过中间件和 Router，找到目标 Endpoint
7. FastAPI 解析依赖和参数，Pydantic Core 校验输入
8. 路径函数开始执行
9. 路径函数 await 数据库、LLM 或 RAG，当前 Task 挂起，线程回到第 2 步
10. 下游 fd 就绪，Task 恢复；响应模型序列化后沿 ASGI send 原路写回 socket

时间 ──────────────────────────────────────────────>
请求 A  执行 ── await LLM ────────────────── 恢复 ── 返回
请求 B           执行 ── await DB ── 恢复 ── 返回
请求 C                    执行 ── await RAG ───── 恢复
线程    [A]      [B]      [C]          [B]          [A][C]

def 路径函数       -> 扔进线程池执行（不占用事件循环）
async def 路径函数 -> 直接在事件循环里 await（协程）

import asyncio
import time
from fastapi import FastAPI

app = FastAPI()

@app.get("/async-right")
async def async_right():
    """正确：async def + 异步等待"""
    await asyncio.sleep(0.5)
    return {"ok": True}

@app.get("/async-wrong")
async def async_wrong():
    """错误：async def 里调阻塞函数"""
    time.sleep(0.5)   # 拖死整个事件循环
    return {"ok": True}

@app.get("/def-blocking")
def def_blocking():
    """正确：def 路径函数，FastAPI 自动扔进线程池"""
    time.sleep(0.5)
    return {"ok": True}

/async-wrong   : 0.51s    <- 罪魁祸首
/async-right   : 1.01s    <- 无辜接口被拖到两倍耗时
/async-right   : 1.01s
/async-right   : 1.01s
/async-right   : 1.01s

40 并发: 0.59s   <- 刚好一批
50 并发: 1.04s   <- 40 个先跑，剩下 10 个排队等第二批

路径函数里要做什么？

├─ 有 IO，且用的是异步库（httpx、asyncpg、aiomysql、redis.asyncio）
│    └─ async def，一路 await 到底
│
├─ 有 IO，但库只有同步版（requests、psycopg2、部分云厂商 SDK）
│    └─ 用 def（FastAPI 自动进线程池）
│       或 async def + await asyncio.to_thread(阻塞函数)
│
├─ 纯 CPU 计算
│    ├─ 几毫秒内能算完：async def 直接写，别折腾
│    └─ 算得久：谁都不合适，交给任务队列或多进程
│
└─ 没有任何 IO，只是读内存/拼数据
     └─ async def，零开销

async with aiohttp.ClientSession() as session:
    for query in queries:
        async with session.post(
            FSRAG_API_URL,
            json=build_payload(query),
            timeout=aiohttp.ClientTimeout(total=30),
        ) as resp:
            data = await resp.json()

query 1:  ├──── 8s ────┤
query 2:                 ├──── 9s ─────┤
query 3:                                ├──── 7s ───┤
当前请求: ├──────────────── 24s ────────────────────┤

T_serial ≈ t1 + t2 + ... + tN
最坏预算 ≈ N × 30s

import asyncio
import json

import aiohttp

PER_QUERY_TIMEOUT = aiohttp.ClientTimeout(
    total=20,
    connect=3,
    sock_read=15,
)
BATCH_TIMEOUT_SECONDS = 50


async def _call_fsrag(
    dataset_id: str,
    messages: list[str],
    rag_key: str,
    session: aiohttp.ClientSession,
    slots: asyncio.Semaphore,
) -> str:
    queries = [m.strip() for m in messages if m and m.strip()]
    headers = {
        "Authorization": f"Bearer {rag_key}",
        "Content-Type": "application/json",
    }

    async def fetch_one(query: str) -> list[dict]:
        payload = {
            "req": {
                "question": query,
                "dataset_ids": [dataset_id],
                "document_ids": [],
                "top_enhance": False,
                "rerank_id": "cohere.rerank-v3-5:0",
                "highlight": True,
            }
        }

        try:
            # slots 在整个 worker 内共享，避免多个 HTTP 请求一起压垮 RAG。
            async with slots:
                async with session.post(
                    FSRAG_API_URL,
                    headers=headers,
                    json=payload,
                    timeout=PER_QUERY_TIMEOUT,
                ) as resp:
                    if resp.status >= 400:
                        logger.error(
                            "FSRAG HTTP error: status=%s query=%s",
                            resp.status,
                            query,
                        )
                        return []

                    data = await resp.json()
        except asyncio.TimeoutError:
            logger.warning("FSRAG query timeout: query=%s", query)
            return []
        except aiohttp.ClientError as exc:
            logger.error("FSRAG request failed: query=%s error=%s", query, exc)
            return []

        if data.get("code") != 10000:
            logger.warning("FSRAG bad response: code=%s", data.get("code"))
            return []

        return data.get("data", {}).get("chunks", [])

    try:
        # gather 才真正让多条 query 同时推进；总 deadline 兜住整个批次。
        async with asyncio.timeout(BATCH_TIMEOUT_SECONDS):
            groups = await asyncio.gather(
                *(fetch_one(query) for query in queries)
            )
    except TimeoutError:
        logger.error("FSRAG batch timeout: queries=%d", len(queries))
        return json.dumps(
            {"success": False, "error": "FSRAG batch timeout"},
            ensure_ascii=False,
        )

    # gather 保持输入顺序。并发任务只返回结果，统一在这里去重，
    # 不让多个任务同时修改 seen/context 这两个共享容器。
    context: list[dict] = []
    seen: set[str] = set()

    for chunks in groups:
        for chunk in chunks:
            content = chunk.get("content", "").strip()
            if not content or content in seen:
                continue
            seen.add(content)
            context.append(
                {
                    "content": content,
                    "recommend_url": chunk.get("document_keyword", ""),
                    "similarity": chunk.get("similarity", 0),
                }
            )

    return json.dumps(
        {"success": True, "context": context},
        ensure_ascii=False,
    )

T_concurrent ≈ ceil(N / C) × 每批最慢请求耗时

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

def bg_work(name: str):
    time.sleep(0.3)
    print(f"[后台任务] {name} 完成", flush=True)

@app.post("/notify")
def notify(background_tasks: BackgroundTasks):
    background_tasks.add_task(bg_work, "任务A")
    background_tasks.add_task(bg_work, "任务B")
    return {"status": "已受理"}

# FastAPI CLI，底层仍然是 Uvicorn
fastapi run app.py --workers 4

# 或者直接使用 Uvicorn
uvicorn app:app --workers 4

gunicorn app:app -w 4 -k uvicorn_worker.UvicornWorker

import os

counter = 0

@app.get("/count")
async def count():
    global counter
    counter += 1
    return {"pid": os.getpid(), "counter": counter}

pid 6297: 处理了 2 个请求,  计数器=[1, 2]
pid 6298: 处理了 18 个请求, 计数器=[7, 8, ..., 24]

                nginx（反向代理）
                    │
          ┌─────────┴─────────┐
          ▼                   ▼
    worker 进程 1        worker 进程 2        <- 进程：吃满多核 CPU
    ┌──────────────┐     ┌──────────────┐
    │  事件循环      │     │  事件循环      │  <- 协程：单线程挂成千上万连接
    │  协程 协程 协程 │     │  协程 协程 协程 │
    │  线程池 x40   │     │  线程池 x40   │  <- 线程：跑 def 函数和同步库
    └──────────────┘     └──────────────┘

用户请求
  -> 等 LLM 首轮推理
  -> 等向量库召回
  -> 等一个或多个 HTTP / 数据库工具
  -> 再等 LLM 汇总
  -> 持续把 token / tool event 推给前端

import asyncio
import json
import os
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

import httpx
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field

MAX_AGENT_RUNS = 20
TOTAL_TIMEOUT_SECONDS = 60


class AgentInput(BaseModel):
    session_id: str = Field(min_length=1, max_length=128)
    message: str = Field(min_length=1, max_length=20_000)


@asynccontextmanager
async def lifespan(app: FastAPI):
    # 一个进程只创建一个客户端，复用连接池；不要每个请求 new 一个。
    async with httpx.AsyncClient(
        base_url=os.environ["AGENT_RUNTIME_URL"],
        timeout=httpx.Timeout(55.0, connect=5.0),
        limits=httpx.Limits(
            max_connections=100,
            max_keepalive_connections=20,
        ),
    ) as client:
        app.state.agent_http = client
        # Semaphore 是进程内限制：4 个 worker 的总上限是 4 × 20。
        app.state.agent_slots = asyncio.Semaphore(MAX_AGENT_RUNS)
        yield


app = FastAPI(lifespan=lifespan)


@app.post("/agents/run")
async def run_agent(body: AgentInput, request: Request):
    client: httpx.AsyncClient = request.app.state.agent_http
    slots: asyncio.Semaphore = request.app.state.agent_slots

    try:
        # 总超时同时覆盖排队时间和真正执行时间。
        async with asyncio.timeout(TOTAL_TIMEOUT_SECONDS):
            async with slots:
                response = await client.post(
                    "/v1/runs",
                    json=body.model_dump(),
                )
                response.raise_for_status()
                return response.json()
    except TimeoutError as exc:
        raise HTTPException(504, "agent run timed out") from exc
    except httpx.HTTPError as exc:
        raise HTTPException(502, "agent runtime unavailable") from exc


def sse_error(code: str) -> bytes:
    data = json.dumps({"type": "error", "code": code})
    return f"event: error\ndata: {data}\n\n".encode()


@app.post("/agents/run/stream")
async def stream_agent(body: AgentInput, request: Request):
    client: httpx.AsyncClient = request.app.state.agent_http
    slots: asyncio.Semaphore = request.app.state.agent_slots

    async def events() -> AsyncIterator[bytes]:
        try:
            async with asyncio.timeout(TOTAL_TIMEOUT_SECONDS):
                async with slots:
                    # 上游返回已经按 SSE 格式编码的数据，这里直接透传。
                    async with client.stream(
                        "POST",
                        "/v1/runs/stream",
                        json=body.model_dump(),
                    ) as upstream:
                        upstream.raise_for_status()
                        async for chunk in upstream.aiter_bytes():
                            if await request.is_disconnected():
                                break
                            yield chunk
        except TimeoutError:
            # 流开始后不能再把 HTTP 状态码改成 504，只能发错误事件。
            yield sse_error("timeout")
        except httpx.HTTPError:
            yield sse_error("upstream_unavailable")

    return StreamingResponse(
        events(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",
        },
    )

FastAPI API 层
  ├─ 鉴权、校验、限流、SSE / WebSocket
  ├─ Redis / PostgreSQL：会话、记忆、运行状态
  ├─ 任务队列：长任务、重试、恢复
  └─ Agent / 模型运行时：独立进程或独立服务