perf(backend): 14 端点缓存+threadpool + N+1 修复 + 性能索引 + system.py 计数

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

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],

apps/api/routers/system.py

@@ -32,15 +32,16 @@ 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 {
             "health": health(),
             "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),
             },

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")

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")

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):