企业官网建设流程全解析

AI 批量生成带货视频的工程实践在近两年已经相当常见。借助大模型写文案、TTS 配音、ffmpeg 合成画面，一套流水线可以在几分钟内产出一批短视频。但这个效率背后藏着一个很容易被忽略的问题：如果商品本身已经被监管机构召回，或者文案使用了没有证据支撑的功效表述，生成速度越快，风险扩散越快。近期关于“AI TikTok Shop slop factory”的讨论，正是这类问题的集中体现：批量生产的内容大量堆上货架，却没有人检查商品是否仍然可以售卖、文案是否合规、平台下架后是否还能自动撤回。

本文不评价某个具体平台或案例，而是从工程角度拆解一套面向短视频电商的 AI 带货视频流水线，重点解决一个核心问题：如何在商品导入、文案生成、视频合成、发布前后这几个环节里，加入可执行的合规网关。读完这篇文章，你可以得到一个最小可运行的设计思路、一组关键代码片段，以及一套用于排查“为什么违规内容没拦住”的排查清单。

1. 先理解“AI 内容工厂”为什么会在合规上失控

1.1 不要把内容自动化和内容垃圾化混为一谈

“内容工厂”本身是一个中性的工程概念。一个团队每天需要产出大量商品短视频，如果完全靠人工写脚本、人工剪辑、人工配音，成本会高到不可接受。于是就有了自动化流水线：商品数据进系统，脚本由大模型生成，音频由 TTS 合成，视频由 ffmpeg 拼接，最后定时发布。

问题出在“只追求产量、不检查质量”的实现方式上。当流水线只关心“能不能生成”，不关心“生成的内容是否合法、商品是否安全”，它就退化成内容垃圾站。英文里用 slop 描述这种内容，意思是量大、低质、没有人工负责。真正的工程问题不是“要不要自动化”，而是“自动化之后谁来把关，把关逻辑放在哪一层”。

1.2 失控链路：商品、文案、视频、发布四个环节各自为政

大部分翻车案例都不是单个环节出错，而是整条链路缺少状态同步。

看一条典型的失控链路：

1. 运营把一批商品导入系统，商品状态标记为“可推广”。
2. 大模型根据商品名生成文案，文案里出现“可以预防某类疾病”之类的表述。
3. TTS 把文案读成音频，视频合成模块把音频、图片、背景音乐拼成成片。
4. 发布模块把视频上传到短视频电商平台。
5. 商品其实已经被监管机构召回，但商品状态在第一步之后就没有再更新。

这里最致命的地方是：每个模块都完成了自己的局部任务，却没有一个模块回答“这个商品现在还能不能卖、这段话有没有证据支持”。召回状态变更通常发生在商品导入之后、视频发布之前，或者在视频发布之后。如果不做二次检查，系统就会持续推广一个已经不具备销售资格的商品。

大模型还会放大这个问题。LLM 生成文案时，很容易产出“夸大功效”“绝对化用语”“没有引用来源”的内容。如果只把大模型当成文案生成器，不限制输出结构，也不做关键词和证据链校验，违规内容就会以很高概率出现。

1.3 合规网关应该放在生成之前和发布之前

正确的做法不是把合规检查全部放到最后，而是设置两个网关：

• 生成前网关：在调用大模型生成文案之前，先确认商品状态合法、品类允许推广、已有的召回名单里没有这个商品。
• 发布前网关：在视频合成完成、准备提交平台之前，再一次检查商品状态和文案内容，防止生成期间状态发生变化。

有些团队还会增加发布后监控任务，比如每隔一段时间拉取一次召回名单，和已经发布的视频做比对。一旦发现某个商品被新召回，立即下线对应视频。这属于生产环境必须具备的“撤回机制”，而不是可选项。

2. 流水线设计与环境准备

2.1 整体流程：一次从商品到成片的完整链路

下面是一条带合规网关的 AI 带货视频流水线。它比普通的一键成片多出两个检查点，但整体结构仍然简单：

商品列表 -> 合规闸门A -> 文案生成 -> 内容校验 -> TTS配音 -> 视频合成 -> 合规闸门B -> 人工抽审 -> 发布 -> 发布后监控

每个环节的职责如下：

• 商品列表：导入商品主数据，包括商品 ID、名称、SKU、品类、宣称功效。
• 合规闸门A：检查商品是否在召回名单、是否属于禁止推广品类、是否存在明显违规属性。
• 文案生成：调用大模型生成口播脚本，同时要求模型按 JSON 结构输出。
• 内容校验：对文案做禁用词、功效词、绝对化用语检测，并检查是否包含免责声明。
• TTS 配音：把通过校验的文案转换为音频。
• 视频合成：用 ffmpeg 把商品图片、音频和字幕合成 MP4。
• 合规闸门B：在发布前重新检查商品状态和文案。
• 人工抽审：对于高风控品类，保留人工审核通道。
• 发布后监控：周期性同步召回名单，发现商品状态变化后下线对应视频。

2.2 开发环境与依赖

学习环境建议使用 Python 3.11 和 SQLite，因为这样可以最快速度跑通链路。生产环境再把存储替换为 PostgreSQL、Redis 等基础设施。

如果原始项目没有固定大模型供应商，代码中会保留一个LLMClient抽象层，方便替换成不同厂商的 SDK。

2.3 项目目录结构

一个最小项目可以按下面的结构组织：

ai_product_video/ ├── main.py # 入口脚本 ├── models.py # 商品、召回状态、文案等数据模型 ├── compliance.py # 合规检查服务 ├── script_generator.py # LLM 文案生成 ├── video_builder.py # TTS + ffmpeg 视频合成 ├── recall_store.py # 召回名单同步和存储 ├── config.yaml # 环境配置 ├── data/ │ ├── products.csv # 示例商品 │ └── recall_products.csv # 示例召回名单 └── out/ # 生成的视频和日志

目录设计的原则是“每个文件只做一件事”。合规检查不依赖具体的大模型 SDK，文案生成不直接操作视频文件，召回名单存储不感知发布逻辑。这样后续替换任何组件，都不会牵动整条链路。

3. 核心代码实现：从商品导入到合规检查

3.1 商品模型与召回状态模型

先定义商品和召回状态的数据结构。这里使用 pydantic 做校验，避免脏数据进入后续环节。

# models.py from datetime import datetime from enum import Enum from pydantic import BaseModel, Field class ProductStatus(str, Enum): NORMAL = "normal" RECALLED = "recalled" SUSPENDED = "suspended" class Product(BaseModel): product_id: str name: str sku: str category: str claims: list[str] = Field(default_factory=list) status: ProductStatus = ProductStatus.NORMAL class RecallRecord(BaseModel): product_identifier: str recall_reason: str recalled_at: datetime source: str

商品识别不建议只用中文名称，因为同一个商品可能有多个别名。学习项目可以同时写入 SKU、UPC/EAN 和商品名，检查时逐项比对。

3.2 召回名单同步逻辑

召回名单可以来自多个渠道，比如 FDA 公开召回数据、内部风控名单、平台下架数据。下面这段代码演示了如何从 CSV 或 HTTP 接口同步名单到 SQLite。

# recall_store.py import csv import sqlite3 from pathlib import Path class RecallStore: def __init__(self, db_path: str = "data/recall.db"): self.conn = sqlite3.connect(db_path) self.conn.execute(""" CREATE TABLE IF NOT EXISTS recall_list ( identifier TEXT PRIMARY KEY, reason TEXT, recalled_at TEXT, source TEXT ) """) def sync_from_csv(self, csv_path: str): with open(csv_path, newline="", encoding="utf-8") as f: reader = csv.DictReader(f) rows = [ ( row["identifier"], row.get("reason", ""), row.get("recalled_at", ""), row.get("source", "csv"), ) for row in reader ] self.conn.executemany( "INSERT OR REPLACE INTO recall_list VALUES (?, ?, ?, ?)", rows ) self.conn.commit() def is_recalled(self, identifier: str) -> bool: cur = self.conn.execute( "SELECT 1 FROM recall_list WHERE identifier = ?", (identifier,) ) return cur.fetchone() is not None

生产环境的同步逻辑不会这么简单。它需要考虑接口分页、增量更新、失败重试、缓存失效，以及多环境之间的数据隔离。学习阶段用 CSV 完全足够。

3.3 合规检查服务

合规检查服务是整个系统的核心。它负责回答三个问题：

1. 商品是否在召回名单里。
2. 商品品类是否允许自动推广。
3. 文案是否包含禁用词、绝对化用语或未经证实的功效表述。

# compliance.py import re from models import Product DISALLOWED_CATEGORIES = {"处方药", "成人用品", "医疗机械"} DISALLOWED_TERMS = ["治愈", "根治", "100%有效", "预防癌症", "无副作用"] REQUIRED_DISCLAIMER = "本品不能代替药物" class ComplianceService: def __init__(self, recall_store): self.recall_store = recall_store def check_product(self, product: Product) -> tuple[bool, list[str]]: errors = [] if product.status.value == "recalled": errors.append("商品状态为已召回") if self.recall_store.is_recalled(product.sku) or \ self.recall_store.is_recalled(product.product_id): errors.append("命中召回名单") if product.category in DISALLOWED_CATEGORIES: errors.append("品类禁止自动推广") return len(errors) == 0, errors def check_script(self, script: str) -> tuple[bool, list[str]]: errors = [] text = re.sub(r"<[^>]+>", "", script).lower() for term in DISALLOWED_TERMS: if term.lower() in text: errors.append(f"文案包含禁用词: {term}") if REQUIRED_DISCLAIMER not in script: errors.append("缺少免责声明") return len(errors) == 0, errors

这段代码并不是一个完整的合规方案，但它说明了最小拦截逻辑应该长什么样。实际项目中，禁用词表需要运营、法务和品控共同维护，并且要做到可配置、可追溯、可更新。

3.4 LLM 文案生成与输出校验

LLM 生成文案时，提示词和输出格式是两个最容易出问题的地方。提示词只写“生成一段带货文案”是不够的，必须把约束写清楚，并且要求返回 JSON。

# script_generator.py import json from openai import OpenAI SYSTEM_PROMPT = """ 你是一个保健品短视频脚本助手。生成内容必须遵守以下规则： 1. 不能使用绝对化用语，如“彻底治愈”“100%有效”。 2. 不能暗示产品可以替代药物。 3. 必须包含免责声明：本品不能代替药物。 4. 只输出 JSON，不要输出其他文字。 5. JSON 字段为 script, claim_level, can_publish。 """ def generate_script(client: OpenAI, product_name: str, claims: list[str]) -> dict: user_prompt = ( f"商品：{product_name}\n" f"商品宣称：{', '.join(claims)}\n" "请生成一段不超过 80 字的短视频口播文案，并判断该商品证据等级。" ) resp = client.chat.completions.create( model="gpt-4o-mini", messages=[ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": user_prompt}, ], temperature=0.3, response_format={"type": "json_object"}, ) data = json.loads(resp.choices[0].message.content) return data

即使提示词写了“不能使用绝对化用语”，也不能完全信任模型输出。文案生成后必须再走一遍ComplianceService.check_script。这里的顺序是：先生成，再校验，校验不通过就重新生成或放弃。

3.5 视频合成：用 ffmpeg 拼接成片

当文案通过校验后，就可以用 edge-tts 生成音频，再用 ffmpeg 把商品图片和音频合成视频。下面是一个最小实现。

# video_builder.py import asyncio import subprocess import edge_tts async def text_to_speech(text: str, output_audio: str): communicator = edge_tts.Communicate(text, voice="zh-CN-XiaoxiaoNeural") await communicator.save(output_audio) def build_video(image_path: str, audio_path: str, output_video: str, duration: int = 10): cmd = [ "ffmpeg", "-y", "-loop", "1", "-i", image_path, "-i", audio_path, "-c:v", "libx264", "-tune", "stillimage", "-c:a", "aac", "-b:a", "192k", "-pix_fmt", "yuv420p", "-t", str(duration), output_video, ] subprocess.run(cmd, check=True) def generate_video(text: str, image_path: str, output_video: str): audio_path = output_video.replace(".mp4", ".mp3") asyncio.run(text_to_speech(text, audio_path)) build_video(image_path, audio_path, output_video)

这里不推荐在视频合成阶段再去做文案合规检查，因为视频合成和转码很耗时。如果文案有问题，应在进入 TTS 之前就拦截掉，节省时间和计算资源。

4. 关键配置与参数说明

4.1 合规检查参数表

4.2 召回名单同步与缓存策略

召回名单不能只在启动时同步一次。生产环境建议采用“定时任务 + 过期缓存”的策略。

# config.yaml recall: sync_interval_seconds: 600 cache_ttl_seconds: 300 sources: - type: http url: "https://example.com/recall-list.json" headers: Authorization: "Bearer ${RECALL_API_TOKEN}" - type: csv path: "data/recall_products.csv"

同步任务需要处理几个边界情况：

• 接口一次只能拉取一页，要支持分页。
• 召回记录可能被撤销，所以不能只做追加，要做增量全量比对。
• 缓存过期后如果没有拉到最新数据，应该拒绝敏感品类的发布，而不是继续使用旧名单。

4.3 双闸门与发布后撤回

这里把两个闸门的检查时机和检查内容整理成一张表：

发布后监控是经常被忽略的环节。商品召回不是只有“导入前”和“生成前”两个时间点，完全可能发生在视频发布之后。没有发布后监控，就等于允许已发布内容永远停留在平台上。

5. 运行验证与测试用例

5.1 最小运行步骤

在本地跑通这条链路，建议先用 CSV 文件准备两个商品：一个正常商品，一个已经被召回的演示商品。

python main.py \ --products data/products.csv \ --recall data/recall_products.csv \ --output out/

入口脚本会依次执行：加载商品、同步召回名单、闸门A检查、生成文案、校验文案、生成音频、合成视频、闸门B检查。

5.2 预期输出：正常商品生成视频

正常商品应该打印下面这样的日志：

[compliance] product=P001 passed gate A [script] generated script for P001 [compliance] script passed check [tts] audio saved to out/P001.mp3 [ffmpeg] video saved to out/P001.mp4 [compliance] product=P001 passed gate B [publish] P001 ready to publish

如果最后能在out/目录下看到一个可播放的 MP4 文件，说明链路跑通了。

5.3 测试用例：召回商品必须被拦截

在products.csv中准备一个 SKU 命中召回名单的商品：

product_id,name,sku,category,status P002,演示召回商品,SKU-RECALL-001,普通食品,normal

执行后应该看到类似输出：

[compliance] product=P002 blocked: 命中召回名单 [script] skip P002 because compliance gate A failed

此时不应生成任何视频文件。这个用例用于验证召回名单确实参与流水线决策，而不只是被打印出来。

5.4 测试用例：功效词必须被拦截

直接构造一段包含“治愈”的文案，传入ComplianceService.check_script。预期结果是返回False，错误信息包含“文案包含禁用词: 治愈”。

这类测试建议接入单元测试框架，比如 pytest。每次修改合规规则后，跑一遍测试用例可以快速发现规则配置错误。

6. 常见问题排查：为什么该拦截的没拦住

6.1 排查顺序

当出现违规内容漏过时，不要急着改大模型提示词。先按下面的顺序排查：

1. 商品是否真的被标记为召回。
2. 召回名单是否同步到了当前环境。
3. 商品匹配用的标识是否一致。
4. 合规检查是否真的被调用。
5. 配置表是否加载了正确的禁用词。
6. 日志里是否有其他系统覆盖了合规结果。

6.2 表格化排查

6.3 一个典型根因：校验逻辑被当成“建议”

很多实现把合规检查结果打印到日志里，却不中断流程。例如：

ok, errors = compliance_service.check_product(product) print(errors) # 继续生成

这是最隐蔽的坑。日志里出现了blocked，但流水线没有停止，视频照样生成，照样发布。合规检查必须是“硬闸门”，只要返回False，就必须中断当前分支。

6.4 另一个典型根因：召回名单不是结构化数据

有时候召回名单是运营手工整理的 PDF 或微信群消息，系统里只有一个模糊的商品名。这种非结构化数据很难支撑自动化匹配。建议把召回信息至少整理成identifier, reason, recalled_at, source四列，并统一标识规则。

7. 学习环境与生产环境的差异

7.1 能力边界对比

7.2 生产环境必须补齐的五类保障

第一，审计日志。每一次合规检查需要记录检查时间、检查项、结果、操作人、关联商品 ID。这样出问题时能定位是规则问题还是数据问题。

第二，发布后撤回。发布模块要保存商品和视频的关系。如果商品被召回，系统可以根据商品 ID 反查所有已发布视频并自动下线。

第三，监控告警。合规拦截率突然下降，可能是禁用词表被误删；召回同步失败，可能是数据源接口变更。需要监控这些指标，而不是只监控视频生成量。

第四，权限隔离。谁可以改禁用词表，谁可以绕过闸门发布，谁可以手动修复召回名单，都需要明确的权限控制。

第五，回滚方案。规则调整后如果误伤正常商品，要能快速回滚到上一个规则版本。

8. 可复用清单与扩展方向

8.1 上线前合规检查清单

• 商品主数据是否包含 SKU、UPC 等可用于召回匹配的稳定标识。
• 召回名单是否支持定时同步、增量更新和缓存过期。
• 合规闸门A是否在调用 LLM 之前执行。
• 文案生成是否要求结构化 JSON 输出。
• 文案校验是否覆盖禁用词、绝对化用语、免责声明。
• 视频合成前是否再次检查文案。
• 发布前是否重新检查商品状态。
• 发布后是否有周期性召回比对任务。
• 违规内容被拦截时，日志是否记录完整链路。
• 是否保留了人工审核通道，尤其是保健品、功能食品等高风控品类。

8.2 扩展方向

这套流水线可以继续扩展的方向包括：接入多模态审核模型，对视频画面、字幕、语音做交叉检查；增加证据链模块，让文案中的每个功效词都能追溯到检测报告或文献来源；把合规规则做成可视化配置平台，让运营和法务可以自行调整禁用词表，而不需要改代码。

如果想在本地搭一个多智能体内容生产环境，用来演练不同商品、不同规则之间的组合效果，可以基于输入材料中提到的开源项目 my_ai_town 做二次开发。先跑通它的玩具场景，再替换成自己的商品库和合规规则，会比直接写生产系统更容易理解整个链路。

AI 带货视频的价值在于规模化，规模化的前提是可控。合规网关不是用来拖慢生产速度的额外负担，它本身就是质量系统的一部分。把召回名单、禁用词、证据校验、发布后撤回做成流水线的内置能力，才能让内容工厂停留在“工厂”而不是“垃圾场”的定义上。