从 curl 到工程封装:内容审核 API 接入实战
1. 适用场景现代互联网应用中UGC用户生成内容的合规审核是刚需。无论是社交平台的帖子、即时通讯的消息、游戏内的聊天还是电商的评价都需要自动过滤色情、政治敏感、广告、联系方式、、谩骂等不良信息。内容审核 API 提供了三重策略敏感词库 正则规则 AI 特征评分能够识别常见的变体绕过手段如谐音、拼音、符号替换并输出safe/low/medium/high四个风险等级。如果你正在建设以下系统该 API 可以显著降低开发维护复杂度社区评论系统的预审核用户昵称/签名的实时校验批量历史数据的合规扫描直播弹幕的实时过滤需结合 WebSocket 但文本分析仍可复用2. 接口能力边界在投入工程化之前需要明确该接口的约束维度说明请求方法POST地址https://v1.apizero.cn/api/content-moderationQPS10 次/秒 (超过会被限流)单次文本长度1–5000 字 (actionmoderate)批量数量最多 50 条 (actionbatch)额外功能支持可选脱敏 (mask 参数)检测类别6 大类色情、政治、广告、联系方式、、谩骂接口支持三种操作moderate单条审核、batch批量审核和categories查询支持检测的类别列表具体返回字段以文档为准。日常最常用的是moderate与batch。3. 接口鉴权API 通过请求头传递密钥。根据素材提供的 curl 示例使用X-API-Key头传递你的 API Key。也可以尝试Authorization头但素材未给出示例建议优先使用文档推荐方式。请将 Key 存放在环境变量或配置中心切勿硬编码。export APIZERO_API_KEYyour_api_key_here4. 请求参数详解请求体为 JSON 对象字段如下字段类型必填说明actionstring否操作类型默认moderate。可选moderate/batch/categoriestextstring条件必填当actionmoderate时必填1–5000 个字符textsarray[string]条件必填当actionbatch时必填最多 50 条每条建议不超过 5000 字接口未明确上限建议保守maskboolean否是否返回脱敏文本默认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 }如果想批量审核改用actionbatch并传入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-moderation6. 代码接入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, jsonpayload, headersself._headers(), timeoutself.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(fRequest 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(fHTTP {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(你真是个傻逼脑残玩意, maskTrue) print(result[data][masked_text]) # 输出脱敏文本7. 返回字段解读成功时 HTTP 状态码 200响应体 JSON 包含字段类型说明codeint业务状态码0 为成功非 0 请参考文档或 msg 字段msgstring状态描述request_idstring请求唯一标识便于排查问题dataobject审核结果数据data对象包含字段类型说明is_passbooleantrue表示审核通过无风险false表示存在敏感内容risk_levelstring风险等级safe/low/medium/highcategoriesarray[string]命中的敏感类别列表如[谩骂, 广告]original_lengthint原始文本字符数masked_textstring当请求masktrue时返回已脱敏的文本敏感词替换为**detailsarray[object]详细命中信息每个元素包含category(类别)、count(命中次数)、matches(命中词数组)、method(检测方式如敏感词、正则、AI)对于批量审核batch模式返回结构可能略有不同素材未提供完整示例建议以实际返回为准通常data可能包含一个列表。8. 常见错误与处理开发中可能遇到的错误及建议处理方式现象可能原因处理建议400 Bad RequestJSON 格式错误、缺少必填字段、text 为空或超长检查请求体格式确保text长度符合要求401 UnauthorizedAPI 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 脱敏策略masktrue返回的脱敏文本可用于展示但注意敏感词被替换为固定长度**可能影响原文排版。如需更精细控制可结合返回的details字段自行实现脱敏。9.5 监控与日志记录每次请求的request_id、risk_level、耗时便于后期审计和问题排查。对审核不通过的内容可落库标记供人工复审。9.6 超时设置网络请求务必设置超时建议 5–10 秒避免线程长时间阻塞。重试时应采用指数退避避免加重服务端压力。10. 参考文档内容审核 API 文档原始文档 Markdown