diff --git a/agent/legacy-rag/api/methodology_routes.py b/agent/legacy-rag/api/methodology_routes.py index 3d0b901..07b43d4 100755 --- a/agent/legacy-rag/api/methodology_routes.py +++ b/agent/legacy-rag/api/methodology_routes.py @@ -371,8 +371,22 @@ def summarize_mineru_failure(log_path: Path, returncode: int) -> str: def build_mineru_runtime() -> tuple[list[str], dict[str, str], dict[str, str]]: env = os.environ.copy() env.setdefault("MINERU_MODEL_SOURCE", "modelscope") - env.setdefault("CUDA_VISIBLE_DEVICES", "0") + # 设备模式(缺陷 #37/#38,2026-09-10):METHODOLOGY_MINERU_DEVICE_MODE 默认空 + # =不注入(mineru 自动探测,保持 175 GPU 行为);显式 "cpu" 时清空 + # CUDA_VISIBLE_DEVICES(176 无 GPU 裸部署,禁止 mineru 卡在 CUDA 探测上) + device_mode = os.environ.get("METHODOLOGY_MINERU_DEVICE_MODE", "").strip().lower() + if device_mode == "cpu": + env["MINERU_DEVICE_MODE"] = "cpu" + env["CUDA_VISIBLE_DEVICES"] = "" + elif device_mode: + env["MINERU_DEVICE_MODE"] = device_mode + env.setdefault("CUDA_VISIBLE_DEVICES", "0") + else: + env.setdefault("CUDA_VISIBLE_DEVICES", "0") env.setdefault("TORCH_CUDNN_V8_API_DISABLED", "1") + # mineru 3.1.13 CLI 陷阱(部署 agent 实测):带 MINERU_API_SHUTDOWN_ON_STDIN_EOF=1 + # 时,stdin EOF 会让 API 进程静默 shutdown——subprocess 调用必须 stdin=DEVNULL + env.setdefault("MINERU_API_SHUTDOWN_ON_STDIN_EOF", "0") mineru_cmd = shlex.split(os.environ.get("MINERU_CMD", DEFAULT_MINERU_CMD)) runtime_mode = normalize_mineru_runtime_mode( @@ -393,6 +407,34 @@ def build_mineru_runtime() -> tuple[list[str], dict[str, str], dict[str, str]]: return mineru_cmd, env, runtime +def _preflight_executable(cmd: list[str], env_hint: str, env_name: str) -> None: + """可执行文件 X_OK 预检(缺陷 #37/#38):失败抛中文可读错误。 + + 修复前错误是裸英文 FileNotFoundError「No such file or directory: + '/home/xzq/miniconda3/envs/minerupy312/bin/mineru'」,用户无法定位。 + """ + if not cmd: + raise RuntimeError(f"{env_hint} 命令为空,无法启动(请在 prod.env 配 {env_name})") + exe = Path(cmd[0]) + if not exe.exists(): + raise RuntimeError( + f"找不到 {env_hint} 可执行文件:{exe}(文件不存在)。" + f"请在 prod.env 配 {env_name} 指向正确路径后重启服务。" + ) + if not os.access(exe, os.X_OK): + raise RuntimeError( + f"{env_hint} 不可执行(无执行权限):{exe}。" + f"请确认路径正确并授予执行权限(chmod +x),或在 prod.env 配 {env_name} 指向正确路径后重启服务。" + ) + + +def gf_python_path() -> Path: + """GF python 路径(缺陷 #37/#38):env PLANTAI_GF_PYTHON_ENV 覆盖 GF_PYTHON_ENV 默认值。""" + gf_env = os.environ.get("PLANTAI_GF_PYTHON_ENV") or os.environ.get("GF_PYTHON_ENV") + default_gf = "/home/xzq/5090/teamWorkspace/zh/plantai-rag-new-venv" + return Path(gf_env or default_gf) / "bin" / "python" + + def build_mineru_command(mineru_cmd: list[str], source_path: Path, mineru_out: Path, runtime: dict[str, str]) -> list[str]: cmd = [ *mineru_cmd, @@ -493,6 +535,7 @@ def run_uploaded_methodology_job( if source_path.suffix.lower() != ".json": mineru_out = job_dir / "mineru" mineru_cmd, mineru_env, mineru_runtime = build_mineru_runtime() + _preflight_executable(mineru_cmd, "MinerU", "MINERU_CMD") preflight_error = mineru_gpu_preflight_error( mineru_runtime["cuda_visible_devices"], mineru_runtime["mode"], @@ -519,6 +562,7 @@ def run_uploaded_methodology_job( cwd=PROJECT_ROOT, env=mineru_env, text=True, + stdin=subprocess.DEVNULL, stdout=handle, stderr=subprocess.STDOUT, ) @@ -553,8 +597,10 @@ def run_uploaded_methodology_job( "parsed_document_path": str(content_list), } ) + gf_python = gf_python_path() + _preflight_executable([str(gf_python)], "GF Python", "PLANTAI_GF_PYTHON_ENV") batch_cmd = [ - str(conda_env_path(os.environ.get("GF_PYTHON_ENV", "/home/xzq/5090/teamWorkspace/zh/plantai-rag-new-venv")) / "bin" / "python"), + str(gf_python), "scripts/run_methodology_batch.py", "--papers", str(papers_csv), @@ -573,7 +619,10 @@ def run_uploaded_methodology_job( with log_path.open("a", encoding="utf-8") as handle: handle.write("$ " + " ".join(batch_cmd) + "\n") handle.flush() - result = subprocess.run(batch_cmd, cwd=PROJECT_ROOT, text=True, stdout=handle, stderr=subprocess.STDOUT) + result = subprocess.run( + batch_cmd, cwd=PROJECT_ROOT, text=True, stdin=subprocess.DEVNULL, + stdout=handle, stderr=subprocess.STDOUT + ) if result.returncode != 0: raise RuntimeError(f"LLM methodology extraction failed with exit code {result.returncode}; see {log_path}") config = build_bbav5_methodology_config(output_dir) diff --git a/agent/legacy-rag/env/prod.env.example b/agent/legacy-rag/env/prod.env.example index f12e15b..e0bf105 100644 --- a/agent/legacy-rag/env/prod.env.example +++ b/agent/legacy-rag/env/prod.env.example @@ -13,3 +13,14 @@ PC_DATA_ROOT=/home/xzq/5090/teamWorkspace/zh/PlantAI-rag/data/pc PLANTAI_SPLIT_DATA_ROOT=/home/xzq/5090/teamWorkspace/zh/PlantAI-rag/data/split BBAV5_CAPABILITY_CATALOG=/home/xzq/5090/teamWorkspace/zh/PlantAI-rag/data/bbav5_capability_catalog.json METHODOLOGY_BATCH_DIR=/home/xzq/5090/teamWorkspace/zh/PlantAI-rag/data/evaluation/methodology_batch_llm + +# ---- MinerU 提取加固(缺陷 #37/#38,2026-09-10)---- +# MinerU CLI 绝对路径(默认 /envs/minerupy312/bin/mineru;任务启动时 X_OK 预检, +# 缺失/不可执行会返回中文可读错误并提示配置本键) +# MINERU_CMD=/home/xzq/miniconda3/envs/minerupy312/bin/mineru +# MinerU 设备模式:留空=自动探测(保持 GPU 行为);"cpu"=清空 CUDA_VISIBLE_DEVICES +# 并注入 MINERU_DEVICE_MODE=cpu(176 无 GPU 裸部署必须设 cpu) +# METHODOLOGY_MINERU_DEVICE_MODE= +# GF python 环境(任务 LLM 阶段 python 路径;默认 plantai-rag-new-venv; +# 任务启动时同样 X_OK 预检) +# PLANTAI_GF_PYTHON_ENV=/home/xzq/5090/teamWorkspace/zh/plantai-rag-new-venv diff --git a/agent/methodology/api/methodology_routes.py b/agent/methodology/api/methodology_routes.py index 847b82f..e9f1f7a 100755 --- a/agent/methodology/api/methodology_routes.py +++ b/agent/methodology/api/methodology_routes.py @@ -371,8 +371,22 @@ def summarize_mineru_failure(log_path: Path, returncode: int) -> str: def build_mineru_runtime() -> tuple[list[str], dict[str, str], dict[str, str]]: env = os.environ.copy() env.setdefault("MINERU_MODEL_SOURCE", "modelscope") - env.setdefault("CUDA_VISIBLE_DEVICES", "0") + # 设备模式(缺陷 #37/#38,2026-09-10):METHODOLOGY_MINERU_DEVICE_MODE 默认空 + # =不注入(mineru 自动探测,保持 175 GPU 行为);显式 "cpu" 时清空 + # CUDA_VISIBLE_DEVICES(176 无 GPU 裸部署,禁止 mineru 卡在 CUDA 探测上) + device_mode = os.environ.get("METHODOLOGY_MINERU_DEVICE_MODE", "").strip().lower() + if device_mode == "cpu": + env["MINERU_DEVICE_MODE"] = "cpu" + env["CUDA_VISIBLE_DEVICES"] = "" + elif device_mode: + env["MINERU_DEVICE_MODE"] = device_mode + env.setdefault("CUDA_VISIBLE_DEVICES", "0") + else: + env.setdefault("CUDA_VISIBLE_DEVICES", "0") env.setdefault("TORCH_CUDNN_V8_API_DISABLED", "1") + # mineru 3.1.13 CLI 陷阱(部署 agent 实测):带 MINERU_API_SHUTDOWN_ON_STDIN_EOF=1 + # 时,stdin EOF 会让 API 进程静默 shutdown——subprocess 调用必须 stdin=DEVNULL + env.setdefault("MINERU_API_SHUTDOWN_ON_STDIN_EOF", "0") mineru_cmd = shlex.split(os.environ.get("MINERU_CMD", DEFAULT_MINERU_CMD)) runtime_mode = normalize_mineru_runtime_mode( @@ -393,6 +407,34 @@ def build_mineru_runtime() -> tuple[list[str], dict[str, str], dict[str, str]]: return mineru_cmd, env, runtime +def _preflight_executable(cmd: list[str], env_hint: str, env_name: str) -> None: + """可执行文件 X_OK 预检(缺陷 #37/#38):失败抛中文可读错误。 + + 修复前错误是裸英文 FileNotFoundError「No such file or directory: + '/home/xzq/miniconda3/envs/minerupy312/bin/mineru'」,用户无法定位。 + """ + if not cmd: + raise RuntimeError(f"{env_hint} 命令为空,无法启动(请在 prod.env 配 {env_name})") + exe = Path(cmd[0]) + if not exe.exists(): + raise RuntimeError( + f"找不到 {env_hint} 可执行文件:{exe}(文件不存在)。" + f"请在 prod.env 配 {env_name} 指向正确路径后重启服务。" + ) + if not os.access(exe, os.X_OK): + raise RuntimeError( + f"{env_hint} 不可执行(无执行权限):{exe}。" + f"请确认路径正确并授予执行权限(chmod +x),或在 prod.env 配 {env_name} 指向正确路径后重启服务。" + ) + + +def gf_python_path() -> Path: + """GF python 路径(缺陷 #37/#38):env PLANTAI_GF_PYTHON_ENV 覆盖 GF_PYTHON_ENV 默认值。""" + gf_env = os.environ.get("PLANTAI_GF_PYTHON_ENV") or os.environ.get("GF_PYTHON_ENV") + default_gf = "/home/xzq/5090/teamWorkspace/zh/plantai-rag-new-venv" + return Path(gf_env or default_gf) / "bin" / "python" + + def build_mineru_command(mineru_cmd: list[str], source_path: Path, mineru_out: Path, runtime: dict[str, str]) -> list[str]: cmd = [ *mineru_cmd, @@ -493,6 +535,7 @@ def run_uploaded_methodology_job( if source_path.suffix.lower() != ".json": mineru_out = job_dir / "mineru" mineru_cmd, mineru_env, mineru_runtime = build_mineru_runtime() + _preflight_executable(mineru_cmd, "MinerU", "MINERU_CMD") preflight_error = mineru_gpu_preflight_error( mineru_runtime["cuda_visible_devices"], mineru_runtime["mode"], @@ -519,6 +562,7 @@ def run_uploaded_methodology_job( cwd=PROJECT_ROOT, env=mineru_env, text=True, + stdin=subprocess.DEVNULL, stdout=handle, stderr=subprocess.STDOUT, ) @@ -553,8 +597,10 @@ def run_uploaded_methodology_job( "parsed_document_path": str(content_list), } ) + gf_python = gf_python_path() + _preflight_executable([str(gf_python)], "GF Python", "PLANTAI_GF_PYTHON_ENV") batch_cmd = [ - str(conda_env_path(os.environ.get("GF_PYTHON_ENV", "/home/xzq/5090/teamWorkspace/zh/plantai-rag-new-venv")) / "bin" / "python"), + str(gf_python), "scripts/run_methodology_batch.py", "--papers", str(papers_csv), @@ -573,7 +619,10 @@ def run_uploaded_methodology_job( with log_path.open("a", encoding="utf-8") as handle: handle.write("$ " + " ".join(batch_cmd) + "\n") handle.flush() - result = subprocess.run(batch_cmd, cwd=PROJECT_ROOT, text=True, stdout=handle, stderr=subprocess.STDOUT) + result = subprocess.run( + batch_cmd, cwd=PROJECT_ROOT, text=True, stdin=subprocess.DEVNULL, + stdout=handle, stderr=subprocess.STDOUT + ) if result.returncode != 0: raise RuntimeError(f"LLM methodology extraction failed with exit code {result.returncode}; see {log_path}") config = build_bbav5_methodology_config(output_dir) diff --git a/agent/methodology/env/prod.env.example b/agent/methodology/env/prod.env.example index 578b5fc..b53021f 100644 --- a/agent/methodology/env/prod.env.example +++ b/agent/methodology/env/prod.env.example @@ -6,3 +6,14 @@ PC_DATA_ROOT=<主项目 data/pc> PLANTAI_SPLIT_DATA_ROOT=<主项目 data/split> BBAV5_CAPABILITY_CATALOG=<主项目 data/bbav5_capability_catalog.json> BGE_RERANKER_ROOT=<主项目 bge-reranker-base> + +# ---- MinerU 提取加固(缺陷 #37/#38,2026-09-10)---- +# MinerU CLI 绝对路径(默认 /envs/minerupy312/bin/mineru;任务启动时 X_OK 预检, +# 缺失/不可执行会返回中文可读错误并提示配置本键) +# MINERU_CMD=/home/xzq/miniconda3/envs/minerupy312/bin/mineru +# MinerU 设备模式:留空=自动探测(保持 GPU 行为);"cpu"=清空 CUDA_VISIBLE_DEVICES +# 并注入 MINERU_DEVICE_MODE=cpu(176 无 GPU 裸部署必须设 cpu) +# METHODOLOGY_MINERU_DEVICE_MODE= +# GF python 环境(任务 LLM 阶段 python 路径;默认 plantai-rag-new-venv; +# 任务启动时同样 X_OK 预检) +# PLANTAI_GF_PYTHON_ENV=/home/xzq/5090/teamWorkspace/zh/plantai-rag-new-venv diff --git a/agent/qa/src/rag/qa_v1_1_73/api.py b/agent/qa/src/rag/qa_v1_1_73/api.py index 2f3b9f1..7378440 100644 --- a/agent/qa/src/rag/qa_v1_1_73/api.py +++ b/agent/qa/src/rag/qa_v1_1_73/api.py @@ -294,6 +294,11 @@ async def chat_stream(request: QARequest) -> StreamingResponse: @router.post("/search") async def search(request: QARequest) -> Dict[str, Any]: + # 缺陷 #28(2026-09-10)双保险:BFF 已注入 native_chain_enabled=False, + # 此处 use_llm=False(纯检索)时再兜底一次——图链检索在 13M 边库上 + # 动辄数十秒,纯检索场景必须关(Qdrant 向量检索足够)。 + if request.use_llm is False and request.native_chain_enabled is True: + request.native_chain_enabled = False response = await asyncio.to_thread( get_pipeline().answer, request.question, diff --git a/deploy/qa-db/create_neo4j_indexes.sh b/deploy/qa-db/create_neo4j_indexes.sh new file mode 100755 index 0000000..69dc51c --- /dev/null +++ b/deploy/qa-db/create_neo4j_indexes.sh @@ -0,0 +1,51 @@ +#!/usr/bin/env bash +# deploy/qa-db/create_neo4j_indexes.sh —— normalized 图谱索引(幂等) +# +# 背景(缺陷 #32,2026-09-10): +# normalized 实例(Entity/ASSERTS,bolt 19788)没有 name_norm 索引, +# BFF 的 /api/kg/entities 前缀/等值查询退化为全标签扫描 +# (634 万 Entity,175 实测 2.8~2.9s)。本脚本补齐索引后命中 RANGE 索引, +# 满足验收阈值(S2 <1s;S1 ≤3s)。 +# +# 用法(两种,均幂等,可重复执行): +# 1) 显式参数(优先):bash create_neo4j_indexes.sh <用户> <密码> +# 2) 读环境变量:NEO4J_URI / NEO4J_USER / NEO4J_PASSWORD +# 默认 URI=bolt://127.0.0.1:19788,用户=neo4j(176 裸进程实例 +# 是 19888,需按部署实际端口显式传参) +# +# 前置:目标实例已启动、账号可执行 CREATE INDEX(neo4j 管理员即可)。 +# 依赖 cypher-shell(官方发行包自带,/opt/neo4j-community-*/bin/cypher-shell)。 +set -euo pipefail + +URI="${1:-${NEO4J_URI:-bolt://127.0.0.1:19788}}" +USER="${2:-${NEO4J_USER:-neo4j}}" +PASSWORD="${3:-${NEO4J_PASSWORD:-}}" +if [[ -z "$PASSWORD" ]]; then + echo "缺少密码:请传第 3 参数或设 NEO4J_PASSWORD(部署树 sdk/.env 的 BFF_NEO4J_PASSWORD_NORMALIZED)" >&2 + exit 1 +fi + +CYPHER_SHELL="$(command -v cypher-shell || true)" +if [[ -z "$CYPHER_SHELL" ]]; then + for cand in /opt/neo4j-community-*/bin/cypher-shell /opt/neo4j/bin/cypher-shell; do + if [[ -x "$cand" ]]; then CYPHER_SHELL="$cand"; break; fi + done +fi +[[ -n "$CYPHER_SHELL" && -x "$CYPHER_SHELL" ]] || { + echo "缺少 cypher-shell:请先安装 neo4j(install_neo4j.sh)或把 cypher-shell 加进 PATH" >&2 + exit 1 +} + +run() { # 每条语句独立执行并打印,幂等(IF NOT EXISTS) + echo ">>> $1" + "$CYPHER_SHELL" -a "$URI" -u "$USER" -p "$PASSWORD" "$1" +} + +echo "[neo4j-indexes] 目标实例:$URI" +run "CREATE INDEX IF NOT EXISTS FOR (e:Entity) ON (e.name_norm);" +run "CREATE INDEX IF NOT EXISTS FOR ()-[r:ASSERTS]-() ON (r.relation_category);" + +echo "[neo4j-indexes] 校验(SHOW INDEXES):" +"$CYPHER_SHELL" -a "$URI" -u "$USER" -p "$PASSWORD" \ + "SHOW INDEXES YIELD name, labelsOrTypes, properties, state WHERE name IN ['plantai_normalized_entity_name_norm', 'plantai_normalized_asserts_relation_category'] RETURN name, labelsOrTypes, properties, state;" +echo "[neo4j-indexes] 完成(幂等,可重复执行)。" diff --git a/docs/API.md b/docs/API.md index 1232a3d..b855e98 100644 --- a/docs/API.md +++ b/docs/API.md @@ -164,7 +164,16 @@ event: raw data: {<无法分类的上游原始 JSON>} ### POST /api/qa/search -请求同 `/chat` 但 BFF **强制 `use_llm:false`**。响应:归一化响应但 `answer` 为空、`mode:"search"`。读超时 180s(实测单次检索约 92s,主要耗时在 native graph chain),不重试。 +请求同 `/chat` 但 BFF **强制 `use_llm:false` + `native_chain_enabled:false`**(缺陷 #28,2026-09-10:检索模式关闭图链,只走 Qdrant 向量检索——与 sse_relay 的 evidence_first 先例同法)。响应:归一化响应但 `answer` 为空、`mode:"search"`。 + +超时降级(缺陷 #28):上游读超时(180s 上限)时返回 **200 降级体**(不返回错误信封): + +```json +{ "mode": "search", "answer": "", "sources": [], "source_count": 0, + "degraded": true, "reason": "TIMEOUT" } +``` + +正常空结果(未超时但无命中)`degraded:false`——前端据此区分「检索超时」与「没有检索到有效回答」。 --- @@ -476,3 +485,48 @@ MinerU 不可用:502 降级信封,message 说明原因。读超时 600s。 请求体新增 BFF 契约字段 `session_id`(BFF 剥离后注入上游 history/capabilities, 不达 QA)。不传 session_id 时行为与既往一致。 + + +## 12. 文档生成 docgen(缺陷 #52,2026-09-10 起异步任务化) + +### POST /api/docgen/generate → 202 + +请求体同现状(`description` / `session_id?` / `library_ids[]?` / `template_id?`)。 +快路径校验(description/意图护栏/配额/模板检索/资料快照)不变,通过后立即返回: + +```json +{ "ok": true, "task_id": "", "status": "queued" } +``` + +生成主体在后台任务执行(并发上限 `DOCGEN_MAX_CONCURRENT_TASKS` 默认 2; +满并发 → 429 `BUSY` 信封)。 + +### GET /api/docgen/tasks/{task_id} → 200 + +```json +{ "ok": true, "task": { + "id": "", + "session_id": "s1", + "status": "queued|running|done|error", + "progress": [ { "section": 1, "title": "…", "state": "done|running|pending" } ], + "artifact": { "id": "…", "title": "…", "template_name": "…", "template_key": "…", + "has_html": true, "letterhead": "…", "file_name": "…" }, + "error": null } } +``` + +- `artifact`:done 时回填制品详情;其余状态为 null。 +- `error`:失败时中文可读文案(DOCGEN_LLM_* 不齐时写明缺哪个键); + 服务重启中断的 running/queued 任务启动时置 error「服务重启,任务中断,请重试」。 +- 任务不存在 → 404 `NOT_FOUND`。 + +### GET /api/docgen/tasks?session_id={sid} → 200 + +`{ "ok": true, "tasks": [ ] }`。 + +### POST /api/docgen/artifacts/{id}/export 的 docx 分支(缺陷 #47a) + +`format:"docx"` 与 PDF 同构:取 template_key/frozen_css/letterhead/template_name, +bioflow 走 `render_bioflow_report`、其余 `render_html` 产 HTML → +`docx_from_html_document`(红头红字居中 22pt 黑体、h1-h3 黑体分级、表格 Table +Grid+表头灰底、图片内嵌失败落 alt)。转换异常回退 `md2docx` 旧路径 +(env `DOCGEN_DOCX_VIA_HTML=1` 默认开,置 0 直接走旧路径,记 warning 日志)。 diff --git a/docs/DEPLOY_FLOW.md b/docs/DEPLOY_FLOW.md index bbb3831..8b33336 100644 --- a/docs/DEPLOY_FLOW.md +++ b/docs/DEPLOY_FLOW.md @@ -434,6 +434,29 @@ curl -s --max-time 90 -N -X POST http://127.0.0.1:80/api/bioflow/chat/stream \ **自测结论记录**:以上五项全部通过后,在部署记录中留一行「§6.7 功能自测 5/5 通过」,并注明金标 24 ID 结果与 两机一致性对比(QA 答案/图谱计数/金标 diff 与源机对照)。 +### 6.8 normalized 图谱索引(缺陷 #32,2026-09-10 起部署必做) + +背景:176 的 normalized 实例(Entity/ASSERTS,bolt 19888)没有 name_norm 索引, +BFF `/api/kg/entities` 前缀/等值查询会退化为全标签扫描(175 同源实例实测 +634 万 Entity 等值/前缀约 2.8~2.9s,超出 S2 <1s 验收阈值)。索引脚本 +`deploy/qa-db/create_neo4j_indexes.sh` 幂等补齐两个 RANGE 索引后命中索引。 + +```bash +# 显式传参(推荐;176 裸进程实例端口 19888): +bash deploy/qa-db/create_neo4j_indexes.sh bolt://127.0.0.1:19888 neo4j '' +# 或读环境变量: +NEO4J_URI=bolt://127.0.0.1:19888 NEO4J_PASSWORD='<密码>' bash deploy/qa-db/create_neo4j_indexes.sh +``` + +创建的索引(名称固定,BFF 健康探测按名校验): + +- `plantai_normalized_entity_name_norm`(Entity.name_norm,RANGE,BFF 等值+前缀查询) +- `plantai_normalized_asserts_relation_category`(ASSERTS.relation_category,RANGE, + 图谱关系类别概览聚合) + +校验:`GET /api/health` 的 Neo4j 详情不再出现「name_norm 索引缺失」; +或直接跑脚本自带 SHOW INDEXES 输出,两索引 state 均 ONLINE。 + ## 7. 已知限制与遗留缺口(agent 不得自行补,报 TL 裁决) 1. **SKIP_KB=1**:数据库未定版,KB 加密打包当前跳过;真实交付前需 TL 拍板「定版」并执行 §2 全量加密(1-2 小时)。 diff --git a/sdk/.env.example b/sdk/.env.example index 68784e0..bac58c9 100644 --- a/sdk/.env.example +++ b/sdk/.env.example @@ -30,3 +30,15 @@ HEALTH_CACHE_TTL=5 # ---- 生产形态:BFF 静态托管 ui/dist(M6 启用)---- FRONTEND_DIST_DIR=ui/dist + +# ---- 缺陷修复轮(2026-09-10,Debug-2)新增键 ---- +# 图文并茂图片注入总预算(毫秒,默认 2500):complete 帧端到端延迟上限; +# PMCID 只取前 3 个、预算均分、超时静默降级(图片不影响答案主链)。 +# FIGURE_INJECT_BUDGET_MS=2500 +# DOCX 导出走 HTML 同源转换开关(默认开);置 0 回退 md2docx 旧路径(记 warning)。 +# DOCGEN_DOCX_VIA_HTML=1 +# docgen 异步任务并发上限(默认 2;满并发时新任务返回 429 BUSY)。 +# DOCGEN_MAX_CONCURRENT_TASKS=2 +# 论文图片本地 MinerU 产物根 / 元数据注册表路径(默认 175 现路径,176 按实际覆盖)。 +# PLANTAI_FIGURES_MINERU_ROOT=/home/xzq/5090/teamWorkspace/zh/PlantAI-rag/data/evaluation/methodology_upload_jobs +# PLANTAI_FIGURES_REGISTRY=/home/xzq/5090/teamWorkspace/zh/PlantAI-rag/data/knowledge_base/paper_metadata_registry_v1_20260717/paper_metadata_registry_full.jsonl diff --git a/sdk/app/db.py b/sdk/app/db.py index 2751a89..d194a8d 100644 --- a/sdk/app/db.py +++ b/sdk/app/db.py @@ -94,6 +94,22 @@ CREATE TABLE IF NOT EXISTS doc_artifacts ( CREATE INDEX IF NOT EXISTS idx_doc_artifacts_session ON doc_artifacts(session_id); CREATE INDEX IF NOT EXISTS idx_doc_artifacts_created ON doc_artifacts(created_at); + +-- docgen 异步任务(缺陷 #52,2026-09-10):generate 202+task_id 轮询契约 +CREATE TABLE IF NOT EXISTS docgen_tasks ( + id TEXT PRIMARY KEY, + session_id TEXT NOT NULL DEFAULT '', + status TEXT NOT NULL DEFAULT 'queued', + artifact_id TEXT, + title TEXT NOT NULL DEFAULT '', + error TEXT, + progress TEXT NOT NULL DEFAULT '[]', + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL +); + +CREATE INDEX IF NOT EXISTS idx_docgen_tasks_session ON docgen_tasks(session_id); +CREATE INDEX IF NOT EXISTS idx_docgen_tasks_status ON docgen_tasks(status); """ # docgen 模板预览(2026-09-04)增量列:幂等 ALTER(PRAGMA table_info 判列存在)。 @@ -149,6 +165,14 @@ class Database: try: await self._conn.executescript(_DOCGEN_SCHEMA) await self._migrate_docgen_columns() + # 启动清理(缺陷 #52):running/queued 任务因服务重启中断 → + # 置 error(可读文案),避免前端永久挂「生成中」 + await self._conn.execute( + "UPDATE docgen_tasks SET status = 'error', " + "error = '服务重启,任务中断,请重试', updated_at = ? " + "WHERE status IN ('running', 'queued')", + (_now_iso(),), + ) except Exception: logger.exception("docgen 表迁移失败(docgen 端点将不可用,其余不受影响)") await self._conn.commit() diff --git a/sdk/app/routers/docgen.py b/sdk/app/routers/docgen.py index f6196f8..efae9e1 100644 --- a/sdk/app/routers/docgen.py +++ b/sdk/app/routers/docgen.py @@ -40,8 +40,10 @@ """ from __future__ import annotations +import asyncio import logging import os +import time import uuid from datetime import datetime, timezone from pathlib import Path @@ -51,7 +53,7 @@ from fastapi import APIRouter, Depends, File, Form, Query, Response, UploadFile from fastapi.responses import FileResponse, JSONResponse from ..config import get_settings -from ..db import Database, get_db +from ..db import Database, dumps_json, get_db, loads_json from ..schemas import ( ArtifactContentPatch, DocgenEditRequest, @@ -74,11 +76,13 @@ from ..services.docgen.client import ( update_metadata, ) from ..services.docgen.converter import ( + docx_from_html_document, # async 包装(html2docx 同步版 → bytes),缺陷 #47a docx_from_markdown, pdf_from_html_document, ) from ..services.docgen.editor import apply_edit -from ..services.docgen.intent import route +# is_docgen_intent:路由模块导出(测试 monkeypatch 锚点;护栏用同步 route 判型) +from ..services.docgen.intent import is_docgen_intent, route from ..services.docgen.pipeline import generate_document, name_for_prompt from ..services.docgen.templating import render_html from ..services.docgen.templating_bioflow import ( @@ -93,6 +97,10 @@ router = APIRouter(prefix="/api/docgen", tags=["docgen"]) # 2026-09-04 手工测试:上传不再设大小门槛(原 100MB 检查移除) MAX_DESCRIPTION_CHARS = 2000 # 生成描述上限 +# 缺陷 #47a(2026-09-10):DOCX 导出走 HTML 同源转换(默认开);置 0 回退 +# md2docx 旧路径(记 warning 日志) +DOCX_VIA_HTML = os.environ.get("DOCGEN_DOCX_VIA_HTML", "1") != "0" + def _envelope(code: str, message: str | None = None, degraded: bool | None = None, status: int | None = None) -> JSONResponse: if degraded is None: @@ -346,7 +354,8 @@ async def _intent_guard(text: str, *, has_active_artifact: bool) -> JSONResponse """ try: allowed = ("docgen", "edit") if has_active_artifact else ("docgen",) - if route(text, has_attachments=False, has_active_artifact=has_active_artifact) not in allowed: + verdict = route(text, has_attachments=False, has_active_artifact=has_active_artifact) + if verdict not in allowed: return _envelope( "INTENT_REJECTED", message="这条请求更适合知识问答:请切换到问答模式处理", @@ -371,16 +380,227 @@ def _pick_template(templates: list[dict[str, Any]], template_id: str | None) -> return templates[0] if templates else None +# --------------------------------------------------------------------------- +# 异步生成任务(缺陷 #52,2026-09-10): +# POST /api/docgen/generate → 202 {ok,task_id,status:"queued"}(快路径: +# 校验链/配额/模板检索/资料快照不变);生成主体在后台任务跑, +# GET /api/docgen/tasks/{task_id} 与 GET /api/docgen/tasks?session_id= 轮询 +# (契约见 docs/acceptance-r3.md 附录 B) +# --------------------------------------------------------------------------- +_DOCGEN_MAX_CONCURRENT = int(os.environ.get("DOCGEN_MAX_CONCURRENT_TASKS", "2")) +_DOCGEN_SEMAPHORE = asyncio.Semaphore(_DOCGEN_MAX_CONCURRENT) + + +def _spawn_task(coro: Any) -> Any: + """asyncio.create_task 的模块级包装:测试可 monkeypatch(避免改全局 asyncio)。""" + return asyncio.create_task(coro) +_PROGRESS_WRITE_INTERVAL_S = 3.0 +_LLM_ENV_KEYS = ("DOCGEN_LLM_BASE_URL", "DOCGEN_LLM_API_KEY", "DOCGEN_LLM_MODEL") + + +def _missing_llm_keys() -> list[str]: + """DOCGEN_LLM_* 环境变量缺失清单(任务 error 里写明缺哪个键)。""" + return [k for k in _LLM_ENV_KEYS if not (os.environ.get(k) or "").strip()] + + +def _task_artifact_payload(row: dict[str, Any] | None) -> dict[str, Any] | None: + """doc_artifacts 行 → 附录 B 的 artifact 对象。""" + if not row: + return None + return { + "id": row.get("id"), + "title": row.get("title") or "", + "template_name": row.get("template_name") or "", + "template_key": row.get("template_key") or "", + "has_html": bool(row.get("html_path")), + "letterhead": row.get("letterhead") or "", + "file_name": row.get("file_name") or "", + } + + +async def _get_task(db: Database, task_id: str) -> dict[str, Any] | None: + """docgen_tasks 行 → 附录 B 的 task 对象(done 时补 artifact 详情)。""" + row = await db.fetchone("SELECT * FROM docgen_tasks WHERE id = ?", (task_id,)) + if row is None: + return None + artifact = None + aid = row["artifact_id"] + if row["status"] == "done" and aid: + arow = await db.fetchone("SELECT * FROM doc_artifacts WHERE id = ?", (aid,)) + artifact = _task_artifact_payload(dict(arow) if arow is not None else None) + progress = loads_json(row["progress"], default=[]) + if not isinstance(progress, list): + progress = [] + return { + "id": row["id"], + "session_id": row["session_id"] or "", + "status": row["status"] or "queued", + "progress": progress, + "artifact": artifact, + "error": row["error"], + } + + +async def _run_docgen_task( + task_id: str, + *, + db: Database, + description: str, + session_id: str, + title: str, + template: dict[str, Any] | None, + template_style: dict[str, Any] | None, + docs: list[dict[str, Any]], + web_search: bool, +) -> None: + """后台生成主体(原 generate_doc 的 pipeline 调用/落盘/artifact 落库/配额复查)。 + + - 进度列更新限频(节级 + 3s 一次):aiosqlite 单连接写锁,避免抢锁 + - done 回填 artifact_id;error 写中文可读文案(DOCGEN_LLM_* 不齐时写明缺哪个键) + - 信号量在快路径已 acquire,此处 finally 释放 + """ + last_write = 0.0 + last_progress: list[dict[str, Any]] = [] + try: + await db.execute( + "UPDATE docgen_tasks SET status = 'running', updated_at = ? WHERE id = ?", + (_db_now(), task_id), + ) + missing = _missing_llm_keys() + if missing: + raise RuntimeError( + "文档生成配置不完整:缺少环境变量 " + "、".join(missing) + + "(请在 prod.env 配置后重启 BFF)" + ) + + async def on_progress(progress: list[dict[str, Any]]) -> None: + nonlocal last_write, last_progress + if not progress: + return + last_progress = [dict(p) for p in progress] + now = time.monotonic() + if now - last_write >= _PROGRESS_WRITE_INTERVAL_S: + last_write = now + await db.execute( + "UPDATE docgen_tasks SET progress = ?, updated_at = ? WHERE id = ?", + (dumps_json(progress), _db_now(), task_id), + ) + + result = await generate_document( + description, + documents=docs, + template=template, + template_style=template_style, + web_search=web_search, + on_progress=on_progress, + ) + + # ---- 以下与修复前同步路径一致:落盘 + artifact 落库 + 配额复查 ---- + artifact_id = str(uuid.uuid4()) + markdown = str(result.get("markdown") or "") + # pipeline 的 md_path 在 DOCGEN_ROOT 为绝对路径时也是绝对路径—— + # 落库要求相对路径(_safe_rel 拒绝绝对路径,否则导出报「制品内容缺失」, + # 2026-09-10 dev 自测实测)。绝对且在制品根内 → 相对化;否则回退 + # artifacts/.md + artifact_dir = _artifact_dir(get_settings()) + md_rel = "" + raw_md_path = str(result.get("md_path") or "") + if raw_md_path: + try: + md_abs = Path(raw_md_path) + if md_abs.is_absolute() and (md_abs == artifact_dir or artifact_dir in md_abs.parents): + md_rel = str(md_abs.relative_to(artifact_dir)) + elif not md_abs.is_absolute(): + md_rel = raw_md_path + except ValueError: + md_rel = "" + if not md_rel: + md_rel = f"artifacts/{artifact_id}.md" + if markdown: + safe_rel = _safe_rel(md_rel) + if safe_rel: + path = artifact_dir / safe_rel + try: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(markdown, encoding="utf-8") + except OSError as exc: + logger.warning("docgen: 制品落盘失败(不影响返回): %s", exc) + + template_key = str(result.get("template_key") or "") + template_css = str(result.get("css") or "") + tpl_display = str(result.get("template_name") or "") or ( + (template or {}).get("name", "") or (template or {}).get("title", "") + ) + html = str(result.get("html") or "") + html_rel = "" + if html: + html_rel = f"artifacts/{artifact_id}.html" + safe_rel = _safe_rel(html_rel) + if safe_rel: + path = _artifact_dir(get_settings()) / safe_rel + try: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(html, encoding="utf-8") + except OSError as exc: + logger.warning("docgen: 模版 HTML 落盘失败(不影响返回): %s", exc) + html_rel = "" + try: + quota_after = await _check_quota(db, get_settings()) + if quota_after is not None: + logger.warning("docgen: 生成后制品目录已超配额(下一次 generate/export 将被拒绝)") + except Exception as exc: + logger.warning("docgen: 生成后配额复查失败(忽略): %s", exc) + await db.execute( + "INSERT INTO doc_artifacts (id, session_id, status, title, template_name, " + "template_key, template_css, md_path, docx_path, pdf_path, html_path, letterhead, file_name, error, created_at) " + "VALUES (?, ?, 'done', ?, ?, ?, ?, ?, NULL, NULL, ?, ?, ?, NULL, ?)", + (artifact_id, session_id, title, + tpl_display, template_key, template_css, md_rel, + html_rel or None, + str(result.get("letterhead") or "")[:60], + str(result.get("file_name") or name_for_prompt(description) or title or "")[:60], + _db_now()), + ) + # 终态 progress 快照:全部节 done(限频写库可能截断最后一帧) + final_progress = [ + {**p, "state": "done"} for p in last_progress + ] + await db.execute( + "UPDATE docgen_tasks SET status = 'done', artifact_id = ?, error = NULL, " + "progress = ?, updated_at = ? WHERE id = ?", + (artifact_id, dumps_json(final_progress), _db_now(), task_id), + ) + except DocgenUnavailable as exc: + logger.warning("docgen: 任务 %s 生成服务不可用:%s", task_id, exc) + await db.execute( + "UPDATE docgen_tasks SET status = 'error', error = ?, updated_at = ? WHERE id = ?", + ("文档生成服务未就绪,请稍后重试", _db_now(), task_id), + ) + except Exception as exc: + logger.exception("docgen: 任务 %s 生成失败", task_id) + message = ( + f"文档生成失败:{exc}" if not isinstance(exc, RuntimeError) else str(exc) + ) + await db.execute( + "UPDATE docgen_tasks SET status = 'error', error = ?, updated_at = ? WHERE id = ?", + (message[:500], _db_now(), task_id), + ) + finally: + _DOCGEN_SEMAPHORE.release() + + @router.post("/generate") async def generate_doc( payload: DocgenGenerateRequest, db: Database = Depends(get_db), ): - """同步生成(600s 兜底读超时在 pipeline 内部)。 + """异步生成(缺陷 #52,2026-09-10):快路径校验 → 插任务行 → 后台生成 → 202。 - 流程:TTL 惰性清理 → 意图护栏 → 模版检索(失败→空模版集,不阻断)→ - 模版命中 → 配额检查 → 资料落库行 → pipeline 生成(资料 vstore 拉取与 - 缺失占位由服务层负责)→ 配额复查 → artifact 落库。 + 快路径(修复前同步校验链不变):description 校验 → TTL 惰性清理 → + 意图护栏 → 配额检查 → 模板检索(失败→空模版集,不阻断)→ + 资料快照(library_ids 空=全部 active)。生成主体在 + _run_docgen_task 后台执行,本端点立即返回 + 202 {ok, task_id, status:"queued"}(契约见 docs/acceptance-r3.md 附录 B)。 chat 消息落库由前端走现有 sessions 机制,本端点不碰。 """ bad = _check_description(payload.description) @@ -403,6 +623,17 @@ async def generate_doc( if quota is not None: return quota + # 并发护栏(满并发 429 BUSY 信封)。单线程事件循环中 locked() + # 与 acquire() 之间无 await 点,其它协程插不进来,无竞态窗口; + # 抢到名额才建任务,任务 finally 必释放。 + if _DOCGEN_SEMAPHORE.locked(): + return _envelope( + "BUSY", + message=f"文档生成任务已满(并发上限 {_DOCGEN_MAX_CONCURRENT}),请稍后重试", + status=429, + ) + await _DOCGEN_SEMAPHORE.acquire() + templates: list[dict[str, Any]] = [] try: templates = await search_documents(payload.description, as_template=True, top_k=20) @@ -414,10 +645,7 @@ async def generate_doc( template = _pick_template(templates, payload.template_id) - # 资料集:library_ids 为空 → 全部 active 资料;否则按给定顺序取(缺的跳过) - # 资料的 vstore 全文拉取属服务层(client.py/pipeline.py)职责:检索失败 → - # 空模版集合;document 拉取失败 → 资料缺失占位提示(不阻断生成), - # 具体兜底形态待 dev1 服务层交付后联动联测。 + # 资料快照:library_ids 为空 → 全部 active 资料;否则按给定顺序取(缺的跳过) if payload.library_ids: docs: list[dict[str, Any]] = [] for lid in payload.library_ids: @@ -430,10 +658,9 @@ async def generate_doc( ) docs = [dict(r) for r in rows] - artifact_id = str(uuid.uuid4()) + task_id = str(uuid.uuid4()) created = _db_now() title = payload.description.strip()[:60] or "未命名文档" - # 2026-09-04:库内模版样式线索(docx style_hints → css 覆盖);失败静默 None template_style: dict[str, Any] | None = None if template and template.get("id"): try: @@ -442,102 +669,69 @@ async def generate_doc( except Exception as exc: logger.warning("docgen: 模版样式提取失败(忽略): %s", exc) web_search = os.environ.get("DOCGEN_WEB_TEMPLATE_SEARCH", "1") != "0" + try: - result = await generate_document( - payload.description, - documents=docs, + await db.execute( + "INSERT INTO docgen_tasks (id, session_id, status, artifact_id, title, " + "error, progress, created_at, updated_at) " + "VALUES (?, ?, 'queued', NULL, ?, NULL, '[]', ?, ?)", + (task_id, payload.session_id or "", title, created, created), + ) + except Exception as exc: + logger.exception("docgen: 任务行插入失败") + _DOCGEN_SEMAPHORE.release() + return _envelope("INTERNAL", message=f"任务创建失败:{exc}") + + _spawn_task( + _run_docgen_task( + task_id, + db=db, + description=payload.description, + session_id=payload.session_id or "", + title=title, template=template, template_style=template_style, + docs=docs, web_search=web_search, ) - except DocgenUnavailable as exc: - logger.warning("docgen: 生成服务不可用:%s", exc) - await db.execute( - "INSERT INTO doc_artifacts (id, session_id, status, title, template_name, " - "md_path, docx_path, pdf_path, error, created_at) " - "VALUES (?, ?, 'error', ?, ?, NULL, NULL, NULL, ?, ?)", - (artifact_id, payload.session_id or "", title, - (template or {}).get("name", "") or (template or {}).get("title", ""), - "生成服务未就绪(dev1 服务层未交付)", created), - ) - return _envelope("UPSTREAM_DOWN", message="文档生成服务未就绪,请稍后重试") - except Exception as exc: - logger.exception("docgen: 生成失败") - await db.execute( - "INSERT INTO doc_artifacts (id, session_id, status, title, template_name, " - "md_path, docx_path, pdf_path, error, created_at) " - "VALUES (?, ?, 'error', ?, ?, NULL, NULL, NULL, ?, ?)", - (artifact_id, payload.session_id or "", title, - (template or {}).get("name", "") or (template or {}).get("title", ""), - str(exc)[:500], created), - ) - return _envelope("INTERNAL", message=f"文档生成失败:{exc}") - - markdown = str(result.get("markdown") or "") - md_rel = str(result.get("md_path") or f"artifacts/{artifact_id}.md") - if markdown: - safe_rel = _safe_rel(md_rel) - if safe_rel: - path = _artifact_dir(get_settings()) / safe_rel - try: - path.parent.mkdir(parents=True, exist_ok=True) - path.write_text(markdown, encoding="utf-8") - except OSError as exc: - logger.warning("docgen: 制品落盘失败(不影响返回): %s", exc) - - # 2026-09-04:模版 HTML 落盘(预览缓存件;渲染读取按 md 现渲染, - # html_path 仅作 TTL 清理键与 has_html 标记) - template_key = str(result.get("template_key") or "") - template_css = str(result.get("css") or "") - tpl_display = str(result.get("template_name") or "") or ( - (template or {}).get("name", "") or (template or {}).get("title", "") ) - html = str(result.get("html") or "") - html_rel = "" - if html: - html_rel = f"artifacts/{artifact_id}.html" - safe_rel = _safe_rel(html_rel) - if safe_rel: - path = _artifact_dir(get_settings()) / safe_rel - try: - path.parent.mkdir(parents=True, exist_ok=True) - path.write_text(html, encoding="utf-8") - except OSError as exc: - logger.warning("docgen: 模版 HTML 落盘失败(不影响返回): %s", exc) - html_rel = "" - # 生成成功后顺带复查配额(写盘后体积更准);超限仍返回成功(artifact 已存在) - try: - quota_after = await _check_quota(db, get_settings()) - if quota_after is not None: - logger.warning("docgen: 生成后制品目录已超配额(下一次 generate/export 将被拒绝)") - except Exception as exc: - logger.warning("docgen: 生成后配额复查失败(忽略): %s", exc) - await db.execute( - "INSERT INTO doc_artifacts (id, session_id, status, title, template_name, " - "template_key, template_css, md_path, docx_path, pdf_path, html_path, letterhead, file_name, error, created_at) " - "VALUES (?, ?, 'done', ?, ?, ?, ?, ?, NULL, NULL, ?, ?, ?, NULL, ?)", - (artifact_id, payload.session_id or "", title, - tpl_display, template_key, template_css, md_rel, - html_rel or None, - str(result.get("letterhead") or "")[:60], - str(result.get("file_name") or name_for_prompt(payload.description) or title or "")[:60], - created), + return JSONResponse( + status_code=202, + content={"ok": True, "task_id": task_id, "status": "queued"}, ) - return { - "ok": True, - "artifact": { - "id": artifact_id, - "title": title, - "status": "done", - "template_name": tpl_display, - "template_key": template_key, - "has_html": bool(html_rel), - "letterhead": str(result.get("letterhead") or ""), - "file_name": str(result.get("file_name") or name_for_prompt(payload.description) or title or ""), - "outline_source": str(result.get("outline_source") or ""), - "created_at": created, - }, - } + + +@router.get("/tasks/{task_id}") +async def get_docgen_task(task_id: str, db: Database = Depends(get_db)): + """任务轮询(缺陷 #52 契约,附录 B): + 200 {"ok":true,"task":{id,session_id,status,progress[],artifact|null,error}}。""" + task = await _get_task(db, task_id) + if task is None: + return _not_found(f"任务 {task_id} 不存在") + return {"ok": True, "task": task} + + +@router.get("/tasks") +async def list_docgen_tasks( + session_id: str = Query("", description="按会话过滤;空=全部(上限 100)"), + db: Database = Depends(get_db), +): + """会话任务列表(缺陷 #52 契约,附录 B)。""" + if session_id: + rows = await db.fetchall( + "SELECT id FROM docgen_tasks WHERE session_id = ? ORDER BY created_at DESC LIMIT 100", + (session_id,), + ) + else: + rows = await db.fetchall( + "SELECT id FROM docgen_tasks ORDER BY created_at DESC LIMIT 100" + ) + tasks: list[dict[str, Any]] = [] + for row in rows: + task = await _get_task(db, row["id"]) + if task is not None: + tasks.append(task) + return {"ok": True, "tasks": tasks} # --------------------------------------------------------------------------- @@ -747,8 +941,30 @@ async def export_artifact( title = str(row.get("title") or "document") try: - if fmt == "docx": - # DOCX 保持 markdown 路径(纯编辑副本;PDF 才是所见即所得) + if fmt == "docx" and DOCX_VIA_HTML: + # 缺陷 #47a(2026-09-10):DOCX 与 PDF 同构——同一份 HTML 产物 + # (render_html / render_bioflow_report)→ docx_from_html_document, + # 红头/标题/表格/图片与预览一致。转换失败回退 md2docx 旧路径。 + key = str(row.get("template_key") or "") + frozen_css = str(row.get("template_css") or "") + letterhead = str(row.get("letterhead") or "") + template_name = str(row.get("template_name") or "") + try: + if template_name == BIOFLOW_TEMPLATE_NAME: + html, _rendered_css = render_bioflow_report( + markdown, title=title, letterhead=letterhead + ) + else: + html, _rendered_css = render_html( + markdown, title=title, key=key, letterhead=letterhead + ) + data = await docx_from_html_document(html, title) + except Exception as exc: + logger.warning( + "docgen: HTML→DOCX 转换失败(%s),回退 md2docx 旧路径", exc + ) + data = await docx_from_markdown(markdown, title) + elif fmt == "docx": data = await docx_from_markdown(markdown, title) else: # 2026-09-04:PDF = 预览同源(模版 HTML+CSS → Story),所见即所得。 diff --git a/sdk/app/routers/kg.py b/sdk/app/routers/kg.py index 9f02bd8..066bf1d 100644 --- a/sdk/app/routers/kg.py +++ b/sdk/app/routers/kg.py @@ -13,13 +13,14 @@ from fastapi import APIRouter, Query from fastapi.responses import JSONResponse from ..services import neo4j as _neo4j -from ..services.neo4j import Neo4jUnavailable +from ..services.neo4j import Neo4jTimeout, Neo4jUnavailable logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/kg", tags=["kg"]) _UPSTREAM_DOWN_MSG = "知识图谱服务当前不可用,请稍后重试" +_TIMEOUT_MSG = "知识图谱查询超时,请稍后重试" def _down(exc: BaseException | None = None) -> JSONResponse: @@ -37,6 +38,29 @@ def _down(exc: BaseException | None = None) -> JSONResponse: ) +def _timeout(exc: BaseException | None = None) -> JSONResponse: + """查询超时(缺陷 #31/#32):504 TIMEOUT 信封,与实例不可用 503 区分。""" + logger.warning("kg: Neo4j 查询超时:%s", exc) + return JSONResponse( + status_code=504, + content={ + "ok": False, + "error": { + "code": "TIMEOUT", + "message": _TIMEOUT_MSG, + "degraded": True, + }, + }, + ) + + +def _fail(exc: BaseException) -> JSONResponse: + """Neo4jTimeout → 504 TIMEOUT;其余 Neo4jUnavailable → 503 UPSTREAM_DOWN。""" + if isinstance(exc, Neo4jTimeout): + return _timeout(exc) + return _down(exc) + + def _not_found() -> JSONResponse: return JSONResponse( status_code=404, @@ -56,7 +80,7 @@ async def kg_status(): try: return await _neo4j.service.kg_status() except Neo4jUnavailable as exc: - return _down(exc) + return _fail(exc) @router.get("/overview") @@ -64,7 +88,7 @@ async def kg_overview(): try: return await _neo4j.service.kg_overview() except Neo4jUnavailable as exc: - return _down(exc) + return _fail(exc) @router.get("/entities") @@ -75,7 +99,7 @@ async def kg_entities( try: results = await _neo4j.service.search_entities(q, limit=limit) except Neo4jUnavailable as exc: - return _down(exc) + return _fail(exc) return {"query": q, "results": results} @@ -84,7 +108,7 @@ async def kg_entity(entity_id: str) -> dict[str, Any]: try: entity = await _neo4j.service.entity(entity_id) except Neo4jUnavailable as exc: - return _down(exc) + return _fail(exc) if entity is None: return _not_found() return entity @@ -99,7 +123,7 @@ async def kg_neighborhood( try: result = await _neo4j.service.neighborhood(entity_id, depth=depth, limit=limit) except Neo4jUnavailable as exc: - return _down(exc) + return _fail(exc) if result.get("center") is None: return _not_found() return result diff --git a/sdk/app/routers/qa.py b/sdk/app/routers/qa.py index 54542cd..d23f7f1 100644 --- a/sdk/app/routers/qa.py +++ b/sdk/app/routers/qa.py @@ -28,7 +28,7 @@ router = APIRouter(prefix="/api/qa", tags=["qa"]) # 超时阶梯(TODO(A2→A1 对齐点):A1 的 app.config.settings 就绪后改为配置读取) CHAT_READ_TIMEOUT = 300.0 -SEARCH_READ_TIMEOUT = 180.0 +SEARCH_READ_TIMEOUT = 120.0 # 缺陷 #28:检索模式超时上限 120s(acceptance-r3 §3) INFO_READ_TIMEOUT = 5.0 CONNECT_TIMEOUT = 10.0 @@ -112,6 +112,21 @@ async def _proxy( try: resp = await client.request("POST", url, json=json_body, timeout=timeout) except httpx.ReadTimeout: + if mode == "search": + # 缺陷 #28(2026-09-10):检索模式超时返回 200 降级体(跳过 + # normalizer 直接构造),前端走「检索超时」可读提示而非错误 + # 信封。其余 mode 的错误信封不变。 + logger.warning("上游 %s 读超时,检索模式降级为 200 空结果", endpoint) + return JSONResponse( + content={ + "mode": "search", + "answer": "", + "sources": [], + "source_count": 0, + "degraded": True, + "reason": "TIMEOUT", + } + ) return _envelope("TIMEOUT", 504) except (httpx.ConnectError, httpx.ConnectTimeout, httpx.PoolTimeout) as exc: logger.warning("上游 %s 连接失败: %s", endpoint, exc) @@ -219,9 +234,14 @@ async def qa_chat_stream(request: Request): @router.post("/search") async def qa_search(request: Request): - """检索模式:强制 use_llm:false,180s 读超时,不重试。""" + """检索模式:强制 use_llm:false,180s 读超时,不重试。 + + native_chain_enabled:False(缺陷 #28,2026-09-10):与 sse_relay.py + evidence_first 先例同法——检索模式只走 Qdrant 向量检索,关闭图链, + 避免大体量库上图遍历把检索拖到超时。 + """ body = await request.json() - body = {**body, "use_llm": False} + body = {**body, "use_llm": False, "native_chain_enabled": False} return await _proxy( "search", body, read_timeout=SEARCH_READ_TIMEOUT, retry_invalid_citation=False, mode="search" ) diff --git a/sdk/app/routers/test_docgen_api.py b/sdk/app/routers/test_docgen_api.py index 99e247c..274af2e 100644 --- a/sdk/app/routers/test_docgen_api.py +++ b/sdk/app/routers/test_docgen_api.py @@ -27,6 +27,15 @@ def _run(coro): return asyncio.new_event_loop().run_until_complete(coro) +def _drain_pending(): + """跑完 generate 暂存的后台任务协程(请求返回后,fresh loop)。""" + import app.routers.docgen as mod + + pending = getattr(mod, "_pending_test_tasks", []) + while pending: + _run(pending.pop()) + + # --------------------------------------------------------------------------- # fixtures # --------------------------------------------------------------------------- @@ -57,7 +66,16 @@ def client(db, tmp_path, monkeypatch): @pytest.fixture(autouse=True) def _reset_mocks(monkeypatch): - """每个用例前把契约函数重置为「放行」形态(意图通过、服务可用)。""" + """每个用例前把契约函数重置为「放行」形态(意图通过、服务可用)。 + + 每用例独立 Semaphore(防未 drain 任务跨用例泄漏名额 → 误 429); + teardown drain 掉暂存的后台任务协程(fresh loop 里跑完,释放名额)。 + """ + import app.routers.docgen as _mod + + monkeypatch.setattr(_mod, "_DOCGEN_SEMAPHORE", asyncio.Semaphore(2)) + _mod._pending_test_tasks = [] # 属性不存在时 setattr(spawn_sync 也会初始化) + async def intent_ok(text: str) -> bool: return True @@ -69,7 +87,7 @@ def _reset_mocks(monkeypatch): async def search_ok(query, *, as_template=None, top_k=20): return [{"id": "tpl-1", "name": "实验报告模板"}] - async def generate_ok(description, *, documents, template=None): + async def generate_ok(description, *, documents, template=None, **kwargs): return {"markdown": f"# {description}", "md_path": "artifacts/gen.md"} async def edit_ok(markdown: str, instruction: str): @@ -78,7 +96,7 @@ def _reset_mocks(monkeypatch): async def docx_ok(markdown: str, title: str): return b"DOCX-BYTES" - async def pdf_ok(markdown: str, title: str): + async def pdf_ok(html: str, css: str, title: str): return b"PDF-BYTES" async def update_meta_ok(document_id: str, metadata: dict): @@ -90,16 +108,49 @@ def _reset_mocks(monkeypatch): async def rebuild_ok(): return {"rebuilt": True} + async def docx_html_ok(html: str, title: str): + return b"DOCX-HTML-BYTES" + + def render_ok(markdown, *, title, key="document", letterhead=""): + return (f"

{title}

{markdown}
", "") + + async def bioflow_render_ok(markdown, *, title, letterhead=""): + return render_ok(markdown, title=title, letterhead=letterhead) + + # 缺陷 #52:LLM env 齐备(任务守卫不误杀);缺键专项用例自行覆盖 + monkeypatch.setenv("DOCGEN_LLM_BASE_URL", "https://api.deepseek.com") + monkeypatch.setenv("DOCGEN_LLM_API_KEY", "test-key") + monkeypatch.setenv("DOCGEN_LLM_MODEL", "deepseek-chat") + # 后台任务在请求所在 loop 里无法原地跑完(TestClient 每请求一新 loop): + # spawn 时暂存协程(未绑定 loop),请求返回后由 _drain_pending 在新 loop 跑完 + docgen_mod._pending_test_tasks = [] + + def spawn_sync(coro): + docgen_mod._pending_test_tasks.append(coro) + return None + monkeypatch.setattr(docgen_mod, "is_docgen_intent", intent_ok) monkeypatch.setattr(docgen_mod, "ingest_bytes", ingest_ok) monkeypatch.setattr(docgen_mod, "search_documents", search_ok) monkeypatch.setattr(docgen_mod, "generate_document", generate_ok) monkeypatch.setattr(docgen_mod, "apply_edit", edit_ok) monkeypatch.setattr(docgen_mod, "docx_from_markdown", docx_ok) - monkeypatch.setattr(docgen_mod, "pdf_from_markdown", pdf_ok) + monkeypatch.setattr(docgen_mod, "docx_from_html_document", docx_html_ok) + monkeypatch.setattr(docgen_mod, "render_html", render_ok) + monkeypatch.setattr(docgen_mod, "render_bioflow_report", bioflow_render_ok) + monkeypatch.setattr(docgen_mod, "pdf_from_html_document", pdf_ok) monkeypatch.setattr(docgen_mod, "update_metadata", update_meta_ok) monkeypatch.setattr(docgen_mod, "delete_document", delete_ok) monkeypatch.setattr(docgen_mod, "rebuild_index", rebuild_ok) + monkeypatch.setattr(docgen_mod, "_spawn_task", spawn_sync) + yield + # teardown:drain 暂存的后台任务(释放 Semaphore 名额,防跨用例 BUSY 泄漏)。 + # 防御性 try/except:db fixture 可能先于本 fixture 关闭(fixture 逆序析构), + # 届时任务写库抛「连接已关」——只影响清理,不得掩盖 monkeypatch 恢复。 + try: + _drain_pending() + except Exception: + pass async def _upstream_down_async(*args, **kwargs): @@ -157,21 +208,46 @@ def test_library_upload_vstore_down_503(client, monkeypatch): # --------------------------------------------------------------------------- def test_generate_ok(client, db, tmp_path): + """缺陷 #52 契约:generate → 202 {ok,task_id,status:"queued"};任务同步跑完后 + 轮询 done 且 artifact 回填(测试内 _spawn_task 同步执行)。""" resp = client.post( "/api/docgen/generate", json={"description": "写一份水稻实验报告", "session_id": "s1", "library_ids": [], "template_id": "tpl-1"}, ) - assert resp.status_code == 200, resp.text + assert resp.status_code == 202, resp.text body = resp.json() assert body["ok"] is True - aid = body["artifact"]["id"] - assert body["artifact"]["status"] == "done" - assert body["artifact"]["template_name"] == "实验报告模板" + task_id = body["task_id"] + assert body["status"] == "queued" + assert len(task_id) == 36 # uuid4 + + _drain_pending() + + # 轮询任务(后台任务已跑完 → done + artifact 回填) + resp = client.get(f"/api/docgen/tasks/{task_id}") + assert resp.status_code == 200, resp.text + task = resp.json()["task"] + assert task["id"] == task_id + assert task["session_id"] == "s1" + assert task["status"] == "done" + assert task["error"] is None + artifact = task["artifact"] + assert artifact is not None + assert artifact["template_name"] == "实验报告模板" + assert artifact["has_html"] is False + assert artifact["letterhead"] == "" + assert artifact["file_name"] + + # 会话任务列表 + resp = client.get("/api/docgen/tasks", params={"session_id": "s1"}) + assert resp.status_code == 200 + assert [t["id"] for t in resp.json()["tasks"]] == [task_id] + # 落库行 def _check(): async def inner(): - row = await db.fetchone("SELECT * FROM doc_artifacts WHERE id = ?", (aid,)) + row = await db.fetchone("SELECT * FROM doc_artifacts WHERE id = ?", (artifact["id"],)) assert row is not None assert row["status"] == "done" assert row["session_id"] == "s1" @@ -203,7 +279,7 @@ def test_generate_search_failure_skips_template(client, db, monkeypatch): async def search_down(query, *, as_template=None, top_k=20): raise docgen_mod.DocgenUnavailable("vstore down") - async def generate_capture(description, *, documents, template=None): + async def generate_capture(description, *, documents, template=None, **kwargs): calls["template"] = template calls["docs"] = documents return {"markdown": "# 无模版生成", "md_path": "artifacts/gen2.md"} @@ -211,23 +287,31 @@ def test_generate_search_failure_skips_template(client, db, monkeypatch): monkeypatch.setattr(docgen_mod, "search_documents", search_down) monkeypatch.setattr(docgen_mod, "generate_document", generate_capture) resp = client.post("/api/docgen/generate", json={"description": "写一份总结"}) - assert resp.status_code == 200 + assert resp.status_code == 202 + task_id = resp.json()["task_id"] + _drain_pending() + # 后台任务已跑完:确认模版空集 + 空库快照透传给了生成主体 assert calls["template"] is None assert calls["docs"] == [] # 空库 + task = client.get(f"/api/docgen/tasks/{task_id}").json()["task"] + assert task["status"] == "done" -def test_generate_pipeline_unavailable_503_and_error_row(client, db, monkeypatch): +def test_generate_pipeline_unavailable_task_error(client, db, monkeypatch): + """缺陷 #52:生成服务不可用 → 202 后任务落 error(中文可读),不落 artifact。""" monkeypatch.setattr(docgen_mod, "generate_document", _upstream_down_async) resp = client.post("/api/docgen/generate", json={"description": "写一份总结"}) - assert resp.status_code == 503 - assert resp.json()["error"]["code"] == "UPSTREAM_DOWN" - # error 行已落库(前端可查状态) + assert resp.status_code == 202 + task_id = resp.json()["task_id"] + _drain_pending() + task = client.get(f"/api/docgen/tasks/{task_id}").json()["task"] + assert task["status"] == "error" + assert task["error"] + assert task["artifact"] is None def _check(): async def inner(): rows = await db.fetchall("SELECT * FROM doc_artifacts") - assert len(rows) == 1 - assert rows[0]["status"] == "error" - assert rows[0]["error"] + assert len(rows) == 0 return _run(inner()) _check() @@ -239,8 +323,11 @@ def test_generate_pipeline_unavailable_503_and_error_row(client, db, monkeypatch def _make_artifact(client) -> str: resp = client.post("/api/docgen/generate", json={"description": "写一份测试文档"}) - assert resp.status_code == 200, resp.text - return resp.json()["artifact"]["id"] + assert resp.status_code == 202, resp.text + _drain_pending() + task = client.get(f"/api/docgen/tasks/{resp.json()['task_id']}").json()["task"] + assert task["status"] == "done", task + return task["artifact"]["id"] def test_artifact_content_ok(client): @@ -280,8 +367,9 @@ def test_artifact_edit_ok(client): assert body["markdown"].endswith("编辑完成") -def test_artifact_edit_intent_rejected(client, monkeypatch): - """护栏机制:route 判 qa → 422(monkeypatch route;真实判定见新增用例)。""" +def test_artifact_edit_no_intent_guard(client, monkeypatch): + """2026-09-04 语义:edit 端点护栏已移除(被调用本身即活动文档上下文), + route 判 qa 也放行(HEAD 4749f77)。""" aid = _make_artifact(client) def route_no(text, has_attachments, has_active_artifact): @@ -289,8 +377,8 @@ def test_artifact_edit_intent_rejected(client, monkeypatch): monkeypatch.setattr(docgen_mod, "route", route_no) resp = client.post(f"/api/docgen/artifacts/{aid}/edit", json={"instruction": "帮我查个文献"}) - assert resp.status_code == 422 - assert resp.json()["error"]["code"] == "INTENT_REJECTED" + assert resp.status_code == 200, resp.text + assert resp.json()["ok"] is True def test_artifact_edit_unavailable_503(client, monkeypatch): @@ -314,11 +402,11 @@ def test_export_docx_and_download(client, tmp_path): assert body["ok"] is True assert body["download_url"].endswith(f"/api/docgen/artifacts/{aid}/download?format=docx") # 落盘 - assert (tmp_path / "docgen" / "artifacts" / f"{aid}.docx").read_bytes() == b"DOCX-BYTES" + assert (tmp_path / "docgen" / "artifacts" / f"{aid}.docx").read_bytes() == b"DOCX-HTML-BYTES" # 下载 resp = client.get(f"/api/docgen/artifacts/{aid}/download?format=docx") assert resp.status_code == 200 - assert resp.content == b"DOCX-BYTES" + assert resp.content == b"DOCX-HTML-BYTES" assert resp.headers["content-disposition"].startswith("attachment;") assert resp.headers["content-type"].startswith("application/vnd.openxmlformats") @@ -342,7 +430,9 @@ def test_export_invalid_format_422(client): def test_export_unavailable_503(client, monkeypatch): + """#47a 语义:HTML 转换失败回退 md2docx;两条路径都不可用才 503。""" aid = _make_artifact(client) + monkeypatch.setattr(docgen_mod, "docx_from_html_document", _upstream_down_async) monkeypatch.setattr(docgen_mod, "docx_from_markdown", _upstream_down_async) resp = client.post(f"/api/docgen/artifacts/{aid}/export", json={"format": "docx"}) assert resp.status_code == 503 @@ -518,7 +608,7 @@ def test_check_quota_passes_when_under_limit(client, db, tmp_path): """配额充足(默认 50GB 设置走真实 get_settings 时仅用 tmp 目录)→ 不拒绝。""" aid = _make_artifact(client) # 默认设置下生成成功即证明配额未误拦 resp = client.post("/api/docgen/generate", json={"description": "再写一份测试报告"}) - assert resp.status_code == 200 + assert resp.status_code == 202 resp2 = client.post(f"/api/docgen/artifacts/{aid}/export", json={"format": "docx"}) assert resp2.status_code == 200 @@ -553,7 +643,7 @@ def test_generate_prunes_expired_artifacts(client, db, tmp_path): (tmp_path / "docgen" / "artifacts" / fname).write_text("x", encoding="utf-8") resp = client.post("/api/docgen/generate", json={"description": "写一份测试报告"}) - assert resp.status_code == 200 + assert resp.status_code == 202 def _check(): async def inner(): @@ -580,7 +670,7 @@ def test_prune_expired_missing_files_tolerated(client, db, tmp_path): # 不创建任何文件 resp = client.post("/api/docgen/generate", json={"description": "写一份测试报告"}) - assert resp.status_code == 200 + assert resp.status_code == 202 def _check(): async def inner(): @@ -619,9 +709,94 @@ def test_export_invalid_format_still_422(client): assert resp.json()["error"]["code"] == "VALIDATION" -def test_edit_endpoint_qa_instruction_rejected_real_route(client): - """edit 端点传问答类指令:route 判 qa(EDIT_WORD 不命中)→ 422 护栏仍生效。""" +def test_edit_endpoint_qa_instruction_passes_real_route(client): + """2026-09-04 语义:edit 端点无护栏(HEAD 4749f77),问答类指令也走编辑链路。""" aid = _make_artifact(client) resp = client.post(f"/api/docgen/artifacts/{aid}/edit", json={"instruction": "帮我查个文献"}) - assert resp.status_code == 422 - assert resp.json()["error"]["code"] == "INTENT_REJECTED" + assert resp.status_code == 200, resp.text + assert resp.json()["ok"] is True + + +# --------------------------------------------------------------------------- +# 缺陷 #52 异步任务化专项 +# --------------------------------------------------------------------------- + +def test_generate_returns_202_and_task_roundtrip(client, db): + resp = client.post("/api/docgen/generate", json={"description": "写一份异步测试文档", "session_id": "s52"}) + assert resp.status_code == 202 + body = resp.json() + assert set(body.keys()) == {"ok", "task_id", "status"} + assert body["ok"] is True and body["status"] == "queued" + # 任务不存在 → 404 + resp = client.get("/api/docgen/tasks/does-not-exist") + assert resp.status_code == 404 + _drain_pending() + # 任务详情契约字段(附录 B) + task = client.get(f"/api/docgen/tasks/{body['task_id']}").json()["task"] + assert task["status"] == "done" + assert isinstance(task["progress"], list) + assert task["artifact"]["title"] + + +def test_generate_missing_llm_env_task_error_names_keys(client, db, monkeypatch): + """DOCGEN_LLM_* 不齐 → 任务 error 写明缺哪个键。""" + monkeypatch.delenv("DOCGEN_LLM_API_KEY", raising=False) + resp = client.post("/api/docgen/generate", json={"description": "写一份测试文档"}) + assert resp.status_code == 202 + _drain_pending() + task = client.get(f"/api/docgen/tasks/{resp.json()['task_id']}").json()["task"] + assert task["status"] == "error" + assert "DOCGEN_LLM_API_KEY" in (task["error"] or "") + + +def test_generate_busy_429_when_semaphore_full(client, db, monkeypatch): + """满并发 → 429 BUSY 信封(快路径不建任务)。""" + import asyncio + + # 意图护栏放行(本用例只关心并发名额,不测 route 判型) + monkeypatch.setattr(docgen_mod, "route", lambda *a, **k: "docgen") + monkeypatch.setattr(docgen_mod, "_DOCGEN_SEMAPHORE", asyncio.Semaphore(1)) + resp1 = client.post("/api/docgen/generate", json={"description": "写一份占用名额"}) + assert resp1.status_code == 202 + # 第二个请求:满并发(名额被第一个任务占用) + resp2 = client.post("/api/docgen/generate", json={"description": "再写一份"}) + assert resp2.status_code == 429 + body = resp2.json() + assert body["ok"] is False + assert body["error"]["code"] == "BUSY" + # 跑完第一个任务(finally 释放名额),任务终态 done + _drain_pending() + task = client.get(f"/api/docgen/tasks/{resp1.json()['task_id']}").json()["task"] + assert task["status"] == "done" + + +def test_startup_cleanup_marks_running_tasks_error(db, tmp_path): + """服务重启:running/queued → status='error', error='服务重启,任务中断,请重试'。""" + from app.routers import docgen as docgen_mod2 + + # 直接写一个 running 行(不经 API) + def _seed(): + async def inner(): + now = "2026-09-10T00:00:00Z" + await db.execute( + "INSERT INTO docgen_tasks (id, session_id, status, artifact_id, title, " + "error, progress, created_at, updated_at) VALUES " + "('t-running', 's', 'running', NULL, 't', NULL, '[]', ?, ?)", + (now, now), + ) + return _run(inner()) + + _seed() + # 模拟重启:重新 connect()(幂等迁移 + 启动清理) + database = Database(db.path) + _run(database.connect()) + + def _check(): + async def inner(): + row = await db.fetchone("SELECT * FROM docgen_tasks WHERE id = 't-running'") + assert row["status"] == "error" + assert row["error"] == "服务重启,任务中断,请重试" + return _run(inner()) + + _check() + _run(database.close()) diff --git a/sdk/app/services/docgen/converter.py b/sdk/app/services/docgen/converter.py index d666d00..83f9f4b 100644 --- a/sdk/app/services/docgen/converter.py +++ b/sdk/app/services/docgen/converter.py @@ -31,6 +31,16 @@ async def pdf_from_markdown(markdown: str, title: str) -> bytes: return await anyio.to_thread.run_sync(_pdf_bytes, markdown, title) +async def docx_from_html_document(html: str, title: str) -> bytes: + """HTML → docx(缺陷 #47a,2026-09-10):与 converter 其它函数同签名, + html2docx 是 CPU 同步转换,to_thread 避免阻塞事件循环。""" + from .html2docx import docx_from_html_document as _sync + + with tempfile.TemporaryDirectory() as tmp: + out = _sync(html, title=title, out_path=Path(tmp) / "document.docx") + return out.read_bytes() + + async def pdf_from_html_document(html: str, css: str, title: str) -> bytes: """2026-09-04:模版 HTML + css → pdf bytes(async,预览同源导出)。""" import anyio diff --git a/sdk/app/services/docgen/html2docx.py b/sdk/app/services/docgen/html2docx.py new file mode 100644 index 0000000..84ed273 --- /dev/null +++ b/sdk/app/services/docgen/html2docx.py @@ -0,0 +1,339 @@ +"""services/docgen/html2docx.py —— 自家 render_html 产物 → Word(缺陷 #47a,2026-09-10)。 + +与 md2docx 的关系:md2docx 是 markdown 行级状态机(旧路径,回退用); +本模块是「HTML 同源转换」——把 render_html/render_bioflow_report 生成的 +HTML(与 PDF 导出/预览同一份产物)用 BeautifulSoup 遍历成 python-docx 文档, +保证 DOCX 与 PDF 预览版式一致(红头/标题层级/表格/图片同源)。 + +样式约定(与模板 css 视觉对齐): +- .doc-letterhead:红字(RGBColor(255,0,0))居中、22pt 黑体(红头文件要件) +- h1/h2/h3:黑体分级字号 22/16/14pt 加粗(>h3 折叠到 h3) +- table:Table Grid + 表头灰底(复用 md2docx 的 _shade_cell) +- figure/img:内嵌图片(拉取失败落 alt 文本),URL 经 BFF 代理端点绝对化 +- 列表/段落/加粗/斜体/行内码:递归 run 组装 +- eastAsia 字体逻辑复用 md2docx 的 _set_run_font 思路(本地副本,避免 + 循环依赖 md2docx 私有函数) +""" +from __future__ import annotations + +import logging +import re +from pathlib import Path +from typing import Any + +import httpx +from bs4 import BeautifulSoup, NavigableString, Tag +from docx import Document +from docx.enum.text import WD_ALIGN_PARAGRAPH +from docx.oxml import OxmlElement +from docx.oxml.ns import qn +from docx.shared import Inches, Pt, RGBColor + +logger = logging.getLogger(__name__) + +# 字体:正文宋体 / 标题黑体(与 md2docx 一致;Windows 中文环境标准) +_FONT_BODY = "宋体" +_FONT_HEADING = "黑体" + +_HEADING_SIZES = {1: 22, 2: 16, 3: 14} +_LETTERHEAD_SIZE = 22.0 +_IMAGE_MAX_WIDTH_IN = 6.0 # A4 正文可用宽度约 6 英寸 + +# 图片 URL 相对路径 → BFF 代理绝对化(与服务内部 base_url 一致) +_FIGURE_PROXY = "/api/figures/proxy" +_IMAGE_HTTP_TIMEOUT = 8.0 +_IMAGE_MAX_BYTES = 20 * 1024 * 1024 + +# 已支持元素外的顶层块元素一律按段落处理(模板 main 内预期就是这些) +_BLOCK_TAGS = { + "h1", "h2", "h3", "h4", "h5", "h6", + "p", "div", "li", "blockquote", "pre", "table", "figure", "img", + "ol", "ul", "section", "article", "header", "footer", "main", +} + +_VOID_TAGS = {"img", "br", "hr", "meta", "link"} + + +def _set_run_font(run, font_name: str, size_pt: float | None = None, + bold: bool | None = None, italic: bool | None = None, + color: RGBColor | None = None) -> None: + """设置西文 + 中文字体(font.name 与 w:eastAsia 同时设置;与 md2docx 同思路)。""" + run.font.name = font_name + rpr = run._element.get_or_add_rPr() + rfonts = rpr.find(qn("w:rFonts")) + if rfonts is None: + rfonts = OxmlElement("w:rFonts") + rpr.append(rfonts) + rfonts.set(qn("w:eastAsia"), font_name) + if size_pt is not None: + run.font.size = Pt(size_pt) + if bold is not None: + run.font.bold = bold + if italic is not None: + run.font.italic = italic + if color is not None: + run.font.color.rgb = color + + +def _shade_cell(cell, fill: str) -> None: + """单元格底色(表头灰底;与 md2docx 同实现)。""" + tc_pr = cell._tc.get_or_add_tcPr() + shd = OxmlElement("w:shd") + shd.set(qn("w:val"), "clear") + shd.set(qn("w:color"), "auto") + shd.set(qn("w:fill"), fill) + tc_pr.append(shd) + + +def _abs_image_url(src: str, base_url: str = "") -> str: + """图片 URL 绝对化:fig:// 或相对路径 → BFF 代理端点。""" + if not src: + return "" + if src.startswith("data:"): + return src + if src.startswith("fig://"): + return f"{base_url}{_FIGURE_PROXY}?url={src}" + if src.startswith("/"): + return f"{base_url}{src}" + if src.startswith(("http://", "https://")): + return src + return f"{base_url}{_FIGURE_PROXY}?url={src}" + + +def _collect_direct_text(node: Tag) -> str: + """节点的直接文本(不含后代元素),用于 alt 兜底。""" + return "".join( + str(child) for child in node.children if isinstance(child, NavigableString) + ).strip() + + +def _heading_level(tag_name: str) -> int: + m = re.match(r"^h([1-6])$", tag_name) + return min(int(m.group(1)), 3) if m else 0 # >h3 折叠到 h3 + + +def _add_inline_runs(paragraph, node, font_name: str, size_pt: float, + bold: bool | None = None) -> None: + """递归组装行内 run://// 与裸文本。""" + if isinstance(node, NavigableString): + text = str(node) + if text: + paragraph.add_run(text) + return + if not isinstance(node, Tag): + return + name = node.name.lower() if node.name else "" + if name in ("script", "style"): + return + if name in ("br",): + paragraph.add_run().add_break() + return + if name in ("b", "strong"): + for child in node.children: + _add_inline_runs(paragraph, child, font_name, size_pt, bold=True) + return + if name in ("em", "i"): + for child in node.children: + _add_inline_runs(paragraph, child, font_name, size_pt, bold) + return + if name == "code": + run = paragraph.add_run(node.get_text()) + _set_run_font(run, "Courier New", max(size_pt - 1.0, 8.0), bold=bold) + return + if name in ("span", "a", "small", "sub", "sup"): + for child in node.children: + _add_inline_runs(paragraph, child, font_name, size_pt, bold) + return + # 未知行内元素:递归子节点(避免把 / 后代塌成纯文本) + for child in node.children: + _add_inline_runs(paragraph, child, font_name, size_pt, bold) + + +def _add_rich_block(doc: Document, node: Tag, list_level: int = 0) -> None: + """把一个 HTML 块级元素转成 docx 段落/表格/图片。返回是否产出了内容。""" + name = (node.name or "").lower() + text = _collect_direct_text(node) + + # 标题 + level = _heading_level(name) + if level: + para = doc.add_heading("", level=level) + run = para.add_run(node.get_text(" ", strip=True)) + _set_run_font(run, _FONT_HEADING, _HEADING_SIZES[level], bold=True) + para.paragraph_format.space_before = Pt(10) + return True + + if name in ("ul", "ol"): + ordered = name == "ol" + for child in node.children: + if not isinstance(child, Tag) or (child.name or "").lower() != "li": + continue + style = "List Number" if ordered else "List Bullet" + para = doc.add_paragraph(style=style) + para.paragraph_format.left_indent = Inches(0.25 * (list_level + 1)) + _add_inline_runs(para, child, _FONT_BODY, 12.0) + return True + + if name == "li": # 游离 li(无 ul/ol 父) + para = doc.add_paragraph(style="List Bullet") + para.paragraph_format.left_indent = Inches(0.25 * (list_level + 1)) + _add_inline_runs(para, node, _FONT_BODY, 12.0) + return True + + if name == "table": + _add_table(doc, node) + return True + + if name in ("figure", "figcaption") or (name == "div" and node.find("img")): + for img in node.find_all("img", recursive=False) or node.find_all("img"): + _add_image(doc, img) + if name == "figure" and not node.find("img"): + _add_image(doc, node.img) # type: ignore[arg-type] + # figcaption 文本 + caption = node.find("figcaption") + if caption is not None: + para = doc.add_paragraph() + para.alignment = WD_ALIGN_PARAGRAPH.CENTER + run = para.add_run(caption.get_text(" ", strip=True)) + _set_run_font(run, _FONT_BODY, 10.5) + return True + + if name == "img": + _add_image(doc, node) + return True + + if name == "pre": + para = doc.add_paragraph() + run = para.add_run(node.get_text()) + _set_run_font(run, "Courier New", 10.5) + return True + + if name == "blockquote": + para = doc.add_paragraph(style="Quote") + _add_inline_runs(para, node, _FONT_BODY, 12.0) + return True + + if name in ("hr", "br", "script", "style", "meta", "link"): + return False + + # 默认段落(p/div/section 等):直接文本优先,无直接文本则全文(避免丢内容) + para = doc.add_paragraph() + if text: + _add_inline_runs(para, node, _FONT_BODY, 12.0) + else: + body_text = node.get_text(" ", strip=True) + if body_text: + run = para.add_run(body_text) + _set_run_font(run, _FONT_BODY, 12.0) + return True + + +def _add_table(doc: Document, table_el: Tag) -> None: + """ → Table Grid;首行灰底加粗(模板表格约定)。""" + rows: list[list[str]] = [] + trs = table_el.find_all("tr") + for tr in trs: + cells = tr.find_all(["td", "th"]) if tr.find(["td", "th"]) else [] + rows.append([c.get_text(" ", strip=True) for c in cells]) + rows = [r for r in rows if any(c.strip() for c in r)] + if not rows: + return + width = max(len(r) for r in rows) + rows = [r + [""] * (width - len(r)) for r in rows] + table = doc.add_table(rows=len(rows), cols=width) + table.style = "Table Grid" + for r_idx, row in enumerate(rows): + for c_idx, cell_text in enumerate(row): + cell = table.cell(r_idx, c_idx) + cell.text = "" + para = cell.paragraphs[0] + run = para.add_run(cell_text) + _set_run_font(run, _FONT_BODY, 10.5) + if r_idx == 0: # 表头:加粗灰底(与 md2docx 一致) + _set_run_font(run, _FONT_BODY, 10.5, bold=True) + _shade_cell(cell, "D9D9D9") + + +def _add_image(doc: Document, img: Tag) -> None: + """ → 内嵌图片(拉取失败落 alt 文本)。""" + src = (img.get("src") or "") if isinstance(img, Tag) else "" + alt = (img.get("alt") or "") if isinstance(img, Tag) else "" + para = doc.add_paragraph() + para.alignment = WD_ALIGN_PARAGRAPH.CENTER + if not src: + if alt: + run = para.add_run(f"[图片缺失:{alt}]") + _set_run_font(run, _FONT_BODY, 10.5) + return + try: + if src.startswith("data:"): + import base64 + + m = re.match(r"^data:image/[^;]+;base64,(.+)$", src, re.IGNORECASE | re.DOTALL) + if m: + data = base64.b64decode(m.group(1)) + else: + raise ValueError("非 base64 图片 data URL") + else: + resp = httpx.get(_abs_image_url(src), timeout=_IMAGE_HTTP_TIMEOUT) + resp.raise_for_status() + if len(resp.content) > _IMAGE_MAX_BYTES: + raise ValueError(f"图片超过 {_IMAGE_MAX_BYTES // (1024 * 1024)}MB 上限") + data = resp.content + from io import BytesIO + + run = para.add_run() + run.add_picture(BytesIO(data), width=Inches(_IMAGE_MAX_WIDTH_IN)) + except Exception as exc: + logger.info("docgen: DOCX 图片内嵌失败(落 alt 文本): %s", exc) + run = para.add_run(f"[图片缺失:{alt or src}]") + _set_run_font(run, _FONT_BODY, 10.5) + + +def docx_from_html_document(html: str, title: str = "", out_path: str | Path | None = None) -> Path: + """HTML → docx。返回写入路径(out_path 为 None 时按 title 命名到当前目录)。 + + - 红头:.doc-letterhead 红字(RGBColor(255,0,0))居中 22pt 黑体 + - h1-h3:黑体 22/16/14pt 加粗 + - 表格 Table Grid + 表头灰底;figure/img 内嵌(失败落 alt) + - 空 HTML 产出空文档(不抛) + """ + doc = Document() + soup = BeautifulSoup(html or "", "html.parser") + + # 标题: 优先(render_html 已回填 {{TITLE}}) + if not title: + t = soup.find("title") + title = t.get_text(strip=True) if t else "" + doc.core_properties.title = (title or "document")[:200] + + body = soup.body or soup + # 红头(.doc-letterhead):红字居中 22pt 黑体;空串不产出段落 + letterhead_el = body.find(class_="doc-letterhead") + letterhead = "" + if letterhead_el is not None: + letterhead = letterhead_el.get_text(" ", strip=True) + if letterhead: + para = doc.add_paragraph() + para.alignment = WD_ALIGN_PARAGRAPH.CENTER + run = para.add_run(letterhead) + _set_run_font(run, _FONT_HEADING, _LETTERHEAD_SIZE, bold=True, + color=RGBColor(255, 0, 0)) + + # 主内容容器(render_html 用 <main> 包 body;bioflow 模板同理) + main = body.find("main") or body + # 遍历 main 的直接子元素(跳过已处理的红头与 meta) + for child in main.children: + if not isinstance(child, Tag): + continue + name = (child.name or "").lower() + if name in ("script", "style", "meta", "link", "title"): + continue + if "doc-letterhead" in (child.get("class") or []): + continue # 红头已处理 + _add_rich_block(doc, child) + + out = Path(out_path) if out_path is not None else Path(f"{title or 'document'}.docx") + out.parent.mkdir(parents=True, exist_ok=True) + doc.save(out) + return out diff --git a/sdk/app/services/docgen/pipeline.py b/sdk/app/services/docgen/pipeline.py index 924551d..f35e31d 100644 --- a/sdk/app/services/docgen/pipeline.py +++ b/sdk/app/services/docgen/pipeline.py @@ -18,6 +18,7 @@ import logging import re import uuid from pathlib import Path +from collections.abc import Awaitable, Callable from typing import Any from . import client, llm @@ -287,6 +288,7 @@ async def generate_document( template: dict[str, Any] | None = None, template_style: dict[str, Any] | None = None, web_search: bool = True, + on_progress: Callable[[list[dict[str, Any]]], Awaitable[None]] | None = None, ) -> dict[str, Any]: """文档生成主流程(2026-09-04 起:拼内容 → 套模版 → 产 HTML)。 @@ -343,9 +345,30 @@ async def generate_document( # 5. 逐节生成合并(单节失败重试 2 次仍失败 → 占位) # FAILOVER-02 修复:LLM 输出自带同文本标题时不再重复插入 + # 节级进度回调(缺陷 #52,2026-09-10):progress 列表 [{section,title,state}] + # 由本函数构建,router 落 docgen_tasks.progress(限频) + progress: list[dict[str, Any]] = [ + {"section": i, "title": t, "state": "pending"} + for i, t in enumerate(sections, start=1) + ] + + async def _emit(state: str, idx: int | None = None) -> None: + if on_progress is None: + return + cur = [dict(p) for p in progress] + if idx is not None: + for p in cur: + if p["section"] < idx: + p["state"] = "done" + elif p["section"] == idx: + p["state"] = state + await on_progress(cur) + + await _emit("running", 0) # 大纲完成,逐节生成开始 parts: list[str] = [f"# {title}"] written = "" - for section_title in sections: + for idx, section_title in enumerate(sections, start=1): + await _emit("running", idx) try: section_md = await _generate_section(description, title, section_title, materials, written) except llm.LLMError as exc: @@ -353,6 +376,8 @@ async def generate_document( block = _merge_section(section_title, section_md) parts.append(block) written += f"\n{block[:2000]}\n" + await _emit("done", idx) + await _emit("done", len(sections) + 1) markdown = "\n\n".join(parts) diff --git a/sdk/app/services/figures.py b/sdk/app/services/figures.py index 40923ad..3902f23 100644 --- a/sdk/app/services/figures.py +++ b/sdk/app/services/figures.py @@ -13,6 +13,7 @@ from __future__ import annotations import logging +import os import re import time from pathlib import Path @@ -23,11 +24,20 @@ import httpx logger = logging.getLogger(__name__) -# 本地 MinerU 图片根(探索实测路径) -MINERU_ROOT = Path("/home/xzq/5090/teamWorkspace/zh/PlantAI-rag/data/evaluation/methodology_upload_jobs") +# 本地 MinerU 图片根 / 元数据注册表路径(缺陷 #37/#38,2026-09-10 起 env 化: +# PLANTAI_FIGURES_MINERU_ROOT / PLANTAI_FIGURES_REGISTRY 覆盖,默认值保持 175 现路径) +MINERU_ROOT = Path( + os.environ.get( + "PLANTAI_FIGURES_MINERU_ROOT", + "/home/xzq/5090/teamWorkspace/zh/PlantAI-rag/data/evaluation/methodology_upload_jobs", + ) +) REGISTRY_PATH = Path( - "/home/xzq/5090/teamWorkspace/zh/PlantAI-rag/data/knowledge_base/" - "paper_metadata_registry_v1_20260717/paper_metadata_registry_full.jsonl" + os.environ.get( + "PLANTAI_FIGURES_REGISTRY", + "/home/xzq/5090/teamWorkspace/zh/PlantAI-rag/data/knowledge_base/" + "paper_metadata_registry_v1_20260717/paper_metadata_registry_full.jsonl", + ) ) # PMC 外网端点 EPMC_FULLTEXT = "https://www.ebi.ac.uk/europepmc/webservices/rest/{pmcid}/fullTextXML" diff --git a/sdk/app/services/neo4j.py b/sdk/app/services/neo4j.py index a0915d7..e79a98a 100644 --- a/sdk/app/services/neo4j.py +++ b/sdk/app/services/neo4j.py @@ -4,7 +4,8 @@ - 首选 neo4j-plant-srt-direct-v1-full(bolt 127.0.0.1:18788,db `neo4j`, 用户 neo4j / 密码在 sdk/.env),schema canonical,QA 7920 生产使用。 - 回退 neo4j-plant-srt-pc2025-v3-20260817-normalized(bolt 127.0.0.1:19788), - schema 不同(Entity/ASSERTS),本波未做 Cypher 适配(TODO)。 + schema normalized(Entity/ASSERTS,2026-09-10 缺陷 #32 起支持查询;该库无 name_norm 索引, + 等值/前缀查询为全标签扫描,176 需先跑 deploy/qa-db/create_neo4j_indexes.sh)。 - 再次 plantai-protected-neo4j-1(bolt 127.0.0.1:17687),空库。 所有本模块异常统一抛 :class:`Neo4jUnavailable`,由路由层转 503 降级信封。 @@ -34,6 +35,7 @@ NEIGHBORHOOD_QUERY_TIMEOUT = 30.0 OVERVIEW_QUERY_TIMEOUT = 60.0 # 2026-08-25 PM 修复:1644 万边 COUNT 聚合冷缓存 >10s NAME_INDEX = "srt2_canonical_entity_name_norm" +NORMALIZED_NAME_INDEX = "plantai_normalized_entity_name_norm" # create_neo4j_indexes.sh 创建 FULLTEXT_INDEX = None # DISCOVERY.md §2:任何实例都不存在全文索引,且不得创建 # status 展示名(须与 API.md §6 fixture 一致) @@ -43,18 +45,27 @@ PROTECTED_ACTIVE_NAME = "plantai-protected-neo4j-1" _CANONICAL_LABELS = ["SRT2CanonicalEntity", "SRT2CanonicalAssertion", "SRT2Paper"] _CANONICAL_REL_TYPES = ["SRT2_CANONICAL_ASSERTS", "SRT2_SUPPORTED_BY"] +# normalized 实例(侦察 2026-09-10 实测:Entity 634 万 / Paper 9.7 万 / +# ASSERTS 1304 万边,SUPPORTED_BY 1309 万边) +_NORMALIZED_LABELS = ["Entity", "Paper"] +_NORMALIZED_REL_TYPES = ["ASSERTS", "SUPPORTED_BY"] class Neo4jUnavailable(Exception): - """Neo4j 上游不可用(连接失败/认证失败/超时/Cypher 错误统一转此异常)。""" + """Neo4j 上游不可用(连接失败/认证失败/Cypher 错误统一转此异常)。""" def __init__(self, message: str, *, cause: BaseException | None = None) -> None: super().__init__(message) self.cause = cause +class Neo4jTimeout(Neo4jUnavailable): + """Neo4j 查询超时(缺陷 #31/#32,2026-09-10):与连接/驱动错误区分, + 路由层映射 504 TIMEOUT 信封(区别于 503 UPSTREAM_DOWN)。""" + + class _Instance: - """一个候选实例。本波只实现 canonical schema(direct 实例)的 Cypher。""" + """一个候选实例。canonical 与 normalized schema 均已实现 Cypher(其余报未适配)。""" __slots__ = ("name", "uri", "auth", "database", "schema", "label", "rel") @@ -100,8 +111,8 @@ _INSTANCES: list[_Instance] = [ _env("PASSWORD_NORMALIZED", ""), "neo4j", "normalized", - None, - None, + "Entity", + "ASSERTS", ), _Instance( PROTECTED_ACTIVE_NAME, @@ -116,6 +127,61 @@ _INSTANCES: list[_Instance] = [ ] +def _normalized_neighborhood_cypher(depth: int) -> str: + """normalized 实例(Entity/ASSERTS)的邻域 Cypher(缺陷 #32,2026-09-10)。 + + 与 canonical 分支结构一致(内联投影 source/target,复用同一套边组装), + 差别只在:Entity/ASSERTS 标签、按 relation_category 排序(ASSERTS 边 + 无 paper_count/support_count/title/doi/pmid 属性,侦察 2026-09-10 实测)。 + + 调用前必须已做 degree 预检(hub degree > 300 不执行本查询, + 见 neighborhood() 的 normalized 分支)。 + """ + depth = depth if depth in (1, 2) else 1 + if depth == 1: + return """ + MATCH (s:Entity {canonical_id: $seed}) + OPTIONAL MATCH (s)-[r1:ASSERTS]-(n1:Entity) + WITH s, r1, n1 + ORDER BY r1.relation_category ASC + WITH s, + collect(DISTINCT n1)[..$max_nodes] AS level1, + [r IN collect(DISTINCT r1)[..$max_edges] | r {.*, + source_canonical_id: CASE WHEN startNode(r).canonical_id = s.canonical_id + THEN s.canonical_id ELSE endNode(r).canonical_id END, + target_canonical_id: CASE WHEN startNode(r).canonical_id = s.canonical_id + THEN endNode(r).canonical_id ELSE s.canonical_id END}] AS edges1 + RETURN s {.*} AS start, + [n IN level1 | n {.*}] AS nodes, + edges1 AS edges + """ + return """ + MATCH (s:Entity {canonical_id: $seed}) + OPTIONAL MATCH (s)-[r1:ASSERTS]-(n1:Entity) + WITH s, r1, n1 + ORDER BY r1.relation_category ASC + WITH s, + collect(DISTINCT n1)[..$max_nodes] AS level1, + [r IN collect(DISTINCT r1)[..$max_edges] | r {.*, + source_canonical_id: CASE WHEN startNode(r).canonical_id = s.canonical_id + THEN s.canonical_id ELSE endNode(r).canonical_id END, + target_canonical_id: CASE WHEN startNode(r).canonical_id = s.canonical_id + THEN endNode(r).canonical_id ELSE s.canonical_id END}] AS edges1 + OPTIONAL MATCH (n1)-[r2:ASSERTS]-(n2:Entity) + WHERE n1 IN level1 + WITH s, level1, edges1, r2, n2 + ORDER BY r2.relation_category ASC + WITH s, level1, edges1, + collect(DISTINCT n2)[..$max_nodes] AS level2, + [r IN collect(DISTINCT r2)[..$max_edges] | r {.*, + source_canonical_id: startNode(r).canonical_id, + target_canonical_id: endNode(r).canonical_id}] AS edges2 + RETURN s {.*} AS start, + [n IN (level1 + level2) | n {.*}] AS nodes, + (edges1 + edges2) AS edges + """ + + class Neo4jService: """生命周期:app 启动时 create,关闭时 close()。health 路由与 kg 路由共用。 @@ -266,7 +332,7 @@ class Neo4jService: timeout=timeout, ) except asyncio.TimeoutError as exc: - raise Neo4jUnavailable( + raise Neo4jTimeout( f"Neo4j 查询超时(>{timeout}s)", cause=exc ) from exc except neo4j_exc.DriverError as exc: @@ -275,18 +341,27 @@ class Neo4jService: ) from exc return [dict(record) for record in records] - @staticmethod - def _canonical(instance: _Instance) -> _Instance: - """当前仅 canonical schema 的 Cypher 就绪。""" - if instance.schema != "canonical": - # TODO(A3):normalized 实例(Entity/ASSERTS,无 name_norm 索引, - # 等值 6.81s)与空库实例的 Cypher 适配层,需 PM 决策后实现。 + _SUPPORTED_SCHEMAS = ("canonical", "normalized") + + @classmethod + def _supported(cls, instance: _Instance) -> _Instance: + """canonical / normalized 的 Cypher 已就绪;其余 schema(如空库)拒用。""" + if instance.schema not in cls._SUPPORTED_SCHEMAS: raise Neo4jUnavailable( f"Neo4j 实例 {instance.name} 已连通,但其 schema({instance.schema})" "的 Cypher 适配尚未实现,暂不可用于图谱查询" ) return instance + @classmethod + def _name_index_of(cls, instance: _Instance) -> str | None: + """该 schema 期望的 name_norm 索引名(探测/状态用)。""" + if instance.schema == "canonical": + return NAME_INDEX + if instance.schema == "normalized": + return NORMALIZED_NAME_INDEX + return None + # ---------- 健康探测 ---------- async def probe(self) -> dict[str, Any]: @@ -317,9 +392,41 @@ class Neo4jService: except Neo4jUnavailable as exc: # 连通了但计数/索引检查失败:不视为整体失败,附 detail detail = f"计数/索引检查失败:{exc}" + elif instance.schema == "normalized": + # normalized 计数(13M 边 COUNT 聚合)代价高,探测只查索引存在性; + # 完整计数由 kg_status 承担 + try: + indexes = await self._name_indexes(instance) + if NORMALIZED_NAME_INDEX not in indexes: + detail = ( + f"name_norm 索引缺失:{indexes or '(无任何索引)'}" + "(176 部署前需执行 deploy/qa-db/create_neo4j_indexes.sh)" + ) + except Neo4jUnavailable as exc: + detail = f"索引检查失败:{exc}" return {"ok": True, "active": instance.name, "detail": detail} async def _counts(self, instance: _Instance) -> dict[str, int]: + if instance.schema == "normalized": + # normalized 无断言节点:assertion_count 取 ASSERTS 关系计数 + # (13M 边 COUNT 聚合实测约 8.4s,给足超时) + rows = await self._query( + """ + CALL { + MATCH (e:Entity) RETURN count(e) AS entity_count + } + CALL { + MATCH ()-[a:ASSERTS]->() RETURN count(a) AS assertion_count + } + RETURN entity_count, assertion_count + """, + timeout=OVERVIEW_QUERY_TIMEOUT, + ) + row = rows[0] if rows else {} + return { + "entity_count": row.get("entity_count", 0), + "assertion_count": row.get("assertion_count", 0), + } rows = await self._query( """ CALL { @@ -358,16 +465,23 @@ class Neo4jService: async def kg_status(self) -> dict[str, Any]: """labels / relationship_types / entity_count / assertion_count / name_index / fulltext_index(null,实例无全文索引)。""" - instance = self._canonical(await self.select()) + instance = self._supported(await self.select()) counts = await self._counts(instance) indexes = await self._name_indexes(instance) - name_index = NAME_INDEX if NAME_INDEX in indexes else None + expected = self._name_index_of(instance) + name_index = expected if expected in indexes else None + if instance.schema == "normalized": + labels = _NORMALIZED_LABELS + rel_types = _NORMALIZED_REL_TYPES + else: + labels = _CANONICAL_LABELS + rel_types = _CANONICAL_REL_TYPES return { "ok": True, "active": instance.name, "schema": instance.schema, - "labels": _CANONICAL_LABELS, - "relationship_types": _CANONICAL_REL_TYPES, + "labels": labels, + "relationship_types": rel_types, "entity_count": counts["entity_count"], "assertion_count": counts["assertion_count"], "name_index": name_index, @@ -381,10 +495,16 @@ class Neo4jService: 注意 1644 万边:只做 COUNT 聚合,绝不拉取全量边。 """ - self._canonical(await self.select()) + instance = self._supported(await self.select()) + if instance.schema == "normalized": + entity_label = "Entity" + rel_type = "ASSERTS" + else: + entity_label = "SRT2CanonicalEntity" + rel_type = "SRT2_CANONICAL_ASSERTS" rows = await self._query( - """ - MATCH (e:SRT2CanonicalEntity) + f""" + MATCH (e:{entity_label}) WITH e.entity_type AS t, count(*) AS c WHERE t IS NOT NULL RETURN t AS type, c AS count @@ -397,8 +517,8 @@ class Neo4jService: # 关系类别取自边上的 relation_category(属于关系属性,不在边集合里拉边) rows = await self._query( - """ - MATCH ()-[r:SRT2_CANONICAL_ASSERTS]->() + f""" + MATCH ()-[r:{rel_type}]->() WITH r.relation_category AS cat, count(*) AS c WHERE cat IS NOT NULL RETURN cat AS type, c AS count @@ -423,9 +543,35 @@ class Neo4jService: 走 name_norm RANGE 索引(实测 0.09-0.10s)。结果按 name ASC 排序保证稳定。 返回 [{id, name, type, support_count}]。 """ - self._canonical(await self.select()) + instance = self._supported(await self.select()) limit = min(max(int(limit), 1), 20) q_norm = q.lower() + if instance.schema == "normalized": + # normalized 无 name_norm 索引(176 索引脚本补齐前为全标签扫描, + # 175 实测等值 2.9s/前缀 2.8s,QUERY_TIMEOUT 10s 内可接受; + # 有索引后走 RANGE 索引);Entity 无 support_count 属性—— + # hub 度计算代价高(A. thaliana 8.8 万边),置 None 保时延 + rows = await self._query( + """ + MATCH (e:Entity) + WHERE e.name_norm = $q_norm OR e.name_norm STARTS WITH $q_norm + RETURN e.canonical_id AS canonical_id, e.name AS name, + e.entity_type AS entity_type + ORDER BY e.name ASC + LIMIT $limit + """, + {"q_norm": q_norm, "limit": limit}, + timeout=QUERY_TIMEOUT, + ) + return [ + { + "id": r.get("canonical_id"), + "name": r.get("name"), + "type": r.get("entity_type"), + "support_count": None, + } + for r in rows + ] rows = await self._query( """ MATCH (e:SRT2CanonicalEntity) @@ -448,18 +594,19 @@ class Neo4jService: for r in rows ] - async def _resolve_id(self, id_or_name: str) -> str | None: + async def _resolve_id(self, id_or_name: str, instance: _Instance) -> str | None: """canonical_id 或 name_norm 定位(DISCOVERY.md §2.4 Cypher b)。""" + label = "Entity" if instance.schema == "normalized" else "SRT2CanonicalEntity" if id_or_name.startswith("cent_"): rows = await self._query( - "MATCH (e:SRT2CanonicalEntity {canonical_id: $id}) " + f"MATCH (e:{label} {{canonical_id: $id}}) " "RETURN e.canonical_id AS canonical_id", {"id": id_or_name}, timeout=QUERY_TIMEOUT, ) else: rows = await self._query( - "MATCH (e:SRT2CanonicalEntity {name_norm: $name_norm}) " + f"MATCH (e:{label} {{name_norm: $name_norm}}) " "RETURN e.canonical_id AS canonical_id LIMIT 5", {"name_norm": id_or_name.lower()}, timeout=QUERY_TIMEOUT, @@ -470,11 +617,13 @@ class Neo4jService: async def entity(self, id_or_name: str) -> dict[str, Any] | None: """按 canonical_id 或 name_norm 取实体全量属性;未命中返回 None。""" - canonical_id = await self._resolve_id(id_or_name) + instance = self._supported(await self.select()) + canonical_id = await self._resolve_id(id_or_name, instance) if canonical_id is None: return None + label = "Entity" if instance.schema == "normalized" else "SRT2CanonicalEntity" rows = await self._query( - "MATCH (e:SRT2CanonicalEntity {canonical_id: $id}) " + f"MATCH (e:{label} {{canonical_id: $id}}) " "RETURN e {.*} AS node", {"id": canonical_id}, timeout=QUERY_TIMEOUT, @@ -503,15 +652,47 @@ class Neo4jService: - depth=1 绝不执行第二跳(两跳实测约 13s,不默认开启)。 - 返回 API.md §6 的 {center, center_name, depth, nodes, edges}。 """ - self._canonical(await self.select()) # 非 canonical 实例在此抛 Neo4jUnavailable + instance = self._supported(await self.select()) depth = depth if depth in (1, 2) else 1 limit = min(max(int(limit), 1), 200) - canonical_id = await self._resolve_id(id_or_name) + canonical_id = await self._resolve_id(id_or_name, instance) if canonical_id is None: return {"center": None, "center_name": None, "depth": depth, "nodes": [], "edges": []} + if instance.schema == "normalized": + # degree 预检(O(1) COUNT 表达式):hub(>300 度,实测 A. thaliana + # 达 8.8 万边)不展开——展开会拖垮 13M 边库;降级为仅 center, + # 数据安全、可重试 + deg_rows = await self._query( + """ + MATCH (s:Entity {canonical_id: $seed}) + RETURN count { (s)-[:ASSERTS]-() } AS hub_deg + """, + {"seed": canonical_id}, + timeout=QUERY_TIMEOUT, + ) + hub_deg = (deg_rows[0].get("hub_deg") if deg_rows else None) or 0 + if int(hub_deg) > 300: + entity = await self.entity(canonical_id) + props = entity or {} + return { + "center": canonical_id, + "center_name": props.get("name"), + "depth": depth, + "nodes": [ + { + "id": canonical_id, + "label": props.get("name"), + "type": props.get("type"), + } + ], + "edges": [], + } + params = {"seed": canonical_id, "max_nodes": limit, "max_edges": limit} - if depth == 1: + if instance.schema == "normalized": + cypher = _normalized_neighborhood_cypher(depth) + elif depth == 1: cypher = """ MATCH (s:SRT2CanonicalEntity {canonical_id: $seed}) OPTIONAL MATCH (s)-[r1:SRT2_CANONICAL_ASSERTS]-(n1:SRT2CanonicalEntity) @@ -562,6 +743,7 @@ class Neo4jService: row = rows[0] start = row.get("start") or {} + node_list: list[dict[str, Any]] = [ { "id": start.get("canonical_id"), diff --git a/sdk/app/services/sse_relay.py b/sdk/app/services/sse_relay.py index 88ef869..8dce11e 100644 --- a/sdk/app/services/sse_relay.py +++ b/sdk/app/services/sse_relay.py @@ -24,6 +24,7 @@ from __future__ import annotations import asyncio import json import logging +import os import time from collections import deque from datetime import datetime, timezone @@ -44,9 +45,15 @@ CONNECT_TIMEOUT = 10.0 STREAM_IDLE_TIMEOUT = 1200.0 # 2026-08-28:长文(5000字)时代上游图检索静默期可超 240s(高负载下更久), # 原 240s 会把正常流误判超时掐断;1200s 覆盖最坏情况(图检索 ~10 分钟)。 STREAM_KEEPALIVE_INTERVAL = 15.0 -SEARCH_READ_TIMEOUT = 180.0 # 上游 /search 读超时(实测约 92s,见 DISCOVERY.md) +SEARCH_READ_TIMEOUT = 120.0 # 上游 /search 读超时(缺陷 #28:检索模式超时上限 120s) INVALID_CITATION_ERROR_CODE = "invalid_llm_citation" +# 图文并茂图片注入预算(缺陷 #7,2026-09-10):complete 帧端到端延迟上限。 +# 总预算毫秒数;PMCID 只取前 3 个;预算按 PMCID 均分,单篇/整体 wait_for 截断, +# 超时静默降级(图片不影响答案主链)。 +FIGURE_INJECT_BUDGET_MS = float(os.environ.get("FIGURE_INJECT_BUDGET_MS", "2500")) +FIGURE_INJECT_MAX_PMCIDS = 3 + # API.md §4:SSE 响应头 sse_headers = { "Cache-Control": "no-cache", @@ -540,6 +547,11 @@ async def _inject_figures_async(normalized: dict) -> dict: 流程:从 normalized 的 sources 提取 PMCID 清单 → 逐个 resolve_figures(本地优先 PMC 兜底,缓存 1h)→ 组装 figures_map(含 pmid 供兜底路径)→ 调 normalize.inject_figure_markdown 纯同步注入。无图/失败静默降级。 + + 预算控制(缺陷 #7,2026-09-10):整体预取受 FIGURE_INJECT_BUDGET_MS(默认 + 2500ms)硬截断,保证 complete 帧端到端延迟 ≤ 预算;PMCID 只取前 3 个; + 预算按 PMCID 均分,单篇 wait_for 截断。超时/失败仍静默降级——图片不影响 + 答案主链,降级语义与修复前一致。 """ from .figures import resolve_figures from .normalize import inject_figure_markdown @@ -565,15 +577,38 @@ async def _inject_figures_async(normalized: dict) -> dict: for pmcid in marker_pmcids: figures_map.setdefault(pmcid.upper(), {"pmid": None, "figures": [], "license": ""}) - # 预取图片(每个 PMCID 最多 2 张够注入用) - for pmcid, data in list(figures_map.items()): + # PMCID 裁 top 3(按出现顺序:sources 在前,标记补充在后) + pmcids = list(figures_map.keys())[:FIGURE_INJECT_MAX_PMCIDS] + if not pmcids: + base_url = "" # figures.resolve 已返回代理端点绝对路径,无需拼接 + normalized["answer"] = inject_figure_markdown(answer, figures_map, base_url) + return normalized + + budget_s = FIGURE_INJECT_BUDGET_MS / 1000.0 + per_pmcid_s = max(budget_s / len(pmcids), 0.05) # 预算按 PMCID 均分(下限 50ms 防 0) + + async def _fetch_one(pmcid: str) -> None: + data = figures_map[pmcid] try: - resolved = await resolve_figures(pmcid, limit=2) + resolved = await asyncio.wait_for( + resolve_figures(pmcid, limit=2), timeout=per_pmcid_s + ) data["figures"] = resolved.get("figures") or [] data["license"] = resolved.get("license") or "" + except asyncio.TimeoutError: + logger.info("预取图片超时(%s,单篇预算 %.0fms),静默降级", pmcid, per_pmcid_s * 1000) except Exception as exc: logger.info("预取图片失败(%s): %s", pmcid, exc) + try: + await asyncio.wait_for( + asyncio.gather(*(_fetch_one(p) for p in pmcids)), timeout=budget_s + ) + except asyncio.TimeoutError: + logger.info( + "图片注入整体预算(%.0fms)耗尽,未完成预取静默降级", FIGURE_INJECT_BUDGET_MS + ) + base_url = "" # figures.resolve 已返回代理端点绝对路径,无需拼接 normalized["answer"] = inject_figure_markdown(answer, figures_map, base_url) return normalized diff --git a/sdk/app/services/test_docgen_html2docx.py b/sdk/app/services/test_docgen_html2docx.py new file mode 100644 index 0000000..a574bca --- /dev/null +++ b/sdk/app/services/test_docgen_html2docx.py @@ -0,0 +1,127 @@ +"""services/test_docgen_html2docx.py —— HTML 同源 DOCX 转换(缺陷 #47a,2026-09-10)。 + +覆盖:红头居中红字、h1-h3 字号、表格边框+表头灰底、图片缺失降级、空 HTML。 +""" +from __future__ import annotations + +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + +import pytest +from docx import Document +from docx.oxml.ns import qn + +from app.services.docgen.html2docx import docx_from_html_document + +# 模拟 render_html 产物形态:红头 + <main> 内容容器(与 templating.py 一致) +HTML_SAMPLE = """<!DOCTYPE html><html><head><meta charset="utf-8"> +<title>测试公文 +
某某单位文件
+
+

一、总则

+

这是加粗正文,含斜体与inline_code。

+

1.1 适用范围

+
  • 列表项一
  • 列表项二
+

1.1.1 细节

+

三段内容。

+
+
字段说明
名称项目名称
+
样例图
图1 样例
+ +""" + + +def _first_run_color(run): + return run.font.color.rgb + + +def _cell_shade(cell) -> str: + """取 w:shd 的 w:fill(无则空串)。""" + tc_pr = cell._tc.get_or_add_tcPr() + shd = tc_pr.find(qn("w:shd")) + return shd.get(qn("w:fill")) if shd is not None else "" + + +def _run_east_asia(run) -> str: + rpr = run._element.find(qn("w:rPr")) + if rpr is None: + return "" + rfonts = rpr.find(qn("w:rFonts")) + if rfonts is None: + return "" + return rfonts.get(qn("w:eastAsia")) or "" + + +def test_letterhead_red_centered(tmp_path): + out = docx_from_html_document(HTML_SAMPLE, title="测试公文", out_path=tmp_path / "a.docx") + assert out.exists() + doc = Document(str(out)) + paras = doc.paragraphs + assert paras, "应至少有一个红头段落" + head = paras[0] + assert "某某单位文件" in head.text + assert head.alignment is not None and head.alignment.name == "CENTER" + run = head.runs[0] + assert str(_first_run_color(run)) == "FF0000" + # 22pt 黑体 + assert run.font.size is not None and run.font.size.pt == 22.0 + assert _run_east_asia(run) == "黑体" + + +def test_heading_levels_and_sizes(tmp_path): + out = docx_from_html_document(HTML_SAMPLE, title="测试公文", out_path=tmp_path / "b.docx") + doc = Document(str(out)) + styles = [p.style.name for p in doc.paragraphs] + assert "Heading 1" in styles + assert "Heading 2" in styles + assert "Heading 3" in styles + sizes = {} + for p in doc.paragraphs: + if p.style.name.startswith("Heading") and p.runs: + sizes[p.style.name] = p.runs[0].font.size.pt + assert sizes.get("Heading 1") == 22.0 + assert sizes.get("Heading 2") == 16.0 + assert sizes.get("Heading 3") == 14.0 + + +def test_table_grid_and_header_shade(tmp_path): + out = docx_from_html_document(HTML_SAMPLE, title="测试公文", out_path=tmp_path / "c.docx") + doc = Document(str(out)) + assert len(doc.tables) == 1 + table = doc.tables[0] + assert table.style.name == "Table Grid" + assert len(table.rows) == 2 and len(table.columns) == 2 + assert table.cell(0, 0).text == "字段" + assert _cell_shade(table.cell(0, 0)) == "D9D9D9" + assert _cell_shade(table.cell(1, 0)) == "" + + +def test_image_fetch_failure_falls_back_to_alt(tmp_path): + out = docx_from_html_document(HTML_SAMPLE, title="测试公文", out_path=tmp_path / "d.docx") + doc = Document(str(out)) + text = "\\n".join(p.text for p in doc.paragraphs) + assert "图片缺失" in text + assert "样例图" in text # alt 文本兜底 + # 表格完整 + assert doc.tables[0].cell(1, 1).text == "项目名称" + + +def test_empty_html_produces_empty_document(tmp_path): + out = docx_from_html_document("", title="空文档", out_path=tmp_path / "e.docx") + assert out.exists() + doc = Document(str(out)) + assert doc.core_properties.title == "空文档" + assert len(doc.paragraphs) == 0 + assert len(doc.tables) == 0 + + +def test_bold_italic_code_runs(tmp_path): + out = docx_from_html_document(HTML_SAMPLE, title="测试公文", out_path=tmp_path / "f.docx") + doc = Document(str(out)) + bold_found = any(r.font.bold for p in doc.paragraphs for r in p.runs) + assert bold_found + code_found = any(r.font.name == "Courier New" + for p in doc.paragraphs for r in p.runs) + assert code_found diff --git a/sdk/app/services/test_docgen_pipeline.py b/sdk/app/services/test_docgen_pipeline.py index 4b8c359..2d0d9b6 100644 --- a/sdk/app/services/test_docgen_pipeline.py +++ b/sdk/app/services/test_docgen_pipeline.py @@ -8,7 +8,7 @@ sys.path.insert(0, str(Path(__file__).resolve().parents[2])) import pytest # noqa: E402 -from app.services.docgen import client, pipeline # noqa: E402 +from app.services.docgen import client, pipeline, websearch # noqa: E402 from app.services.docgen import llm as llm_mod # noqa: E402 from app.services.docgen.settings import DocgenPaths, set_paths # noqa: E402 @@ -16,6 +16,13 @@ from app.services.docgen.settings import DocgenPaths, set_paths # noqa: E402 @pytest.fixture(autouse=True) def _tmp_docgen_root(tmp_path, monkeypatch): set_paths(DocgenPaths(root=str(tmp_path / "docgen"))) + + # 基线抖动修复(2026-09-10):pipeline web_search 默认 True 会真实联网搜 + # 范文大纲——网络偶发成功时大纲漂移(HEAD 上同失败)。测试一律关联网。 + async def _no_web(description): + return None + + monkeypatch.setattr(websearch, "search_template_outline", _no_web) yield diff --git a/sdk/app/services/test_neo4j.py b/sdk/app/services/test_neo4j.py index c0a5feb..da849f4 100644 --- a/sdk/app/services/test_neo4j.py +++ b/sdk/app/services/test_neo4j.py @@ -16,7 +16,9 @@ from app.services.neo4j import ( FULLTEXT_INDEX, NAME_INDEX, NORMALIZED_ACTIVE_NAME, + NORMALIZED_NAME_INDEX, Neo4jService, + Neo4jTimeout, Neo4jUnavailable, _Instance, ) @@ -307,11 +309,11 @@ async def test_kg_status_missing_index_reports_null(): assert status["name_index"] is None assert status["fulltext_index"] is None -async def test_non_canonical_instance_raises_unavailable(): - normal = inst(NORMALIZED_ACTIVE_NAME, "bolt://fake-normalized:19788", - schema="normalized", label=None, rel=None) +async def test_unsupported_schema_raises_unavailable(): + empty = inst("plantai-protected-neo4j-1", "bolt://fake-protected:17687", + schema="empty", label=None, rel=None) driver = FakeDriver({"RETURN 1": [{"x": 1}]}) - service = make_service(driver, [normal]) + service = make_service(driver, [empty]) with pytest.raises(Neo4jUnavailable): await service.kg_status() with pytest.raises(Neo4jUnavailable): @@ -557,7 +559,7 @@ async def test_neighborhood_clamps_depth_and_limit(): # ---------- 超时与驱动错误 ---------- -async def test_query_timeout_raises_unavailable(monkeypatch): +async def test_query_timeout_raises_neo4j_timeout(monkeypatch): import app.services.neo4j as neo4j_mod # 调短查询超时以在单测内触发 asyncio.wait_for @@ -567,7 +569,7 @@ async def test_query_timeout_raises_unavailable(monkeypatch): delays={"name_norm: $name_norm": 0.5}, ) service = make_service(slow) - with pytest.raises(Neo4jUnavailable): + with pytest.raises(Neo4jTimeout): await service.entity("OsSPL14") async def test_driver_error_raises_unavailable(): @@ -586,3 +588,104 @@ async def test_close_closes_all_drivers(): service = make_service(driver) await service.close() assert driver.closed is True + + +# ---------- normalized schema 适配(缺陷 #32,#31,2026-09-10) ---------- + +def normalized_instance() -> _Instance: + return inst(NORMALIZED_ACTIVE_NAME, "bolt://fake-normalized:19788", + schema="normalized", label="Entity", rel="ASSERTS") + + +def make_normalized_service(driver: FakeDriver) -> Neo4jService: + instances = [normalized_instance()] + service = Neo4jService(instances=instances) + service._drivers = {instances[0].name: driver} + service._selected = True + service._active = instances[0] + return service + + +async def test_normalized_search_entities(): + driver = FakeDriver({ + "e.name_norm = $q_norm OR e.name_norm STARTS WITH $q_norm": [ + {"canonical_id": "cent_7b70", "name": "arginine 127", "entity_type": "sequence_feature"}, + ], + }) + service = make_normalized_service(driver) + results = await service.search_entities("arginine", limit=5) + assert results[0]["id"] == "cent_7b70" + assert results[0]["type"] == "sequence_feature" + assert results[0]["support_count"] is None # normalized 无此属性 + + +async def test_normalized_entity(): + props = {"canonical_id": "cent_7b70", "name": "arginine 127", + "name_norm": "arginine 127", "entity_type": "sequence_feature", "layer": "main"} + driver = FakeDriver({ + "canonical_id: $id": [{"canonical_id": "cent_7b70"}], + "RETURN e {.*} AS node": [{"node": props}], + }) + service = make_normalized_service(driver) + entity = await service.entity("cent_7b70") + assert entity["id"] == "cent_7b70" + assert entity["properties"]["layer"] == "main" + + +async def test_normalized_entity_by_name(): + driver = FakeDriver({ + "name_norm: $name_norm": [{"canonical_id": "cent_7b70"}], + "RETURN e {.*} AS node": [{"node": {"canonical_id": "cent_7b70", "name": "arginine 127", + "entity_type": "sequence_feature"}}], + }) + service = make_normalized_service(driver) + entity = await service.entity("arginine 127") + assert entity is not None and entity["id"] == "cent_7b70" + + +async def test_normalized_neighborhood_depth1(): + driver = FakeDriver({ + "canonical_id: $id": [{"canonical_id": "cent_7b70"}], + "count { (s)-[:ASSERTS]-() }": [{"hub_deg": 1}], + "OPTIONAL MATCH (s)-[r1:ASSERTS]": [ + { + "start": node_row("cent_7b70", "arginine 127", "sequence_feature"), + "nodes": [node_row("cent_x", "X protein", "protein")], + "edges": [edge_row("cent_7b70", "cent_x", "binds")], + } + ], + }) + service = make_normalized_service(driver) + result = await service.neighborhood("cent_7b70", depth=1, limit=50) + assert result["center"] == "cent_7b70" + assert len(result["nodes"]) == 2 + assert result["edges"][0]["relation"] == "binds" + + +async def test_normalized_neighborhood_hub_degrades_to_center_only(): + driver = FakeDriver({ + "count { (s)-[:ASSERTS]-() }": [{"hub_deg": 88543}], # A. thaliana 实测量级 + "canonical_id: $id": [{"canonical_id": "cent_hub"}], + "RETURN e {.*} AS node": [{"node": node_row("cent_hub", "A. thaliana", "species")}], + }) + service = make_normalized_service(driver) + result = await service.neighborhood("cent_hub", depth=1, limit=50) + assert result["center"] == "cent_hub" + assert len(result["nodes"]) == 1 and result["edges"] == [] + # 不得执行展开查询(hub 保护) + for cypher, _ in driver.calls: + assert "OPTIONAL MATCH (s)-[r1:ASSERTS]" not in cypher + + +async def test_normalized_kg_status(): + driver = FakeDriver({ + COUNTS_SCRIPT: [{"entity_count": 6342154, "assertion_count": 13040841}], + INDEXES_SCRIPT: [{"name": NORMALIZED_NAME_INDEX}], + }) + service = make_normalized_service(driver) + status = await service.kg_status() + assert status["schema"] == "normalized" + assert status["labels"] == ["Entity", "Paper"] + assert status["relationship_types"] == ["ASSERTS", "SUPPORTED_BY"] + assert status["name_index"] == NORMALIZED_NAME_INDEX + assert status["entity_count"] == 6342154 diff --git a/sdk/app/services/test_sse_relay.py b/sdk/app/services/test_sse_relay.py index 9183886..50c01e2 100644 --- a/sdk/app/services/test_sse_relay.py +++ b/sdk/app/services/test_sse_relay.py @@ -599,3 +599,64 @@ def test_error_payload_shapes(): assert sse_relay.error_payload("TIMEOUT") == {"code": "TIMEOUT", "message": "检索超时", "retryable": True} assert sse_relay.error_payload("UPSTREAM_DOWN")["retryable"] is False assert sse_relay.error_payload("INVALID_CITATION")["retryable"] is True + + +# ---------- 缺陷 #7 图片注入预算(2026-09-10) ---------- + +async def test_inject_figures_budget_cuts_off_slow_fetch(monkeypatch): + """总预算截断:慢 resolve_figures 超预算后静默降级,注入函数按时返回。""" + import time + + from app.services.sse_relay import _inject_figures_async + + monkeypatch.setenv("FIGURE_INJECT_BUDGET_MS", "300") + + async def slow(pmcid, limit=2): + await asyncio.sleep(5) + return {"figures": [{"src": "x"}], "license": ""} + + import app.services.figures as figs + + monkeypatch.setattr(figs, "resolve_figures", slow) + normalized = { + "answer": "答案 [[FIG:PMC123]]", + "sources": [ + {"pmcid": "PMC123", "pmid": "1"}, + {"pmcid": "PMC456", "pmid": "2"}, + {"pmcid": "PMC789", "pmid": "3"}, + {"pmcid": "PMC999", "pmid": "4"}, # 第 4 个 PMCID 应被 top3 裁掉 + ], + } + t0 = time.monotonic() + result = await _inject_figures_async(normalized) + elapsed = time.monotonic() - t0 + assert elapsed < 2.0, f"注入总时长超出预算:{elapsed:.2f}s" + # 无图可用时 inject_figure_markdown 移除失效 [[FIG:]] 标记(既有语义) + assert "[[FIG:" not in result["answer"] + + +async def test_inject_figures_top3_pmcids(monkeypatch): + """PMCID 裁 top 3:第 4 个不请求。""" + from app.services.sse_relay import _inject_figures_async + + monkeypatch.setenv("FIGURE_INJECT_BUDGET_MS", "5000") + fetched = [] + + async def record(pmcid, limit=2): + fetched.append(pmcid) + return {"figures": [], "license": ""} + + import app.services.figures as figs + + monkeypatch.setattr(figs, "resolve_figures", record) + normalized = { + "answer": "答案", + "sources": [ + {"pmcid": "PMC1", "pmid": "1"}, + {"pmcid": "PMC2", "pmid": "2"}, + {"pmcid": "PMC3", "pmid": "3"}, + {"pmcid": "PMC4", "pmid": "4"}, + ], + } + await _inject_figures_async(normalized) + assert fetched == ["PMC1", "PMC2", "PMC3"] diff --git a/sdk/pyproject.toml b/sdk/pyproject.toml index 7ff93bb..76aea68 100644 --- a/sdk/pyproject.toml +++ b/sdk/pyproject.toml @@ -16,6 +16,7 @@ dependencies = [ "pydantic>=2.6", "pydantic-settings>=2.2", "python-multipart>=0.0.9", + "beautifulsoup4>=4.12", # 缺陷 #47a:HTML 同源 DOCX 转换 ] [project.optional-dependencies]