从 curl 到工程封装：内容审核 API 接入实战

1. 适用场景

现代互联网应用中，UGC（用户生成内容）的合规审核是刚需。无论是社交平台的帖子、即时通讯的消息、游戏内的聊天还是电商的评价，都需要自动过滤色情、政治敏感、广告、联系方式、、谩骂等不良信息。内容审核 API 提供了三重策略（敏感词库 + 正则规则 + AI 特征评分），能够识别常见的变体绕过手段（如谐音、拼音、符号替换），并输出safe/low/medium/high四个风险等级。如果你正在建设以下系统，该 API 可以显著降低开发维护复杂度：

• 社区评论系统的预审核
• 用户昵称/签名的实时校验
• 批量历史数据的合规扫描
• 直播弹幕的实时过滤（需结合 WebSocket 但文本分析仍可复用）

2. 接口能力边界

在投入工程化之前，需要明确该接口的约束：

接口支持三种操作：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 对象，字段如下：

注意：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 包含：

data对象包含：

对于批量审核batch模式，返回结构可能略有不同（素材未提供完整示例，建议以实际返回为准，通常data可能包含一个列表）。

8. 常见错误与处理

开发中可能遇到的错误及建议处理方式：

注意：素材未提供具体业务错误码列表，实际集成时应参考官方文档获取完整错误码表。

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