基于环境自适应架构的降低AIGC检测率系统

基于环境自适应架构的降低AIGC检测率系统——及其背后工程设计一套代码两个灵魂。Linux 服务器跑 DockerWindows 双击 EXE后端逻辑零修改。项目地址https://github.com/math89423-star/AI-Academic-PolisherLicense: MITDisclaimer: 本工具仅供辅助学术写作与语言润色使用旨在帮助作者提升论文的表达质量与可读性。使用者应确保最终提交的学术成果符合所在机构的学术诚信规范本工具不应被用于规避学术诚信审查。写在前面寒假帮学弟改毕业论文他用千问辅助写的初稿被维普的 AIGC 检测标记了 80% 以上的 AI 率。问我有没有什么办法看了一圈市面上的工具大多是套壳收费效果参差不齐。后来我自己折腾了一套提示词方案先在本地跑通了又顺手做了 Web 端方便几个朋友用。再后来想着干脆做个桌面版让不懂技术的人也能直接用。这一折腾就是几个月索性整理成了一个完整的开源项目AI Academic Polisher。这篇文章不是 README 的翻版。我想聊聊背后的几个技术决策——为什么用 RQ 不用 Celery、为什么 Desktop 模式要自己造一个 MemoryRedis、SSE 和 WebSocket 怎么选、长文档怎么并发切片还不丢顺序。希望能给做类似项目一些参考。一、润色效果验证工具好不好用数据说话。以下是部分测试数据用gemini-3.1-pro-preview模型润色后提交到主流文本检测平台的评估结果2026 年 4 月测试检测平台润色前 AIGC 识别率润色后 AIGC 识别率PaperPass75.24%0.41%维普42.79%3.34%朱雀 AI英文100%0%朱雀 AI中文100%0%PaperPass检测维普检测朱雀检测需要说明的是这不是所谓的降重工具。它做的事情是把 AI 生成的、带有明显机器味的句子改写成更符合人类学术写作习惯的表达。原意保留AIGC 识别率的下降是润色质量的自然体现。具体的检测截图都在仓库的docs/目录下。二、为什么要做双模式架构核心思路用工厂模式在启动时决定基础设施上层代码完全无感知。痛点最早只做了 Web 版部署在自己的服务器上给几个朋友用。问题很快暴露出来使用门槛高朋友里有非技术人员让他们 SSH 进服务器看日志不太现实隐私顾虑有人不放心把论文上传到别人的服务器资源冲突大家共用一个 API Key一个人密集调用就把别人的额度用光了最直接的方案是做一个 Windows 桌面版但又不想维护两份代码。于是就有了一套代码两个灵魂的双模式架构。模式自动检测项目通过DEPLOY_MODE环境变量控制运行模式支持三个值server、desktop、auto。默认是auto会根据操作系统自动判断# config.pydef_resolve_deploy_mode():modeos.environ.get(DEPLOY_MODE,auto)ifmodeauto:returndesktopifplatform.system()WindowselseserverreturnmodeWindows 上双击 EXE 自动进入 Desktop 模式Linux 服务器上 Docker 启动自动进入 Server 模式不需要手动配置。工厂模式切换基础设施# extensions.py 简化版ifDEPLOY_MODEserver:redis_clientredis.Redis(host...,port...)task_queuerq.Queue(ai_tasks,connectionredis_client)else:# desktopredis_clientMemoryRedis()# 内存字典 threading.Locktask_queueMemoryQueue()# queue.Queue 守护线程上层代码完全不知道自己跑在哪种模式下。Processor 调用redis_client.publish()推进度Server 模式下走真 RedisDesktop 模式下走内存 Pub/Sub接口签名一模一样。MemoryRedis在内存里造一个 Redis这是项目里我个人比较满意的一块设计。需求很明确实现 redis-py 的方法子集让上层代码零感知切换。classMemoryRedis:def__init__(self):self._kv{}# 普通 KVself._hashdefaultdict(dict)# Hashself._setdefaultdict(set)# Setself._channelsdefaultdict(list)# Pub/Sub 订阅者self._lockthreading.Lock()# 全局锁defpublish(self,channel,message):withself._lock:forqinself._channels[channel]:q.put(message)defpubsub(self):returnMemoryPubSub(self)# 返回兼容 redis-py 的 pubsub 对象Pub/Sub 部分用queue.Queue模拟阻塞订阅——每个订阅者持有一个 Queue发布者往所有订阅者的 Queue 里塞消息。配合get_message(timeout...)接口和 redis-py 完全一致SSE 推送那一层代码不用改一行。MemoryQueue用守护线程替代 RQ WorkerServer 模式下 RQ Worker 是一个独立进程通过 Redis 拿任务。Desktop 模式下没有独立进程的概念但又不能阻塞主线程所以用了守护线程classMemoryQueue:def__init__(self):self._qqueue.Queue()self._appNonedefenqueue(self,func,*args):self._q.put((func,args))def_worker_loop(self):whileTrue:func,argsself._q.get()withself._app.app_context():# 关键手动注入 app contexttry:func(*args)exceptException:logger.exception(Task failed)defstart_worker(self):threading.Thread(targetself._worker_loop,daemonTrue).start()这里有一个容易踩的坑Flask 的app_context。RQ Worker 是子进程启动时会自己调用create_app()天然有上下文。但守护线程跑在主进程里必须手动with self._app.app_context():包裹否则db.session会直接报错。这个问题排查了相当长时间才定位到。数据库的差异处理Server 用 MySQLDesktop 用 SQLite。SQLAlchemy 已经把大部分差异抽象掉了只有一个字段类型需要特殊处理# models.pyLongTextdb.TextifDEPLOY_MODEdesktopelsemysql.LONGTEXT()classTask(db.Model):polished_textdb.Column(LongText)# Desktop 用 TextServer 用 LONGTEXT原因是 SQLite 的TEXT没有长度限制而 MySQL 的TEXT只有 64KB长论文必须用LONGTEXT。这是少数几个不能完全抽象掉的数据库差异。三、异步任务队列为什么选 RQ 而不是 Celery对于一个学术工具项目RQ 的简洁性远比 Celery 的功能全面性更重要。很多人第一反应是 Celery我也考虑过。但 Celery 有几个让我犹豫的地方配置复杂broker、backend、各种 worker 参数文档要啃一阵子依赖偏重对于一个学术工具引入 Celery RabbitMQ 的组合太重了任务函数耦合Celery 用task装饰器跟代码结构绑定较深RQ的优势在于 Python 原生、依赖只有 Redis、API 简单到几行就能上手fromrqimportQueue queueQueue(ai_tasks,connectionredis_client)queue.enqueue(process_task,task_id)# 就这么简单实际使用中RQ 还有一个很方便的地方rq info命令可以直接在终端查看队列状态、Worker 数量、任务积压情况排查问题非常直观。而且 RQ Worker 是纯 Python 进程出问题直接看日志就能定位不像 Celery 的 prefork/eventlet/gevent 模型那样排查起来比较曲折。任务派发工厂模式 模板方法任务有三种类型文本、DOCX、PDF。一开始我在process_task里写了一堆if task_type text: ...后来重构成了工厂方法def_get_processor(task):return{text:TextTaskProcessor,docx:DocxTaskProcessor,pdf:PdfTaskProcessor,}[task.task_type](task,redis_client)defprocess_task(task_id):taskTask.query.get(task_id)processor_get_processor(task)processor.run()# 模板方法BaseTaskProcessor.run()是模板方法定义了初始化 AI 服务 → 更新状态 → 处理 → 推送完成事件的标准流程子类只需要实现process()。这样新增一种文件类型只需要写一个 Processor 类其他代码完全不用动。四、实时推送为什么选 SSE 而不是 WebSocket单向推送场景下SSE 是比 WebSocket 更轻量、更省心的选择。任务是异步执行的前端怎么实时拿到进度三个方案对比方案优点缺点轮询实现简单延迟高浪费请求WebSocket双向通信实时性好需额外协议Nginx 配置相对复杂SSEHTTP 原生浏览器自动重连只能服务器→客户端单向我的需求是服务器单向推送润色进度不需要客户端通过同一条连接发送命令。SSE 完美匹配# 后端app.route(/api/tasks/stream/task_id)defstream_results(task_id):defgenerate():pubsubredis_client.pubsub()pubsub.subscribe(fprogress:{task_id})formsginpubsub.listen():ifmsg[type]message:yieldfdata:{msg[data]}\n\nreturnResponse(generate(),mimetypetext/event-stream)// 前端constesnewEventSource(/api/tasks/stream/${taskId})es.addEventListener(stream,etask.polished_textJSON.parse(e.data).chunk)es.addEventListener(done,ees.close())EventSource是浏览器原生 API内置断线自动重连机制前端代码非常简洁。Nginx 配置 SSE 的关键参数部署时 SSE 有一个容易踩的坑Nginx 默认会缓冲后端响应导致流式数据被攒成一大块才发给客户端。必须显式关闭缓冲location ~ ^/api/tasks/\d/stream$ { proxy_pass http://backend_api; proxy_http_version 1.1; proxy_set_header Connection ; proxy_buffering off; # 关键关闭响应缓冲 proxy_cache off; chunked_transfer_encoding off; gzip off; # SSE 不要压缩 proxy_read_timeout 600s; # 长连接超时要设够 }另外要注意的是SSE 连接会占用一个 HTTP 连接。Gunicorn 的 sync worker 会被长连接阻塞所以 worker 类型必须用gthread线程模型否则一个 SSE 连接就会占满一个 worker。五、长文档并发处理怎么快还不丢顺序索引化 ThreadPoolExecutor用 future-to-index 映射保证结果有序。DOCX 论文动辄上百段每段都要调一次 AI串行处理一篇 50 页的论文可能要十几分钟。必须并发但有一个关键约束段落顺序必须保留。方案索引化 ThreadPoolExecutordef_process_paragraphs_concurrent(self,paragraphs):results[None]*len(paragraphs)withThreadPoolExecutor(max_workers5)aspool:future_to_idx{pool.submit(self._process_single_paragraph,p,i):ifori,pinenumerate(paragraphs)ifneeds_polishing(p)}forfutureinas_completed(future_to_idx):idxfuture_to_idx[future]results[idx]future.result()returnresults# 顺序和原文一致关键点用dict[future, index]把 future 和原始位置绑定。as_completed哪个先完成就先处理哪个但最终results数组的顺序由 index 保证。这个模式在很多需要并发但保序的场景里都适用。任务取消机制长文档处理可能要几分钟用户中途想取消怎么办项目实现了一个CancellationChecker通过 Redis 信号实现跨线程的取消通知# 用户点击取消 → 写入 Redis 信号redis_client.set(fcancel:{task_id},1)# Worker 每处理完一个段落就检查一次classCancellationChecker:defis_cancelled(self,task_id):returnself.redis.exists(fcancel:{task_id})每个段落处理前都会调用is_cancelled()检查一旦检测到取消信号就提前退出不会浪费后续的 API 调用。Desktop 模式下 MemoryRedis 的exists()方法签名完全一致所以取消逻辑也是双模式通用的。为什么不用 asyncioOpenAI 官方 SDK 的同步版本基于requests。虽然有AsyncOpenAI但线程池对于 IO 密集型任务来说已经够用——5 个并发就能把 API 调用耗时从 15 分钟压缩到 3 分钟左右。再多反而容易触发 API 的速率限制得不偿失。六、提示词热插拔提示词是 Markdown 文件每次任务执行时按需读取修改后无需重启即刻生效。策略系统是这个项目里比较有意思的软设计。两个核心需求提示词要能不重启服务就更新调试阶段每次改完都要重启 Worker 太低效了多套策略可切换标准润色 / 深度改写 / 自定义最终方案提示词以 Markdown 文件存放在prompts/目录启动时不加载每次任务执行时按需读盘。Linux 文件系统的页缓存会处理掉重复 IO 的性能损耗实测开销可以忽略。# prompts_config.pydefload_strategy_prompt(strategy:str,lang:str)-str:pathPROMPTS_DIR/STRATEGIES[strategy][lang]returnpath.read_text(encodingutf-8)# 每次都读简单直接策略注册在一个 dict 里新增策略只需要加一行配置STRATEGIES{standard:{zh:cn_standard.md,en:en_standard.md},strict:{zh:cn_strict.md,en:en_strict.md},}前端通过ConfigSwitcher组件让用户在界面上直接切换策略选择后立即生效。非技术用户反馈说我想让它别把’其次’改成’第二点’我直接改 Markdown 文件他刷新页面就能看到效果迭代效率很高。七、踩过的坑挑几个有代表性的希望能帮后来者少走弯路。1. ResponseExtractor 的二次 AI 调用最初版本里从 AI 输出中提取干净文本去掉润色结果“这种前缀用的是再调一次 AI 让它只保留正文”。结果长文本场景下 API 调用量直接翻倍成本显然不可接受。后来改成正则优先的策略defextract_clean_text(text:str)-str:cleanedre.sub(r^(润色结果|结果|输出)[:]\s*,,text)cleanedre.sub(r^[\w]*\n,,cleaned)ifcleanedandlen(cleaned)10:returncleaned# 90% 的情况正则就够了return_ai_extract_fallback(text)# 实在不行再回退到 AI正则覆盖了 90% 以上的情况API 调用量直接减半。2. PyInstaller 打包的隐式依赖打 Desktop 模式 EXE 时各种隐式导入需要手动声明到hiddenimportshiddenimports[sqlalchemy.dialects.sqlite,# 不写SQLAlchemy 跑不起来pydantic.deprecated.decorator,# 不写OpenAI SDK 报错lxml.etree,# 不写docx 解析报错# ... 几十个]PyInstaller.utils.hooks.collect_all能帮你收集flask_sqlalchemy这种带数据文件的包但很多间接依赖还是要靠运行时报错逐个补充。建议做这种打包时用一台干净的机器测试不然本地环境的全局包会掩盖问题。3. SQLite 的线程安全Desktop 模式下 Flask 主线程和 MemoryQueue 的守护线程会同时访问 SQLite。SQLite 默认的线程安全级别不允许跨线程共享连接需要在连接字符串里加上check_same_threadFalse并且确保 SQLAlchemy 的连接池配置正确。这个问题在开发环境不容易复现因为请求量小但在多任务并发时会偶发报错。4. RedisKeyManager为什么硬编码 key 是技术债一开始 Redis 的 key 散落在十几个文件里redis_client.set(fcancel:{task_id},1)# task_service.pyredis_client.publish(fprogress:{task_id},...)# progress_publisher.pyredis_client.exists(fdocx_done:{task_id})# docx_processor.py改一次命名规范要全局搜索改十几个地方而且容易漏改导致诡异 bug。后来抽成了RedisKeyManagerclassRedisKeyManager:staticmethoddefcancel_key(task_id):returnfcancel:{task_id}staticmethoddefprogress_key(task_id):returnfprogress:{task_id}staticmethoddefdocx_done_key(task_id):returnfdocx_done:{task_id}任何分布式系统中的 key 都应该集中管理这是我下一个项目从第一天就会做的事情。八、扩展方向项目目前功能完整、运行稳定但还有不少值得探索的方向。批量 APIOpenAI 的 Batch API 价格便宜一半适合非实时的 DOCX 任务段落级缓存同一段落重复润色时直接命中缓存目前已实现任务级的 text_hash 去重本地模型优化项目已经兼容 Ollama 本地模型但提示词还没有针对本地模型的特点做专门调优多语言提示词目前支持中英文后续可以扩展到日语、韩语等学术论文常见语种技术栈总结层Server 模式Desktop 模式后端框架Flask GunicornFlask (threaded)数据库MySQL (LONGTEXT)SQLite (Text)缓存/消息RedisMemoryRedis (内存字典)任务队列RQ 独立 Worker 进程MemoryQueue 守护线程实时推送SSE Redis Pub/SubSSE 内存 Pub/Sub前端Vue 3 Pinia Vite同左打包进 EXEAI 调用OpenAI 兼容 API官方 / 代理 / Ollama同左部署Docker Compose (Nginx)PyInstaller EXE写在最后做这个项目最大的收获不是写了多少代码而是被迫想清楚了一些以前没认真思考过的设计问题。比如双模式听起来很酷但本质就是把基础设施当成可替换的依赖——抽象出接口然后按环境选择实现。这个思路一旦想通后面要加 Mac 模式、加云函数模式都是顺理成章的事。再比如异步任务的设计以前我习惯把状态塞数据库里然后轮询这次认真做了 Pub/Sub SSE做完才发现有些事情看起来轮询也能凑合但做对了之后用户体验会有质的提升。如果你也在做类似的工具——文档处理、异步任务系统、或者需要双模式部署的项目希望这篇文章里至少有一个点能帮到你。仓库地址在文章开头Issue 和 PR 都欢迎。