从零实现AI图文自动发布系统：小红书内容生成与发布的工程实践

1次阅读

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

作为一名内容创作者，我深知手动发布图文内容的繁琐。每天绞尽脑汁想创意、修图、写文案、找话题标签，还要研究平台规则，效率极其低下。这种重复劳动不仅消耗时间，还容易让人产生创意枯竭的困扰。于是我开始探索如何通过技术手段实现全自动化的内容生产与发布。

在构建这个系统时，首先需要选择合适的 AI 生成模型。经过对比测试，我最终选择了以下方案：

图像生成 ：Stable Diffusion XL 1.0
开源免费，可本地部署
通过 LoRA 微调可适配小红书风格
Diffusers 库提供友好的 Python 接口
文本生成 ：GPT-3.5-turbo
生成质量稳定
可通过 prompt 工程精确控制输出
API 调用简单高效

使用 Diffusers 库调用 Stable Diffusion 模型生成图片：

from diffusers import StableDiffusionXLPipeline
import torch

# 初始化模型
pipe = StableDiffusionXLPipeline.from_pretrained(
    "stabilityai/stable-diffusion-xl-base-1.0",
    torch_dtype=torch.float16,
    variant="fp16",
).to("cuda")

# 生成图片
def generate_image(prompt):
    try:
        image = pipe(
            prompt=prompt,
            height=1024,
            width=768,
            num_inference_steps=50,
        ).images[0]
        return image
    except Exception as e:
        logging.error(f"Image generation failed: {str(e)}")
        raise

通过 OpenAI API 生成文案，并优化话题标签：

import openai

def generate_caption(topic):
    try:
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=[{"role": "system", "content": "你是一个专业的小红书文案写手"},
                {"role": "user", "content": f"请为'{topic}'创作一篇小红书文案，包含 3 个相关话题标签"}
            ],
            temperature=0.7,
        )
        return response.choices[0].message.content
    except Exception as e:
        logging.error(f"Caption generation failed: {str(e)}")
        raise

调用小红书开放平台 API 实现自动发布：

import requests
from io import BytesIO

def upload_image(image):
    # 将图片转换为字节流
    img_byte_arr = BytesIO()
    image.save(img_byte_arr, format='PNG')
    img_byte_arr = img_byte_arr.getvalue()

    # 上传图片
    files = {'file': ('image.png', img_byte_arr, 'image/png')}
    response = requests.post(
        'https://open.xiaohongshu.com/api/upload',
        files=files,
        headers={'Authorization': f'Bearer {access_token}'}
    )
    return response.json()['data']['image_id']

def create_post(caption, image_ids):
    data = {
        "content": caption,
        "image_ids": image_ids,
        "visibility": 1  # 公开可见
    }
    response = requests.post(
        'https://open.xiaohongshu.com/api/post/create',
        json=data,
        headers={'Authorization': f'Bearer {access_token}'}
    )
    return response.json()

使用 Celery+RabbitMQ 实现任务队列：

from celery import Celery

app = Celery('tasks', broker='pyamqp://guest@localhost//')

@app.task(bind=True)
def publish_post(self, topic):
    try:
        # 生成图片
        image = generate_image(topic)

        # 生成文案
        caption = generate_caption(topic)

        # 上传图片
        image_id = upload_image(image)

        # 发布内容
        result = create_post(caption, [image_id])

        return result
    except Exception as e:
        self.retry(exc=e, countdown=60)

实现敏感内容过滤：

from transformers import pipeline

# 初始化敏感词检测模型
classifier = pipeline("text-classification", model="bert-base-chinese")

def check_sensitive_content(text):
    result = classifier(text)
    return result[0]['label'] == 'NEGATIVE'