智学文枢：QThread 异步引擎与变式状态

analysis_store / variant_storeQThread 异步引擎与变式状态一句话摘要AI 分析、讲解词生成、变式题生成都是秒级起步的重活AnalysisStore用QThread子类把脏活甩到后台线程用finished / error信号把结果安全送回主线程再配合防重入检查与阶段推进标记让页面始终流畅、进度永远一致。一、引言先问第一性问题为什么 AI 调用必须异步一次 AI 分析请求从组装 Prompt 到 HTTP 请求再到模型推理动辄 520 秒。如果这段代码直接跑在 UI 线程主线程里Qt 的事件循环会被完全阻塞——窗口拖不动、按钮点不了、界面假死操作系统甚至会弹出程序未响应。更隐蔽的问题是完成时机不可控。用户点完AI分析按钮后结果什么时候回来1 秒还是 20 秒如果代码写成顺序执行用户就只能干等如果让页面轮询数据库又回到第 19 篇说的老路。第一性原理的答案很清楚长任务离开主线程执行完成后通过信号把结果投递回主线程。PySide6 的QThread子类 信号槽恰好是完成这件事的最小工具——不用引入concurrent.futures的线程池对象也不用自己写线程安全的队列。二、核心原理2.1 QThread 的正确打开方式PySide6 中推荐的做法是子类化QThread把任务写进run()方法线程启动后run()自动在新线程里执行。关键点finished/error是 Worker 自己声明的信号信号发射跨线程时由 Qt 队列投递槽函数在接收方主线程执行——这就是结果安全回主线程的底层机制Worker 的run()里绝不能直接操作任何 UI 控件那是主线程的地盘Store 扮演异步门面页面只调analysis_store.analyze_question(...)Store 负责创建 Worker、连接信号、转发广播。2.2 阶段推进mark_stageAI 工作流有明确的阶段分析analysis→ 讲解lecture→ 追问followup→ 变式variant。Worker 在关键节点调用mark_stage(question.id, analysis)推进习题状态再通过question_store.refresh_question_status()通知列表页刷新进度列。异步任务的终点要回写到共享 Store否则后台算完列表页还停留在旧进度。startrun 中理科文科emit finished异常UI 线程主线程AnalysisWorkerQThread 子类读题 repo学科判断ScienceAnalyzerHumanitiesAnalyzermark_stage analysis可选 TeachingEngine 生成讲解词mark_stage lecturerefresh_question_statusemit error三、代码实现3.1 AnalysisWorker后台线程的骨架classAnalysisWorker(QThread):AI分析后台线程finishedSignal(object)# AIAnalysis or NoneerrorSignal(str)# error messagedef__init__(self,question_id:str,grade:str,config:dictNone,student_id:strNone,generate_teaching:boolFalse,reuse_existing:boolFalse):super().__init__()self._question_idquestion_id self._gradegrade self._configconfig self._student_idstudent_id self._generate_teachinggenerate_teaching self._reuse_existingreuse_existingdefrun(self):try:withQuestionRepository()asrepo:questionrepo.get_by_id(self._question_id)ifnotquestion:self.error.emit(习题不存在)return# 根据学科选择分析引擎ifquestion.subjectinSCIENCE_SUBJECTS:analyzerScienceAnalyzer()else:analyzerHumanitiesAnalyzer()...# 推进阶段AI分析已完成mark_stage(question.id,analysis)...# 通知列表页刷新进度列question_store.refresh_question_status(question.id)self.finished.emit(result)exceptExceptionase:self.error.emit(str(e))源码位置ui/stores/analysis_store.py L19-L105结构值得细品构造时只收参数习题 ID、年级、Prompt 配置不碰任何外部状态——Worker 是自包含的run()整体 try/except任何异常都变成error信号绝不静默吞掉也不让线程带着异常死掉读题用with QuestionRepository() as repo上下文管理器线程里也保证连接归还末尾统一refresh_question_statusfinished.emit(result)把数据回写和结果通知收口。3.2 讲解词生成可选分支 幂等初始化generate_teachingTrue时Worker 在分析后追加生成讲解词并把分析结果写入库、初始化追问对话记录ifself._generate_teaching:engineTeachingEngine()talkengine.generate(question,result,gradeself._grade,student_idself._student_id,configself._config,)result.has_teaching_talk1result.teaching_talktalk result.teaching_talk_modelConfigManager().get(ai,model,default)result.teaching_talk_generated_atint(time.time())withAnalysisRepository()asarepo:arepo.create_or_update(result)# 初始化追问对话记录幂等带容错try:fromservices.conversation.conversation_serviceimportConversationManager mgrConversationManager()mgr.init_conversation(question_idself._question_id,student_idself._student_idor,subjectquestion.subjector,analysis_result{core_concept:result.core_conceptor,knowledge_gap:result.knowledge_gapor,step_by_step:result.step_by_stepor[],remediation_suggestion:result.remediation_suggestionor,},teaching_talktalk,source_typesingle,)exceptException:pass# 推进阶段AI讲解已完成mark_stage(question.id,lecture)源码位置ui/stores/analysis_store.py L65-L99init_conversation包裹在try/except Exception: pass里——对话初始化是锦上添花失败不能拖垮整个分析流程这是后台任务里常见的降级容错写法。3.3 AnalysisStore防重入的异步门面defanalyze_question(self,question_id:str,grade:str,config:dictNone,student_id:strNone,generate_teaching:boolFalse,reuse_existing:boolFalse):启动AI分析异步讲解页可复用已有分析并生成讲解词ifself._workerandself._worker.isRunning():self.analysis_error.emit(正在处理中请等待...)returnself.analysis_started.emit(question_id)self._workerAnalysisWorker(question_id,grade,config,student_id,generate_teaching,reuse_existing)self._worker.finished.connect(self._on_analysis_finished)self._worker.error.connect(self._on_analysis_error)self._worker.start()源码位置ui/stores/analysis_store.py L154-L177isRunning()防重入是必须的用户连点两次按钮第二次调用直接以analysis_error信号提示正在处理中而不是再起一个线程打爆 API。3.4 VariantStore同步侧的状态容器与 AnalysisStore 不同VariantStore是纯同步的——变式题是已生成的结果数据加载、删除都很快不需要后台线程classVariantStore(BaseStore):变式题状态管理器variants_loadedSignal(list)# List[Variant]variant_deletedSignal(str)# variant_iddef__init__(self):super().__init__()self._variants:List[Variant][]self._current_question_id:Optional[str]Nonedefload_variants(self,question_id:str):加载某道原题的变式题self._current_question_idquestion_id repoVariantRepository()try:self._variantsrepo.get_by_original_question(question_id)self.variants_loaded.emit(self._variants)finally:repo.close()源码位置ui/stores/variant_store.py L11-L38它记录了当前原题 IDget_variant先查内存缓存命中失败再回库兜底——两级查找尽量少打数据库。VariantStoreVariantWorker(QThread)AnalysisStore变式训练页用户VariantStoreVariantWorker(QThread)AnalysisStore变式训练页用户run() 在新线程点击「生成变式题」generate_variants(qid)start()VariantEngine.generate(question)mark_stage(qid, variant)emit variant_finished(list)emit variant_finished(list)load_variants(qid)若未联动渲染变式卡片四、设计权衡 / 踩坑记录4.1 为什么用 QThread 子类而不是 QThreadPool / asyncioQThreadPoolQRunnable适合大量短任务而这里一次只有一个 AI 请求在跑简单子类即可且天然持有finished/error两个语义清晰的信号asyncio可以并发 IO但 AI 调用链里有同步的 SDK、本地文件操作混编成本高直接threading.Thread也能跑但结果回主线程要自己QMetaObject.invokeMethod或信号桥QThread子类把这些全包了。4.2 坑Worker 里禁止碰 UI早期版本曾尝试在 Worker 里直接更新按钮文字结果界面偶发卡死——Qt 控件不是线程安全的。现在的约定是Worker 只发信号UI 更新全部由主线程的槽函数完成。4.3 坑Worker 生命周期与悬挂引用self._worker AnalysisWorker(...)每次重建、保留引用防止 QThread 被 GC 后对象析构导致崩溃finished连接的是 Store 的私有槽_on_analysis_finishedStore 只做一件事——把信号原样转发成自己的公开信号。多一层转发看似冗余却让页面永远只依赖 Store 的公开接口不感知 Worker 的存在。4.4 权衡复用已有分析reuse_existing讲解页场景下用户可能已经分析过这道题。reuse_existingTrue时 Worker 先查库有结果就直接跳过 AI 调用、只生成讲解词——省一次昂贵的 API 调用也符合分析结果稳定不变的业务直觉。五、小结AI 调用是秒级操作必须放QThread子类的run()里执行绝不能阻塞主线程Worker 的finished/error信号跨线程队列投递结果安全回到主线程Store 作为异步门面统一转发isRunning()防重入避免连点导致并发请求异常统一转error信号绝不静默异步任务结束时回写共享 Storerefresh_question_status并推进阶段mark_stage保证列表页进度一致数据量小的纯查询变式列表仍走同步 Store异步不是银弹按需使用。