AI智能体与Coze工作流实战：如何实现小红书图文一键生成

1次阅读

共计 2126 个字符，预计需要花费 6 分钟才能阅读完成。

小红书作为内容创作的热门平台，图文制作一直是创作者的核心需求。然而，传统的手动创作流程存在几个显著问题：

效率低下 ：从选题、文案创作、图片处理到排版发布，全流程耗时平均 2 - 3 小时 / 篇
内容同质化 ：人工创作难以突破思维定式，导致内容相似度高
质量不稳定 ：受创作者状态影响大，难以保证持续输出高质量内容
平台规则适应 ：需要不断调整内容以适应算法推荐机制

我们对比了多种 AI 模型和工作流平台，最终选择以下技术组合：

AI 模型层 ：
GPT-4 Turbo：用于文案生成，优势在于长文本连贯性和创意表达
Stable Diffusion XL：图片生成，平衡质量与生成速度
CLIP：图像内容审核，确保合规性
工作流平台 ：
Coze：相比 Zapier/Make，提供更灵活的 AI 集成和本地化支持
优势点：
- 可视化流程设计
- 内置 AI 节点
- 支持自定义代码
- 完善的错误处理机制

graph LR
    A[用户输入主题] --> B(主题分析)
    B --> C{是否需要细化?}
    C -->| 是 | D[生成关键词矩阵]
    C -->| 否 | E[直接生成文案]
    D --> E
    E --> F[调用 GPT 文案生成]
    F --> G[图片 Prompt 生成]
    G --> H[调用 SD 生成图片]
    H --> I[图文合成]
    I --> J[内容合规检查]
    J --> K[输出小红书格式]

主题清洗：
去除特殊字符
提取核心关键词
情感倾向分析
需求细化：
通过 3 轮问答式交互（Coze Bot）
生成风格指令：” 年轻女性视角，轻松幽默语气 ”

文案生成：
temperature=0.7
max_tokens=800
添加小红书特有指令：” 包含 3 个 emoji，每段不超过 3 行 ”
图片生成：
使用 LoRA 适配小红书风格
负面提示词：”text, watermark, lowres”
固定种子值保证风格统一

使用 Pillow 库实现：
自适应图片裁剪（9:16）
文字叠加抗锯齿处理
品牌水印自动添加
模板系统：
5 种预设版式
根据文字量自动选择

# 文案生成封装函数
def generate_copywriting(topic: str, style: str = "casual") -> str:
    """
    调用 GPT 生成小红书文案

    参数:
        topic: 核心主题
        style: 写作风格

    返回:
        格式化后的文案内容
    """
    try:
        prompt = f""" 你是一位小红书爆款文案写手，请根据以下要求创作：主题：{topic}
        风格：{style}
        要求：- 添加 3 个相关 emoji
        - 每段不超过 3 行
        - 包含互动问题 """

        response = openai.ChatCompletion.create(
            model="gpt-4-turbo",
            messages=[{"role": "user", "content": prompt}],
            temperature=0.7,
            max_tokens=800
        )

        # 后处理
        result = response.choices[0].message.content
        return format_for_xiaohongshu(result)

    except Exception as e:
        logger.error(f"文案生成失败: {str(e)}")
        raise ContentGenerationError("文案生成服务暂时不可用")


def format_for_xiaohongshu(text: str) -> str:
    """小红书特定格式处理"""
    # 添加话题标签
    hashtags = "# 好物分享 #生活记录 #经验分享"
    return f"{text}\n\n{hashtags}"

采用 Redis 实现请求队列
设置单用户 QPS= 2 的限制
异步生成 + 回调通知机制

指数退避重试策略：

def call_with_retry(api_func, max_retries=3):
    retry_delay = 1
    for attempt in range(max_retries):
        try:
            return api_func()
        except RateLimitError:
            sleep(retry_delay * (2 ** attempt))
    raise APILimitExceededError