图像生成流水线(GPU版)
1063 字约 4 分钟
2026-02-15
1. 概述
图像生成流水线的核心功能是根据用户提供的文本生成目标图片,为后续图像理解、图像编辑等任务提供图片数据。
本版本使用本地GPU模型进行文本到图片生成,支持在本地部署FLUX.1-dev等模型。
💡 提示:如果要使用云端API模型进行文本到图片生成,请查看 图像生成流水线(API版)
2. 快速开始
第一步:创建新的 DataFlow 工作文件夹
mkdir run_dataflow_mm
cd run_dataflow_mm第二步:配置模型路径
在流水线代码中配置模型路径。支持两种方式:
(1)方式1:使用Hugging Face模型路径(自动下载)
hf_model_name_or_path="black-forest-labs/FLUX.1-dev"(2)方式2:使用本地模型路径(已下载的模型)
hf_model_name_or_path="/path/to/your/local/FLUX.1-dev"在 text_to_image_generation_pipeline.py 中修改 LocalImageGenServing 的 hf_model_name_or_path 参数:
self.serving = LocalImageGenServing(
image_io=ImageIO(save_path=image_save_path),
batch_size=4,
hf_model_name_or_path="black-forest-labs/FLUX.1-dev", # 模型路径
hf_cache_dir="./cache_local", # Hugging Face模型缓存目录
hf_local_dir="./ckpt/models/", # 本地模型存储目录
diffuser_num_inference_steps=20, # 扩散模型推理步数,可调整以平衡速度和质量
diffuser_image_height=512, # 生成图片的高度
diffuser_image_width=512, # 生成图片的宽度
)第三步:准备文本数据
我们使用 jsonl 文件来保存文本数据,每行一个样本。下面是一个简单的输入数据样例:
{"conversations": [{"content": "a fox darting between snow-covered pines at dusk", "role": "user"}]}
{"conversations": [{"content": "a kite surfer riding emerald waves under a cloudy sky", "role": "user"}]}conversations 包含图片生成描述的对话列表,content 字段是其中的文本提示词。
第四步:运行流水线
python dataflow/statics/pipelines/gpu_pipelines/text_to_image_generation_pipeline.py \
--first_entry_file_name <your_input_text_file_path>生成的文件会默认保存在 ./cache_local/text2image_local 文件夹内。
3. 数据流与流水线逻辑
1. 输入数据
该流程的输入数据包括以下字段:
- conversations:对话格式数据,包含文本提示词。
这些输入数据存储在 jsonl 文件中,并通过 FileStorage 对象进行管理和读取:
self.storage = FileStorage(
first_entry_file_name="<your_jsonl_file_path>",
cache_path="./cache_local/text2image_local",
file_name_prefix="dataflow_cache_step",
cache_type="jsonl"
)2. 文本到图像生成(PromptedImageGenerator)
流程的核心步骤是使用提示式图像生成器(PromptedImageGenerator)结合本地GPU模型为每个文本提示词生成对应的图像。
功能:
- 利用本地GPU模型(如FLUX.1-dev)根据文本提示词生成图像
- 支持配置推理步数、图像尺寸等参数
- 可调整批次大小以优化GPU利用率
- 自动保存生成的图像到指定路径
输入:对话格式数据(包含文本提示词)
输出:生成的图像文件路径
本地GPU服务配置:
self.serving = LocalImageGenServing(
image_io=ImageIO(save_path=image_save_path), # 图像保存路径
batch_size=4, # 批次大小
hf_model_name_or_path="black-forest-labs/FLUX.1-dev", # 模型路径
hf_cache_dir="./cache_local", # Hugging Face模型缓存目录
hf_local_dir="./ckpt/models/", # 本地模型存储目录
diffuser_num_inference_steps=20, # 扩散模型推理步数
diffuser_image_height=512, # 生成图片的高度
diffuser_image_width=512, # 生成图片的宽度
)算子初始化:
self.text_to_image_generator = PromptedImageGenerator(
t2i_serving=self.serving, # 文本到图像服务
save_interval=10 # 保存间隔
)算子运行:
self.text_to_image_generator.run(
storage=self.storage.step(),
input_conversation_key="conversations", # 输入对话字段
output_image_key="images", # 输出图像字段
)3. 输出数据
最终,流水线生成的输出数据将包含以下内容:
- conversations:原始对话数据(包含文本提示词)
- images:生成的图像文件路径列表
输出数据示例:
{"conversations":[{"content":"a fox darting between snow-covered pines at dusk","role":"user"}],"images":["./cache_local/text2image_local/sample0_condition0/sample0_condition0_0.png"]}4. 流水线示例
下面给出使用本地FLUX模型的文本到图片生成流水线示例:
import os
from pathlib import Path
from dataflow.operators.core_vision import PromptedImageGenerator
from dataflow.serving.local_image_gen_serving import LocalImageGenServing
from dataflow.utils.storage import FileStorage
from dataflow.io import ImageIO
class ImageGenerationPipeline():
def __init__(self):
current_file = Path(__file__).resolve()
project_root = current_file.parent.parent.parent.parent.parent
prompts_file = project_root / "dataflow" / "example" / "image_gen" / "text2image" / "prompts.jsonl"
# -------- 存储配置 --------
self.storage = FileStorage(
first_entry_file_name=str(prompts_file),
cache_path="./cache_local/text2image_local",
file_name_prefix="dataflow_cache_step",
cache_type="jsonl"
)
image_save_path = str(project_root / "cache_local" / "text2image_local")
# -------- 本地GPU图像生成服务 --------
self.serving = LocalImageGenServing(
image_io=ImageIO(save_path=image_save_path),
batch_size=4,
hf_model_name_or_path="black-forest-labs/FLUX.1-dev", # 或模型在本地的路径
hf_cache_dir="./cache_local",
hf_local_dir="./ckpt/models/",
diffuser_num_inference_steps=20,
diffuser_image_height=512,
diffuser_image_width=512,
)
# -------- 文本到图像生成算子 --------
self.text_to_image_generator = PromptedImageGenerator(
t2i_serving=self.serving,
save_interval=10
)
def forward(self):
# 调用 PromptedImageGenerator 生成图像
self.text_to_image_generator.run(
storage=self.storage.step(),
input_conversation_key="conversations",
output_image_key="images",
)
if __name__ == "__main__":
# -------- 流水线入口 --------
model = ImageGenerationPipeline()
model.forward()
