03 · 后端骨架

⏱ 40 分钟 ｜ ⭐ 这一章的每个零件，都是后面几章往上挂东西的挂载点

🎯 一句话

后端不是「给模型调用套一层 HTTP」，而是立四根柱子：边界在哪、等待怎么处理、每个请求都要做的事挂在哪、出错时对外长什么样。 柱子立歪了，第 8 章的认证、第 12 章的限流、第 15 章的可观测性就没地方生根，只能一路 if 补下去。

🧩 一、为什么中间必须有个后端

「浏览器直接调模型 API」在 Demo 里能跑。上不了线的理由有四条，每条都足够致命：

⭐ 模型 API 是「谁拿到密钥谁就是老板」的接口，产品需要的是「谁是谁、能做什么、做过什么」。这两者之间的翻译层就是你的后端。

🧱 二、最小骨架：两个模型和一个路由

形状是一个入口 + 一份「进来的长什么样」+ 一份「出去的长什么样」，任何框架都有这三样，只是叫法不同。

# 最小骨架：一对请求/响应模型 + 一个路由
from fastapi import FastAPI
from pydantic import BaseModel, Field

app = FastAPI()


class ChatIn(BaseModel):          # ⭐ 边界合同：不合规的请求进不到业务代码
    message: str = Field(min_length=1, max_length=4000)      # 卡住【输入】的花费
    max_tokens: int = Field(default=512, ge=1, le=4096)      # ⭐ 卡住【输出】的花费，更贵的是这头
    temperature: float = Field(default=0.0, ge=0.0, le=2.0)


class ChatOut(BaseModel):         # ⭐ 同时是裁剪器：没写进来的字段不会被吐出去
    reply: str
    tokens: int


@app.post("/chat", response_model=ChatOut)
def chat(body: ChatIn) -> ChatOut:
    reply = "你说的是：" + body.message      # 下一章换成真的模型调用，max_tokens 原样传给上游
    return ChatOut(reply=reply, tokens=len(reply))


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app)              # 空 message / max_tokens=99999 / temperature=9 都被挡在 422

为什么值得单独定义这两个模型：① max_length=4000 这类限制只写一处，而成本变量在请求发出前能卡住两个：输入长度和输出上限（max_tokens）；② 非法请求进业务之前就被 422 挡掉，业务函数拿到的对象一定合法；③ 响应模型是个裁剪器——内部对象上挂的 raw_prompt 只要没写进 ChatOut 就漏不出去，比「记得删掉」可靠。

⭐ 两个里更该先卡的是输出：按 04 章那张价格表，输出单价是输入的 3～4 倍（🗓️ 占位数字，量级上普遍如此）。 ⚠️ 而 max_tokens 特别容易被漏掉——它常常有个很大的默认值，message 再短也拦不住一次跑满上限的回答。

⚡ 三、async 到底在解什么

一次模型调用要 3 秒。这 3 秒里服务器在干什么？什么也没干——它在等网络，CPU 完全空着。

⭐⭐ 普通 Web 应用的瓶颈通常是「算不完」，AI 应用的瓶颈是「等不完」。 等待型负载正是 async 的主场，所以 AI 应用比一般 Web 应用更需要 async。

同步模型下一个工作进程被这 3 秒完全占住，8 个进程就只能同时服务 8 个人；async 下等待期间控制权交回事件循环，一个进程能同时挂着上千个「正在等模型」的请求。

⚠️ 但 async 函数里一个同步阻塞调用，会卡死整个进程

# 一段能自己跑出结论的对照：async 里写同步阻塞调用，会把整个事件循环卡死。
import asyncio
import time

N, WAIT = 8, 0.5          # 8 个并发请求，每个"等模型"0.5 秒


async def good(i):
    await asyncio.sleep(WAIT)     # ⭐ 等的时候把控制权交回事件循环，别人能跑
    return i


async def bad(i):
    time.sleep(WAIT)              # 💀 同步阻塞：这 0.5 秒里整个进程什么都干不了
    return i


async def main():
    for name, fn in (("await asyncio.sleep", good), ("time.sleep", bad)):
        t0 = time.perf_counter()
        await asyncio.gather(*(fn(i) for i in range(N)))
        print("%-22s %d 个请求耗时 %.2f 秒" % (name, N, time.perf_counter() - t0))


asyncio.run(main())

实跑结果：await asyncio.sleep 版 0.51 秒，time.sleep 版 4.01 秒——8 个请求被排成了一队。⚠️ bad() 长得完全像个正常协程，没有语法错误、没有任何警告，而且本地单人开发时一切正常，上线并发一起来才一起爆炸。

真实代码里的 time.sleep 伪装成：requests.post(...)（换 httpx.AsyncClient）、同步数据库 / Redis 驱动（换 async 驱动或走线程池）、只有同步版的 SDK（当阻塞处理）。另有一类是真 CPU 活——解析 50 MB JSON、本地算 embedding——async 救不了，只能扔线程池或独立进程。

⭐ FastAPI 的逃生舱：路由写成普通 def（不是 async def）时，它会自动把这个函数扔进线程池。本章示例用的都是 def。🗓️ 线程数有限且随版本变，按官方文档核对。

⭐ 判据背下来：async def 里只能写「会 await 的等待」。拿不准某个库阻不阻塞，就写成 def 交给线程池——慢一点，但不会全站一起躺下。

🔌 四、依赖注入不是花架子

它解决一个很具体的问题：有些事每个请求都要做，却不属于任何一个业务函数——鉴权、限流、拿连接、生成 trace_id、识别租户。写进业务函数就是十个路由十份复制粘贴，改一次漏三处。

# 依赖注入：把"每个请求都要做的事"挪出业务函数，变成可替换的挂载点
from typing import Annotated
from fastapi import Depends, FastAPI, Header, HTTPException

app = FastAPI()


def current_user(authorization: Annotated[str, Header()] = ""):
    if not authorization.startswith("Bearer "):
        raise HTTPException(401, "缺少 Bearer token")
    return {"uid": authorization[7:]}             # 第 8 章换成真校验


def quota(user: Annotated[dict, Depends(current_user)]):
    if user["uid"] == "u_banned":                 # ⭐ 依赖可以套依赖
        raise HTTPException(429, "免费额度已用完")
    return user


@app.post("/chat")
def chat(user: Annotated[dict, Depends(quota)]):
    return {"uid": user["uid"], "reply": "..."}   # ⭐ 业务函数只剩业务


# ⭐ 测试里一行把整条鉴权链换掉，不用起认证服务、不用造 token
# app.dependency_overrides[quota] = lambda: {"uid": "u_test"}

⭐ 最后那行注释和第 4 章「把模型客户端换成假客户端」是同一个思路——「能被换掉」是这个骨架最值钱的性质。

这两个依赖现在是空壳，但是已经焊好的挂载点：current_user 留给第 8 章（真 token 校验、多租户隔离），quota 留给第 12 章（限流与配额扣减），再加一个 trace 就是第 15 章（trace_id 串起日志与指标）的位置。没有这层，那三章只能塞进每个路由函数的开头。

🚨 五、错误处理：对外一种形状，对内保留细节

模型应用比普通 CRUD 多一大类错误：上游 429、上游超时、内容被拒、输出校验失败。每个路由自己 try/except 自己拼 JSON，前端就得认十种格式。

# 统一错误响应：一种形状、一个可报的编号、一句人话；细节只进日志
import logging, uuid
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

log, app = logging.getLogger("api"), FastAPI()


@app.exception_handler(Exception)
def on_unknown(r: Request, e: Exception):
    tid = uuid.uuid4().hex[:12]
    log.exception("unhandled trace=%s path=%s", tid, r.url.path)   # 细节只进日志
    # ⚠️ 绝不把上游报错原样吐出去：既泄露内部结构，用户也看不懂
    return JSONResponse({"error": {"code": "internal_error",       # ⭐ 稳定机器码
                                   "message": "服务开小差了，稍后再试",
                                   "trace_id": tid}}, status_code=500)

业务自己的错误同理：定义一个带「状态码 + 机器码 + 一句人话」的 AppError，再注册一个形状一样的 handler。三条原则：

⚠️ 一个 AI 应用特有的取舍：上游 429 该翻译成什么？ 原样透传，用户会以为是自己发太快了。诚实的做法是分开——用户超限用 429，你的上游额度不够用 503 +「服务繁忙」。两者的止损动作完全不同，混在一起监控也分不清。

📁 六、什么时候从单文件拆出来

别一上来就建八个目录。 判据只有一条：当你为了改一件事必须在同一个文件里上下反复横跳时就该拆，通常在 300～500 行、第三个路由出现时。按「变化的理由」分，不要按「技术种类」分：

app/
├─ main.py       # 只做装配：建 app、挂路由、注册异常处理器
├─ api/          # HTTP 层：路由、请求/响应模型
├─ deps.py       # 依赖：current_user / quota / trace
├─ services/     # 业务编排：一次对话该做哪几步
├─ llm/          # ⭐ 第 4 章的调用层：超时/重试/记账/缓存/假客户端
├─ store/        # 第 6/7 章：数据库与向量库
└─ settings.py   # 配置一律从环境变量读

⭐ 唯一强制的一条：api/ 里不许直接出现模型 SDK 的名字。为什么，下一章第一节整节都在讲。

🔄 七、换个栈怎么对应

⭐ 这张表真正想说的：三个栈里只有「依赖注入」是 FastAPI 的叫法，另两个栈用中间件做同一件事。 「每请求要做的事得有统一的地方挂」这个概念不会过期，Depends 这个名字会。

🔗 这一章连到哪里

✅ 检查点

1. 前端直接调模型 API 会踩到哪四类问题？
2. 单独定义请求/响应模型带来的三个好处是什么？
3. 为什么说 AI 应用比一般 Web 应用更需要 async？
4. async def 里调一个同步阻塞的库会怎样？对照代码跑出了什么结果？这个 bug 为什么本地开发时发现不了？
5. FastAPI 的「逃生舱」是什么？什么时候不该靠它？
6. 依赖注入解决的具体问题是什么？本章埋的两个依赖分别留给哪两章？
7. 统一错误响应的三条原则是什么？上游 429 直接透传会造成什么误解？
8. 什么时候该拆文件？api/ 层唯一强制的规矩是什么？

🛑 可以停在这里

⚡ 走神救援

后端不是给模型调用套层 HTTP，是立四根柱子。① 为什么要有它：密钥在前端等于公开、没有配额边界、没有审计、业务逻辑白盒——模型 API 是「谁拿到密钥谁是老板」，产品需要「谁是谁、能做什么、做过什么」。② 边界：单独写请求/响应模型，非法请求进业务前就 422 被挡，响应模型同时是裁剪器；⭐ 请求发出前能卡死的成本变量有两个——输入长度（max_length）和输出上限（max_tokens），而按 04 章那张价格表输出单价是输入的 3～4 倍，所以更该先卡的是输出那个，⚠️ 它还常带一个很大的默认值，光限制输入拦不住。③ ⭐⭐ async 在解什么：普通 Web 瓶颈是「算不完」，AI 应用瓶颈是「等不完」——一次调用等 3 秒 CPU 全空，所以 AI 应用更需要 async。⚠️ 最贵的坑：async def 里一个同步阻塞调用会卡死整个进程。本章可跑对照：8 个各等 0.5 秒的任务，await asyncio.sleep 版 0.51 秒，time.sleep 版 4.01 秒；错误写法语法完全正确、不报警，本地单人开发一切正常，上线并发一起来才一起爆炸。伪装形态是 requests、同步 DB/Redis 驱动、只有同步版的 SDK。⭐ 逃生舱：路由写成普通 def，FastAPI 自动扔线程池——但线程数有限。④ 依赖注入不是花架子：它是「每请求都要做、又不属于任何业务函数」的统一挂载点。current_user 是第 8 章认证/多租户的位置，quota 是第 12 章限流的位置，再加个 trace 就是第 15 章可观测性的位置；没有这层，那三章只能塞进每个路由开头。dependency_overrides 能在测试里把整条链换掉，和第 4 章的假客户端同源——「能被换掉」是这个骨架最值钱的性质。⑤ 错误处理：对外一种形状（code + message + trace_id），⚠️ 未知异常绝不原样透出上游报错（里面常有密钥前缀、内网 IP）；上游 429 别直接透传，用户超限才 429，上游额度不够用 503。⑥ 拆文件：判据是「改一件事要在同一文件里反复横跳」，api/ 里不许出现模型 SDK 的名字。⑦ 换个栈：Node 用中间件、Go 用 context.Context 做同一件事——概念不会过期，Depends 这个名字会。

下一节 👉 03b-接口的形状.md