From 3889a2a29bdaaab20c06c1e1e50a94b381263b67 Mon Sep 17 00:00:00 2001 From: admin Date: Fri, 11 Sep 2026 14:12:46 +0800 Subject: [PATCH 1/3] =?UTF-8?q?fix(r4-38):=20mineru=20=E5=90=AF=E5=8A=A8?= =?UTF-8?q?=E6=9C=9F=E5=A4=B1=E8=B4=A5=E8=87=AA=E5=8A=A8=E9=87=8D=E8=AF=95?= =?UTF-8?q?+=E4=B8=AD=E6=96=87=E5=8F=AF=E8=AF=BB=E9=94=99=E8=AF=AF+PYTHONU?= =?UTF-8?q?NBUFFERED?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 缺陷 #38(验收 §4):任务在最早阶段(pipeline 日志出现前)抛出无消息异常, mineru API 侧 task.error 空串 → CLI 拼出空 message;API 异常日志打向 stdout 被块缓冲,CLI 失败后立即杀 API 进程,真因永久丢失。 修复(agent/methodology 与 agent/legacy-rag 两处同改): - build_mineru_runtime 的 env 注入 PYTHONUNBUFFERED=1(子进程异常日志即写即落) - returncode≠0 且 job.log 无 doc_analyze_streaming 行 = 启动期失败 → 自动重试 1 次(stdout 追加同一 job.log,日志留 [retry] 标记,重试块含重复启动命令行) - 重试仍失败且仍无 pipeline 行 → status failed + 中文可读错误 「提取启动阶段失败(重试后),可能受主机 GPU/显示重配置干扰,请重试」, 附 job.log 原始错误尾部 - 已有 pipeline 行的中途失败不得触发重试,走原错误路径(含 CUDA OOM 中文错误) 测试:agent/{methodology,legacy-rag}/api/test_mineru_retry.py(两树 8 用例): env 含 PYTHONUNBUFFERED;首败无 pipeline 行 → 重试;重试仍败 → 中文可读错误; 有 pipeline 行失败 → 不重试;json 直投跳过 mineru 段不重试;G6-2 边界。 Co-Authored-By: Claude Fable 5 --- agent/legacy-rag/api/methodology_routes.py | 70 ++++- agent/legacy-rag/api/test_mineru_retry.py | 328 ++++++++++++++++++++ agent/methodology/api/methodology_routes.py | 68 ++++ agent/methodology/api/test_mineru_retry.py | 328 ++++++++++++++++++++ 4 files changed, 793 insertions(+), 1 deletion(-) mode change 100755 => 100644 agent/legacy-rag/api/methodology_routes.py create mode 100644 agent/legacy-rag/api/test_mineru_retry.py mode change 100755 => 100644 agent/methodology/api/methodology_routes.py create mode 100644 agent/methodology/api/test_mineru_retry.py diff --git a/agent/legacy-rag/api/methodology_routes.py b/agent/legacy-rag/api/methodology_routes.py old mode 100755 new mode 100644 index 07b43d4..06137a4 --- a/agent/legacy-rag/api/methodology_routes.py +++ b/agent/legacy-rag/api/methodology_routes.py @@ -29,7 +29,7 @@ PAPERS_CSV = PROJECT_ROOT / "data" / "evaluation" / "methodology_papers.csv" GOLD_CSV = PROJECT_ROOT / "data" / "evaluation" / "methodology_gold.csv" REVIEW_RESULTS_JSON = BATCH_DIR / "review_results.json" REVIEW_RESULTS_CSV = BATCH_DIR / "review_results.csv" -# 2026-08-31 新工程适配:上传目录/产出根支持 env 覆盖 +# 2026-08-31 新工程适配:上传目录/产出根支持 env 覆盖(数据根迁移时无需改代码) UPLOAD_JOBS_DIR = Path(os.environ.get("METHODOLOGY_UPLOAD_JOBS_DIR", PROJECT_ROOT / "data" / "evaluation" / "methodology_upload_jobs")) UPLOAD_OUTPUT_ROOT = Path(os.environ.get("METHODOLOGY_UPLOAD_OUTPUT_ROOT", PROJECT_ROOT / "data" / "evaluation" / "0505")) DEFAULT_MINERU_CMD = str(conda_env_path("minerupy312") / "bin" / "mineru") @@ -351,6 +351,22 @@ def mineru_gpu_preflight_error( ) +# 缺陷 #38（2026-09-11）：成功任务的 job.log 必然有 doc_analyze_streaming 行（176 实测）。 +# returncode≠0 且日志中无此行 = 启动期失败（pipeline 起来前就死了，真因被块缓冲吞掉）。 +# 判定条件（methodology 与 legacy-rag 两处 build_mineru_runtime/run_uploaded_methodology_job 同改） +MINERU_PIPELINE_LOG_MARKER = "doc_analyze_streaming" +MINERU_STARTUP_RETRIES = 1 + + +def job_log_has_pipeline_line(log_path: Path) -> bool: + """job.log 是否已出现 pipeline 日志行（def _job_log_has_pipeline_line）。""" + try: + text = log_path.read_text(encoding="utf-8", errors="replace") + except OSError: + return False + return MINERU_PIPELINE_LOG_MARKER in text + + def summarize_mineru_failure(log_path: Path, returncode: int) -> str: message = f"MinerU failed with exit code {returncode}; see {log_path}" if not log_path.exists(): @@ -368,6 +384,28 @@ def summarize_mineru_failure(log_path: Path, returncode: int) -> str: return message +def summarize_mineru_startup_failure(log_path: Path, returncode: int, retried: bool) -> str: + """启动期失败（无 pipeline 行）的中文可读错误（缺陷 #38）。 + + 附上 job.log 中可见的原始错误尾部，让用户能看到真因（修复前真因 + 被 stdout 块缓冲 + CLI 立即杀 API 进程永久丢失）。 + """ + tail = "" + try: + text = log_path.read_text(encoding="utf-8", errors="replace") + lines = [ln for ln in text.splitlines() if ln.strip()] + if lines: + tail = " | ".join(ln.strip() for ln in lines[-3:])[:500] + except OSError: + tail = "" + detail = f"原始日志尾部：{tail}" if tail else f"原始错误见日志：{log_path}" + action = "重试后" if retried else "" + return ( + f"提取启动阶段失败{action}，可能受主机 GPU/显示重配置干扰，请重试" + f"（MinerU 退出码 {returncode}；{detail}）。" + ) + + def build_mineru_runtime() -> tuple[list[str], dict[str, str], dict[str, str]]: env = os.environ.copy() env.setdefault("MINERU_MODEL_SOURCE", "modelscope") @@ -384,6 +422,9 @@ def build_mineru_runtime() -> tuple[list[str], dict[str, str], dict[str, str]]: else: env.setdefault("CUDA_VISIBLE_DEVICES", "0") env.setdefault("TORCH_CUDNN_V8_API_DISABLED", "1") + # 缺陷 #38（2026-09-11）：子进程 API 的异常日志即写即落（不再被 stdout + # 块缓冲吞掉）——CLI 失败后立即杀 API 进程前真因已落到 job.log。 + env["PYTHONUNBUFFERED"] = "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") @@ -566,7 +607,34 @@ def run_uploaded_methodology_job( stdout=handle, stderr=subprocess.STDOUT, ) + # 缺陷 #38：启动期失败自动重试 1 次——returncode≠0 且 job.log 无 + # doc_analyze_streaming 行 = pipeline 起来前就失败（真因曾因块缓冲 + # 丢失）。重试的 stdout 继续 append 到同一 job.log，日志/状态留 + # retry 标记（G6-2：已有 pipeline 行的中途失败不得触发重试）。 + startup_retried = False + if result.returncode != 0 and not job_log_has_pipeline_line(log_path): + startup_retried = True + with log_path.open("a", encoding="utf-8") as handle: + handle.write( + f"[retry] MinerU 启动期失败（exit {result.returncode}，" + "无 pipeline 阶段日志），自动重试 1/1\n" + ) + handle.write("$ " + " ".join(cmd) + "\n") + handle.flush() + result = subprocess.run( + cmd, + cwd=PROJECT_ROOT, + env=mineru_env, + text=True, + stdin=subprocess.DEVNULL, + stdout=handle, + stderr=subprocess.STDOUT, + ) if result.returncode != 0: + if not job_log_has_pipeline_line(log_path): + raise RuntimeError( + summarize_mineru_startup_failure(log_path, result.returncode, retried=startup_retried) + ) raise RuntimeError(summarize_mineru_failure(log_path, result.returncode)) matches = sorted(mineru_out.rglob("*_content_list.json")) if not matches: diff --git a/agent/legacy-rag/api/test_mineru_retry.py b/agent/legacy-rag/api/test_mineru_retry.py new file mode 100644 index 0000000..cad974c --- /dev/null +++ b/agent/legacy-rag/api/test_mineru_retry.py @@ -0,0 +1,328 @@ +"""test_mineru_retry.py —— MinerU 启动期失败重试与可读错误（缺陷 #38，验收 §4 + 回归 G6）。 + +运行（worktree 根目录，无需部署、不触服务）： + /home/xzq/5090/teamWorkspace/zh/plantai-rag-new-venv/bin/python -m pytest agent/methodology/api/test_mineru_retry.py -v + /home/xzq/5090/teamWorkspace/zh/plantai-rag-new-venv/bin/python -m pytest agent/legacy-rag/api/test_mineru_retry.py -v +说明：agent/methodology 与 agent/legacy-rag 共用同一份 routes（文件名同、内容同）， +同一测试文件复制两处、两处同跑（各用各树的 src/）。 + +断言： +1. build_mineru_runtime 返回 env 含 PYTHONUNBUFFERED=1； +2. subprocess.run 首次失败 + job.log 无 pipeline 行 → 自动重试 1 次 + （两次启动行、job.log 有 [retry] 标记，第二次 stdout 追加同一 job.log）； +3. 重试仍失败 → status failed + 中文可读「提取启动阶段失败」+ 原始错误尾部； +4. 已有 pipeline 行的中途失败 → 不重试，走原错误路径（含 CUDA OOM 中文错误）。 +""" +from __future__ import annotations + +import json +import subprocess +from dataclasses import dataclass, field +from pathlib import Path + +import pytest + +import api.methodology_routes as mr + +from api.methodology_routes import ( + MINERU_PIPELINE_LOG_MARKER, + build_mineru_runtime, + job_log_has_pipeline_line, + run_uploaded_methodology_job, + summarize_mineru_startup_failure, +) + + +# --------------------------------------------------------------------------- +# 固定夹具 +# --------------------------------------------------------------------------- + +@dataclass +class FakeResult: + returncode: int + + +@dataclass +class RunnerRecorder: + """按调用顺序返回预设 returncode；每次调用把 stdout 写进所给 handle（模拟子进程日志）。""" + + sequence: list + calls: list = field(default_factory=list) + + def run(self, cmd, **kwargs): + n = len(self.calls) + self.calls.append({"cmd": cmd, "kwargs": kwargs}) + rc = int(self.sequence[min(n, len(self.sequence) - 1)]) + handle = kwargs.get("stdout") + if handle is not None and kwargs.get("text"): + handle.write(f"[fake stdout call#{n}] rc={rc}\n") + handle.flush() + return FakeResult(returncode=rc) + + +@pytest.fixture +def fake_success(monkeypatch): + """LLM 批次与 bbav5 配置阶段全部假成功（不触真实脚本/目录/真实函数）。 + + - gf_python_path / _preflight_executable 桩掉（不依赖真实 python 路径） + - build_bbav5_methodology_config 桩掉（不要求真实批次产物目录） + - figure_table 阶段走 routes 内部 try/except，不阻断 + """ + captured = {} + + def fake_batch(cmd, **kwargs): + captured["batch_cmd"] = cmd + return FakeResult(returncode=0) + + monkeypatch.setattr(mr, "gf_python_path", lambda: Path("/usr/bin/python3")) + monkeypatch.setattr(mr, "_preflight_executable", lambda cmd, hint, name: None) + monkeypatch.setattr(mr.subprocess, "run", fake_batch) + monkeypatch.setattr( + mr, + "build_bbav5_methodology_config", + lambda output_dir: captured.setdefault("config", {}), + ) + return captured + + +@pytest.fixture +def job_root(monkeypatch, tmp_path): + """把 UPLOAD_JOBS_DIR 重定向到 tmp（绝不写真实数据目录）。""" + root = tmp_path / "upload_jobs" + monkeypatch.setattr(mr, "UPLOAD_JOBS_DIR", root) + return root + + +def read_status(job_dir: Path) -> dict: + return json.loads((job_dir / "status.json").read_text(encoding="utf-8")) + + +def run_job(monkeypatch, tmp_path, runner: RunnerRecorder, source_path: Path, fake_success, job_root): + """以 runner 重绑 subprocess.run 跑完整任务流（fake_success 保证后续阶段假成功）。 + + 注意：fake_success 已把 mr.subprocess.run 桩为假批次；这里再次 setattr + 为 runner（同一模块对象），此后所有子进程调用（mineru 两次 + LLM 批次） + 都由 runner 按 sequence 依次应答——测试 sequence 需覆盖全部调用。 + """ + job_dir = job_root / "testjob" + # 模拟 mineru 成功产出：在 job/mineru 下伪造 *_content_list.json + # （测试不真正执行 mineru） + (job_dir / "mineru").mkdir(parents=True, exist_ok=True) + (job_dir / "mineru" / "fake_content_list.json").write_text("{}", encoding="utf-8") + # GPU 预检禁用（否则 nvidia-smi 会消耗 runner 的 sequence 槽位） + monkeypatch.setenv("METHODOLOGY_MINERU_GPU_PREFLIGHT", "0") + monkeypatch.setattr(subprocess, "run", runner.run) + run_uploaded_methodology_job( + job_id="testjob", + source_path=source_path, + paper_id="paper1", + family="", + output_dir=tmp_path / "out", + llm_model="deepseek-v4-flash", + llm_request_timeout=75, + sandbox=False, + sandbox_run_id="", + ) + return job_dir + + +@pytest.fixture +def pdf_source(tmp_path): + src = tmp_path / "sample.pdf" + src.write_bytes(b"%PDF-1.4\nfake pdf bytes for startup-failure test\n") + return src + + +@pytest.fixture +def json_source(tmp_path): + src = tmp_path / "sample.json" + src.write_text("{}", encoding="utf-8") + return src + + +# --------------------------------------------------------------------------- +# 修复点 1：env 注入 PYTHONUNBUFFERED +# --------------------------------------------------------------------------- + +def test_build_mineru_runtime_env_unbuffered(monkeypatch): + monkeypatch.delenv("MINERU_CMD", raising=False) + cmd, env, runtime = build_mineru_runtime() + assert env.get("PYTHONUNBUFFERED") == "1" + assert cmd # 默认 MINERU_CMD 路径（不要求存在，build 阶段不校验） + + +# --------------------------------------------------------------------------- +# 修复点 2：启动期失败自动重试 1 次 +# --------------------------------------------------------------------------- + +def test_startup_failure_retries_once(monkeypatch, tmp_path, pdf_source, fake_success, job_root): + runner = RunnerRecorder(sequence=[1, 0, 0]) # mineru 首调失败 → 重试成功 → LLM 批次成功 + job_dir = run_job(monkeypatch, tmp_path, runner, pdf_source, fake_success, job_root) + + log_text = (job_dir / "job.log").read_text(encoding="utf-8") + startup_lines = [ln for ln in log_text.splitlines() if ln.startswith("$ ")] + assert len(startup_lines) >= 2, f"应出现两次 mineru 启动（1 次自动重试），实际：{startup_lines}" + assert "[retry]" in log_text, "job.log 应有可识别重试标记" + # 第二次 subprocess.run 的 stdout 追加到同一 job.log + assert "[fake stdout call#0]" in log_text + assert "[fake stdout call#1]" in log_text + + status = read_status(job_dir) + assert status.get("status") == "completed" + + +def test_startup_failure_retry_then_fail_readable_error(monkeypatch, tmp_path, pdf_source, fake_success, job_root): + runner = RunnerRecorder(sequence=[1, 1]) # 两次都失败，且无 pipeline 行 + job_dir = run_job(monkeypatch, tmp_path, runner, pdf_source, fake_success, job_root) + + log_text = (job_dir / "job.log").read_text(encoding="utf-8") + assert len([ln for ln in log_text.splitlines() if ln.startswith("$ /")]) == 2 # 两次 mineru 启动 + assert "[retry]" in log_text + # pipeline 判定：仅检查子进程输出行（启动命令行本身含 marker 关键字） + stdout_lines = [ln for ln in log_text.splitlines() if "[fake stdout" in ln or ln.startswith("[retry]")] + assert MINERU_PIPELINE_LOG_MARKER not in "\n".join(stdout_lines) + + status = read_status(job_dir) + assert status.get("status") == "failed" + message = str(status.get("message") or "") + assert "提取启动阶段失败" in message, f"中文可读错误缺失：{message}" + assert "退出码" in message and "1" in message + assert "原始日志尾部" in message or "原始错误见日志" in message + assert message # 非空 + + +def test_startup_failure_with_marker_retries_once_then_no_second_retry(monkeypatch, tmp_path, pdf_source, job_root): + """G6-2 边界：第一次失败无 marker → 触发重试；重试后才有 marker 但 rc≠0 → + 不二次重试，走 summarize_mineru_failure（有 marker 的中途失败路径）。""" + sequence = [1, 1] + calls = [] + + def run_with_marker(cmd, **kwargs): + n = len(calls) + rc = int(sequence[min(n, len(sequence) - 1)]) + calls.append({"cmd": cmd, "kwargs": kwargs}) + handle = kwargs.get("stdout") + if handle is not None and kwargs.get("text"): + if n == 1: # 第二次调用（重试）时写入 pipeline 行 + handle.write(f"[fake] {MINERU_PIPELINE_LOG_MARKER} started\n") + handle.write(f"[fake stdout call#{n}] rc={rc}\n") + handle.flush() + return FakeResult(returncode=rc) + + monkeypatch.setenv("METHODOLOGY_MINERU_GPU_PREFLIGHT", "0") + monkeypatch.setattr(subprocess, "run", run_with_marker) + monkeypatch.setattr(mr, "_preflight_executable", lambda cmd, hint, name: None) + monkeypatch.setattr( + mr, + "summarize_mineru_failure", + lambda log_path, rc: f"CUDA out of memory fake path rc={rc}", + ) + job_dir = job_root / "testjob" + run_uploaded_methodology_job( + job_id="testjob", + source_path=pdf_source, + paper_id="paper1", + family="", + output_dir=tmp_path / "out", + llm_model="deepseek-v4-flash", + llm_request_timeout=75, + sandbox=False, + sandbox_run_id="", + ) + log_text = (job_dir / "job.log").read_text(encoding="utf-8") + assert MINERU_PIPELINE_LOG_MARKER in log_text + assert log_text.count("[retry]") == 1, "有 pipeline 行后不得再次重试" + # 中途失败（有 marker）→ 原错误路径（CUDA 中文错误） + status = read_status(job_dir) + assert status.get("status") == "failed" + assert "CUDA" in str(status.get("message") or "") + + +# --------------------------------------------------------------------------- +# 修复点 4：有 pipeline 行的中途失败不重试 +# --------------------------------------------------------------------------- + +def test_mid_failure_with_pipeline_line_no_retry(monkeypatch, tmp_path, pdf_source, job_root): + sequence = [1] + calls = [] + + def run_with_marker(cmd, **kwargs): + n = len(calls) + rc = int(sequence[min(n, len(sequence) - 1)]) + calls.append({"cmd": cmd, "kwargs": kwargs}) + handle = kwargs.get("stdout") + if handle is not None and kwargs.get("text"): + if n == 0: # 第一次调用即写 pipeline 行 + handle.write(f"[fake] {MINERU_PIPELINE_LOG_MARKER} started\n") + handle.write(f"[fake stdout call#{n}] rc={rc}\n") + handle.flush() + return FakeResult(returncode=rc) + + monkeypatch.setenv("METHODOLOGY_MINERU_GPU_PREFLIGHT", "0") + monkeypatch.setattr(subprocess, "run", run_with_marker) + monkeypatch.setattr(mr, "_preflight_executable", lambda cmd, hint, name: None) + monkeypatch.setattr( + mr, + "summarize_mineru_failure", + lambda log_path, rc: f"CUDA out of memory fake path rc={rc}", + ) + job_dir = job_root / "testjob" + run_uploaded_methodology_job( + job_id="testjob", + source_path=pdf_source, + paper_id="paper1", + family="", + output_dir=tmp_path / "out", + llm_model="deepseek-v4-flash", + llm_request_timeout=75, + sandbox=False, + sandbox_run_id="", + ) + log_text = (job_dir / "job.log").read_text(encoding="utf-8") + assert MINERU_PIPELINE_LOG_MARKER in log_text + assert "[retry]" not in log_text, "已有 pipeline 行的中途失败不得触发重试" + # 原始启动块 = 1 条 `$ env ...` + 1 条 `$ `；无重试追加 + assert len([ln for ln in log_text.splitlines() if ln.startswith("$ /")]) == 1 + status = read_status(job_dir) + assert status.get("status") == "failed" + assert "CUDA" in str(status.get("message") or "") + + +# --------------------------------------------------------------------------- +# JSON 直接投递（跳过 mineru 段）不重试、不破坏状态机 +# --------------------------------------------------------------------------- + +def test_json_source_skips_mineru_no_retry(monkeypatch, tmp_path, json_source, fake_success, job_root): + runner = RunnerRecorder(sequence=[0]) # 仅 LLM 批次一次 + job_dir = run_job(monkeypatch, tmp_path, runner, json_source, fake_success, job_root) + log_text = (job_dir / "job.log").read_text(encoding="utf-8") + assert "[retry]" not in log_text + assert MINERU_PIPELINE_LOG_MARKER not in log_text + status = read_status(job_dir) + assert status.get("status") == "completed" + + +# --------------------------------------------------------------------------- +# 工具函数单元 +# --------------------------------------------------------------------------- + +def test_job_log_has_pipeline_line(tmp_path): + log_path = tmp_path / "job.log" + assert job_log_has_pipeline_line(log_path) is False + log_path.write_text("noise\n", encoding="utf-8") + assert job_log_has_pipeline_line(log_path) is False + log_path.write_text(f"2026-09-11 [INFO] {MINERU_PIPELINE_LOG_MARKER} begin\n", encoding="utf-8") + assert job_log_has_pipeline_line(log_path) is True + + +def test_summarize_startup_failure_readable(tmp_path): + log_path = tmp_path / "job.log" + log_path.write_text("line one\nline two\nTraceback: boom\n", encoding="utf-8") + msg = summarize_mineru_startup_failure(log_path, 1, retried=True) + assert "提取启动阶段失败" in msg + assert "重试后" in msg + assert "退出码 1" in msg + assert "Traceback: boom" in msg + missing = summarize_mineru_startup_failure(tmp_path / "nope.log", 1, retried=False) + assert "提取启动阶段失败" in missing + assert "重试后" not in missing diff --git a/agent/methodology/api/methodology_routes.py b/agent/methodology/api/methodology_routes.py old mode 100755 new mode 100644 index e9f1f7a..06137a4 --- a/agent/methodology/api/methodology_routes.py +++ b/agent/methodology/api/methodology_routes.py @@ -351,6 +351,22 @@ def mineru_gpu_preflight_error( ) +# 缺陷 #38（2026-09-11）：成功任务的 job.log 必然有 doc_analyze_streaming 行（176 实测）。 +# returncode≠0 且日志中无此行 = 启动期失败（pipeline 起来前就死了，真因被块缓冲吞掉）。 +# 判定条件（methodology 与 legacy-rag 两处 build_mineru_runtime/run_uploaded_methodology_job 同改） +MINERU_PIPELINE_LOG_MARKER = "doc_analyze_streaming" +MINERU_STARTUP_RETRIES = 1 + + +def job_log_has_pipeline_line(log_path: Path) -> bool: + """job.log 是否已出现 pipeline 日志行（def _job_log_has_pipeline_line）。""" + try: + text = log_path.read_text(encoding="utf-8", errors="replace") + except OSError: + return False + return MINERU_PIPELINE_LOG_MARKER in text + + def summarize_mineru_failure(log_path: Path, returncode: int) -> str: message = f"MinerU failed with exit code {returncode}; see {log_path}" if not log_path.exists(): @@ -368,6 +384,28 @@ def summarize_mineru_failure(log_path: Path, returncode: int) -> str: return message +def summarize_mineru_startup_failure(log_path: Path, returncode: int, retried: bool) -> str: + """启动期失败（无 pipeline 行）的中文可读错误（缺陷 #38）。 + + 附上 job.log 中可见的原始错误尾部，让用户能看到真因（修复前真因 + 被 stdout 块缓冲 + CLI 立即杀 API 进程永久丢失）。 + """ + tail = "" + try: + text = log_path.read_text(encoding="utf-8", errors="replace") + lines = [ln for ln in text.splitlines() if ln.strip()] + if lines: + tail = " | ".join(ln.strip() for ln in lines[-3:])[:500] + except OSError: + tail = "" + detail = f"原始日志尾部：{tail}" if tail else f"原始错误见日志：{log_path}" + action = "重试后" if retried else "" + return ( + f"提取启动阶段失败{action}，可能受主机 GPU/显示重配置干扰，请重试" + f"（MinerU 退出码 {returncode}；{detail}）。" + ) + + def build_mineru_runtime() -> tuple[list[str], dict[str, str], dict[str, str]]: env = os.environ.copy() env.setdefault("MINERU_MODEL_SOURCE", "modelscope") @@ -384,6 +422,9 @@ def build_mineru_runtime() -> tuple[list[str], dict[str, str], dict[str, str]]: else: env.setdefault("CUDA_VISIBLE_DEVICES", "0") env.setdefault("TORCH_CUDNN_V8_API_DISABLED", "1") + # 缺陷 #38（2026-09-11）：子进程 API 的异常日志即写即落（不再被 stdout + # 块缓冲吞掉）——CLI 失败后立即杀 API 进程前真因已落到 job.log。 + env["PYTHONUNBUFFERED"] = "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") @@ -566,7 +607,34 @@ def run_uploaded_methodology_job( stdout=handle, stderr=subprocess.STDOUT, ) + # 缺陷 #38：启动期失败自动重试 1 次——returncode≠0 且 job.log 无 + # doc_analyze_streaming 行 = pipeline 起来前就失败（真因曾因块缓冲 + # 丢失）。重试的 stdout 继续 append 到同一 job.log，日志/状态留 + # retry 标记（G6-2：已有 pipeline 行的中途失败不得触发重试）。 + startup_retried = False + if result.returncode != 0 and not job_log_has_pipeline_line(log_path): + startup_retried = True + with log_path.open("a", encoding="utf-8") as handle: + handle.write( + f"[retry] MinerU 启动期失败（exit {result.returncode}，" + "无 pipeline 阶段日志），自动重试 1/1\n" + ) + handle.write("$ " + " ".join(cmd) + "\n") + handle.flush() + result = subprocess.run( + cmd, + cwd=PROJECT_ROOT, + env=mineru_env, + text=True, + stdin=subprocess.DEVNULL, + stdout=handle, + stderr=subprocess.STDOUT, + ) if result.returncode != 0: + if not job_log_has_pipeline_line(log_path): + raise RuntimeError( + summarize_mineru_startup_failure(log_path, result.returncode, retried=startup_retried) + ) raise RuntimeError(summarize_mineru_failure(log_path, result.returncode)) matches = sorted(mineru_out.rglob("*_content_list.json")) if not matches: diff --git a/agent/methodology/api/test_mineru_retry.py b/agent/methodology/api/test_mineru_retry.py new file mode 100644 index 0000000..cad974c --- /dev/null +++ b/agent/methodology/api/test_mineru_retry.py @@ -0,0 +1,328 @@ +"""test_mineru_retry.py —— MinerU 启动期失败重试与可读错误（缺陷 #38，验收 §4 + 回归 G6）。 + +运行（worktree 根目录，无需部署、不触服务）： + /home/xzq/5090/teamWorkspace/zh/plantai-rag-new-venv/bin/python -m pytest agent/methodology/api/test_mineru_retry.py -v + /home/xzq/5090/teamWorkspace/zh/plantai-rag-new-venv/bin/python -m pytest agent/legacy-rag/api/test_mineru_retry.py -v +说明：agent/methodology 与 agent/legacy-rag 共用同一份 routes（文件名同、内容同）， +同一测试文件复制两处、两处同跑（各用各树的 src/）。 + +断言： +1. build_mineru_runtime 返回 env 含 PYTHONUNBUFFERED=1； +2. subprocess.run 首次失败 + job.log 无 pipeline 行 → 自动重试 1 次 + （两次启动行、job.log 有 [retry] 标记，第二次 stdout 追加同一 job.log）； +3. 重试仍失败 → status failed + 中文可读「提取启动阶段失败」+ 原始错误尾部； +4. 已有 pipeline 行的中途失败 → 不重试，走原错误路径（含 CUDA OOM 中文错误）。 +""" +from __future__ import annotations + +import json +import subprocess +from dataclasses import dataclass, field +from pathlib import Path + +import pytest + +import api.methodology_routes as mr + +from api.methodology_routes import ( + MINERU_PIPELINE_LOG_MARKER, + build_mineru_runtime, + job_log_has_pipeline_line, + run_uploaded_methodology_job, + summarize_mineru_startup_failure, +) + + +# --------------------------------------------------------------------------- +# 固定夹具 +# --------------------------------------------------------------------------- + +@dataclass +class FakeResult: + returncode: int + + +@dataclass +class RunnerRecorder: + """按调用顺序返回预设 returncode；每次调用把 stdout 写进所给 handle（模拟子进程日志）。""" + + sequence: list + calls: list = field(default_factory=list) + + def run(self, cmd, **kwargs): + n = len(self.calls) + self.calls.append({"cmd": cmd, "kwargs": kwargs}) + rc = int(self.sequence[min(n, len(self.sequence) - 1)]) + handle = kwargs.get("stdout") + if handle is not None and kwargs.get("text"): + handle.write(f"[fake stdout call#{n}] rc={rc}\n") + handle.flush() + return FakeResult(returncode=rc) + + +@pytest.fixture +def fake_success(monkeypatch): + """LLM 批次与 bbav5 配置阶段全部假成功（不触真实脚本/目录/真实函数）。 + + - gf_python_path / _preflight_executable 桩掉（不依赖真实 python 路径） + - build_bbav5_methodology_config 桩掉（不要求真实批次产物目录） + - figure_table 阶段走 routes 内部 try/except，不阻断 + """ + captured = {} + + def fake_batch(cmd, **kwargs): + captured["batch_cmd"] = cmd + return FakeResult(returncode=0) + + monkeypatch.setattr(mr, "gf_python_path", lambda: Path("/usr/bin/python3")) + monkeypatch.setattr(mr, "_preflight_executable", lambda cmd, hint, name: None) + monkeypatch.setattr(mr.subprocess, "run", fake_batch) + monkeypatch.setattr( + mr, + "build_bbav5_methodology_config", + lambda output_dir: captured.setdefault("config", {}), + ) + return captured + + +@pytest.fixture +def job_root(monkeypatch, tmp_path): + """把 UPLOAD_JOBS_DIR 重定向到 tmp（绝不写真实数据目录）。""" + root = tmp_path / "upload_jobs" + monkeypatch.setattr(mr, "UPLOAD_JOBS_DIR", root) + return root + + +def read_status(job_dir: Path) -> dict: + return json.loads((job_dir / "status.json").read_text(encoding="utf-8")) + + +def run_job(monkeypatch, tmp_path, runner: RunnerRecorder, source_path: Path, fake_success, job_root): + """以 runner 重绑 subprocess.run 跑完整任务流（fake_success 保证后续阶段假成功）。 + + 注意：fake_success 已把 mr.subprocess.run 桩为假批次；这里再次 setattr + 为 runner（同一模块对象），此后所有子进程调用（mineru 两次 + LLM 批次） + 都由 runner 按 sequence 依次应答——测试 sequence 需覆盖全部调用。 + """ + job_dir = job_root / "testjob" + # 模拟 mineru 成功产出：在 job/mineru 下伪造 *_content_list.json + # （测试不真正执行 mineru） + (job_dir / "mineru").mkdir(parents=True, exist_ok=True) + (job_dir / "mineru" / "fake_content_list.json").write_text("{}", encoding="utf-8") + # GPU 预检禁用（否则 nvidia-smi 会消耗 runner 的 sequence 槽位） + monkeypatch.setenv("METHODOLOGY_MINERU_GPU_PREFLIGHT", "0") + monkeypatch.setattr(subprocess, "run", runner.run) + run_uploaded_methodology_job( + job_id="testjob", + source_path=source_path, + paper_id="paper1", + family="", + output_dir=tmp_path / "out", + llm_model="deepseek-v4-flash", + llm_request_timeout=75, + sandbox=False, + sandbox_run_id="", + ) + return job_dir + + +@pytest.fixture +def pdf_source(tmp_path): + src = tmp_path / "sample.pdf" + src.write_bytes(b"%PDF-1.4\nfake pdf bytes for startup-failure test\n") + return src + + +@pytest.fixture +def json_source(tmp_path): + src = tmp_path / "sample.json" + src.write_text("{}", encoding="utf-8") + return src + + +# --------------------------------------------------------------------------- +# 修复点 1：env 注入 PYTHONUNBUFFERED +# --------------------------------------------------------------------------- + +def test_build_mineru_runtime_env_unbuffered(monkeypatch): + monkeypatch.delenv("MINERU_CMD", raising=False) + cmd, env, runtime = build_mineru_runtime() + assert env.get("PYTHONUNBUFFERED") == "1" + assert cmd # 默认 MINERU_CMD 路径（不要求存在，build 阶段不校验） + + +# --------------------------------------------------------------------------- +# 修复点 2：启动期失败自动重试 1 次 +# --------------------------------------------------------------------------- + +def test_startup_failure_retries_once(monkeypatch, tmp_path, pdf_source, fake_success, job_root): + runner = RunnerRecorder(sequence=[1, 0, 0]) # mineru 首调失败 → 重试成功 → LLM 批次成功 + job_dir = run_job(monkeypatch, tmp_path, runner, pdf_source, fake_success, job_root) + + log_text = (job_dir / "job.log").read_text(encoding="utf-8") + startup_lines = [ln for ln in log_text.splitlines() if ln.startswith("$ ")] + assert len(startup_lines) >= 2, f"应出现两次 mineru 启动（1 次自动重试），实际：{startup_lines}" + assert "[retry]" in log_text, "job.log 应有可识别重试标记" + # 第二次 subprocess.run 的 stdout 追加到同一 job.log + assert "[fake stdout call#0]" in log_text + assert "[fake stdout call#1]" in log_text + + status = read_status(job_dir) + assert status.get("status") == "completed" + + +def test_startup_failure_retry_then_fail_readable_error(monkeypatch, tmp_path, pdf_source, fake_success, job_root): + runner = RunnerRecorder(sequence=[1, 1]) # 两次都失败，且无 pipeline 行 + job_dir = run_job(monkeypatch, tmp_path, runner, pdf_source, fake_success, job_root) + + log_text = (job_dir / "job.log").read_text(encoding="utf-8") + assert len([ln for ln in log_text.splitlines() if ln.startswith("$ /")]) == 2 # 两次 mineru 启动 + assert "[retry]" in log_text + # pipeline 判定：仅检查子进程输出行（启动命令行本身含 marker 关键字） + stdout_lines = [ln for ln in log_text.splitlines() if "[fake stdout" in ln or ln.startswith("[retry]")] + assert MINERU_PIPELINE_LOG_MARKER not in "\n".join(stdout_lines) + + status = read_status(job_dir) + assert status.get("status") == "failed" + message = str(status.get("message") or "") + assert "提取启动阶段失败" in message, f"中文可读错误缺失：{message}" + assert "退出码" in message and "1" in message + assert "原始日志尾部" in message or "原始错误见日志" in message + assert message # 非空 + + +def test_startup_failure_with_marker_retries_once_then_no_second_retry(monkeypatch, tmp_path, pdf_source, job_root): + """G6-2 边界：第一次失败无 marker → 触发重试；重试后才有 marker 但 rc≠0 → + 不二次重试，走 summarize_mineru_failure（有 marker 的中途失败路径）。""" + sequence = [1, 1] + calls = [] + + def run_with_marker(cmd, **kwargs): + n = len(calls) + rc = int(sequence[min(n, len(sequence) - 1)]) + calls.append({"cmd": cmd, "kwargs": kwargs}) + handle = kwargs.get("stdout") + if handle is not None and kwargs.get("text"): + if n == 1: # 第二次调用（重试）时写入 pipeline 行 + handle.write(f"[fake] {MINERU_PIPELINE_LOG_MARKER} started\n") + handle.write(f"[fake stdout call#{n}] rc={rc}\n") + handle.flush() + return FakeResult(returncode=rc) + + monkeypatch.setenv("METHODOLOGY_MINERU_GPU_PREFLIGHT", "0") + monkeypatch.setattr(subprocess, "run", run_with_marker) + monkeypatch.setattr(mr, "_preflight_executable", lambda cmd, hint, name: None) + monkeypatch.setattr( + mr, + "summarize_mineru_failure", + lambda log_path, rc: f"CUDA out of memory fake path rc={rc}", + ) + job_dir = job_root / "testjob" + run_uploaded_methodology_job( + job_id="testjob", + source_path=pdf_source, + paper_id="paper1", + family="", + output_dir=tmp_path / "out", + llm_model="deepseek-v4-flash", + llm_request_timeout=75, + sandbox=False, + sandbox_run_id="", + ) + log_text = (job_dir / "job.log").read_text(encoding="utf-8") + assert MINERU_PIPELINE_LOG_MARKER in log_text + assert log_text.count("[retry]") == 1, "有 pipeline 行后不得再次重试" + # 中途失败（有 marker）→ 原错误路径（CUDA 中文错误） + status = read_status(job_dir) + assert status.get("status") == "failed" + assert "CUDA" in str(status.get("message") or "") + + +# --------------------------------------------------------------------------- +# 修复点 4：有 pipeline 行的中途失败不重试 +# --------------------------------------------------------------------------- + +def test_mid_failure_with_pipeline_line_no_retry(monkeypatch, tmp_path, pdf_source, job_root): + sequence = [1] + calls = [] + + def run_with_marker(cmd, **kwargs): + n = len(calls) + rc = int(sequence[min(n, len(sequence) - 1)]) + calls.append({"cmd": cmd, "kwargs": kwargs}) + handle = kwargs.get("stdout") + if handle is not None and kwargs.get("text"): + if n == 0: # 第一次调用即写 pipeline 行 + handle.write(f"[fake] {MINERU_PIPELINE_LOG_MARKER} started\n") + handle.write(f"[fake stdout call#{n}] rc={rc}\n") + handle.flush() + return FakeResult(returncode=rc) + + monkeypatch.setenv("METHODOLOGY_MINERU_GPU_PREFLIGHT", "0") + monkeypatch.setattr(subprocess, "run", run_with_marker) + monkeypatch.setattr(mr, "_preflight_executable", lambda cmd, hint, name: None) + monkeypatch.setattr( + mr, + "summarize_mineru_failure", + lambda log_path, rc: f"CUDA out of memory fake path rc={rc}", + ) + job_dir = job_root / "testjob" + run_uploaded_methodology_job( + job_id="testjob", + source_path=pdf_source, + paper_id="paper1", + family="", + output_dir=tmp_path / "out", + llm_model="deepseek-v4-flash", + llm_request_timeout=75, + sandbox=False, + sandbox_run_id="", + ) + log_text = (job_dir / "job.log").read_text(encoding="utf-8") + assert MINERU_PIPELINE_LOG_MARKER in log_text + assert "[retry]" not in log_text, "已有 pipeline 行的中途失败不得触发重试" + # 原始启动块 = 1 条 `$ env ...` + 1 条 `$ `；无重试追加 + assert len([ln for ln in log_text.splitlines() if ln.startswith("$ /")]) == 1 + status = read_status(job_dir) + assert status.get("status") == "failed" + assert "CUDA" in str(status.get("message") or "") + + +# --------------------------------------------------------------------------- +# JSON 直接投递（跳过 mineru 段）不重试、不破坏状态机 +# --------------------------------------------------------------------------- + +def test_json_source_skips_mineru_no_retry(monkeypatch, tmp_path, json_source, fake_success, job_root): + runner = RunnerRecorder(sequence=[0]) # 仅 LLM 批次一次 + job_dir = run_job(monkeypatch, tmp_path, runner, json_source, fake_success, job_root) + log_text = (job_dir / "job.log").read_text(encoding="utf-8") + assert "[retry]" not in log_text + assert MINERU_PIPELINE_LOG_MARKER not in log_text + status = read_status(job_dir) + assert status.get("status") == "completed" + + +# --------------------------------------------------------------------------- +# 工具函数单元 +# --------------------------------------------------------------------------- + +def test_job_log_has_pipeline_line(tmp_path): + log_path = tmp_path / "job.log" + assert job_log_has_pipeline_line(log_path) is False + log_path.write_text("noise\n", encoding="utf-8") + assert job_log_has_pipeline_line(log_path) is False + log_path.write_text(f"2026-09-11 [INFO] {MINERU_PIPELINE_LOG_MARKER} begin\n", encoding="utf-8") + assert job_log_has_pipeline_line(log_path) is True + + +def test_summarize_startup_failure_readable(tmp_path): + log_path = tmp_path / "job.log" + log_path.write_text("line one\nline two\nTraceback: boom\n", encoding="utf-8") + msg = summarize_mineru_startup_failure(log_path, 1, retried=True) + assert "提取启动阶段失败" in msg + assert "重试后" in msg + assert "退出码 1" in msg + assert "Traceback: boom" in msg + missing = summarize_mineru_startup_failure(tmp_path / "nope.log", 1, retried=False) + assert "提取启动阶段失败" in missing + assert "重试后" not in missing From 60f7360cec96491ae599ba286a745299f0a1e5dc Mon Sep 17 00:00:00 2001 From: admin Date: Fri, 11 Sep 2026 14:12:57 +0800 Subject: [PATCH 2/3] =?UTF-8?q?fix(r4-52b):=20vault.add=20=E6=8C=89?= =?UTF-8?q?=E5=86=85=E5=AE=B9=20SHA256=20=E5=8E=BB=E9=87=8D,=E9=87=8D?= =?UTF-8?q?=E5=A4=8D=E4=B8=8A=E4=BC=A0=E8=A6=86=E7=9B=96=E6=97=A7=20doc=5F?= =?UTF-8?q?id?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 缺陷 #52b(验收 §6 第 2 条 + 回归 G3):同一 PDF 重复上传 4 次 → 4 个 doc_id × 200 chunk = 800 条,FAISS 检索 top_k 被同内容不同 doc_id 的块占满。 修复(vstore/vault.py): - add() 先算内容哈希 SHA256(原始文件字节),查 index.json 的 content_sha 字段判重;同内容 → 移除旧 doc 的检索行/密文/chunk 行,再落一条新 doc (新 doc_id,meta 以本次上传为准,created_at 刷新) - 去重后该内容在库中恒仅 1 个 doc_id、chunk 总数不翻倍;同名不同内容 按内容哈希不误判(G3-2);不同内容互不覆盖(G3-1) - content_sha 落 index.json(doc_meta 新字段);rebuild_index 对历史密文 缺 content_sha 时以密文内容补算(重建后去重依据不丢) - 与现有删除机制一致:index.remove_item 过滤重建 faiss、chunks.remove_document 过滤重建 chunk faiss(IndexFlatIP 不支持逐行删,现有机制即重建) 测试:vstore/tests/test_vault_dedup.py(5 用例):同内容两次 → 1 个 doc_id/ chunk 不翻倍/旧密文清理;meta 以本次为准;不同内容 → 2 个 doc_id; 同名不同内容不误判;重启(新实例 load)后仍去重。 Co-Authored-By: Claude Fable 5 --- vstore/tests/test_vault_dedup.py | 123 +++++++++++++++++++++++++++++++ vstore/vault.py | 100 ++++++++++++++++++++++--- 2 files changed, 212 insertions(+), 11 deletions(-) create mode 100644 vstore/tests/test_vault_dedup.py diff --git a/vstore/tests/test_vault_dedup.py b/vstore/tests/test_vault_dedup.py new file mode 100644 index 0000000..8526516 --- /dev/null +++ b/vstore/tests/test_vault_dedup.py @@ -0,0 +1,123 @@ +"""test_vault_dedup.py —— 内容 SHA256 去重（缺陷 #52b，验收 §6 第 2 条 + 回归 G3）。 + +运行：cd vstore && /home/xzq/5090/teamWorkspace/zh/plantai-rag-new-venv/bin/python -m pytest tests/test_vault_dedup.py -v +（或 vstore/pyproject.toml testpaths 下全量 pytest；venv = plantai-rag-new-venv） + +G3 断言： +- 同内容 add 两次 → 库中该内容仅 1 个 doc_id、chunk 总数不翻倍； +- 不同内容 add → 2 个 doc_id 互不覆盖； +- 同名不同内容 → 按内容哈希不误判（保留 2 个 doc_id）。 +""" +from __future__ import annotations + +import hashlib + +import pytest + +from vstore.vault import Vault + + +@pytest.fixture +def vault(tmp_path, embed_fn, monkeypatch): + from vstore.settings import VstorePaths, set_paths + + set_paths(VstorePaths(root=str(tmp_path), vault_path="", index_path="", chunk_path="")) + # 桩掉 chunk 批量嵌入（vault._chunked_embed 动态查找），不触网真 Ollama + dim = 1024 + monkeypatch.setattr( + "vstore.embed.embed_batch_chunked", + lambda texts, *a, **k: [[1.0 / dim] * dim for _ in texts], + ) + return Vault(tmp_path / "vault", tmp_path / "index") + + +def _doc_ids_for_content(vault, content: bytes) -> list[str]: + """库中 content_sha == sha256(content) 的 doc 行 id 列表（磁盘事实）。""" + sha = hashlib.sha256(content).hexdigest() + import json + + rows = [] + try: + data = json.loads(vault.index.json_path.read_text(encoding="utf-8")) + items = data.get("items", []) if isinstance(data, dict) else [] + except (ValueError, OSError): + items = [] + for item in items: + if isinstance(item, dict) and str(item.get("content_sha") or "") == sha: + rows.append(item) + return [r.get("id") for r in rows] + + +def _make_pdf(text: str) -> bytes: + """真实最小 PDF 字节（pymupdf 可解析、含目标文本）。""" + import pymupdf + + doc = pymupdf.open() + page = doc.new_page() + page.insert_text((72, 72), text) + return doc.tobytes() + + +def test_same_content_added_twice_keeps_one_doc_id(vault): + """#52b：同内容 add 两次 → 1 个 doc_id、chunk 数不翻倍、旧密文被清。""" + content = _make_pdf("同一份论文内容。") + first = vault.add(content, "paper.pdf", {"name": "论文", "category": "资料"}) + assert first["content_sha"] == hashlib.sha256(content).hexdigest() + chunks_after_first = vault.chunks.count() + assert chunks_after_first >= 1 + + second = vault.add(content, "paper.pdf", {"name": "论文", "category": "资料"}) + + ids = _doc_ids_for_content(vault, content) + assert ids == [second["document_id"]], f"库中该内容 doc_id 应为仅新 doc：{ids}" + assert second["document_id"] != first["document_id"] + # 旧密文被清理（覆盖语义不留两份） + assert not (vault.vault_dir / f"{first['document_id']}.enc").exists() + assert (vault.vault_dir / f"{second['document_id']}.enc").exists() + # chunk 总数不翻倍（旧 doc_id 的 chunk 已清理） + assert vault.chunks.count() == chunks_after_first + # chunk 行全部属于新 doc_id（旧 doc_id 不再出现在检索结果） + assert {r["doc_id"] for r in vault.chunks.rows()} == {second["document_id"]} + + +def test_duplicate_reuses_latest_meta(vault): + """#52b：覆盖语义——重复上传 meta 以本次为准（created_at 刷新、name 覆盖）。""" + content = "内容甲。".encode("utf-8") + first = vault.add(content, "a.txt", {"name": "旧名"}) + second = vault.add(content, "a.txt", {"name": "新名"}) + rows = vault.list_rows() + assert len(rows) == 1 + assert rows[0]["name"] == "新名" + assert rows[0]["id"] == second["document_id"] + assert first["document_id"] not in [r["id"] for r in rows] + + +def test_different_content_keeps_two_doc_ids(vault): + """G3-1：不同内容上传 → 2 个 doc_id 互不覆盖。""" + a = vault.add("内容甲。".encode("utf-8"), "a.txt", {"name": "甲"}) + b = vault.add("内容乙。".encode("utf-8"), "b.txt", {"name": "乙"}) + assert a["document_id"] != b["document_id"] + assert len(vault.list_rows()) == 2 + assert set(_doc_ids_for_content(vault, "内容甲。".encode("utf-8"))) == {a["document_id"]} + assert set(_doc_ids_for_content(vault, "内容乙。".encode("utf-8"))) == {b["document_id"]} + + +def test_same_name_different_content_not_deduped(vault): + """G3-2：同名不同内容 → 按 SHA256 不误判（保留 2 个 doc_id）。""" + a = vault.add(_make_pdf("内容甲。"), "same.pdf", {"name": "同名"}) + b = vault.add(_make_pdf("内容乙。"), "same.pdf", {"name": "同名"}) + assert a["document_id"] != b["document_id"] + assert len(vault.list_rows()) == 2 + assert vault.chunks.count() >= 2 + + +def test_duplicate_dedup_persists_across_reload(vault, tmp_path, monkeypatch): + """#52b：去重依据（content_sha）落盘持久——重启后（新实例 load）仍不判重。""" + content = "持久化内容。".encode("utf-8") + vault.add(content, "a.txt", {"name": "甲"}) + vault2 = Vault(tmp_path / "vault", tmp_path / "index") + vault2.index.load() + vault2.chunks.load() + item = vault2.add(content, "a.txt", {"name": "甲"}) + assert _doc_ids_for_content(vault2, content) == [item["document_id"]] + assert vault2.chunks.count() >= 1 diff --git a/vstore/vault.py b/vstore/vault.py index 2a60a42..d9d32a7 100644 --- a/vstore/vault.py +++ b/vstore/vault.py @@ -130,22 +130,22 @@ class Vault: # ---- 增删改查 ---- - def add( + def _add_file( self, + doc_id: str, file_bytes: bytes, filename: str, - meta: dict[str, Any], - as_template: bool = False, + name: str, + category: str, + tags: str, + description: str, + as_template: bool, ) -> dict[str, Any]: - """入库一条。原文只在函数内流转，末尾丢弃引用。""" - doc_id = str(uuid.uuid4()) - name = str(meta.get("name") or filename or "未命名") - category = str(meta.get("category") or "") - tags = str(meta.get("tags") or "") - description = str(meta.get("description") or "") + """入库一条（覆盖语义已由调用方保证：同内容 doc 已被移除）。 - if not file_bytes: - raise VaultError("上传文件为空") + 与旧 add() 唯一差异是 doc_id 由调用方传入；函数内失败照常抛 VaultError， + 由调用方决定失败回滚语义。 + """ try: extracted = extract(file_bytes, filename) except ParsingError as exc: @@ -162,6 +162,7 @@ class Vault: "description": description, "as_template": bool(as_template), "size": len(file_bytes), + "content_sha": hashlib.sha256(bytes(file_bytes)).hexdigest(), "created_at": _now_iso(), "embed_chunks": 0, } @@ -252,6 +253,7 @@ class Vault: "description": description, "as_template": bool(as_template), "enc_path": str(enc_path), + "content_sha": doc_meta["content_sha"], "embed_sha": embed_sha, "embed_chunks": embed_chunks, # FR-PRIV-002：chunk 级索引结果（失败不阻断 doc 级） @@ -260,6 +262,75 @@ class Vault: "chunk_error": chunk_error, } + def _find_content_duplicate(self, content_sha: str) -> dict[str, Any] | None: + """库中同内容哈希的 doc 行（无则 None）。 + + 内容哈希存于 index.json 每行的 content_sha 字段（#52b 新增）；历史行 + 无该字段（字段缺失/不一致时绝不判重）。直接读 index.json 而非内存 + self.index._rows——覆盖语义要求与磁盘一致（同内容密文先删后建）。 + """ + try: + data = json.loads(self.index.json_path.read_text(encoding="utf-8")) + items = data.get("items", []) if isinstance(data, dict) else [] + except (ValueError, OSError): + return None + for item in items: + if isinstance(item, dict) and str(item.get("content_sha") or "") == content_sha: + return item + return None + + def add( + self, + file_bytes: bytes, + filename: str, + meta: dict[str, Any], + as_template: bool = False, + ) -> dict[str, Any]: + """入库一条（内容 SHA256 去重，缺陷 #52b）。原文只在函数内流转，末尾丢弃引用。 + + 去重语义（验收 §6 第 2 条 + 回归 G3）： + - 内容哈希 = SHA256(原文件字节)。同内容重复上传 → 移除旧 doc 的密文/ + 检索行/chunk 行 → 落一条新 doc；该内容在库中始终仅 1 个 doc_id， + chunk 总数不翻倍。 + - 旧行清理先于新 doc 落盘；新 doc 建库失败时旧 doc 已删（覆盖语义， + 失败由调用方重传即可，绝无同内容双份）。 + - 新 doc 的 meta 以本次上传为准（name/category/tags/description 覆盖， + created_at 刷新）；as_template 以本次为准。 + """ + doc_id = str(uuid.uuid4()) + name = str(meta.get("name") or filename or "未命名") + category = str(meta.get("category") or "") + tags = str(meta.get("tags") or "") + description = str(meta.get("description") or "") + + if not file_bytes: + raise VaultError("上传文件为空") + + # ---- 内容去重（#52b）：同内容哈希 → 先清旧 doc 再落新 doc ---- + content_sha = hashlib.sha256(bytes(file_bytes)).hexdigest() + dup = self._find_content_duplicate(content_sha) + if dup: + old_doc_id = str(dup.get("id") or "") + if old_doc_id and old_doc_id != doc_id: + try: + self.index.remove_item(old_doc_id) + except Exception as exc: + logger.warning("vault: 去重移除旧检索行失败（%s）：%s", old_doc_id, exc) + old_enc = self.vault_dir / f"{old_doc_id}.enc" + if old_enc.exists(): + try: + secure_delete(old_enc) + except OSError as exc: + logger.warning("vault: 去重移除旧密文失败（%s）：%s", old_enc, exc) + try: + self.chunks.remove_document(old_doc_id) + except Exception as exc: + logger.warning("vault: 去重移除旧 chunk 行失败（%s）：%s", old_doc_id, exc) + + return self._add_file( + doc_id, file_bytes, filename, name, category, tags, description, as_template + ) + def get_decrypted(self, doc_id: str) -> dict[str, Any] | None: """取解密后的文档结构（meta + extracted）。不存在返回 None。""" enc_path = self.vault_dir / f"{doc_id}.enc" @@ -362,6 +433,13 @@ class Vault: row = dict(meta) # keywords 只用 meta，不含正文派生的 title（明文纪律） row["keywords"] = _meta_to_keywords(meta) + # content_sha（#52b）：历史密文 meta 若缺失则以密文内容补算（与入库时一致）， + # 保证重建后去重依据不丢 + if not row.get("content_sha"): + try: + row["content_sha"] = hashlib.sha256(load_encrypted(enc)).hexdigest() + except (CryptoError, VaultError, OSError): + row["content_sha"] = "" items.append((row, vectors[0])) try: count = self.index.rebuild(items) From b5dc12e8a5db444bb3f96017132ab0c531dd9645 Mon Sep 17 00:00:00 2001 From: admin Date: Fri, 11 Sep 2026 14:13:08 +0800 Subject: [PATCH 3/3] =?UTF-8?q?fix(r4-52c):=20=5Fpick=5Fitems=20=E5=85=88?= =?UTF-8?q?=E6=8C=89=20doc=5Fid=20=E8=81=9A=E5=90=88=E5=86=8D=E5=8F=96?= =?UTF-8?q?=E5=9D=97,=E9=87=8D=E5=A4=8D=E6=96=87=E6=A1=A3=E4=B8=8D?= =?UTF-8?q?=E5=86=8D=E5=8D=A0=E6=BB=A1=E5=90=8D=E9=A2=9D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 缺陷 #52c(验收 §6 第 3 条):_pick_items 原实现按 score 序逐条取 top_k 再 每 doc ≤3 去重,4 份重复上传的作者页块(得分几乎相同)占满 top-8,正文块 从未进入上下文,模型只能答「资料不足」。 修复(sdk/app/services/privateqa.py): - _pick_items 改为先按 doc_id 聚合:doc 级得分 = 组内 max score; 按 doc 级得分降序取 top 文档(与 top_k 对齐);每选中文档取组内得分 最高的 ≤ PRIVATE_QA_PER_DOC_MAX(3) 块;块内按得分排序 - 组内近重复块(归一化 snippet 前 48 字符指纹相同——重复上传残留的 同一页面切块结果逐字相同)只取 1 条,防 200 条同内容作者页块把该 doc 的 3 个名额占满、正文块仍进不了上下文;正常相邻块(重叠 1 段)不受影响 - ask_private 候选池从 top_k 放宽到 min(top_k×4, 20):只取 8 条时重复 文档块把池占满、聚合阶段无块可选;最终 _pick_items 仍收敛到 top_k 条 - _is_self_deprecating/_build_context(12k 预算、[资料n] 引用编号)不动 测试:sdk/app/services/test_privateqa_pick.py(8 用例):4 doc × 200 作者页块 + 正文块 → 多 doc 覆盖且含正文块;正常场景 top_k/每 doc 上限语义不变; 单 doc 上限;同分并列稳定;ask_private 候选池 top_k×4(≤20)与 e2e 全链。 Co-Authored-By: Claude Fable 5 --- sdk/app/services/privateqa.py | 80 +++++++++-- sdk/app/services/test_privateqa_pick.py | 183 ++++++++++++++++++++++++ 2 files changed, 250 insertions(+), 13 deletions(-) create mode 100644 sdk/app/services/test_privateqa_pick.py diff --git a/sdk/app/services/privateqa.py b/sdk/app/services/privateqa.py index cd9016a..72879ff 100644 --- a/sdk/app/services/privateqa.py +++ b/sdk/app/services/privateqa.py @@ -3,7 +3,8 @@ BFF 直连 DeepSeek 生成 300 字内带 [资料n] 引用的 markdown 短答： - ask_private(question, *, top_k=8, min_score=0.30) -> dict 检索 chunk 索引 → 空则 {hit:false, mode:"empty", answer:"", items:[]}（不调 LLM） - → 同 doc 按 score 序去重（每 doc ≤3 块）再取 top_k + → 先按 doc_id 聚合（doc 级 max score 排序取 top 文档）再取各 doc 内 + 最相关块（每 doc ≤3，缺陷 #52c） → 上下文拼 `[资料n]《name》（相关度 x.xx）\nsnippet`（12k 预算截断） → complete_text(system 约束, temperature=0.2, max_tokens=800, timeout=40) → {hit:true, mode:"vector", answer, items:[...带 ref], elapsed_ms} @@ -65,24 +66,73 @@ def _safe_float(value: Any) -> bool: def _pick_items(items: list[dict[str, Any]], top_k: int) -> list[dict[str, Any]]: - """同 doc 按 score 序去重（每 doc ≤ PRIVATE_QA_PER_DOC_MAX 块），再取 top_k。 + """先按 doc_id 聚合再取块（缺陷 #52c）：单一文档不再占满全部名额。 - 输入已按 score 降序（vstore faiss 顺序）；同 doc 靠前块优先。 + 原实现按输入 score 序逐条取（每 doc ≤ PRIVATE_QA_PER_DOC_MAX 块）直到 + top_k——重复上传的作者页块（得分几乎相同）会占满 top-8、正文块进不了 + 上下文。新语义： + 1. 按 doc_id 分组，doc 级得分 = 组内 max score； + 2. doc 级得分降序取前 top_k 个文档； + 3. 每选中文档取组内得分最高的 ≤ PRIVATE_QA_PER_DOC_MAX 个块， + 且组内近重复块（内容指纹相同，重复上传残留产物）只取 1 条—— + 否则「同内容 200 条作者页块」仍会把该 doc 的 3 个名额占满、 + 正文块永远进不了上下文； + 4. 全部选中块按 score 降序稳定排序、截断到 top_k 返回（_build_context + 的 [资料n] 编号与 items 顺序不变，回归语义保留）。 """ picked: list[dict[str, Any]] = [] - per_doc: dict[str, int] = {} + groups: dict[str, list[dict[str, Any]]] = {} + doc_score: dict[str, float] = {} + doc_order: list[str] = [] for item in items: if not isinstance(item, dict): continue doc_id = str(item.get("doc_id") or "") - used = per_doc.get(doc_id, 0) - if used >= PRIVATE_QA_PER_DOC_MAX: - continue - per_doc[doc_id] = used + 1 - picked.append(dict(item)) - if len(picked) >= top_k: - break - return picked + if doc_id not in doc_order: + doc_order.append(doc_id) + groups.setdefault(doc_id, []).append(item) + try: + score = float(item.get("score")) + except (TypeError, ValueError): + score = 0.0 + if score > doc_score.get(doc_id, float("-inf")): + doc_score[doc_id] = score + # 先聚合后排序：得分相同时按输入首现顺序（稳定） + top_docs = sorted(doc_order, key=lambda d: doc_score.get(d, 0.0), reverse=True)[:top_k] + for doc_id in top_docs: + group = sorted(groups.get(doc_id, []), key=_score_of, reverse=True) + taken_fingerprints: set[str] = set() + for item in group: + fingerprint = _chunk_fingerprint(item) + if fingerprint in taken_fingerprints: + continue # 组内近重复块（重复上传残留）不重复占名额 + taken_fingerprints.add(fingerprint) + picked.append(dict(item)) + if len(taken_fingerprints) >= PRIVATE_QA_PER_DOC_MAX: + break + picked.sort(key=_score_of, reverse=True) + return picked[:top_k] + + +def _chunk_fingerprint(item: dict[str, Any]) -> str: + """组内近重复指纹：归一化（去空白）后的 snippet 前 48 字符。 + + 重复上传残留的块内容逐字相同（同内容多次入库切块结果一致）； + 正常相邻块（重叠 1 段）不会整段相同，不受影响。 + """ + import re + + snippet = str(item.get("snippet") or "") + normalized = re.sub(r"\s+", "", snippet) + return normalized[:48] + + +def _score_of(item: dict[str, Any]) -> float: + """item 得分（无/非法 → 0.0）。""" + try: + return float(item.get("score")) + except (TypeError, ValueError): + return 0.0 def _build_context(picked: list[dict[str, Any]]) -> tuple[str, list[dict[str, Any]]]: @@ -119,7 +169,11 @@ async def ask_private( 异常（DocgenUnavailable / Exception / LLMError）原样上抛。 """ started = time.perf_counter() - payload = await search_chunks(question, top_k=top_k, min_score=min_score, timeout=12.0) + # 缺陷 #52c：先聚合再取块需要更宽的候选池——只取 top_k 条时重复文档 + # 的块会把池占满、聚合阶段无块可选。取 top_k×4（vstore 上限 20 clamp）， + # 最终 _pick_items 仍收敛到 top_k 条，响应量与 [资料n] 编号语义不变。 + pool_k = min(max(top_k * 4, top_k), 20) + payload = await search_chunks(question, top_k=pool_k, min_score=min_score, timeout=12.0) items = [i for i in payload.get("items") or [] if isinstance(i, dict)] if not items: return {"hit": False, "mode": "empty", "answer": "", "items": []} diff --git a/sdk/app/services/test_privateqa_pick.py b/sdk/app/services/test_privateqa_pick.py new file mode 100644 index 0000000..e826ee0 --- /dev/null +++ b/sdk/app/services/test_privateqa_pick.py @@ -0,0 +1,183 @@ +"""test_privateqa_pick.py —— _pick_items 先按 doc_id 聚合再取块（缺陷 #52c，验收 §6 第 3 条）。 + +运行：cd sdk && /home/xzq/5090/teamWorkspace/zh/plantai-rag-new-venv/bin/python -m pytest app/services/test_privateqa_pick.py -v +（venv = plantai-rag-new-venv；sdk 的 app/ 布局与 pyproject testpaths 一致） + +构造合成数据（不真跑 FAISS）： +- 4 个 doc_id 各若干「作者页块」（高分、分接近）+ 1 条低分「正文块」。 +- 旧实现会把高分的作者页块占满 top-8、正文块进不了上下文； + 新实现按 doc 级 max score 取 top 文档 → 各 doc 只占 ≤3 块， + 正文块（本组内最高低分块）能进入 picked。 +""" +from __future__ import annotations + +from app.services.privateqa import ( + PRIVATE_QA_PER_DOC_MAX, + _build_context, + _pick_items, +) + + +def make_items(doc_specs): + """doc_specs: [(doc_id, [(score, snippet), ...]), ...] → item 列表（输入序打散， + 模拟 vstore 全局 score 序）。""" + items = [] + for doc_id, chunks in doc_specs: + for score, snippet in chunks: + items.append({"doc_id": doc_id, "score": score, "snippet": snippet, "doc_name": f"资料{doc_id}"}) + return sorted(items, key=lambda it: -it["score"]) + + +def test_dup_docs_no_longer_monopolize_topk(): + """#52c 主场景：4 个 doc_id 各 200 条近同分作者页块（同 doc 内内容相同， + 模拟重复上传残留——同一页面被反复切块/复传）+ 正文块 → + 选出的 items 覆盖多个 doc 且含正文块。""" + specs = [] + for i in range(4): + author = [(0.94 - i * 0.001, f"作者页块 {i}") for _ in range(200)] + body = [(0.30, f"正文块 {i} 方法与结论")] + specs.append((f"doc-{i}", author + body)) + items = make_items(specs) + picked = _pick_items(items, top_k=8) + + doc_ids = [it["doc_id"] for it in picked] + assert len(set(doc_ids)) >= 2, f"重复文档仍占满名额：{doc_ids}" + for doc_id in set(doc_ids): + assert doc_ids.count(doc_id) <= PRIVATE_QA_PER_DOC_MAX + snippets = [it["snippet"] for it in picked] + assert any("正文块" in s for s in snippets), "正文块未进入上下文" + # 总分 ≤ top_k；块按分数降序 + assert len(picked) <= 8 + scores = [it["score"] for it in picked] + assert scores == sorted(scores, reverse=True) + + +def test_normal_case_semantics_preserved(): + """回归：正常场景 top_k/每 doc 上限语义不变。""" + items = [ + {"doc_id": "A", "score": 0.95, "snippet": "a1"}, + {"doc_id": "A", "score": 0.90, "snippet": "a2"}, + {"doc_id": "B", "score": 0.85, "snippet": "b1"}, + {"doc_id": "C", "score": 0.80, "snippet": "c1"}, + {"doc_id": "D", "score": 0.75, "snippet": "d1"}, + ] + picked = _pick_items(items, top_k=5) + assert len(picked) == 5 + for doc_id in {it["doc_id"] for it in picked}: + assert sum(1 for it in picked if it["doc_id"] == doc_id) <= PRIVATE_QA_PER_DOC_MAX + # 每 doc 内按分数降序取 ≤3 块 + a = [it for it in picked if it["doc_id"] == "A"] + assert [it["snippet"] for it in a] == ["a1", "a2"] + + +def test_single_doc_caps_at_per_doc_max(): + """回归：单一 doc 大量块时仍 ≤ per-doc 上限。""" + items = [{"doc_id": "X", "score": 0.9 - i * 0.001, "snippet": f"x{i}"} for i in range(20)] + picked = _pick_items(items, top_k=8) + assert 1 <= len(picked) <= PRIVATE_QA_PER_DOC_MAX + + +def test_ties_are_stable(): + """回归：同分并列时文档选择稳定（按输入首现顺序），结果确定。""" + items = make_items( + [ + ("A", [(0.9, "a1"), (0.9, "a2")]), + ("B", [(0.9, "b1")]), + ("C", [(0.9, "c1")]), + ] + ) + p1 = _pick_items(items, top_k=4) + p2 = _pick_items(items, top_k=4) + assert [it["doc_id"] for it in p1] == [it["doc_id"] for it in p2] + assert len(p1) == 4 + + +def test_malformed_items_ignored(): + """回归：非 dict / 缺 doc_id 条目不炸。""" + picked = _pick_items( + [None, "bad", {"score": 0.9, "snippet": "s"}, {"doc_id": "Z", "score": 0.5, "snippet": "z"}], + top_k=4, + ) + assert picked + + +def test_build_context_integration(): + """回归：_build_context 12k 预算截断 + [资料n] 引用编号与 items 顺序一致。""" + picked = _pick_items( + [ + {"doc_id": "A", "score": 0.95, "snippet": "正文一"}, + {"doc_id": "A", "score": 0.90, "snippet": "正文二"}, + {"doc_id": "B", "score": 0.80, "snippet": "正文三"}, + ], + top_k=3, + ) + context, out_items = _build_context(picked) + assert "[资料1]" in context + assert out_items[0]["ref"] == 1 + assert [it["snippet"] for it in out_items] == [it["snippet"] for it in picked] + + +async def test_ask_private_fetches_wider_pool(monkeypatch): + """#52c：ask_private 以 top_k×4 向 vstore 取候选池，再收敛到 top_k。""" + import app.services.privateqa as pq + + calls = {} + + async def fake_search(query, *, top_k, min_score, timeout): + calls["requested_top_k"] = top_k + items = [] + for doc_i in range(4): + for j in range(200): + items.append( + { + "doc_id": f"doc-{doc_i}", + "score": 0.94 - doc_i * 0.001, + # 同 doc 内作者页块内容相同（重复上传残留的真实形态） + "snippet": f"作者页块 {doc_i}", + "doc_name": f"资料{doc_i}", + } + ) + items.append( + { + "doc_id": f"doc-{doc_i}", + "score": 0.30, + "snippet": f"正文块 {doc_i} 方法与结论", + "doc_name": f"资料{doc_i}", + } + ) + return {"ok": True, "mode": "vector", "items": items} + + async def fake_llm(system, user, **kwargs): + return "方法与结论见[资料1]。" + + monkeypatch.setattr(pq, "search_chunks", fake_search) + monkeypatch.setattr(pq, "complete_text", fake_llm) + + result = await pq.ask_private("总结这篇论文的主要方法和结论", top_k=8) + assert calls["requested_top_k"] == 20, "候选池应取 min(top_k×4, vstore 上限 20)" + assert result["hit"] is True + assert result["mode"] == "vector" + doc_ids = [it["doc_id"] for it in result["items"]] + assert len(set(doc_ids)) >= 2 + assert any("正文块" in (it["snippet"] or "") for it in result["items"]) + assert len(result["items"]) <= 8 + assert all(it["ref"] >= 1 for it in result["items"]) + + +async def test_ask_private_pool_capped_at_20(monkeypatch): + """#52c：候选池 = min(top_k×4, 20)（vstore CHUNK_TOP_K_MAX clamp）。""" + import app.services.privateqa as pq + + calls = {} + + async def fake_search(query, *, top_k, min_score, timeout): + calls["requested_top_k"] = top_k + return {"ok": True, "mode": "vector", "items": []} + + monkeypatch.setattr(pq, "search_chunks", fake_search) + await pq.ask_private("问题", top_k=8) + assert calls["requested_top_k"] == 20 + await pq.ask_private("问题", top_k=5) + assert calls["requested_top_k"] == 20 # 5×4=20 + await pq.ask_private("问题", top_k=3) + assert calls["requested_top_k"] == 12 # 3×4=12（低于 20）