From 681c3bf1eed5b0e7f3092c2e719193d2c889c333 Mon Sep 17 00:00:00 2001 From: Color2333 <1552429809@qq.com> Date: Fri, 17 Jul 2026 23:51:18 +0800 Subject: [PATCH] =?UTF-8?q?perf(backend):=2014=20=E7=AB=AF=E7=82=B9?= =?UTF-8?q?=E7=BC=93=E5=AD=98+threadpool=20+=20N+1=20=E4=BF=AE=E5=A4=8D=20?= =?UTF-8?q?+=20=E6=80=A7=E8=83=BD=E7=B4=A2=E5=BC=95=20+=20system.py=20?= =?UTF-8?q?=E8=AE=A1=E6=95=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PR3 — 后端 API 性能 graph.py（14 端点补缓存 + threadpool，内联 cluster_map 模式）: - overview/bridges/frontier：补 run_in_threadpool（此前 sync+cache 仍阻塞线程池槽） - cocitation-clusters O(N²) 配对：加 300s 缓存 + threadpool - citation-tree/citation-network/topic/similar-via-citation：加 60s 缓存 + threadpool - timeline/quality：加 120s 缓存 + threadpool - weekly_evolution/survey（含 LLM）：加 300s 缓存 + threadpool - research-gaps（最重，2×timeline+2×LLM）：加 600s 缓存 + threadpool - citation-detail/topic_deep_trace/auto_link：含外部 API 副作用，不加缓存（如实标注） N+1 修复: - papers.similar：循环 get_by_id 改 list_by_ids 一次查完 + 按 ids 顺序构建（5-20 次查询→1 次） - topics.list_topics：每主题 2 次查询改 2 次批量聚合（paper_count GROUP BY + 最近 action 按 topic 分组取首条） 性能索引迁移（alembic d4e5f6a7b8c9）: - papers.publication_date 索引（list_paginated 排序/frontier 过滤/年份分组，此前全表扫） - collection_actions.created_at 索引（ORDER BY created_at DESC/每主题最近行动查询） - models.py 对应字段加 index=True 声明 - CREATE INDEX IF NOT EXISTS 幂等（PG/SQLite 通用） system.py: - 拉 200 行 ORM 仅为 len() 改 count_all() 一次 COUNT 查询 --- apps/api/routers/graph.py | 118 +++++++++++++----- apps/api/routers/papers.py | 42 ++++--- apps/api/routers/system.py | 5 +- apps/api/routers/topics.py | 54 +++++++- .../versions/d4e5f6a7b8c9_add_perf_indexes.py | 34 +++++ packages/storage/models.py | 6 +- 6 files changed, 204 insertions(+), 55 deletions(-) create mode 100644 infra/migrations/versions/d4e5f6a7b8c9_add_perf_indexes.py diff --git a/apps/api/routers/graph.py b/apps/api/routers/graph.py index 9231128..b37f058 100644 --- a/apps/api/routers/graph.py +++ b/apps/api/routers/graph.py @@ -146,116 +146,174 @@ async def similar_via_citation( @router.get("/graph/citation-tree/{paper_id}") -def citation_tree( +async def citation_tree( paper_id: str, depth: int = Query(default=2, ge=1, le=5), ) -> dict: - return graph_service.citation_tree(root_paper_id=paper_id, depth=depth) + """引用树 BFS（60s 缓存，线程池执行避免阻塞事件循环）""" + cache_key = f"graph_citation_tree_{paper_id}_{depth}" + cached = cache.get(cache_key) + if cached is not None: + return cached + result = await run_in_threadpool( + graph_service.citation_tree, root_paper_id=paper_id, depth=depth + ) + cache.set(cache_key, result, ttl=60) + return result @router.get("/graph/citation-detail/{paper_id}") def citation_detail(paper_id: str) -> dict: - """获取单篇论文的丰富引用详情（含参考文献和被引列表）""" + """获取单篇论文的丰富引用详情（含参考文献和被引列表，含外部 API 副作用，不缓存）""" return graph_service.citation_detail(paper_id=paper_id) @router.get("/graph/citation-network/topic/{topic_id}") -def topic_citation_network(topic_id: str) -> dict: - """获取主题内论文的互引网络""" - return graph_service.topic_citation_network(topic_id=topic_id) +async def topic_citation_network(topic_id: str) -> dict: + """获取主题内论文的互引网络（60s 缓存，线程池执行）""" + cache_key = f"graph_topic_citation_network_{topic_id}" + cached = cache.get(cache_key) + if cached is not None: + return cached + result = await run_in_threadpool(graph_service.topic_citation_network, topic_id=topic_id) + cache.set(cache_key, result, ttl=60) + return result @router.post("/graph/citation-network/topic/{topic_id}/deep-trace") def topic_deep_trace(topic_id: str) -> dict: - """对主题内论文执行深度溯源，拉取外部引用并进行共引分析""" + """对主题内论文执行深度溯源，拉取外部引用并进行共引分析（含外部 API 副作用，不缓存）""" return graph_service.topic_deep_trace(topic_id=topic_id) @router.get("/graph/overview") -def graph_overview() -> dict: - """全库引用概览 — 节点 + 边 + PageRank + 统计（60s 缓存）""" +async def graph_overview() -> dict: + """全库引用概览 — 节点 + 边 + PageRank + 统计（60s 缓存，线程池执行）""" cached = cache.get("graph_overview") if cached is not None: return cached - result = graph_service.library_overview() + result = await run_in_threadpool(graph_service.library_overview) cache.set("graph_overview", result, ttl=60) return result @router.get("/graph/bridges") -def graph_bridges() -> dict: - """跨主题桥接论文（60s 缓存）""" +async def graph_bridges() -> dict: + """跨主题桥接论文（60s 缓存，线程池执行）""" cached = cache.get("graph_bridges") if cached is not None: return cached - result = graph_service.cross_topic_bridges() + result = await run_in_threadpool(graph_service.cross_topic_bridges) cache.set("graph_bridges", result, ttl=60) return result @router.get("/graph/frontier") -def graph_frontier( +async def graph_frontier( days: int = Query(default=90, ge=7, le=365), ) -> dict: - """研究前沿检测（60s 缓存）""" + """研究前沿检测（60s 缓存，线程池执行）""" cache_key = f"graph_frontier_{days}" cached = cache.get(cache_key) if cached is not None: return cached - result = graph_service.research_frontier(days=days) + result = await run_in_threadpool(graph_service.research_frontier, days=days) cache.set(cache_key, result, ttl=60) return result @router.get("/graph/cocitation-clusters") -def graph_cocitation_clusters( +async def graph_cocitation_clusters( min_cocite: int = Query(default=2, ge=1, le=10), ) -> dict: - """共引聚类分析""" - return graph_service.cocitation_clusters(min_cocite=min_cocite) + """共引聚类分析 O(N²) 配对（300s 缓存，线程池执行）""" + cache_key = f"graph_cocitation_clusters_{min_cocite}" + cached = cache.get(cache_key) + if cached is not None: + return cached + result = await run_in_threadpool(graph_service.cocitation_clusters, min_cocite=min_cocite) + cache.set(cache_key, result, ttl=300) + return result @router.post("/graph/auto-link") def graph_auto_link(paper_ids: list[str]) -> dict: - """手动触发引用自动关联""" + """手动触发引用自动关联（含外部 API 副作用，不缓存）""" return graph_service.auto_link_citations(paper_ids) @router.get("/graph/timeline") -def graph_timeline( +async def graph_timeline( keyword: str, limit: int = Query(default=100, ge=1, le=500), ) -> dict: - return graph_service.timeline(keyword=keyword, limit=limit) + """领域时间线 PageRank（120s 缓存，线程池执行）""" + cache_key = f"graph_timeline_{keyword}_{limit}" + cached = cache.get(cache_key) + if cached is not None: + return cached + result = await run_in_threadpool(graph_service.timeline, keyword=keyword, limit=limit) + cache.set(cache_key, result, ttl=120) + return result @router.get("/graph/quality") -def graph_quality( +async def graph_quality( keyword: str, limit: int = Query(default=120, ge=1, le=500), ) -> dict: - return graph_service.quality_metrics(keyword=keyword, limit=limit) + """图谱质量指标（120s 缓存，线程池执行）""" + cache_key = f"graph_quality_{keyword}_{limit}" + cached = cache.get(cache_key) + if cached is not None: + return cached + result = await run_in_threadpool(graph_service.quality_metrics, keyword=keyword, limit=limit) + cache.set(cache_key, result, ttl=120) + return result @router.get("/graph/evolution/weekly") -def graph_weekly_evolution( +async def graph_weekly_evolution( keyword: str, limit: int = Query(default=160, ge=1, le=500), ) -> dict: - return graph_service.weekly_evolution(keyword=keyword, limit=limit) + """周演化（含 LLM，300s 缓存，线程池执行）""" + cache_key = f"graph_weekly_evolution_{keyword}_{limit}" + cached = cache.get(cache_key) + if cached is not None: + return cached + result = await run_in_threadpool(graph_service.weekly_evolution, keyword=keyword, limit=limit) + cache.set(cache_key, result, ttl=300) + return result @router.get("/graph/survey") -def graph_survey( +async def graph_survey( keyword: str, limit: int = Query(default=120, ge=1, le=500), ) -> dict: - return graph_service.survey(keyword=keyword, limit=limit) + """领域综述（含 LLM，300s 缓存，线程池执行）""" + cache_key = f"graph_survey_{keyword}_{limit}" + cached = cache.get(cache_key) + if cached is not None: + return cached + result = await run_in_threadpool(graph_service.survey, keyword=keyword, limit=limit) + cache.set(cache_key, result, ttl=300) + return result @router.get("/graph/research-gaps") -def graph_research_gaps( +async def graph_research_gaps( keyword: str, limit: int = Query(default=120, ge=1, le=500), ) -> dict: - return graph_service.detect_research_gaps(keyword=keyword, limit=limit) + """研究空白检测（含 2×timeline + 2×LLM，最重图谱端点，600s 缓存，线程池执行）""" + cache_key = f"graph_research_gaps_{keyword}_{limit}" + cached = cache.get(cache_key) + if cached is not None: + return cached + result = await run_in_threadpool( + graph_service.detect_research_gaps, keyword=keyword, limit=limit + ) + cache.set(cache_key, result, ttl=600) + return result diff --git a/apps/api/routers/papers.py b/apps/api/routers/papers.py index 8c5cbc0..0b9a78b 100644 --- a/apps/api/routers/papers.py +++ b/apps/api/routers/papers.py @@ -560,28 +560,30 @@ def similar( raise HTTPException(status_code=404, detail=str(exc)) from exc items = [] if ids: + # N+1 修复：一次 list_by_ids 查完，按 ids 顺序构建（缺失的补 UUID 占位） with session_scope() as session: repo = PaperRepository(session) - for pid in ids: - try: - p = repo.get_by_id(pid) - items.append( - { - "id": str(p.id), - "title": p.title, - "arxiv_id": p.arxiv_id, - "read_status": p.read_status.value if p.read_status else "unread", - } - ) - except Exception: - items.append( - { - "id": str(pid), - "title": str(pid), - "arxiv_id": None, - "read_status": "unread", - } - ) + by_id = {str(p.id): p for p in repo.list_by_ids([str(i) for i in ids])} + for pid in ids: + p = by_id.get(str(pid)) + if p is not None: + items.append( + { + "id": str(p.id), + "title": p.title, + "arxiv_id": p.arxiv_id, + "read_status": p.read_status.value if p.read_status else "unread", + } + ) + else: + items.append( + { + "id": str(pid), + "title": str(pid), + "arxiv_id": None, + "read_status": "unread", + } + ) return { "paper_id": str(paper_id), "similar_ids": [str(x) for x in ids], diff --git a/apps/api/routers/system.py b/apps/api/routers/system.py index 4f0da62..72a2d38 100644 --- a/apps/api/routers/system.py +++ b/apps/api/routers/system.py @@ -32,7 +32,8 @@ def health() -> dict: def system_status() -> dict: with session_scope() as session: topics = TopicRepository(session).list_topics(enabled_only=False) - papers = PaperRepository(session).list_latest(limit=200) + # 浪费的全量加载修复：此前拉 200 行 ORM 仅为 len()，改 count_all() 一次 COUNT 查询 + papers_total = PaperRepository(session).count_all() runs = PipelineRunRepository(session).list_latest(limit=50) failed = [r for r in runs if r.status.value == "failed"] return { @@ -40,7 +41,7 @@ def system_status() -> dict: "counts": { "topics": len(topics), "enabled_topics": len([t for t in topics if t.enabled]), - "papers_latest_200": len(papers), + "papers_latest_200": papers_total, "runs_latest_50": len(runs), "failed_runs_latest_50": len(failed), }, diff --git a/apps/api/routers/topics.py b/apps/api/routers/topics.py index cd32bb0..8f7d121 100644 --- a/apps/api/routers/topics.py +++ b/apps/api/routers/topics.py @@ -66,9 +66,61 @@ def _topic_dict(t, session=None) -> dict: @router.get("/topics") def list_topics(enabled_only: bool = False) -> dict: + from sqlalchemy import func, select + + from packages.storage.models import CollectionAction, PaperTopic + with session_scope() as session: topics = TopicRepository(session).list_topics(enabled_only=enabled_only) - return {"items": [_topic_dict(t, session) for t in topics]} + if not topics: + return {"items": []} + topic_ids = [t.id for t in topics] + + # N+1 修复：2 次批量聚合代替每主题 2 次查询（2N+1 → 3） + # 1. 批量论文计数（GROUP BY topic_id） + count_rows = session.execute( + select(PaperTopic.topic_id, func.count()) + .where(PaperTopic.topic_id.in_(topic_ids)) + .group_by(PaperTopic.topic_id) + ).all() + paper_counts = {row[0]: row[1] for row in count_rows} + + # 2. 批量最近一次行动（用窗口函数或每组取首条；SQLite/PG 通用：按 topic 分组取 created_at 最大） + # 一次查询拿所有相关 topic 的最新 action + latest_actions: dict = {} + action_rows = ( + session.execute( + select(CollectionAction) + .where(CollectionAction.topic_id.in_(topic_ids)) + .order_by(CollectionAction.topic_id, CollectionAction.created_at.desc()) + ) + .scalars() + .all() + ) + for a in action_rows: + if a.topic_id not in latest_actions: # 已按 topic + created_at desc 排序，首个即最新 + latest_actions[a.topic_id] = a + + items = [] + for t in topics: + d = { + "id": str(t.id), + "name": t.name, + "query": t.query, + "enabled": t.enabled, + "created_at": t.created_at.isoformat() if t.created_at else None, + "paper_count": paper_counts.get(t.id, 0), + "last_run_at": None, + "last_run_count": None, + } + last_action = latest_actions.get(t.id) + if last_action: + d["last_run_at"] = ( + last_action.created_at.isoformat() if last_action.created_at else None + ) + d["last_run_count"] = last_action.paper_count + items.append(d) + return {"items": items} @router.post("/topics") diff --git a/infra/migrations/versions/d4e5f6a7b8c9_add_perf_indexes.py b/infra/migrations/versions/d4e5f6a7b8c9_add_perf_indexes.py new file mode 100644 index 0000000..7ec9480 --- /dev/null +++ b/infra/migrations/versions/d4e5f6a7b8c9_add_perf_indexes.py @@ -0,0 +1,34 @@ +"""add performance indexes on papers.publication_date and collection_actions.created_at + +Revision ID: d4e5f6a7b8c9 +Revises: c3d4e5f6a7b8 +Create Date: 2026-07-17 14:00:00.000000 + +目的：为高频排序/过滤字段加索引，消除全表扫描。 +- papers.publication_date：用于 list_paginated 排序/过滤、frontier 过滤、年份分组 +- collection_actions.created_at：用于 ORDER BY created_at DESC、每主题最近行动查询 +用 CREATE INDEX IF NOT EXISTS 保证幂等（PG/SQLite 均支持）。 +""" +from alembic import op + + +# revision identifiers, used by Alembic. +revision = "d4e5f6a7b8c9" +down_revision = "c3d4e5f6a7b8" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.execute( + "CREATE INDEX IF NOT EXISTS ix_papers_publication_date ON papers (publication_date)" + ) + op.execute( + "CREATE INDEX IF NOT EXISTS ix_collection_actions_created_at " + "ON collection_actions (created_at)" + ) + + +def downgrade() -> None: + op.execute("DROP INDEX IF EXISTS ix_collection_actions_created_at") + op.execute("DROP INDEX IF EXISTS ix_papers_publication_date") diff --git a/packages/storage/models.py b/packages/storage/models.py index 3811fa7..35607d8 100644 --- a/packages/storage/models.py +++ b/packages/storage/models.py @@ -37,7 +37,7 @@ class Paper(Base): arxiv_id: Mapped[str] = mapped_column(String(64), nullable=False, unique=True, index=True) abstract: Mapped[str] = mapped_column(Text, nullable=False, default="") pdf_path: Mapped[str | None] = mapped_column(String(1024), nullable=True) - publication_date: Mapped[date | None] = mapped_column(Date, nullable=True) + publication_date: Mapped[date | None] = mapped_column(Date, nullable=True, index=True) embedding: Mapped[list[float] | None] = mapped_column( "embedding_vec", Vector_or_JSON(1024), nullable=True ) @@ -397,7 +397,9 @@ class CollectionAction(Base): index=True, ) paper_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0) - created_at: Mapped[datetime] = mapped_column(DateTime, default=_utcnow, nullable=False) + created_at: Mapped[datetime] = mapped_column( + DateTime, default=_utcnow, nullable=False, index=True + ) class ActionPaper(Base):