FastAPI 零基础手把手｜手把手教程

怎么用

每一步都产出一个可运行文件。不要只看概念，至少跑通 /health、/chat、/chat/stream 三个接口。

Step 1 · 建项目骨架

mkdir fastapi-ai-service && cd fastapi-ai-service
python -m venv .venv
source .venv/bin/activate
pip install fastapi uvicorn pydantic-settings httpx pytest pytest-asyncio
mkdir app tests
touch app/main.py app/settings.py tests/test_health.py

验收：uvicorn app.main:app --reload 能启动。

Step 2 · 健康检查与配置

app/settings.py：

from pydantic_settings import BaseSettings
 
class Settings(BaseSettings):
    app_name: str = "fastapi-ai-service"
    llm_timeout: float = 20.0
 
settings = Settings()

app/main.py：

from fastapi import FastAPI
from app.settings import settings
 
app = FastAPI(title=settings.app_name)
 
@app.get("/health")
def health():
    return {"ok": True, "app": settings.app_name}

验收：curl http://127.0.0.1:8000/health 返回 ok: true。

Step 3 · 请求与响应模型

加入 Pydantic schema，明确输入输出边界。面试里要能说清：API 合同先于业务实现。

from pydantic import BaseModel, Field
 
class ChatRequest(BaseModel):
    query: str = Field(min_length=1, max_length=2000)
 
class ChatResponse(BaseModel):
    answer: str
    trace_id: str

Step 4 · 模拟 LLM 调用

先不要急着接真实模型。写一个 fake_llm()，把超时、异常、日志结构先搭好。

import uuid
from fastapi import HTTPException
 
async def fake_llm(query: str) -> str:
    if "timeout" in query:
        raise TimeoutError("llm timeout")
    return f"收到：{query}"
 
@app.post("/chat", response_model=ChatResponse)
async def chat(req: ChatRequest):
    trace_id = str(uuid.uuid4())
    try:
        answer = await fake_llm(req.query)
    except TimeoutError as exc:
        raise HTTPException(status_code=504, detail={"code": "LLM_TIMEOUT", "trace_id": trace_id}) from exc
    return ChatResponse(answer=answer, trace_id=trace_id)

Step 5 · 流式输出

import asyncio
from fastapi.responses import StreamingResponse
 
async def stream_tokens(text: str):
    for token in text:
        yield f"data: {token}\n\n"
        await asyncio.sleep(0.02)
    yield "event: done\ndata: [DONE]\n\n"
 
@app.post("/chat/stream")
async def chat_stream(req: ChatRequest):
    return StreamingResponse(stream_tokens(req.query), media_type="text/event-stream")

验收：浏览器或 curl 能看到分段返回。

Step 6 · 测试

from fastapi.testclient import TestClient
from app.main import app
 
client = TestClient(app)
 
def test_health():
    res = client.get("/health")
    assert res.status_code == 200
    assert res.json()["ok"] is True

Step 7 · Docker

写最小 Dockerfile，能被 Compose 或 Vercel 外的服务器复用。

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY app ./app
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

Step 8 · 面试复述

用 3 分钟讲清：

为什么 API 层先定义 schema。
流式输出如何减少感知延迟。
超时、错误码、trace_id 如何帮助排障。
Docker 如何保证别人一条命令启动。

怎么用