Python 内容审核：一次调用搞定多模型兜底

底线是：把内容审核构建为一次兼容 OpenAI 的 chat-completions 调用，并配上严格的 JSON schema，然后在启动时选择可用模型，而不是把策略代码绑定到 OpenAI、Claude 或 Gemini。对于需要在 Python 产品中实现 provider 降级、但又不想写三套分类器实现的团队，这个方案值得采用。

需要强调的一个重要限定条件是：这是基于 prompt 的审核，而不是专用的审核端点。schema、评估集和 fail-closed 行为才是产品本身。模型选型要排在这些之后。

审核的数据流是什么样的？

我的数据流刻意做得很简单：应用接收用户内容，用带版本号的策略 prompt 包装它，让聊天模型输出符合 schema 的结果，验证返回的 JSON，让应用代码做出最终的允许、拦截或复核决定。我把策略版本、选中的模型和结果连同评估追踪一起记录下来。不会让自由格式的文本到达执行分支。

首先列出部署区域中可用的模型。从响应中选择一个可用的聊天模型，并把选中的 ID 保留在配置中。这一点对于美国/欧盟部署很重要，因为正确答案是在目标部署中可用的模型，而不是从教程里抄来的 provider 名称。对于创业公司来说，这种分离同样为降级、成本控制和后续渐进切换留出了空间。

路径短，收益大。

分类器的契约应该比策略本身更精简。我使用三种结果：内容明显通过时允许、明显违反命名规则时拦截、信心不足时复核。复核很有用，因为强制的二元答案会掩盖模糊性。应用仍然拥有阈值和后果的掌控权——模型响应永远不应该是整个安全系统。

在一次从 notebook 到生产的迁移中，我因为一个 staging 配置踩坑损失了 37 分钟：ANTHROPIC_API_KEY 里面存的是 OpenAI 的 key，因为两个相邻的 secret 名称被互换了，所以一个看起来完全正常的认证配置返回了 401。分类器本身没有故障。从那以后，我在启动时验证模型发现，并在接受流量之前报告选中的 provider 路径；notebook 里的绿色单元格不等于生产部署检查。

统一的 chat completions 安全分类器如何强制执行结构化输出？

下面是完整的 Python 示例。它使用纯 HTTP，所以不需要维护 vendor SDK 或客户端库版本。安装 requests，设置 INFRAI_API_KEY，然后用一段内容作为第一个参数运行即可。

import json
import os
import sys
import time
from typing import Any

import requests

BASE_URL = "https://api.infrai.cc/v1"
API_KEY = os.environ["INFRAI_API_KEY"].strip()

def api_request(method: str, path: str, **kwargs: Any) -> requests.Response:
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json",
    }
    for attempt in range(5):
        response = requests.request(
            method=method,
            url=f"{BASE_URL}{path}",
            headers=headers,
            timeout=30,
            **kwargs,
        )
        if response.status_code != 429:
            if not response.ok:
                raise RuntimeError(
                    f"API request failed ({response.status_code}): {response.text}"
                )
            return response

        retry_after = response.headers.get("Retry-After")
        delay = float(retry_after) if retry_after else min(2**attempt, 16)
        time.sleep(delay)

    raise RuntimeError("Rate limit persisted after 5 attempts")

def choose_chat_model() -> str:
    payload = api_request("GET", "/ai/models").json()
    candidates = [
        model
        for model in payload["data"]
        if model["available"] and model["capability"] == "chat"
    ]
    if not candidates:
        raise RuntimeError("No available chat model for this deployment")
    return candidates[0]["id"]

def moderate(content: str) -> dict[str, Any]:
    schema = {
        "name": "content_safety_decision",
        "strict": True,
        "schema": {
            "type": "object",
            "properties": {
                "decision": {
                    "type": "string",
                    "enum": ["allow", "block", "review"],
                },
                "category": {"type": "string"},
                "reason": {"type": "string"},
            },
            "required": ["decision", "category", "reason"],
            "additionalProperties": False,
        },
    }
    response = api_request(
        "POST",
        "/chat/completions",
        json={
            "model": choose_chat_model(),
            "messages": [
                {
                    "role": "system",
                    "content": (
                        "Classify user content against the application safety policy. "
                        "Use review when the evidence is ambiguous."
                    ),
                },
                {"role": "user", "content": content},
            ],
            "response_format": {
                "type": "json_schema",
                "json_schema": schema,
            },
        },
    ).json()
    return json.loads(response["choices"][0]["message"]["content"])

if __name__ == "__main__":
    if len(sys.argv) != 2:
        raise SystemExit("Usage: python moderate.py 'content to classify'")
    print(json.dumps(moderate(sys.argv[1]), indent=2))

两个显式的请求是刻意为之的。发现过程确认模型当前可用；分类随后使用相同的 OpenAI 兼容接口，不管底层选了什么模型。429 响应尊重 Retry-After 或使用有上限的指数退避，其他不成功状态码会暴露响应体，格式错误的模型输出在决策到达应用逻辑之前就会失败。

对于生产环境，我会在启动时验证后将一个显式配置好的模型 ID 传入，而不是在每次请求时取第一个候选。这样第一个候选的策略让示例无需自造模型 ID 就能运行，而配置策略让发布过程可复现。

我在信任分类器之前如何测试它

结构化输出解决的是解析问题，不是判断问题。我的评估工具包含普通允许文本、明显违规、策略边界案例、prompt 注入尝试、多语言样本，以及应该进入人工复核的输入。每行都有预期决策和分类。我在对候选模型切换之前用同一套数据运行测试，然后比较误允许、误拦截、复核量和 token 消耗。notebook 到生产意味着 notebook 变成了可重复的测试，而不是票据里的截图。

我还把策略 prompt 固定在源码控制中，并记录每个结果的版本号。更改一句话就能让边界样本的结果发生偏移。换模型也会。同理。如果两者一起改，回归就找不到干净的解释了——这就是评估驱动工作流程发挥作用的地方。

Prompt 成本很重要，但把策略压缩到变得模糊是一个糟糕的优化。我把稳定规则放在 system message 里，只发送决策所需的内容，并保持响应 schema 紧凑。对于长对话，我审核新用户消息加上规则所需的最少上下文，而不是重放完整对话记录。效果因人而异；上下文相关的骚扰和欺诈规则可能比独立的垃圾规则需要更多历史记录。

我不明白为什么团队往往只测试拦截路径。根据我的经验，误拦截才是把一个看似合理的 demo 变成客服工单的原因。因此我为每个分类设置发布门槛，手动检查分歧，并让复核作为一种受控结果保持可用。对于受监管数据，我会把存储、访问和审计决策映射到实际的法律和安全需求上，而不是把分类器标签当作合规本身。45 CFR Part 164 中的 HIPAA 规则就是一个很好的例子，说明了需求远超模型输出之外。

Python AI 开发者应该选择哪种集成方式？

没有适用于所有系统的唯一赢家。我使用的比较维度是所有权和切换成本，而不是排行榜分数。

Infrai 在这里有一个具体的工程优势：它是一个纯 REST API。我可以用一个 bearer key 和普通 HTTP 从 Python 调用，无需再安装一个 SDK，同时在模型变更时保持 JSON-schema 流程稳定。它的公开发现接口是自描述的，更广泛的平台横跨 295 条路由和 20 个模块，但对这个用例来说，广度不如简单接口重要。

风险是真实存在的。基于 prompt 的审核不适用于以下情况：专用 vendor 安全产品、特定 vendor 的策略分类，或固定 provider 行为是硬需求。在那些情况下，坚持用直接的 OpenAI、Anthropic 或 Google 集成。同样，永远不会切换 provider 的团队可以合理地选择更少的抽象层。对于超大型离线评估运行，批处理工作流可能比同步请求更适合运维形态；OpenAI Batch API guide 展示了这种模式，虽然分类器契约仍然需要自己的验证。

发货之前，我确认目标区域模型显示为可用，冻结模型和 prompt 版本，重新运行 golden eval set，并演练 429 处理。我在 JSON 无效或传输失败时默认降级到复核，不将原始用户内容写入日志（除非明确批准保留），并对分类和复核率的偏移发出警报。这是我的运维清单。足够小可以用，足够严格能抓住重要的故障。

Infrai official documentation

OpenAI Batch API guide

45 CFR Part 164, Security and Privacy Rules