feat: add review_service with plant genomics presubmit, journals endpoint, and BFF proxy

This commit is contained in:
potaser
2026-10-09 16:54:03 +08:00
parent d37e0d9c3a
commit d43d30ee3f
24 changed files with 2322 additions and 0 deletions
+8
View File
@@ -42,3 +42,11 @@ paper_writer_service/paper.db
paper_writer_service/paper.db-shm
paper_writer_service/paper.db-wal
paper_writer_service/paper.log
# review_service 运行时产物(不入库)
review_service/review.db
review_service/review.db-wal
review_service/review.db-shm
review_service/review.log
# 期刊种子 journals.json 是源码:顶层 **/data/ 会误伤 review_service/data/,显式放行
!review_service/data/
+27
View File
@@ -0,0 +1,27 @@
# review_service/.env.example —— 运行时配置模板
# 用法: cp .env.example .env 后填写(.env 已被仓库 .gitignore 忽略,勿提交)
# restart.sh 会 `set -a; source .env` 导出给 uvicorn 进程。
# ---- LLM(DeepSeek;REVIEW_LLM_* 优先 → DOCGEN_LLM_* 回退 → 内置默认,见 services/llm.py)----
# 最简做法:从部署树抄一行即可复用同一把 key:
# grep '^DOCGEN_LLM_API_KEY=' /home/xzq/5090/teamWorkspace/zh/PlantAI-rag-new/sdk/.env >> .env
REVIEW_LLM_BASE_URL=https://api.deepseek.com/v1
REVIEW_LLM_API_KEY=
REVIEW_LLM_MODEL=deepseek-chat
#REVIEW_LLM_TIMEOUT=300
# ---- 上游(只读消费)----
#REVIEW_VSTORE_BASE_URL=http://127.0.0.1:8094
#REVIEW_METHODOLOGY_BASE_URL=http://127.0.0.1:8092 # 预留(MinerU 重链路,当前未用)
# ---- PDF 解析(经 vstore)----
#REVIEW_PARSE_TIMEOUT=180 # ingest→取文→清理 总读超时(秒)
#REVIEW_PARSE_DELETE_AFTER=1 # 1=解析后删除 vstore 临时文档(0=保留便于排查)
# ---- 自审稿 ----
#REVIEW_PRESUBMIT_READ_TIMEOUT=360 # LLM 长调用读超时(秒)
#REVIEW_MAX_TEXT_CHARS=60000 # 送 LLM 的正文截断上限
#REVIEW_MAX_PDF_BYTES=52428800 # 上传 PDF 上限 50MB
# ---- 服务(restart.sh 读取)----
#REVIEW_BIND_HOST=127.0.0.1
+7
View File
@@ -0,0 +1,7 @@
"""review_service —— 科研投稿与智能审稿助手(8100)。
独立部署树(模式同 paper_writer_service 8097 / bioflow_service 8098):
- BFF(sdk 8090) 经 routers/review_proxy.py 条件反代 /api/review/*(REVIEW_ENABLED)。
- 上游只读消费:vstore 8094(PDF 解析)、DeepSeek(LLM,REVIEW_LLM_*/DOCGEN_LLM_*)。
- 约束见根目录 AGENTS.md:审稿相关新代码只允许落在本目录。
"""
+1
View File
@@ -0,0 +1 @@
"""review 服务(8100)应用包。"""
+78
View File
@@ -0,0 +1,78 @@
"""config.py —— review 服务(8100)环境变量驱动配置(pydantic-settings)。
全部键可用环境变量覆盖(前缀 REVIEW_)。LLM 沿用 docgen 的裸键模式:
优先 REVIEW_LLM_*,回退 DOCGEN_LLM_*(与 paper_writer_service 共享同一套
DeepSeek 凭据,由 services/llm.py 直接读 os.environ)。零硬编码宿主路径。
"""
from __future__ import annotations
from functools import lru_cache
from pathlib import Path
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
model_config = SettingsConfigDict(
env_file=".env",
env_file_encoding="utf-8",
extra="ignore",
env_prefix="REVIEW_",
)
app_name: str = "plantai-review-service"
version: str = "0.1.0"
# ---- 上游(只读消费;不修改 vstore/methodology 任何代码)----
# vstore 8094:POST /api/vstore/ingest + GET/DELETE /api/vstore/documents/{id}
# (pymupdf 段落解析,同步秒级;methodology 8092 的 MinerU 链路是
# 分钟级异步 job,不适配同步自审稿,仅预留地址供后续扩展)
vstore_base_url: str = "http://127.0.0.1:8094"
methodology_base_url: str = "http://127.0.0.1:8092"
# ---- PDF 解析 ----
parse_timeout: float = 180.0 # vstore ingest→fetch→delete 读超时(秒)
parse_delete_after: bool = True # 解析完成后删除 vstore 临时文档(不污染私有库)
# ---- 自审稿(presubmit)----
presubmit_read_timeout: float = 360.0 # LLM 长调用读超时(秒)
max_text_chars: int = 60000 # 送 LLM 的正文截断上限(约 3 万汉字/6 万字符)
max_pdf_bytes: int = 50 * 1024 * 1024 # 上传 PDF 上限 50MB(对齐 extraction FR-EXTRACT-004)
# ---- 期刊情报与投稿推荐(recommend)----
db_path: str = "review.db" # 独立 SQLite(相对 review_service/;REVIEW_DB_PATH 覆盖)
journals_seed_path: str = "data/journals/journals.json" # 种子(相对 review_service/)
vstore_match_timeout: float = 30.0 # scope 入库/向量检索读超时(秒)
recommend_read_timeout: float = 60.0 # LLM 重排读超时(秒)
recommend_query_chars: int = 3000 # 送 vstore 的检索 query 截断(防嵌入模型超长)
recommend_scope_top_k: int = 10 # search-chunks top_k(vstore 上限 20)
@property
def _service_root(self) -> "Path":
# review_service/app/config.py → review_service/(相对本文件定位,不依赖 cwd)
return Path(__file__).resolve().parent.parent
@property
def db_path_abs(self) -> "Path":
# 绝对 db_path 会被 Path 除法原样返回;相对路径落在 review_service/ 下
return self._service_root / self.db_path
@property
def journals_seed_abs(self) -> "Path":
return self._service_root / self.journals_seed_path
@property
def vstore_base(self) -> str:
return self.vstore_base_url.rstrip("/")
@property
def methodology_base(self) -> str:
return self.methodology_base_url.rstrip("/")
@lru_cache(maxsize=1)
def get_settings() -> Settings:
return Settings()
settings = get_settings()
+191
View File
@@ -0,0 +1,191 @@
"""app/db.py —— review 服务独立 SQLite(aiosqlite + WAL,无 ORM):journals 表。
- 单连接 + asyncio 写锁(模式对齐 paper_writer_service/bioflow,但表结构与种子
逻辑为本模块自研);路径默认 review_service/review.db(REVIEW_DB_PATH 覆盖)。
- 种子:data/journals/journals.json → journals 表(幂等 upsert,vstore_doc_id 列
永不被种子覆盖——该列记录 scope_desc 在 vstore 8094 的 document_id)。
- 只在 review_service/ 内使用,不触碰 BFF/生产服务数据库。
"""
from __future__ import annotations
import asyncio
import json
from pathlib import Path
from typing import Any, Iterable
import aiosqlite
from .config import get_settings
_SCHEMA = """
CREATE TABLE IF NOT EXISTS journals (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
scope_desc TEXT NOT NULL,
impact_factor REAL,
review_time_weeks REAL,
acceptance_rate REAL,
submission_deadline TEXT,
official_url TEXT,
last_updated TEXT NOT NULL,
vstore_doc_id TEXT
);
"""
# 种子 upsert 列(刻意不含 vstore_doc_id:种子刷新不得清掉向量库映射)
_SEED_COLS = (
"id",
"name",
"scope_desc",
"impact_factor",
"review_time_weeks",
"acceptance_rate",
"submission_deadline",
"official_url",
"last_updated",
)
_UPSERT_SQL = (
"INSERT INTO journals (" + ", ".join(_SEED_COLS) + ") VALUES ("
+ ", ".join("?" for _ in _SEED_COLS) + ") "
+ "ON CONFLICT(id) DO UPDATE SET "
+ ", ".join(f"{c}=excluded.{c}" for c in _SEED_COLS if c != "id")
)
class SeedError(Exception):
"""种子文件缺失/损坏(journals.json 不可读)。"""
class Database:
"""aiosqlite 单连接封装(WAL;写串行化)。"""
def __init__(self, path: str | Path | None = None) -> None:
self._path: Path | None = Path(path) if path is not None else None
self._conn: aiosqlite.Connection | None = None
self._lock = asyncio.Lock()
@property
def path(self) -> Path:
if self._path is None:
self._path = get_settings().db_path_abs
return self._path
def set_path(self, path: str | Path) -> None:
"""测试注入:在进入 TestClient lifespan 之前预置库路径(须尚未 connect)。"""
if self._conn is not None:
raise RuntimeError("Database 已连接,无法再改路径(先 close_db())")
self._path = Path(path)
@property
def connected(self) -> bool:
return self._conn is not None
async def connect(self) -> None:
self.path.parent.mkdir(parents=True, exist_ok=True)
self._conn = await aiosqlite.connect(self.path)
self._conn.row_factory = aiosqlite.Row
await self._conn.execute("PRAGMA journal_mode=WAL")
await self._conn.execute("PRAGMA busy_timeout=5000")
await self._conn.executescript(_SCHEMA)
await self._conn.commit()
async def close(self) -> None:
if self._conn is not None:
await self._conn.close()
self._conn = None
@property
def conn(self) -> aiosqlite.Connection:
if self._conn is None:
raise RuntimeError("Database 未初始化:先 await init_db()")
return self._conn
async def fetchall(self, sql: str, params: Iterable[Any] = ()) -> list[dict[str, Any]]:
async with self.conn.execute(sql, tuple(params)) as cur:
rows = await cur.fetchall()
return [dict(r) for r in rows]
async def fetchone(self, sql: str, params: Iterable[Any] = ()) -> dict[str, Any] | None:
async with self.conn.execute(sql, tuple(params)) as cur:
row = await cur.fetchone()
return dict(row) if row is not None else None
async def execute(self, sql: str, params: Iterable[Any] = ()) -> int:
async with self._lock:
cur = await self.conn.execute(sql, tuple(params))
await self._conn.commit() # type: ignore[union-attr]
return cur.lastrowid or 0
async def executemany(self, sql: str, seq: list[tuple[Any, ...]]) -> None:
async with self._lock:
await self.conn.executemany(sql, seq)
await self._conn.commit() # type: ignore[union-attr]
db = Database()
async def init_db(path: str | Path | None = None) -> None:
"""幂等初始化:已连接则跳过(测试可在 lifespan 前 set_path 注入路径)。"""
if path is not None:
db.set_path(path)
if db.connected:
return
await db.connect()
async def close_db() -> None:
await db.close()
def set_path(path: str | Path) -> None:
"""同步注入库路径(测试 fixture 用;须在 init_db 之前调用)。"""
db.set_path(path)
async def ensure_journals_seeded() -> int:
"""装载种子期刊(幂等)。返回 journals 表行数。"""
row = await db.fetchone("SELECT COUNT(*) AS c FROM journals")
if row and int(row["c"]) > 0:
return int(row["c"])
seed_file = get_settings().journals_seed_abs
try:
payload = json.loads(Path(seed_file).read_text(encoding="utf-8"))
except FileNotFoundError as exc:
raise SeedError(f"期刊种子文件不存在:{seed_file}") from exc
except (OSError, ValueError) as exc:
raise SeedError(f"期刊种子文件不可读:{exc}") from exc
journals = payload.get("journals") if isinstance(payload, dict) else None
if not isinstance(journals, list) or not journals:
raise SeedError(f"期刊种子文件缺少 journals 数组:{seed_file}")
rows: list[tuple[Any, ...]] = []
for item in journals:
if not isinstance(item, dict) or not item.get("id") or not item.get("name"):
continue
rows.append(tuple(item.get(c) for c in _SEED_COLS))
if not rows:
raise SeedError(f"期刊种子文件无有效条目:{seed_file}")
await db.executemany(_UPSERT_SQL, rows)
return len(rows)
async def list_journals() -> list[dict[str, Any]]:
"""全部期刊(impact_factor 降序,NULL 靠后)。"""
return await db.fetchall(
"SELECT * FROM journals ORDER BY impact_factor DESC, name ASC"
)
async def get_journal(journal_id: str) -> dict[str, Any] | None:
return await db.fetchone("SELECT * FROM journals WHERE id = ?", (journal_id,))
async def set_vstore_doc_id(journal_id: str, doc_id: str) -> None:
"""记录 scope_desc 在 vstore 的 document_id(向量白名单检索用)。"""
await db.execute(
"UPDATE journals SET vstore_doc_id = ? WHERE id = ?", (doc_id, journal_id)
)
+71
View File
@@ -0,0 +1,71 @@
"""main.py —— review 服务应用入口(uvicorn 入口)。
- 独立部署树 review_service/(科研投稿与智能审稿助手,端口 8100)。
- 端点:/health(GET,BFF 探活反代)+ /api/review/*(自审稿等业务路由)。
- 不依赖 BFF/生产服务(8090 等绝不探测/连接);上游仅消费 vstore 8094。
"""
from __future__ import annotations
import logging
from contextlib import asynccontextmanager
import httpx
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from .config import get_settings
from .db import close_db, init_db
from .routers import journals, presubmit
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def create_app() -> FastAPI:
settings = get_settings()
app = FastAPI(title=settings.app_name, version=settings.version)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=False,
allow_methods=["*"],
allow_headers=["*"],
)
@asynccontextmanager
async def _lifespan(_app: FastAPI):
# journals 推荐依赖独立 SQLite:启动即建表(幂等;测试可先 set_path 注入
# tmp 路径,本函数沿用注入路径)
await init_db()
try:
yield
finally:
await close_db()
app.router.lifespan_context = _lifespan
@app.get("/health")
async def health() -> dict[str, object]:
vstore_ok = False
vstore_error: str | None = None
try:
async with httpx.AsyncClient(timeout=2.0) as client:
resp = await client.get(f"{settings.vstore_base}/api/vstore/health")
resp.raise_for_status()
vstore_ok = bool(resp.json().get("ok"))
except Exception as exc: # noqa: BLE001
vstore_error = str(exc)
return {
"ok": True,
"status": "ok" if vstore_ok else "degraded",
"upstreams": {"vstore": {"connected": vstore_ok, "error": vstore_error}},
"version": settings.version,
}
app.include_router(presubmit.router)
app.include_router(journals.router)
return app
app = create_app()
+1
View File
@@ -0,0 +1 @@
"""review 服务路由包。"""
+292
View File
@@ -0,0 +1,292 @@
"""routers/journals.py —— 期刊情报与智能投稿推荐。
- GET /api/review/journals 只读:journals 表全量列表(?field=/&order= 排序),
meta 含 count / last_updated / data_source_note。无任何写接口;空表返回
空数组 + meta.count=0(HTTP 200,绝不 500)。
- POST /api/review/recommend 智能投稿推荐:vstore 向量匹配(摘要 vs
scope_desc,白名单检索)→ LLM 按 id 白名单重排(幻觉期刊丢弃、缺漏规则
补位)→ 三档输出:挑战型 / 适配型 / 稳妥型,附匹配理由与风险提示。
降级哲学(同 presubmit):向量失败 → 全量候选 + vector_ok=false;
LLM 失败 → 规则分档 + llm_ok=false;推荐永不 5xx。仅入参非法返回
VALIDATION 错误信封(HTTP 200,与本服务其余端点一致)。
"""
from __future__ import annotations
import logging
from typing import Any, Literal
from fastapi import APIRouter
from pydantic import BaseModel, Field
from ..config import get_settings
from ..db import SeedError, ensure_journals_seeded, list_journals
from ..services import journal_kb, journal_rank, llm
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/review", tags=["review-journals"])
# ---- 请求/响应模型 ----
class Preferences(BaseModel):
max_review_weeks: float | None = Field(default=None, ge=1, le=104)
min_impact_factor: float | None = Field(default=None, ge=0, le=200)
tier_focus: Literal["all", "challenge", "fit", "safe"] = "all"
class RecommendRequest(BaseModel):
abstract: str
preferences: Preferences = Preferences()
class JournalItem(BaseModel):
id: str
name: str
impact_factor: float | None = None
review_time_weeks: float | None = None
acceptance_rate: float | None = None
official_url: str | None = None
similarity: float | None = None
tier: str
match_reason: str
risk_note: str
class TierBox(BaseModel):
label: str
items: list[JournalItem]
class RecommendResponse(BaseModel):
ok: bool
tiers: dict[str, TierBox] | None = None
meta: dict[str, Any] | None = None
error: dict[str, Any] | None = None
# ---- GET /api/review/journals(只读列表)----
class JournalBrief(BaseModel):
"""期刊公开字段(journals 表全列;不含 vstore_doc_id 内部映射)。"""
id: str
name: str
scope_desc: str = ""
impact_factor: float | None = None
review_time_weeks: float | None = None
acceptance_rate: float | None = None
submission_deadline: str | None = None
official_url: str | None = None
last_updated: str = ""
class JournalsListResponse(BaseModel):
ok: bool
items: list[JournalBrief] = []
meta: dict[str, Any] | None = None
error: dict[str, Any] | None = None
# ?field= 排序白名单(列名无法参数化,白名单防注入;非法值 → VALIDATION 信封)
_SORT_FIELDS = frozenset({
"id",
"name",
"impact_factor",
"review_time_weeks",
"acceptance_rate",
"last_updated",
})
_DATA_SOURCE_NOTE = (
"IF/审稿周期为公开近似值,acceptance_rate 为 null 表示暂无公开数据,需人工核验。"
)
def _sort_rows(rows: list[dict[str, Any]], field: str, order: str) -> list[dict[str, Any]]:
"""按白名单字段排序;NULL 恒排最后;次序不稳定时按 db 默认序(稳定排序保留)。"""
present = [r for r in rows if r.get(field) is not None]
missing = [r for r in rows if r.get(field) is None]
present.sort(key=lambda r: r.get(field), reverse=(order == "desc"))
return present + missing
def _list_error(message: str) -> JournalsListResponse:
return JournalsListResponse(
ok=False, items=[], meta=None,
error={"code": "VALIDATION", "message": message, "degraded": False},
)
@router.get("/journals", response_model=JournalsListResponse)
async def journals_list(
field: str = "name",
order: str = "asc",
) -> JournalsListResponse:
"""只读:journals 表全量列表(无写/删/改接口)。
- ?field= 排序字段白名单:id/name/impact_factor/review_time_weeks/
acceptance_rate/last_updated,默认 name;
- ?order=asc|desc,默认 asc;
- 空表 → items=[] + meta.count=0(HTTP 200,不报错)。种子装载由
POST /recommend 首次调用幂等完成,本端点严格只读。
"""
col = (field or "").strip()
if col not in _SORT_FIELDS:
return _list_error(
f"不支持的排序字段 field={field!r},可选:{', '.join(sorted(_SORT_FIELDS))}"
)
ord_norm = (order or "asc").strip().lower()
if ord_norm not in ("asc", "desc"):
return _list_error(f"不支持的排序方向 order={order!r},可选:asc, desc")
rows = await list_journals()
rows = _sort_rows(rows, col, ord_norm)
items = [
JournalBrief(
id=str(r.get("id") or ""),
name=str(r.get("name") or ""),
scope_desc=str(r.get("scope_desc") or ""),
impact_factor=r.get("impact_factor"),
review_time_weeks=r.get("review_time_weeks"),
acceptance_rate=r.get("acceptance_rate"),
submission_deadline=r.get("submission_deadline"),
official_url=r.get("official_url"),
last_updated=str(r.get("last_updated") or ""),
)
for r in rows
]
latest = max((str(r.get("last_updated") or "") for r in rows), default="")
meta: dict[str, Any] = {
"count": len(items),
"last_updated": latest or None,
"data_source_note": _DATA_SOURCE_NOTE,
}
return JournalsListResponse(ok=True, items=items, meta=meta, error=None)
def _error(code: str, message: str, *, degraded: bool = False) -> RecommendResponse:
return RecommendResponse(
ok=False, tiers=None, meta=None,
error={"code": code, "message": message, "degraded": degraded},
)
def _item_sort_key(it: JournalItem) -> tuple[int, float, float]:
return (
1 if it.similarity is None else 0,
-(it.similarity or 0.0),
-(it.impact_factor or 0.0),
)
def _apply_soft_preferences(item: JournalItem, prefs: Preferences) -> None:
"""软偏好:不满足只加风险提示,不删条目。"""
weeks = item.review_time_weeks
if prefs.max_review_weeks is not None and weeks is not None and weeks > prefs.max_review_weeks:
prefix = f"⚠ 审稿约 {weeks:g} 周,超出你 {prefs.max_review_weeks:g} 周的周期偏好;"
item.risk_note = prefix + ("无" if item.risk_note == "无" else item.risk_note)
inf = item.impact_factor
if prefs.min_impact_factor is not None:
if inf is None:
item.risk_note = "⚠ 影响因子未收录,无法确认满足你的要求;" + item.risk_note
elif inf < prefs.min_impact_factor:
prefix = f"⚠ 影响因子 {inf:g} 低于你 {prefs.min_impact_factor:g} 的要求;"
item.risk_note = prefix + ("无" if item.risk_note == "无" else item.risk_note)
@router.post("/recommend", response_model=RecommendResponse)
async def recommend(req: RecommendRequest) -> RecommendResponse:
settings = get_settings()
# ---- 入参校验(手动,错误信封与本服务其余端点一致)----
abstract = (req.abstract or "").strip()
if len(abstract) < 50:
return _error("VALIDATION", "abstract 至少需要 50 个字符(请提供摘要或正文片段)")
if len(abstract) > 20000:
return _error("VALIDATION", "abstract 超过 20000 字符上限")
prefs = req.preferences
# ---- 种子 + 期刊元数据 ----
try:
await ensure_journals_seeded()
journals = await list_journals()
except SeedError as exc:
logger.error("recommend: 种子装载失败:%s", exc)
return _error("SEED_UNAVAILABLE", str(exc), degraded=True)
if not journals:
return _error("SEED_UNAVAILABLE", "期刊库为空", degraded=True)
# ---- ① vstore 向量匹配(摘要 vs scope_desc)----
try:
candidates, vector_ok = await journal_kb.match(abstract, journals)
except journal_kb.MatchError as exc:
logger.warning("recommend: 向量匹配降级:%s", exc)
candidates, vector_ok = [], False
if not candidates:
# 无有效语义分 → 全量候选(similarity=None),LLM 仍可按 scope 文本重排
candidates = [{"journal": j, "similarity": None} for j in journals]
vector_ok = False
# ---- ② LLM 重排(id 白名单硬校验)----
prefs_dict = prefs.model_dump(exclude_none=True)
llm_ok = True
dropped = 0
repaired = 0
try:
rankings, dropped, repaired = await journal_rank.rerank(abstract, candidates, prefs_dict)
except (llm.LLMError, journal_rank.RankError, ValueError) as exc:
# ValueError:complete_json 最终 json.loads 修复仍失败(同 presubmit 宽容口径)
logger.warning("recommend: LLM 重排降级为规则分档:%s", exc)
rankings = journal_rank.fallback_rankings(candidates)
llm_ok = False
# ---- ③ 三档组装(名称/指标全部来自 DB,不采信 LLM 文本)----
by_id = {str(c["journal"]["id"]): c for c in candidates}
tiers: dict[str, list[JournalItem]] = {t: [] for t in journal_rank.TIERS}
for r in rankings:
cand = by_id.get(r.journal_id)
if cand is None:
continue
j = cand["journal"]
sim = cand.get("similarity")
item = JournalItem(
id=str(j["id"]),
name=str(j.get("name") or j["id"]),
impact_factor=j.get("impact_factor"),
review_time_weeks=j.get("review_time_weeks"),
acceptance_rate=j.get("acceptance_rate"),
official_url=j.get("official_url"),
similarity=round(float(sim), 4) if sim is not None else None,
tier=r.tier,
match_reason=r.match_reason,
risk_note=r.risk_note,
)
_apply_soft_preferences(item, prefs)
tiers.setdefault(r.tier, []).append(item)
# 档内排序:相似度(None 靠后)→ 影响因子
for box in tiers.values():
box.sort(key=_item_sort_key)
# tier_focus:软展示聚焦(默认 all;聚焦时其余档 items 置空,label 保留)
out_tiers: dict[str, TierBox] = {}
for t in journal_rank.TIERS:
items = tiers.get(t, [])
if prefs.tier_focus != "all" and t != prefs.tier_focus:
items = []
out_tiers[t] = TierBox(label=journal_rank.TIER_LABELS[t], items=items)
meta: dict[str, Any] = {
"model": llm._llm_config()["model"],
"vector_ok": vector_ok,
"llm_ok": llm_ok,
"seed_count": len(journals),
"candidate_count": len(candidates),
"dropped_hallucinated": dropped,
"repaired": repaired,
"preferences": prefs_dict,
"abstract_chars": len(abstract),
}
return RecommendResponse(ok=True, tiers=out_tiers, meta=meta, error=None)
+87
View File
@@ -0,0 +1,87 @@
"""routers/presubmit.py —— 投稿前自审稿(POST /api/review/presubmit)。
流程:上传 PDF → paper_parse(vstore 8094 解析)→ LLM 结构化审稿
(DeepSeek json_object,System Prompt 为植物基因组学六维专属清单,
见 services/self_review.py)→ 归一化为 SelfReviewReport。
"""
from __future__ import annotations
import logging
from fastapi import APIRouter, File, Form, UploadFile
from pydantic import BaseModel
from ..config import get_settings
from ..services import llm, paper_parse, self_review
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/review", tags=["review-presubmit"])
class PresubmitResponse(BaseModel):
ok: bool
report: self_review.SelfReviewReport | None = None
error: dict[str, object] | None = None
meta: dict[str, object] | None = None
def _error(code: str, message: str, *, degraded: bool = False) -> PresubmitResponse:
return PresubmitResponse(
ok=False, report=None,
error={"code": code, "message": message, "degraded": degraded},
)
@router.post("/presubmit", response_model=PresubmitResponse)
async def presubmit_review(
file: UploadFile = File(..., description="待审稿 PDF"),
target_journal: str = Form("", description="目标期刊(可选,影响格式类意见)"),
) -> PresubmitResponse:
settings = get_settings()
filename = file.filename or "paper.pdf"
if not filename.lower().endswith(".pdf"):
return _error("VALIDATION", "仅支持 PDF 文件")
content = await file.read()
if not content:
return _error("VALIDATION", "上传文件为空")
if len(content) > settings.max_pdf_bytes:
return _error("VALIDATION", "PDF 超过 50MB 上限")
# ---- 解析(vstore 8094)----
try:
text = await paper_parse.parse_pdf_bytes(content, filename=filename)
except paper_parse.ParseError as exc:
logger.warning("presubmit: 解析失败:%s", exc)
return _error("PARSE_FAILED", str(exc), degraded=True)
truncated = len(text) > settings.max_text_chars
feed_text = text[: settings.max_text_chars]
# ---- LLM 结构化自审(植物基因组学六维清单)----
try:
payload = await llm.complete_json(
self_review.SYSTEM_PROMPT,
self_review.build_user_prompt(feed_text, (target_journal or "").strip()),
temperature=0.2,
max_tokens=8000,
timeout=settings.presubmit_read_timeout,
)
except llm.LLMError as exc:
logger.error("presubmit: LLM 审稿失败:%s", exc)
return _error("UPSTREAM_DOWN", f"审稿模型调用失败:{exc}", degraded=True)
report = self_review.coerce_report(payload)
return PresubmitResponse(
ok=True,
report=report,
meta={
"filename": filename,
"target_journal": (target_journal or "").strip(),
"text_chars": len(feed_text),
"truncated": truncated,
"model": llm._llm_config()["model"],
},
)
+1
View File
@@ -0,0 +1 @@
"""review 服务服务层包。"""
+223
View File
@@ -0,0 +1,223 @@
"""services/journal_kb.py —— 期刊 scope_desc → vstore 入库 + 向量匹配(只读消费 8094)。
链路(全部复用 vstore 既有 API,零改动):
1. ensure_scope_docs:各期刊 scope_desc 以 .txt 入库(category=journal_scope,
name=期刊名),document_id 回写 journals.vstore_doc_id(幂等:已有 id 不重灌)。
2. match:GET /api/vstore/search-chunks?q=<摘要截断>&doc_ids=<id 白名单>&top_k=N
→ 按 doc_id 映射回期刊,取该期刊最高 chunk 分数为 similarity。
3. 自愈:白名单命中 0 条 → GET /documents/{id} 探活 → 失效的重灌(vstore 去重
语义下同内容重灌会换 doc_id,需回写)→ 重试一次 → 仍 0 视为语义低相似。
错误语义:连通性/上游异常抛 MatchError(路由降级为全量候选);
返回 ([], False) 表示检索成功但无有效匹配(路由同样降级)。
"""
from __future__ import annotations
import logging
from typing import Any
import httpx
from ..config import get_settings
from ..db import set_vstore_doc_id
logger = logging.getLogger(__name__)
SCOPE_CATEGORY = "journal_scope"
class MatchError(Exception):
"""vstore 向量匹配失败(不可达/响应异常)。"""
def _timeout() -> httpx.Timeout:
t = get_settings().vstore_match_timeout
return httpx.Timeout(connect=5.0, read=t, write=t, pool=5.0)
def _err_body(resp: httpx.Response) -> str:
try:
payload = resp.json()
except ValueError:
return (resp.text or "").strip()[:500] or "(空响应体)"
if isinstance(payload, dict):
err = payload.get("error")
if isinstance(err, dict) and err.get("message"):
return str(err["message"])
if isinstance(err, str):
return err
return str(payload)[:500]
async def _ingest_scope(client: httpx.AsyncClient, base: str, journal: dict[str, Any]) -> str:
"""单本期刊 scope_desc 入库,返回 document_id。"""
content = str(journal.get("scope_desc") or "").strip()
if not content:
raise MatchError(f"期刊 scope_desc 为空:{journal.get('id')}")
jid = str(journal.get("id") or "")
try:
resp = await client.post(
f"{base}/api/vstore/ingest",
files={"file": (f"{jid}.txt", content.encode("utf-8"), "text/plain")},
data={
"name": str(journal.get("name") or jid),
"category": SCOPE_CATEGORY,
"description": "review_service 期刊范围种子(scope_desc 向量匹配)",
},
)
except httpx.HTTPError as exc:
raise MatchError(f"vstore 不可达({base}):{exc}") from exc
if resp.status_code >= 400:
raise MatchError(f"vstore ingest 失败(HTTP {resp.status_code}):{_err_body(resp)}")
try:
payload = resp.json()
except ValueError as exc:
raise MatchError(f"vstore ingest 响应非 JSON:{exc}") from exc
if not isinstance(payload, dict) or not payload.get("ok"):
raise MatchError(
str((payload.get("error") or {}).get("message") if isinstance(payload, dict) else "") or "vstore ingest 失败"
)
doc_id = str((payload.get("item") or {}).get("document_id") or "")
if not doc_id:
raise MatchError("vstore ingest 响应缺少 document_id")
return doc_id
async def ensure_scope_docs(journals: list[dict[str, Any]]) -> list[dict[str, Any]]:
"""补齐缺失的 scope 入库映射(就地回写 vstore_doc_id 并持久到 DB)。"""
settings = get_settings()
missing = [j for j in journals if not j.get("vstore_doc_id")]
if not missing:
return journals
async with httpx.AsyncClient(timeout=_timeout()) as client:
for j in missing:
try:
doc_id = await _ingest_scope(client, settings.vstore_base, j)
except MatchError as exc:
# 单本失败不阻断其余(该期刊将不进白名单,由路由降级兜底)
logger.warning("journal_kb: scope 入库失败(%s):%s", j.get("id"), exc)
continue
j["vstore_doc_id"] = doc_id
await set_vstore_doc_id(str(j["id"]), doc_id)
return journals
async def _search(
client: httpx.AsyncClient, base: str, query: str, doc_ids: list[str]
) -> list[dict[str, Any]]:
"""chunk 级白名单检索。连通/上游错误抛 MatchError。"""
if not doc_ids:
# doc_ids 为空时 vstore 侧等于不过滤 → 会命中用户私有库,必须拦死
raise MatchError("白名单为空,拒绝全库检索")
try:
resp = await client.get(
f"{base}/api/vstore/search-chunks",
params={
"q": query,
"top_k": get_settings().recommend_scope_top_k,
"doc_ids": ",".join(doc_ids),
},
)
except httpx.HTTPError as exc:
raise MatchError(f"vstore search-chunks 不可达:{exc}") from exc
if resp.status_code >= 400:
raise MatchError(f"vstore search-chunks 失败(HTTP {resp.status_code}):{_err_body(resp)}")
try:
payload = resp.json()
except ValueError as exc:
raise MatchError(f"vstore search-chunks 响应非 JSON:{exc}") from exc
if not isinstance(payload, dict) or not payload.get("ok"):
raise MatchError(
str((payload.get("error") or {}).get("message") if isinstance(payload, dict) else "") or "vstore search-chunks 失败"
)
items = payload.get("items")
return items if isinstance(items, list) else []
async def _doc_alive(client: httpx.AsyncClient, base: str, doc_id: str) -> bool:
try:
resp = await client.get(f"{base}/api/vstore/documents/{doc_id}")
except httpx.HTTPError:
return False
if resp.status_code != 200:
return False
try:
payload = resp.json()
except ValueError:
return False
return bool(isinstance(payload, dict) and payload.get("ok"))
async def _refresh_dead_docs(
client: httpx.AsyncClient, base: str, journals: list[dict[str, Any]]
) -> None:
"""白名单 0 命中时探活:失效文档重灌并回写新 id。"""
for j in journals:
doc_id = str(j.get("vstore_doc_id") or "")
if not doc_id or await _doc_alive(client, base, doc_id):
continue
try:
new_id = await _ingest_scope(client, base, j)
except MatchError as exc:
logger.warning("journal_kb: scope 重灌失败(%s):%s", j.get("id"), exc)
continue
j["vstore_doc_id"] = new_id
await set_vstore_doc_id(str(j["id"]), new_id)
async def match(
abstract: str, journals: list[dict[str, Any]]
) -> tuple[list[dict[str, Any]], bool]:
"""摘要 vs scope_desc 向量匹配。
返回 (candidates, vector_ok);candidates 元素:
{"journal": <journals 行>, "similarity": float}
按 similarity 降序。MatchError → 路由降级;([], False) → 语义低相似。
"""
settings = get_settings()
query = abstract.strip()[: settings.recommend_query_chars]
if not query:
return [], False
await ensure_scope_docs(journals)
whitelisted = [j for j in journals if j.get("vstore_doc_id")]
if not whitelisted:
raise MatchError("无任何期刊 scope 入库成功(vstore_doc_id 为空)")
doc_map = {str(j["vstore_doc_id"]): j for j in whitelisted}
timeout = _timeout()
async with httpx.AsyncClient(timeout=timeout) as client:
items = await _search(client, settings.vstore_base, query, list(doc_map))
if not items:
await _refresh_dead_docs(client, settings.vstore_base, whitelisted)
doc_map = {str(j["vstore_doc_id"]): j for j in whitelisted if j.get("vstore_doc_id")}
if not doc_map:
return [], False
items = await _search(client, settings.vstore_base, query, list(doc_map))
# 聚合:每本期刊取最高 chunk 分
best: dict[str, float] = {}
for item in items:
if not isinstance(item, dict):
continue
j = doc_map.get(str(item.get("doc_id") or ""))
if j is None:
continue
try:
score = float(item.get("score"))
except (TypeError, ValueError):
continue
jid = str(j["id"])
if jid not in best or score > best[jid]:
best[jid] = score
if not best:
return [], False
by_id = {str(j["id"]): j for j in journals}
candidates = [
{"journal": by_id[jid], "similarity": score}
for jid, score in best.items()
if jid in by_id
]
candidates.sort(key=lambda c: c["similarity"], reverse=True)
return candidates, bool(candidates)
+210
View File
@@ -0,0 +1,210 @@
"""services/journal_rank.py —— LLM 重排 + 幻觉防护 + 三档归类。
职责:
1. rerank:把带语义分的候选期刊(来自 journal_kb)+ 用户偏好交给 LLM,
要求按 id 白名单返回 {rankings:[{id, tier, match_reason, risk_note}]}。
2. validate:硬校验——白名单外 id 丢弃(幻觉拦截);tier 非法/LLM 缺漏的
期刊按规则补位(绝不因 LLM 输出丢失候选期刊);期刊名称/指标一律不采信
LLM,最终展示全部用 DB 字段(防止改名型幻觉)。
3. fallback_rankings:LLM 不可用时的确定性规则分档(按 impact_factor)。
三档语义:challenge 挑战型(高影响/慢审稿)、fit 适配型(均衡)、
safe 稳妥型(快速/门槛低)。
"""
from __future__ import annotations
import json
from dataclasses import dataclass
from typing import Any
from ..config import get_settings
from . import llm
TIERS = ("challenge", "fit", "safe")
TIER_LABELS = {
"challenge": "挑战型(高影响因子/慢审稿)",
"fit": "适配型(中等影响/周期适中)",
"safe": "稳妥型(快速发表/门槛较低)",
}
# 规则分档阈值(近似值,可后续配置化)
_CHALLENGE_MIN_IF = 9.0
_SAFE_MAX_IF = 4.0
# tier 同义词归一(LLM 偶发中文标签)
_TIER_ALIASES = {
"挑战": "challenge",
"挑战型": "challenge",
"高风险": "challenge",
"适配": "fit",
"适配型": "fit",
"稳妥": "safe",
"稳妥型": "safe",
"低风险": "safe",
}
SYSTEM_PROMPT = """你是学术投稿推荐专家(植物基因组学方向)。根据给定候选期刊、论文摘要与用户偏好,为**每一个**候选期刊输出推荐分档与理由。
硬规则(违反即视为无效输出):
1. id 白名单:只能使用输入 candidates 中出现的 id;绝对禁止添加、替换或臆造列表之外的任何期刊。
2. 每个输入 id 恰好输出一次,不得遗漏、不得重复;期刊名称等元数据由系统填充,你只输出 id。
3. tier 只能取 challenge | fit | safe:
- challenge(挑战型):高影响因子、审稿较慢或竞争激烈的期刊
- fit(适配型):影响力与发表周期均衡、与论文契合的期刊
- safe(稳妥型):审稿较快、门槛较低、命中概率较高的期刊
4. match_reason:中文,1-2 句,结合摘要的研究主题与该刊 scope_desc 及指标(如有 similarity 一并参考),说明为何匹配/不匹配。
5. risk_note:中文,1-2 句,指出投稿风险(超出用户偏好、IF 门槛偏高、范围匹配偏弱、审稿周期长等);无明显风险写「无」。
6. 用户 preferences 为软约束:不满足偏好的期刊仍必须输出,并在 risk_note 中明确说明。
7. 不确定时宁可保守陈述,禁止编造统计数据(录用率、审稿天数等)。
输出:仅一个 JSON 对象,格式:
{"rankings":[{"id":"...","tier":"challenge","match_reason":"...","risk_note":"..."}]}
不要输出任何 JSON 之外的内容。"""
class RankError(Exception):
"""LLM 输出形态异常(无法按 JSON 解析/结构非法)。"""
@dataclass
class Ranking:
journal_id: str
tier: str
match_reason: str
risk_note: str
def rule_tier(journal: dict[str, Any]) -> str:
"""确定性分档规则(LLM 缺漏/不可用时兜底)。"""
inf = journal.get("impact_factor")
weeks = journal.get("review_time_weeks")
if isinstance(inf, (int, float)):
if inf >= _CHALLENGE_MIN_IF:
return "challenge"
if inf <= _SAFE_MAX_IF:
return "safe"
elif isinstance(weeks, (int, float)) and weeks <= 4:
return "safe"
return "fit"
def build_user_prompt(
abstract: str,
candidates: list[dict[str, Any]],
preferences: dict[str, Any],
) -> str:
"""候选 + 偏好 + 摘要 → 用户消息(JSON)。"""
payload = {
"abstract": abstract.strip()[:4000],
"preferences": preferences or {},
"candidates": [
{
"id": str(c["journal"]["id"]),
"name": str(c["journal"].get("name") or ""),
"impact_factor": c["journal"].get("impact_factor"),
"review_time_weeks": c["journal"].get("review_time_weeks"),
"acceptance_rate": c["journal"].get("acceptance_rate"),
"similarity": c.get("similarity"),
"scope_desc": str(c["journal"].get("scope_desc") or "")[:400],
}
for c in candidates
],
}
return json.dumps(payload, ensure_ascii=False)
def _normalize_tier(raw: Any) -> str | None:
tier = str(raw or "").strip().lower()
tier = _TIER_ALIASES.get(str(raw or "").strip(), tier)
return tier if tier in TIERS else None
def validate_rankings(
payload: Any, candidates: list[dict[str, Any]]
) -> tuple[list[Ranking], int, int]:
"""校验 LLM 输出。
返回 (rankings, dropped_hallucinated, repaired):
- 白名单外 id / 重复 id → 丢弃并计数(repaired 中重复不计);
- tier 非法、LLM 缺漏的 id → 按规则补位(repaired 计数)。
"""
if not isinstance(payload, dict) or not isinstance(payload.get("rankings"), list):
raise RankError("LLM 输出缺少 rankings 数组")
by_id = {str(c["journal"]["id"]): c["journal"] for c in candidates}
input_ids = list(by_id)
seen: set[str] = set()
rankings: list[Ranking] = []
dropped = 0
repaired = 0
for entry in payload["rankings"]:
if not isinstance(entry, dict):
dropped += 1
continue
jid = str(entry.get("id") or "").strip()
if jid in seen:
continue
if jid not in by_id:
dropped += 1 # 幻觉期刊:绝不进输出
continue
seen.add(jid)
tier = _normalize_tier(entry.get("tier"))
if tier is None:
tier = rule_tier(by_id[jid])
repaired += 1
reason = str(entry.get("match_reason") or "").strip() or "基于期刊范围与指标的匹配"
risk = str(entry.get("risk_note") or "").strip() or "无"
rankings.append(Ranking(journal_id=jid, tier=tier, match_reason=reason, risk_note=risk))
for jid in input_ids:
if jid in seen:
continue
rankings.append(
Ranking(
journal_id=jid,
tier=rule_tier(by_id[jid]),
match_reason="LLM 未给出该期刊条目,已按影响因子/周期规则补位",
risk_note="未提供(LLM 输出缺漏,规则兜底)",
)
)
repaired += 1
return rankings, dropped, repaired
async def rerank(
abstract: str,
candidates: list[dict[str, Any]],
preferences: dict[str, Any],
) -> tuple[list[Ranking], int, int]:
"""LLM 重排 + 校验。LLMError / RankError 上抛(路由降级为规则分档)。"""
settings = get_settings()
payload = await llm.complete_json(
SYSTEM_PROMPT,
build_user_prompt(abstract, candidates, preferences),
temperature=0.2,
max_tokens=3000,
timeout=settings.recommend_read_timeout,
)
return validate_rankings(payload, candidates)
def fallback_rankings(candidates: list[dict[str, Any]]) -> list[Ranking]:
"""LLM 不可用时的确定性分档(不丢任何候选期刊)。"""
rankings: list[Ranking] = []
for c in candidates:
j = c["journal"]
inf = j.get("impact_factor")
weeks = j.get("review_time_weeks")
rankings.append(
Ranking(
journal_id=str(j["id"]),
tier=rule_tier(j),
match_reason=(
f"规则分档:影响因子 {inf if inf is not None else '未收录'}、"
f"审稿约 {weeks if weeks is not None else '未收录'} 周"
),
risk_note="LLM 不可用,本条按影响因子/周期规则估算",
)
)
return rankings
+203
View File
@@ -0,0 +1,203 @@
"""services/llm.py —— LLM 调用(httpx 裸调 DeepSeek,模式同 paper_writer_service)。
- complete_text(system, user, *, temperature=0.3, max_tokens=8000, timeout=None) -> str
- complete_json(...)(response_format json_object + 一次 JSON 修复重试)
- 4 次尝试(1.5s / 3s / 6s 指数退避,429 / 5xx / 超时)
- env 键(优先 REVIEW_LLM_*,回退 DOCGEN_LLM_*,再回退内置默认):
*_LLM_BASE_URL / *_LLM_API_KEY / *_LLM_MODEL / *_LLM_TIMEOUT
"""
from __future__ import annotations
import asyncio
import json
import logging
import os
from typing import Any
import httpx
logger = logging.getLogger(__name__)
DEFAULT_LLM_BASE_URL = "https://api.deepseek.com/v1"
DEFAULT_LLM_MODEL = "deepseek-chat"
DEFAULT_LLM_TIMEOUT = 300.0
BACKOFF_SECONDS = (1.5, 3.0, 6.0)
MAX_ATTEMPTS = 4
class LLMError(Exception):
"""LLM 调用失败(重试尽/响应异常)。"""
def _env(keys: tuple[str, ...], default: str = "") -> str:
for key in keys:
val = os.environ.get(key)
if val:
return val
return default
def _llm_config() -> dict[str, Any]:
return {
"base_url": _env(
("REVIEW_LLM_BASE_URL", "DOCGEN_LLM_BASE_URL"), DEFAULT_LLM_BASE_URL
).rstrip("/"),
"api_key": _env(("REVIEW_LLM_API_KEY", "DOCGEN_LLM_API_KEY"), ""),
"model": _env(("REVIEW_LLM_MODEL", "DOCGEN_LLM_MODEL"), DEFAULT_LLM_MODEL),
"timeout": float(
_env(("REVIEW_LLM_TIMEOUT", "DOCGEN_LLM_TIMEOUT"), str(DEFAULT_LLM_TIMEOUT))
),
}
def _extract_text(payload: dict[str, Any]) -> str:
try:
choices = payload.get("choices") or []
if choices:
content = choices[0].get("message", {}).get("content")
if isinstance(content, str):
return content
if isinstance(content, list):
return "".join(
str(c.get("text", "")) if isinstance(c, dict) else str(c) for c in content
)
except Exception:
pass
return ""
async def _chat_once(
client: httpx.AsyncClient,
cfg: dict[str, Any],
system: str,
user: str,
*,
temperature: float,
max_tokens: int,
json_mode: bool,
timeout: float | None,
) -> str:
headers = {"Content-Type": "application/json"}
if cfg["api_key"]:
headers["Authorization"] = f"Bearer {cfg['api_key']}"
body: dict[str, Any] = {
"model": cfg["model"],
"messages": [
{"role": "system", "content": system},
{"role": "user", "content": user},
],
"temperature": temperature,
"max_tokens": max_tokens,
}
if json_mode:
body["response_format"] = {"type": "json_object"}
resp = await client.post(
f"{cfg['base_url']}/chat/completions",
headers=headers,
json=body,
timeout=timeout if timeout is not None else cfg["timeout"],
)
resp.raise_for_status()
payload = resp.json()
if not isinstance(payload, dict):
raise LLMError("LLM 响应形态异常(非 JSON 对象)")
text = _extract_text(payload)
if not text:
raise LLMError("LLM 响应缺少 choices[0].message.content")
return text
async def _complete(
system: str,
user: str,
*,
temperature: float,
max_tokens: int,
json_mode: bool,
timeout: float | None,
) -> str:
cfg = _llm_config()
last_exc: BaseException | None = None
async with httpx.AsyncClient() as client:
for attempt in range(MAX_ATTEMPTS):
try:
return await _chat_once(
client, cfg, system, user,
temperature=temperature, max_tokens=max_tokens, json_mode=json_mode,
timeout=timeout,
)
except LLMError:
raise
except (httpx.HTTPStatusError, httpx.HTTPError, httpx.TimeoutException) as exc:
last_exc = exc
status = getattr(getattr(exc, "response", None), "status_code", None)
retryable = status is None or status == 429 or status >= 500
if not retryable or attempt == MAX_ATTEMPTS - 1:
raise LLMError(f"LLM 调用失败(HTTP {status or '网络'}):{exc}") from exc
await asyncio.sleep(BACKOFF_SECONDS[min(attempt, len(BACKOFF_SECONDS) - 1)])
raise LLMError(f"LLM 调用失败:{last_exc}")
async def complete_text(
system: str,
user: str,
*,
temperature: float = 0.3,
max_tokens: int = 8000,
timeout: float | None = None,
) -> str:
"""文本补全。失败抛 LLMError。"""
return await _complete(
system, user, temperature=temperature, max_tokens=max_tokens, json_mode=False,
timeout=timeout,
)
def _repair_json(text: str) -> str:
text = text.strip()
if text.startswith("```"):
text = text.strip("`")
if text.lower().startswith("json"):
text = text[4:]
text = text.strip()
if not text:
raise LLMError("LLM 返回空内容")
if text.startswith("{"):
return text
start = text.find("{")
end = text.rfind("}")
if start >= 0 and end > start:
return text[start:end + 1]
raise LLMError("LLM 输出中未找到 JSON 对象")
async def complete_json(
system: str,
user: str,
*,
temperature: float = 0.3,
max_tokens: int = 8000,
timeout: float | None = None,
) -> dict[str, Any]:
"""JSON 补全(json_object + 一次修复重试)。"""
text = await _complete(
system, user, temperature=temperature, max_tokens=max_tokens, json_mode=True,
timeout=timeout,
)
try:
return json.loads(text)
except ValueError:
pass
text2 = await _complete(
system + "\n(重要:你的回答必须是单个合法 JSON 对象,不要输出其他任何内容)",
user,
temperature=temperature,
max_tokens=max_tokens,
json_mode=True,
timeout=timeout,
)
try:
return json.loads(text2)
except ValueError:
repaired = _repair_json(text2)
return json.loads(repaired)
+139
View File
@@ -0,0 +1,139 @@
"""paper_parse.py —— PDF → 纯文本(经 vstore 8094 现有解析接口,零改动复用)。
链路(全部只读消费 vstore 既有 API,不修改 vstore 代码):
1. POST {vstore}/api/vstore/ingest multipart file(vstore 内部 pymupdf 解析)
2. GET {vstore}/api/vstore/documents/{doc_id} → document.paragraphs 拼接纯文本
3. DELETE {vstore}/api/vstore/documents/{doc_id} (临时解析件 best-effort 清理,
REVIEW_PARSE_DELETE_AFTER=0 可关闭——保留文档便于人工排查)
为何不用 methodology 8092:其 /extract-upload 是 MinerU 分钟级异步 job
(MinerU 子进程 + LLM 抽取两阶段),不适配同步自审稿;vstore 段落解析秒级
返回,满足 presubmit 场景。methodology 地址在 config 中预留供后续扩展。
"""
from __future__ import annotations
import logging
from typing import Any
import httpx
from ..config import get_settings
logger = logging.getLogger(__name__)
class ParseError(Exception):
"""PDF 解析失败(上游不可达/响应异常/空文本)。"""
async def parse_pdf(path: str) -> str:
"""接收 PDF 文件路径,返回拼接后的纯文本。失败抛 ParseError。"""
try:
with open(path, "rb") as fh:
file_bytes = fh.read()
except OSError as exc:
raise ParseError(f"读取 PDF 失败:{exc}") from exc
return await parse_pdf_bytes(file_bytes, filename=str(path).rsplit("/", 1)[-1] or "paper.pdf")
async def parse_pdf_bytes(file_bytes: bytes, filename: str = "paper.pdf") -> str:
"""接收 PDF 字节流,返回拼接后的纯文本。失败抛 ParseError。"""
if not file_bytes:
raise ParseError("PDF 内容为空")
settings = get_settings()
base = settings.vstore_base
timeout = httpx.Timeout(
connect=5.0, read=settings.parse_timeout, write=settings.parse_timeout, pool=5.0
)
async with httpx.AsyncClient(timeout=timeout) as client:
# ---- 1) ingest(vstore 内部 pymupdf 解析 + 嵌入 + 加密落盘)----
try:
resp = await client.post(
f"{base}/api/vstore/ingest",
files={"file": (filename, file_bytes, "application/pdf")},
data={
"name": filename,
"category": "review_presubmit",
"description": "review_service 临时解析件",
},
)
except httpx.HTTPError as exc:
raise ParseError(f"vstore 不可达({base}):{exc}") from exc
if resp.status_code >= 400:
# 读上游错误体(信封 JSON 或裸文本),区分“不可达”与“上游报错”
raise ParseError(f"vstore ingest 失败(HTTP {resp.status_code}):{_err_body(resp)}")
try:
payload = resp.json()
except ValueError as exc:
raise ParseError(f"vstore ingest 响应非 JSON:{exc}") from exc
if not isinstance(payload, dict) or not payload.get("ok"):
message = _err_message(payload) or "vstore ingest 失败"
raise ParseError(message)
doc_id = str((payload.get("item") or {}).get("document_id") or "")
if not doc_id:
raise ParseError("vstore ingest 响应缺少 document_id")
# ---- 2) 取明文全文(paragraphs 拼接)+ 3) 清理临时件 ----
try:
text = await _fetch_text(client, base, doc_id)
finally:
if settings.parse_delete_after:
await _best_effort_delete(client, base, doc_id)
text = text.strip()
if not text:
raise ParseError("PDF 解析结果为空(可能是扫描件/图片型 PDF,vstore 无 OCR)")
return text
async def _fetch_text(client: httpx.AsyncClient, base: str, doc_id: str) -> str:
"""取 vstore 文档明文并按段落拼接纯文本。"""
try:
resp = await client.get(f"{base}/api/vstore/documents/{doc_id}")
except httpx.HTTPError as exc:
raise ParseError(f"vstore 取文档失败({doc_id}):{exc}") from exc
if resp.status_code >= 400:
raise ParseError(
f"vstore 取文档失败({doc_id},HTTP {resp.status_code}):{_err_body(resp)}"
)
try:
payload = resp.json()
except ValueError as exc:
raise ParseError(f"vstore 取文档响应非 JSON({doc_id}):{exc}") from exc
if not isinstance(payload, dict) or not payload.get("ok"):
raise ParseError(_err_message(payload) or f"vstore 取文档失败({doc_id})")
document: dict[str, Any] = payload.get("document") or {}
paragraphs = document.get("paragraphs") or []
return "\n".join(str(p) for p in paragraphs if str(p).strip())
async def _best_effort_delete(client: httpx.AsyncClient, base: str, doc_id: str) -> None:
try:
await client.delete(f"{base}/api/vstore/documents/{doc_id}")
except Exception: # noqa: BLE001 - 清理失败不阻断(文档级 secure_delete 已尽力)
logger.warning("paper_parse: vstore 临时文档清理失败(%s)", doc_id, exc_info=True)
def _err_message(payload: Any) -> str:
if isinstance(payload, dict):
err = payload.get("error")
if isinstance(err, dict):
return str(err.get("message") or "")
if isinstance(err, str):
return err
return ""
def _err_body(resp: httpx.Response) -> str:
"""上游错误体:优先 JSON 信封 message,回退裸文本(截断 500 字符)。"""
try:
payload = resp.json()
except ValueError:
return (resp.text or "").strip()[:500] or "(空响应体)"
return _err_message(payload) or str(payload)[:500]
+155
View File
@@ -0,0 +1,155 @@
"""services/self_review.py —— 植物基因组学专属自审稿(System Prompt + 报告模型)。
六大领域维度检查清单硬编码在 SYSTEM_PROMPT(发给 DeepSeek):
1. 基因组组装质量(N50/L50/BUSCO 完整格式/Hi-C 挂载率,缺项即 Major)
2. 基因注释可靠性(MAKER/AUGUSTUS 流程、RNA-seq 证据、基因数与已知范围一致性)
3. 比较基因组学与进化(Bootstrap/后验概率、OrthoFinder 等直系同源策略)
4. 数据可用性合规(SRA/BioProject + 组装/GFF3 Accession,缺失即 Major)
5. 统计与实验设计(GWAS/QTL 样本量·多重校正·群体结构;转录组生物学重复)
6. 伦理与合规·植物特异性(凭证标本 voucher、ABS 惠益分享声明)
Major/Minor 每条必须给 Evidence(论文章节位置或具体指标数值),由 coerce
宽容归一化(缺失补空串,不丢条目)。
"""
from __future__ import annotations
from pydantic import BaseModel, Field
RECOMMENDATIONS = ("accept", "minor_revision", "major_revision", "reject")
SYSTEM_PROMPT = """你是一位植物基因组学领域资深审稿人,长期为 Plant Cell、Nature Plants、
Genome Biology、Molecular Plant、Plant Journal 等期刊审稿。请对用户提供的论文正文做
投稿前自审(pre-submission self review),严格按以下六个植物基因组学专属维度逐项检查,
输出结构化审稿意见。所有条目使用中文撰写(专业术语、指标名、工具名保留英文)。
【维度一:基因组组装质量】必查指标:
- contig N50 / scaffold N50 / L50(报了哪个要写明哪个)
- BUSCO 完整度:必须按完整格式报告——例如
"Complete 95.3% [S:93.1%, D:2.2%], Missing 4.7%, n=1614 (embryophyta_odb10)"
(含 Complete/Single-copy/Duplicated/Missing 四项与数据集版本、n 值)
- Hi-C 挂载率(anchoring rate,chromosome-level 比例)及挂载到染色体的 scaffold 比例
判定规则:若论文只报 N50 而未报 BUSCO 完整度或 Hi-C 挂载率(组装类论文),直接列为
Major Weakness。同时检查组装是否为 pseudo-chromosome 级别、污染评估(如 k-mer/coverage
检查)与 gap 比例。
【维度二:基因注释可靠性】检查:
- 是否使用标准注释流程(MAKER、AUGUSTUS、BRAKER、GeMoMa、EVidenceModeler 等),参数与版本是否交代
- 是否有转录组证据支持:RNA-seq(组织/时期/物种内或近缘种)比对支撑比例,或蛋白同源证据
- 基因数量是否与该物种已知基因数范围一致(如水稻 ~37-40k、拟南芥 ~27k、玉米 ~39-45k、
小麦 ~107k、大豆 ~46k;近缘种偏差过大须说明);有无转座元件基因误注释的排查
- 注释质量评估:BRAKER/AED 分值、BUSCO 在蛋白集上的复核
【维度三:比较基因组学与进化】检查:
- 系统发育树:每个节点必须标注 Bootstrap 支持率(建议 ≥1000 次重复)或贝叶斯后验概率;
未标注或仅"高/低支持"的描述 → Major 或 Minor(视严重程度)
- 直系同源基因鉴定策略是否说明清楚:OrthoFinder/OrthoMCL/RBX 等工具、版本、参数、
聚类阈值(inflation/coverage)、single-copy 标记基因的筛选方式
- 分子进化分析是否规范:dN/dS 检验方法与多重校正、共线性(synteny)分析参数、
分化时间估算是否校准(化石/二次校准点)
【维度四:数据可用性合规——极其重要】必查:
- 原始测序数据是否上传 NCBI SRA / BioProject(或 ENA/DDBJ),是否给出 Accession 号
(PRJNA/PRJEB/ SRP 开头)
- 基因组组装是否有 GenBank/ENA accession(GCA_/GCF_),GFF3 注释文件、蛋白/CDS 文件
是否有可公开获取的位置(仓库 accession 或补充数据链接)
- 转录组数据是否同样有 SRA accession
判定规则:组装/注释类论文若缺失上述任何一环,直接列为 Major Weakness。
【维度五:统计与实验设计】检查:
- GWAS/QTL 定位:样本量/群体大小是否充分;多重检验校正是否明确(Bonferroni/FDR/
permutation 阈值/PCA 显著性水平);群体结构控制(PCA、亲缘关系矩阵 K、分层分析)
是否交代;表型测量描述是否充分
- 转录组分析:生物学重复数量(≥3 为底线)、差异分析工具(DESeq2/edgeR 等)与
校正方法、批次效应处理、|log2FC| 与 padj 阈值
- 其他统计:t 检验/ANOVA 是否满足前提、重复类型(生物学 vs 技术重复)写明与否
【维度六:伦理与合规(植物特异性)】检查:
- 野生植物材料:是否提供凭证标本(voucher specimen)信息——标本馆编号、存放机构、
采集地点(GPS)与采集许可;无凭证标本的野生资源取样 → Major 或 Minor(视用途)
- 遗传资源获取与惠益分享(ABS)合规性:是否声明符合《名古屋议定书》/本国遗传资源
获取条例(如中国《生物安全法》/《种子法》相关条款)、是否说明材料来源与授权
- 农业转基因/基因编辑材料的合规声明(若涉及)
输出要求(必须输出单个 JSON 对象,不要输出其他任何内容):
{
"summary": "总体评价(3-5 句,须点明论文类型与所适用的维度)",
"major_weaknesses": [{"title": "...", "detail": "...", "evidence": "...", "suggestion": "..."}],
"minor_weaknesses": [{"title": "...", "detail": "...", "evidence": "...", "suggestion": "..."}],
"presentation_format": [{"title": "...", "detail": "...", "evidence": "...", "suggestion": "..."}],
"questions_for_authors": ["..."],
"recommendation": "major_revision"
}
硬性规则:
1. major_weaknesses 3-8 条、minor_weaknesses 3-8 条;上述维度中命中的硬性判定规则
(组装缺 BUSCO/挂载率、缺数据 Accession)必须出现。
2. **每条 major_weaknesses 的 evidence 字段必填**:必须引用论文的具体章节位置
(如 "Methods 2.1"、"Table 2"、"Results 第 3 段"、"Data Availability 一句话未见")
或具体指标数值(如 "仅报 scaffold N50=42.6 Mb,未提供 BUSCO 与 Hi-C 挂载率")。
evidence 严禁编造:引用不到具体位置时,写明"正文未见相关报告"。
3. minor_weaknesses/presentation_format 的 evidence 尽量给出,无法定位时可留空字符串。
4. 论文不涉及的维度(如纯 RNA-seq 论文不谈组装)在 summary 中说明并跳过,不硬凑条目。
5. recommendation 只能是 accept / minor_revision / major_revision / reject 之一,
缺失关键合规项(维度四)时通常不高于 major_revision。
6. presentation_format 覆盖:图表规范(图注含单位/误差棒说明)、参考文献格式、
字数与结构、英文语法(若投英文刊)。
"""
class WeaknessItem(BaseModel):
title: str
detail: str = ""
evidence: str = ""
suggestion: str = ""
class SelfReviewReport(BaseModel):
summary: str = ""
major_weaknesses: list[WeaknessItem] = Field(default_factory=list)
minor_weaknesses: list[WeaknessItem] = Field(default_factory=list)
presentation_format: list[WeaknessItem] = Field(default_factory=list)
questions_for_authors: list[str] = Field(default_factory=list)
recommendation: str = "major_revision"
def build_user_prompt(text: str, target_journal: str) -> str:
header = f"目标期刊:{target_journal}\n\n" if target_journal else ""
return f"{header}论文正文:\n\n{text}"
def _coerce_items(raw: object) -> list[WeaknessItem]:
items: list[WeaknessItem] = []
if not isinstance(raw, list):
return items
for entry in raw:
if isinstance(entry, str) and entry.strip():
items.append(WeaknessItem(title=entry.strip()))
elif isinstance(entry, dict):
title = str(entry.get("title") or "").strip()
detail = str(entry.get("detail") or "").strip()
if not title and not detail:
continue
items.append(WeaknessItem(
title=title or "(未命名)",
detail=detail,
evidence=str(entry.get("evidence") or "").strip(),
suggestion=str(entry.get("suggestion") or "").strip(),
))
return items
def coerce_report(payload: dict) -> SelfReviewReport:
questions = payload.get("questions_for_authors") or []
if not isinstance(questions, list):
questions = []
recommendation = str(payload.get("recommendation") or "").strip().lower()
if recommendation not in RECOMMENDATIONS:
recommendation = "major_revision"
return SelfReviewReport(
summary=str(payload.get("summary") or "").strip(),
major_weaknesses=_coerce_items(payload.get("major_weaknesses")),
minor_weaknesses=_coerce_items(payload.get("minor_weaknesses")),
presentation_format=_coerce_items(payload.get("presentation_format")),
questions_for_authors=[str(q).strip() for q in questions if str(q).strip()],
recommendation=recommendation,
)
@@ -0,0 +1,62 @@
{
"version": 1,
"updated": "2026-10-09",
"notes": "植物基因组学领域种子期刊。acceptance_rate 无公开权威数据一律为 null(不编造);impact_factor / review_time_weeks 为公开 JCR/期刊官网常识近似值,以 last_updated 标注刷新时间,接入真实数据源前需人工核验。scope_desc 为英文范围描述,供 vstore 向量匹配使用。",
"journals": [
{
"id": "plant-cell",
"name": "The Plant Cell",
"scope_desc": "Broad-interest plant biology with emphasis on molecular, cellular and physiological mechanisms, including genomics, systems biology, gene regulation, signaling, and plant-microbe interactions. Publishes genome-scale and functional genomics studies with strong mechanistic insight.",
"impact_factor": 11.6,
"review_time_weeks": 6.0,
"acceptance_rate": null,
"submission_deadline": null,
"official_url": "https://academic.oup.com/plcell",
"last_updated": "2026-10-09"
},
{
"id": "genome-biology",
"name": "Genome Biology",
"scope_desc": "High-impact open-access journal for genome-scale research across all organisms, including genome assembly and annotation, comparative and evolutionary genomics, computational methods, single-cell and multi-omics studies, and reproducible bioinformatics.",
"impact_factor": 10.1,
"review_time_weeks": 4.0,
"acceptance_rate": null,
"submission_deadline": null,
"official_url": "https://genomebiology.biomedcentral.com",
"last_updated": "2026-10-09"
},
{
"id": "plant-journal",
"name": "The Plant Journal",
"scope_desc": "Molecular and physiological plant sciences, including genome biology and genomics, gene function, cell and developmental biology, stress and environmental responses, and plant biotechnology; favors experimental studies with clear mechanistic advance.",
"impact_factor": 7.2,
"review_time_weeks": 5.0,
"acceptance_rate": null,
"submission_deadline": null,
"official_url": "https://onlinelibrary.wiley.com/journal/1365313X",
"last_updated": "2026-10-09"
},
{
"id": "theoretical-applied-genetics",
"name": "Theoretical and Applied Genetics",
"scope_desc": "Plant breeding and genetics research combining theoretical and applied approaches, including quantitative genetics, QTL mapping, genome-wide association studies, genomic selection, population genetics, and molecular marker applications for crop improvement.",
"impact_factor": 5.3,
"review_time_weeks": 5.0,
"acceptance_rate": null,
"submission_deadline": null,
"official_url": "https://link.springer.com/journal/338",
"last_updated": "2026-10-09"
},
{
"id": "plant-genome",
"name": "The Plant Genome",
"scope_desc": "Plant and crop genome research with practical breeding applications, including genome assembly and annotation, structural variation, GWAS and QTL analysis, marker-assisted and genomic selection, and translational genomics for crop improvement.",
"impact_factor": 3.3,
"review_time_weeks": 4.0,
"acceptance_rate": null,
"submission_deadline": null,
"official_url": "https://acsess.onlinelibrary.wiley.com/journal/1940-3372",
"last_updated": "2026-10-09"
}
]
}
+6
View File
@@ -0,0 +1,6 @@
[pytest]
python_files = test_*.py
pythonpath = .
markers =
review_unit: 审稿服务单元测试
asyncio_mode = auto
+33
View File
@@ -0,0 +1,33 @@
#!/usr/bin/env bash
# restart.sh —— REVIEW 服务重启脚本(8100;不触碰生产服务 8090/8091/8092/8093/8094/8097/8098)。
# 用法: restart.sh [python] [port]
set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
# 默认 venv:本机统一 venv 固定路径(原先 $REPO_ROOT/../ 推导在本工作区不存在)
PYTHON="${1:-/home/xzq/5090/teamWorkspace/zh/plantai-rag-new-venv/bin/python}"
PORT="${2:-8100}"
HOST="${REVIEW_BIND_HOST:-127.0.0.1}"
[[ -x "$PYTHON" ]] || { echo "Python not found: $PYTHON" >&2; exit 2; }
APP_MODULE="app.main:app"
if command -v lsof >/dev/null 2>&1; then
PID="$(lsof -ti tcp:"$PORT" -sTCP:LISTEN 2>/dev/null || true)"
if [[ -n "$PID" ]]; then
echo "[restart] 停止 $PORT 旧进程 pid=$PID"
kill $PID 2>/dev/null || true
sleep 2
fi
fi
cd "$SCRIPT_DIR"
# 加载运行时配置(.env 不入 git):services/llm.py 直接读 REVIEW_LLM_*/DOCGEN_LLM_*
# 裸键,paper_parse.py 读 REVIEW_*——需 set -a 显式导出(同 paper_writer_service 先例)。
set -a
[ -f .env ] && source .env
set +a
echo "[restart] 启动审稿服务 :$PORT($PYTHON)"
nohup "$PYTHON" -m uvicorn "$APP_MODULE" --host "$HOST" --port "$PORT" \
> "$SCRIPT_DIR/review.log" 2>&1 &
echo "[restart] 完成;日志 review_service/review.log"
+163
View File
@@ -0,0 +1,163 @@
"""tests/test_presubmit.py —— presubmit 离线单元测试(mock 解析与 LLM,不依赖上游)。
覆盖:非 PDF 校验 / 报告归一化 / LLM 失败降级 / 解析失败降级 / 非法 recommendation 回退。
运行: cd review_service && <venv>/bin/python -m pytest -q
"""
from __future__ import annotations
import pytest
from fastapi.testclient import TestClient
from app.main import app
from app.services import llm, paper_parse
FAKE_PDF = b"%PDF-1.4 fake-bytes-for-test"
LLM_PAYLOAD = {
"summary": "研究问题清晰,但统计与证据链存在重大缺口。",
"major_weaknesses": [
{
"title": "统计方法不当",
"detail": "未做多重检验校正",
"evidence": "Methods 2.1 仅报 p<0.05,未见 FDR/Bonferroni",
"suggestion": "补充 FDR 校正",
},
"纯字符串条目", # 宽容归一化:字符串 → WeaknessItem(title=...)
],
# minor 故意不带 evidence 字段 → 归一化应补空串
"minor_weaknesses": [{"detail": "缺少缩写表", "title": "缩写"}],
"presentation_format": [{"title": "图注不完整", "detail": "Fig.2 无单位", "suggestion": "补单位"}],
"questions_for_authors": ["群体结构如何校正?", ""],
"recommendation": "MAJOR_REVISION", # 应归一化为小写
}
@pytest.fixture()
def client() -> TestClient:
return TestClient(app)
@pytest.fixture(autouse=True)
def _mock_upstreams(monkeypatch):
async def fake_parse(file_bytes: bytes, filename: str = "paper.pdf") -> str:
assert file_bytes == FAKE_PDF
return "论文正文……" * 10
async def fake_llm_json(system: str, user: str, **kwargs):
assert "论文正文" in user
return dict(LLM_PAYLOAD)
async def boom_llm(*args, **kwargs):
raise llm.LLMError("余额不足")
async def boom_parse(*args, **kwargs):
raise paper_parse.ParseError("PDF 解析结果为空")
monkeypatch.setattr(paper_parse, "parse_pdf_bytes", fake_parse)
monkeypatch.setattr(llm, "complete_json", fake_llm_json)
yield
# monkeypatch 自动还原
def _post(client: TestClient, filename: str, content: bytes, journal: str = ""):
data = {"target_journal": journal} if journal else {}
return client.post(
"/api/review/presubmit",
files={"file": (filename, content, "application/pdf")},
data=data,
)
def test_non_pdf_rejected(client: TestClient):
r = client.post(
"/api/review/presubmit",
files={"file": ("a.txt", b"x", "text/plain")},
)
body = r.json()
assert body["ok"] is False
assert body["error"]["code"] == "VALIDATION"
def test_empty_file_rejected(client: TestClient):
body = _post(client, "p.pdf", b"").json()
assert body["ok"] is False
assert body["error"]["code"] == "VALIDATION"
def test_normal_report_normalized(client: TestClient):
body = _post(client, "p.pdf", FAKE_PDF, journal="Plant Cell").json()
assert body["ok"] is True
report = body["report"]
# recommendation 大写 → 小写
assert report["recommendation"] == "major_revision"
# 字符串条目宽容归一化
assert len(report["major_weaknesses"]) == 2
assert report["major_weaknesses"][1]["title"] == "纯字符串条目"
# Evidence 字段:有则透传,字符串条目/缺失则为空串
assert report["major_weaknesses"][0]["evidence"] == "Methods 2.1 仅报 p<0.05,未见 FDR/Bonferroni"
assert report["major_weaknesses"][1]["evidence"] == ""
assert report["minor_weaknesses"][0]["evidence"] == ""
# 空问题剔除
assert report["questions_for_authors"] == ["群体结构如何校正?"]
assert report["minor_weaknesses"] and report["presentation_format"] and report["summary"]
# meta 透传
assert body["meta"]["target_journal"] == "Plant Cell"
assert body["meta"]["text_chars"] > 0
assert body["meta"]["truncated"] is False
def test_llm_failure_degrades(client: TestClient, monkeypatch):
async def boom(*args, **kwargs):
raise llm.LLMError("余额不足")
monkeypatch.setattr(llm, "complete_json", boom)
body = _post(client, "p.pdf", FAKE_PDF).json()
assert body["ok"] is False
assert body["error"]["code"] == "UPSTREAM_DOWN"
assert body["error"]["degraded"] is True
def test_parse_failure_degrades(client: TestClient, monkeypatch):
async def boom(*args, **kwargs):
raise paper_parse.ParseError("PDF 解析结果为空")
monkeypatch.setattr(paper_parse, "parse_pdf_bytes", boom)
body = _post(client, "p.pdf", FAKE_PDF).json()
assert body["ok"] is False
assert body["error"]["code"] == "PARSE_FAILED"
assert body["error"]["degraded"] is True
def test_invalid_recommendation_falls_back(client: TestClient, monkeypatch):
async def weird(*args, **kwargs):
return {"summary": "x", "recommendation": "strong_accept", "major_weaknesses": "not-a-list"}
monkeypatch.setattr(llm, "complete_json", weird)
body = _post(client, "p.pdf", FAKE_PDF).json()
assert body["ok"] is True
assert body["report"]["recommendation"] == "major_revision"
assert body["report"]["major_weaknesses"] == []
def test_domain_system_prompt_wired(client: TestClient, monkeypatch):
"""植物基因组学六维清单 + Evidence 规则必须注入 System Prompt。"""
captured: dict[str, str] = {}
async def spy(system: str, user: str, **kwargs):
captured["system"] = system
return dict(LLM_PAYLOAD)
monkeypatch.setattr(llm, "complete_json", spy)
body = _post(client, "p.pdf", FAKE_PDF, journal="Plant Cell").json()
assert body["ok"] is True
system = captured["system"]
for marker in (
"BUSCO", "N50", "L50", "Hi-C", "挂载率", # 维度1 组装质量
"MAKER", "AUGUSTUS", "RNA-seq", # 维度2 注释可靠性
"Bootstrap", "OrthoFinder", # 维度3 比较基因组
"SRA", "BioProject", "GFF3", "accession", # 维度4 数据合规
"GWAS", "多重检验校正", "生物学重复", # 维度5 统计设计
"凭证标本", "ABS", "惠益分享", # 维度6 伦理合规
"evidence", "章节", # Evidence 硬性要求
):
assert marker in system, f"System Prompt 缺少领域标记: {marker}"
+271
View File
@@ -0,0 +1,271 @@
"""tests/test_recommend.py —— /api/review/recommend 离线单元测试。
mock vstore 向量匹配(journal_kb.match)与 LLM(llm.complete_json),不依赖上游。
核心回归:LLM 幻觉期刊名绝不进输出(白名单 id 硬校验 + 名称一律取 DB)。
运行: cd review_service && <venv>/bin/python -m pytest -q
"""
from __future__ import annotations
import asyncio
import pytest
from fastapi.testclient import TestClient
from app import db as db_mod
from app.main import app
from app.services import journal_kb, journal_rank, llm
ABSTRACT = (
"We report a chromosome-level genome assembly of wild rice Oryza rufipogon "
"using PacBio HiFi and Hi-C sequencing, annotate 38,412 protein-coding genes "
"with RNA-seq evidence, and perform comparative genomics and GWAS analyses "
"to reveal the evolutionary basis of agronomic traits in rice."
)
SEED_NAMES = {
"The Plant Cell",
"Genome Biology",
"The Plant Journal",
"Theoretical and Applied Genetics",
"The Plant Genome",
}
# 覆盖全部 5 本种子期刊的合法重排结果
ALL_RANKINGS = [
{"id": "plant-cell", "tier": "challenge", "match_reason": "功能基因组机制研究契合", "risk_note": "竞争激烈"},
{"id": "genome-biology", "tier": "challenge", "match_reason": "组装与比较基因组学范围契合", "risk_note": "IF 门槛高"},
{"id": "plant-journal", "tier": "fit", "match_reason": "植物分子生物学范围契合", "risk_note": "无"},
{"id": "theoretical-applied-genetics", "tier": "fit", "match_reason": "GWAS/驯化分析契合", "risk_note": "偏重育种"},
{"id": "plant-genome", "tier": "safe", "match_reason": "作物基因组组装契合", "risk_note": "IF 较低"},
]
def make_fake_match(scores: dict[str, float] | None = None, fail: bool = False):
async def fake(abstract: str, journals: list[dict]):
if fail:
raise journal_kb.MatchError("vstore 不可达(mock)")
out = [
{"journal": j, "similarity": (scores or {}).get(str(j["id"]), 0.5)}
for j in journals
]
return out, True
return fake
def make_fake_llm(rankings: list[dict] | None = None, fail: bool = False, payload=None):
async def fake(system: str, user: str, **kwargs):
if fail:
raise llm.LLMError("余额不足(mock)")
if payload is not None:
return payload
return {"rankings": rankings if rankings is not None else []}
return fake
@pytest.fixture()
def client(tmp_path, monkeypatch):
db_mod.set_path(tmp_path / "review_test.db") # 进 lifespan 前注入 tmp 路径
monkeypatch.setattr(journal_kb, "match", make_fake_match())
monkeypatch.setattr(llm, "complete_json", make_fake_llm(ALL_RANKINGS))
with TestClient(app) as c: # CM 触发 lifespan:init_db(注入路径) + 收尾 close
yield c
def _all_items(body: dict) -> list[dict]:
items: list[dict] = []
for tier in ("challenge", "fit", "safe"):
items.extend(body["tiers"][tier]["items"])
return items
def test_recommend_happy_path(client: TestClient):
r = client.post("/api/review/recommend", json={"abstract": ABSTRACT})
assert r.status_code == 200
body = r.json()
assert body["ok"] is True
assert set(body["tiers"]) == {"challenge", "fit", "safe"}
assert "挑战型" in body["tiers"]["challenge"]["label"]
assert "适配型" in body["tiers"]["fit"]["label"]
assert "稳妥型" in body["tiers"]["safe"]["label"]
items = _all_items(body)
assert len(items) == 5
assert {it["name"] for it in items} == SEED_NAMES
for it in items:
assert it["match_reason"], f"缺少匹配理由:{it['name']}"
assert it["risk_note"], f"缺少风险提示:{it['name']}"
assert it["similarity"] == 0.5
# 三档内容与 LLM 输出一致
assert [it["name"] for it in body["tiers"]["challenge"]["items"]][0] == "The Plant Cell"
assert body["tiers"]["safe"]["items"][0]["name"] == "The Plant Genome"
meta = body["meta"]
assert meta["llm_ok"] is True and meta["vector_ok"] is True
assert meta["dropped_hallucinated"] == 0
assert meta["seed_count"] == 5
def test_seed_has_no_fabricated_acceptance_rate(client: TestClient):
"""种子诚实性:无公开数据的 acceptance_rate 必须为 null。"""
body = client.post("/api/review/recommend", json={"abstract": ABSTRACT}).json()
for it in _all_items(body):
assert it["acceptance_rate"] is None, f"编造了录用率:{it['name']}"
def test_hallucinated_journal_dropped(client: TestClient, monkeypatch):
"""LLM 幻觉期刊(白名单外 id)绝不进输出;缺漏期刊规则补位。"""
hallucinated = {
"id": "nature-genetics",
"tier": "challenge",
"match_reason": "幻觉期刊,输入列表中并不存在",
"risk_note": "x",
}
monkeypatch.setattr(
llm,
"complete_json",
make_fake_llm([hallucinated, ALL_RANKINGS[1]]), # 1 幻觉 + 1 真实,其余缺漏
)
body = client.post("/api/review/recommend", json={"abstract": ABSTRACT}).json()
assert body["ok"] is True
items = _all_items(body)
raw = str(body)
assert "nature-genetics" not in raw and "Nature Genetics" not in raw
assert "幻觉期刊" not in raw # 幻觉条目的理由同样被丢弃
assert {it["name"] for it in items} <= SEED_NAMES
assert len(items) == 5 # 4 本缺漏由规则补位,不丢候选
assert body["meta"]["dropped_hallucinated"] == 1
assert body["meta"]["repaired"] == 4
def test_llm_name_tamper_ignored(client: TestClient, monkeypatch):
"""LLM 篡改期刊名不采信:输出名称一律取 DB。"""
tampered = [
{"id": "plant-cell", "tier": "fit", "name": "Fake Journal X",
"match_reason": "被篡改的名称", "risk_note": "无"}
]
monkeypatch.setattr(llm, "complete_json", make_fake_llm(tampered))
body = client.post("/api/review/recommend", json={"abstract": ABSTRACT}).json()
assert body["ok"] is True
raw = str(body)
assert "Fake Journal X" not in raw
plant_cell = [it for it in _all_items(body) if it["id"] == "plant-cell"]
assert plant_cell and plant_cell[0]["name"] == "The Plant Cell"
def test_llm_failure_fallback(client: TestClient, monkeypatch):
"""LLM 失败 → 规则分档,推荐仍可用(永不 5xx)。"""
monkeypatch.setattr(llm, "complete_json", make_fake_llm(fail=True))
body = client.post("/api/review/recommend", json={"abstract": ABSTRACT}).json()
assert body["ok"] is True
assert body["meta"]["llm_ok"] is False
items = _all_items(body)
assert len(items) == 5
assert {it["name"] for it in items} == SEED_NAMES
# 规则分档:IF≥9 → challenge(The Plant Cell 11.6 / Genome Biology 10.1)
challenge = {it["name"] for it in body["tiers"]["challenge"]["items"]}
assert challenge == {"The Plant Cell", "Genome Biology"}
for it in items:
assert it["match_reason"] and it["risk_note"]
def test_vector_failure_degrades(client: TestClient, monkeypatch):
"""向量匹配失败 → 全量候选 + similarity 为 null,不报错。"""
monkeypatch.setattr(journal_kb, "match", make_fake_match(fail=True))
body = client.post("/api/review/recommend", json={"abstract": ABSTRACT}).json()
assert body["ok"] is True
assert body["meta"]["vector_ok"] is False
assert body["meta"]["candidate_count"] == 5
items = _all_items(body)
assert len(items) == 5
assert all(it["similarity"] is None for it in items)
def test_invalid_tier_repaired(client: TestClient, monkeypatch):
"""非法 tier 按规则补位,不丢期刊。"""
bad = [
{"id": "plant-genome", "tier": "超越顶刊", "match_reason": "tier 非法", "risk_note": "无"},
{"id": "plant-cell", "tier": "challenge", "match_reason": "合法", "risk_note": "无"},
]
monkeypatch.setattr(llm, "complete_json", make_fake_llm(bad))
body = client.post("/api/review/recommend", json={"abstract": ABSTRACT}).json()
assert body["ok"] is True
assert len(_all_items(body)) == 5
# plant-genome 规则分档 → IF 3.3 ≤ 4 → safe
safe_ids = [it["id"] for it in body["tiers"]["safe"]["items"]]
assert "plant-genome" in safe_ids
assert body["meta"]["repaired"] >= 1
def test_short_abstract_validation(client: TestClient):
body = client.post("/api/review/recommend", json={"abstract": "too short"}).json()
assert body["ok"] is False
assert body["error"]["code"] == "VALIDATION"
assert body["tiers"] is None
def test_tier_focus_filters_other_tiers(client: TestClient):
body = client.post(
"/api/review/recommend",
json={"abstract": ABSTRACT, "preferences": {"tier_focus": "safe"}},
).json()
assert body["ok"] is True
assert body["tiers"]["challenge"]["items"] == []
assert body["tiers"]["fit"]["items"] == []
assert body["tiers"]["safe"]["items"] # 聚焦档保留
assert body["meta"]["preferences"]["tier_focus"] == "safe"
def test_soft_preference_risk_note(client: TestClient):
"""软偏好:超周期偏好只加风险提示,不删条目。"""
body = client.post(
"/api/review/recommend",
json={"abstract": ABSTRACT, "preferences": {"max_review_weeks": 5}},
).json()
assert body["ok"] is True
items = _all_items(body)
assert len(items) == 5 # 不删
# The Plant Cell 审稿约 6 周 > 5 周 → 带 ⚠ 提示
pc = next(it for it in items if it["id"] == "plant-cell")
assert "超出你 5 周" in pc["risk_note"]
def test_seed_idempotent(tmp_path):
"""种子装载幂等:连跑两次仍恰好 5 行,且不编造 acceptance_rate。"""
async def work():
await db_mod.init_db(tmp_path / "seed.db")
try:
await db_mod.ensure_journals_seeded()
await db_mod.ensure_journals_seeded()
return await db_mod.list_journals()
finally:
await db_mod.close_db()
rows = asyncio.run(work())
assert len(rows) == 5
assert {r["id"] for r in rows} == {
"plant-cell", "genome-biology", "plant-journal",
"theoretical-applied-genetics", "plant-genome",
}
assert all(r["acceptance_rate"] is None for r in rows)
assert all(r["scope_desc"] for r in rows)
assert all(r["official_url"] for r in rows)
def test_rule_fallback_covers_all_tiers():
"""纯规则分档覆盖三档(不依赖 LLM 与网络)。"""
candidates = [
{"journal": {"id": "plant-cell", "impact_factor": 11.6, "review_time_weeks": 6.0}, "similarity": None},
{"journal": {"id": "plant-journal", "impact_factor": 7.2, "review_time_weeks": 5.0}, "similarity": None},
{"journal": {"id": "plant-genome", "impact_factor": 3.3, "review_time_weeks": 4.0}, "similarity": None},
]
rankings = journal_rank.fallback_rankings(candidates)
assert {r.journal_id for r in rankings} == {"plant-cell", "plant-journal", "plant-genome"}
by_id = {r.journal_id: r.tier for r in rankings}
assert by_id["plant-cell"] == "challenge"
assert by_id["plant-journal"] == "fit"
assert by_id["plant-genome"] == "safe"
+5
View File
@@ -81,6 +81,11 @@ class Settings(BaseSettings):
paper_writer_url: str = "http://127.0.0.1:8097"
paper_writer_read_timeout: float = 360.0 # write_paper_section 同步长调用读超时(秒)
# ---- 科研审稿服务(review_service 8100;REVIEW_ENABLED 为假时路由不注册)----
review_enabled: bool = False
review_service_url: str = "http://127.0.0.1:8100"
review_read_timeout: float = 360.0 # presubmit 上传+LLM 长调用读超时(秒)
# ---- 生产形态:BFF 静态托管 ui/dist(ARCHITECTURE.md §3,M6 挂载)----
frontend_dist_dir: str = "ui/dist" # 生产形态:静态托管 ui/dist
+3
View File
@@ -26,6 +26,7 @@ from .routers import capabilities, docgen, extraction, figures, health, kg, lega
from .routers import bioflow_proxy
from .routers import paper
from .routers import private_docs
from .routers import review_proxy
from .upstream import close_client
logging.basicConfig(level=logging.INFO)
@@ -62,6 +63,8 @@ def create_app() -> FastAPI:
app.include_router(bioflow_proxy.router) # 生信反代(BIOFLOW_ENABLED 开关,FR-BIO-028)
if get_settings().paper_writer_enabled:
app.include_router(paper.router) # 论文撰写(PAPER_WRITER_ENABLED 开关)
if get_settings().review_enabled:
app.include_router(review_proxy.router) # 科研审稿反代(REVIEW_ENABLED 开关)
@app.exception_handler(RequestValidationError)
async def _validation_handler(_: Request, exc: RequestValidationError) -> JSONResponse:
+85
View File
@@ -0,0 +1,85 @@
"""routers/review_proxy.py —— BFF 科研审稿反代(REVIEW_ENABLED 条件注册)。
- GET /api/review/health → 探活 review 服务 /health(离线返回 offline 标记,
前端据此隐藏审稿入口;模式同 routers/paper.py)。
- /api/review/{path:path} → 透明反代 review_service(8100);presubmit 为
上传+LLM 长调用,读超时取 settings.review_read_timeout。
- REVIEW_ENABLED 为假时路由不注册(main.py 按开关 include,三形态部署)。
约束:本文件是 BFF 层对审稿模块的唯一新增面(新文件,不改既有路由逻辑)。
review_service 失败统一 503 UPSTREAM_DOWN 信封(经 app.upstream.envelope)。
"""
from __future__ import annotations
import logging
from typing import Any
import httpx
from fastapi import APIRouter, Request
from fastapi.responses import JSONResponse, Response
from ..config import get_settings
from ..upstream import envelope
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/review", tags=["review-proxy"])
PROXY_CONNECT_TIMEOUT = 5.0
def _service_base() -> str:
return get_settings().review_service_url.rstrip("/")
@router.get("/health")
async def review_health() -> dict[str, Any]:
"""探活反代 8100 /health;离线返回 offline 标记。"""
s = get_settings()
try:
async with httpx.AsyncClient(timeout=10.0) as client:
resp = await client.get(f"{s.review_service_url.rstrip('/')}/health")
resp.raise_for_status()
return resp.json()
except Exception: # noqa: BLE001
return {"ok": False, "status": "offline"}
async def _proxy(request: Request, path: str) -> Any:
"""httpx 透传反代(非 SSE;presubmit 长读超时);失败→503 信封。"""
s = get_settings()
base = _service_base()
url = f"{base}/api/review/{path.lstrip('/')}"
if request.url.query:
url = f"{url}?{request.url.query}"
body = await request.body()
headers = {k: v for k, v in request.headers.items()
if k.lower() not in ("host", "content-length", "connection")}
timeout = httpx.Timeout(
connect=PROXY_CONNECT_TIMEOUT,
read=s.review_read_timeout,
write=s.review_read_timeout,
pool=5.0,
)
async with httpx.AsyncClient(timeout=timeout) as client:
try:
upstream = await client.build_request(
request.method, url, content=body, headers=headers
)
resp = await client.send(upstream)
except httpx.HTTPError as exc:
logger.warning("review_proxy: 审稿服务不可达 %s: %s", url, exc)
payload, status = envelope(
"UPSTREAM_DOWN", message="审稿服务当前不可用,请稍后重试",
degraded=True, status=503,
)
return JSONResponse(status_code=status, content=payload)
content = await resp.aread()
media_type = resp.headers.get("content-type", "") or "application/json"
return Response(status_code=resp.status_code, content=content, media_type=media_type)
@router.api_route("/{path:path}", methods=["GET", "POST", "PUT", "PATCH", "DELETE"])
async def proxy_all(path: str, request: Request) -> Any:
return await _proxy(request, path)