从 curl 到工程封装：内容审核 API 接入实战
1. 适用场景
现代互联网应用中，UGC（用户生成内容）的合规审核是刚需。无论是社交平台的帖子、即时通讯的消息、游戏内的聊天还是电商的评价，都需要自动过滤色情、政治敏感、广告、联系方式、、谩骂等不良信息。内容审核 API 提供了三重策略（敏感词库 + 正则规则 + AI 特征评分），能够识别常见的变体绕过手段（如谐音、拼音、符号替换），并输出safe/low/medium/high四个风险等级。如果你正在建设以下系统，该 API 可以显著降低开发维护复杂度：
- 社区评论系统的预审核
- 用户昵称/签名的实时校验
- 批量历史数据的合规扫描
- 直播弹幕的实时过滤（需结合 WebSocket 但文本分析仍可复用）
2. 接口能力边界
在投入工程化之前，需要明确该接口的约束：
| 维度 | 说明 |
|---|---|
| 请求方法 | POST |
| 地址 | https://v1.apizero.cn/api/content-moderation |
| QPS | 10 次/秒 (超过会被限流) |
| 单次文本长度 | 1–5000 字 (action=moderate) |
| 批量数量 | 最多 50 条 (action=batch) |
| 额外功能 | 支持可选脱敏 (mask 参数) |
| 检测类别 | 6 大类：色情、政治、广告、联系方式、、谩骂 |
接口支持三种操作：moderate（单条审核）、batch（批量审核）和categories（查询支持检测的类别列表，具体返回字段以文档为准）。日常最常用的是moderate与batch。
3. 接口鉴权
API 通过请求头传递密钥。根据素材提供的 curl 示例，使用X-API-Key头传递你的 API Key。也可以尝试Authorization头（但素材未给出示例，建议优先使用文档推荐方式）。请将 Key 存放在环境变量或配置中心，切勿硬编码。
export APIZERO_API_KEY="your_api_key_here"4. 请求参数详解
请求体为 JSON 对象，字段如下：
| 字段 | 类型 | 必填 | 说明 |
|---|---|---|---|
| action | string | 否 | 操作类型，默认moderate。可选moderate/batch/categories |
| text | string | 条件必填 | 当action=moderate时必填，1–5000 个字符 |
| texts | array[string] | 条件必填 | 当action=batch时必填，最多 50 条，每条建议不超过 5000 字（接口未明确上限，建议保守） |
| mask | boolean | 否 | 是否返回脱敏文本，默认false。设为true时，响应中会包含masked_text字段 |
注意：text和texts不能同时为非空；一次请求只能选择一种模式。
5. curl 调试示例
以下是单体审核 + 启用脱敏的完整请求：
curl -sS \ -X POST \ -H "X-API-Key: $APIZERO_API_KEY" \ -H "Content-Type: application/json" \ -d '{"action": "moderate", "text": "你真是个傻逼，脑残玩意！", "mask": true}' \ "https://v1.apizero.cn/api/content-moderation"执行后应得到类似响应：
{ "code": 0, "data": { "categories": ["谩骂"], "details": [ { "category": "谩骂", "count": 2, "matches": ["傻逼", "脑残"], "method": "敏感词" } ], "is_pass": false, "masked_text": "你真是个**，**玩意！", "original_length": 10, "risk_level": "high" }, "msg": "成功", "request_id": "abc123" }如果想批量审核，改用action=batch并传入texts数组：
curl -sS \ -X POST \ -H "X-API-Key: $APIZERO_API_KEY" \ -H "Content-Type: application/json" \ -d '{"action": "batch", "texts": ["正常文本", "你他妈的傻逼"], "mask": false}' \ "https://v1.apizero.cn/api/content-moderation"6. 代码接入：Python 工程封装
生产环境中，我们不应每次手动拼 curl，而是封装为一个可复用的客户端。以下使用requests库，并加入超时、指数退避重试、环境变量管理等常见工程化要素。
import os import time import requests from typing import Optional, List, Dict, Any class ContentModerationClient: BASE_URL = "https://v1.apizero.cn/api/content-moderation" DEFAULT_TIMEOUT = 10 # seconds MAX_RETRIES = 3 def __init__(self, api_key: Optional[str] = None): self.api_key = api_key or os.environ.get("APIZERO_API_KEY") if not self.api_key: raise ValueError("API Key is required. Set it via environment variable or constructor.") def _headers(self) -> Dict[str, str]: return { "X-API-Key": self.api_key, "Content-Type": "application/json" } def _request_with_retry(self, payload: Dict[str, Any]) -> Dict[str, Any]: for attempt in range(1, self.MAX_RETRIES + 1): try: resp = requests.post( self.BASE_URL, json=payload, headers=self._headers(), timeout=self.DEFAULT_TIMEOUT ) resp.raise_for_status() return resp.json() except (requests.ConnectionError, requests.Timeout) as e: if attempt < self.MAX_RETRIES: wait = 2 ** attempt # exponential backoff time.sleep(wait) else: raise RuntimeError(f"Request failed after {self.MAX_RETRIES} retries: {e}") except requests.HTTPError as e: # 4xx/5xx 不重试，直接抛出 try: detail = e.response.json() except: detail = e.response.text raise RuntimeError(f"HTTP {e.response.status_code}: {detail}") def moderate(self, text: str, mask: bool = False) -> Dict[str, Any]: """单条文本审核""" payload = {"action": "moderate", "text": text, "mask": mask} return self._request_with_retry(payload) def batch(self, texts: List[str], mask: bool = False) -> Dict[str, Any]: """批量文本审核（最多50条）""" if len(texts) > 50: raise ValueError("Batch size limited to 50 texts.") payload = {"action": "batch", "texts": texts, "mask": mask} return self._request_with_retry(payload) # 使用示例 if __name__ == "__main__": client = ContentModerationClient() result = client.moderate("你真是个傻逼，脑残玩意！", mask=True) print(result["data"]["masked_text"]) # 输出脱敏文本7. 返回字段解读
成功时 HTTP 状态码 200，响应体 JSON 包含：
| 字段 | 类型 | 说明 |
|---|---|---|
| code | int | 业务状态码，0 为成功，非 0 请参考文档或 msg 字段 |
| msg | string | 状态描述 |
| request_id | string | 请求唯一标识，便于排查问题 |
| data | object | 审核结果数据 |
data对象包含：
| 字段 | 类型 | 说明 |
|---|---|---|
| is_pass | boolean | true表示审核通过（无风险），false表示存在敏感内容 |
| risk_level | string | 风险等级：safe/low/medium/high |
| categories | array[string] | 命中的敏感类别列表，如["谩骂", "广告"] |
| original_length | int | 原始文本字符数 |
| masked_text | string | 当请求mask=true时返回，已脱敏的文本（敏感词替换为**） |
| details | array[object] | 详细命中信息，每个元素包含category(类别)、count(命中次数)、matches(命中词数组)、method(检测方式，如"敏感词"、"正则"、"AI") |
对于批量审核batch模式，返回结构可能略有不同（素材未提供完整示例，建议以实际返回为准，通常data可能包含一个列表）。
8. 常见错误与处理
开发中可能遇到的错误及建议处理方式：
| 现象 | 可能原因 | 处理建议 |
|---|---|---|
| 400 Bad Request | JSON 格式错误、缺少必填字段、text 为空或超长 | 检查请求体格式，确保text长度符合要求 |
| 401 Unauthorized | API Key 无效或缺失 | 检查请求头 Key 是否配置正确 |
| 429 Too Many Requests | 超出 QPS 限制 (10/s) | 添加本地限流或请求队列，等待后重试 |
| 502/503 | 服务端临时故障 | 启用重试机制，注意指数退避 |
| code ≠ 0 | 业务逻辑错误 | 根据msg和request_id查阅文档或联系支持 |
注意：素材未提供具体业务错误码列表，实际集成时应参考官方文档获取完整错误码表。
9. 工程化注意事项
9.1 API Key 安全
- 通过环境变量或密钥管理服务（如 Vault）注入，不要硬编码。
- 前端代码不要直接暴露 Key，应由后端代理请求。
9.2 限流控制
虽然 QPS 为 10，但实际调用中应主动控制并发数。可以使用asyncio.Semaphore或threading.Semaphore限制同时进行中的请求数量。
9.3 批量优化
- 对单条文本较短且数量巨大时，使用
batch模式可显著降低网络开销。 - 注意
texts数组长度不超过 50，若超过需拆分。
9.4 脱敏策略
mask=true返回的脱敏文本可用于展示，但注意敏感词被替换为固定长度**，可能影响原文排版。如需更精细控制，可结合返回的details字段自行实现脱敏。
9.5 监控与日志
- 记录每次请求的
request_id、risk_level、耗时，便于后期审计和问题排查。 - 对审核不通过的内容可落库标记，供人工复审。
9.6 超时设置
网络请求务必设置超时（建议 5–10 秒），避免线程长时间阻塞。重试时应采用指数退避，避免加重服务端压力。
10. 参考文档
- 内容审核 API 文档
- 原始文档 Markdown