Qwen-Image-Lightning部署教程：Airflow调度定时绘图任务实践

1. 项目简介与核心价值

Qwen-Image-Lightning是一个基于Qwen/Qwen-Image-2512旗舰模型构建的文生图应用，集成了最新的Lightning LoRA加速技术。这个镜像最大的特点是极致轻量和高稳定性，专门为需要批量生成图片的用户设计。

传统的文生图模型通常需要几十步推理计算，生成一张图片可能要几分钟。而Qwen-Image-Lightning采用了4步极速推理方案，将生成时间压缩到毫秒级响应。更重要的是，它针对24G显存环境做了深度优化，使用Sequential CPU Offload策略，在生成1024x1024高清大图时显存占用极低，彻底解决了让人头疼的"显存不足"问题。

为什么需要Airflow调度？

• 自动执行：可以设置每天固定时间自动生成图片
• 批量处理：一次可以设置多个绘图任务
• 任务监控：清晰看到每个任务的执行状态和结果
• 错误处理：任务失败会自动重试或通知

2. 环境准备与快速部署

2.1 系统要求

在开始之前，请确保你的环境满足以下要求：

• 操作系统：Ubuntu 20.04/22.04 或 CentOS 7+
• 显卡：NVIDIA RTX 3090/4090（24G显存）
• 驱动：NVIDIA驱动版本 525.60.13+
• Docker：版本 20.10+
• Docker Compose：版本 2.20+

2.2 一键部署Qwen-Image-Lightning

首先我们来部署文生图服务：

# 创建项目目录
mkdir qwen-airflow-project && cd qwen-airflow-project

# 创建docker-compose.yml文件
cat > docker-compose.yml << 'EOF'
version: '3.8'

services:
  qwen-image:
    image: qwen-image-lightning:latest
    ports:
      - "8082:8082"
    deploy:
      resources:
        reservations:
          devices:
            - driver: nvidia
              count: 1
              capabilities: [gpu]
    volumes:
      - ./output:/app/output
    restart: unless-stopped

  airflow:
    image: apache/airflow:2.8.1
    environment:
      - AIRFLOW__CORE__EXECUTOR=LocalExecutor
      - AIRFLOW__CORE__LOAD_EXAMPLES=false
    volumes:
      - ./airflow/dags:/opt/airflow/dags
      - ./airflow/logs:/opt/airflow/logs
      - ./output:/opt/airflow/output
      - ./scripts:/opt/airflow/scripts
    ports:
      - "8080:8080"
    depends_on:
      - qwen-image
    restart: unless-stopped
EOF

2.3 启动服务

# 启动所有服务
docker-compose up -d

# 查看服务状态
docker-compose ps

# 查看日志（等待2分钟左右服务启动完成）
docker-compose logs -f qwen-image

服务启动后，你可以通过以下方式访问：

• Qwen文生图界面：http://你的服务器IP:8082
• Airflow管理界面：http://你的服务器IP:8080

3. Airflow任务配置实战

3.1 创建绘图任务脚本

首先创建用于调用文生图服务的Python脚本：

mkdir -p scripts

创建 scripts/generate_image.py：

import requests
import json
import time
import os
from datetime import datetime

def generate_image(prompt, output_dir="/opt/airflow/output"):
    """
    调用Qwen-Image-Lightning生成图片
    """
    url = "http://qwen-image:8082/generate"
    
    payload = {
        "prompt": prompt,
        "steps": 4,
        "width": 1024,
        "height": 1024
    }
    
    try:
        # 发送生成请求
        response = requests.post(url, json=payload, timeout=120)
        response.raise_for_status()
        
        # 解析返回的图片数据
        result = response.json()
        image_data = result.get("image", "")
        
        if image_data:
            # 生成文件名
            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            safe_prompt = "".join(c for c in prompt[:50] if c.isalnum() or c in (' ', '-', '_')).rstrip()
            filename = f"{timestamp}_{safe_prompt}.png"
            filepath = os.path.join(output_dir, filename)
            
            # 保存图片（实际使用时需要base64解码）
            # 这里简化处理，实际需要根据API返回格式调整
            print(f"图片生成成功: {filename}")
            return filepath
        else:
            print("生成失败：未返回图片数据")
            return None
            
    except requests.exceptions.RequestException as e:
        print(f"请求失败: {e}")
        return None
    except json.JSONDecodeError as e:
        print(f"JSON解析失败: {e}")
        return None

if __name__ == "__main__":
    # 测试生成
    test_prompt = "清晨的阳光透过森林，雾气缭绕，4K高清"
    result = generate_image(test_prompt)
    print(f"生成结果: {result}")

3.2 配置Airflow DAG任务

创建Airflow任务调度文件：

mkdir -p airflow/dags

创建 airflow/dags/daily_image_generation.py：

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator
import sys
import os

# 添加脚本路径
sys.path.insert(0, '/opt/airflow/scripts')

from generate_image import generate_image

default_args = {
    'owner': 'ai-artist',
    'depends_on_past': False,
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
}

# 定义每天要生成的图片主题
DAILY_PROMPTS = {
    0: "周一早安！朝阳下的城市天际线，现代建筑，4K高清",  # 周一
    1: "周二激励：高山巅峰，云海日出，壮丽景观",        # 周二
    2: "周三加油：科幻未来城市，霓虹灯光，赛博朋克风格", # 周三
    3: "周四坚持：森林中的小鹿，晨雾，自然风光",        # 周四
    4: "周五愉快：海滩日落，温暖色调，度假风格",        # 周五
    5: "周末放松：星空下的雪山，极光，梦幻场景",        # 周六
    6: "周日反思：水墨山水画，中国风，意境深远"         # 周日
}

def get_daily_prompt():
    """获取当天的图片主题"""
    weekday = datetime.now().weekday()
    return DAILY_PROMPTS.get(weekday, "美丽的自然风景，4K高清")

def generate_daily_image():
    """生成每日图片任务"""
    prompt = get_daily_prompt()
    print(f"开始生成今日图片: {prompt}")
    
    result = generate_image(prompt)
    if result:
        print(f"图片生成成功: {result}")
        return result
    else:
        raise Exception("图片生成失败")

# 定义DAG
with DAG(
    'daily_image_generation',
    default_args=default_args,
    description='每天自动生成一张AI图片',
    schedule_interval='0 9 * * *',  # 每天上午9点执行
    start_date=datetime(2024, 1, 1),
    catchup=False,
    tags=['ai', 'image-generation'],
) as dag:

    # 创建输出目录
    create_output_dir = BashOperator(
        task_id='create_output_dir',
        bash_command='mkdir -p /opt/airflow/output/$(date +%Y-%m-%d)',
    )

    # 生成图片任务
    generate_task = PythonOperator(
        task_id='generate_daily_image',
        python_callable=generate_daily_image,
    )

    # 备份任务
    backup_task = BashOperator(
        task_id='backup_results',
        bash_command='cp /opt/airflow/output/*.png /opt/airflow/output/$(date +%Y-%m-%d)/ 2>/dev/null || true',
    )

    # 设置任务依赖
    create_output_dir >> generate_task >> backup_task

4. 高级调度技巧与实践

4.1 批量图片生成任务

如果你需要一次性生成多张图片，可以创建批量任务：

# 在scripts目录下创建 batch_generate.py
def generate_batch_images(prompts_list, output_dir):
    """批量生成多张图片"""
    results = []
    for i, prompt in enumerate(prompts_list):
        print(f"生成第 {i+1}/{len(prompts_list)} 张图片: {prompt}")
        result = generate_image(prompt, output_dir)
        if result:
            results.append(result)
        time.sleep(2)  # 避免请求过于频繁
    return results

# 对应的Airflow DAG任务
def generate_weekly_batch():
    """生成一周的所有图片"""
    prompts = list(DAILY_PROMPTS.values())
    output_dir = f"/opt/airflow/output/weekly_{datetime.now().strftime('%Y-%U')}"
    os.makedirs(output_dir, exist_ok=True)
    
    return generate_batch_images(prompts, output_dir)

4.2 错误处理与重试机制

在实际使用中，我们需要完善的错误处理：

def robust_generate_image(prompt, max_retries=3):
    """带重试机制的图片生成"""
    for attempt in range(max_retries):
        try:
            result = generate_image(prompt)
            if result:
                return result
            else:
                print(f"第 {attempt+1} 次尝试失败，重试...")
                time.sleep(10)
        except Exception as e:
            print(f"第 {attempt+1} 次尝试异常: {e}")
            time.sleep(10)
    
    print(f"所有 {max_retries} 次尝试都失败了")
    return None

4.3 监控与通知配置

配置任务完成通知：

def send_notification(success=True, image_path=None, prompt=None):
    """发送任务通知"""
    if success:
        message = f" 图片生成成功！\n主题: {prompt}\n生成时间: {datetime.now()}"
    else:
        message = f" 图片生成失败！\n主题: {prompt}\n请检查服务状态"
    
    print(message)
    # 这里可以集成邮件、Slack、微信等通知方式

5. 实战演示与效果验证

5.1 测试任务执行

让我们测试一下整个流程：

# 手动触发Airflow任务测试
docker exec -it qwen-airflow-project-airflow-1 airflow tasks test daily_image_generation generate_daily_image 2024-01-01

# 查看生成的图片
ls -la output/

# 检查服务日志
docker-compose logs qwen-image

5.2 效果验证与优化

生成效果检查要点：

1. 图片质量：1024x1024分辨率是否清晰
2. 生成速度：单张图片40-50秒是否正常
3. 显存占用：使用nvidia-smi检查显存使用情况
4. 内容匹配：生成的图片是否与提示词相符

常见问题排查：

# 检查服务状态
docker-compose ps

# 查看详细日志
docker-compose logs --tail=100 qwen-image

# 检查显存使用
nvidia-smi

# 测试API接口
curl -X POST http://localhost:8082/generate \
  -H "Content-Type: application/json" \
  -d '{"prompt": "测试图片，蓝天白云", "steps": 4}'

6. 总结与最佳实践

通过本教程，我们成功搭建了基于Qwen-Image-Lightning和Airflow的自动绘图系统。这个方案的优势在于：

核心价值：

• 🕐 全自动运行：每天自动生成图片，无需人工干预
• 批量处理：支持一次性生成多张不同主题的图片
• 🔧 稳定可靠：内置重试机制和错误处理
• 👀 实时监控：通过Airflow界面清晰掌握任务状态

实践建议：

1. 资源规划：建议在业务低峰期执行批量生成任务
2. 提示词优化：积累高质量的提示词模板库
3. 存储管理：定期清理旧的图片文件，避免磁盘占满
4. 监控告警：设置任务失败通知，及时处理问题

扩展应用场景：

• 社交媒体内容自动生成
• 电商产品图批量制作
• 设计素材库自动扩充
• 个性化每日问候图片

这个方案不仅解决了手动生成图片的繁琐问题，更重要的是提供了一个可扩展的自动化框架。你可以根据自己的需求，进一步扩展功能，比如添加图片后处理、自动上传到云存储、集成到内容管理系统等。

---

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。