AI内容生成后如何自动评估与决策？用Python实现规则引擎驱动的数字生死簿

在 AI 全民制作人的内容生产链路里“AI 生成一段剧情、旁白、文案、分镜脚本”只是起点。真正麻烦的是生成之后怎么办这段内容质量够不够、能不能发布、需不需要人工确认、被拒的话是哪个维度出了问题。如果你把“数字生死簿”理解成一个娱乐化的算法决策场景——由算法为每段生成内容打分并决定它的去向那它背后的技术骨架就是一套“内容生成 自动评估 决策分流”的工程链路。这篇文章就用 Python 从零搭一个最小可运行版本。它不依赖复杂框架核心只做四件事调用大模型生成内容、用规则引擎对内容做多维评分、根据评分给出通过/转人工/拒绝的决策、把决策结果和日志落库。写完这套最小闭环你就能理解为什么“算法控制”不是一个黑盒而是一系列可配置、可排查、可调整的规则和参数。适合正在做 AI 内容工具、AI 短剧工作流或者想入门规则引擎和自动决策系统的开发者阅读。1. 先理解“生成-评估-决策”三段式里算法层到底管什么1.1 从 AI 短剧生成流程看内容流水线的真实需求不管做 AI 短剧、AI 漫剧还是一键成片系统内容流水线一般都可以拆成三段生成阶段根据用户输入的题材、角色、剧情方向调用大模型生成文案、分镜、配音稿或视频素材描述。评估阶段对生成结果做质量、格式、合规性、风格匹配度等多维度检查。决策阶段根据评估分数决定内容进入发布、进入人工复审还是被直接丢弃。很多人在刚接触 AI 内容工具时会把精力全部放在“提示词怎么写、模型怎么调”上忽略了中间的评估层和决策层。结果就是模型输出什么就发布什么出了问题只能人工一条条看。放到“数字生死簿”这个场景里相当于只有判官没有规则也没有申诉通道。1.2 “数字生死簿”示例的业务链路设计这个示例不讨论真实的伦理判断只是借“生死簿”这个意象来模拟一段内容生成后由算法自动评估它的“状态”并决定它的命运。完整的业务链路如下输入剧情主题 - 大模型生成剧情文案 - 规则引擎提取特征长度、关键词、风险词、评分 - 多维度加权计算 - 决策输出pass / review / reject - 写入 SQLite 决策日志在这个链路里算法层的主要职责是把模型输出从“一段文本”变成“一组结构化指标”再根据指标组合给出决策。这个“指标化”的过程就是规则引擎和评分算法发挥作用的地方。1.3 算法层可以用什么技术实现算法层不一定要上机器学习。实际项目里最常见的是这几种组合规则引擎把“如果关键词 X 命中则扣分”“如果长度小于 N则转人工”这类业务规则独立出来方便运维调整不必改代码。加权评分模型为每个评估维度设置权重用加权和得到总分。基础算法能力比如用 KMP、AC 自动机做敏感词匹配用 DFA 做关键词过滤用编辑距离做相似内容去重。很多时候简单算法就够用。机器学习模型当规则复杂到无法人工维护时再考虑训练分类模型或打分模型。本文的示例采用“规则引擎 加权评分”的组合因为它是可解释性最强、最容易排错、也最适合作为教学起点的方式。2. 环境准备依赖、目录与配置要一次做对2.1 Python 版本与依赖清单开发学习环境建议使用 Python 3.10 或更高版本。依赖尽量少降低环境折腾成本依赖版本建议用途Flask3.0.x提供 Web 接口方便演示和测试requests2.31.x调用大模型 HTTP APIPyYAML6.0.x读取 yaml 配置文件SQLite3Python 内置存储决策日志无需单独安装安装命令mkdir digital-ledger cd digital-ledger python -m venv venv source venv/bin/activate pip install flask requests pyyaml注意这里选择 SQLite 是为了让示例在本地零依赖跑通。生产环境如果并发量大应换成 MySQL 或 PostgreSQL并在数据库层增加连接池和索引设计。2.2 项目目录结构推荐下面这个结构逻辑清楚后续扩展也方便digital-ledger/ ├── app.py # 主流程入口 ├── config.yaml # 配置文件 ├── models.py # 数据模型定义 ├── rules.py # 规则引擎实现 ├── llm_client.py # 大模型客户端封装 ├── database.py # SQLite 操作封装 ├── tests/ │ └── demo_cases.py # 演示用例 └── logs/ └── decision.log # 决策日志运行时生成2.3 配置文件编写把模型接口、评分规则、决策阈值都放到配置文件里。这样调整阈值时不必重新部署代码。llm: base_url: https://your-llm-service.example.com/v1 api_key: ${LLM_API_KEY} model: qwen-plus temperature: 0.8 max_tokens: 800 evaluation: risk_words: - 禁忌词A - 禁忌词B keyword_bonus: 修仙: 10 逆袭: 8 复仇: 5 weights: length_score: 0.2 keyword_score: 0.3 risk_score: 0.3 format_score: 0.2 decision: pass_threshold: 80 review_threshold: 60这里要解释几个关键参数temperature控制生成随机性取值 0 到 1。内容创作场景可以设 0.7 到 0.9让剧情更多样如果用于知识问答或稳定输出建议降到 0.3 以下。max_tokens限制单次生成长度防止模型输出过长导致接口超时或成本失控。weights四个维度权重加起来必须是 1.0否则总分会失真。pass_threshold和review_threshold是决策分界点。2.4 大模型服务接入说明示例代码使用兼容 OpenAI Chat Completions 接口的服务。你可以根据自己可用的服务商调整base_url、model和认证方式。# llm_client.py import os import requests class LLMClient: def __init__(self, config): self.base_url config[base_url] self.api_key os.getenv(LLM_API_KEY) or config.get(api_key, ) self.model config[model] self.temperature config.get(temperature, 0.8) self.max_tokens config.get(max_tokens, 800) def generate(self, prompt: str) - str: headers { Authorization: fBearer {self.api_key}, Content-Type: application/json } payload { model: self.model, messages: [ {role: system, content: 你是一个擅长创作短篇剧情的写手。}, {role: user, content: prompt} ], temperature: self.temperature, max_tokens: self.max_tokens } resp requests.post( f{self.base_url}/chat/completions, headersheaders, jsonpayload, timeout30 ) resp.raise_for_status() return resp.json()[choices][0][message][content]实际项目中还要补上超时重试、流式输出、错误码分类和调用成本统计这里只保留最小逻辑。3. 实现一个最小可运行的“内容生成-评估-决策”系统3.1 定义数据模型把文本变成结构化字段评估系统不能直接对字符串做加减法需要先把内容映射成一组可计算的字段。# models.py from dataclasses import dataclass, field from datetime import datetime dataclass class GenerationInput: topic: str style: str extra: dict field(default_factorydict) dataclass class EvaluationResult: content: str length: int hit_risk_words: list hit_keywords: list length_score: float keyword_score: float risk_score: float format_score: float total_score: float decision: str created_at: str field(default_factorylambda: datetime.now().isoformat())这里有一个容易误解的地方total_score是多个维度的加权和而不是某个单一维度的原始值。决策层只依赖total_score和个别“一票否决”字段比如命中高危险词时直接拒绝不参与总分。3.2 规则引擎核心逻辑这个规则引擎不做复杂抽象用“规则列表 顺序执行”实现。它的可读性最好也最容易加日志。# rules.py from models import EvaluationResult class RuleEngine: def __init__(self, config): self.risk_words config.get(risk_words, []) self.keyword_bonus config.get(keyword_bonus, {}) self.weights config.get(weights, { length_score: 0.2, keyword_score: 0.3, risk_score: 0.3, format_score: 0.2 }) def evaluate(self, content: str) - EvaluationResult: length len(content) hit_risk_words [w for w in self.risk_words if w in content] hit_keywords [k for k in self.keyword_bonus if k in content] length_score self._score_length(length) keyword_score self._score_keywords(hit_keywords) risk_score self._score_risk(hit_risk_words) format_score self._score_format(content) total_score ( length_score * self.weights[length_score] keyword_score * self.weights[keyword_score] risk_score * self.weights[risk_score] format_score * self.weights[format_score] ) total_score round(max(0, min(100, total_score)), 2) decision self._decide(total_score, hit_risk_words) return EvaluationResult( contentcontent, lengthlength, hit_risk_wordshit_risk_words, hit_keywordshit_keywords, length_scorelength_score, keyword_scorekeyword_score, risk_scorerisk_score, format_scoreformat_score, total_scoretotal_score, decisiondecision ) def _score_length(self, length: int) - float: if length 200: return 100.0 if length 100: return 70.0 return 40.0 def _score_keywords(self, hit_keywords: list) - float: score 0.0 for kw in hit_keywords: score self.keyword_bonus.get(kw, 0) return min(100.0, score * 10) def _score_risk(self, hit_risk_words: list) - float: if hit_risk_words: return 0.0 return 100.0 def _score_format(self, content: str) - float: if len(content) 20 and content[-1] in 。.!?: return 100.0 return 60.0 def _decide(self, total_score: float, hit_risk_words: list) - str: if hit_risk_words: return reject if total_score 80: return pass if total_score 60: return review return reject3.3 主流程生成、评估、决策、落库主流程要把前面几个模块串起来。注意判断是否配置了真实的模型服务如果没配就用内置示例文本保证演示环境能跑通。# app.py import yaml from flask import Flask, request, jsonify from llm_client import LLMClient from rules import RuleEngine from database import DecisionDatabase app Flask(__name__) with open(config.yaml, r, encodingutf-8) as f: config yaml.safe_load(f) llm LLMClient(config[llm]) engine RuleEngine(config[evaluation]) db DecisionDatabase(decisions.db) def build_prompt(topic: str, style: str ) - str: style_part f风格要求{style}。 if style else return f请根据主题“{topic}”创作一段200字左右的短剧剧情。{style_part}要求在结尾有一个反转。只输出剧情正文不要额外说明。 def generate_and_evaluate(topic: str, style: str ): try: content llm.generate(build_prompt(topic, style)) except Exception as e: # 没有配置模型服务时使用示例文本保证流程可演示 print(f[LLM调用失败使用示例内容] {e}) content f在修仙世界里主角原本只是一个杂役弟子。{topic}让他在绝境中觉醒。他凭借逆袭的意志最终复仇成功。这个故事告诉我们命运是可以被改变的。 result engine.evaluate(content) db.insert(result) return result app.route(/api/judge, methods[POST]) def judge(): data request.get_json(forceTrue) topic data.get(topic, 逆袭) style data.get(style, ) result generate_and_evaluate(topic, style) return jsonify({ content: result.content, total_score: result.total_score, decision: result.decision, hit_risk_words: result.hit_risk_words, hit_keywords: result.hit_keywords, scores: { length: result.length_score, keyword: result.keyword_score, risk: result.risk_score, format: result.format_score } }) if __name__ __main__: app.run(host0.0.0.0, port5000, debugTrue)3.4 SQLite 落库保证每次决策都可追溯决策日志是这套系统里容易被忽略的部分。没有日志就没法回查“为什么这条内容被拒了”。# database.py import sqlite3 class DecisionDatabase: def __init__(self, db_path: str): self.conn sqlite3.connect(db_path, check_same_threadFalse) self.conn.execute( CREATE TABLE IF NOT EXISTS decisions ( id INTEGER PRIMARY KEY AUTOINCREMENT, content TEXT, total_score REAL, decision TEXT, hit_risk_words TEXT, hit_keywords TEXT, created_at TEXT ) ) self.conn.commit() def insert(self, result) - None: self.conn.execute( INSERT INTO decisions (content, total_score, decision, hit_risk_words, hit_keywords, created_at) VALUES (?, ?, ?, ?, ?, ?), ( result.content, result.total_score, result.decision, ,.join(result.hit_risk_words), ,.join(result.hit_keywords), result.created_at ) ) self.conn.commit()落库之后出现争议时可以直接查表确认当时的评分明细而不是凭印象判断。4. 关键代码解析与参数取值说明4.1 提示词模板的稳定性和变量控制build_prompt里用了一个约定要求模型“只输出剧情正文不要额外说明”。这个约束很重要。很多模型的输出会带“好的以下是……”这类前缀会污染评估结果影响长度评分和格式评分。如果你在真实项目中遇到模型总爱加前缀可以考虑两种方式在提示词里加强约束例如“不允许出现任何解释性文字”。在后处理里用正则把“好的/以下是/当然可以”这类开头去掉。4.2 评分维度设计为什么是四维示例选择了长度、关键词、风险词、格式四个维度原因如下长度分防止生成过短或过长。关键词分奖励贴近目标题材的内容让“修仙”“逆袭”这类词产生正反馈。风险分作为一票否决项命中即归零决策层直接拒绝。格式分检查结尾符号保证内容适合直接展示或配音。四个维度各有各的目的而且每一项都能单独解释。这比一个黑盒模型输出一个分数更容易排查。4.3 决策阈值的边界处理当前_decide的逻辑总分 80 - pass 60 总分 80 - review 总分 60 - reject这里容易踩坑。比如一个内容总分恰好是 60会进入 review。但如果总分恰好是 79.99由于浮点精度问题可能被判定为低于 80 而进入 review。示例代码里已经用round(..., 2)做了处理但实际项目中如果有 80.0 这种边界值建议在配置里明确闭区间还是开区间。4.4 权重配置改变后会发生什么假设把risk_score的权重从 0.3 调成 0.5那么风险维度对总分的影响更大更多内容会被判为 reject。这个影响是全局的所以生产环境调整权重时必须跑一批历史样本回归对比而不是上线后靠肉眼观察。权重配错时的典型表现现象可能原因几乎所有内容都被拒risk 权重过高或风险词表过宽几乎全部通过pass 阈值过低或权重分配不合理review 数量暴增review 区间过宽阈值需要收敛总分出现 0 或 100 占比很高某个维度评分函数过于极端5. 运行验证从启动到看决策结果5.1 启动服务export LLM_API_KEYyour-api-key python app.py如果没配置 API Key代码会捕获异常并走示例文本分支不影响流程演示。5.2 调用接口验证curl -X POST http://127.0.0.1:5000/api/judge \ -H Content-Type: application/json \ -d {topic: 修仙逆袭, style: 热血}预期返回{ content: 在修仙世界里主角原本只是一个杂役弟子。修仙逆袭让他在绝境中觉醒。他凭借逆袭的意志最终复仇成功。这个故事告诉我们命运是可以被改变的。, total_score: 76.0, decision: review, hit_risk_words: [], hit_keywords: [修仙, 逆袭], scores: { length: 40.0, keyword: 100.0, risk: 100.0, format: 100.0 } }这里可以看到内容长度较短导致 length_score 只有 40虽然关键词和格式都拿满分总分还是被拉到了 76进入 review。这个结果能直观说明多维度加权的作用。5.3 验证拒绝分支用一段包含风险词的示例验证curl -X POST http://127.0.0.1:5000/api/judge \ -H Content-Type: application/json \ -d {topic: 测试风险词, style: dark}如果生成的内容命中config.yaml里的risk_words决策会直接是reject并且hit_risk_words数组会给出具体命中了哪个词。5.4 查看决策日志sqlite3 decisions.db SELECT id, total_score, decision, hit_keywords, created_at FROM decisions ORDER BY id DESC LIMIT 5;输出示例2|76.0|review|修仙,逆袭|2025-01-01T10:20:30 1|85.5|pass|修仙|2025-01-01T10:20:15日志能让你快速确认当前系统到底放行了什么、拒绝了什么这是评估系统上线后最基础的可观测能力。6. 常见问题排查从现象倒推原因6.1 大模型 API 调用失败或超时现象启动服务后第一次请求一直转圈最后报错。检查项操作是否配置 API Key确认LLM_API_KEY环境变量已设置base_url 是否正确确认是chat/completions接口的根路径网络是否可达直接 curl 测试接口连通性超时时间代码里timeout30模型生成慢时可以调到 60推荐做法把 LLM 调用失败和超时都做成降级策略比如返回错误码而不是让整个服务崩溃或者走示例文本分支保证演示流程不断。6.2 规则不生效现象明明在config.yaml里加了关键词但评分没有变化。排查顺序确认修改的是启动时读取的配置文件而不是另一个目录下的同名文件。确认 Flask debug 模式是否自动重启。如果没重启需要手动重启进程。在RuleEngine.__init__里打印self.keyword_bonus确认配置加载成功。检查关键词是否真的出现在内容里。例如“修仙”和“修仙世界”是包含关系但“修仙界”并不包含“修仙世界”。常见原因是修改配置后没有重启进程或者路径写错。建议在启动日志里打印配置文件的绝对路径。6.3 中文乱码与 JSON 解析失败现象返回内容正常但控制台出现乱码或者模型输出带 Markdown 代码块导致 JSON 解析失败。处理方式代码里读取文件和 HTTP 响应时固定使用encodingutf-8。调用模型接口时在headers里声明Accept: application/json。如果模型喜欢输出 json 包裹的内容需要先剥离代码块标记再解析。import re def clean_model_output(text: str) - str: text re.sub(r^(?:json)?\s*, , text.strip()) text re.sub(r\s*$, , text) return text.strip()6.4 阈值边界判断错误现象总分 60 的内容进入了 review但需求是 60 分以上进入 review。原因判断条件写成了total_score 60包含 60 分。如果需求是“60 以下拒绝60 及以上 review80 及以上 pass”那么边界逻辑要明确。推荐写法是把这个逻辑做成配置decision: pass_threshold: 80 review_threshold: 60 pass_include_equal: true review_include_equal: true不要在代码里写死边界否则产品调整需求时又要走一次发布流程。7. 生产环境最佳实践从最小演示到可维护系统7.1 学习环境与生产环境的差异项目学习示例生产环境数据库SQLiteMySQL/PostgreSQL 连接池配置yaml 本地文件配置中心支持热更新LLM 调用单次请求异步队列 重试 熔断日志print结构化日志 追踪 ID安全无鉴权接口鉴权 限流 审计测试手动 curl自动化回归 灰度对比7.2 规则配置外置化和热更新生产环境不允许每次改阈值都重启服务。常见的做法是把规则配置放到 Nacos、Apollo 或 etcd 这类配置中心。在内存里维护一份规则快照。监听配置变更事件更新快照。每次评估时加一个规则版本号写进日志方便追溯“这条内容是用哪个版本的规则判的”。规则版本号是很容易被忽略的点。没有版本号规则变更后你就无法解释历史数据为什么和现在的判断不一致。7.3 人工复核与灰度放量当 review 类内容比较多时要设计一个人工复核后台。建议列表展示待审内容以及每个维度的评分明细。支持人工修改决策并记录修改人、修改时间、修改原因。定期分析“人工修改决策”的数据用它来调整规则阈值。这相当于把算法当成初审人员而不是最终决策者。在“数字生死簿”的场景里可以理解为算法先做初判保留一个人工赦免通道避免规则误杀。7.4 可复用清单评估系统上线前检查表每个想要上线的内容评估系统建议至少过一遍下面这个清单每条评估记录都有唯一请求 ID 和规则版本号。所有评分维度的原始值和加权值都能在日志中查到。大模型异常时有降级策略不会直接 500。风险词表有更新流程而不是沉淀在代码里。决策阈值调整前先用历史样本做回归对比。review 类内容有人工处理流程不会积压。数据库表有索引查询决策日志不会全表扫描。接口有鉴权和限流防止被刷。8. 扩展方向从“数字生死簿”到完整的 AI 内容工作流这个最小系统解决的是“文本内容生成后如何评估决策”的问题。如果继续往下拓展可以往三个方向走。方向一把文本评估扩展到视频和短剧素材。AI 短剧、AI 漫剧的一键成片系统通常要评估的不仅是文案还有分镜脚本、画面描述、音频文本。可以在生成每个分镜时都做一次评估提前拦截问题内容而不是等成片后再返工。方向二引入更复杂的匹配算法。当前关键词匹配是简单包含实际项目里可以替换成 AC 自动机处理大规模词表或者用编辑距离做近义词判定甚至用向量召回做语义相似度判断。基础算法在规则层依然有很高的性价比。方向三加入机器学习模型。当规则条目膨胀到几十条甚至上百条维护成本会明显上升。这时可以基于历史决策日志训练一个打分模型把规则引擎的结果作为特征之一。注意模型上线后要保留规则引擎作为兜底和解释机制不要让决策完全黑盒化。对新手来说最有价值的练习不是一开始就写完整平台而是把这个最小闭环跑通再逐步替换其中某一块把 SQLite 换成 MySQL把规则引擎换成 Drools把关键词匹配换成 AC 自动机。每一次替换都能加深对“生成-评估-决策”这条链路某个环节的理解。需要记住的一点是算法控制从来不只是“模型给个分数”这么简单。真正决定系统质量的是规则设计是否合理、数据是否能追溯、异常是否有人兜底。把这些做扎实“数字生死簿”式的自动化决策系统才能从玩具变成工具。