Python 内容审核：Schema 门控批量分类 + 人工复核队列

决策、不变式与失败边界

决策的核心是优化有效且可追溯的决策，而非最低的 token 价格。批量执行之所以有用，是因为这类工作负载不需要在上传请求中立即返回答案，但批量处理是一种调度选择，而非正确性机制。相同的验证和升级规则必须适用于每一个条目。

四个不变式定义了边界：

每个被接受的结果必须匹配一个固定的 schema 版本和一个封闭的标签集。

每个自动拦截或放行都必须指向不可变的源哈希、prompt 版本、分类器版本和决策 ID。

重试不能产生第二个 moderation 操作或第二个审查工单。

缺失字段、未知标签、矛盾的理由以及不确定的发票提取都必须进入审查流程；它们永远不会变成猜测值。

第三条不变式很容易被低估。RFC 9110 区分了幂等方法，因为当操作具有副作用时，自动重试会改变风险。队列 worker 应该将这个理念延伸到 HTTP 方法选择之外：通过稳定的内容哈希来声明工作，在幂等键下记录决策，并使最终状态转换变为有条件的。如果 worker 在分类后丢失了租约，另一个 worker 可以重复计算，但绝不能重复用户可见的操作。

将失败类别分开处理。传输失败意味着带上限和抖动的重试。Schema 失败意味着隔离响应并创建一个审查项。策略歧义意味着正常审查。容量耗尽意味着在摄入时施加背压。将这些状态混合到一个通用的失败桶中会使成本预测变得不可能，并会引发重试风暴。

对于发票上传，验证也保护下游会计数据。currency: USD 但没有发票总额是不完整的提取，而 label: allow 附带一个仅为被拦截内容保留的原因码是矛盾的 moderation 结果。两者都是结构上可读的 JSON。但都不适合自动化。

批量 LLM 分类如何为大容量的内容审查队列提供支持？

在摄入、分类和审查之间使用持久化条目记录。摄入请求应该在存储源引用和哈希之后就返回；它不应该等待分类完成。然后 worker 根据年龄和大小限制形成批次，即使模型请求将多个条目分组在一起，也要保留每个条目的身份。当结果返回时，按身份拆分，独立验证，并将每个决策通过幂等状态转换提交。

人工队列需要明确的理由，而不是单一的 needs_review 标签。至少要区分：策略歧义、无效的结构化输出、提取冲突和手动抽样。这些通道有不同的 staffing 影响。合规专员可能需要策略通道，而数据运营审查员可以解决发票字段冲突。垃圾邮件过滤工作在这里提供了一个有用的架构教训——送达失败、速率限制和内容决策远看很相似，但将它们合并到一个重试循环中会破坏修复任何一个问题所需的信号。

不要将模型的自由格式解释暴露为审查员记录。存储受限的原因码，在策略允许的情况下，存储与源偏移量绑定的简短证据片段。源始终是权威的。解释是辅助材料。这也限制了包含指令的供应商备注成为审查界面中的操作命令的可能性。

一个实用的队列策略使用三个出口：自动化满足每个不变式的决策，升级歧义但有效的结果，或隔离无效的结果。后两者可以共享用户界面，但不应该共享指标。歧义率描述策略边界；无效输出率描述分类契约。

我不确定任何固定的置信度阈值可以在游戏类别、语言和供应商模板之间不变地转移。带标签的评估集可以解决这种不确定性。按策略切片测量结果，然后仅为有足够代表性样本的切片设置阈值；否则将该切片路由到审查。随着上传组合的变化，你的里程会有所不同，因此阈值版本应该与 prompt 版本和 schema 版本放在一起。

通过结构化输出正确性比较操作选项

最便宜的设计取决于坏自动化要花多少成本，以及存在多少审查容量。这张表将价格作为一个输入，而不是裁决。

对于这个游戏发票案例，在 LLM 之前选择规则，然后在自动化之前要求 schema 验证。这不适合上传稀少且策略每天都在变化的情况；在团队构建标签集并稳定决策分类法时，坚持人工审查。当供应商使用少量合同模板且 moderation 关注点是精确的、可检查的模式时，仅规则方案仍然有效。

比较也暴露了一个被拒绝的捷径：使用宽松的 JSON 解析器作为正确性测试。可解析的输出证明了语法，而不是语义。封闭枚举、必填字段、跨字段检查和源链接证据定义了有用的契约。

Token 计数和能随队列增长的成本估算

从实际上传组合的样本中构建估算。使用精确用于计费的 tokenizer 计算 token 数，在应用实际 prompt 模板和序列化之后。字符计数启发式方法可以接受，用于粗略的容量范围，但它们不是账单台账，特别是当供应商名称、零件代码和多语言备注改变 tokenization 时。

对于一个周期，计算：

model_cost = input_tokens * input_rate + output_tokens * output_rate

review_cost = reviewed_items * average_review_minutes * loaded_reviewer_rate_per_minute

total_cost = model_cost + review_cost + queue_infrastructure_cost

将费率放在带日期的配置中，而不是文章正文或 worker 代码中。有用的预测报告 token 数、审查速率和审查时间的范围。单一的点估计隐藏了昂贵的分支：无效或歧义结果的微小变化可能将大量工作转移到人工队列中。

假设一个规划样本包含 10,000 次上传。不要将一个平均 prompt 乘以 10,000 然后停止。将样本按语言、文件类型、供应商模板和策略标签分区；测量每个切片的输入 token、输出 token、schema 有效率、自动决策率和审查员分钟数。然后使用预期的生产组合对切片进行加权。这些是规划数量，而不是声称的基准。当组合、策略、prompt、schema 或分类器发生变化时，重新运行计算。

Token 计数应该发生两次。在分发之前，它强制执行每个条目和每个批次的上限，这样一个巨大的 OCR 载荷不会挤掉其余的。在完成之后，记录的使用量协调估算值并为下一次预测提供数据。如果运行时没有返回权威的使用量，明确记录这个差距并与账单导出进行协调；不要将估算值静默地视为观察到的支出。

审查容量需要自己的公式：required_reviewer_minutes = arrival_count * review_rate * average_review_minutes。将那个需求与同一时间间隔内的值班分钟数进行比较，并包括到达高峰假设。批量折扣或较低的 token 费率无法拯救审查到达率超过服务容量的队列。

每处理条目的成本是有用的，但每正确最终决策的成本是更好的比较轴。它将无效输出和错误自动化记到产生它们的架构账上。建立这个分母需要带标签的审计样本，包括自动批准和拦截；只审查升级无法揭示绕过队列的错误。

Python 中的关键路径

下面的示例将商业 API 放在通用分类器接口后面。它验证一个小的结果契约，应用跨字段规则，并从策略版本和源字节派生幂等键。生产代码会将这些转换原子化；这里重要的是自动化在哪里停止。

from dataclasses import dataclass
from decimal import Decimal
from hashlib import sha256
from typing import Any, Literal, Protocol

Decision = Literal["allow", "block", "review"]
REASONS = {"safe", "prohibited_content", "ambiguous", "invalid_invoice"}

@dataclass(frozen=True)
class ModerationResult:
    decision: Decision
    reason: str
    invoice_id: str | None
    supplier: str | None
    currency: str | None
    total: Decimal | None

class Classifier(Protocol):
    def classify(self, payload: bytes, schema_version: str) -> dict[str, Any]: ...

def parse_result(raw: dict[str, Any]) -> ModerationResult:
    required = {"decision", "reason", "invoice_id", "supplier", "currency", "total"}
    if set(raw) != required:
        raise ValueError("result keys do not match the pinned schema")
    if raw["decision"] not in {"allow", "block", "review"}:
        raise ValueError("unknown decision")
    if raw["reason"] not in REASONS:
        raise ValueError("unknown reason")

    total = Decimal(str(raw["total"])) if raw["total"] is not None else None
    result = ModerationResult(
        decision=raw["decision"],
        reason=raw["reason"],
        invoice_id=raw["invoice_id"],
        supplier=raw["supplier"],
        currency=raw["currency"],
        total=total,
    )
    if result.decision == "allow" and (not result.invoice_id or not result.currency or total is None):
        raise ValueError("an allowed invoice requires complete accounting fields")
    if result.decision == "allow" and result.reason != "safe":
        raise ValueError("allow requires the safe reason")
    return result

def classify_for_queue(
    content: bytes, policy_version: str, schema_version: str, classifier: Classifier
) -> tuple[str, ModerationResult | None, str]:
    item_key = sha256(policy_version.encode() + b":" + content).hexdigest()
    try:
        result = parse_result(classifier.classify(content, schema_version))
    except (ValueError, ArithmeticError):
        return item_key, None, "invalid_output_review"

    if result.decision == "review":
        return item_key, result, "policy_review"
    return item_key, result, "automated_decision"

代码故意不在 classify_for_queue 内部重试。重试策略属于 worker，在那里它可以区分传输错误和契约错误，并尊重条目的尝试预算。状态存储应该在 item_key 上强制执行唯一性；然后条件写入使在决策边界处的重投递变得无害。

Prompt 结构仍然很重要。明确地定义策略、允许的标签、schema 和不受信任的发票文本的处理方式，并根据带标签的示例评估变更。Prompt Engineering Guide 是 prompt 技术的有用目录，但没有 prompt 指令可以取代生成后的验证。

被拒绝的选项及其适用场景

被拒绝的选项是在上传请求内进行同步分类。它看起来更简单，因为没有可见的队列，但它将用户延迟和请求重试与分类延迟耦合在一起，使负载峰值更难吸收，并诱使处理程序在不确定的超时后重复副作用。对于大容量 moderation 来说，这是错误的失败边界。

它有一个有效的用例。当容量低、调用者真正需要在继续之前做出决策、请求截止日期轻松覆盖操作，且仍强制执行幂等状态转换时，选择同步处理。预发布聊天消息可能有那个要求；可以显示处理状态的供应商发票通常没有。

所选批量设计的一个更广泛的限制是反馈延迟。当策略解释变化速度快于带标签评估集的维护速度时，它也是一个糟糕的选择。在这些条件下，将更多条目路由到人工并接受更高的审查成本。自动化应该一次一个策略切片地赢得其范围。

没有通用最便宜的运行时。防御性的选择是能最小化每个正确、可审计决策的成本，同时保持在队列容量和错误预算内的那个选择。对于带有发票的用户内容，schema 有效性和跨字段一致性优先于 token 价格；然后批量处理、token 上限和确定性预过滤器在不削弱该契约的情况下减少可避免的工作。