8.0
热点
AI SCORE
技术实践2026-08-12 12:51
多租户文本审核这样设计：先分类再出图，成本边界清晰
dev.to · AI#AI工程#多租户#内容审核
Editor brief · 编辑速览
在图像生成前增加独立的分类审核层，根据完整请求（描述、风格、负向提示）返回 allow/review/block，分类结果与图像任务绑定同一租户账本，费用可解释。
简短回答：在图像生成之前对完整的图像请求进行分类，返回一个小的 JSON 决策，并将分类器和图像调用附加到同一个租户账本上。真正有用的设计选择是成本边界，而不是某个特定的审核端点。
对于一个物流产品来说，这个账本非常重要。承运人、仓库运营商和内部支持团队可能都通过同一个图像功能提交提示词，但他们的审核率和提示词大小不同。如果分类是一个不可见的辅助调用，团队就无法解释为什么某个租户的月度 AI 支出出现了变动。我更倾向于一种从 Notebook 到生产的路径，使得策略契约、租户密钥和评估记录从第一个示例开始就是明确的。
这个闸门在图像作业进入队列之前运行。它接收将发送给图像模型的确切文本：用户的描述、所选风格、负面提示词以及任何模板字段。它发出 allow、review 或 block 三种决策。只有 allow 才能创建图像作业。
这个边界是不可协商的。
将分类器视为策略组件，而不是安全神谕。应用程序拥有策略类别以及对每个结果采取的操作。JSON schema 使边界易于验证，但仅有效的 JSON 并不能说明策略是有用的。
下面的示例使用纯 HTTP 形式的配置，这样应用程序可以指向其选择的聊天分类器和图像服务。使用 Python 是因为我的生产示例需要贴近评估测试框架；相同的请求体适用于 Node.js HTTP 客户端。URL 是配置项，而不是推荐。
import json
import os
import urllib.request
import uuid
DECISIONS = {"allow", "review", "block"}
def post_json(url, payload, headers):
request = urllib.request.Request(
url,
data=json.dumps(payload).encode("utf-8"),
headers={**headers, "Content-Type": "application/json"},
method="POST",
)
with urllib.request.urlopen(request, timeout=60) as response:
return json.loads(response.read().decode("utf-8"))
def moderate_and_queue(tenant_id, prompt, style, negative_prompt=""):
candidate = {
"prompt": prompt,
"style": style,
"negative_prompt": negative_prompt,
}
headers = {"Authorization": f"Bearer {os.environ['AI_API_KEY']}"}
result = post_json(
os.environ["CLASSIFIER_URL"],
{
"input": candidate,
"response_schema": {
"type": "object",
"properties": {
"decision": {
"type": "string",
"enum": ["allow", "review", "block"],
},
"reason": {"type": "string"},
},
"required": ["decision", "reason"],
"additionalProperties": False,
},
},
headers,
)
decision = result.get("decision")
if decision not in DECISIONS:
return {"decision": "review", "reason": "Invalid policy result"}
if decision != "allow":
return result
job_id = str(uuid.uuid4())
job = post_json(
os.environ["IMAGE_URL"],
{"job_id": job_id, "prompt": candidate},
{**headers, "Idempotency-Key": job_id},
)
return {"decision": "allow", "job": job}
print(moderate_and_queue(
tenant_id="warehouse-west",
prompt="A clean loading dock diagram at sunrise",
style="technical illustration",
))
有一个细节是刻意保持平淡的：格式错误的结果会变成 review，绝不会是 allow。网络故障应该在周围的工作进程中采取相同的非生成路径，带有持久化的原因和重试策略。图像请求携带一个幂等性密钥，因为重试不能意外创建两个作业；HTTP 重试行为和幂等性是应用程序的关注点，应该基于 RFC 9110 的语义来设计，而不是从客户端库中猜测。
示例还将 tenant_id 传递到函数中，尽管远程负载不需要它。这是为了提醒我们在本地记录所有权。在分类之前，创建一个包含租户、请求 ID、策略版本和提示词 token 估算值的会计记录。每次调用后，追加服务报告的使用量（如果可用的话）。不要将分类器和图像的使用量合并为一个数字。
第一个故障是部分输入。分类器看到主提示词，但看不到模板的风格字段，所以一个看起来无害的请求可能在后续获得含义。构建一个规范的候选对象，为分类序列化该对象，并使用相同的对象进行图像生成。这消除了一类非常常见的策略漂移。
第二个故障是结果模糊。"可能没问题"不是一种生产状态。保持契约小而精，拒绝未知的枚举值，并将 review 发送到带有足够上下文的人工审核队列。block 响应应该对请求者是通用的；详细的策略推理属于受限的操作日志。
第三个故障是成本归属。一个提示词较长的租户可能消耗更多的分类器 token，即使每张图像都被阻止了。这是一个真实的成本，也是一个有用的信号。分别跟踪分类器输入大小、图像输入大小、结果、延迟和重试次数。租户视图应该同时回答"这个请求花了什么？"和"它在哪里停止的？"
以下是我会放在队列设计旁边的决策表：
这是一张小表，但它防止了一个常见的设计错误：将安全和会计视为独立的中间件。它们观察同一个请求，所以它们应该共享同一个请求 ID。
考虑一个租户提交了一个很长的提示词，由路线描述、仓库预设和用户选择的风格组合而成。分类器可能返回 block，因此不会生成图像，但分类尝试仍然消耗了时间和 token。如果账本只记录成功的图像，租户会在其活动仪表板和发票之间看到一个无法解释的差距。如果账本只记录原始文本，运营团队在试图解决财务问题时制造了一个隐私问题。有用的记录更窄：租户 ID、请求 ID、策略版本、阶段、测量到的使用量、决策和一个短期指纹。该记录支持对账，而不会将每份成本报告变成提示词存储的副本。
从一个分为普通创意请求、明确策略违规、模糊案例以及在可编辑字段中隐藏指令的尝试的评估集开始。存储预期的操作，而不仅仅是标签。当策略提示词、分类器模型、JSON schema 或图像模板发生变化时，重放这个集合。我不确定任何单一阈值都适合每个物流工作流程；审核队列的裁定结果才是应该解决这个问题的。
按租户细分测量误 allow 和误 block。一个仓库图表工具可能需要一个不同于公共创意功能的审核阈值，但这种差异应该存在于版本化的策略配置中，而不是在未跟踪的提示词编辑中。将原始文本保留期保持在产品和合规要求允许的最短时间内，并在常规成本报告中使用指纹。
重试需要边界。遵守服务器提供的重试延迟（如果存在），限制尝试次数，并区分可重试的传输响应和策略决策。RFC 9110 是思考方法语义和重试安全性的有用基准。对于图像作业，幂等性密钥加上持久化的请求 ID 让工作进程能够恢复而不会静默重复工作；它不会使每个故障都可重试。
当你需要自定义审核状态、完整的输入覆盖以及按租户对分类器和生成费用进行解释时，使用这种架构。它与面向标准的 HTTP 边界和小型的应用程序自有 schema 配合良好。
关键在于策略所有权。当托管审核工作流、固定安全分类或特定提供商的审计界面是硬性要求时，这不适合；选择那条托管路径，当其运营保证比可移植契约更重要时。如果没有人能够标记 review 案例或维护评估集，这也很不合适。共享接口无法弥补缺席的策略治理。
上线前，用文字验证五件事：每个用户可编辑的字段都能到达闸门；只有经过验证的 allow 才能将图像加入队列；格式错误或不可用的分类器响应会关闭式地失败为 review；重试是有界的和幂等的；账本分别记录租户、策略版本、分类器使用量、图像使用量和结果。然后在 CI 中运行评估集，并对实时 review 案例进行采样。保持实现简洁。简洁更易于审计。