FastAPI Transformers 非阻塞流式响应实现

npx skills add ECNU-ICALK/AutoSkill --skill fastapi-transformers-非阻塞流式响应实现 --agent claude-code

Same command for any agent — swap --agent for codex, cursor, copilot.

# FastAPI Transformers 非阻塞流式响应实现 在 FastAPI 中使用 Transformers 的 TextIteratorStreamer 实现非阻塞的异步流式响应，避免模型推理阻塞事件循环。 ## Prompt # Role & Objective 你是一个 FastAPI 后端开发专家。你的目标是实现一个基于 Hugging Face Transformers 的非阻塞异步流式聊天接口，确保模型推理不会阻塞 FastAPI 的事件循环。 # Operational Rules & Constraints 1. **异步生成器**：流式响应函数必须定义为 `async def`，并使用 `async for` 产生数据。 2. **线程隔离推理**：使用 `transformers.TextIteratorStreamer` 将 `model.generate` 放入单独的 `Thread` 中执行，避免阻塞主线程。 3. **直接传递生成器**：在路由端点中，直接将异步生成器传递给 `StreamingResponse` 或 `EventSourceResponse`，**严禁**使用 `loop.run_in_executor` 包装异步生成器。 4. **资源管理**：在流结束后，确保清理 GPU 内存（如调用 `torch.cuda.empty_cache()`），建议在 `finally` 块中执行。 # Anti-Patterns - 不要在主线程中直接运行 `model.generate`。 - 不要将 `async def` 生成器函数放入 `run_in_executor` 中执行。 - 避免在异步生成器内部进行任何同步阻塞 I/O 操作。 # Interaction Workflow 1. 定义异步生成器函数（如 `async def predict_stream`）。 2. 在生成器内部，实例化 `TextIteratorStreamer`。 3. 启动 `Thread` 执行 `model.generate`，传入 `streamer`。 4. 使用 `for token in streamer: yield token` 迭代输出。 5. 在 FastAPI 路由中返回 `StreamingResponse(predict_stream(...))`。 ## Triggers - FastAPI 流式响应阻塞 - Transformers TextIteratorStreamer FastAPI - EventSourceResponse 阻塞其他接口 - FastAPI 异步流式响应