From 9090283e7027018bb3fb16d60ac2d1e71d966a26 Mon Sep 17 00:00:00 2001 From: Color2333 <1552429809@qq.com> Date: Sat, 18 Jul 2026 20:24:40 +0800 Subject: [PATCH] =?UTF-8?q?fix(idle):=20=E8=A1=A5=E5=81=BF=E8=A7=A6?= =?UTF-8?q?=E5=8F=91=20bug=20+=20=E5=AD=98=E9=87=8F=E5=9B=9E=E5=A1=AB?= =?UTF-8?q?=E8=84=9A=E6=9C=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 发现 idle 补偿触发 bug:Critical #6 的 _compensate_stuck_skimmed 此前 挂在 _process_batch 末尾,但无 unread 论文时 _process_batch 提前 return 0(line 225-227),补偿永远不触发。库里 1239 篇卡在 skimmed 未精读(有 summary_md 但 deep_dive_md 空),idle_processor 反复报 "没有需要处理的未读论文"却从不补偿,stuck 论文永久卡死。 修复: - packages/ai/idle_processor.py:把 _compensate_stuck_skimmed 调用从 _process_batch 末尾移到 _run_loop,独立于 skim 批次触发。无论有无 unread 论文,每次 idle 检查都尝试补偿 stuck 论文。_process_batch 只负责 skim,补偿在循环层独立调用。 回填脚本: - scripts/redeep_stuck_skimmed.py:一次性回填存量 stuck 论文。按 skim_score 降序优先精读高价值论文(1233 篇高分>=0.8),支持 --dry-run/--limit/--min-score/--concurrency。deep_dive 幂等覆盖, 重跑安全。 测试:新增 TestIdleCompensationTrigger 2 项(_process_batch 无 unread 返回0且不调补偿、补偿独立捞 stuck 论文),全套 63 passed。 --- packages/ai/idle_processor.py | 9 +- scripts/redeep_stuck_skimmed.py | 155 ++++++++++++++++++++++++++++++++ tests/test_repositories.py | 49 ++++++++++ 3 files changed, 210 insertions(+), 3 deletions(-) create mode 100644 scripts/redeep_stuck_skimmed.py diff --git a/packages/ai/idle_processor.py b/packages/ai/idle_processor.py index 6355165..7bb94f1 100644 --- a/packages/ai/idle_processor.py +++ b/packages/ai/idle_processor.py @@ -318,9 +318,7 @@ def _process_batch(self) -> int: self._papers_processed += processed self.detector.mark_task_executed() - # Critical #6 补偿:对已 skim 但卡住未精读的论文补一次精读(独立配额受限) - deep_compensated = self._compensate_stuck_skimmed() - return processed + deep_compensated + return processed def _compensate_stuck_skimmed(self) -> int: """补偿已 skim 但未精读的论文(Critical #6)。 @@ -380,6 +378,11 @@ def _run_loop(self): if not self._is_processing: self._is_processing = True self._process_batch() + # Critical #6 补偿:独立于 skim 批次触发。此前补偿挂在 + # _process_batch 末尾,但无 unread 论文时它提前 return 0, + # 补偿永远不跑 → 1239 篇卡在 skimmed 未精读。改为在 _run_loop + # 独立调用,无论有无 unread 都尝试补偿 stuck 论文。 + self._compensate_stuck_skimmed() self._is_processing = False else: if self._is_processing: diff --git a/scripts/redeep_stuck_skimmed.py b/scripts/redeep_stuck_skimmed.py new file mode 100644 index 0000000..3ef0f4c --- /dev/null +++ b/scripts/redeep_stuck_skimmed.py @@ -0,0 +1,155 @@ +#!/usr/bin/env python3 +""" +补偿精读卡在 skimmed 的存量论文 - 一次性回填脚本。 + +背景:Critical #6 修复让 idle_processor 补偿 skimmed 未精读的论文,但补偿逻辑 +此前挂在 _process_batch 末尾,无 unread 论文时提前 return 导致永远不触发。 +库里积累 1239 篇卡在 skimmed(有 summary_md 但 deep_dive_md 空),其中 1233 篇 +高分(>=0.8)。本脚本一次性回填这些存量论文,按 skim_score 降序优先精读高价值论文。 + +修复 idle 补偿 bug 后,新抓取的 stuck 论文由 idle_processor 自动补偿;本脚本只 +处理修复前已卡住的存量。deep_dive 是幂等覆盖(upsert_deep_dive 覆写 deep_dive_md), +直接重跑安全。 + +@author Color2333 + +使用方式: + # 预览要精读的论文数量(不调用 API) + python scripts/redeep_stuck_skimmed.py --dry-run + + # 回填全部 stuck 论文(按分数降序) + python scripts/redeep_stuck_skimmed.py + + # 限制数量(如先跑 100 篇) + python scripts/redeep_stuck_skimmed.py --limit 100 + + # 只回填高分(>=0.8)的 + python scripts/redeep_stuck_skimmed.py --min-score 0.8 +""" + +from __future__ import annotations + +import argparse +import logging +import pathlib +import sys +from concurrent.futures import ThreadPoolExecutor, as_completed + +# 让脚本能直接 `python scripts/x.py` 运行(不依赖 PYTHONPATH 环境变量) +sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent)) + +from sqlalchemy import select + +from packages.ai.pipelines import PaperPipelines +from packages.ai.rate_limiter import acquire_api, get_rate_limiter +from packages.storage.db import session_scope +from packages.storage.models import AnalysisReport, Paper + +# 配置日志 +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", +) +logger = logging.getLogger(__name__) + + +def get_stuck_skimmed_papers( + limit: int = 0, min_score: float = 0.0 +) -> list[tuple[str, str, float]]: + """获取卡在 skimmed 未精读的论文(有 summary_md 但 deep_dive_md 空)。 + + 按 skim_score 降序,优先精读高价值论文。 + Returns: [(paper_id, title, skim_score), ...] + """ + with session_scope() as session: + stmt = ( + select(Paper.id, Paper.title, AnalysisReport.skim_score) + .join(AnalysisReport, AnalysisReport.paper_id == Paper.id) + .where(Paper.read_status == "skimmed") + .where(AnalysisReport.summary_md.is_not(None)) + .where(AnalysisReport.deep_dive_md.is_(None)) + .where(AnalysisReport.skim_score >= min_score) + .order_by(AnalysisReport.skim_score.desc()) + ) + if limit > 0: + stmt = stmt.limit(limit) + rows = session.execute(stmt).all() + return [(str(r.id), r.title, r.skim_score or 0.0) for r in rows] + + +def redeep_single(paper_id: str, title: str, score: float) -> dict: + """精读单篇论文。""" + pipelines = PaperPipelines() + limiter = get_rate_limiter() + result = { + "paper_id": paper_id[:8], + "title": title[:50], + "skim_score": score, + "deep_success": False, + "error": None, + } + + if not limiter.start_task(): + result["error"] = "并发满" + return result + try: + if not acquire_api("llm", timeout=30.0): + result["error"] = "LLM API 限流" + logger.warning(f"⚠️ [{paper_id[:8]}] LLM 限流,跳过") + return result + logger.info(f"📖 [{paper_id[:8]}] 开始精读 (skim={score:.2f}) {title[:40]}") + pipelines.deep_dive(paper_id) + result["deep_success"] = True + logger.info(f"✅ [{paper_id[:8]}] 精读完成 (skim={score:.2f})") + except Exception as exc: + result["error"] = str(exc)[:120] + logger.warning(f"❌ [{paper_id[:8]}] 精读失败: {exc}") + finally: + limiter.end_task() + return result + + +def main() -> None: + parser = argparse.ArgumentParser(description="回填卡在 skimmed 未精读的存量论文") + parser.add_argument("--dry-run", action="store_true", help="只统计不精读") + parser.add_argument("--limit", type=int, default=0, help="限制处理数量(0=全部)") + parser.add_argument("--min-score", type=float, default=0.0, help="只精读 skim_score >= 此值的") + parser.add_argument("--concurrency", type=int, default=2, help="并发数") + args = parser.parse_args() + + papers = get_stuck_skimmed_papers(limit=args.limit, min_score=args.min_score) + logger.info(f"找到 {len(papers)} 篇卡在 skimmed 未精读的论文 (min_score={args.min_score})") + + if not papers: + logger.info("无需回填") + return + + # 分数分布预览 + high = sum(1 for _, _, s in papers if s >= 0.8) + mid = sum(1 for _, _, s in papers if 0.65 <= s < 0.8) + low = sum(1 for _, _, s in papers if s < 0.65) + logger.info(f"分数分布: 高分(>=0.8)={high}, 中(0.65-0.8)={mid}, 低(<0.65)={low}") + + if args.dry_run: + logger.info("dry-run 模式:不精读,退出") + return + + success = 0 + failed = 0 + with ThreadPoolExecutor(max_workers=args.concurrency) as pool: + futures = { + pool.submit(redeep_single, pid, title, score): pid for pid, title, score in papers + } + for fut in as_completed(futures): + r = fut.result() + if r["deep_success"]: + success += 1 + else: + failed += 1 + + logger.info(f"回填完成:成功={success}, 失败={failed}, 总计={len(papers)}") + + +if __name__ == "__main__": + main() diff --git a/tests/test_repositories.py b/tests/test_repositories.py index 8c9bc6e..c7eadd6 100644 --- a/tests/test_repositories.py +++ b/tests/test_repositories.py @@ -493,3 +493,52 @@ def test_last_error_persisted_and_readable(self, db_session): repo.update_run_status(topic.id, error=None) db_session.refresh(topic) assert topic.last_error is None + + +class TestIdleCompensationTrigger: + """Critical #6 补偿触发 bug 回归:补偿须独立于 skim 批次,无 unread 时也跑 + + 此前 _compensate_stuck_skimmed 挂在 _process_batch 末尾,但无 unread 论文时 + _process_batch 提前 return 0,补偿永远不触发 → stuck 论文卡死。 + """ + + def test_process_batch_returns_zero_when_no_unread(self, db_session, monkeypatch): + """无 unread 论文时 _process_batch 返回 0(主路径提前返回)""" + from packages.ai.idle_processor import IdleProcessor + + ip = IdleProcessor() + # 无论文时 _get_unread_papers 返回空 → _process_batch 直接 return 0 + monkeypatch.setattr(ip, "_get_unread_papers", lambda limit=10: []) + # 不应触发补偿(补偿已移出 _process_batch) + called = [] + monkeypatch.setattr(ip, "_compensate_stuck_skimmed", lambda: called.append(1) or 0) + result = ip._process_batch() + assert result == 0 + assert called == [], "_process_batch 不应再调用补偿(已移到 _run_loop)" + + def test_compensate_runs_independently_of_unread(self, db_session, monkeypatch): + """补偿独立触发:无 unread 但有 stuck skimmed 时仍补偿精读""" + from uuid import uuid4 + + from packages.domain.enums import ReadStatus + from packages.storage.models import AnalysisReport + + repo = PaperRepository(db_session) + paper = repo.upsert_paper( + PaperCreate(arxiv_id="2401.00301", title="stuck", abstract="a", metadata={}) + ) + repo.update_read_status(paper.id, ReadStatus.skimmed) + db_session.add( + AnalysisReport(id=str(uuid4()), paper_id=paper.id, summary_md="skim", deep_dive_md=None) + ) + db_session.commit() + + from packages.ai.idle_processor import IdleProcessor + + ip = IdleProcessor() + # 无 unread 论文 + monkeypatch.setattr(ip, "_get_unread_papers", lambda limit=10: []) + # 补偿应能捞到这篇 stuck 论文(验证查询独立可用) + stuck = ip._get_stuck_skimmed_papers(limit=5) + assert len(stuck) == 1, "应捞到 1 篇 stuck skimmed 论文" + assert stuck[0][0] == paper.id