先说结论LLM 应用的 HTTP 层90% 的工作不是“写个接口”而是处理四件事异步 IO模型调用是网络等待不能让一个请求堵住整个进程流式输出首 token 延迟决定体感SSE 是默认选择资源生命周期HTTP 连接池、数据库、检查点、模型客户端必须在启动时建、关闭时收失败模式超时、断连、限流、上游 5xx、流开始后的错误——这些不会因为 Demo 能跑就消失。这一篇从最小服务开始一步步把这些补上。代码可以直接作为后续 LangGraph 项目的底座。1. 为什么 LLM 服务默认选 FastAPI需求FastAPI 的优势大量网络等待原生async/await基于 Starlette/ASGI单进程可扛高并发长连接输入输出结构复杂Pydantic v2 做请求校验、响应序列化、OpenAPI 文档流式响应支持StreamingResponse新版内置fastapi.sse依赖管理Depends适合注入配置、数据库、模型客户端、鉴权测试TestClient ASGI transport无需真的起端口生态LangServe、LangGraph Platform、OpenAI 兼容网关都能对接不适合的场景也要说清楚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, extraignore) 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( timeouthttpx.Timeout(settings.request_timeout_seconds, connect10.0), limitshttpx.Limits(max_connections100, max_keepalive_connections20), ) # 模型客户端是线程安全的进程内复用即可 app.state.model init_chat_model( settings.chat_model, temperature0.2, timeoutsettings.request_timeout_seconds, ) yield await app.state.http.aclose() def create_app() - FastAPI: settings get_settings() app FastAPI( titlesettings.app_name, version0.1.0, lifespanlifespan, 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 上所有请求都卡住。正确做法有三种优先用异步客户端httpx.AsyncClient、openai.AsyncOpenAI、LangChain 的ainvoke/astream只有同步客户端时把路由声明成def不是async defFastAPI 会自动放进线程池必须在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_length1, max_length20_000) class ChatRequest(BaseModel): messages: list[ChatMessage] Field(min_length1, max_length50) session_id: str | None Field(defaultNone, max_length128) 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 三种方案怎么选方案方向适用普通 JSON 一次性返回请求-响应短输出、后台任务SSEtext/event-stream服务器 → 客户端单向聊天、生成、进度推送WebSocket双向语音、实时协作、需要客户端中途发指令SSE 的优势是简单基于 HTTP自带重连语义代理和负载均衡友好配置正确的前提下。浏览器原生EventSource只支持 GET所以很多聊天产品用fetchReadableStream解析 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](contentm.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_classEventSourceResponse) async def chat_stream( body: ChatRequest, request: Request, ) - AsyncIterable[ServerSentEvent]: model request.app.state.model yield ServerSentEvent( eventmeta, data{request_id: getattr(request.state, request_id, None)}, ) try: async for token in stream_chat(model, body.messages): yield ServerSentEvent(eventtoken, data{text: token}) except Exception as exc: # 流已开始只能通过事件报错 yield ServerSentEvent( eventerror, data{code: GENERATION_FAILED, message: str(exc)[:300]}, ) return yield ServerSentEvent(eventdone, data{finish_reason: stop})ServerSentEvent的data会自动 JSON 序列化。字符串也会被加引号如果要发原始文本用raw_data。版本提示fastapi.sse是较新版本才有的模块。如果你的 FastAPI 版本较旧用下一节的StreamingResponse写法效果等价。5.5 兼容写法StreamingResponseimport json from collections.abc import AsyncIterator from fastapi.responses import StreamingResponse def sse_encode(event: str, data: dict) - str: payload json.dumps(data, ensure_asciiFalse) return fevent: {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_typetext/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(), timeoutinterval) 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(commentping) elif kind token: yield ServerSentEvent(eventtoken, data{text: payload}) elif kind error: yield ServerSentEvent(eventerror, data{message: payload}) return这个模式还顺便解决了“客户端断开时取消上游生成”的问题生成器被取消时finally会 cancel 掉pump任务进而取消模型流。否则用户关掉页面你还在为 token 付费。5.7 流式的六个坑Nginx / 网关缓冲必须proxy_buffering off并返回X-Accel-Buffering: noGZipMiddleware 缓冲压缩中间件会攒数据SSE 路由上要绕开或关闭Content-Length流式响应不能带固定长度交给 chunked错误时机流开始前的错误可以返回 4xx/5xx开始后只能发error事件客户端断连不取消上游就会持续计费用生成器的finally或request.is_disconnected()超时层级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(nameX-API-Key, auto_errorFalse) 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_codestatus.HTTP_401_UNAUTHORIZED, detailinvalid api key, )面向终端用户时换成 JWT / OAuth2Depends的写法不变。关键是把“当前用户”变成依赖的返回值后续接口直接拿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, frl:{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_codeexc.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_asciiFalse))至少记录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 body8.2 异步测试import httpx import pytest from httpx import ASGITransport from app.main import app pytest.mark.anyio async def test_health(): transport ASGITransport(appapp) async with httpx.AsyncClient(transporttransport, base_urlhttp://test) as client: resp await client.get(/health) assert resp.status_code 2008.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 DockerfileFROM python:3.12-slim ENV PYTHONDONTWRITEBYTECODE1 \ PYTHONUNBUFFERED1 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。
