用 FastAPI 写了一个 AI 应用的后端，踩了这些坑

最近在做一个 AI 聊天应用的后端，用的 FastAPI。功能不复杂，但在流式响应和并发处理上踩了不少坑。 ## 流式响应 AI 聊天的回复需要一个字一个字地流式返回，不能等整个回复生成完再返回（那样用户要等好几秒才能看到第一个字）。 FastAPI 的 `StreamingResponse` 可以做到： ```python from fastapi.responses import StreamingResponse @app.post("/chat") async def chat(request: ChatRequest): async def generate(): async for chunk in call_llm_stream(request.messages): yield f"data: {json.dumps({'content': chunk})}\n\n" yield "data: [DONE]\n\n" return StreamingResponse(generate(), media_type="text/event-stream") ``` ### 坑 1：Nginx 缓冲 部署到线上后发现流式响应变成了"一坨一坨"返回，不是一个字一个字的。原因是 Nginx 默认会缓冲后端的响应。解决： ```nginx location /api/ { proxy_buffering off; proxy_cache off; } ``` ### 坑 2：超时 大模型生成长文本可能要 30 秒以上，Nginx 和 uvicorn 的默认超时都不够。需要调大： ```nginx proxy_read_timeout 300s; proxy_send_timeout 300s; ``` ```bash uvicorn main:app --timeout-keep-alive 300 ``` ## 并发处理 ### 坑 3：同步阻塞 调用大模型 API 如果用了同步的 requests 库，会阻塞整个事件循环，其他请求全都等着。必须用异步的 httpx： ```python import httpx async def call_llm_stream(messages): async with httpx.AsyncClient() as client: async with client.stream("POST", LLM_API_URL, json={"messages": messages}) as response: async for line in response.aiter_lines(): if line.startswith("data: "): yield line[6:] ``` ### 坑 4：并发限制 大模型 API 通常有并发限制（比如同时最多 10 个请求）。需要用 Semaphore 控制： ```python llm_semaphore = asyncio.Semaphore(10) async def call_llm_with_limit(messages): async with llm_semaphore: return await call_llm(messages) ``` ## Token 计算 ### 坑 5：上下文长度溢出 用户的对话越来越长，消息列表越来越大，很容易超过模型的上下文限制。需要在发送前截断历史消息： ```python def trim_messages(messages, max_tokens=4000): total = 0 trimmed = [] for msg in reversed(messages): tokens = count_tokens(msg["content"]) if total + tokens > max_tokens: break trimmed.insert(0, msg) total += tokens return trimmed ``` 总的来说 FastAPI 做 AI 应用后端还是很合适的，异步特性和类型提示都很加分。但上面这些坑基本每个人都会踩一遍，希望能帮到正在做类似项目的同学。