先说结论

LLM 应用的 HTTP 层，90% 的工作不是“写个接口”，而是处理四件事：

1. 异步 IO：模型调用是网络等待，不能让一个请求堵住整个进程；

2. 流式输出：首 token 延迟决定体感，SSE 是默认选择；

3. 资源生命周期：HTTP 连接池、数据库、检查点、模型客户端，必须在启动时建、关闭时收；

4. 失败模式：超时、断连、限流、上游 5xx、流开始后的错误——这些不会因为 Demo 能跑就消失。

这一篇从最小服务开始，一步步把这些补上。代码可以直接作为后续 LangGraph 项目的底座。

1. 为什么 LLM 服务默认选 FastAPI

不适合的场景也要说清楚：

• CPU 密集（本地大模型推理、批量 embedding）：应该放到独立 worker / 推理服务，不要塞进 API 进程；

• 超长任务（几分钟以上的批处理）：用队列（Celery / Arq / Dramatiq）+ 任务状态查询，HTTP 只负责提交；

• 需要双向实时通信（语音、协同编辑）：WebSocket 更合适，SSE 只能服务器单向推。

2. 最小可用结构

先别急着堆功能。一个能长大的 FastAPI 项目，从第一天就该有清晰的目录：

app/ ├── main.py # 应用工厂、路由挂载、中间件 ├── config.py # pydantic-settings 配置 ├── api/ │ ├── deps.py # 依赖：鉴权、限流、资源获取 │ └── chat.py # 聊天接口 ├── services/ │ └── llm.py # 模型调用、流式适配 └── schemas.py # 请求 / 响应模型

2.1 配置：不要用os.getenv散落各处

# app/config.py from functools import lru_cache from pydantic_settings import BaseSettings, SettingsConfigDict ​ ​ class Settings(BaseSettings): model_config = SettingsConfigDict(env_file=".env", extra="ignore") ​ app_name: str = "llm-app" environment: str = "dev" ​ # 模型 openai_api_key: str openai_base_url: str | None = None chat_model: str = "openai:gpt-4.1-mini" ​ # 服务 service_api_key: str = "dev-key-change-me" request_timeout_seconds: float = 90.0 max_input_chars: int = 20_000 ​ ​ @lru_cache def get_settings() -> Settings: return Settings()

lru_cache让配置只解析一次；测试时可以用get_settings.cache_clear()重置。生产环境不要把.env打进镜像，用平台注入的环境变量或密钥管理服务。

2.2 应用工厂 + lifespan

# app/main.py from contextlib import asynccontextmanager ​ import httpx from fastapi import FastAPI from langchain.chat_models import init_chat_model ​ from app.api.chat import router as chat_router from app.config import get_settings ​ ​ @asynccontextmanager async def lifespan(app: FastAPI): settings = get_settings() ​ # 共享 HTTP 客户端：复用连接、统一超时 app.state.http = httpx.AsyncClient( timeout=httpx.Timeout(settings.request_timeout_seconds, connect=10.0), limits=httpx.Limits(max_connections=100, max_keepalive_connections=20), ) ​ # 模型客户端是线程安全的，进程内复用即可 app.state.model = init_chat_model( settings.chat_model, temperature=0.2, timeout=settings.request_timeout_seconds, ) ​ yield ​ await app.state.http.aclose() ​ ​ def create_app() -> FastAPI: settings = get_settings() app = FastAPI( title=settings.app_name, version="0.1.0", lifespan=lifespan, docs_url="/docs" if settings.environment != "prod" else None, ) app.include_router(chat_router, prefix="/v1") return app ​ ​ app = create_app()

几个容易忽略的点：

• 不要在模块顶层创建AsyncClient。它绑定事件循环，在uvicornreload、多 worker、测试环境下容易出Event loop is closed。

• init_chat_model返回的模型对象内部通常持有同步/异步客户端，进程内复用没问题；不要每个请求 new 一次。

• lifespan里抛异常会导致启动失败，这比“启动成功但第一个请求才炸”好得多。

3. async 的边界：什么时候会堵住事件循环

这是 FastAPI 新手最容易踩的坑。

# 反例：在 async 函数里调用同步阻塞 SDK @app.post("/chat") async def chat(req: ChatRequest): result = some_blocking_llm_sdk.call(req.message) # 阻塞整个事件循环 return {"text": result}

事件循环是单线程的。一个同步调用卡住 30 秒，这个 worker 上所有请求都卡住。

正确做法有三种：

1. 优先用异步客户端（httpx.AsyncClient、openai.AsyncOpenAI、LangChain 的ainvoke/astream）；

2. 只有同步客户端时，把路由声明成def（不是async def），FastAPI 会自动放进线程池；

3. 必须在async def里调同步代码时，显式丢线程池：

from starlette.concurrency import run_in_threadpool ​ result = await run_in_threadpool(some_blocking_llm_sdk.call, req.message)

注意线程池大小有限（默认约 40），大量并发阻塞调用会排队。这不是扩展方案，只是避免堵住事件循环。

超时：每一层都要有

import asyncio ​ async def generate_answer(model, messages): async with asyncio.timeout(60): # 整体上限 return await model.ainvoke(messages)

• 模型 SDK 自己的timeout：单次 HTTP 请求；

• asyncio.timeout：包含重试、工具调用的整体预算；

• 网关 / Nginx 的proxy_read_timeout：必须大于应用超时，否则客户端先断；

• 数据库、向量库客户端：各自设置连接超时和读超时。

只设一层，通常等于没设。

4. 请求模型：把“自然语言接口”变成“有契约的接口”

LLM 应用容易写成{"message": "..."}一把梭。更好的做法是把业务字段显式建模：

# app/schemas.py from typing import Literal from pydantic import BaseModel, Field ​ ​ class ChatMessage(BaseModel): role: Literal["system", "user", "assistant"] content: str = Field(min_length=1, max_length=20_000) ​ ​ class ChatRequest(BaseModel): messages: list[ChatMessage] = Field(min_length=1, max_length=50) session_id: str | None = Field(default=None, max_length=128) stream: bool = True ​ ​ class ChatChunk(BaseModel): type: Literal["token", "done", "error"] text: str | None = None code: str | None = None

好处：

• OpenAPI 自动生成，前端/调用方有明确契约；

• max_length在入口就挡住超长输入，避免把成本问题带到模型层；

• 后续加字段（user_id、locale、retrieval_scope）不需要改路由签名。

不要把 LLM 的结构化输出直接当 HTTP 响应模型。模型输出是不可信数据，必须先经过 Pydantic 校验、业务校验，再返回。

5. 流式输出：SSE 是 LLM 应用的默认交互

5.1 三种方案怎么选

SSE 的优势是简单：基于 HTTP，自带重连语义，代理和负载均衡友好（配置正确的前提下）。浏览器原生EventSource只支持 GET，所以很多聊天产品用fetch+ReadableStream解析 SSE，或者用 POST + 自定义解析。

5.2 协议设计：别只发data: token

一个能上线的流式协议，至少要有这些事件：

event: meta data: {"request_id": "...", "model": "..."} event: token data: {"text": "你"} event: tool data: {"name": "search", "status": "running"} event: done data: {"finish_reason": "stop", "usage": {...}} event: error data: {"code": "UPSTREAM_TIMEOUT", "message": "..."}

约定：

• done和error都是终止事件，客户端收到后停止读取；

• 流开始后 HTTP 状态码已经固定为 200，业务错误只能放在事件里；

• 每个事件带id，便于客户端断线后从Last-Event-ID恢复（如果业务支持）；

• 用comment发心跳，避免代理在长时间无数据时断开。

5.3 先把模型流接出来

路由只负责协议，模型调用放到 service 层：

# app/services/llm.py from collections.abc import AsyncIterator from langchain.messages import AIMessage, HumanMessage, SystemMessage _MESSAGE_TYPES = { "system": SystemMessage, "user": HumanMessage, "assistant": AIMessage, } def _chunk_text(chunk) -> str: """兼容 content 为 str 或 content blocks 列表两种情况。""" content = chunk.content if isinstance(content, str): return content return "".join( block.get("text", "") for block in content if isinstance(block, dict) and block.get("type") == "text" ) async def stream_chat(model, messages: list) -> AsyncIterator[str]: lc_messages = [_MESSAGE_TYPES[m.role](content=m.content) for m in messages] async for chunk in model.astream(lc_messages): text = _chunk_text(chunk) if text: yield text

这样路由、协议、模型调用各自独立，测试时可以单独替换stream_chat。

5.4 新版 FastAPI：内置EventSourceResponse

较新的 FastAPI 在fastapi.sse里提供了原生 SSE 支持，路由直接yield事件：

# app/api/chat.py from collections.abc import AsyncIterable from fastapi import APIRouter, Depends, Request from fastapi.sse import EventSourceResponse, ServerSentEvent from app.api.deps import require_api_key from app.schemas import ChatRequest from app.services.llm import stream_chat router = APIRouter(dependencies=[Depends(require_api_key)]) @router.post("/chat/stream", response_class=EventSourceResponse) async def chat_stream( body: ChatRequest, request: Request, ) -> AsyncIterable[ServerSentEvent]: model = request.app.state.model yield ServerSentEvent( event="meta", data={"request_id": getattr(request.state, "request_id", None)}, ) try: async for token in stream_chat(model, body.messages): yield ServerSentEvent(event="token", data={"text": token}) except Exception as exc: # 流已开始，只能通过事件报错 yield ServerSentEvent( event="error", data={"code": "GENERATION_FAILED", "message": str(exc)[:300]}, ) return yield ServerSentEvent(event="done", data={"finish_reason": "stop"})

ServerSentEvent的data会自动 JSON 序列化。字符串也会被加引号；如果要发原始文本，用raw_data。

版本提示：fastapi.sse是较新版本才有的模块。如果你的 FastAPI 版本较旧，用下一节的StreamingResponse写法，效果等价。

5.5 兼容写法：StreamingResponse

import json from collections.abc import AsyncIterator from fastapi.responses import StreamingResponse def sse_encode(event: str, data: dict) -> str: payload = json.dumps(data, ensure_ascii=False) return f"event: {event}\ndata: {payload}\n\n" async def event_stream(body: ChatRequest, model) -> AsyncIterator[str]: yield sse_encode("meta", {"model": "..."}) try: async for token in stream_chat(model, body.messages): yield sse_encode("token", {"text": token}) except Exception as exc: yield sse_encode("error", {"code": "GENERATION_FAILED", "message": str(exc)[:300]}) return yield sse_encode("done", {"finish_reason": "stop"}) @router.post("/chat/stream") async def chat_stream_compat(body: ChatRequest, request: Request): return StreamingResponse( event_stream(body, request.app.state.model), media_type="text/event-stream", headers={ "Cache-Control": "no-cache, no-transform", "Connection": "keep-alive", "X-Accel-Buffering": "no", # 关闭 Nginx 缓冲 }, )

5.6 心跳：长生成场景的必需品

模型思考 30 秒不出 token 很常见。很多代理在 60 秒无数据时断开连接。解法是定期发:注释行（SSE 心跳）。

一个把“数据流”和“定时心跳”合并的通用实现：

import asyncio from collections.abc import AsyncIterator async def stream_with_heartbeat( source: AsyncIterator[str], interval: float = 15.0, ) -> AsyncIterator[tuple[str, str]]: queue: asyncio.Queue[tuple[str, str | None]] = asyncio.Queue() async def pump() -> None: try: async for token in source: await queue.put(("token", token)) await queue.put(("end", None)) except asyncio.CancelledError: raise except Exception as exc: await queue.put(("error", str(exc))) task = asyncio.create_task(pump()) try: while True: try: kind, payload = await asyncio.wait_for(queue.get(), timeout=interval) except TimeoutError: yield ("heartbeat", None) continue if kind == "end": return if kind == "error": yield ("error", payload) return yield ("token", payload or "") finally: task.cancel() with contextlib.suppress(asyncio.CancelledError): await task

然后在路由里：

async for kind, payload in stream_with_heartbeat(stream_chat(...)): if kind == "heartbeat": yield ServerSentEvent(comment="ping") elif kind == "token": yield ServerSentEvent(event="token", data={"text": payload}) elif kind == "error": yield ServerSentEvent(event="error", data={"message": payload}) return

这个模式还顺便解决了“客户端断开时取消上游生成”的问题：生成器被取消时，finally会 cancel 掉pump任务，进而取消模型流。否则用户关掉页面，你还在为 token 付费。

5.7 流式的六个坑

1. Nginx / 网关缓冲：必须proxy_buffering off，并返回X-Accel-Buffering: no；

2. GZipMiddleware 缓冲：压缩中间件会攒数据，SSE 路由上要绕开或关闭；

3. Content-Length：流式响应不能带固定长度，交给 chunked；

4. 错误时机：流开始前的错误可以返回 4xx/5xx；开始后只能发error事件；

5. 客户端断连：不取消上游，就会持续计费；用生成器的finally或request.is_disconnected()；

6. 超时层级：Nginxproxy_read_timeout、负载均衡空闲超时、应用整体超时，必须从外到内递增。

6. 鉴权、限流与多租户

6.1 服务间鉴权

# app/api/deps.py import secrets from fastapi import Depends, HTTPException, status from fastapi.security import APIKeyHeader from app.config import Settings, get_settings api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False) async def require_api_key( key: str | None = Depends(api_key_header), settings: Settings = Depends(get_settings), ) -> None: if not key or not secrets.compare_digest(key, settings.service_api_key): raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="invalid api key", )

面向终端用户时，换成 JWT / OAuth2；Depends的写法不变。关键是把“当前用户”变成依赖的返回值，后续接口直接拿：

async def current_user(token: str = Depends(oauth2_scheme)) -> User: ...

6.2 限流

单机可以用slowapi；多实例必须用共享存储（Redis）。一个简单的令牌桶依赖：

async def rate_limit(user: User = Depends(current_user)) -> None: allowed = await redis.eval(RATE_LIMIT_LUA, 1, f"rl:{user.id}", 60, 30) if not allowed: raise HTTPException(429, "rate limit exceeded")

限流维度建议至少两层：

• 请求数：每秒/每分钟多少次；

• token 预算：每天/每月多少 token 或多少钱。LLM 应用里，后者才是成本失控的真正来源。

6.3 多租户隔离

• 每个请求带tenant_id，从鉴权结果推导，不要信任请求体里的 tenant 字段；

• 向量库、数据库、检查点都要按 tenant 做命名空间或行级隔离；

• 日志和 trace 带tenant_id，否则线上排查无法按客户切分；

• 模型 API Key 可以按租户做网关转发，便于配额和审计。

7. 错误处理与可观测性

7.1 统一异常

from fastapi import Request from fastapi.responses import JSONResponse class AppError(Exception): def __init__(self, code: str, message: str, status_code: int = 400): self.code = code self.message = message self.status_code = status_code @app.exception_handler(AppError) async def app_error_handler(request: Request, exc: AppError): return JSONResponse( status_code=exc.status_code, content={"code": exc.code, "message": exc.message, "request_id": getattr(request.state, "request_id", None)}, )

原则：

• 对外返回稳定的错误码，不要把上游 traceback、密钥、内部 URL 暴露出去；

• 对内记录完整异常 +request_id；

• 区分“可重试”（超时、429、5xx）和“不可重试”（参数错误、内容审核拒绝）。

7.2 Request ID 中间件

import uuid from fastapi import Request @app.middleware("http") async def request_id_middleware(request: Request, call_next): rid = request.headers.get("X-Request-ID") or uuid.uuid4().hex request.state.request_id = rid response = await call_next(request) response.headers["X-Request-ID"] = rid return response

这个 ID 要贯穿：HTTP 响应头 → 结构化日志 → LangSmith/Langfuse trace metadata → 前端错误上报。否则用户说“刚才那条回答不对”，你无从查起。

7.3 结构化日志

import logging, json logger = logging.getLogger("app") def log_event(event: str, **fields): logger.info(json.dumps({"event": event, **fields}, ensure_ascii=False))

至少记录：request_id、tenant_id、model、input_tokens、output_tokens、latency_ms、tool_calls、error_code。这些字段后面可以直接喂给 Langfuse 的 metadata 或自建看板。

8. 测试：不要只测“200 OK”

8.1 lifespan 必须被触发

from fastapi.testclient import TestClient from app.main import app def test_chat_stream(): with TestClient(app) as client: # with 才会执行 lifespan with client.stream( "POST", "/v1/chat/stream", json={"messages": [{"role": "user", "content": "你好"}]}, ) as response: assert response.status_code == 200 body = "".join(response.iter_text()) assert "event: token" in body assert "event: done" in body

8.2 异步测试

import httpx import pytest from httpx import ASGITransport from app.main import app @pytest.mark.anyio async def test_health(): transport = ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: resp = await client.get("/health") assert resp.status_code == 200

8.3 用假模型做确定性测试

不要在 CI 里真的调模型。把模型客户端做成依赖，测试时替换：

class FakeModel: async def astream(self, messages): for token in ["你", "好"]: yield type("Chunk", (), {"content": token})() app.dependency_overrides[get_model] = lambda: FakeModel()

需要覆盖的场景：

• 正常流式；

• 上游超时；

• 上游 429；

• 流中途异常；

• 客户端断连（生成器被取消）；

• 超长输入被 422 拒绝；

• 无鉴权 / 错误鉴权。

9. 部署

9.1 进程模型

uvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 4

• --workers是多进程，不是多线程；

• 每个 worker 有独立内存：内存里的会话、限流计数、检查点都不共享；

• LLM 应用大部分时间在等网络，worker 数不需要按 CPU 核数死算，按并发连接和内存压测；

• 生产更常见的是“一个容器一个 worker”，由 K8s 做水平扩展。

9.2 Dockerfile

FROM python:3.12-slim ENV PYTHONDONTWRITEBYTECODE=1 \ PYTHONUNBUFFERED=1 WORKDIR /app COPY pyproject.toml uv.lock ./ RUN pip install --no-cache-dir uv && uv sync --frozen --no-dev COPY app ./app EXPOSE 8000 CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--proxy-headers"]

9.3 健康检查

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

区分：

• liveness：进程活着即可，不要在里面调模型；

• readiness：依赖是否可用（数据库、Redis），失败时摘流量；

• startup：初始化慢时用，避免被误杀。

9.4 Nginx 关键配置

location /v1/chat/stream { proxy_pass http://api_upstream; proxy_http_version 1.1; proxy_set_header Connection ""; proxy_buffering off; proxy_cache off; proxy_read_timeout 3600s; chunked_transfer_encoding on; }

10. 上线前检查清单

• 模型调用全部走异步，或明确放在线程池
• 每层都有超时：连接、读、整体预算、网关
• SSE 关闭了代理缓冲，并有心跳
• 客户端断开会取消上游生成
• 流开始后的错误通过error事件返回
• 鉴权从依赖注入，租户 ID 不可由请求体伪造
• 有限流，且有 token / 成本预算
• 所有日志带request_id，并能关联到 trace
• 测试覆盖超时、429、流中断、断连
• 多 worker 下没有依赖进程内共享状态
• 健康检查不会调用模型
• 镜像里没有.env和密钥

小结

FastAPI 在 LLM 系统里的角色，是把不确定的模型调用包装成确定的 HTTP 契约。这一层做稳了，后面的 LangChain / LangGraph 才能安心地专注于编排逻辑。

下一篇进入模型调用层：02 | LangChain 1.x：模型、工具与 Agent。我们会讲清楚 1.0 到底改了什么、消息格式怎么统一、工具调用和结构化输出的底层机制，以及什么时候该用create_agent，什么时候该下沉到 LangGraph。