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 或更高版本。依赖尽量少，降低环境折腾成本：

安装命令：

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": f"Bearer {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", headers=headers, json=payload, timeout=30 ) 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_factory=dict) @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_factory=lambda: 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( content=content, length=length, hit_risk_words=hit_risk_words, hit_keywords=hit_keywords, length_score=length_score, keyword_score=keyword_score, risk_score=risk_score, format_score=format_score, total_score=total_score, decision=decision ) 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 "reject"

3.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", encoding="utf-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(force=True) 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(host="0.0.0.0", port=5000, debug=True)

3.4 SQLite 落库，保证每次决策都可追溯

决策日志是这套系统里容易被忽略的部分。没有日志，就没法回查“为什么这条内容被拒了”。

# database.py import sqlite3 class DecisionDatabase: def __init__(self, db_path: str): self.conn = sqlite3.connect(db_path, check_same_thread=False) 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。这个影响是全局的，所以生产环境调整权重时，必须跑一批历史样本回归对比，而不是上线后靠肉眼观察。

权重配错时的典型表现：

5. 运行验证：从启动到看决策结果

5.1 启动服务

export LLM_API_KEY="your-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 调用失败或超时

现象：启动服务后第一次请求一直转圈，最后报错。

推荐做法：把 LLM 调用失败和超时都做成降级策略，比如返回错误码而不是让整个服务崩溃，或者走示例文本分支保证演示流程不断。

6.2 规则不生效

现象：明明在config.yaml里加了关键词，但评分没有变化。

排查顺序：

1. 确认修改的是启动时读取的配置文件，而不是另一个目录下的同名文件。
2. 确认 Flask debug 模式是否自动重启。如果没重启，需要手动重启进程。
3. 在RuleEngine.__init__里打印self.keyword_bonus，确认配置加载成功。
4. 检查关键词是否真的出现在内容里。例如“修仙”和“修仙世界”是包含关系，但“修仙界”并不包含“修仙世界”。

常见原因是修改配置后没有重启进程，或者路径写错。建议在启动日志里打印配置文件的绝对路径。

6.3 中文乱码与 JSON 解析失败

现象：返回内容正常，但控制台出现乱码，或者模型输出带 Markdown 代码块导致 JSON 解析失败。

处理方式：

• 代码里读取文件和 HTTP 响应时，固定使用encoding="utf-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 及以上 review，80 及以上 pass”，那么边界逻辑要明确。

推荐写法是把这个逻辑做成配置：

decision: pass_threshold: 80 review_threshold: 60 pass_include_equal: true review_include_equal: true

不要在代码里写死边界，否则产品调整需求时又要走一次发布流程。

7. 生产环境最佳实践：从最小演示到可维护系统

7.1 学习环境与生产环境的差异

7.2 规则配置外置化和热更新

生产环境不允许每次改阈值都重启服务。常见的做法是：

1. 把规则配置放到 Nacos、Apollo 或 etcd 这类配置中心。
2. 在内存里维护一份规则快照。
3. 监听配置变更事件，更新快照。
4. 每次评估时加一个规则版本号，写进日志，方便追溯“这条内容是用哪个版本的规则判的”。

规则版本号是很容易被忽略的点。没有版本号，规则变更后你就无法解释历史数据为什么和现在的判断不一致。

7.3 人工复核与灰度放量

当 review 类内容比较多时，要设计一个人工复核后台。建议：

• 列表展示待审内容，以及每个维度的评分明细。
• 支持人工修改决策，并记录修改人、修改时间、修改原因。
• 定期分析“人工修改决策”的数据，用它来调整规则阈值。

这相当于把算法当成初审人员，而不是最终决策者。在“数字生死簿”的场景里，可以理解为：算法先做初判，保留一个人工赦免通道，避免规则误杀。

7.4 可复用清单：评估系统上线前检查表

每个想要上线的内容评估系统，建议至少过一遍下面这个清单：

• 每条评估记录都有唯一请求 ID 和规则版本号。
• 所有评分维度的原始值和加权值都能在日志中查到。
• 大模型异常时有降级策略，不会直接 500。
• 风险词表有更新流程，而不是沉淀在代码里。
• 决策阈值调整前，先用历史样本做回归对比。
• review 类内容有人工处理流程，不会积压。
• 数据库表有索引，查询决策日志不会全表扫描。
• 接口有鉴权和限流，防止被刷。

8. 扩展方向：从“数字生死簿”到完整的 AI 内容工作流

这个最小系统解决的是“文本内容生成后如何评估决策”的问题。如果继续往下拓展，可以往三个方向走。

方向一：把文本评估扩展到视频和短剧素材。AI 短剧、AI 漫剧的一键成片系统，通常要评估的不仅是文案，还有分镜脚本、画面描述、音频文本。可以在生成每个分镜时都做一次评估，提前拦截问题内容，而不是等成片后再返工。

方向二：引入更复杂的匹配算法。当前关键词匹配是简单包含，实际项目里可以替换成 AC 自动机处理大规模词表，或者用编辑距离做近义词判定，甚至用向量召回做语义相似度判断。基础算法在规则层依然有很高的性价比。

方向三：加入机器学习模型。当规则条目膨胀到几十条甚至上百条，维护成本会明显上升。这时可以基于历史决策日志训练一个打分模型，把规则引擎的结果作为特征之一。注意，模型上线后要保留规则引擎作为兜底和解释机制，不要让决策完全黑盒化。

对新手来说，最有价值的练习不是一开始就写完整平台，而是把这个最小闭环跑通，再逐步替换其中某一块：把 SQLite 换成 MySQL，把规则引擎换成 Drools，把关键词匹配换成 AC 自动机。每一次替换都能加深对“生成-评估-决策”这条链路某个环节的理解。

需要记住的一点是：算法控制从来不只是“模型给个分数”这么简单。真正决定系统质量的，是规则设计是否合理、数据是否能追溯、异常是否有人兜底。把这些做扎实，“数字生死簿”式的自动化决策系统才能从玩具变成工具。