Bài 4: Async endpoint — vì sao quan trọng cho AI inference

Sau bài này bạn sẽ:

• Hiểu event loop của ASGI hoạt động như thế nào và tại sao blocking là vấn đề nghiêm trọng.
• Biết FastAPI xử lý endpoint def (sync) khác endpoint async def ra sao.
• Phân biệt 3 loại workload AI — CPU-bound inference, I/O-bound API call, mixed pipeline — và chọn đúng pattern cho từng loại.
• Nhận ra 2 anti-pattern phổ biến và sửa được ngay.
• Biết cách cấu hình số worker phù hợp với từng loại workload.

FastAPI chạy trên Starlette, giao diện ASGI. Mỗi Uvicorn worker process có đúng 1 event loop chạy trên 1 thread chính. Event loop này là nơi xử lý tất cả coroutine async def.

async def endpoint

FastAPI gọi hàm trực tiếp trong event loop. Nếu hàm await một coroutine I/O (HTTP call, DB query), event loop tạm dừng coroutine đó và chuyển sang xử lý request khác — đây là điểm mạnh của async. Tuy nhiên, nếu hàm thực hiện tác vụ blocking mà không await gì, event loop bị treo và toàn bộ request khác phải chờ cho đến khi hàm đó kết thúc.

def endpoint (sync)

FastAPI không gọi hàm trong event loop. Thay vào đó, nó offload hàm vào ThreadPoolExecutor bằng asyncio.run_in_executor(None, func). Default pool của anyio có 40 thread. Event loop gọi await executor.run() và tiếp tục nhận request khác trong thời gian thread đang chạy. Khi thread xong, event loop nhận kết quả và trả response.

Ngắn gọn:

• async def → chạy trong event loop → KHÔNG được blocking.
• def → chạy trong thread pool → có thể blocking, FastAPI lo phần offload.

from fastapi import FastAPI

app = FastAPI()

# Chạy trong event loop — KHÔNG được blocking
@app.get("/async-ok")
async def async_endpoint():
    return {"status": "ok"}

# Chạy trong ThreadPoolExecutor — có thể blocking
@app.get("/sync-ok")
def sync_endpoint():
    return {"status": "ok"}

GIL (Global Interpreter Lock) là cơ chế trong CPython ngăn hai thread Python chạy bytecode đồng thời trên cùng 1 process. Điều này có nghĩa thread pool 40 thread không thể thực sự song song hóa pure Python code.

Tuy nhiên, GIL có một ngoại lệ quan trọng: khi một C extension giải phóng GIL, các thread Python khác có thể chạy trong thời gian đó. NumPy, PyTorch, scikit-learn đều giải phóng GIL khi thực hiện tính toán nặng trong C/CUDA. Điều này có nghĩa:

• torch.Tensor.matmul, model.forward(), sklearn.predict() — giải phóng GIL phần lớn thời gian tính toán. Thread pool vẫn có thể xử lý request khác trong thời gian đó.
• Pure Python code (vòng lặp, dict manipulation) — bị GIL giới hạn, các thread thực sự tuần tự.

Từ Python 3.13, GIL có thể tắt thông qua build flag --disable-gil (PEP 703), nhưng hầu hết production vẫn dùng CPython thông thường. Đừng tính vào đây khi thiết kế.

Kết luận thực tế: với PyTorch / NumPy inference, def endpoint + thread pool vẫn cho phép nhiều request xử lý gần đồng thời vì GIL được giải phóng trong phần tính toán nặng nhất.

CPU-bound inference

Ví dụ: PyTorch model forward pass, sklearn predict, OpenCV image processing. Tác vụ không chờ I/O, chỉ tính toán CPU/GPU. Phần lớn giải phóng GIL qua C extension.

Pattern phù hợp: def endpoint — FastAPI offload vào thread pool, không chiếm event loop.

I/O-bound — external API call

Ví dụ: gọi OpenAI Chat Completions API, Anthropic Claude API, query vector DB qua HTTP (Pinecone, Weaviate cloud). Tác vụ phần lớn là chờ network — CPU gần như idle trong thời gian đó.

Pattern phù hợp: async def + httpx.AsyncClient hoặc SDK async. Event loop không bị chiếm, xử lý song song nhiều request.

Mixed — RAG pipeline

Ví dụ: embedding văn bản (CPU) → query vector DB (HTTP I/O) → gọi LLM API (HTTP I/O). Có cả CPU-bound lẫn I/O-bound trong 1 request.

Pattern phù hợp: async def + asyncio.to_thread() cho phần CPU-bound, await cho phần I/O. Chi tiết ở mục 8.

Bảng quyết định nhanh:

Trường hợp nào tuyệt đối không làm: đặt code blocking (time.sleep, requests.post, model.forward không qua to_thread) bên trong async def mà không await.

Đây là lỗi phổ biến nhất khi AI Engineer chuyển từ Flask sang FastAPI. Vì Flask không có async nên khi sang FastAPI, người ta vẫn quen viết code blocking nhưng thêm từ khóa async vào trước def.

Lỗi sai

import torch
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

# model được load ở đâu đó (bài 5 sẽ dạy cách đúng)
model = torch.load("model.pt")
model.eval()

class PredictRequest(BaseModel):
    data: list[float]

# SAI: async def nhưng lại chạy blocking code bên trong
@app.post("/predict")
async def predict(req: PredictRequest):
    tensor = torch.tensor(req.data).unsqueeze(0)
    # model.forward() là blocking — chiếm event loop
    # Trong thời gian này, KHÔNG request nào khác được xử lý
    result = model(tensor)
    return {"output": result.tolist()}

Nếu model.forward() mất 200ms và có 50 request đồng thời, request cuối sẽ chờ 50 × 200ms = 10 giây trước khi được xử lý, dù server chỉ đang chạy 1 tác vụ.

Cách sửa — dùng asyncio.to_thread

import asyncio
import torch
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

model = torch.load("model.pt")
model.eval()

class PredictRequest(BaseModel):
    data: list[float]

def run_inference(data: list[float]) -> list:
    """Hàm sync thuần túy — an toàn để chạy trong thread."""
    tensor = torch.tensor(data).unsqueeze(0)
    with torch.no_grad():
        result = model(tensor)
    return result.squeeze().tolist()

@app.post("/predict")
async def predict(req: PredictRequest):
    # asyncio.to_thread offload run_inference sang thread pool
    # Event loop tự do nhận request khác trong thời gian này
    output = await asyncio.to_thread(run_inference, req.data)
    return {"output": output}

asyncio.to_thread có từ Python 3.9. Nó là shorthand cho loop.run_in_executor(None, func, *args). Với Python 3.8 trở xuống, dùng run_in_executor trực tiếp.

Cách thay thế — dùng def endpoint

# Đơn giản hơn: dùng def — FastAPI lo offload
@app.post("/predict-simple")
def predict_simple(req: PredictRequest):
    tensor = torch.tensor(req.data).unsqueeze(0)
    with torch.no_grad():
        result = model(tensor)
    return {"output": result.squeeze().tolist()}

Cả hai cách đều đúng. Dùng def đơn giản hơn khi endpoint chỉ có inference. Dùng async def + asyncio.to_thread khi endpoint còn có I/O async khác cần kết hợp.

Library requests là synchronous HTTP client. Dùng nó bên trong async def là blocking — đặc biệt nguy hiểm khi gọi LLM API có latency 1-10 giây.

Lỗi sai

import requests
from fastapi import FastAPI

app = FastAPI()
OPENAI_API_KEY = "sk-..."

# SAI: requests.post là blocking sync
@app.post("/chat")
async def chat(prompt: str):
    response = requests.post(
        "https://api.openai.com/v1/chat/completions",
        headers={"Authorization": f"Bearer {OPENAI_API_KEY}"},
        json={
            "model": "gpt-4o-mini",
            "messages": [{"role": "user", "content": prompt}]
        }
    )
    return response.json()

Cách sửa — dùng httpx.AsyncClient

import httpx
from fastapi import FastAPI

app = FastAPI()
OPENAI_API_KEY = "sk-..."

# Tạo client 1 lần, tái dùng cho mọi request (bài 5 sẽ dạy cách đúng qua lifespan)
http_client = httpx.AsyncClient(timeout=30.0)

@app.post("/chat")
async def chat(prompt: str):
    response = await http_client.post(
        "https://api.openai.com/v1/chat/completions",
        headers={"Authorization": f"Bearer {OPENAI_API_KEY}"},
        json={
            "model": "gpt-4o-mini",
            "messages": [{"role": "user", "content": prompt}]
        }
    )
    response.raise_for_status()
    return response.json()

httpx.AsyncClient là async HTTP client tương thích với requests API. Phần lớn SDK của OpenAI, Anthropic, Cohere đều có async client riêng — ưu tiên dùng SDK async thay vì gọi HTTP thủ công.

Demo tác động latency

Để thấy rõ sự khác biệt khi có 100 concurrent request gọi endpoint cần 500ms latency từ external API:

"""
Dùng requests (blocking trong async):
- 100 request gửi cùng lúc
- Event loop xử lý tuần tự từng cái
- Request cuối phải chờ ~100 × 500ms = 50 giây
- P99 latency ≈ 50s

Dùng httpx.AsyncClient (non-blocking):
- 100 request gửi cùng lúc
- Event loop await tất cả song song
- Tất cả hoàn thành sau ~500ms + overhead nhỏ
- P99 latency ≈ 1-2s (phụ thuộc rate limit API)
"""

Thử nghiệm thực tế có thể chạy với asyncio.gather:

import asyncio
import time
import httpx

async def call_endpoint(client: httpx.AsyncClient, i: int):
    start = time.monotonic()
    r = await client.post("http://localhost:8000/chat", json={"prompt": f"test {i}"})
    elapsed = time.monotonic() - start
    print(f"Request {i}: {elapsed:.2f}s")

async def main():
    async with httpx.AsyncClient() as client:
        tasks = [call_endpoint(client, i) for i in range(20)]
        await asyncio.gather(*tasks)

asyncio.run(main())

RAG pipeline điển hình gồm: embedding (CPU) → vector search (HTTP) → LLM call (HTTP). Đây là trường hợp mixed — cần kết hợp cả hai pattern.

import asyncio
import httpx
from fastapi import FastAPI
from pydantic import BaseModel
from sentence_transformers import SentenceTransformer

app = FastAPI()

# Embedding model — CPU-bound, sync
embed_model = SentenceTransformer("all-MiniLM-L6-v2")

# HTTP client — async
http_client = httpx.AsyncClient(timeout=30.0)

VECTOR_DB_URL = "https://your-vector-db.example.com"
LLM_API_URL = "https://api.openai.com/v1/chat/completions"
LLM_API_KEY = "sk-..."

class RAGRequest(BaseModel):
    query: str
    top_k: int = 3

def compute_embedding(text: str) -> list[float]:
    """Sync — sẽ được offload qua asyncio.to_thread."""
    return embed_model.encode(text).tolist()

async def search_vectors(embedding: list[float], top_k: int) -> list[str]:
    """Async HTTP — gọi vector DB."""
    resp = await http_client.post(
        f"{VECTOR_DB_URL}/search",
        json={"vector": embedding, "top_k": top_k}
    )
    resp.raise_for_status()
    return [doc["text"] for doc in resp.json()["results"]]

async def call_llm(query: str, context: list[str]) -> str:
    """Async HTTP — gọi LLM API."""
    context_str = "\n".join(context)
    messages = [
        {"role": "system", "content": f"Context:\n{context_str}"},
        {"role": "user", "content": query}
    ]
    resp = await http_client.post(
        LLM_API_URL,
        headers={"Authorization": f"Bearer {LLM_API_KEY}"},
        json={"model": "gpt-4o-mini", "messages": messages}
    )
    resp.raise_for_status()
    return resp.json()["choices"][0]["message"]["content"]

@app.post("/rag")
async def rag_endpoint(req: RAGRequest):
    # Bước 1: Embedding — CPU-bound, offload qua to_thread
    embedding = await asyncio.to_thread(compute_embedding, req.query)

    # Bước 2: Vector search — I/O async, await trực tiếp
    docs = await search_vectors(embedding, req.top_k)

    # Bước 3: LLM call — I/O async, await trực tiếp
    answer = await call_llm(req.query, docs)

    return {"answer": answer, "sources": docs}

Lưu ý: nếu bước 2 và bước 3 độc lập nhau (không phụ thuộc output của nhau), có thể chạy song song bằng asyncio.gather để giảm tổng latency.

Mỗi Uvicorn worker là 1 process Python riêng biệt, có event loop riêng và thread pool riêng. Các worker không chia sẻ memory. Điều này có nghĩa nếu load model vào memory, mỗi worker sẽ load 1 bản riêng.

# Chạy với 4 worker
uvicorn main:app --workers 4 --host 0.0.0.0 --port 8000

Thread pool của anyio (backend mặc định của FastAPI) có 40 thread per process. Có thể cấu hình:

import anyio
from fastapi import FastAPI

app = FastAPI()

# Đặt thread pool limit cho anyio — gọi trước khi server start
# Mặc định là 40
anyio.from_thread.start_blocking_portal()

# Hoặc qua biến môi trường khi chạy uvicorn:
# ANYIO_WORKER_THREADS=20 uvicorn main:app

Chọn số worker theo workload

Chủ yếu I/O async (gọi external API): 1-2 worker là đủ vì event loop xử lý được nhiều concurrent request. Tăng worker chỉ tốn RAM thêm mà không giúp được nhiều.

# Gọi OpenAI API, vector DB cloud — chủ yếu await
uvicorn main:app --workers 2

Chủ yếu CPU inference (torch model lớn): Tăng worker để tận dụng CPU cores. Nhưng chú ý RAM — mỗi worker load model riêng.

# PyTorch model, sklearn — CPU-bound
# N ≈ số CPU cores, trừ đi 1 cho OS
uvicorn main:app --workers 4  # máy 4 core

# Nếu model ăn 4GB RAM × 4 worker = 16GB — cần tính trước

GPU inference: Thường chỉ 1 worker vì GPU là shared resource. Nhiều worker cạnh tranh GPU gây context switching overhead. Giải pháp scale GPU là dùng batching (bài 51) hoặc nhiều máy.

# GPU inference — 1 worker, nhiều thread
uvicorn main:app --workers 1

Trước khi quyết định cấu hình, đo thực tế. Dùng wrk hoặc ApacheBench (ab).

wrk

# Cài wrk (macOS)
brew install wrk

# Test endpoint trong 10 giây, 4 thread, 100 concurrent connection
wrk -t4 -c100 -d10s http://localhost:8000/predict

# Test POST với body
wrk -t4 -c100 -d10s -s post.lua http://localhost:8000/predict

File post.lua cho wrk:

wrk.method = "POST"
wrk.body   = '{"data": [1.0, 2.0, 3.0, 4.0]}'
wrk.headers["Content-Type"] = "application/json"

Output của wrk:

Running 10s test @ http://localhost:8000/predict
  4 threads and 100 connections
  Thread Stats   Avg      Stdev     Max   +/- Stdev
    Latency    45.23ms   12.18ms 198.45ms   89.12%
    Req/Sec   543.21     67.34   712.00     74.25%
  21678 requests in 10.01s, 3.82MB read
Requests/sec: 2165.63
Transfer/sec: 390.82KB

ApacheBench (ab)

# Cài trên Ubuntu
sudo apt install apache2-utils

# 1000 request, concurrency 100
ab -n 1000 -c 100 -T application/json -p payload.json http://localhost:8000/predict

File payload.json:

{"data": [1.0, 2.0, 3.0, 4.0]}

Metrics cần theo dõi

• P50 (median latency): Phần lớn request có latency bao nhiêu.
• P95: 95% request hoàn thành trong thời gian này. Thực tế thường dùng P95 làm SLA.
• P99: Latency của 1% request chậm nhất — phát hiện outlier, timeout issues.
• Requests/sec: Throughput tổng của server.

Quy trình đo: chạy với def endpoint → ghi P50/P95/P99, sau đó thử async def + to_thread → so sánh. Nếu kết quả tương đương, chọn cách đơn giản hơn.

Flowchart quyết định nhanh:

Endpoint có gọi async library không?
├─ Có (httpx, asyncpg, SDK async) → dùng async def, await trực tiếp
└─ Không
    ├─ Chỉ chạy sync code (torch, sklearn) → dùng def
    └─ Có cả hai → dùng async def + asyncio.to_thread cho phần sync

Các điểm cần nhớ:

• async def không tự động làm hàm nhanh hơn — nó chỉ cho phép event loop xử lý request khác trong thời gian await.
• Blocking code trong async def mà không qua to_thread là lỗi nghiêm trọng hơn so với dùng def thuần.
• requests là sync — không dùng trong async def. Thay bằng httpx.AsyncClient hoặc SDK async của provider.
• Với GPU inference, 1 worker thường đủ. Scale theo chiều ngang (nhiều instance) thay vì nhiều worker trên 1 máy.
• asyncio.to_thread yêu cầu Python 3.9+. Python 3.8 dùng loop.run_in_executor(None, func, *args).

Bài 5: Load model 1 lần khi startup (lifespan event) — Cách dùng lifespan context manager của FastAPI để load model đúng 1 lần, tránh load lại mỗi request, và xử lý cleanup khi shutdown.