最后使用时间： 2026-03-24
记忆引用次数： 2
状态： 活跃

FastAPI Studio 模板

为生成式 AI 对比、A/B 测试和人工评估搭建深色主题的 FastAPI + HTMX Studio 应用，支持实时进度流式传输。

适用场景

• - 任何工作室类应用：图像生成对比、文本模型 A/B 测试、人工评估界面
• 需要实时进度更新的应用（生成过程可能耗时 30 秒至 15 分钟）
• 盲测/评估界面：评估者不应知道哪个模型生成了哪个输出
• 生成式 AI 对比工具的快速原型开发

不适用场景

• - 简单的 CRUD 应用（请使用标准 FastAPI + Jinja2）
• 无需实时进度的应用（SSE 会增加复杂性）
• 超过 100 个并发用户的生产级应用（请使用 WebSocket 替代 SSE）

核心模式

SSE 异步模式（关键）

必须使用 threading.SimpleQueue + asyncio 轮询。 不要使用带阻塞读取的 runinexecutor——这会导致事件循环死锁。

python
import asyncio
import threading
from queue import SimpleQueue

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

async def event_stream(queue: SimpleQueue):
从线程安全队列中生成 SSE 事件。
while True:
try:
msg = queue.get_nowait()
except Exception:
await asyncio.sleep(0.1)
continue
if msg is None: # 哨兵值
yield fdata: {{\done\: true}}\n\n
break
yield fdata: {msg}\n\n

@app.get(/generate/stream)
async def generate_stream(prompt: str, model: str):
queue = SimpleQueue()

def _run():
# 后台线程中的繁重生成工作
for step in range(10):
import time; time.sleep(1)
queue.put(f{{step: {step}, total: 10}})
queue.put(None) # 完成哨兵值

threading.Thread(target=_run, daemon=True).start()
return StreamingResponse(eventstream(queue), mediatype=text/event-stream)

为什么不用 runinexecutor？ FastAPI 的执行器运行在线程池上，但 SSE 需要增量地生成事件。在执行器中阻塞意味着你无法流式传输部分进度——你必须等待整个生成过程完成。队列模式将生成与流式传输解耦。

盲测模式

生成 N 个变体（每个模型一个），随机化显示顺序，仅在用户对所有变体评分后揭示模型身份。

python
import random
import uuid

def createblindtest(prompt: str, models: list[str]) -> dict:
test_id = str(uuid.uuid4())
variants = []
for model in models:
variants.append({
variant_id: str(uuid.uuid4()),
model: model, # 在揭示前对 UI 隐藏
prompt: prompt,
})
random.shuffle(variants)
return {
testid: testid,
variants: variants,
displayorder: [v[variantid] for v in variants],
}

在 HTMX 前端中，将变体渲染为选项 A、选项 B等。在提交评分时，返回从选项字母到模型名称的映射。

热加载模型单例（ModelRegistry）

冷加载 SDXL 或类似模型需要 6-14 分钟。将加载的模型缓存在注册表单例中。

python
class ModelRegistry:
_instance = None
_models: dict = {}
_lock = threading.Lock()

@classmethod
def get(cls, model_name: str):
with cls._lock:
if modelname not in cls.models:
cls.models[modelname] = cls.loadmodel(model_name)
return cls.models[modelname]

@classmethod
def loadmodel(cls, name: str):
# 导入并加载模型
if name == sdxl:
from mflux import Flux1
return Flux1.from_alias(schnell, quantize=8)
raise ValueError(f未知模型: {name})

在启动时预加载：通过 FastAPI 的生命周期钩子加载你确定需要的模型。

MPS 上 SDXL 的 float32 要求

在 Apple Silicon (MPS) 上，torch 2.10 使用 float16 进行 SDXL 推理会产生 NaN 输出。 强制使用 float32：

python
import torch
torch.setdefaultdtype(torch.float32)

或按模型设置：model = model.to(dtype=torch.float32)

这会使 VRAM 使用量翻倍，但在 MPS float16 错误修复之前，这是唯一可靠的选择。

评分系统的 SQLite 模式

sql
CREATE TABLE ratings (
id INTEGER PRIMARY KEY AUTOINCREMENT,
test_id TEXT NOT NULL,
variant_id TEXT NOT NULL,
model TEXT NOT NULL,
rater TEXT DEFAULT anonymous,
score INTEGER CHECK(score BETWEEN 1 AND 5),
preferred BOOLEAN DEFAULT FALSE, -- 成对比较的胜者
notes TEXT,
createdat TIMESTAMP DEFAULT CURRENTTIMESTAMP
);

CREATE INDEX idxratingstest ON ratings(test_id);
CREATE INDEX idxratingsmodel ON ratings(model);

Langfuse 追踪

用 Langfuse 追踪包装生成调用，用于成本跟踪和延迟监控：

python
from langfuse import Langfuse

langfuse = Langfuse()

def generatewithtrace(prompt, model_name):
trace = langfuse.trace(name=studio-generation, metadata={model: model_name})
span = trace.span(name=generate, input={prompt: prompt})
result = ModelRegistry.get(model_name).generate(prompt)
span.end(output={length: len(result)})
return result

完整示例：最小化 Studio 应用

python
带有 SSE 进度的最小化 FastAPI+HTMX Studio 应用。
import asyncio
import json
import threading
from queue import SimpleQueue
from pathlib import Path

from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse, StreamingResponse
from fastapi.staticfiles import StaticFiles

app = FastAPI()

HTML =

🎨 Studio

@app.get(/, response_class=HTMLResponse)
async def index():
return HTML

async def event_stream(queue