如何用Nano Banana Pro批量生成广告素材?电商模板化投放4大实战策略

import openai
import asyncio
import aiohttp
import json
import os
from typing import List, Dict, Optional
from dataclasses import dataclass
from enum import Enum
import hashlib
from datetime import datetime

class Platform(Enum):
    """广告平台枚举"""
    TAOBAO = {"name": "淘宝", "size": "800x800", "ratio": "1:1"}
    JD = {"name": "京东", "size": "800x800", "ratio": "1:1"}
    PINDUODUO = {"name": "拼多多", "size": "750x750", "ratio": "1:1"}
    WECHAT_MOMENTS = {"name": "微信朋友圈", "size": "1280x720", "ratio": "16:9"}
    DOUYIN = {"name": "抖音", "size": "1080x1920", "ratio": "9:16"}
    XIAOHONGSHU = {"name": "小红书", "size": "1242x1660", "ratio": "3:4"}

@dataclass
class AdTemplate:
    """广告模板数据类"""
    name: str
    prompt_template: str
    style: str
    platform: Platform
    negative_prompt: Optional[str] = None

@dataclass
class ProductData:
    """产品数据类"""
    sku: str
    name: str
    category: str
    selling_points: List[str]
    price: float
    discount: Optional[str] = None
    reference_images: List[str] = None

class NanoBananaProAdGenerator:
    """Nano Banana Pro 广告素材批量生成器"""

    def __init__(self, api_key: str, base_url: str = "https://api.apiyi.com/v1"):
        self.client = openai.OpenAI(api_key=api_key, base_url=base_url)
        self.cache_dir = "./ad_cache"
        self.output_dir = "./generated_ads"
        os.makedirs(self.cache_dir, exist_ok=True)
        os.makedirs(self.output_dir, exist_ok=True)

    def _generate_cache_key(self, product: ProductData, template: AdTemplate) -> str:
        """生成缓存键"""
        data = f"{product.sku}_{template.name}_{template.platform.name}"
        return hashlib.md5(data.encode()).hexdigest()

    def _check_cache(self, cache_key: str) -> Optional[str]:
        """检查缓存"""
        cache_file = f"{self.cache_dir}/{cache_key}.json"
        if os.path.exists(cache_file):
            with open(cache_file, 'r', encoding='utf-8') as f:
                cache_data = json.load(f)
                return cache_data.get('image_url')
        return None

    def _save_cache(self, cache_key: str, image_url: str, metadata: Dict):
        """保存缓存"""
        cache_file = f"{self.cache_dir}/{cache_key}.json"
        cache_data = {
            "image_url": image_url,
            "generated_at": datetime.now().isoformat(),
            "metadata": metadata
        }
        with open(cache_file, 'w', encoding='utf-8') as f:
            json.dump(cache_data, f, ensure_ascii=False, indent=2)

    def build_prompt(self, product: ProductData, template: AdTemplate) -> str:
        """构建提示词"""
        selling_point = product.selling_points[0] if product.selling_points else "优质商品"
        discount_text = product.discount if product.discount else "热卖中"

        prompt = template.prompt_template.format(
            product_name=product.name,
            category=product.category,
            selling_point=selling_point,
            price=product.price,
            discount=discount_text,
            style=template.style,
            platform=template.platform.value["name"]
        )

        return prompt

    async def generate_single_ad(
        self,
        product: ProductData,
        template: AdTemplate,
        use_cache: bool = True
    ) -> Dict:
        """生成单个广告素材"""

        # 检查缓存
        cache_key = self._generate_cache_key(product, template)
        if use_cache:
            cached_url = self._check_cache(cache_key)
            if cached_url:
                return {
                    "success": True,
                    "from_cache": True,
                    "product_sku": product.sku,
                    "template": template.name,
                    "platform": template.platform.name,
                    "image_url": cached_url
                }

        # 构建提示词
        prompt = self.build_prompt(product, template)

        try:
            # 调用 API 生成
            response = self.client.images.generate(
                model="nano-banana-pro",
                prompt=prompt,
                size=template.platform.value["size"],
                n=1,
                response_format="url"
            )

            image_url = response.data[0].url

            # 保存缓存
            metadata = {
                "product": product.name,
                "sku": product.sku,
                "template": template.name,
                "platform": template.platform.name,
                "prompt": prompt
            }
            self._save_cache(cache_key, image_url, metadata)

            return {
                "success": True,
                "from_cache": False,
                "product_sku": product.sku,
                "template": template.name,
                "platform": template.platform.name,
                "image_url": image_url,
                "prompt": prompt
            }

        except Exception as e:
            return {
                "success": False,
                "product_sku": product.sku,
                "template": template.name,
                "platform": template.platform.name,
                "error": str(e)
            }

    async def batch_generate(
        self,
        products: List[ProductData],
        templates: List[AdTemplate],
        max_concurrent: int = 10,
        use_cache: bool = True
    ) -> List[Dict]:
        """批量生成广告素材"""

        tasks = []
        for product in products:
            for template in templates:
                tasks.append(self.generate_single_ad(product, template, use_cache))

        # 控制并发数
        results = []
        for i in range(0, len(tasks), max_concurrent):
            batch = tasks[i:i + max_concurrent]
            batch_results = await asyncio.gather(*batch)
            results.extend(batch_results)

            # 进度显示
            print(f"进度: {min(i + max_concurrent, len(tasks))}/{len(tasks)} 已完成")

        return results

    def generate_report(self, results: List[Dict]) -> Dict:
        """生成生成报告"""
        total = len(results)
        success = sum(1 for r in results if r.get("success"))
        from_cache = sum(1 for r in results if r.get("from_cache"))
        failed = total - success

        # 按平台分组统计
        by_platform = {}
        for result in results:
            if result.get("success"):
                platform = result["platform"]
                if platform not in by_platform:
                    by_platform[platform] = 0
                by_platform[platform] += 1

        return {
            "total_tasks": total,
            "success_count": success,
            "from_cache_count": from_cache,
            "new_generated_count": success - from_cache,
            "failed_count": failed,
            "success_rate": f"{(success/total*100):.1f}%",
            "by_platform": by_platform
        }

# 使用示例
async def main():
    # 初始化生成器
    generator = NanoBananaProAdGenerator(
        api_key="your_api_key_here"
    )

    # 定义广告模板
    templates = [
        AdTemplate(
            name="主图模板",
            prompt_template="""电商主图设计:
产品: {product_name}
类别: {category}
核心卖点: {selling_point}
价格: ¥{price}
促销标签: {discount}

设计风格: {style}
平台: {platform}

设计要求:
1. 产品居中,占比60-70%
2. 白色或浅色纯净背景
3. 左上角红色促销标签
4. 右下角显示价格
5. 专业商业摄影质感

高质量电商主图,适合{platform}平台。""",
            style="简约现代",
            platform=Platform.TAOBAO
        ),
        AdTemplate(
            name="社交媒体模板",
            prompt_template="""社交媒体广告设计:
产品: {product_name}
卖点: {selling_point}
优惠: {discount}

设计风格: {style},生活化场景
文字: 产品名称和卖点文案

设计要求:
1. 生活场景中展示产品使用
2. 温馨自然的氛围
3. 上方显示产品名称
4. 底部显示核心卖点
5. 适合{platform}平台分享

真实感强,吸引目光。""",
            style="清新自然",
            platform=Platform.XIAOHONGSHU
        )
    ]

    # 准备产品数据
    products = [
        ProductData(
            sku="BT-001",
            name="蓝牙耳机 Pro Max",
            category="数码配件",
            selling_points=["主动降噪", "40小时续航", "HiFi音质"],
            price=299.0,
            discount="限时5折"
        ),
        ProductData(
            sku="WB-002",
            name="保温杯商务款",
            category="生活用品",
            selling_points=["24小时保温", "316不锈钢", "防漏设计"],
            price=159.0,
            discount="买一送一"
        )
    ]

    # 批量生成
    print("开始批量生成广告素材...")
    results = await generator.batch_generate(
        products=products,
        templates=templates,
        max_concurrent=5,
        use_cache=True
    )

    # 生成报告
    report = generator.generate_report(results)

    print("\n" + "="*50)
    print("生成报告")
    print("="*50)
    print(f"总任务数: {report['total_tasks']}")
    print(f"成功: {report['success_count']} ({report['success_rate']})")
    print(f"从缓存读取: {report['from_cache_count']}")
    print(f"新生成: {report['new_generated_count']}")
    print(f"失败: {report['failed_count']}")
    print(f"\n各平台分布:")
    for platform, count in report['by_platform'].items():
        print(f"  {platform}: {count} 张")

    # 保存结果
    with open("generation_results.json", "w", encoding="utf-8") as f:
        json.dump(results, f, ensure_ascii=False, indent=2)

    print(f"\n详细结果已保存到 generation_results.json")

if __name__ == "__main__":
    asyncio.run(main())

保证批量生成素材质量一致性需要三个层面的控制:

模板层控制:

• 使用标准化提示词模板,确保描述语言一致
• 在模板中明确禁止的元素(否定提示词)
• 固定品牌色调、Logo位置等核心视觉元素

流程层控制:

• 实施自动化质量检查(图像识别验证关键元素)
• 设置质量阈值,不合格自动重新生成
• 建立人工抽检机制(抽检 10% 样本)

数据层控制:

• 使用高质量、一致性强的参考图
• 产品数据规范化(统一字段格式和命名)
• 记录生成参数,便于复现和优化

通过 API易 apiyi.com 平台的质量管理功能,可以设置自动化质量检查规则,不符合要求的素材会自动进入人工审核队列或触发重新生成。

批量生成中的失败处理是保证系统稳定性的关键:

预防措施:

• 验证输入数据完整性和格式正确性
• 设置合理的 API 调用超时时间
• 控制并发数量避免触发限流

失败处理策略:

• 实现指数退避重试机制(第1次等1秒,第2次等2秒,第3次等4秒)
• 最多重试 3 次,仍失败则记录到失败队列
• 定期批量处理失败队列(人工介入或调整参数)

监控告警:

• 实时监控成功率,低于 90% 触发告警
• 记录详细错误日志便于问题排查
• 生成失败报告,分析失败原因分布

代码示例:

async def generate_with_retry(task, max_retries=3):
    for attempt in range(max_retries):
        try:
            result = await generate_single_ad(task)
            if result["success"]:
                return result
        except Exception as e:
            if attempt == max_retries - 1:
                # 最后一次失败,记录到失败队列
                save_to_failed_queue(task, str(e))
            else:
                # 等待后重试
                await asyncio.sleep(2 ** attempt)

是的,不同产品类别的广告素材需求差异很大,建议建立「分类模板库」:

标准品类模板 (适用大多数产品):

• 3C 数码类: 科技感、功能展示、参数突出
• 服装鞋包: 场景化、模特展示、搭配建议
• 美妆个护: 质感展示、使用效果、成分说明
• 食品饮料: 食欲感、新鲜度、口味描述
• 家居日用: 场景融入、使用场景、材质展示

特殊品类定制:

• 奢侈品: 高端感、细节特写、品牌故事
• 母婴产品: 安全感、温馨场景、材质认证
• 健康保健: 专业背书、数据支持、权威认证

模板复用策略:

• 同类别产品共享模板,只替换产品信息
• 跨类别的通用元素(如促销标签)可以复用
• 定期分析各类别转化数据,优化模板库

建议初期为每个主要类别准备 3-5 个核心模板,根据实际效果逐步扩充和优化。

大部分情况下可以,但建议建立「分级审核」机制:

自动投放 (占比 70-80%):

• 使用验证过的成熟模板
• 产品信息完整准确
• 自动化质量检查全部通过
• 历史数据表现良好

快速审核 (占比 15-20%):

• 新模板首次使用
• 高价值产品/重要活动
• 涉及品牌敏感信息
• 人工抽检 2-3 分钟通过

严格审核 (占比 5-10%):

• 特殊品类(医疗/金融/教育)
• 涉及明星/IP 授权内容
• 超大规模投放(百万级预算)
• 法务/品牌部门联合审核

平台规范检查:
在投放前务必验证:

• 图片尺寸和格式符合平台要求
• 文案中无违禁词
• 无侵犯第三方版权内容
• 广告标识和免责声明完整

通过 API易 apiyi.com 平台的「智能审核」功能,可以自动检测素材是否符合各大广告平台的规范,包括尺寸验证、违禁词检测和内容安全扫描,大幅提升投放效率。

建立完整的效果追踪体系是优化素材生成策略的关键:

素材级指标:

• CTR (点击率): 衡量素材吸引力
• CVR (转化率): 衡量素材说服力
• CPC (单次点击成本): 衡量投放效率
• ROI (投资回报率): 衡量整体收益

模板级指标:

• 不同模板的平均 CTR/CVR 对比
• 各模板在不同品类的表现差异
• 模板迭代前后的效果提升

追踪方法:

1. 为每个生成的素材分配唯一 ID
2. 在广告投放时记录素材 ID 与投放数据关联
3. 定期(日/周)汇总各素材的表现数据
4. 建立数据看板可视化展示

优化闭环:

生成素材 → 投放测试 → 收集数据 →
分析表现 → 优化模板 → 再次生成

代码示例:

# 记录素材元数据
metadata = {
    "asset_id": "AD-2026-001",
    "template": "主图模板A",
    "product_sku": "BT-001",
    "generated_at": "2026-01-19",
    "platforms": ["淘宝", "京东"]
}

# 定期拉取投放数据
ad_performance = fetch_ad_performance(
    asset_id="AD-2026-001",
    date_range="last_7_days"
)

# 分析并优化
if ad_performance["ctr"] > 0.05:  # CTR > 5%
    mark_template_as_high_performing("主图模板A")

API易 apiyi.com 平台提供素材效果追踪 API,可与主流广告平台(腾讯广告/巨量引擎/阿里妈妈)打通数据,自动关联素材 ID 和投放效果,生成优化建议报告。