From fde46e3b5e31e6ce6643720f4a1d4d77d71f0c47 Mon Sep 17 00:00:00 2001
From: suhan <72542107+suhan42@users.noreply.github.com>
Date: Tue, 13 Oct 2026 16:10:43 +0800
Subject: [PATCH] =?UTF-8?q?fix(evaluation):=20=E6=94=B6=E6=95=9B=E8=AF=84?=
=?UTF-8?q?=E4=BC=B0=E5=8F=96=E6=B6=88=E4=B8=8E=E6=9F=A5=E8=AF=A2=E5=A4=B1?=
=?UTF-8?q?=E8=B4=A5=E8=AF=AD=E4=B9=89?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
- 评估链路强制检索 fail-fast：run_evaluation 写入 raise_on_error=true，
evaluate_question 拆分开关后透传，避免作为普通检索参数泄漏。
- 查询失败语义归位：KnowledgeBaseManager.aquery 只透传 raise_on_error，
不再捕获配置/实例/查询异常；Dify 与 Notion 在执行器层保留默认静默降级和严格上抛两种模式，
Milvus 保持原生 fail-fast。
- 新增评估运行取消接口 POST /databases/{kb_id}/runs/{run_id}/cancel，
使用知识库 manage 权限定位活跃 Durable Task 并复用通用取消；
cancel_requested 投影为 Run 的 cancelled 状态，其余失败保持 failed。
- running Run 删除前拒绝，防止任务继续写入已删除的评估记录。
- Dataset 删除前同步生成任务终态，并拒绝 pending/running 状态，防止活跃生成任务继续写入。
- 前端补齐 cancelled 状态展示与取消按钮，取消后刷新任务中心和评估概览。
- 新增查询层回归测试与评估服务生成/取消/删除保护测试。
---
.../package/yuxi/knowledge/eval/evaluator.py | 10 +-
.../package/yuxi/knowledge/eval/service.py | 44 ++-
.../yuxi/knowledge/implementations/dify.py | 5 +
.../yuxi/knowledge/implementations/milvus.py | 2 +
.../yuxi/knowledge/implementations/notion.py | 3 +
backend/package/yuxi/knowledge/manager.py | 12 +-
.../server/routers/knowledge_eval_router.py | 22 ++
.../unit/knowledge/eval/test_evaluator.py | 19 +-
.../knowledge/eval/test_service_generation.py | 261 ++++++++++++++++++
.../test_knowledge_manager_aquery.py | 145 ++++++++++
...evaluation-run-cancel-and-query-failure.md | 40 +++
web/src/apis/knowledge_api.js | 4 +
.../KnowledgeEvaluationWorkspace.vue | 46 ++-
.../views/EvaluationBenchmarkDetailView.vue | 10 +-
14 files changed, 614 insertions(+), 9 deletions(-)
create mode 100644 backend/test/unit/knowledge/test_knowledge_manager_aquery.py
create mode 100644 docs/develop-guides/decisions/implemented/2026-10-12-evaluation-run-cancel-and-query-failure.md
diff --git a/backend/package/yuxi/knowledge/eval/evaluator.py b/backend/package/yuxi/knowledge/eval/evaluator.py
index c850458d0c..59f295de0f 100644
--- a/backend/package/yuxi/knowledge/eval/evaluator.py
+++ b/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,
diff --git a/backend/package/yuxi/knowledge/eval/service.py b/backend/package/yuxi/knowledge/eval/service.py
index 00f59f605d..f7e65ace13 100644
--- a/backend/package/yuxi/knowledge/eval/service.py
+++ b/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()
diff --git a/backend/package/yuxi/knowledge/implementations/dify.py b/backend/package/yuxi/knowledge/implementations/dify.py
index 4837a6dcf1..2d2fe1502d 100644
--- a/backend/package/yuxi/knowledge/implementations/dify.py
+++ b/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 []
diff --git a/backend/package/yuxi/knowledge/implementations/milvus.py b/backend/package/yuxi/knowledge/implementations/milvus.py
index 86611a00a9..8f60770ac0 100644
--- a/backend/package/yuxi/knowledge/implementations/milvus.py
+++ b/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:
diff --git a/backend/package/yuxi/knowledge/implementations/notion.py b/backend/package/yuxi/knowledge/implementations/notion.py
index 868ed37be9..44425bb4b0 100644
--- a/backend/package/yuxi/knowledge/implementations/notion.py
+++ b/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(
diff --git a/backend/package/yuxi/knowledge/manager.py b/backend/package/yuxi/knowledge/manager.py
index 7950420908..fa3a1b9cac 100644
--- a/backend/package/yuxi/knowledge/manager.py
+++ b/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,
)
diff --git a/backend/server/routers/knowledge_eval_router.py b/backend/server/routers/knowledge_eval_router.py
index 1bbc09d30b..3113f6a10d 100644
--- a/backend/server/routers/knowledge_eval_router.py
+++ b/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,
diff --git a/backend/test/unit/knowledge/eval/test_evaluator.py b/backend/test/unit/knowledge/eval/test_evaluator.py
index 9e3fa85d23..0f9d2d08b8 100644
--- a/backend/test/unit/knowledge/eval/test_evaluator.py
+++ b/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}],
diff --git a/backend/test/unit/knowledge/eval/test_service_generation.py b/backend/test/unit/knowledge/eval/test_service_generation.py
index 5f8bd19659..5ce0b991e2 100644
--- a/backend/test/unit/knowledge/eval/test_service_generation.py
+++ b/backend/test/unit/knowledge/eval/test_service_generation.py
@@ -1,5 +1,6 @@
from contextlib import asynccontextmanager
from types import SimpleNamespace
+from datetime import timedelta
import pytest
@@ -13,6 +14,7 @@ def __init__(self):
self.updated_dataset = None
self.dataset = None
self.created_run = None
+ self.deleted_dataset = False
async def create_dataset_in_session(self, session, payload):
self.created_dataset = payload
@@ -23,6 +25,10 @@ async def update_dataset(self, dataset_id, payload):
async def get_dataset(self, dataset_id):
return self.dataset
+ async def delete_dataset(self, dataset_id):
+ """记录数据集删除副作用。"""
+ self.deleted_dataset = True
+
async def create_run_in_session(self, session, payload):
self.created_run = payload
@@ -217,6 +223,238 @@ async def list_by_payload_values(self, **kwargs):
assert service.eval_repo.updated_run[0] == "run_12345678"
+@pytest.mark.asyncio
+async def test_list_runs_projects_cancelled_durable_task_to_cancelled_run():
+ started_at = eval_service_module.utc_now_naive()
+ run = SimpleNamespace(
+ run_id="run_12345678",
+ name="评估",
+ dataset_id="dataset_1",
+ status="running",
+ started_at=started_at,
+ completed_at=None,
+ total_items=1,
+ completed_items=7,
+ overall_score=None,
+ retrieval_config={},
+ metrics={},
+ )
+
+ class RunRepository(FakeEvaluationRepository):
+ async def list_runs(self, kb_id):
+ return [run]
+
+ async def update_run(self, run_id, data):
+ self.updated_run = (run_id, data)
+
+ class CancelledTaskRepository:
+ async def list_by_payload_values(self, **kwargs):
+ return [
+ SimpleNamespace(
+ payload={"run_id": "run_12345678"},
+ status="cancelled",
+ error="任务已取消",
+ message="任务已取消",
+ completed_at=started_at,
+ )
+ ]
+
+ service = EvaluationService()
+ service.eval_repo = RunRepository()
+ service.task_repo = CancelledTaskRepository()
+
+ result = await service.list_runs("kb_1")
+
+ assert result[0]["status"] == "cancelled"
+ assert result[0]["completed_items"] == 7
+ assert service.eval_repo.updated_run[1]["status"] == "cancelled"
+
+
+@pytest.mark.asyncio
+async def test_list_runs_projects_orphan_running_run_to_failed():
+ started_at = eval_service_module.utc_now_naive() - timedelta(minutes=2)
+ run = SimpleNamespace(
+ run_id="run_12345678",
+ name="评估",
+ dataset_id="dataset_1",
+ status="running",
+ started_at=started_at,
+ completed_at=None,
+ total_items=1,
+ completed_items=7,
+ overall_score=None,
+ retrieval_config={},
+ metrics={},
+ )
+
+ class RunRepository(FakeEvaluationRepository):
+ async def list_runs(self, kb_id):
+ return [run]
+
+ async def update_run(self, run_id, data):
+ self.updated_run = (run_id, data)
+
+ class MissingTaskRepository:
+ async def list_by_payload_values(self, **kwargs):
+ return []
+
+ service = EvaluationService()
+ service.eval_repo = RunRepository()
+ service.task_repo = MissingTaskRepository()
+
+ result = await service.list_runs("kb_1")
+
+ assert result[0]["status"] == "failed"
+ assert result[0]["completed_items"] == 7
+ assert service.eval_repo.updated_run[1]["status"] == "failed"
+ assert service.eval_repo.updated_run[1]["metrics"] == {"error": "评估任务提交中断"}
+
+
+@pytest.mark.asyncio
+async def test_cancel_run_requests_active_durable_task(monkeypatch):
+ run = SimpleNamespace(run_id="run_12345678", kb_id="kb_1", status="running")
+
+ class RunRepository:
+ async def get_run(self, run_id):
+ return run
+
+ class TaskRepository:
+ async def find_latest_by_payload(self, **kwargs):
+ assert kwargs["task_type"] == "rag_evaluation"
+ assert kwargs["payload_match"] == {"run_id": "run_12345678"}
+ assert kwargs["statuses"] == {"pending", "running"}
+ return SimpleNamespace(id="task_1")
+
+ class Tasker:
+ async def cancel_task(self, task_id):
+ assert task_id == "task_1"
+ return SimpleNamespace(status="cancelled", cancel_requested=True)
+
+ service = EvaluationService()
+ service.eval_repo = RunRepository()
+ service.task_repo = TaskRepository()
+ monkeypatch.setattr(eval_service_module, "tasker", Tasker())
+
+ result = await service.cancel_run("kb_1", "run_12345678")
+
+ assert result == {
+ "run_id": "run_12345678",
+ "task_id": "task_1",
+ "status": "cancelled",
+ "cancel_requested": True,
+ }
+
+
+@pytest.mark.asyncio
+async def test_cancel_run_rejects_terminal_run():
+ class RunRepository:
+ async def get_run(self, run_id):
+ return SimpleNamespace(run_id="run_12345678", kb_id="kb_1", status="completed")
+
+ service = EvaluationService()
+ service.eval_repo = RunRepository()
+
+ with pytest.raises(ValueError, match="已结束"):
+ await service.cancel_run("kb_1", "run_12345678")
+
+
+@pytest.mark.asyncio
+async def test_delete_run_rejects_running_run():
+ class RunRepository:
+ def __init__(self):
+ self.deleted = False
+
+ async def get_run(self, run_id):
+ return SimpleNamespace(run_id="run_12345678", kb_id="kb_1", status="running")
+
+ async def delete_run(self, run_id):
+ self.deleted = True
+
+ repo = RunRepository()
+ service = EvaluationService()
+ service.eval_repo = repo
+
+ with pytest.raises(ValueError, match="先取消"):
+ await service.delete_run("kb_1", "run_12345678")
+
+ assert repo.deleted is False
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("status", ["pending", "running"])
+async def test_delete_dataset_rejects_active_generation(status):
+ """等待中和运行中的生成任务都阻止数据集删除。"""
+
+ class ActiveTaskRepository:
+ """提供尚未结束的生成任务。"""
+
+ async def get_by_id(self, task_id):
+ """返回数据集关联的活跃任务。"""
+ assert task_id == "task_1"
+ return SimpleNamespace(status=status, progress=30, message="生成中")
+
+ repo = FakeEvaluationRepository()
+ repo.dataset = SimpleNamespace(
+ dataset_id="dataset_1",
+ build_metadata={"source": "generated", "status": status, "task_id": "task_1"},
+ )
+ service = EvaluationService()
+ service.eval_repo = repo
+ service.task_repo = ActiveTaskRepository()
+
+ with pytest.raises(ValueError, match="先取消生成任务"):
+ await service.delete_dataset("dataset_1")
+
+ assert repo.deleted_dataset is False
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("status", ["completed", "failed", None])
+async def test_delete_dataset_allows_inactive_dataset(status):
+ """终态及没有生成元数据的数据集保持可删除。"""
+ repo = FakeEvaluationRepository()
+ repo.dataset = SimpleNamespace(
+ dataset_id="dataset_1",
+ build_metadata={"source": "generated", "status": status} if status else None,
+ )
+ service = EvaluationService()
+ service.eval_repo = repo
+
+ await service.delete_dataset("dataset_1")
+
+ assert repo.deleted_dataset is True
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("task_status", ["success", "failed", "cancelled"])
+async def test_delete_dataset_projects_terminal_task_before_delete(task_status):
+ """先同步任务终态，避免陈旧 running 元数据误拦删除。"""
+
+ class TerminalTaskRepository:
+ """提供已经结束的生成任务。"""
+
+ async def get_by_id(self, task_id):
+ """返回数据集关联的终态任务。"""
+ assert task_id == "task_1"
+ return SimpleNamespace(status=task_status, error=None, message="任务结束")
+
+ repo = FakeEvaluationRepository()
+ repo.dataset = SimpleNamespace(
+ dataset_id="dataset_1",
+ build_metadata={"source": "generated", "status": "running", "task_id": "task_1"},
+ )
+ service = EvaluationService()
+ service.eval_repo = repo
+ service.task_repo = TerminalTaskRepository()
+
+ await service.delete_dataset("dataset_1")
+
+ expected_status = "completed" if task_status == "success" else "failed"
+ assert repo.dataset.build_metadata["status"] == expected_status
+ assert repo.updated_dataset == ("dataset_1", {"build_metadata": repo.dataset.build_metadata})
+ assert repo.deleted_dataset is True
+
+
def test_build_evaluation_run_name_uses_eval_date_hash_format():
name = build_evaluation_run_name(hash_value="abcdef12")
@@ -252,3 +490,26 @@ async def test_run_evaluation_saves_custom_name(task_submission):
assert repo.created_run["name"] == "回归评估"
assert repo.created_run["retrieval_config"]["top_k"] == 3
assert repo.created_run["retrieval_config"]["answer_llm"] == "test:model"
+
+
+@pytest.mark.asyncio
+async def test_run_evaluation_forces_query_error_fail_fast(task_submission):
+ repo = FakeEvaluationRepository()
+ repo.dataset = SimpleNamespace(
+ dataset_id="dataset_1",
+ kb_id="db_1",
+ name="dataset",
+ item_count=1,
+ build_metadata={"status": "completed"},
+ )
+ service = EvaluationService()
+ service.eval_repo = repo
+ service.kb_repo = FakeKnowledgeBaseRepository()
+
+ await service.run_evaluation(
+ kb_id="db_1",
+ dataset_id="dataset_1",
+ model_config={"raise_on_error": False},
+ )
+
+ assert repo.created_run["retrieval_config"]["raise_on_error"] is True
diff --git a/backend/test/unit/knowledge/test_knowledge_manager_aquery.py b/backend/test/unit/knowledge/test_knowledge_manager_aquery.py
new file mode 100644
index 0000000000..acc9d6724d
--- /dev/null
+++ b/backend/test/unit/knowledge/test_knowledge_manager_aquery.py
@@ -0,0 +1,145 @@
+"""验证知识库查询层与真实执行器的静默降级边界。"""
+
+from types import SimpleNamespace
+
+import pytest
+
+from yuxi.knowledge.implementations.dify import DifyKB
+from yuxi.knowledge.implementations.notion import NotionKB
+from yuxi.knowledge.manager import KnowledgeBaseManager
+
+
+async def test_dify_executor_raises_when_opted_in(monkeypatch, tmp_path):
+ """开启严格模式时，Dify 请求失败必须上抛。"""
+ executor = DifyKB(work_dir=str(tmp_path))
+
+ async def failing_request(**kwargs):
+ """模拟真实请求失败。"""
+ raise RuntimeError("dify unavailable")
+
+ monkeypatch.setattr(executor, "_request_dify", failing_request)
+
+ with pytest.raises(RuntimeError, match="dify unavailable"):
+ await executor.aquery("query", "kb_1", config=_kb_config(), raise_on_error=True)
+
+
+async def test_dify_executor_preserves_default_silent_degradation(monkeypatch, tmp_path):
+ """Dify 默认保留执行器内部的空结果降级。"""
+ executor = DifyKB(work_dir=str(tmp_path))
+
+ async def failing_request(**kwargs):
+ """模拟真实请求失败。"""
+ raise RuntimeError("dify unavailable")
+
+ monkeypatch.setattr(executor, "_request_dify", failing_request)
+
+ assert await executor.aquery("query", "kb_1", config=_kb_config()) == []
+
+
+async def test_notion_executor_raises_when_opted_in(monkeypatch, tmp_path):
+ """有效配置下的 Notion 检索故障在严格模式中上抛。"""
+ executor = NotionKB(work_dir=str(tmp_path))
+
+ async def failing_search(*args, **kwargs):
+ """在通过连接配置校验后模拟检索失败。"""
+ raise ValueError("notion unavailable")
+
+ monkeypatch.setattr(executor, "_search_candidate_pages", failing_search)
+
+ with pytest.raises(ValueError, match="notion unavailable"):
+ await executor.aquery("query", "kb_1", config=_notion_config(), raise_on_error=True)
+
+
+async def test_notion_executor_preserves_default_silent_degradation(monkeypatch, tmp_path):
+ """Notion 默认保留执行器内部的空结果降级。"""
+ executor = NotionKB(work_dir=str(tmp_path))
+
+ async def failing_search(*args, **kwargs):
+ """在通过连接配置校验后模拟检索失败。"""
+ raise ValueError("notion unavailable")
+
+ monkeypatch.setattr(executor, "_search_candidate_pages", failing_search)
+
+ assert await executor.aquery("query", "kb_1", config=_notion_config()) == []
+
+
+@pytest.mark.parametrize("query_options", [{}, {"raise_on_error": True}])
+async def test_aquery_propagates_executor_error(monkeypatch, query_options):
+ """共享入口在默认和严格模式中都保留执行器的异常。"""
+ manager = KnowledgeBaseManager.__new__(KnowledgeBaseManager)
+
+ async def failing_executor(*args, **kwargs):
+ """模拟主检索失败。"""
+ raise RuntimeError("query failed")
+
+ async def get_config(kb_id):
+ """返回有效的知识库类型。"""
+ return SimpleNamespace(kb_type="milvus")
+
+ async def get_instance(kb_type):
+ """返回会抛出主检索异常的执行器。"""
+ return SimpleNamespace(aquery=failing_executor)
+
+ monkeypatch.setattr(manager, "get_kb_config", get_config)
+ monkeypatch.setattr(manager, "_get_or_create_kb_instance", get_instance)
+
+ with pytest.raises(RuntimeError, match="query failed"):
+ await manager.aquery("query", "kb_1", **query_options)
+
+
+async def test_aquery_propagates_config_error_by_default(monkeypatch):
+ """共享入口不把配置读取故障降级成成功空结果。"""
+ manager = KnowledgeBaseManager.__new__(KnowledgeBaseManager)
+
+ async def get_config(kb_id):
+ """模拟配置读取失败。"""
+ raise RuntimeError("config unavailable")
+
+ monkeypatch.setattr(manager, "get_kb_config", get_config)
+
+ with pytest.raises(RuntimeError, match="config unavailable"):
+ await manager.aquery("query", "kb_1")
+
+
+@pytest.mark.parametrize("raise_on_error", [False, True])
+async def test_aquery_forwards_error_contract_to_executor(monkeypatch, raise_on_error):
+ """共享入口透传错误策略和普通查询选项。"""
+ manager = KnowledgeBaseManager.__new__(KnowledgeBaseManager)
+ received_options = {}
+
+ async def executor_query(*args, **kwargs):
+ """记录查询选项并返回空结果。"""
+ received_options.update(kwargs)
+ return []
+
+ async def get_config(kb_id):
+ """返回有效的知识库类型。"""
+ return SimpleNamespace(kb_type="dify")
+
+ async def get_instance(kb_type):
+ """返回记录查询参数的执行器。"""
+ return SimpleNamespace(aquery=executor_query)
+
+ monkeypatch.setattr(manager, "get_kb_config", get_config)
+ monkeypatch.setattr(manager, "_get_or_create_kb_instance", get_instance)
+
+ assert await manager.aquery("query", "kb_1", raise_on_error=raise_on_error, final_top_k=3) == []
+ assert received_options["raise_on_error"] is raise_on_error
+ assert received_options["final_top_k"] == 3
+
+
+def _kb_config():
+ """构造 Dify 执行器测试所需的最小配置。"""
+ return SimpleNamespace(
+ kb_type="dify",
+ query_options={},
+ additional_params={"dify_api_url": "https://dify.example/v1", "dify_token": "token", "dify_dataset_id": "ds"},
+ )
+
+
+def _notion_config():
+ """构造 Notion 执行器测试所需的最小连接配置。"""
+ return SimpleNamespace(
+ query_options={},
+ additional_params={"notion_token": "token", "notion_data_source_id": "data_source"},
+ )
diff --git a/docs/develop-guides/decisions/implemented/2026-10-12-evaluation-run-cancel-and-query-failure.md b/docs/develop-guides/decisions/implemented/2026-10-12-evaluation-run-cancel-and-query-failure.md
new file mode 100644
index 0000000000..0e95925481
--- /dev/null
+++ b/docs/develop-guides/decisions/implemented/2026-10-12-evaluation-run-cancel-and-query-failure.md
@@ -0,0 +1,40 @@
+# 评估运行取消与查询失败语义
+
+状态：implemented
+类型：bug-fix
+Owner：backend/package/yuxi/knowledge/eval/service.py
+
+## 问题
+
+RAG 评估的 Task 取消、超时和异常都由同一个失败钩子收敛成 `failed`，评估页无法区分用户停止和真实错误；取消入口只存在于管理员任务中心，普通知识库管理者没有对应业务操作。检索链路可能把底层查询错误降级成空列表，评估会把故障算成零命中。运行中的 Run 还能被直接删除，任务可能继续消耗资源或写入已删除的关联数据。
+
+## 决策
+
+### 实现方案
+
+评估提交在 `retrieval_config` 中强制写入 `raise_on_error=true`；`KnowledgeBaseManager.aquery` 只把开关作为显式关键字参数传给执行器，不捕获配置读取、执行器创建或底层查询错误。评估链路开启时由真实异常 Owner 上抛：Dify 配置缺失或两次请求均失败、Notion 连接/查询失败会中断评估并进入既有失败钩子，Milvus 主检索原本就上抛。Dify 的兼容重试成功仍返回结果，不把成功重试当故障。`fail_rag_evaluation_task` 和 `list_runs` 的读时投影按 Task 的 `cancel_requested` / `cancelled` 状态把 Run 收敛为 `cancelled`，其余失败仍是 `failed`，取消保留已完成题目数量。评估路由新增 `POST /api/evaluation/databases/{kb_id}/runs/{run_id}/cancel`，使用知识库管理权限定位活跃 Durable Task 后调用通用取消；running Run 的删除被拒绝，前端补取消按钮、取消中和 `cancelled` 展示。
+
+Dataset 删除由 `EvaluationService.delete_dataset` 在 service 层保护：先复用生成元数据同步逻辑，把关联 Durable Task 的终态投影到 Dataset，再拒绝 `build_metadata.status` 为 `pending` 或 `running` 的删除并提示先取消生成任务。排队中的生成任务也受保护，因为它仍可启动并写入数据集；终态同步避免已结束任务的陈旧 running 元数据误拦删除。删除操作不联动取消任务，路由沿用 `ValueError` 到 HTTP 400 的转换。
+
+## 替代方案
+
+把所有任务失败都保持成 `failed` 最小，但会让用户主动停止和系统错误共用一个含义。把检索失败记录成逐项 `retrieval_status` 可以保留更多结果，但会扩大评估表、指标汇总和 UI 契约，当前不需要。删除 running Run 时静默取消任务看起来更顺手，但会把破坏性操作和长耗时任务终止隐式耦合。
+
+## 后果
+
+用户取消保留部分评估进度并显示“已取消”；检索故障会让整条评估失败，不再伪装成低分结果。知识库管理权限可以取消自己的评估，不再必须进入全局任务中心。running Run 必须先取消评估任务并等待终态再删除；pending / running Dataset 必须先取消生成任务并等待终态再删除。共享查询层保留既有的失败传播语义；默认静默降级只属于 Dify 和 Notion 执行器，Milvus 与共享层故障继续由普通查询路由返回 `failed`。
+
+Dataset 删除保护是 service 层的状态检查，检查与删除不在同一加锁事务中，并发恢复生成与删除的互斥未验证。缺少 task_id 且找不到活跃任务的生成元数据沿用既有同步逻辑，保持 pending / running 并拒绝删除。
+
+## 验证
+
+| 验证 | 结果 |
+|---|---|
+| `backend/.venv/Scripts/python.exe -m pytest test/unit/knowledge/test_evaluation_task_cancellation.py test/unit/knowledge/test_knowledge_manager_aquery.py test/unit/knowledge/eval/test_service_generation.py test/unit/knowledge/eval/test_evaluator.py test/unit/services/test_tasker_behavior.py` | 既有验证记录：38 passed，仅覆盖 Run 取消与查询失败变更，不包含新增 Dataset 删除保护用例 |
+| `python -B -m pytest backend/test/unit/knowledge/eval/test_service_generation.py`（本机补充验证 Dataset 删除保护用例） | Not run：系统 Python 3.11 缺少 pytest，无法在本机执行 |
+| `python -B -m pytest backend/test/unit/knowledge/test_knowledge_manager_aquery.py`（执行器 fail-fast 回归） | Not run：本机系统 Python 3.11 缺少 pytest，且没有 backend 虚拟环境、uv 和 Docker；新增用例覆盖 manager 失败传播与透传、Dify 请求失败的开启/关闭语义及 Notion 上抛与默认降级路径 |
+| `python -B scripts/verify_engineering_contracts.py` | √ 工程契约检查通过 |
+| `python -B -m unittest scripts.test_verify_engineering_contracts` | 63 项中 62 项成功、1 项错误；符号链接隔离用例因 Windows WinError 1314 无创建符号链接权限失败，沙盒外重跑结果相同 |
+| Dataset 删除保护真实 HTTP / PostgreSQL / worker 验证、文档构建 | Not run：本机缺少对应运行环境与工具，尚未验证真实删除结果和取消后删除链路 |
+| `git diff --check` | 通过；Windows 仅提示既有 CRLF 转换警告 |
+| `python -B -m py_compile`（manager、Dify、Notion、Milvus 及查询回归测试） | √ 使用临时 Python 缓存目录通过 |
diff --git a/web/src/apis/knowledge_api.js b/web/src/apis/knowledge_api.js
index 08ea9c0584..79c41b8717 100644
--- a/web/src/apis/knowledge_api.js
+++ b/web/src/apis/knowledge_api.js
@@ -646,6 +646,10 @@ export const evaluationApi = {
return apiAdminGet(`/api/evaluation/databases/${kbId}/runs`)
},
+ cancelRun: async (kbId, runId) => {
+ return apiAdminPost(`/api/evaluation/databases/${kbId}/runs/${runId}/cancel`, {})
+ },
+
getRunResults: async (kbId, runId, params = {}) => {
const queryParams = new URLSearchParams()
diff --git a/web/src/components/evaluation/KnowledgeEvaluationWorkspace.vue b/web/src/components/evaluation/KnowledgeEvaluationWorkspace.vue
index 0120107075..0016a59e41 100644
--- a/web/src/components/evaluation/KnowledgeEvaluationWorkspace.vue
+++ b/web/src/components/evaluation/KnowledgeEvaluationWorkspace.vue
@@ -234,6 +234,7 @@
/>
+
@@ -282,6 +283,16 @@
/>
+
const getRunName = (run) => run?.name || run?.run_name || run?.run_id?.slice(0, 8) || '-'
const getRunStatusText = (status) =>
- ({ running: '运行中', completed: '已完成', failed: '失败', paused: '已暂停' })[status] || status
+ ({
+ running: '运行中',
+ completed: '已完成',
+ failed: '失败',
+ cancelled: '已取消',
+ paused: '已暂停'
+ })[status] || status
const getRunProgress = (run) => {
if (Number.isFinite(run?.progress)) return Math.max(0, Math.min(Math.round(run.progress), 100))
@@ -539,6 +558,21 @@ const handleDatasetAction = async (key, dataset) => {
}
}
+const cancelRun = async (run) => {
+ cancellingRunId.value = run.run_id
+ try {
+ await evaluationApi.cancelRun(props.kbId, run.run_id)
+ message.success('已请求取消评估')
+ taskerStore.loadTasks()
+ await loadOverview()
+ } catch (error) {
+ console.error('取消评估记录失败:', error)
+ message.error(error.message || '取消评估记录失败')
+ } finally {
+ cancellingRunId.value = ''
+ }
+}
+
const deleteRun = async (run) => {
try {
await evaluationApi.deleteRun(props.kbId, run.run_id)
@@ -715,6 +749,11 @@ defineExpose({ loadOverview })
background: var(--color-error-50);
color: var(--color-error-700);
}
+
+ &.run-cancelled {
+ background: var(--gray-100);
+ color: var(--gray-600);
+ }
}
.evaluation-row-main {
@@ -805,6 +844,11 @@ defineExpose({ loadOverview })
color: var(--color-error-700);
}
+ &.status-cancelled {
+ background: var(--gray-100);
+ color: var(--gray-600);
+ }
+
&.status-pending,
&.status-paused {
background: var(--color-warning-50);
diff --git a/web/src/views/EvaluationBenchmarkDetailView.vue b/web/src/views/EvaluationBenchmarkDetailView.vue
index 9e063fd19e..282befcc2b 100644
--- a/web/src/views/EvaluationBenchmarkDetailView.vue
+++ b/web/src/views/EvaluationBenchmarkDetailView.vue
@@ -399,7 +399,13 @@ const runOptions = computed(() =>
const getRunName = (run) => run?.name || run?.run_name || '未命名测试'
const getRunStatusText = (status) =>
- ({ running: '运行中', completed: '已完成', failed: '失败', paused: '已暂停' })[status] || '未知状态'
+ ({
+ running: '运行中',
+ completed: '已完成',
+ failed: '失败',
+ cancelled: '已取消',
+ paused: '已暂停'
+ })[status] || '未知状态'
const formatMetric = (value) => (Number.isFinite(value) ? Number(value).toFixed(2) : '-')
const formatScore = (value) => (Number.isFinite(value) ? `${(value * 100).toFixed(1)}%` : '-')
const formatTime = (value) => (value ? new Date(value).toLocaleString('zh-CN') : '-')
@@ -408,6 +414,7 @@ const formatRunItems = (run) => `${run?.completed_items || 0}/${run?.total_items
const formatRunDuration = (run) => {
if (!run) return '-'
if (run.status === 'running') return '进行中'
+ if (run.status === 'cancelled') return '已停止'
if (!run.started_at || !run.completed_at) return '-'
const seconds = Math.max(0, (new Date(run.completed_at) - new Date(run.started_at)) / 1000)
if (seconds < 60) return `${Math.round(seconds)} 秒`
@@ -956,6 +963,7 @@ onUnmounted(() => {
}
&.status-failed,
+ &.status-cancelled,
&.score-error {
background: var(--color-error-50);
color: var(--color-error-700);