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

基于环境自适应架构的降低AIGC检测率系统——及其背后工程设计

一套代码，两个灵魂。Linux 服务器跑 Docker，Windows 双击 EXE，后端逻辑零修改。
项目地址：https://github.com/math89423-star/AI-Academic-Polisher
License: MIT

Disclaimer: 本工具仅供辅助学术写作与语言润色使用，旨在帮助作者提升论文的表达质量与可读性。使用者应确保最终提交的学术成果符合所在机构的学术诚信规范，本工具不应被用于规避学术诚信审查。

写在前面

寒假帮学弟改毕业论文，他用千问辅助写的初稿被维普的 AIGC 检测标记了 80% 以上的 AI 率。问我有没有什么办法，看了一圈市面上的工具，大多是套壳收费，效果参差不齐。

后来我自己折腾了一套提示词方案，先在本地跑通了，又顺手做了 Web 端方便几个朋友用。再后来想着干脆做个桌面版，让不懂技术的人也能直接用。这一折腾就是几个月，索性整理成了一个完整的开源项目：AI Academic Polisher。

这篇文章不是 README 的翻版。我想聊聊背后的几个技术决策——为什么用 RQ 不用 Celery、为什么 Desktop 模式要自己造一个 MemoryRedis、SSE 和 WebSocket 怎么选、长文档怎么并发切片还不丢顺序。希望能给做类似项目一些参考。

一、润色效果验证

工具好不好用，数据说话。以下是部分测试数据，用gemini-3.1-pro-preview模型润色后，提交到主流文本检测平台的评估结果（2026 年 4 月测试）：

PaperPass检测：

维普检测：

朱雀检测：

需要说明的是：这不是所谓的"降重"工具。它做的事情是把 AI 生成的、带有明显"机器味"的句子改写成更符合人类学术写作习惯的表达。原意保留，AIGC 识别率的下降是润色质量的自然体现。

具体的检测截图都在仓库的docs/目录下。

二、为什么要做双模式架构

核心思路：用工厂模式在启动时决定基础设施，上层代码完全无感知。

痛点

最早只做了 Web 版，部署在自己的服务器上给几个朋友用。问题很快暴露出来：

1. 使用门槛高：朋友里有非技术人员，让他们 SSH 进服务器看日志不太现实
2. 隐私顾虑：有人不放心把论文上传到别人的服务器
3. 资源冲突：大家共用一个 API Key，一个人密集调用就把别人的额度用光了

最直接的方案是做一个 Windows 桌面版，但又不想维护两份代码。于是就有了一套代码，两个灵魂的双模式架构。

模式自动检测

项目通过DEPLOY_MODE环境变量控制运行模式，支持三个值：server、desktop、auto。默认是auto，会根据操作系统自动判断：

# config.pydef_resolve_deploy_mode():mode=os.environ.get("DEPLOY_MODE","auto")ifmode=="auto":return"desktop"ifplatform.system()=="Windows"else"server"returnmode

Windows 上双击 EXE 自动进入 Desktop 模式，Linux 服务器上 Docker 启动自动进入 Server 模式，不需要手动配置。

工厂模式切换基础设施

# extensions.py 简化版ifDEPLOY_MODE=="server":redis_client=redis.Redis(host=...,port=...)task_queue=rq.Queue("ai_tasks",connection=redis_client)else:# desktopredis_client=MemoryRedis()# 内存字典 + threading.Locktask_queue=MemoryQueue()# queue.Queue + 守护线程

上层代码完全不知道自己跑在哪种模式下。Processor 调用redis_client.publish()推进度，Server 模式下走真 Redis，Desktop 模式下走内存 Pub/Sub，接口签名一模一样。

MemoryRedis：在内存里造一个 Redis

这是项目里我个人比较满意的一块设计。需求很明确：实现 redis-py 的方法子集，让上层代码零感知切换。

classMemoryRedis:def__init__(self):self._kv={}# 普通 KVself._hash=defaultdict(dict)# Hashself._set=defaultdict(set)# Setself._channels=defaultdict(list)# Pub/Sub 订阅者self._lock=threading.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 Worker

Server 模式下 RQ Worker 是一个独立进程，通过 Redis 拿任务。Desktop 模式下没有独立进程的概念，但又不能阻塞主线程，所以用了守护线程：

classMemoryQueue:def__init__(self):self._q=queue.Queue()self._app=Nonedefenqueue(self,func,*args):self._q.put((func,args))def_worker_loop(self):whileTrue:func,args=self._q.get()withself._app.app_context():# 关键：手动注入 app contexttry:func(*args)exceptException:logger.exception("Task failed")defstart_worker(self):threading.Thread(target=self._worker_loop,daemon=True).start()

这里有一个容易踩的坑：Flask 的app_context。RQ Worker 是子进程，启动时会自己调用create_app()，天然有上下文。但守护线程跑在主进程里，必须手动with self._app.app_context():包裹，否则db.session会直接报错。这个问题排查了相当长时间才定位到。

数据库的差异处理

Server 用 MySQL，Desktop 用 SQLite。SQLAlchemy 已经把大部分差异抽象掉了，只有一个字段类型需要特殊处理：

# models.pyLongText=db.TextifDEPLOY_MODE=="desktop"elsemysql.LONGTEXT()classTask(db.Model):polished_text=db.Column(LongText)# Desktop 用 Text，Server 用 LONGTEXT

原因是 SQLite 的TEXT没有长度限制，而 MySQL 的TEXT只有 64KB，长论文必须用LONGTEXT。这是少数几个不能完全抽象掉的数据库差异。

三、异步任务队列：为什么选 RQ 而不是 Celery

对于一个学术工具项目，RQ 的简洁性远比 Celery 的功能全面性更重要。

很多人第一反应是 Celery，我也考虑过。但 Celery 有几个让我犹豫的地方：

1. 配置复杂：broker、backend、各种 worker 参数，文档要啃一阵子
2. 依赖偏重：对于一个学术工具，引入 Celery + RabbitMQ 的组合太重了
3. 任务函数耦合：Celery 用@task装饰器，跟代码结构绑定较深

RQ的优势在于 Python 原生、依赖只有 Redis、API 简单到几行就能上手：

fromrqimportQueue queue=Queue("ai_tasks",connection=redis_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):task=Task.query.get(task_id)processor=_get_processor(task)processor.run()# 模板方法

BaseTaskProcessor.run()是模板方法，定义了"初始化 AI 服务 → 更新状态 → 处理 → 推送完成事件"的标准流程，子类只需要实现process()。这样新增一种文件类型只需要写一个 Processor 类，其他代码完全不用动。

四、实时推送：为什么选 SSE 而不是 WebSocket

单向推送场景下，SSE 是比 WebSocket 更轻量、更省心的选择。

任务是异步执行的，前端怎么实时拿到进度？三个方案对比：

我的需求是服务器单向推送润色进度，不需要客户端通过同一条连接发送命令。SSE 完美匹配：

# 后端@app.route("/api/tasks/stream/<task_id>")defstream_results(task_id):defgenerate():pubsub=redis_client.pubsub()pubsub.subscribe(f"progress:{task_id}")formsginpubsub.listen():ifmsg["type"]=="message":yieldf"data:{msg['data']}\n\n"returnResponse(generate(),mimetype="text/event-stream")

// 前端constes=newEventSource(`/api/tasks/stream/${taskId}`)es.addEventListener("stream",e=>task.polished_text+=JSON.parse(e.data).chunk)es.addEventListener("done",e=>es.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 页的论文可能要十几分钟。必须并发，但有一个关键约束：段落顺序必须保留。

方案：索引化 + ThreadPoolExecutor

def_process_paragraphs_concurrent(self,paragraphs):results=[None]*len(paragraphs)withThreadPoolExecutor(max_workers=5)aspool:future_to_idx={pool.submit(self._process_single_paragraph,p,i):ifori,pinenumerate(paragraphs)ifneeds_polishing(p)}forfutureinas_completed(future_to_idx):idx=future_to_idx[future]results[idx]=future.result()returnresults# 顺序和原文一致

关键点：用dict[future, index]把 future 和原始位置绑定。as_completed哪个先完成就先处理哪个，但最终results数组的顺序由 index 保证。这个模式在很多需要并发但保序的场景里都适用。

任务取消机制

长文档处理可能要几分钟，用户中途想取消怎么办？项目实现了一个CancellationChecker，通过 Redis 信号实现跨线程的取消通知：

# 用户点击取消 → 写入 Redis 信号redis_client.set(f"cancel:{task_id}","1")# Worker 每处理完一个段落就检查一次classCancellationChecker:defis_cancelled(self,task_id):returnself.redis.exists(f"cancel:{task_id}")

每个段落处理前都会调用is_cancelled()检查，一旦检测到取消信号就提前退出，不会浪费后续的 API 调用。Desktop 模式下 MemoryRedis 的exists()方法签名完全一致，所以取消逻辑也是双模式通用的。

为什么不用 asyncio

OpenAI 官方 SDK 的同步版本基于requests。虽然有AsyncOpenAI，但线程池对于 IO 密集型任务来说已经够用——5 个并发就能把 API 调用耗时从 15 分钟压缩到 3 分钟左右。再多反而容易触发 API 的速率限制，得不偿失。

六、提示词热插拔

提示词是 Markdown 文件，每次任务执行时按需读取，修改后无需重启即刻生效。

策略系统是这个项目里比较有意思的"软"设计。两个核心需求：

1. 提示词要能不重启服务就更新（调试阶段每次改完都要重启 Worker 太低效了）
2. 多套策略可切换（标准润色 / 深度改写 / 自定义）

最终方案：提示词以 Markdown 文件存放在prompts/目录，启动时不加载，每次任务执行时按需读盘。Linux 文件系统的页缓存会处理掉重复 IO 的性能损耗，实测开销可以忽略。

# prompts_config.pydefload_strategy_prompt(strategy:str,lang:str)->str:path=PROMPTS_DIR/STRATEGIES[strategy][lang]returnpath.read_text(encoding="utf-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:cleaned=re.sub(r'^(润色结果|结果|输出)[:：]\s*','',text)cleaned=re.sub(r'^```[\w]*\n','',cleaned)ifcleanedandlen(cleaned)>10:returncleaned# 90% 的情况正则就够了return_ai_extract_fallback(text)# 实在不行再回退到 AI

正则覆盖了 90% 以上的情况，API 调用量直接减半。

2. PyInstaller 打包的隐式依赖

打 Desktop 模式 EXE 时，各种隐式导入需要手动声明到hiddenimports：

hiddenimports=["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_thread=False，并且确保 SQLAlchemy 的连接池配置正确。这个问题在开发环境不容易复现（因为请求量小），但在多任务并发时会偶发报错。

4. RedisKeyManager：为什么硬编码 key 是技术债

一开始 Redis 的 key 散落在十几个文件里：

redis_client.set(f"cancel:{task_id}","1")# task_service.pyredis_client.publish(f"progress:{task_id}",...)# progress_publisher.pyredis_client.exists(f"docx_done:{task_id}")# docx_processor.py

改一次命名规范要全局搜索改十几个地方，而且容易漏改导致诡异 bug。后来抽成了RedisKeyManager：

classRedisKeyManager:@staticmethoddefcancel_key(task_id):returnf"cancel:{task_id}"@staticmethoddefprogress_key(task_id):returnf"progress:{task_id}"@staticmethoddefdocx_done_key(task_id):returnf"docx_done:{task_id}"

任何分布式系统中的 key 都应该集中管理，这是我下一个项目从第一天就会做的事情。

八、扩展方向

项目目前功能完整、运行稳定，但还有不少值得探索的方向。

• 批量 API：OpenAI 的 Batch API 价格便宜一半，适合非实时的 DOCX 任务
• 段落级缓存：同一段落重复润色时直接命中缓存（目前已实现任务级的 text_hash 去重）
• 本地模型优化：项目已经兼容 Ollama 本地模型，但提示词还没有针对本地模型的特点做专门调优
• 多语言提示词：目前支持中英文，后续可以扩展到日语、韩语等学术论文常见语种

技术栈总结

写在最后

做这个项目最大的收获不是写了多少代码，而是被迫想清楚了一些以前没认真思考过的设计问题。

比如"双模式"听起来很酷，但本质就是把基础设施当成可替换的依赖——抽象出接口，然后按环境选择实现。这个思路一旦想通，后面要加 Mac 模式、加云函数模式都是顺理成章的事。

再比如异步任务的设计，以前我习惯把状态塞数据库里然后轮询，这次认真做了 Pub/Sub + SSE，做完才发现：有些事情看起来"轮询也能凑合"，但做对了之后用户体验会有质的提升。

如果你也在做类似的工具——文档处理、异步任务系统、或者需要双模式部署的项目，希望这篇文章里至少有一个点能帮到你。仓库地址在文章开头，Issue 和 PR 都欢迎。