From fcc601fcf3bb1a9ca846cf678461672cafa0b22e Mon Sep 17 00:00:00 2001 From: yxx <2412119399@qq.com> Date: Thu, 3 Sep 2026 22:20:11 +0800 Subject: [PATCH] =?UTF-8?q?fix(backend):=20=E8=90=BD=E5=AE=9E=20PR=20#11?= =?UTF-8?q?=20=E7=AC=AC=E4=BA=8C=E8=BD=AE=E8=AF=84=E5=AE=A1=E6=84=8F?= =?UTF-8?q?=E8=A7=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Benchmark 容量淘汰只删终态 run，满容量且全活动时返回 BENCHMARK_CAPACITY_EXCEEDED - 创建 run 前校验索引兼容性（BENCHMARK_INDEX_INCOMPATIBLE） - 取消 run 补发 RunCancelled 终止事件；失败分支脱敏（BENCHMARK_RUN_FAILED） - 失败样本计入汇总分母，报告输出 total/successful/failed/failure_rate - load_dataset 按文件名隔离无关损坏文件，顶层非对象拒绝 - FTS score_threshold 先于计数/分页，total 与 items 一致 - Benchmark SSE 支持 Last-Event-ID 游标 - 移除 Agent Benchmark 501 占位接口 - 同步第二阶段接口契约与开发说明文档 Co-Authored-By: Claude Code --- backend/app/benchmarks/datasets.py | 18 ++- backend/app/benchmarks/rag.py | 44 +++++-- backend/app/benchmarks/service.py | 102 ++++++++++++--- backend/app/contracts.py | 9 ++ backend/app/retrieval/engine.py | 33 ++--- backend/app/retrieval/vectorstore.py | 8 ++ backend/app/routes.py | 44 ++++--- backend/tests/test_benchmark.py | 118 +++++++++++++++++- backend/tests/test_retrieval.py | 38 ++++++ docs/contracts/第二阶段接口契约-开发版.md | 27 +++- .../Knowledge与Retrieval-Core开发说明.md | 6 +- 11 files changed, 378 insertions(+), 69 deletions(-) diff --git a/backend/app/benchmarks/datasets.py b/backend/app/benchmarks/datasets.py index 3a3866b..fff0661 100644 --- a/backend/app/benchmarks/datasets.py +++ b/backend/app/benchmarks/datasets.py @@ -172,11 +172,23 @@ def list_datasets(kind: BenchmarkKind) -> list[BenchmarkDatasetInfo]: def load_dataset(dataset_id: str, kind: BenchmarkKind) -> RAGDataset: - """按 id 加载并校验数据集；找不到抛 BENCHMARK_DATASET_NOT_FOUND。""" + """按文件名加载并校验数据集；找不到抛 BENCHMARK_DATASET_NOT_FOUND。 + + 只读取与请求 dataset_id 同名的文件（{dataset_id}.json），无关文件的损坏（JSON 语法 + 错误、UTF-8 解码错误、顶层非对象）不会阻断目标数据集加载；只有目标文件本身损坏 + 才抛 BENCHMARK_DATASET_INVALID。按现有文件 stem 精确匹配，不拼接调用方传入的路径。 + """ for path in _dataset_files(): - raw, raw_bytes = _read_json(path) - if raw.get("dataset_id") != dataset_id: + if path.stem != dataset_id: continue + raw, raw_bytes = _read_json(path) + if not isinstance(raw, dict): + raise ApiError( + 422, + "BENCHMARK_DATASET_INVALID", + "Dataset top-level must be a JSON object.", + {"dataset_id": dataset_id, "path": path.name}, + ) return _dataset_from_raw(raw, raw_bytes, kind) raise ApiError( 404, diff --git a/backend/app/benchmarks/rag.py b/backend/app/benchmarks/rag.py index 2ec4ec3..54cec1d 100644 --- a/backend/app/benchmarks/rag.py +++ b/backend/app/benchmarks/rag.py @@ -1,12 +1,13 @@ """RAG Benchmark Runner：调用检索引擎对数据集逐 Case 求值并聚合指标。 只读操作，直接复用 app.retrieval.engine 的 search()，不旁路检索链路。指标按 -(mode, case, repeat) 逐样本计算，再按 mode 聚合；失败样本保留在报告中但不计入汇总， -避免异常样本污染指标。 +(mode, case, repeat) 逐样本计算，再按 mode 聚合；失败样本按零分计入质量指标分母， +避免把执行失败误判为检索质量（同时保留 total/successful/failed/failure_rate）。 """ from __future__ import annotations +import logging import time from collections.abc import Callable @@ -22,6 +23,8 @@ from app.contracts import ( ) from app.retrieval.engine import engine +logger = logging.getLogger(__name__) + class BenchmarkCancelled(Exception): """运行在 Case 之间被取消时抛出，用于中断后台执行并标记 cancelled。""" @@ -75,12 +78,19 @@ async def _evaluate_one( response = await engine.search(search_request) latency_ms = (time.perf_counter() - start) * 1000.0 except Exception as exc: # 单个样本失败不中断整个 Benchmark + # 详细异常只进日志，公开响应只带项目错误码与安全消息，避免泄露路径/SQL 等敏感信息 + logger.warning( + "RAG case evaluation failed: case=%s mode=%s", case.case_id, mode.value, + exc_info=exc, + ) return RAGCaseResult( case_id=case.case_id, mode=mode, repeat=repeat, latency_ms=(time.perf_counter() - start) * 1000.0, - error=str(exc), + citation_applicable=case.citation_required, + error="RAG case evaluation failed.", + error_code="BENCHMARK_CASE_EVALUATION_FAILED", ) retrieved_note_ids = [item.note_id for item in response.items] @@ -107,19 +117,27 @@ async def _evaluate_one( def _aggregate(cases: list[RAGCaseResult], mode: SearchMode) -> RAGMetrics: samples = [c for c in cases if c.mode == mode] - ok = [c for c in samples if c.error is None] - if not ok: + total = len(samples) + failed = sum(1 for c in samples if c.error is not None) + successful = total - failed + if total == 0: return RAGMetrics() - latencies = [c.latency_ms for c in ok] - # citation_hit_rate 只统计声明了 expected_block_ids 的样本 - citation_samples = [c for c in ok if c.citation_applicable] + # 延迟只统计成功样本；失败样本按零分计入质量指标分母，避免汇总虚高 + latencies = [c.latency_ms for c in samples if c.error is None] + citation_samples = [c for c in samples if c.citation_applicable] return RAGMetrics( - hit_at_1=m.mean([1.0 if c.hit_at_1 else 0.0 for c in ok]), - hit_at_5=m.mean([1.0 if c.hit_at_5 else 0.0 for c in ok]), - recall_at_k=m.mean([c.recall for c in ok]), - mrr=m.mean([c.reciprocal_rank for c in ok]), - citation_hit_rate=m.mean([1.0 if c.citation_hit else 0.0 for c in citation_samples]), + hit_at_1=m.mean([1.0 if (c.error is None and c.hit_at_1) else 0.0 for c in samples]), + hit_at_5=m.mean([1.0 if (c.error is None and c.hit_at_5) else 0.0 for c in samples]), + recall_at_k=m.mean([c.recall if c.error is None else 0.0 for c in samples]), + mrr=m.mean([c.reciprocal_rank if c.error is None else 0.0 for c in samples]), + citation_hit_rate=m.mean( + [1.0 if (c.error is None and c.citation_hit) else 0.0 for c in citation_samples] + ), p50_latency_ms=m.percentile(latencies, 50.0), p95_latency_ms=m.percentile(latencies, 95.0), + total_cases=total, + successful_cases=successful, + failed_cases=failed, + failure_rate=failed / total, ) diff --git a/backend/app/benchmarks/service.py b/backend/app/benchmarks/service.py index dcad814..9b04a27 100644 --- a/backend/app/benchmarks/service.py +++ b/backend/app/benchmarks/service.py @@ -9,6 +9,7 @@ RAG Benchmark 采用「创建即返回 queued、后台 Task 异步执行」的 from __future__ import annotations import asyncio +import logging import sys from datetime import datetime, timezone from uuid import uuid4 @@ -28,10 +29,13 @@ from app.contracts import ( RAGCaseResult, RAGMetrics, RAGRunRequest, + SearchMode, ) from app.errors import ApiError from app.retrieval.engine import engine +logger = logging.getLogger(__name__) + _runs: dict[str, BenchmarkRun] = {} _events: dict[str, list[BenchmarkEvent]] = {} _reports: dict[str, BenchmarkReport] = {} @@ -45,16 +49,31 @@ def _now() -> datetime: return datetime.now(timezone.utc) -def _remember(run: BenchmarkRun) -> None: - _runs[run.run_id] = run - while len(_runs) > MAX_RUNS: - oldest = next(iter(_runs)) - _runs.pop(oldest, None) - _events.pop(oldest, None) - _reports.pop(oldest, None) - _tasks.pop(oldest, None) - _subscribers.pop(oldest, None) - _cancel_flags.pop(oldest, None) +def _forget(run_id: str) -> None: + """移除一条 run 的全部内存态；仅在 run 处于终态时调用，避免打断活动任务。""" + _runs.pop(run_id, None) + _events.pop(run_id, None) + _reports.pop(run_id, None) + _tasks.pop(run_id, None) + _subscribers.pop(run_id, None) + _cancel_flags.pop(run_id, None) + + +def _evict_terminal() -> bool: + """超过容量时淘汰最旧的终态 run；全部为活动 run 无法淘汰时返回 False。 + + 绝不能删除仍在运行（queued/running）的 run：那会连带移除其 _cancel_flags 与 + _subscribers，使后台 Task 访问时抛出 KeyError。 + """ + terminal = (BenchmarkStatus.completed, BenchmarkStatus.failed, BenchmarkStatus.cancelled) + while len(_runs) >= MAX_RUNS: + victim = next( + (rid for rid, run in _runs.items() if run.status in terminal), None + ) + if victim is None: + return False + _forget(victim) + return True def _config_snapshot(request: RAGRunRequest, dataset: RAGDataset) -> dict: @@ -83,12 +102,57 @@ def _config_snapshot(request: RAGRunRequest, dataset: RAGDataset) -> dict: } +async def _validate_index_compatibility(request: RAGRunRequest) -> None: + """创建 RAG Run 前校验索引已建立且与当前 Embedding 模型/维度兼容。 + + 空索引或不兼容索引会让所有模式得到全 0 指标，把环境/索引错误误判为检索质量差， + 故在创建时即拒绝，返回 BENCHMARK_INDEX_INCOMPATIBLE。 + """ + stats = repository.stats() + meta = repository.get_index_meta() + needs_vector = any(m in (SearchMode.vector, SearchMode.hybrid) for m in request.modes) + + reasons: list[str] = [] + if stats["blocks"] == 0: + reasons.append("index is empty (no indexed blocks; run /api/index/rebuild first)") + if needs_vector: + if meta.get("embedding_model") != engine.embedding.model_id: + reasons.append( + f"embedding model mismatch: index={meta.get('embedding_model')!r}, " + f"engine={engine.embedding.model_id!r}" + ) + if meta.get("embedding_dim") != str(engine.embedding.dim): + reasons.append( + f"embedding dimension mismatch: index={meta.get('embedding_dim')!r}, " + f"engine={engine.embedding.dim}" + ) + if await engine.vector_store.count() == 0: + reasons.append("vector index is empty") + if reasons: + raise ApiError( + 409, + "BENCHMARK_INDEX_INCOMPATIBLE", + "Benchmark index is not built or is incompatible with the current retrieval engine.", + {"reasons": reasons}, + ) + + async def create_rag_run(request: RAGRunRequest) -> BenchmarkRun: """创建一次 RAG Benchmark，立即返回 queued 的 BenchmarkRun，由后台 Task 执行。""" dataset = datasets.load_dataset(request.dataset_id, BenchmarkKind.rag) + await _validate_index_compatibility(request) + + # 容量检查：先淘汰终态 run 腾空间；满容量且全为活动 run 时拒绝创建 + if not _evict_terminal(): + raise ApiError( + 429, + "BENCHMARK_CAPACITY_EXCEEDED", + "Benchmark run capacity exceeded; wait for active runs to finish.", + {}, + ) + run_id = "benchmark_" + uuid4().hex[:12] snapshot = _config_snapshot(request, dataset) - run = BenchmarkRun( run_id=run_id, kind=BenchmarkKind.rag, @@ -99,7 +163,7 @@ async def create_rag_run(request: RAGRunRequest) -> BenchmarkRun: config_snapshot=snapshot, created_at=_now(), ) - _remember(run) + _runs[run_id] = run _events[run_id] = [] _subscribers[run_id] = [] _cancel_flags[run_id] = asyncio.Event() @@ -155,6 +219,7 @@ async def _execute_rag( "completed_at": _now(), } ) + emit(BenchmarkEventType.run_cancelled, {"status": BenchmarkStatus.cancelled.value}) _reports[run_id] = BenchmarkReport( run_id=run_id, kind=BenchmarkKind.rag, @@ -166,15 +231,21 @@ async def _execute_rag( finish() return except Exception as exc: # 单次运行失败不拖垮服务，记录错误后结束 + # 详细异常只进日志，公开响应仅带项目错误码与安全消息，避免泄露路径/SQL 等敏感信息 + logger.exception("Benchmark run failed: run_id=%s", run_id) _runs[run_id] = _runs[run_id].model_copy( update={ "status": BenchmarkStatus.failed, "progress": 1.0, - "error": str(exc), + "error": "Benchmark run failed.", + "error_code": "BENCHMARK_RUN_FAILED", "completed_at": _now(), } ) - emit(BenchmarkEventType.run_failed, {"error": str(exc)}) + emit( + BenchmarkEventType.run_failed, + {"error": "Benchmark run failed.", "error_code": "BENCHMARK_RUN_FAILED"}, + ) _reports[run_id] = BenchmarkReport( run_id=run_id, kind=BenchmarkKind.rag, @@ -182,7 +253,8 @@ async def _execute_rag( dataset_hash=dataset.content_hash, status=BenchmarkStatus.failed, config_snapshot=snapshot, - error=str(exc), + error="Benchmark run failed.", + error_code="BENCHMARK_RUN_FAILED", ) finish() return diff --git a/backend/app/contracts.py b/backend/app/contracts.py index aac441d..ce73a7a 100644 --- a/backend/app/contracts.py +++ b/backend/app/contracts.py @@ -720,6 +720,11 @@ class RAGMetrics(Contract): citation_hit_rate: float = 0.0 p50_latency_ms: float = 0.0 p95_latency_ms: float = 0.0 + # 样本构成：失败样本按零分计入质量指标，汇总不虚高；报告据此可知实际分母 + total_cases: int = 0 + successful_cases: int = 0 + failed_cases: int = 0 + failure_rate: float = 0.0 class BenchmarkDatasetInfo(Contract): @@ -745,6 +750,7 @@ class BenchmarkRun(Contract): metrics: dict[str, Any] | None = None config_snapshot: dict[str, Any] = Field(default_factory=dict) error: str | None = None + error_code: str | None = None created_at: datetime started_at: datetime | None = None completed_at: datetime | None = None @@ -760,6 +766,7 @@ class BenchmarkEventType(str, Enum): case_completed = "CaseCompleted" run_completed = "RunCompleted" run_failed = "RunFailed" + run_cancelled = "RunCancelled" class BenchmarkEvent(Contract): @@ -785,6 +792,7 @@ class RAGCaseResult(Contract): # 该 Case 是否声明了 expected_block_ids（决定是否计入 citation_hit_rate 分母） citation_applicable: bool = False error: str | None = None + error_code: str | None = None class BenchmarkReport(Contract): @@ -797,3 +805,4 @@ class BenchmarkReport(Contract): metrics: dict[str, Any] = Field(default_factory=dict) cases: list[RAGCaseResult] = Field(default_factory=list) error: str | None = None + error_code: str | None = None diff --git a/backend/app/retrieval/engine.py b/backend/app/retrieval/engine.py index 3ac3c2d..15d4ba6 100644 --- a/backend/app/retrieval/engine.py +++ b/backend/app/retrieval/engine.py @@ -29,6 +29,8 @@ from app.textutils import make_snippet, match_query CANDIDATE_POOL = 50 # 分页窗口上限：候选池至少覆盖 offset+limit，但设上限防止超大 offset 撑爆内存 MAX_CANDIDATE_POOL = 200 +# FTS 全量取回上限：统一归一化 + 阈值过滤后再分页，保证阈值语义跨页一致 +FTS_FETCH_LIMIT = 5000 # 带 metadata 过滤时放大召回倍数，缓解「先截断候选池再过滤」造成的漏召回 OVERSCAN_FACTOR = 4 @@ -137,15 +139,18 @@ class RetrievalEngine: ) def _search_fts(self, request: SearchRequest) -> SearchResponse: - """FTS 专用路径：过滤、COUNT 与分页全部在 SQLite 中完成。""" + """FTS 专用路径：先取全量命中（≤FTS_FETCH_LIMIT），统一归一化 + 阈值过滤后再分页。 + + 阈值过滤必须在计数与分页之前完成，否则 score_threshold 只作用于当前页， + 且返回的 total 与 items 数量不一致（如 items 为空但 total 非零）。""" match = match_query(request.query) if not match: return self._empty(request) - fts_hits, total = repository.fts_search_page( + fts_hits, _ = repository.fts_search_page( match=match, - limit=request.limit, - offset=request.offset, + limit=FTS_FETCH_LIMIT, + offset=0, folders=request.folders, note_ids=request.note_ids, tags=request.tags, @@ -155,18 +160,18 @@ class RetrievalEngine: updated_to=request.updated_to, ) if not fts_hits: - return SearchResponse( - query=request.query, - mode=request.mode, - page=PageMeta(total=total, limit=request.limit, offset=request.offset), - ) + return self._empty(request) - hits = {h.block_id: h for h in repository.get_block_hits([hit.block_id for hit in fts_hits])} - ordered = normalize_scores( - [(hit.block_id, -hit.bm25) for hit in fts_hits if hit.block_id in hits] - ) + ordered = normalize_scores([(hit.block_id, -hit.bm25) for hit in fts_hits]) ordered = [(bid, score) for bid, score in ordered if score >= request.score_threshold] - items = [self._build_result(hits[block_id], request, score) for block_id, score in ordered] + total = len(ordered) + page = ordered[request.offset : request.offset + request.limit] + hits = {h.block_id: h for h in repository.get_block_hits([bid for bid, _ in page])} + items = [ + self._build_result(hits[block_id], request, score) + for block_id, score in page + if block_id in hits + ] return SearchResponse( query=request.query, mode=request.mode, diff --git a/backend/app/retrieval/vectorstore.py b/backend/app/retrieval/vectorstore.py index 99e5c1e..07687d0 100644 --- a/backend/app/retrieval/vectorstore.py +++ b/backend/app/retrieval/vectorstore.py @@ -35,6 +35,7 @@ class VectorStore(Protocol): async def upsert(self, records: list[VectorRecord]) -> None: ... async def delete(self, ids: list[str]) -> None: ... async def search(self, vector: list[float], *, top_k: int) -> list[VectorHit]: ... + async def count(self) -> int: ... class SqliteVecStore: @@ -92,3 +93,10 @@ class SqliteVecStore: conn.execute("DELETE FROM vec_blocks") finally: conn.close() + + async def count(self) -> int: + conn = connect() + try: + return conn.execute("SELECT COUNT(*) FROM vec_blocks").fetchone()[0] + finally: + conn.close() diff --git a/backend/app/routes.py b/backend/app/routes.py index b012830..efce047 100644 --- a/backend/app/routes.py +++ b/backend/app/routes.py @@ -72,7 +72,7 @@ from app.agent import AgentCapacityError, AgentRunNotFoundError from app.benchmarks import datasets as benchmark_datasets from app.benchmarks import service as benchmark_service from app.container import container -from app.errors import ApiError, not_implemented +from app.errors import ApiError from app.extensions import ExtensionError from app.providers.registry import ProviderNotFoundError from app.providers.factory import UnsupportedProviderError @@ -868,18 +868,6 @@ async def create_rag_benchmark(request: RAGRunRequest) -> BenchmarkRun: return await benchmark_service.create_rag_run(request) -@router.post( - "/benchmarks/agent/runs", - response_model=BenchmarkRun, - status_code=202, - tags=["Benchmark"], -) -async def create_agent_benchmark() -> BenchmarkRun: - # Agent Benchmark 基础设施在 RAG Benchmark 之后单独交付，先占位契约 - not_implemented("Agent Benchmark") - raise AssertionError("unreachable") - - @router.get( "/benchmarks/runs", response_model=BenchmarkRunListResponse, @@ -945,18 +933,38 @@ async def cancel_benchmark_run(run_id: str) -> OperationResponse: async def benchmark_events( run_id: str, after_sequence: int = Query(default=-1, ge=-1), + last_event_id: str | None = Header(default=None, alias="Last-Event-ID"), ) -> StreamingResponse: if benchmark_service.get_run(run_id) is None: raise ApiError( 404, "BENCHMARK_RUN_NOT_FOUND", "benchmark run not found", {"run_id": run_id} ) + # SSE 断线重连：Last-Event-ID 优先于 after_sequence，用于从上次收到的事件继续 + cursor = after_sequence + if last_event_id is not None: + try: + cursor = int(last_event_id) + except ValueError as exc: + raise ApiError( + 400, + "BENCHMARK_EVENT_CURSOR_INVALID", + "Last-Event-ID must be an integer sequence.", + {"last_event_id": last_event_id}, + ) from exc + if cursor < -1: + raise ApiError( + 400, + "BENCHMARK_EVENT_CURSOR_INVALID", + "Last-Event-ID must be greater than or equal to -1.", + ) + async def stream() -> AsyncIterator[str]: # 先订阅（保证订阅之后产生的事件也能收到），再回放历史事件，最后实时输出新事件 queue = benchmark_service.subscribe(run_id) - last_sequence = after_sequence + last_sequence = cursor for event in benchmark_service.get_events(run_id): - if event.sequence <= after_sequence: + if event.sequence <= cursor: continue yield as_sse(event.event.value, event.model_dump_json(), event_id=event.sequence) last_sequence = event.sequence @@ -969,7 +977,11 @@ async def benchmark_events( continue yield as_sse(event.event.value, event.model_dump_json(), event_id=event.sequence) last_sequence = event.sequence - if event.event in (BenchmarkEventType.run_completed, BenchmarkEventType.run_failed): + if event.event in ( + BenchmarkEventType.run_completed, + BenchmarkEventType.run_failed, + BenchmarkEventType.run_cancelled, + ): break finally: benchmark_service.unsubscribe(run_id, queue) diff --git a/backend/tests/test_benchmark.py b/backend/tests/test_benchmark.py index 72b3507..b2fea65 100644 --- a/backend/tests/test_benchmark.py +++ b/backend/tests/test_benchmark.py @@ -17,7 +17,13 @@ from pydantic import ValidationError from app.benchmarks import datasets, metrics as m, service from app.config import get_settings -from app.contracts import BenchmarkKind, RAGRunRequest, SearchMode +from app.contracts import ( + BenchmarkKind, + BenchmarkRun, + BenchmarkStatus, + RAGRunRequest, + SearchMode, +) from app.errors import ApiError @@ -323,3 +329,113 @@ def test_benchmark_run_not_found_raises() -> None: with pytest.raises(ApiError) as exc: asyncio.run(routes.get_benchmark_run("benchmark_missing")) assert exc.value.code == "BENCHMARK_RUN_NOT_FOUND" + + +# --------------------------------------------------------------------------- # +# 审阅回归：索引兼容 / 容量 / 失败样本 / 取消事件 / 数据集隔离 +# --------------------------------------------------------------------------- # +def test_create_rag_run_requires_built_index() -> None: + # 空索引（无已索引 block）会让所有模式得到全 0 指标，应在创建时拒绝而非跑出误导结果 + _write_dataset("empty-index-v1", [{"case_id": "x", "query": "q", "expected_note_ids": ["n"]}]) + with pytest.raises(ApiError) as exc: + asyncio.run( + service.create_rag_run( + RAGRunRequest(dataset_id="empty-index-v1", modes=[SearchMode.fts]) + ) + ) + assert exc.value.status_code == 409 + assert exc.value.code == "BENCHMARK_INDEX_INCOMPATIBLE" + + +def test_capacity_exceeded_when_all_runs_active(monkeypatch) -> None: + # 满容量且全为活动（非终态）run 时，无法淘汰，应拒绝创建而非删掉正在运行的 run + _, _, case = _single_note_case() + _write_dataset("capacity-v1", [case]) + + monkeypatch.setattr(service, "MAX_RUNS", 1) + fake_id = "benchmark_fake_active" + service._runs[fake_id] = BenchmarkRun( + run_id=fake_id, + kind=BenchmarkKind.rag, + dataset_id="capacity-v1", + dataset_hash="sha256:fake", + status=BenchmarkStatus.queued, + created_at=service._now(), + ) + try: + with pytest.raises(ApiError) as exc: + asyncio.run( + service.create_rag_run( + RAGRunRequest(dataset_id="capacity-v1", modes=[SearchMode.fts]) + ) + ) + assert exc.value.status_code == 429 + assert exc.value.code == "BENCHMARK_CAPACITY_EXCEEDED" + finally: + service._runs.pop(fake_id, None) + + +def test_failed_samples_counted_as_zero_in_aggregate() -> None: + from app.benchmarks import rag as rag_module + from app.contracts import RAGCaseResult + + cases = [ + RAGCaseResult( + case_id="ok", mode=SearchMode.fts, repeat=0, latency_ms=10.0, + hit_at_1=True, recall=1.0, reciprocal_rank=1.0, + citation_hit=True, citation_applicable=True, + ), + RAGCaseResult( + case_id="boom", mode=SearchMode.fts, repeat=0, latency_ms=0.0, + error="RAG case evaluation failed.", + error_code="BENCHMARK_CASE_EVALUATION_FAILED", + ), + ] + metrics = rag_module._aggregate(cases, SearchMode.fts) + + assert metrics.total_cases == 2 + assert metrics.successful_cases == 1 + assert metrics.failed_cases == 1 + assert metrics.failure_rate == 0.5 + # 失败样本按零分计入质量指标分母，汇总不虚高 + assert metrics.hit_at_1 == 0.5 + assert metrics.recall_at_k == 0.5 + # 延迟只统计成功样本 + assert metrics.p50_latency_ms == 10.0 + + +def test_cancel_emits_run_cancelled_event() -> None: + _, _, case = _single_note_case() + _write_dataset("cancel-event-v1", [case]) + + async def _scenario(): + run = await service.create_rag_run( + RAGRunRequest(dataset_id="cancel-event-v1", modes=[SearchMode.fts]) + ) + service.cancel_run(run.run_id) + return await service.wait_for_run(run.run_id) + + run = asyncio.run(_scenario()) + assert run.status.value == "cancelled" + events = service.get_events(run.run_id) + assert events[-1].event.value == "RunCancelled" + + +def test_load_dataset_ignores_corrupted_unrelated_files() -> None: + # 无关文件损坏（非法 JSON / 顶层非对象）不应阻断目标数据集加载 + directory = get_settings().benchmark_datasets_path + directory.mkdir(parents=True, exist_ok=True) + (directory / "broken.json").write_text("{ not valid json", encoding="utf-8") + (directory / "array.json").write_text('["a", "b"]', encoding="utf-8") + _write_dataset("ok-v1", [{"case_id": "x", "query": "q", "expected_note_ids": ["n"]}]) + + dataset = datasets.load_dataset("ok-v1", BenchmarkKind.rag) + assert dataset.dataset_id == "ok-v1" + assert len(dataset.cases) == 1 + + +def test_load_dataset_top_level_must_be_object() -> None: + _write_raw("array-top", ["a", "b"]) + with pytest.raises(ApiError) as exc: + datasets.load_dataset("array-top", BenchmarkKind.rag) + assert exc.value.code == "BENCHMARK_DATASET_INVALID" diff --git a/backend/tests/test_retrieval.py b/backend/tests/test_retrieval.py index 1199381..e9d4e62 100644 --- a/backend/tests/test_retrieval.py +++ b/backend/tests/test_retrieval.py @@ -436,6 +436,44 @@ def test_fts_pagination_is_not_truncated_at_one_thousand(vault) -> None: assert len(response.items) == 10 +def test_fts_score_threshold_filters_before_total(vault) -> None: + """score_threshold 先于计数与分页生效：total 反映过滤后数量，与 items 一致。 + + 高阈值过滤掉全部结果时 total==0 且 items 为空，杜绝「空页但 total>0」的 + 不一致（审阅 P2-7）。 + """ + from app.retrieval.engine import engine + from app.services import note_service + + # 10 个 block，含「目标」次数递增，bm25 分数各异，min-max 归一化后分数落在 [0,1] + markdown = "\n\n".join(f"{'目标' * i} 分隔内容" for i in range(1, 11)) + asyncio.run( + note_service.create_note(title="阈值过滤", markdown=markdown, folder="", tags=[]) + ) + + all_hits = asyncio.run( + engine.search( + SearchRequest(query="目标", mode=SearchMode.fts, limit=20, score_threshold=0.0) + ) + ) + filtered = asyncio.run( + engine.search( + SearchRequest(query="目标", mode=SearchMode.fts, limit=20, score_threshold=0.5) + ) + ) + none = asyncio.run( + engine.search( + SearchRequest(query="目标", mode=SearchMode.fts, limit=20, score_threshold=2.0) + ) + ) + + assert all_hits.page.total >= 10 + assert 0 < filtered.page.total < all_hits.page.total # 阈值过滤掉部分而非全部 + assert filtered.page.total == len(filtered.items) + assert none.page.total == 0 + assert none.items == [] + + # --------------------------------------------------------------------------- # # 审阅回归：PATCH tags 语义 / 向量-块一致性 / 过滤漏召回 / rebuild 语义与回滚 # --------------------------------------------------------------------------- # diff --git a/docs/contracts/第二阶段接口契约-开发版.md b/docs/contracts/第二阶段接口契约-开发版.md index 68e0d9b..a2d71e4 100644 --- a/docs/contracts/第二阶段接口契约-开发版.md +++ b/docs/contracts/第二阶段接口契约-开发版.md @@ -55,9 +55,9 @@ | Provider | 现有路径 | `/api/providers/*`、`POST /api/chat` | 扩展 | 补齐协议能力和统一行为 | | Retrieval | GET/POST | `/api/index/status`、`/api/index/rebuild` | 扩展 | 暴露 Embedding 兼容状态并安全重建向量 | | Benchmark | GET | `/api/benchmarks/datasets` | 计划新增 | 枚举受控 Dataset | -| Benchmark | POST | `/api/benchmarks/rag/runs` | 计划新增 | 创建 RAG Benchmark | -| Benchmark | POST | `/api/benchmarks/agent/runs` | 计划新增 | 创建 Agent Benchmark | -| Benchmark | GET | `/api/benchmarks/runs` | 计划新增 | 分页获取 Benchmark Run | +| Benchmark | POST | `/api/benchmarks/rag/runs` | 已实现 | 创建 RAG Benchmark | +| Benchmark | POST | `/api/benchmarks/agent/runs` | 暂缓 | 创建 Agent Benchmark（依赖 Agent Runtime 完成后交付） | +| Benchmark | GET | `/api/benchmarks/runs` | 已实现 | 分页获取 Benchmark Run | | Benchmark | GET/POST | `/api/benchmarks/runs/{run_id}/*` | 计划新增 | 查询、订阅、取消和读取报告 | | Export | POST | `/api/exports` | 计划新增 | 创建 HTML/PDF/DOCX 导出任务 | | Export | GET | `/api/exports` | 计划新增 | 分页获取导出任务 | @@ -817,7 +817,7 @@ Dataset 从仓库或受控导入目录注册。API 不接受调用方提交任 配置快照必须记录 Embedding model ID/version/dimension、Reranker、索引版本、Dataset Hash 和运行环境。 -### 9.5 创建 Agent Benchmark +### 9.5 创建 Agent Benchmark（暂缓，未暴露接口） `POST /api/benchmarks/agent/runs` @@ -851,12 +851,16 @@ RAG 和 Agent 创建接口均返回 `202 BenchmarkRun`： "metrics": null, "config_snapshot": {}, "error": null, + "error_code": null, "created_at": "2026-08-31T10:30:00Z", "started_at": null, "completed_at": null } ``` +`status` 取值：`queued` → `running` → `completed` | `failed` | `cancelled`。失败/取消时 `error` 与 +`error_code` 只返回项目错误码与安全消息，不暴露第三方堆栈。 + 公共接口： | 方法 | 路径 | 用途 | @@ -867,6 +871,10 @@ RAG 和 Agent 创建接口均返回 `202 BenchmarkRun`： | POST | `/api/benchmarks/runs/{run_id}/cancel` | 取消运行 | | GET | `/api/benchmarks/runs/{run_id}/report` | 获取结构化完整报告 | +SSE 事件流（`RunStarted` → `CaseCompleted`* → `RunCompleted` | `RunFailed` | `RunCancelled`）： +`GET /api/benchmarks/runs/{run_id}/events` 支持 `Last-Event-ID` 与 `?after_sequence=` 游标恢复， +`RunCompleted` / `RunFailed` / `RunCancelled` 为终止事件，收到后即断流。 + ### 9.7 指标 Contract RAG： @@ -879,10 +887,17 @@ RAG： "mrr": 0.81, "citation_hit_rate": 0.89, "p50_latency_ms": 24.5, - "p95_latency_ms": 67.3 + "p95_latency_ms": 67.3, + "total_cases": 50, + "successful_cases": 48, + "failed_cases": 2, + "failure_rate": 0.04 } ``` +失败样本按零分计入质量指标分母，`total_cases` / `successful_cases` / `failed_cases` / +`failure_rate` 让报告明确实际分母；延迟仅统计成功样本。 + Agent： ```json @@ -906,8 +921,10 @@ BENCHMARK_DATASET_NOT_FOUND BENCHMARK_DATASET_INVALID BENCHMARK_CONFIG_INVALID BENCHMARK_INDEX_INCOMPATIBLE +BENCHMARK_CAPACITY_EXCEEDED BENCHMARK_RUN_NOT_FOUND BENCHMARK_RUN_FAILED +BENCHMARK_CASE_EVALUATION_FAILED ``` ### 9.9 Retrieval Profile 与索引兼容 diff --git a/docs/development/Knowledge与Retrieval-Core开发说明.md b/docs/development/Knowledge与Retrieval-Core开发说明.md index 8aee584..f6299c6 100644 --- a/docs/development/Knowledge与Retrieval-Core开发说明.md +++ b/docs/development/Knowledge与Retrieval-Core开发说明.md @@ -198,7 +198,7 @@ cd backend uv run pytest -q ``` -当前后端完整测试共 71 个用例通过（单元 + 端到端）。测试通过 `tests/conftest.py` 的 autouse fixture 把 +当前后端完整测试共 120 个用例通过（单元 + 端到端）。测试通过 `tests/conftest.py` 的 autouse fixture 把 数据目录/DB/Vault 重定向到临时目录，不读写真实 `backend/data`，任何本机状态下结果确定。 ## 配置 @@ -232,4 +232,6 @@ rag.search - Embedding / Reranker 为轻量实现，后续替换为真实模型（接口不变）。 - 小语料下 hybrid 检索召回偏宽（向量 Top-K 覆盖全部 block），可加相关性阈值收紧。 - 重建为同步 + 全量，后续接入增量索引与异步任务队列。 -- 检索 Benchmark 待建立。 +- RAG Benchmark 已建立：`POST /api/benchmarks/rag/runs` 创建即返回 queued、后台 Task 执行， + 通过 SSE 实时推送进度，报告含逐 Case 结果与 `total_cases` / `successful_cases` / `failed_cases` / `failure_rate`。 +- Agent Benchmark 暂缓，待 Agent Runtime 完成后交付。