fix(evaluation): 收敛评估取消与查询失败语义

backend/package/yuxi/knowledge/eval/evaluator.py

@@ -14,6 +14,13 @@ def normalize_query_result(query_result: Any) -> tuple[str, list[dict[str, Any]]
     return "", []
 
 
+def split_raise_on_error_config(retrieval_config: dict[str, Any]) -> tuple[dict[str, Any], bool]:
+    """拆出查询层错误开关，避免它作为普通检索参数泄漏给底层实现。"""
+    config = dict(retrieval_config or {})
+    raise_on_error = bool(config.pop("raise_on_error", False))
+    return config, raise_on_error
+
+
 def build_answer_prompt(query: str, retrieved_chunks: list[dict[str, Any]], max_docs: int = 5) -> str:
     context_docs = []
     for idx, chunk in enumerate(retrieved_chunks[:max_docs]):
@@ -67,7 +74,8 @@ async def evaluate_question(
     select_model_fn: Callable[..., Any],
 ) -> dict[str, Any]:
     query = question_data["query"]
-    query_result = await kb_manager.aquery(query, kb_id, **retrieval_config)
+    retrieval_options, raise_on_error = split_raise_on_error_config(retrieval_config)
+    query_result = await kb_manager.aquery(query, kb_id, raise_on_error=raise_on_error, **retrieval_options)
     generated_answer, retrieved_chunks = normalize_query_result(query_result)
     generated_answer = await generate_answer_if_needed(
         query=query,

backend/package/yuxi/knowledge/eval/service.py

@@ -160,6 +160,7 @@ async def _sync_evaluation_run(self, row, task: Any = _TASK_NOT_LOADED):
         completed_at = None
         if task is not None and task.status in {"failed", "cancelled"}:
             error = task.error or task.message or "评估任务失败"
+            status = "cancelled" if task.status == "cancelled" else "failed"
             completed_at = task.completed_at or utc_now()
             if completed_at.tzinfo is None:
                 completed_at = completed_at.replace(tzinfo=UTC)
@@ -168,15 +169,16 @@ async def _sync_evaluation_run(self, row, task: Any = _TASK_NOT_LOADED):
             and row.started_at
             and coerce_any_to_utc_datetime(row.started_at).replace(tzinfo=None) < utc_now_naive() - timedelta(minutes=1)
         ):
+            status = "failed"
             error = "评估任务提交中断"
             completed_at = utc_now()
 
         if error is not None:
             await self.eval_repo.update_run(
                 row.run_id,
-                {"status": "failed", "metrics": {"error": error}, "completed_at": completed_at},
+                {"status": status, "metrics": {"error": error}, "completed_at": completed_at},
             )
-            row.status = "failed"
+            row.status = status
             row.metrics = {"error": error}
             row.completed_at = completed_at
         return task
@@ -333,6 +335,10 @@ async def delete_dataset(self, dataset_id: str) -> None:
             row = await self.eval_repo.get_dataset(dataset_id)
             if row is None:
                 raise ValueError("Dataset not found")
+            # 先按 Durable Task 终态收敛元数据，避免任务已取消或失败后仍被陈旧 running 状态误拦。
+            await self._sync_dataset_build_metadata(row)
+            if (row.build_metadata or {}).get("status") in {"pending", "running"}:
+                raise ValueError("评估数据集正在生成，请先取消生成任务")
             await self.eval_repo.delete_dataset(dataset_id)
             logger.info(f"成功删除评估数据集: {dataset_id}")
         except Exception as e:
@@ -663,6 +669,8 @@ async def run_evaluation(
 
             if model_config:
                 retrieval_config.update(model_config)
+            # 评估必须区分真实零命中和检索故障；底层接口失败时让整条 Run 进入 failed。
+            retrieval_config["raise_on_error"] = True
 
             task_payload = {
                 "run_id": run_id,
@@ -672,6 +680,7 @@ async def run_evaluation(
                 "retrieval_config": retrieval_config,
                 "created_by": created_by,
             }
+
             async with pg_manager.get_async_session_context() as session:
                 task = await tasker.create_in_session(
                     session,
@@ -704,6 +713,32 @@ async def run_evaluation(
             logger.error(f"启动评估失败: {e}")
             raise
 
+    async def cancel_run(self, kb_id: str, run_id: str) -> dict[str, Any]:
+        """取消一条运行中的评估，并把终态语义交给 Durable Task 失败钩子。"""
+        if not re.match(r"^run_[a-f0-9]{8}$", run_id):
+            raise ValueError("Invalid run_id format")
+        run = await self.eval_repo.get_run(run_id)
+        if run is None or run.kb_id != kb_id:
+            raise ValueError(f"Run not found for {run_id}")
+        if run.status != "running":
+            raise ValueError("评估任务已结束")
+        task = await self.task_repo.find_latest_by_payload(
+            task_type="rag_evaluation",
+            payload_match={"run_id": run_id},
+            statuses={"pending", "running"},
+        )
+        if task is None:
+            raise ValueError("评估任务不存在")
+        cancelled = await tasker.cancel_task(task.id)
+        if cancelled is None:
+            raise ValueError("评估任务已结束")
+        return {
+            "run_id": run_id,
+            "task_id": task.id,
+            "status": cancelled.status,
+            "cancel_requested": cancelled.cancel_requested,
+        }
+
     async def _run_evaluation_task(self, context: TaskContext):
         try:
             payload = context.payload
@@ -927,6 +962,8 @@ async def delete_run(self, kb_id: str, run_id: str) -> None:
         row = await self.eval_repo.get_run(run_id)
         if row is None or row.kb_id != kb_id:
             raise ValueError("Run not found")
+        if row.status == "running":
+            raise ValueError("评估任务仍在运行，请先取消")
         await self.eval_repo.delete_run(run_id)
         logger.info(f"成功删除评估运行: {run_id}")
 
@@ -995,6 +1032,7 @@ async def fail_rag_evaluation_task(session, task_record, error: str) -> None:
     run = await session.scalar(select(EvaluationRun).where(EvaluationRun.run_id == run_id).with_for_update())
     if run is None or run.status == "completed":
         return
-    run.status = "failed"
+    run.status = "cancelled" if task_record.cancel_requested else "failed"
+    run.completed_items = int(run.completed_items or 0)
     run.metrics = {"error": error}
     run.completed_at = utc_now()

backend/package/yuxi/knowledge/implementations/dify.py

@@ -70,6 +70,7 @@ async def aquery(
         kb_id: str,
         *,
         config: KnowledgeBaseConfig,
+        raise_on_error: bool = False,
         agent_call: bool = False,
         **kwargs,
     ) -> list[dict]:
@@ -81,6 +82,8 @@ async def aquery(
 
         if not api_url or not token or not dataset_id:
             logger.error(f"Dify config incomplete for kb_id={kb_id}")
+            if raise_on_error:
+                raise ValueError(f"Dify 配置缺失，无法查询 kb_id={kb_id}")
             return []
 
         merged = {**config.query_options, **kwargs}
@@ -130,6 +133,8 @@ async def aquery(
                 logger.error(
                     f"Dify query fallback failed for kb_id={kb_id}: {fallback_error}, {traceback.format_exc()}"
                 )
+                if raise_on_error:
+                    raise e
                 return []
 
         records = response_json.get("records", []) if isinstance(response_json, dict) else []

backend/package/yuxi/knowledge/implementations/milvus.py

@@ -874,10 +874,12 @@ async def aquery(
         kb_id: str,
         *,
         config: KnowledgeBaseConfig,
+        raise_on_error: bool = False,
         agent_call: bool = False,
         **kwargs,
     ) -> list[dict]:
         """异步查询知识库"""
+        del raise_on_error
         embedding_model_spec = config.embedding_model_spec
         collection = await self._get_or_create_milvus_collection(kb_id, embedding_model_spec)
         if not collection:

backend/package/yuxi/knowledge/implementations/notion.py

@@ -234,6 +234,7 @@ async def aquery(
         kb_id: str,
         *,
         config: KnowledgeBaseConfig,
+        raise_on_error: bool = False,
         agent_call: bool = False,
         **kwargs,
     ) -> list[dict]:
@@ -279,6 +280,8 @@ async def aquery(
             return sorted(results, key=lambda item: item.get("score", 0.0), reverse=True)[:final_top_k]
         except (NotionAPIError, httpx.HTTPError, ValueError) as exc:
             logger.error(f"Notion query failed for kb_id={kb_id}: {exc}, {traceback.format_exc()}")
+            if raise_on_error:
+                raise
             return []
 
     async def _search_candidate_pages(

backend/package/yuxi/knowledge/manager.py

@@ -641,14 +641,22 @@ async def update_file_params(self, kb_id: str, file_id: str, params: dict, opera
             additional_params=config.additional_params,
         )
 
-    async def aquery(self, query_text: str, kb_id: str, **kwargs) -> str:
-        """异步查询知识库"""
+    async def aquery(
+        self,
+        query_text: str,
+        kb_id: str,
+        *,
+        raise_on_error: bool = False,
+        **kwargs,
+    ) -> str:
+        """异步查询知识库；`raise_on_error` 供评估等调用方关闭执行器层静默降级。"""
         config = await self.get_kb_config(kb_id)
         executor = await self._get_or_create_kb_instance(config.kb_type)
         return await executor.aquery(
             query_text,
             kb_id,
             config=config,
+            raise_on_error=raise_on_error,
             **kwargs,
         )
 

backend/server/routers/knowledge_eval_router.py

@@ -336,6 +336,28 @@ async def get_evaluation_run_results(
         raise HTTPException(status_code=500, detail=f"获取评估运行结果失败: {str(e)}")
 
 
+@evaluation.post("/databases/{kb_id}/runs/{run_id}/cancel")
+async def cancel_evaluation_run(
+    kb_id: str,
+    run_id: str,
+    current_user: User = Depends(require_knowledge_base_manage),
+):
+    """取消运行中的 RAG 评估"""
+    try:
+        service = EvaluationService()
+        result = await service.cancel_run(kb_id, run_id)
+        return {"message": "success", "data": result}
+    except ValueError as e:
+        if "not found" in str(e).lower():
+            raise HTTPException(status_code=404, detail=str(e))
+        raise HTTPException(status_code=400, detail=str(e))
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.exception(f"取消评估运行失败: {e}")
+        raise HTTPException(status_code=500, detail=f"取消评估运行失败: {str(e)}")
+
+
 @evaluation.delete("/databases/{kb_id}/runs/{run_id}")
 async def delete_evaluation_run(
     kb_id: str,

backend/test/unit/knowledge/eval/test_evaluator.py

@@ -48,10 +48,27 @@ async def test_evaluate_question_uses_runtime_kb_manager(monkeypatch):
         select_model_fn=lambda **_: None,
     )
 
-    aquery.assert_awaited_once_with("问题", "kb-1", final_top_k=3)
+    aquery.assert_awaited_once_with("问题", "kb-1", raise_on_error=False, final_top_k=3)
     assert result["detail"]["retrieved_chunks"] == [{"content": "检索结果"}]
 
 
+async def test_evaluate_question_splits_and_forwards_raise_on_error(monkeypatch):
+    aquery = AsyncMock(return_value=[])
+    monkeypatch.setattr(evaluator, "kb_manager", SimpleNamespace(aquery=aquery))
+
+    await evaluate_question(
+        kb_id="kb-1",
+        question_data={"query": "问题"},
+        retrieval_config={"final_top_k": 3, "raise_on_error": True},
+        has_gold_chunks=False,
+        has_gold_answers=False,
+        judge_llm=None,
+        select_model_fn=lambda **_: None,
+    )
+
+    aquery.assert_awaited_once_with("问题", "kb-1", final_top_k=3, raise_on_error=True)
+
+
 def test_aggregate_metrics_matches_service_output_shape():
     metrics, _ = aggregate_metrics(
         [{"recall@1": 1.0, "f1@1": 0.0}, {"recall@1": 0.0, "f1@1": 1.0}],