# FastAPI Transformers 非阻塞流式响应实现

> 在 FastAPI 中使用 Transformers 的 TextIteratorStreamer 实现非阻塞的异步流式响应，避免模型推理阻塞事件循环。

- Skill: `ecnu-icalk/fastapi-transformers` (Agent Skill)
- Install (CLI): `npx skillmds@latest add ecnu-icalk/fastapi-transformers`
- Raw SKILL.md: https://api.skillmd.com/api/skills/ecnu-icalk/fastapi-transformers/raw
- Safety review: pending (external: skill-scanner PASS, skillspector PASS)
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Coding & Dev Tools
- Author: ECNU-ICALK (https://skillmd.com/u/ecnu-icalk)
- Updated: 2026-09-08
- Page: https://skillmd.com/skills/ecnu-icalk/fastapi-transformers

---


# 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 异步流式响应