其他
Flux 1.0：使用Streamlit和Replicate API批量生成AI图像 | 附完整代码
引言
Flux 1.0 是由 Black Forest Labs 开发的文本到图像合成模型，利用 AI 技术从文本生成高质量图像。它包含三个针对不同用途优化的模型：
FLUX.1 [pro]：旗舰模型，提供卓越的图像质量、细节和多样性，适合专业应用。 FLUX.1 [dev]：pro 模型的蒸馏变体，适合研究和开发，非商业用途，权重可在 HuggingFace 获取。 FLUX.1 [schnell]：速度优化模型，适合本地开发和个人项目，支持高效本地部署。
Flux dev 8月初发布，现在是最强开源模型。
Flux模型示例
以下是我用Flux dev和schnell生成的图片
在不同平台上使用Flux
我在Poe、Replicate、Fal、Grok、本地应用Draw things APP 都试过Flux，效果很不错，但也有些问题。本地跑 FLUX.1 [dev] 8bit版本，电脑（32G内存）风扇呜呜响，生成时间也比较长。
使用Streamlit和Replicate API批量生成图片
今天，介绍如何通过Streamlit调用Replicate API来批量生成图片。
Replicate积分活动
Replicate有一个10美元积分活动，点击即可领取
https://replicate.com/invites/572ae3c6-1385-41b3-afdc-1e99e11836b8
Replicate图片生成效率
用Replicate生成的单张图片大小约1.5M。（100%png，schnell和dev模型）
1美元可以生成333张schnell图片或33张dev图片。
使用FLUX.1 [schnell]批量生成图片
非常推荐用FLUX.1 [schnell]来批量生成图片
Streamlit应用实例
我编写了Streamlit应用眨眼之间就能生成图片。
我设定的逻辑是上传txt，txt里每一段就是一个提示词，可以批量生成图片。
上面小猫找妈妈风格一致的图片是这样制作的：
和Claude沟通AI绘图提示词
对应的中文如下：
"超写实高分辨率图像。一只小橘猫，绿色的大眼睛，粉红色的鼻子和白色的爪子，独自坐在一个舒适的藤篮里。小猫看起来困惑且担忧，毛发的细节清晰可见。柔和的晨光从附近的窗户洒进来，营造出逼真的阴影，照亮空气中的灰尘颗粒。场景捕捉到了篮子的纹理、小猫毛发的柔软感，以及光影交错的摄影般的准确性。"
"超写实高分辨率图像。同一只绿色眼睛、粉红色鼻子和白色爪子的小橘猫小心翼翼地探索着阳光充足的客厅。小猫偷偷瞥向家具下方和窗帘后面，它的胡须和单根毛发清晰可见。阳光在柔软的地毯上形成了图案，每一根纤维都清晰分明。这种仿照片的画质展现了表面的反光效果和类似专业摄影的景深。"
"超写实高分辨率图像。勇敢的小橘猫，有着显眼的绿色眼睛和白色的爪子，进入了一个郁郁葱葱的后院花园。高耸的花朵和草围绕着这个小小的生灵，每一片花瓣和草叶都以清晰的细节呈现。附近有蝴蝶翩翩起舞，它们脆弱的翅膀被捕捉到在飞行的瞬间。高分辨率风格展现了小猫每一根胡须和它毛色的细微变化。"
"超写实高分辨率图像。雨开始下了，小橘猫带着绿色眼睛，找到了一个木质门廊下的庇护所。它的白色爪子上沾着一点泥，质感真实。水滴在附近的小水坑里激起了涟漪，逼真地反映出小猫的脸。场景捕捉了小猫湿润的毛发、门廊木纹的细节，以及雨滴折射光线的效果。"
"超写实高分辨率图像。这只坚定的小橘猫，绿色的眼睛充满警觉，白色的爪子略显疲惫，继续在安静的郊区人行道上寻找。随着黄昏的来临，街灯发出温暖的光芒，在小猫的毛发上创造出逼真的光效。高分辨率风格显示了水泥人行道的质地，边缘草叶的细节，以及光影的微妙渐变。"
"超写实高分辨率图像。最终，小橘猫带着绿色的眼睛和白色的爪子，与同样毛色的母猫喜悦地团聚了。他们在家门前亲密地互相蹭着。超写实风格捕捉到了每一个细节：它们毛发的柔软，眼中的光泽，鼻子的纹理，和每一根胡须。场景的灯光突出了重逢时的情感温暖，猫咪们被锐利地聚焦，背景略显模糊。"
运行程序时，在高级设置里固定种子。
因为Replicate里3个模型参数稍有差别，我设定的Streamlit应用可选schnell和dev来批量生成图片。
完整代码
import streamlit as st
import replicate
import io
import base64
from zipfile import ZipFile
import requests
import asyncio
from PIL import Image
import logging
# 设置日志记录
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# 设置页面标题
st.set_page_config(page_title="Flux 图像生成器", layout="wide")
# 标题
st.title("Flux 图像生成器")
# 选择模型
model = st.selectbox(
"选择模型",
["black-forest-labs/flux-schnell", "black-forest-labs/flux-dev"],
help="选择要使用的Flux模型"
)
# 输入提示词
prompt = st.text_area("输入提示词", help="描述您想生成的图像")
# 上传txt文件
uploaded_file = st.file_uploader("上传提示词文件 (每行一个提示词)", type="txt")
# 设置参数
with st.expander("高级设置"):
seed = st.number_input("随机种子", min_value=0, help="设置随机种子以获得可重复的生成结果")
num_outputs = st.slider("每个提示词的输出数量", min_value=1, max_value=4, value=1, help="每个提示词生成的图像数量")
aspect_ratio = st.selectbox(
"宽高比",
["16:9", "1:1", "21:9", "2:3", "3:2", "4:5", "5:4", "9:16", "9:21"],
index=0,
help="生成图像的宽高比"
)
output_format = st.selectbox("输出格式", ["png", "webp", "jpg"], index=0, help="输出图像的格式")
output_quality = st.slider("输出质量", min_value=0, max_value=100, value=100, help="输出图像的质量,从0到100。100是最佳质量,0是最低质量。对于.png输出不相关")
disable_safety_checker = st.checkbox("禁用安全检查器", help="禁用生成图像的安全检查器。此功能仅通过API可用。")
async def generate_image_async(prompt, model, input_data, timeout=45):
try:
prediction = await replicate.predictions.async_create(
model=model,
input=input_data
)
start_time = asyncio.get_event_loop().time()
while prediction.status != "succeeded":
if asyncio.get_event_loop().time() - start_time > timeout:
raise asyncio.TimeoutError(f"生成图像超时（{timeout}秒）")
await asyncio.sleep(1)
prediction = await replicate.predictions.async_get(prediction.id)
logger.info(f"API 响应: {prediction.output}")
logger.info(f"完整的 API 响应: {prediction}")
if prediction.output:
if isinstance(prediction.output, list):
return prediction.output[0] if prediction.output else None
elif isinstance(prediction.output, str):
return prediction.output
else:
logger.error(f"未知的 API 响应格式: {type(prediction.output)}")
return None
else:
logger.error("API 响应为空")
return None
except asyncio.TimeoutError as e:
logger.error(str(e))
return None
except Exception as e:
logger.error(f"生成图像时发生错误: {str(e)}")
return None
def get_image_download_link(img_url, filename):
response = requests.get(img_url)
response.raise_for_status()
img_data = response.content
b64 = base64.b64encode(img_data).decode()
file_size = len(img_data)
logger.info(f"下载的图像大小: {file_size} 字节")
return f'<a href="data:image/png;base64,{b64}" download="{filename}">下载图片 ({file_size/1024:.2f} KB)</a>'
# 生成按钮
if st.button("生成图像"):
prompts = []
if uploaded_file:
prompts = [line.decode("utf-8").strip() for line in uploaded_file]
elif prompt:
prompts = [prompt]
if not prompts:
st.error("请输入提示词或上传提示词文件")
else:
with st.spinner(f"正在生成 {len(prompts) * num_outputs} 张图像..."):
async def generate_all_images():
tasks = []
for current_prompt in prompts:
for _ in range(num_outputs):
input_data = {
"prompt": current_prompt,
"seed": seed if seed else None,
"aspect_ratio": aspect_ratio,
"output_format": output_format,
"output_quality": output_quality,
"disable_safety_checker": disable_safety_checker
}
input_data = {k: v for k, v in input_data.items() if v is not None}
tasks.append(generate_image_async(current_prompt, model, input_data, timeout=45))
return await asyncio.gather(*tasks)
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
generated_images = loop.run_until_complete(generate_all_images())
# 显示生成的图像
for i, image_url in enumerate(generated_images):
if image_url:
logger.info(f"处理图像 URL: {image_url}")
col1, col2 = st.columns([3, 1])
with col1:
try:
if not image_url.startswith(('http://', 'https://')):
raise ValueError(f"无效的 URL: {image_url}")
response = requests.get(image_url)
response.raise_for_status()
image = Image.open(io.BytesIO(response.content))
logger.info(f"图像 {i+1} 尺寸: {image.size}")
st.image(image, caption=f"生成的图像 {i+1}", use_column_width=True)
logger.info(f"成功显示图像 {i+1}")
except requests.RequestException as e:
logger.error(f"下载图像时发生错误: {str(e)}")
st.error(f"无法加载图像 {i+1}: {str(e)}")
except ValueError as e:
logger.error(str(e))
st.error(f"无效的图像 URL {i+1}: {str(e)}")
except Exception as e:
logger.error(f"处理图像时发生未知错误: {str(e)}")
st.error(f"处理图像 {i+1} 时发生错误: {str(e)}")
with col2:
st.markdown(get_image_download_link(image_url, f"generated_image_{i+1}.{output_format}"), unsafe_allow_html=True)
st.text_area("提示词", prompts[i // num_outputs], height=100)
else:
logger.warning(f"图像 {i+1} 的 URL 为空")
# 批量下载
if len(generated_images) > 1:
zip_buffer = io.BytesIO()
with ZipFile(zip_buffer, 'w') as zip_file:
for i, image_url in enumerate(generated_images):
if image_url:
response = requests.get(image_url)
zip_file.writestr(f"generated_image_{i+1}.{output_format}", response.content)
zip_buffer.seek(0)
b64 = base64.b64encode(zip_buffer.getvalue()).decode()
href = f'<a href="data:application/zip;base64,{b64}" download="generated_images.zip">下载所有图片</a>'
st.markdown(href, unsafe_allow_html=True)
# 添加说明
st.markdown("""
## 使用说明
- 选择要使用的Flux模型。
- 在文本框中输入描述您想要生成的图像的提示词，或上传包含多个提示词的txt文件（每行一个提示词）。
- 如果需要，可以展开“高级设置”调整其他参数。
- 点击“生成图像”按钮开始生成过程。
- 生成的图像将显示在页面上，您可以单独下载每张图片或批量下载所有图片。
注意：确保您已经设置了REPLICATE_API_TOKEN环境变量，否则API调用将失败。
""")
运行说明
环境准备：
确保您的系统已安装Python（建议使用Python 3.9或更高版本）。
安装依赖：
打开命令行终端。
安装必要的库。您可以使用以下命令：
pip install streamlit replicate pillow requests
设置API令牌：
在Windows上： set REPLICATE_API_TOKEN=your_token_here在Mac或Linux上： export REPLICATE_API_TOKEN=your_token_here您需要一个Replicate API令牌。如果没有，请在Replicate网站上注册并获取令牌。 设置环境变量REPLICATE_API_TOKEN。在终端中运行：
保存代码：
将提供的代码保存为一个Python文件，例如 flux_image_generator.py。
运行应用：
在终端中，导航到保存代码的目录。
运行以下命令启动Streamlit应用：
streamlit run flux_image_generator.py
使用应用：
Streamlit会自动在您的默认web浏览器中打开应用。 如果没有自动打开，您可以在浏览器中访问终端显示的URL（通常是 http://localhost:8501）。
在应用中：
选择Flux模型。 输入提示词或上传包含提示词的txt文件。 调整高级设置（如果需要）。 点击"生成图像"按钮。
查看结果：
生成的图像将显示在页面上。 您可以下载单个图像或所有图像的压缩包。
注意事项：
确保您的网络连接良好，因为程序需要与Replicate API进行通信。 如果遇到任何错误，请检查终端输出以获取更多信息。
结语
通过结合Streamlit和Replicate API，我们可以轻松实现批量图像生成，大大提高了效率。
希望本文的介绍和代码示例能够激发您的创意，帮助您在AI图像生成领域展开探索。
广告
过去我已写了170+篇AI主题原创长文，我对继续写作充满信心，因为这是我的爱好，我非常热爱这件事。
最近我开通了知识星球，你加入后可以访问公众号收费文章，向我提问，第一时间获取AI资讯。
精选历史文章，请看这里：
解锁 Claude 3.5 Sonnet 创意潜能：10+ 个 Web 应用实战