更新一手 在 Open WebUI 中使用硅基流动的绘图模型
更新内容
- 一次绘制多张图片
- 添加 seed 参数
- 优化错误处理
在 User Valves 中配置图片数量以及 seed（可选）：
更新方式
- 手动更新
如果之前导入过这个函数，可以选择编辑，然后将新代码粘贴保存。
import aiohttp
import asyncio
from typing import Optional
from pydantic import BaseModel, Field
async def emit(emitter, msg, done):
await emitter(
{
"type": "status",
"data": {
"done": done,
"description": msg,
},
}
)
class Filter:
class Valves(BaseModel):
priority: int = Field(
default=0,
description="Priority level for the filter operations.",
)
api_url: str = Field(
default="https://api.siliconflow.cn/v1",
description="Base URL for the Siliconflow API.",
)
api_key: str = Field(
default="",
description="API Key for the Siliconflow API.",
)
class UserValves(BaseModel):
size: str = Field(
default="1024x1024",
description="1024x1024, 512x1024, 768x512, 768x1024, 1024x576, 576x1024.",
)
steps: int = Field(
default=20,
description="The number of inference steps to be performed (1-100).",
)
model: str = Field(
default="black-forest-labs/FLUX.1-dev",
description="The name of the model.",
)
pnum: int = Field(
default=1,
description="The number of pictures.",
)
seed: Optional[int] = Field(
default=None,
description="The seed.",
)
def __init__(self):
self.valves = self.Valves()
async def inlet(self, body, __user__, __event_emitter__):
await emit(__event_emitter__, "Generating prompt, please wait...", False)
return body
async def request(self, prompt, __user__):
url = f"{self.valves.api_url}/image/generations"
headers = {
"accept": "application/json",
"content-type": "application/json",
"authorization": f"Bearer {self.valves.api_key}",
}
payload = {
"prompt": prompt,
"model": __user__["valves"].model,
"image_size": __user__["valves"].size,
"num_inference_steps": __user__["valves"].steps,
}
if seed := __user__["valves"].seed:
payload["seed"] = seed
pnum = __user__["valves"].pnum
async with aiohttp.ClientSession() as sess:
tasks = [sess.post(url, json=payload, headers=headers) for _ in range(pnum)]
res = await asyncio.gather(*tasks)
ret = []
for i, r in enumerate(res):
if (s := r.status) == 200:
json = await r.json()
url = json["images"][0]["url"]
ret.append(f"")
else:
text = await r.text()
ret.append(f"> The {i} request failed ({s}): {text}.")
return ret
async def outlet(self, body, __user__, __event_emitter__):
await emit(__event_emitter__, f"Generating pictures, please wait...", False)
last = body["messages"][-1]
res = await self.request(last["content"], __user__)
for r in res:
last["content"] += f"\n\n{r}"
await emit(
__event_emitter__, f"Generated successfully, click to preview!", True
)
return body
- 官网导入
新代码我也同步到 Open WebUI 官网了，可以删除重新导入（导入方式见旧帖）：地址。
大佬