Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 29 additions & 10 deletions apps/api/routers/graph.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,8 @@

from uuid import UUID

from fastapi import APIRouter, Query
from fastapi import APIRouter, HTTPException, Query
from fastapi.concurrency import run_in_threadpool

from apps.api.deps import cache, get_paper_title, graph_service
from packages.domain.task_tracker import global_tracker
Expand Down Expand Up @@ -100,30 +101,48 @@ def _fn(progress_callback=None):


@router.get("/graph/similarity-map")
def similarity_map(
async def similarity_map(
topic_id: str | None = None,
limit: int = Query(default=200, ge=5, le=500),
) -> dict:
"""论文相似度 2D 散点图(UMAP 降维)"""
return graph_service.similarity_map(topic_id=topic_id, limit=limit)
"""论文相似度 2D 散点图(UMAP 降维,60s 缓存,线程池执行避免阻塞事件循环)"""
cache_key = f"graph_similarity_map_{topic_id}_{limit}"
cached = cache.get(cache_key)
if cached is not None:
return cached
result = await run_in_threadpool(graph_service.similarity_map, topic_id=topic_id, limit=limit)
cache.set(cache_key, result, ttl=60)
return result


@router.get("/graph/cluster-map")
def cluster_map(
async def cluster_map(
n_clusters: int = Query(default=12, ge=2, le=30),
limit: int = Query(default=5000, ge=10, le=20000),
) -> dict:
"""全库论文 embedding k-means 聚类(研究领域地图,每簇用 keywords 自动命名)"""
return graph_service.cluster_map(n_clusters=n_clusters, limit=limit)
"""全库论文 embedding k-means 聚类(研究领域地图,60s 缓存,线程池执行)"""
cache_key = f"graph_cluster_map_{n_clusters}_{limit}"
cached = cache.get(cache_key)
if cached is not None:
return cached
result = await run_in_threadpool(graph_service.cluster_map, n_clusters=n_clusters, limit=limit)
cache.set(cache_key, result, ttl=60)
return result


@router.get("/graph/similar-via-citation/{paper_id}")
def similar_via_citation(
paper_id: str,
async def similar_via_citation(
paper_id: UUID,
top_k: int = Query(default=5, ge=1, le=20),
) -> dict:
"""引用同一篇论文且语义相近的论文(co-citation + 向量补强)"""
return graph_service.similar_via_citation(paper_id=paper_id, top_k=top_k)
try:
return await run_in_threadpool(
graph_service.similar_via_citation, paper_id=str(paper_id), top_k=top_k
)
except ValueError as exc:
# get_by_id 在论文不存在时抛 ValueError,统一转 404(此前返回 500)
raise HTTPException(status_code=404, detail=str(exc)) from exc


@router.get("/graph/citation-tree/{paper_id}")
Expand Down
17 changes: 15 additions & 2 deletions apps/api/routers/papers.py
Original file line number Diff line number Diff line change
Expand Up @@ -290,6 +290,8 @@ def toggle_favorite(paper_id: UUID) -> dict:
@router.patch("/papers/{paper_id}/reject")
def toggle_reject(paper_id: UUID) -> dict:
"""切换论文"不感兴趣"状态(推荐系统负反馈)"""
from packages.ai.recommendation_service import invalidate_recommendations

with session_scope() as session:
repo = PaperRepository(session)
try:
Expand All @@ -298,7 +300,10 @@ def toggle_reject(paper_id: UUID) -> dict:
raise HTTPException(status_code=404, detail=str(exc)) from exc
p.rejected = not getattr(p, "rejected", False)
session.commit()
# 失效 folder_stats(左侧文件夹计数)+ 推荐缓存(被拒论文立即从推荐列表移除,
# 否则 recommend:{top_k} / today_summary 最长 5min 仍含该论文)
cache.invalidate("folder_stats")
invalidate_recommendations()
return {"id": str(p.id), "rejected": p.rejected}


Expand Down Expand Up @@ -548,7 +553,11 @@ def similar(
paper_id: UUID,
top_k: int = Query(default=5, ge=1, le=20),
) -> dict:
ids = rag_service.similar_papers(paper_id, top_k=top_k)
try:
ids = rag_service.similar_papers(paper_id, top_k=top_k)
except ValueError as exc:
# get_by_id 在论文不存在时抛 ValueError,统一转 404(此前返回 500)
raise HTTPException(status_code=404, detail=str(exc)) from exc
items = []
if ids:
with session_scope() as session:
Expand Down Expand Up @@ -589,7 +598,11 @@ def paper_duplicates(
from packages.ai.pipelines import PaperPipelines

pipelines = PaperPipelines()
return pipelines.detect_duplicates(paper_id, threshold=threshold)
try:
return pipelines.detect_duplicates(paper_id, threshold=threshold)
except ValueError as exc:
# get_by_id 在论文不存在时抛 ValueError,统一转 404(此前返回 500)
raise HTTPException(status_code=404, detail=str(exc)) from exc


@router.post("/papers/{paper_id}/reasoning")
Expand Down
21 changes: 13 additions & 8 deletions packages/ai/graph/similarity.py
Original file line number Diff line number Diff line change
Expand Up @@ -185,8 +185,6 @@ def similar_via_citation(self, paper_id: str, top_k: int = 5) -> dict:
with session_scope() as session:
repo = PaperRepository(session)
seed = repo.get_by_id(paper_id)
if not seed:
return {"paper_id": str(paper_id), "items": [], "note": "论文不存在"}
if not seed.embedding:
return {
"paper_id": str(paper_id),
Expand All @@ -195,13 +193,15 @@ def similar_via_citation(self, paper_id: str, top_k: int = 5) -> dict:
}
seed_vec = list(seed.embedding)

# 1. 取种子论文引用 / 被引的论文(双向)
# 1. 取种子论文引用 / 被引的论文(双向,限 200 防高被引 hub 过载)
cited_rows = (
session.execute(
_sa_select(Citation).where(
_sa_select(Citation)
.where(
(Citation.source_paper_id == str(paper_id))
| (Citation.target_paper_id == str(paper_id))
)
.limit(200)
)
.scalars()
.all()
Expand All @@ -215,13 +215,15 @@ def similar_via_citation(self, paper_id: str, top_k: int = 5) -> dict:
if not cited_ids:
return {"paper_id": str(paper_id), "items": [], "note": "无引用关系"}

# 2. 找也引用了 cited 中任一篇的论文(co-citation)
# 2. 找也引用了 cited 中任一篇的论文(co-citation,限 2000 防无界扩张)
co_cite_rows = (
session.execute(
_sa_select(Citation).where(
_sa_select(Citation)
.where(
Citation.source_paper_id.in_(list(cited_ids))
| Citation.target_paper_id.in_(list(cited_ids))
)
.limit(2000)
)
.scalars()
.all()
Expand All @@ -236,13 +238,16 @@ def similar_via_citation(self, paper_id: str, top_k: int = 5) -> dict:
if not co_cite_ids:
return {"paper_id": str(paper_id), "items": [], "note": "无 co-citation 候选"}

# 3. 在 co-citation 集合里按 embedding cosine 排序
# 3. 在 co-citation 集合里按 embedding cosine 排序(排除被屏蔽论文)
candidates = (
session.execute(
_sa_select(Paper).where(
_sa_select(Paper)
.where(
Paper.id.in_(list(co_cite_ids)),
Paper.embedding.is_not(None),
Paper.rejected.is_(False),
)
.limit(500)
)
.scalars()
.all()
Expand Down
11 changes: 11 additions & 0 deletions packages/ai/recommendation_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,17 @@ def _set_cache(key: str, value: object):
_ttl_cache[key] = (time.monotonic(), value)


def invalidate_recommendations() -> None:
"""清空推荐与今日简报缓存(toggle_reject 后调用,使负反馈立即生效)。

覆盖 `recommend:{top_k}` 与 `today_summary`(简报内嵌 recommend(top_k=5))。
与 deps.cache 是两套独立 dict,不能复用 deps.cache.invalidate。
"""
with _ttl_lock:
for k in [k for k in _ttl_cache if k.startswith("recommend:") or k == "today_summary"]:
del _ttl_cache[k]


def _mean_vector(vectors: list[list[float]]) -> list[float]:
"""计算向量集合的质心(自动过滤维度不一致的向量)"""
if not vectors:
Expand Down
4 changes: 4 additions & 0 deletions packages/storage/repositories/paper.py
Original file line number Diff line number Diff line change
Expand Up @@ -392,6 +392,7 @@ def similar_by_embedding(
select(Paper)
.where(Paper.id != str(exclude))
.where(Paper.embedding.is_not(None))
.where(Paper.rejected.is_(False))
.order_by(Paper.embedding.cosine_distance(vector))
.limit(limit)
)
Expand All @@ -401,6 +402,7 @@ def similar_by_embedding(
select(Paper)
.where(Paper.id != str(exclude))
.where(Paper.embedding.is_not(None))
.where(Paper.rejected.is_(False))
.order_by(Paper.created_at.desc())
.limit(max_candidates)
)
Expand Down Expand Up @@ -438,6 +440,7 @@ def semantic_candidates(
q = (
select(Paper)
.where(Paper.embedding.is_not(None))
.where(Paper.rejected.is_(False))
.order_by(Paper.embedding.cosine_distance(query_vector))
.limit(limit)
)
Expand All @@ -446,6 +449,7 @@ def semantic_candidates(
q = (
select(Paper)
.where(Paper.embedding.is_not(None))
.where(Paper.rejected.is_(False))
.order_by(Paper.created_at.desc())
.limit(max_candidates)
)
Expand Down