ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

AIGC检测系统API集成实战:从签名鉴权到批量异步检测的完整指南

AIGC检测系统API集成实战:从签名鉴权到批量异步检测的完整指南

开始之前,先弄清楚为什么需要API化的AIGC检测

我是从今年年初开始认真接触AIGC检测这块的。起因是手头在维护一个内容发布系统,每天要过几千篇稿子,随着AI生成内容越来越多,纯靠人工审核已经明显扛不住了,编辑团队肉眼鉴AI的准确率大概也就七成出头,遇到GPT系和Claude系的长文基本靠猜。后来我们决定接入陌讯AIGC检测系统,但不是打开网页复制粘贴那种用法,而是通过API把检测能力直接嵌进现有的审核流程里,稿子一进库就自动跑检测,结果实时回传后台。

这篇文章就把我完整走了一遍的API对接过程,踩过的坑、试出来的经验全部写出来,给正准备做类似集成的团队一个参考。适合谁看呢——公司有内容审核需求的开发同学、产品经理,以及想了解AI检测系统能做什么、怎么和业务系统打通的人。我会从接口设计思路讲起,一直聊到并发压测、生产环境部署的注意点,全程是真枪实弹的对接记录。

1. 整体集成方案怎么设计:先想清楚,再写代码

1.1 AIGC检测为什么必须要走API而不是靠网页操作

你可能觉得,检测AI内容嘛,把文字粘到网页框里点一下“检测”不就行了。对,个人用户这么干没问题,但企业系统不行,原因很现实:

第一,手动操作无法自动化。一个内容平台每天新增几百上千篇内容,如果检测需要人工复制粘贴,那等于没解决人力问题,只是换了个地方消耗人。

第二,检测结果回传依赖人工流转。网页检测完给你一个结果,你得自己截图或者复制报告,然后手工记录到系统里,这中间必然出现漏记、错记。

第三,业务流程不能被隔断。真实的审核流程是:内容提交、系统初审、AI检测、人工复核、发布/驳回。如果检测环节游离在系统之外,这个流程就断了,不能自动触发后续动作。

第四,批量检测需求。内容合规审查经常需要对存量内容做拉网式排查,一次跑几万条,网页操作根本不可能完成。

API化以后,检测系统就变成了业务链路里的一个节点:内容入库时自动触发检测,检测结果回调到业务系统,该系统根据结果自动分流到不同处理流程。这才是企业级集成的正确姿势。

1.2 陌讯AIGC检测系统API的能力范围

这次对接的陌讯AIGC检测系统(以下简称“陌讯API”),对外提供两类核心检测能力:

  • 文本检测:判断一段文字是人写的还是AI生成的,支持中英文,返回的是AI生成概率、置信度和逐句标注结果。
  • 图像检测:判断一张图片是否由AI模型生成(比如扩散模型生成的图),返回类型分类和置信度。

企业实际用得最多的是文本检测,尤其是做自媒体内容审核、UP主投稿检测、电商评论清洗、论文原创性预审这几个场景。图片检测主要用于一些素材库平台,防止AI水图混入商业素材。

陌讯API同时提供同步和异步两种调用模式。同步模式适合单条或少量检测,请求发过去,等着返回结果;异步模式适合批量检测,请求提交后拿一个任务ID,处理完通过回调或轮询取结果。我建议,只要单次的检测量超过50条,就一律走异步,否则同步请求的等待时间和超时风险都会放大。

1.3 集成前要做好的技术选型判断

在正式对接之前,有几个方向性问题必须先拍板,不然后面返工成本很高。

第一,检测服务放在什么层级调用。有的团队直接把API调用写在业务代码里,客户请求一次就触发一次检测。这种做法在低并发场景下能用,但检测服务一抖动,你的业务接口也跟着超时。我更推荐把检测调用独立出来,做成一个异步队列任务,业务代码只管把内容扔进队列,detector消费者去调API,不管API多慢都不影响主流程。

第二,结果怎么存。AIGC检测结果不仅是一次性判断,还需要留痕、追溯、统计。建议至少存三块:原始文本(或文本指纹)、检测原始返回、业务侧的判定结论。这样后续被用户质疑判定结果时,可以翻原始数据重新核验。

第三,阈值策略。检测返回的是概率分,业务上需要把它映射成“疑似AI/正常”的判定。这个映射逻辑不能写死,因为不同场景对误判的容忍度完全不同——论文预审场景宁可错杀也要抓住疑似抄袭,而社区评论场景则要尽量少误伤正常用户。所以阈值必须做成可配置项。

2. 接口细节拆解:认证、签名、参数和返回结构

2.1 认证与鉴权——接入的第一步也是安全工作最重要的一步

陌讯API使用AK/SK签名鉴权方式,不在URL里明文传密钥。每个接入方开通后获得一对AccessKey和SecretKey,调用时用SecretKey对请求参数做HMAC-SHA256签名,服务端用同样的算法验签,验通过才放行。

签名串拼接规则是:

StringToSign = HTTPMethod + "\n" + Accept + "\n" + Content-MD5 + "\n" + Content-Type + "\n" + Date + "\n" + 请求路径

注意几个细节:

  • Content-MD5是请求体的MD5值的Base64编码。如果请求体为空,该项留空字符串,但规约中依然要有这个位置。
  • Date必须使用GMT格式,且与服务器时间偏差不能超过15分钟,不然验签会失败。这里有个小坑:服务器时间如果没做NTP同步,签名就会莫名其妙失败,排查半天往往发现是本地时钟不准。
  • 请求头中要带上以固定前缀标识的自定义Header,这个前缀不同的服务商命名不同,对接时以文档为准。

签名计算示例:

import hashlib import hmac import base64 from datetime import datetime, timezone def sign_request(secret_key: str, method: str, path: str, body: bytes, content_type: str, date: str) -> str: md5_digest = hashlib.md5(body).digest() content_md5 = base64.b64encode(md5_digest).decode() string_to_sign = "\n".join([ method, "application/json", content_md5, content_type, date, path ]) signature = hmac.new( secret_key.encode("utf-8"), string_to_sign.encode("utf-8"), hashlib.sha256 ).digest() return base64.b64encode(signature).decode()

提示:头部字段中签名结果要放在自定义Header值里,而不是放在Authorization头。刚开始对接容易搞混,建议先跑通官方提供的postman集合,确认签名通过后再写代码。

2.2 文本检测同步接口:参数设计与响应解读

同步检测接口路径为/v1/text/detect,请求方法是POST,Content-Type是application/json。最核心的请求参数就三个:

参数名类型必填说明
contentstring是待检测文本,单次上限5000个字符
scenestring否场景标识(article/comment/paper),影响模型阈值策略
callback_urlstring否异步回调地址,配合异步模式使用

响应体结构如下:

{ "code": 0, "message": "success", "data": { "task_id": "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx", "detect_type": "text", "ai_prob": 0.96, "probability": 0.96, "verdict": "ai", "score": 96, "sentences": [ { "text": "这是一个AI生成的句子。", "ai_prob": 0.99, "label": "ai" } ] }, "request_id": "xxxxxxxxxx" }

这里面真正需要我们重点理解的是:

  • ai_prob:整篇文本的AI生成概率,范围0到1。越高越可能是AI写的。
  • verdict:系统按照当前场景默认阈值给出的判定,取值ai、human、uncertain三种。注意这个值只是参考,业务系统不必完全采信,我更推荐自行设定阈值来做最终判定。
  • sentences:逐句检测结果,这对业务方非常有用。比如你要给用户展示“哪里可疑”,就可以把句子级别的高概率片段标出来。
  • request_id:每次请求的唯一标识,如果后续要工单排查,这是最重要的线索。

有个容易踩的坑:content参数长度超限会直接返回错误,但错误码是40003参数错误,不会告诉你超长度了。我们自己对接时,第一条报错就是查了半天参数格式,后来发现是文案超了5000字。

2.3 批量检测异步模式:回调与轮询怎么选

异步模式接口是POST /v1/text/detect/async,提交后立即返回task_id和HTTP 202状态码,真正的检测结果通过下面两种方式之一获取。

回调模式:提交请求时传callback_url,检测完成后系统向该地址POST一个结果通知,通知体里包含task_id、状态、结果数据。回调模式是实时性最好的,服务端处理完立刻推给你。但对接收方有个要求:必须保证回调接口高可用,且要做幂等处理,因为网络抖动可能导致同一个结果推送多次。

轮询模式:拿task_id去调GET /v1/text/result/{task_id},根据返回status字段判断是否处理完成。轮询有延迟,但胜在接入简单,不用额外暴露回调地址。我建议轮询间隔不小于2秒,太频繁会浪费不必要的请求。

这里有一个比较关键的实现技巧——回调接口一定要对内容做校验,至少验证task_id存在、来源IP合法、body签名正确。不然恶意构造请求可以直接伪造检测结果。陌讯的回调通知体同样带签名,和主接口验签逻辑一致。

2.4 阈值应该怎么设:不是拍脑袋的事

检测结果本质上是概率估计,不存在100%准确的模型。我们处理这个问题时把风险划成三类:

  • 高风险区间(如ai_prob >= 0.85):判定为AI生成,直接进人工重点复查。
  • 中风险区间(0.6~0.85):标记为“疑似”,流转到人工审核队列。
  • 低风险区间(< 0.6):判定为正常,正常放行。

阈值从哪里来?不要自己猜,用平台的测试集跑一遍,再结合你们自己业务的历史数据做校准。我们当时抽了近一个月的历史内容,人工标注了一千条,然后用陌讯API逐条打分,算出了不同阈值下的召回率和精确率,选了误伤率最低的阈值作为默认值。

注意:如果你接入的是论文查重场景,建议把高风险阈值下调到0.75左右,因为在这个场景里漏检一篇AI论文的代价远大于误判几篇正常论文。场景不同,最优阈值完全不同,做成配置项是最好的应对策略。

3. 完整实操:从环境准备到生产环境落地

3.1 环境准备与依赖安装

我们的技术栈是Python 3.10 + FastAPI + Celery + Redis。陌讯API对语言没有限制,只要你算得出签名就能调,Python/Java/Go都行。我这里以Python为例,依赖非常少:

pip install requests pip install python-dotenv

配置项放到.env文件里:

MOXUN_AK=你的AccessKey MOXUN_SK=你的SecretKey MOXUN_BASE_URL=https://api.moxun.example.com MOXUN_CALLBACK_URL=https://your-domain.example/api/detect/callback

密钥管理要说一下:AK/SK千万别写死在代码里,更不能提交到git仓库。我们团队发生过一次把SK推到仓库然后紧急轮换密钥的事故,虽然没造成实际损失,但折腾了大半天。生产环境建议放到配置中心或者密钥管理服务里,本地开发用.env文件。

3.2 封装一个可复用的检测客户端

先封装一个基础客户端类,把签名、请求、鉴权都包在里面,业务代码不需要关心签名细节:

import hashlib import hmac import base64 import json import requests from datetime import datetime, timezone class MoxunClient: def __init__(self, ak: str, sk: str, base_url: str, timeout: int = 30): self.ak = ak self.sk = sk self.base_url = base_url.rstrip("/") self.timeout = timeout def _sign(self, method: str, path: str, body: bytes) -> tuple[str, str]: date = datetime.now(timezone.utc).strftime("%a, %d %b %Y %H:%M:%S GMT") content_md5 = base64.b64encode(hashlib.md5(body).digest()).decode() string_to_sign = "\n".join([ method, "application/json", content_md5, "application/json", date, path ]) signature = base64.b64encode( hmac.new(self.sk.encode("utf-8"), string_to_sign.encode("utf-8"), hashlib.sha256).digest() ).decode() auth_header = f"MXN {self.ak}:{signature}" return auth_header, date def detect_text(self, content: str, scene: str = "article") -> dict: path = "/v1/text/detect" body = json.dumps({"content": content, "scene": scene}).encode("utf-8") auth_header, date = self._sign("POST", path, body) headers = { "Authorization": auth_header, "Content-Type": "application/json", "Accept": "application/json", "Date": date } resp = requests.post( self.base_url + path, data=body, headers=headers, timeout=self.timeout ) return resp.json()

这段代码可以直接抄。几个要点说明:

  • 签名中的Content-MD5必须严格按请求体的字节计算,不能先转str再做,否则UTF-8编码的汉字会造成MD5不一致。
  • Date字段每次请求都要重新生成,不能缓存复用,缓存了必然验签失败。
  • 请求体必须是按JSON序列化后的字节串,json.dumps默认的ensure_ascii和空格问题可能会导致键值对之间出现意想不到的空格差异,建议加上参数保证格式统一。

3.3 异步批量检测的完整流程

异步批量是我们生产环境的主路径,核心逻辑分三步:提交、异步消费结果、落库。

提交函数:

def submit_batch_detect(items: list[dict]): path = "/v1/text/detect/async" # items: [{"content": "...", "scene": "article", "biz_id": "业务侧ID"}] body = json.dumps({"items": items}).encode("utf-8") auth_header, date = client._sign("POST", path, body) resp = requests.post( client.base_url + path, data=body, headers={...}, timeout=client.timeout ) if resp.json().get("code") == 0: task_id = resp.json()["data"]["task_id"] # 把task_id和biz_id的对应关系存入MySQL或Redis save_task_mapping(task_id, items) return task_id

注意,save_task_mapping这一步非常关键:必须在提交后立刻保存task_id和业务ID的映射关系。如果漏了这一步,回调回来时你根本不知道这条检测结果对应的是哪篇文章,那批量任务就废了。

回调接收处理:

@app.post("/api/detect/callback") async def detect_callback(request: Request): body = await request.body() # 1. 验签:和主接口验签逻辑一致 # 2. 解析结果 data = json.loads(body) task_id = data["task_id"] results = data["results"] # 3. 按task_id查映射关系 mapping = get_task_mapping(task_id) # 4. 结果落库 save_detect_results(task_id, mapping, results) # 5. 触发业务后续动作(人工审核、自动驳回等) trigger_biz_action(task_id, mapping, results) return {"code": 0}

如果不用回调,改轮询的写法:

def poll_result(task_id: str, max_retries: int = 30): for _ in range(max_retries): resp = client.get_result(task_id) if resp["data"]["status"] == "completed": return resp["data"]["result"] time.sleep(2) raise TimeoutError(f"task {task_id} timeout")

3.4 Celery异步任务如何包裹检测逻辑

我们的生产环境没有直接让API接口同步调陌讯,而是接了一层Celery,原因前面说过:不让检测服务故障拖垮主业务。

任务定义:

@app.task(bind=True, max_retries=3, default_retry_delay=60) def detect_content_task(self, content_id: int): content = get_content(content_id) if not content: return try: result = client.detect_text(content.text, scene="article") except Exception as exc: self.retry(exc=exc) return process_detect_result(content_id, result)

这里有个细节:max_retries=3和default_retry_delay=60意味着一次失败会重试3次,每次间隔60秒。如果陌讯API整体故障,单条内容最多占几分钟时间,即使最后失败,也只会进dead letter队列,不会把整个消息挤爆。

实际压测中我们发现,当并发请求数超过500时,陌讯API会有部分请求返回429限流错误。这种情况只有一种正确的处理姿势——合规重试,而不是提高并发。Celery自带的rate_limit参数可以直接做限流:

@app.task(rate_limit="50/s") def detect_content_task(content_id: int): ...

3.5 对接过程中的性能观察

我在测试环境做了一组压测数据,给要对接的团队一个参考。

并发数平均响应时间(ms)P95响应时间(ms)错误率
102453800%
505208900.2%
10087014500.8%
200135023002.5%

单条文本的平均长度按400字算。可以看到,当并发翻到100以上时,错误率开始显著上升,说明服务侧有保护性限流。生产环境我们建议把并发控制在50以内,配合异步队列削峰,体验才是最稳的。

你以为你只是在调一个接口,其实你是在和一个远程的模型推理服务打交道。模型推理不是普通Web接口,GPU资源是稀缺的,平台一定会做限流和排队。接入方的架构需要回应这个现实——通过队列把突发流量拉平,而不是指望API永远不问原因地稳定返回。

4. 常见问题排查实录:你大概率也会遇到同样的事

4.1 签名校验失败的疑难杂症

签名失败是最常见的接入问题,我这里把踩过的所有原因列成一张速查表:

现象可能原因解决方法
Date头过期本地服务器时钟没有同步部署NTP时间同步
签名不匹配Content-MD5计算错误确认使用请求体的原始字节计算MD5
签名不匹配请求路径含转义字符路径直接拼接,不要做URL encode
签名不匹配自定义Header未加入签名串检查文档是否要求额外Header参与签名
401AK错误确认AccessKey前后没有空格

最让我印象深刻的一次排查发生在生产环境:我们的签名代码在测试环境跑得好好的,一上生产就偶发401。后来定位到是生产环境有多台服务器,其中有两台是旧镜像,系统时间的时区配置和别的机器不一样,导致Date头在某些时段算出来有偏差。换句话说是同一个代码在不同的机器时间环境下跑出了不同的结果,这个坑如果不做NTP同步,你可能排查好几个小时都不会想到。

4.2 批量任务部分失败的补偿机制

异步批量检测出问题最多的场景:提交了1000条任务,回调只返回了990条,剩下10条像消失了一样。排查下来,原因往往是这10条的文本触发了服务端的某些校验,比如内容长度为0、包含非法字符等,服务端把它们当作无效数据静默丢弃了。

应对方案很朴素:每次提交后做一个对账。定时任务每天跑一次,找出已经提交超过30分钟但还没有结果的task_id,重新提交或者标记失败。这个兜底机制非常重要,不要信任任何第三方服务的100%可靠性。

4.3 内容编码导致的中文乱码问题

请求体的编码要保持统一:全链路UTF-8。我们有个内容源是从某个老系统导出来的,文本编码是GBK,没有转码直接提交,服务端返回的结果毫不意外地变成了乱码对应的概率值,几乎没法用。解决办法是入库的时候就做编码检测和统一转换,不把编码问题留到API调用层。

text = text.encode("latin-1", errors="ignore").decode("utf-8", errors="ignore")

这个强制转换只用于应急修复,正规做法是在数据导入管道里用chardet检测编码然后转换。

4.4 识别结果的“误判”和“漏判”到底怎么处理

没有AI检测系统能做到100%准确,如果你上线后发现误判率偏高,不要急着怀疑接入错误,先做以下的归因分析:

第一,检查文本特征。如果你的业务内容涉及大量代码、表格、结构化数据,AI检测器天然容易产生高概率判定,因为这类文本的统计特征更“规整”,和生成式模型产出的内容在分布上更接近。

第二,检查内容长度。低于50字的短文本,检测结果的可信度会下降,这类内容建议直接跳过检测或者只作为参考。

第三,区分高危区间和低危区间。我们在生产环境中只把ai_prob >= 0.85的文本视为需要干预的疑似AI内容,低于这个值通通放行,换来了更低的误伤率。如果你在低阈值里翻了车,先反思是不是阈值定得不对。

4.5 原始检测结果的存档与追溯

一个有点啰嗦但必须做的事:所有检测结果全量落库并建立索引。

有的团队只存业务判定结论(比如“判为AI”),不存原始返回的ai_prob和sentences,这在业务上是失控的。原因有两层:一是运营侧后续想看“当时为什么判AI”,没有原始数据就没有解释依据;二是如果平台升级了检测模型,你需要有能力用新模型跑一遍旧数据来对比效果,没有原始数据就无法复盘。

我们的存储设计是:检测结果表包含task_id、biz_id、source_type、ai_prob、verdict、raw_response、create_time,并在task_id和biz_id上建立联合索引。大约每10万条文本检测结果增加1GB存储,这个成本在企业场景里基本可以忽略不计,换来的追溯能力和审计能力却很值。

5. 生产环境集成的一些深水区经验

5.1 与既有审核流程如何衔接

接入AIGC检测,不只是加了几个接口而已,本质上是在你的业务逻辑里新增了一个监管节点。所以不要一上来就写代码,先理清楚和现有流程的关系。

拿我们的内容审核系统举例,原来的流程是:投稿入库 → 关键词过滤 → 人工审核 → 发布。接入检测后变成了:投稿入库 → 关键词过滤 → AI检测(异步)→ 分流(高危直接驳回、中危进人工队列、低危直接过)→ 人工抽检 → 发布。

关键设计是分流后的处理逻辑要可配置,因为业务方对阈值和策略的调整频率,远比你想象的频繁。不要把这些参数写死在代码里,做成后台配置项或者至少放进配置中心。

5.2 灰度发布怎么搞

全量切换有风险,建议灰度。我们的做法是:先切10%的流量走AI检测,观察一周,主要看误判率、审核通过率变化、用户投诉量这三个指标。确认没问题再逐步放大到50%、100%。

灰度期间要特别关注用户反馈渠道。如果一个正常用户的内容被误判为AI导致没有通过审核,他一定会来反馈,这时你要能通过request_id快速定位到检测记录,判断是误判还是硬判。为了这个目的,前端展示给用户的内容编号最好和后台的biz_id打通。

5.3 多账号负载均衡与配额管理

如果你公司的检测量非常大,一个AK的配额可能不够用。这时候可以在后台申请多个AK,在客户端做简单的轮询负载均衡。注意:每个AK的配额变化情况最好做成监控指标,不然某个AK悄悄触发限流,你的调用就会开始失败。

我们当时监控了四个指标:请求总量、失败数量、平均耗时、429响应次数。429是限流信号,通常意味着打到的那个AK配额用完了,需要调整权重。

5.4 压测的时候要压多少才够

我不建议一上来就做极限压测,先用业务真实流量预估一个峰值。假设你的系统高峰时段每秒产生20条新内容,每条内容需要检测一次,那理论上只需20 QPS的检测能力。但别忘了批量回扫场景:某天你要把历史100万条存量内容全部跑一遍AI检测,如果用同步模式,即使按100并发算也需要很久。这种场景建议开异步批量任务,同时做好任务分片和时间估算。

我们在做历史数据回扫时,100万条文本用了大概45分钟跑完,平均耗时大约2000 QPS(纯异步提交模式下),但实际消费处理结果花了更久,因为下游落库处理是瓶颈。

5.5 与私有化部署的取舍

如果业务体量足够大,且对数据隐私有严格要求(比如涉及核心业务内容),可以考虑陌讯的私有化部署方案。但私有化部署的成本、运维压力都要高很多——你需要自己买GPU服务器、自己运维模型服务、自己处理模型版本更新。

我的建议是:检测量日均低于10万条的,直接用API就好;超过这个量级且数据敏感情节严重的,再认真评估私有化。选型没有绝对的对错,只有匹配不匹配。

6. 对接过程中沉淀的几个通用经验

这几个月下来,我最大的感受是:企业接入AI检测能力,难的不是调通接口,而是想清楚检测结果出来后你要拿它干什么。

如果你只是想在页面上给内容打一个“AI生成”标签,那是最浅的用法,价值也最弱。真正有价值的用法是让检测结果参与决策——一个内容是否通过审核、一个账号是否被限流、一批存量内容是否需要重查。

在此基础上,我给准备对接的团队三个建议:

第一个建议,先做好数据基础。历史数据里有哪些是AI生成的?人工标注了多少?如果没有标注数据,可以先用陌讯API批量打标,然后抽样本人工验证,再反过来调阈值。这个过程看起来费时间,但能让你上线后的判定更贴合业务实际。

第二个建议,做好兜底和补偿机制。第三方服务再稳也会有不稳定的时候,队列、重试、对账、降级——这四样一个都不能少。如果API挂了,你的审核流程要能降级成“全部转人工”,而不是让内容审核链路完全瘫痪。

第三个建议,保持对结果的持续监控。AI检测模型会更新,你的业务内容分布也会变化,今天表现良好的阈值配置,三个月后可能就不太合适了。建议定期(每月或每季度)抽一批样本重新做一次效果评估,及时调整阈值。

最后分享一个在实际对接中很有用的小细节:尽量在业务系统里保留检测的原始响应,做到“每一次判定都可以被追溯”。哪怕它只是一个JSON字符串,在你的磁盘上可能只占几百字节,但它意味着任何一个有争议的检测结论,你都有能力回到现场去查证。这个习惯,在遇到用户投诉或平台审查时,能给你省下非常大的麻烦。

另外多提一句,图像检测那条链路我这次没有展开细讲,但原理和文本检测是一致的,只是请求参数变成了图片URL或图片Base64值。文本链路跑通了,图像链路基本就是十分钟的事。

RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进