第27篇｜FastAPI 写一个推理服务接口

本篇定位：《Python 教程》专栏二「Python 进阶与工程」第 27 篇。承接第 26 篇「客户端如何发 HTTP、如何打大模型」之后，站到对面：用 FastAPI 暴露一个可被调用的推理接口。串起第 18 篇 Pydantic、第 25 篇 asyncio、第 26 篇 httpx，为第 28 篇最小 Agent 提供「可调用的大脑 / 工具端点」。

读完你将能够：用 uv 起一个最小 FastAPI + uvicorn 服务；用 Pydantic 模型定义请求/响应并理解 422；写出同步与异步端点并说明选型；实现非流式 JSON 与流式 StreamingResponse（SSE）推理接口；用 lifespan 共享 httpx.AsyncClient；用本篇客户端自测自己的服务。

预计用时：阅读 35～45 分钟，动手 60～90 分钟。

第 26 篇把「发一封信、等回信」练熟了；本篇把你变成收信的人。一句话定心：FastAPI = 路由 + Pydantic 合同 + ASGI 异步能力——你声明「这条路径收什么形状、返回什么形状」，框架帮你校验、生成文档、在事件环上叠等待。

建议边读边开练习目录：uv init --no-package，再 uv add fastapi "uvicorn[standard]" httpx pydantic。没有上游大模型时，用本地 echo / mock 也能把协议与流式跑通；有 Key 时再把转发换成真实兼容端点。阅读顺序：心智图 → 最小应用 → Pydantic → 异步与 lifespan → 流式 → 推理服务骨架 → 踩坑。

和第 18 篇对照：注解与 BaseModel 在本篇变成入站门口。和第 25 篇对照：async def 端点跑在事件环上，出站必须用 AsyncClient。和第 26 篇对照：同一套 JSON / SSE，只是角色对调——你会用昨天的客户端打今天的服务。

动手时建议「假推理 → 真路由 → 真流式 → 可选真上游」四步：先返回固定字符串；再接 Pydantic；再开 StreamingResponse；最后 lifespan 里挂 AsyncClient 转发。做完后打开 /docs，你会看到自动生成的 OpenAPI——这就是「合同即文档」。

本篇不穷尽 FastAPI 全部依赖注入花样、不替代完整生产网关（鉴权网关、多租户、K8s 部署）、不绑定某一家推理引擎。主线是：ASGI 心智 → 最小路由 → Pydantic 合同 → 异步端点 → 流式响应 → 可自测的推理薄层。和「抄一个脚手架就上线」相比，正确价值是：你能看懂每个装饰器在干什么，并能在出问题时拆到一层一层修。

写服务时保持工程直觉：入站必校验；出站必超时；共享 Client 进 lifespan；流式两端配对；密钥不进仓库。 「在 async def 里狂调同步 requests」与「每个请求新建一个 Client」——两个月后的夜班你会感谢今天躲开它们。

也可以把本篇看成专栏二「入站面」的收官课之一：前面把协议、类型、异步、出站 HTTP 收齐，本篇把「可被调用的服务」立住。下一篇 Agent 会在循环里反复打这类接口或本地工具。

读官方 FastAPI 教程时，按本篇目录对照即可。入门阶段把「会起服务、会写模型、会异步转发、会流式、会自测」做扎实，比背完整依赖树更重要。

写路由时再记一条：服务端的第一原则仍是合同优先于炫——错误的「全异步包装」比同步更糟，因为它摧毁可读性并引入取消、泄漏与假吞吐。排障时一条最小可复现（curl 打 /health + 一次带 Body 的 POST），往往比一百次「再加中间件」更值钱。

和「一切都上某某 Agent 框架」的神话相比，更值钱的是自己会暴露一个干净的 HTTP 面：你会校验、会流式、会转发，框架只是上层编排。把这个品味带进 code review，比强制「必须用某某脚手架」温和且有效。

若你刚从第 26 篇「出站 HTTP」过来，本篇前两节会换一种体感：同样是 JSON 与状态码，但挂钩点从「我怎么发」换成「我怎么收、怎么拒、怎么回」。把 echo 接口写通，再用 httpx 自打——对称感一出来，REST / 推理 API 就不再神秘。

和「在聊天里口头说会写 FastAPI」相比，可运行示例的价值是：机器可重复、可测、可对照 /docs——本地与 CI 上同一套结论。排障时你面对的是「谁 422、谁堵环、谁流式断」，而不是「我记得好像起过服务」。

动手前再立一条纪律：先非流式正确，再开 stream；先 mock 上游正确，再接真 LLM。 先对单一路径跑通并在 /docs 点 Try it out，再引入 lifespan 与转发。很多人倒过来「先写兼容 OpenAI 全集再修校验」，修的时候容易把阻塞调用塞进协程——事件环又变成复杂度放大器。

---

一、从客户端到服务端：FastAPI 心智

1.1 一张图看清入站路径

该图片为标题为“FastAPI：一次入站请求怎么走”的架构流程图，对应文档中“1.1一张图看清入站路径”的内容。图中清晰展示了FastAPI入站请求的完整流程：客户端发起POST JSON格式的HTTP请求至uvicorn服务器，uvicorn作为ASGI服务器接收请求后将其交给FastAPI应用；FastAPI应用内部经路由匹配、Pydantic校验、业务处理等环节后，向客户端返回JSON或流形式的响应。图下方的注解点明了该内容的核心心智，对应文档中关于入站请求路径的要点说明。

图1 FastAPI：一次入站请求怎么走

零件	职责	本篇用法
客户端	发 HTTP	第 26 篇的 httpx / curl
uvicorn	ASGI 服务器	接 TCP、跑应用、可选多 worker
FastAPI	路由与合同	@app.post、Pydantic、依赖
业务函数	真正干活	echo / 转发上游 / 流式生成
响应	JSON 或流	return model / StreamingResponse

# 最小「起一个服务」——后面各节往里加肉
from fastapi import FastAPI

app = FastAPI(title="mini-infer")

@app.get("/health")
def health():
    return {"ok": True}

uv run uvicorn main:app --reload --port 8000
# 另开终端：
curl -s http://127.0.0.1:8000/health

{"ok":true}

要点：

1. ASGI：异步版 WSGI；uvicorn 是服务器，FastAPI 是应用。
2. 装饰器即路由：路径 + 方法绑定到函数；返回 dict / 模型会自动变 JSON。
3. 文档免费送：浏览器打开 http://127.0.0.1:8000/docs 即 Swagger UI。

1.2 和前后篇的挂钩

第 26 篇站在客户端；本篇站在服务端——协议对称。第 18 篇的 Pydantic 在这里成为默认入参类型。第 25 篇的事件环，是 async def 端点的舞台。第 28 篇 Agent 会把本篇接口当成可调用端点之一。

AI Infra 视角：vLLM / SGLang / 各云推理网关，对外往往就是「FastAPI 风格」或 OpenAI 兼容 HTTP。你写的薄网关负责校验、鉴权、路由与流式透传；重计算在上游。先把薄层写对，比先纠结「选哪家引擎」更划算。

1.3 安装与目录约定

uv init --no-package
uv add fastapi "uvicorn[standard]" httpx pydantic
# 可选：密钥仍走环境变量（转发上游时）
export OPENAI_API_KEY="sk-..."
export OPENAI_BASE_URL="https://api.openai.com/v1"

统一约定：下文示例假设工作目录里有 main.py（或拆成 app/ 包）；本地用 --reload 开发，生产再谈 worker 与反向代理。没有上游时用内置 echo，仍能练完所有形状。

1.4 和「随便起一个 Flask」差在哪

不是贬低其他框架，而是说明：本系列选 FastAPI，是因为它和类型注解、异步、OpenAPI、流式推理的契合度高。

---

二、最小应用：路由、状态码与自动文档

2.1 GET / POST 与路径参数

from fastapi import FastAPI, HTTPException

app = FastAPI(title="mini-infer")

@app.get("/health")
def health():
    return {"ok": True}

@app.get("/items/{item_id}")
def read_item(item_id: int, q: str | None = None):
    return {"item_id": item_id, "q": q}

@app.post("/echo")
def echo(payload: dict):
    # 裸 dict 能跑，但不推荐——下一节换成模型
    if "text" not in payload:
        raise HTTPException(status_code=400, detail="missing text")
    return {"echo": payload["text"]}

curl -s http://127.0.0.1:8000/items/42?q=hi
curl -s -X POST http://127.0.0.1:8000/echo \
  -H 'Content-Type: application/json' \
  -d '{"text":"你好"}'

{"item_id":42,"q":"hi"}
{"echo":"你好"}

item_id: int 会自动把路径段转成整数；转失败直接 422。HTTPException 是主动返回业务错误的标准方式——比裸抛 ValueError 更可控。

2.2 看看自动文档在帮你什么

打开 /docs：每个路由的参数、Body 示例、响应模型都会列出来。改注解 → 刷新文档，这就是「合同即文档」。团队协作时，前端或评测同学往往先看 /docs，而不是翻你的源码。

2.3 响应模型与状态码

from pydantic import BaseModel
from fastapi import FastAPI

app = FastAPI()

class Msg(BaseModel):
    message: str

@app.post("/hi", response_model=Msg, status_code=201)
def hi():
    return Msg(message="created")

response_model 会过滤多余字段、按模型序列化；status_code 改默认 200。推理接口通常保持 200 + JSON；创建类资源才常用 201。

2.4 开发期热重载

uvicorn main:app --reload 会监视文件变化并重启 worker。注意：全局状态（内存计数器、未放进 lifespan 的连接）会在重载时丢失——这是特性不是 bug。真正的共享资源，放到下一节的 lifespan。

读到这里自检：你能否不看文档写出带路径参数的 GET 和返回 dict 的 POST？能否解释 uvicorn 与 FastAPI 的分工？卡壳就回到图 1。

---

三、Pydantic 模型：请求与响应合同

3.1 把裸 dict 换成模型

这张图展示了FastAPI中Pydantic模型作为请求与响应合同的核心作用，标题为“Pydantic：请求进门、响应出门”。左侧是待校验的原始JSON数据，右侧箭头指向代表请求校验的InferRequest模型，该模型为BaseModel子类，定义了prompt、max_tokens、stream三个字段及各自类型，若数据缺字段或类型错误会返回422状态码。经校验通过的请求数据会流入业务环节，最终生成右侧的InferResponse模型，该模型同样为Pydantic模型，定义了text、tokens、model等响应字段，可自动对接OpenAPI标准，实现数据校验与格式规范化，确保非法输入无法进入业务逻辑。

图2 Pydantic：请求进门、响应出门

from pydantic import BaseModel, Field
from fastapi import FastAPI

app = FastAPI(title="mini-infer")

class InferRequest(BaseModel):
    prompt: str = Field(min_length=1, max_length=4000)
    max_tokens: int = Field(default=64, ge=1, le=2048)
    stream: bool = False
    temperature: float = Field(default=0.7, ge=0.0, le=2.0)

class InferResponse(BaseModel):
    text: str
    tokens: int
    model: str = "echo-mini"

@app.post("/v1/infer", response_model=InferResponse)
def infer(req: InferRequest) -> InferResponse:
    # 假推理：原样回显，便于先通协议
    text = f"echo:{req.prompt[:80]}"
    return InferResponse(text=text, tokens=len(text))

curl -s -X POST http://127.0.0.1:8000/v1/infer \
  -H 'Content-Type: application/json' \
  -d '{"prompt":"你好，世界"}'

{"text":"echo:你好，世界","tokens":12,"model":"echo-mini"}

故意传坏数据：

curl -s -X POST http://127.0.0.1:8000/v1/infer \
  -H 'Content-Type: application/json' \
  -d '{"prompt":"","max_tokens":99999}'

你会看到 422 Unprocessable Entity，Body 里是结构化的 detail 列表——字段名、错误类型一目了然。这比业务函数里到处 if not prompt 干净得多。

3.2 嵌套模型与列表

class Message(BaseModel):
    role: str
    content: str

class ChatRequest(BaseModel):
    model: str = "echo-mini"
    messages: list[Message]
    stream: bool = False

class ChatChoice(BaseModel):
    index: int = 0
    message: Message
    finish_reason: str = "stop"

class ChatResponse(BaseModel):
    id: str = "chatcmpl-echo"
    choices: list[ChatChoice]

这和 OpenAI 兼容形状已经很接近。第 26 篇客户端拼的 messages，本篇服务端用同一套字段接住——联调时少翻译一层。

3.3 为何不要在业务里「先收 dict 再手验」

做法	后果	推理场景例子
payload: dict + 手写 if	文档空白；漏检；错误格式不一	prompt 上限防撑爆
InferRequest 注解	自动 422；/docs 有示例；IDE 补全	max_tokens、temperature
业务内再 model_validate	可以，但路由层已校验时多为重复	stream=False
pattern	正则	模型名白名单（慎用过严）
阻塞调用	不在 async 路径里裸跑	防泄漏

AI Infra 视角：LLM 工具参数、网关入参、评测用例配置，都适合 Pydantic。模型输出的 JSON「看起来像」不够，进门必须验——第 18 篇的结论在本篇落地。

3.4 Field 约束常用表

字段级能表达的优先 Field；跨字段规则（例如「stream 为真时禁止某字段」）再用 model 级 validator——需要时查 Pydantic v2 文档即可。

---

四、异步端点、依赖注入与 lifespan

4.1 sync def vs async def

这张图用于说明FastAPI中同步def端点与异步async def端点的差异及选型，标题为“同步 def vs 异步 async def 端点”。左侧粉色区域对应@ app.post + def的同步端点，说明其会交由FastAPI的线程池执行，适合短计算、暂无异步阻塞的场景，但在同步端点中调用同步requests请求会占用线程池致整站变慢；右侧绿色区域对应@ app.post + async def的异步端点，运行在事件环上，适合异步出站请求、调用LLM等场景，推荐在lifespan中创建共享异步客户端。底部的选型口诀进一步明确，可等待I/O用async def，纯CPU或无异步库用def，并提醒不要在async def中直接调用同步requests。

图3 同步 def vs 异步 async def 端点

import asyncio
from fastapi import FastAPI

app = FastAPI()

@app.get("/sync-sleep")
def sync_sleep():
    # 演示：同步睡眠会占线程池工人
    import time
    time.sleep(0.2)
    return {"mode": "sync"}

@app.get("/async-sleep")
async def async_sleep():
    await asyncio.sleep(0.2)
    return {"mode": "async"}

口诀：

1. 出站 I/O 能 await → 写 async def，里面用 httpx.AsyncClient。
2. 纯短计算或只有同步库 → 可以 def，让 FastAPI 丢线程池（知代价：池子有限）。
3. 禁止：在 async def 里直接 requests.get / time.sleep 长阻塞。

4.2 lifespan：共享 AsyncClient

进程级长连接不要每个请求 async with AsyncClient() 新建。用 lifespan 在启动时创建、关闭时 aclose：

from contextlib import asynccontextmanager
import httpx
from fastapi import FastAPI, Request

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.http = httpx.AsyncClient(
        timeout=httpx.Timeout(60.0, connect=5.0),
        headers={"User-Agent": "mini-infer/27"},
    )
    yield
    await app.state.http.aclose()

app = FastAPI(title="mini-infer", lifespan=lifespan)

@app.post("/v1/infer")
async def infer(req: InferRequest, request: Request):
    client: httpx.AsyncClient = request.app.state.http
    # 此处可转发上游；无 Key 时走 echo
    text = f"echo:{req.prompt[:80]}"
    return InferResponse(text=text, tokens=len(text))

和「在全局模块 import 时创建 Client」相比，lifespan 的好处是：启动失败可见、关闭可清理、测试可替换。

4.3 依赖注入浅尝

from fastapi import Depends, Header, HTTPException

async def require_api_key(x_api_key: str | None = Header(default=None)):
    import os
    expected = os.environ.get("SERVICE_API_KEY")
    if expected and x_api_key != expected:
        raise HTTPException(status_code=401, detail="invalid api key")
    return x_api_key

@app.post("/v1/infer")
async def infer(
    req: InferRequest,
    request: Request,
    _: str | None = Depends(require_api_key),
):
    ...

Depends 把鉴权、取配置、取 DB session 从业务函数里拆出去。本篇点到为止；生产级 OAuth / JWT 是另一门课，但「门口验 Key」已经够防「裸奔端口」。

4.4 和第 25、26 篇的拼接检查清单

读到这里自检：你能否说明为何推理转发要用 async？能否画出 lifespan 里 Client 的生死？卡壳就回到图 3。

---

五、流式响应：把 LLM token 推给客户端

5.1 StreamingResponse 与 SSE 形状

图片展示了流式推理中StreamingResponse + SSE的工作流程。服务端生成器通过yield生成分块数据，如“你”“好”等，不等整段生成完就返回。StreamingResponse将这些数据以text/event-stream格式边产边推，客户端通过httpx stream / alter读取，读到“你”立刻显示，读到“好”也显示，读到“DONE”结束，体验类似打字机。此图与上下文紧密相关，直观呈现了流式响应的实现原理。

图4 流式推理：StreamingResponse + SSE

聊天 UI 要的是打字机：首 token 尽快到达，而不是等整段生成完。服务端用生成器 yield，FastAPI 用 StreamingResponse 边产边推。

import json
import asyncio
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

async def fake_token_stream(prompt: str):
    # 假流式：按字吐出；真场景改为读上游 aiter_bytes
    for ch in f"回复:{prompt[:20]}":
        payload = {"choices": [{"delta": {"content": ch}}]}
        yield f"data: {json.dumps(payload, ensure_ascii=False)}\n\n"
        await asyncio.sleep(0.05)
    yield "data: [DONE]\n\n"

@app.post("/v1/infer/stream")
async def infer_stream(req: InferRequest):
    return StreamingResponse(
        fake_token_stream(req.prompt),
        media_type="text/event-stream",
    )

curl -N -X POST http://127.0.0.1:8000/v1/infer/stream \
  -H 'Content-Type: application/json' \
  -d '{"prompt":"流式你好","stream":true}'

data: {"choices": [{"delta": {"content": "回"}}]}

data: {"choices": [{"delta": {"content": "复"}}]}

...
data: [DONE]

-N 让 curl 不缓冲，才能看见打字机效果。

5.2 同一路由按 stream 字段分支

@app.post("/v1/infer")
async def infer(req: InferRequest, request: Request):
    if req.stream:
        return StreamingResponse(
            fake_token_stream(req.prompt),
            media_type="text/event-stream",
        )
    text = f"echo:{req.prompt[:80]}"
    return InferResponse(text=text, tokens=len(text))

注意：流式分支的返回类型与 response_model=InferResponse 可能冲突——实践中常把流式路径拆成独立路由，或流式时不用 response_model。本篇推荐 /v1/infer 非流式 + /v1/infer/stream 流式 两条清晰路径。

5.3 用第 26 篇客户端自测流式

import httpx

with httpx.Client(timeout=30.0) as client:
    with client.stream(
        "POST",
        "http://127.0.0.1:8000/v1/infer/stream",
        json={"prompt": "自测", "stream": True},
    ) as resp:
        resp.raise_for_status()
        for line in resp.iter_lines():
            if line:
                print(line)

服务端 yield 与客户端 iter_lines 必须配对——这是第 26 篇与本篇的对称美。若服务端整包 return 大 JSON，客户端再怎么 stream 也变不成打字机。

5.4 透传上游流式（要点）

真接上游时，常见模式是：async with client.stream("POST", url, json=...) → async for chunk in resp.aiter_bytes() → yield chunk。注意：

1. 及时检查上游状态码，错误不要静默当 token 推。
2. 客户端断开时，生成器被取消——用 try/finally 或上下文确保上游连接关闭。
3. 不要在透传前把整个上游 body read() 进内存。

AI Infra 视角：网关层「流式透传」是降低首字延迟的关键路径。错误的「先拼完整再返回」会把 TTFT（Time To First Token）拖成整段生成时间——用户体感直接崩。

---

六、AI Infra：推理服务骨架与自测

6.1 一张图看清薄网关

这张图对应文档中提到的“图5最小推理服务”，展示了最小推理服务的流转逻辑。流程从左侧的调用方（聊天UI、评测脚本、Agent等）发起请求，接入中间的FastAPI推理网关，该网关包含Pydantic入参合同校验、鉴权/限流、async转发、统一响应形状等环节，之后将请求发往右侧的上游推理（兼容OpenAI的vLLM/SGLang、本地模型模块），最后完成响应。图中还标注了AI Infra视角的总结，说明FastAPI常作为“薄网关”完成校验、鉴权、路由与流式透传，重负载由上游推理引擎承担。

图5 最小推理服务：入站 · 校验 · 上游 · 出站

你的 FastAPI 往往是薄网关：Pydantic 校验 → 可选鉴权/限流 → AsyncClient 转发 → 统一响应或 SSE 透传。重活（连续 batch、KV cache、调度）在 vLLM / SGLang / 云端点。本篇把薄层写对，就具备对接任何兼容上游的能力。

6.2 完整最小骨架（可粘贴）

# main.py —— 本地可跑的最小推理服务
from __future__ import annotations

import json
import os
from contextlib import asynccontextmanager

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


class InferRequest(BaseModel):
    prompt: str = Field(min_length=1, max_length=4000)
    max_tokens: int = Field(default=64, ge=1, le=2048)
    stream: bool = False


class InferResponse(BaseModel):
    text: str
    tokens: int
    model: str = "echo-mini"


@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.http = httpx.AsyncClient(timeout=httpx.Timeout(60.0, connect=5.0))
    yield
    await app.state.http.aclose()


app = FastAPI(title="mini-infer", version="0.1.0", lifespan=lifespan)


def _check_key(x_api_key: str | None) -> None:
    expected = os.environ.get("SERVICE_API_KEY")
    if expected and x_api_key != expected:
        raise HTTPException(status_code=401, detail="invalid api key")


@app.get("/health")
def health():
    return {"ok": True}


async def _echo_stream(prompt: str):
    for ch in f"echo:{prompt[:40]}":
        yield f"data: {json.dumps({'token': ch}, ensure_ascii=False)}\n\n"
    yield "data: [DONE]\n\n"


@app.post("/v1/infer", response_model=InferResponse)
async def infer(
    req: InferRequest,
    request: Request,
    x_api_key: str | None = Header(default=None),
):
    _check_key(x_api_key)
    if req.stream:
        return StreamingResponse(
            _echo_stream(req.prompt),
            media_type="text/event-stream",
        )
    # 有上游 Key 时可改为转发；此处默认 echo
    base = os.environ.get("OPENAI_BASE_URL")
    key = os.environ.get("OPENAI_API_KEY")
    if base and key:
        client: httpx.AsyncClient = request.app.state.http
        r = await client.post(
            f"{base.rstrip('/')}/chat/completions",
            headers={"Authorization": f"Bearer {key}"},
            json={
                "model": os.environ.get("OPENAI_MODEL", "gpt-4o-mini"),
                "messages": [{"role": "user", "content": req.prompt}],
                "max_tokens": req.max_tokens,
            },
        )
        r.raise_for_status()
        data = r.json()
        text = data["choices"][0]["message"]["content"]
        return InferResponse(text=text, tokens=data.get("usage", {}).get("total_tokens", 0), model=data.get("model", "upstream"))
    text = f"echo:{req.prompt[:80]}"
    return InferResponse(text=text, tokens=len(text))

uv run uvicorn main:app --reload --port 8000

无上游时永远走 echo；设了 OPENAI_* 就转发。自测脚本用第 26 篇的 httpx 即可。

6.3 选型表

6.4 和后续篇章的分工

• 第 25 篇：事件环、gather、禁阻塞。
• 第 26 篇：出站 HTTP、流式客户端、兼容 Chat Completions。
• 本篇：入站 FastAPI、Pydantic、流式服务端、lifespan。
• 第 28 篇：Agent 循环里调用工具 / 模型（HTTP 或本地函数）。

读到这里自检：你能否画出「客户端 → uvicorn → 校验 → 业务 → 流式/JSON」？能否解释为何 Client 要进 lifespan？卡壳就回到图 1、图 5。

和「等线上偶发再加校验」相比，现在付出的代价极低：模型字段、Field 范围、可选 API Key。事故夜里你会感谢今天的自己。

再强调边界：本篇教你用 FastAPI 暴露干净的推理 HTTP 面；水平扩展、GPU 调度、前缀缓存是上游与基础设施的事。不要因为「听说有更炫的一键部署」就省略校验与超时。

若你带团队做 code review，可收成三条门槛：新增入站是否有 Pydantic；async 路径是否真用异步出站；流式是否真 yield 而非整包。门槛不在「必须抄某脚手架」，而在于不许用裸 dict 与假流式换虚假进度。

6.5 本地联调三板斧

1. /health 必须 200。
2. /docs 里 Try it out 跑通非流式。
3. curl -N 或 httpx stream 跑通流式，确认能看到多帧而非一帧。

三条过关，再谈鉴权、限流、对接真上游。很多人跳过第 3 条，结果「本地觉得流式了、浏览器仍在等」——其实是缓冲或 media_type 不对。

动手时再记一条：服务端的可观测优先——每次入站至少能回答：谁打的、状态码、耗时、是否流式、上游是否 429。缺这五问，线上「偶发变慢」会变成玄学。

和「等框架帮你处理好一切」相比，自己会写薄网关的价值是：换上游、改字段、临时 mock，你不至于停摆。 薄封装 + 清晰合同，比厚魔法更耐折腾。

---

七、踩坑大全

通用

坑 1：在 async def 里用同步 requests / time.sleep。 现象：延迟尖刺、吞吐崩。原因：堵事件环。解决：httpx.AsyncClient + await；或改 def 知代价。

坑 2：入参用裸 dict 不校验。 现象：KeyError、脏数据进模型。解决：Pydantic 模型；让 422 在门口发生。

坑 3：每个请求新建 AsyncClient。 现象：握手过多、FD 涨。解决：lifespan 共享；请求结束不要 aclose 掉共享实例。

坑 4：流式却先拼完整字符串再 return。 现象：无首字优势。解决：生成器 + StreamingResponse；media_type="text/event-stream"。

坑 5：response_model 与流式返回混用导致怪异。 现象：文档/序列化不符合预期。解决：流式独立路由，或不给该路由绑 JSON response_model。

坑 6：忘记 Content-Type: application/json。 现象：422 或 Body 解析失败。解决：客户端显式声明；/docs 试调可对照。

坑 7：把服务 API Key 与上游 LLM Key 混为一谈。 现象：鉴权逻辑乱。解决：SERVICE_API_KEY 管入站；OPENAI_API_KEY 管出站。

macOS / Linux / Windows

坑 8：--reload 下全局状态「丢了」。 现象：计数器归零。原因：重载重启进程。解决：开发期接受；或用外部 Redis 等。

坑 9：Linux 文件描述符不够。 现象：高并发 Too many open files。解决：提 ulimit；复用 Client；限并发。

坑 10：Windows 下 curl 流式显示异常。 现象：SSE 一次喷出。解决：确认关闭缓冲；用 httpx 脚本验证更稳。

与 AI / 网关叠加

坑 11：上游 429/5xx 未转义给调用方。 现象：调用方只看到 500 或挂起。解决：捕获后返回明确状态码与 detail；可映射 502/504。

坑 12：透传流式时不处理客户端断开。 现象：上游仍在生成、浪费算力。解决：取消传播；finally 关闭上游流。

坑 13：CORS 未配置导致浏览器打不通。 现象：本地 JS 报跨域。解决：开发期 CORSMiddleware；生产按域名白名单。

---

八、小结

合上文章后，试着用自己的话说明：uvicorn 与 FastAPI 如何分工、Pydantic 如何把非法请求挡在 422、为何推理转发要用 async + lifespan、流式 SSE 如何 yield。讲得清，才算真的会了。动手至少跑通：/health → 带模型的 /v1/infer → 故意坏数据看 422 → 流式 curl -N → httpx 自测。

1. 心智：客户端发信（26）↔︎ 服务端收信（本篇）；uvicorn 管 ASGI，FastAPI 管路由与合同。
2. 路由：装饰器绑定方法与路径；/docs 即活文档。
3. Pydantic：请求/响应模型；Field 约束；非法 → 422。
4. 异步：出站 I/O 用 async def + AsyncClient；共享连接进 lifespan。
5. 流式：StreamingResponse + SSE；与客户端 iter_lines 配对。
6. 薄网关：校验、鉴权、转发、统一形状；重推理在上游。

把故事连起来：第 23～25 篇 concurrent 三件套 → 第 26 篇出站 HTTP → 本篇入站 FastAPI → 下一站最小 Agent。专栏二工程刀具接近成套。

若只能带走一句话：入站 Pydantic；出站 AsyncClient；流式真 yield；Client 进 lifespan。

也可以用「三个不要」自检：不要裸 dict 入参；不要在 async 里跑同步 HTTP；不要假流式整包返回。

再留一个自测：起服务 → /docs Try it out → 坏数据 422 → 流式多帧 → 用第 26 篇客户端打通——五问都顺，本篇就站稳了。

本篇技能会立刻复利：读开源推理网关先看它如何校验与流式；PR 里要求说清是否真 async 出站；Agent 篇把本篇端点当工具。后续章节都假设这层已稳。

和专栏一串起来复习也不亏：函数与作用域（07）对应依赖拆分；异常（11）要在 HTTPException 与上游错误里浮出；类型与 Pydantic（18）是合同；asyncio（25）与 httpx（26）提供编排与出站；本篇是专栏二把「推理服务接口」说透的一课。

也可以把收获压成检查清单：会起 uvicorn；会写模型与 422；会 async 端点；会 lifespan Client；会 SSE 流式；会自测。六项过关，即可进入 Agent 篇。

若把本篇压成给同事的三十秒讲解：先画「入站路径」；再强调「Pydantic + 别堵环」；最后提「流式要 yield」。三张草图比贴一篇博客更管用。等对方能独立把 echo 服务改成带校验的转发，并解释 lifespan 为何存在，就可以一起 code review 服务端 PR 了。

动手清单再压缩一版：/health；InferRequest POST；故意 422；StreamingResponse；lifespan AsyncClient。五步做完，比只「看完文章」扎实一个数量级。

---

九、课后练习

1. （基础） 用 uv 创建项目，写出带 /health 的 FastAPI，用 uvicorn 启动并用 curl 验证。
2. （基础） 定义 InferRequest / InferResponse，实现非流式 /v1/infer echo；故意传空 prompt，确认 422。
3. （基础） 给接口加可选 Header X-API-Key，环境变量存在时校验，错误返回 401。
4. （理解） 把 /v1/infer 改成 async def，用 lifespan 挂上 httpx.AsyncClient，在有/无上游 Key 两种模式下都能返回。
5. （理解） 实现 /v1/infer/stream，按字 yield SSE；用 curl -N 与 httpx iter_lines 两种方式消费。
6. （理解） 在 /docs 截图或文字记录：请求体 Schema 是否出现 max_tokens 的范围约束；解释注解如何变成文档。
7. （进阶） 将第 26 篇的 Chat Completions 客户端改成打本篇服务；对比非流式与流式总耗时与「首包时间」。
8. （进阶 · AI Infra） 画一张你设想的推理网关图：鉴权、限流、路由到 vLLM/SGLang、SSE 透传；写 200 字说明「薄网关」与「引擎」的边界，以及假流式的危害。

---

下一篇预告

下一篇是《第28篇｜实战：用纯 Python 写一个最小 Agent（工具调用循环）》——在搞清「如何暴露推理接口」之后，把 模型 + 工具 + 循环 收成最小 Agent：解析工具调用、执行函数、把结果塞回 messages、直到结束。你会看到：和本篇 HTTP 面如何对接、和纯本地工具如何混用、以及如何避免失控循环。

专栏二收官路线：FastAPI（本篇）→ 最小 Agent（28）。带着本篇的校验、异步与流式纪律，写 Agent 时记得：模型出入要有形状；工具失败要可观测；循环要有上限。

若想提前热身：把本篇 /v1/infer 保持可跑，下一篇直接当「大脑」调用；同时准备一两个本地工具函数（算数、查字典）——对接时你会带着具体体感。

也可以把第 25～27 篇对照：协程叠等待、出站真实 HTTP、本篇入站服务——工程刀具的轮廓已经清晰，下一篇把「Agent 循环」说透。

专栏一用十四篇把「对象与协议」立住；专栏二从迭代器（15）走到 FastAPI（27），下一站是最小 Agent。下一篇见。

若时间充裕，再多做一件事：打开你最近用过的一个推理或 Bot 服务，找到入站校验与流式代码，记三条「它如何定义模型、如何 yield、Client 放哪」。这种阅读比空背装饰器更能把纪律刻进肌肉记忆。

进入专栏二至此，可验证面、可观测面、并发面、联网面与入站面开始交汇：会自动验回归，会在正确层级叠任务，会安全地发请求，也会暴露可调用的服务。下一篇开始把「会思考并调用工具的循环」配齐。若某一节读着抽象，回到对应插图把箭头走一遍，再手打 lifespan 与 StreamingResponse——多数卡住的地方，其实是「同步脚本习惯」被原样搬进了异步服务。

写服务或读延迟报告时再记一条：错误往往来自裸 dict 与假异步（该 await 却用 requests、该 yield 却整包 return）。和「栈很长但根因很浅」相比，先把合同与事件环写对能省很多时间。

也可以把本篇看成专栏二的「入站 HTTP」补课：前面把协议、积木、环境、眼睛、断言、并发与出站收齐，本篇把服务接口收齐。后面 Agent 篇则把模型与工具打进循环。

若阅读时觉得「Flask / Django / FastAPI」总搅在一起，回到图 1 与图 3：先定同步还是异步，再定合同是否在门口。日常编程三者都可能遇到，但本系列主线是 FastAPI + Pydantic + asyncio——混谈是多数初学者卡住的根源。

收官前对照练习：同一段「echo 推理」，分别用裸 dict、Pydantic、再加流式。你会清楚感受到：合同与流式不是装饰，而是可联调、可观测的服务面——也是专栏二与 Agent 对接的入场券之一。

写完练习后，不妨在自己的仓库里开一个小 PR：只把某条裸 JSON 接口改成 Pydantic + 可选流式，并补上 /health，不改业务语义。这种「小步服务化」比空读文档更能把工作流刻进肌肉记忆，也更适合团队推广。

若你带新人，可以把本篇插图打印成一页速查：左列场景（健康检查 / 校验入参 / 转发 LLM / 打字机流式），右列策略（def/async def / BaseModel / lifespan Client / StreamingResponse）。新人写服务时对着贴，两周后就会自然说出「先 422」与「先禁止环内 requests」。工具文化是带出来的，不是靠一次分享会喊出来的。

最后把专栏二的位置再钉一次：协议篇（15～18）解决「怎么正确接线」；标准库与依赖篇（19～20）解决「积木与环境」；可观测与可验证篇（21～22）解决「如何被看见与自动验证」；并发篇（23～25）解决「线程、进程、协程如何分层」；HTTP 篇（26）解决「出站怎么发」；本篇解决「推理服务接口怎么暴露」；下一篇解决「最小 Agent 怎么转起来」。七者齐备，脚本才既写得快，又服务得敢——下一篇再给它配上 Agent 循环的实话。