AI带货视频流水线合规网关：从商品召回检测到发布前审核的工程实践

AI 批量生成带货视频的工程实践在近两年已经相当常见。借助大模型写文案、TTS 配音、ffmpeg 合成画面一套流水线可以在几分钟内产出一批短视频。但这个效率背后藏着一个很容易被忽略的问题如果商品本身已经被监管机构召回或者文案使用了没有证据支撑的功效表述生成速度越快风险扩散越快。近期关于“AI TikTok Shop slop factory”的讨论正是这类问题的集中体现批量生产的内容大量堆上货架却没有人检查商品是否仍然可以售卖、文案是否合规、平台下架后是否还能自动撤回。本文不评价某个具体平台或案例而是从工程角度拆解一套面向短视频电商的 AI 带货视频流水线重点解决一个核心问题如何在商品导入、文案生成、视频合成、发布前后这几个环节里加入可执行的合规网关。读完这篇文章你可以得到一个最小可运行的设计思路、一组关键代码片段以及一套用于排查“为什么违规内容没拦住”的排查清单。1. 先理解“AI 内容工厂”为什么会在合规上失控1.1 不要把内容自动化和内容垃圾化混为一谈“内容工厂”本身是一个中性的工程概念。一个团队每天需要产出大量商品短视频如果完全靠人工写脚本、人工剪辑、人工配音成本会高到不可接受。于是就有了自动化流水线商品数据进系统脚本由大模型生成音频由 TTS 合成视频由 ffmpeg 拼接最后定时发布。问题出在“只追求产量、不检查质量”的实现方式上。当流水线只关心“能不能生成”不关心“生成的内容是否合法、商品是否安全”它就退化成内容垃圾站。英文里用 slop 描述这种内容意思是量大、低质、没有人工负责。真正的工程问题不是“要不要自动化”而是“自动化之后谁来把关把关逻辑放在哪一层”。1.2 失控链路商品、文案、视频、发布四个环节各自为政大部分翻车案例都不是单个环节出错而是整条链路缺少状态同步。看一条典型的失控链路运营把一批商品导入系统商品状态标记为“可推广”。大模型根据商品名生成文案文案里出现“可以预防某类疾病”之类的表述。TTS 把文案读成音频视频合成模块把音频、图片、背景音乐拼成成片。发布模块把视频上传到短视频电商平台。商品其实已经被监管机构召回但商品状态在第一步之后就没有再更新。这里最致命的地方是每个模块都完成了自己的局部任务却没有一个模块回答“这个商品现在还能不能卖、这段话有没有证据支持”。召回状态变更通常发生在商品导入之后、视频发布之前或者在视频发布之后。如果不做二次检查系统就会持续推广一个已经不具备销售资格的商品。大模型还会放大这个问题。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 等基础设施。依赖版本建议作用Python3.10 或 3.11主开发语言openai1.x调用 LLM 接口示例中用于文案生成edge-tts6.x免费 TTS适用于中文和英文配音ffmpeg6.0视频拼接、转码pydantic2.x数据模型和输入校验SQLite内置学习环境存储商品、召回名单和日志如果原始项目没有固定大模型供应商代码中会保留一个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_factorylist) 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, encodingutf-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 合规检查服务合规检查服务是整个系统的核心。它负责回答三个问题商品是否在召回名单里。商品品类是否允许自动推广。文案是否包含禁用词、绝对化用语或未经证实的功效表述。# 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( modelgpt-4o-mini, messages[ {role: system, content: SYSTEM_PROMPT}, {role: user, content: user_prompt}, ], temperature0.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, voicezh-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, checkTrue) 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 合规检查参数表参数含义推荐值调大影响调小影响错误配置表现DISALLOWED_CATEGORIES禁止自动推广的品类集合按业务维护拦截范围变广减少违规可能放过高风险品类违规商品进入文案生成DISALLOWED_TERMS文案禁用词列表由法务和运营维护模型可用表达变少违规表述漏网生成大量绝对化文案temperatureLLM 采样随机性0.2 到 0.4文案更多样但更不可控文案更稳定但略死板偏离合规约束max_script_length口播文案长度上限80 到 120 字信息更完整但视频变长表达受限视频时长不稳定recall_sync_interval召回名单同步周期学习环境 1 小时生产环境 10 分钟名单更新慢召回响应慢接口压力大已召回商品仍在推广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 双闸门与发布后撤回这里把两个闸门的检查时机和检查内容整理成一张表检查点检查时机检查内容失败处理合规闸门A生成文案之前商品状态、召回名单、品类停止生成并记录原因文案校验生成文案之后禁用词、免责声明、输出格式重新生成重试上限 2 次合规闸门B发布之前商品状态、召回名单、文案再次校验拒绝发布并进入人工队列发布后监控发布后周期性执行召回名单增量比对自动下线对应视频并告警发布后监控是经常被忽略的环节。商品召回不是只有“导入前”和“生成前”两个时间点完全可能发生在视频发布之后。没有发布后监控就等于允许已发布内容永远停留在平台上。5. 运行验证与测试用例5.1 最小运行步骤在本地跑通这条链路建议先用 CSV 文件准备两个商品一个正常商品一个已经被召回的演示商品。python main.py \ --products data/products.csv \ --recall data/recall_products.csv \ --output out/入口脚本会依次执行加载商品、同步召回名单、闸门A检查、生成文案、校验文案、生成音频、合成视频、闸门B检查。5.2 预期输出正常商品生成视频正常商品应该打印下面这样的日志[compliance] productP001 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] productP001 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] productP002 blocked: 命中召回名单 [script] skip P002 because compliance gate A failed此时不应生成任何视频文件。这个用例用于验证召回名单确实参与流水线决策而不只是被打印出来。5.4 测试用例功效词必须被拦截直接构造一段包含“治愈”的文案传入ComplianceService.check_script。预期结果是返回False错误信息包含“文案包含禁用词: 治愈”。这类测试建议接入单元测试框架比如 pytest。每次修改合规规则后跑一遍测试用例可以快速发现规则配置错误。6. 常见问题排查为什么该拦截的没拦住6.1 排查顺序当出现违规内容漏过时不要急着改大模型提示词。先按下面的顺序排查商品是否真的被标记为召回。召回名单是否同步到了当前环境。商品匹配用的标识是否一致。合规检查是否真的被调用。配置表是否加载了正确的禁用词。日志里是否有其他系统覆盖了合规结果。6.2 表格化排查问题现象常见原因检查方式处理建议已召回商品仍生成视频商品状态未同步或匹配标识不一致检查商品库和召回名单的主键字段统一使用 SKU 或 UPC 作为匹配键召回名单同步后仍不生效缓存过期时间过长检查缓存 TTL 和同步日志缩短 TTL在同步完成后主动刷新缓存文案出现明显违规功效词提示词缺少约束或输出未校验检查生成日志和check_script是否被调用增加输出 JSON schema 校验禁止直接信任模型输出闸门A通过但闸门B拦截生成期间商品状态发生变化查看闸门A和闸门B的时间戳以发布前检查结果为准视频下线失败发布后监控没有绑定商品 ID检查视频和商品的关联表发布时保存product_id - video_url映射6.3 一个典型根因校验逻辑被当成“建议”很多实现把合规检查结果打印到日志里却不中断流程。例如ok, errors compliance_service.check_product(product) print(errors) # 继续生成这是最隐蔽的坑。日志里出现了blocked但流水线没有停止视频照样生成照样发布。合规检查必须是“硬闸门”只要返回False就必须中断当前分支。6.4 另一个典型根因召回名单不是结构化数据有时候召回名单是运营手工整理的 PDF 或微信群消息系统里只有一个模糊的商品名。这种非结构化数据很难支撑自动化匹配。建议把召回信息至少整理成identifier, reason, recalled_at, source四列并统一标识规则。7. 学习环境与生产环境的差异7.1 能力边界对比能力学习环境生产环境召回名单来源本地 CSV监管接口、内部风控系统、平台下架数据存储SQLitePostgreSQL关键表做主从同步LLM 调用个人 API Key独立账号按项目隔离配额有内容审计日志审核不设人工高风控品类必须人工抽审日志控制台输出集中式日志平台保留 180 天发布后撤回手动清理定时任务 消息队列 告警配置本地 yaml配置中心变更可回滚7.2 生产环境必须补齐的五类保障第一审计日志。每一次合规检查需要记录检查时间、检查项、结果、操作人、关联商品 ID。这样出问题时能定位是规则问题还是数据问题。第二发布后撤回。发布模块要保存商品和视频的关系。如果商品被召回系统可以根据商品 ID 反查所有已发布视频并自动下线。第三监控告警。合规拦截率突然下降可能是禁用词表被误删召回同步失败可能是数据源接口变更。需要监控这些指标而不是只监控视频生成量。第四权限隔离。谁可以改禁用词表谁可以绕过闸门发布谁可以手动修复召回名单都需要明确的权限控制。第五回滚方案。规则调整后如果误伤正常商品要能快速回滚到上一个规则版本。8. 可复用清单与扩展方向8.1 上线前合规检查清单商品主数据是否包含 SKU、UPC 等可用于召回匹配的稳定标识。召回名单是否支持定时同步、增量更新和缓存过期。合规闸门A是否在调用 LLM 之前执行。文案生成是否要求结构化 JSON 输出。文案校验是否覆盖禁用词、绝对化用语、免责声明。视频合成前是否再次检查文案。发布前是否重新检查商品状态。发布后是否有周期性召回比对任务。违规内容被拦截时日志是否记录完整链路。是否保留了人工审核通道尤其是保健品、功能食品等高风控品类。8.2 扩展方向这套流水线可以继续扩展的方向包括接入多模态审核模型对视频画面、字幕、语音做交叉检查增加证据链模块让文案中的每个功效词都能追溯到检测报告或文献来源把合规规则做成可视化配置平台让运营和法务可以自行调整禁用词表而不需要改代码。如果想在本地搭一个多智能体内容生产环境用来演练不同商品、不同规则之间的组合效果可以基于输入材料中提到的开源项目 my_ai_town 做二次开发。先跑通它的玩具场景再替换成自己的商品库和合规规则会比直接写生产系统更容易理解整个链路。AI 带货视频的价值在于规模化规模化的前提是可控。合规网关不是用来拖慢生产速度的额外负担它本身就是质量系统的一部分。把召回名单、禁用词、证据校验、发布后撤回做成流水线的内置能力才能让内容工厂停留在“工厂”而不是“垃圾场”的定义上。