From 9379f93633ce6f642ac7c2f7cd15c5e066f9b50d Mon Sep 17 00:00:00 2001 From: wuyongtao Date: Wed, 19 Aug 2026 10:44:56 +0800 Subject: [PATCH 1/6] =?UTF-8?q?feat:=20=E6=8E=A8=E7=90=86=E4=B8=8E?= =?UTF-8?q?=E8=AF=84=E6=B5=8B=E6=94=AF=E6=8C=81=E5=A4=9A=E5=8D=A1=20GPU=20?= =?UTF-8?q?=E9=80=89=E6=8B=A9=EF=BC=8C=E8=AE=AD=E7=BB=83=E4=BB=BB=E5=8A=A1?= =?UTF-8?q?=E5=AE=9E=E6=97=B6=20GPU=20=E7=9B=91=E6=8E=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 GPU 选择归一化 helper，统一前端各形态的选择（gpu_indices/gpus/gpu_id） - 评测与推理支持同一节点内多卡选择，校验所选 GPU 空闲后再派发 - 新增 /fine-tune/{id}/gpu-status 接口，训练日志页展示实时 GPU 指标 - 算力节点推理加载支持 CUDA_VISIBLE_DEVICES 多卡可见，nvidia-smi 进程级监控 - GPU 占用跟踪细化为按卡记录，覆盖评测任务、推理模型与对比任务 Co-Authored-By: Claude --- backend/app/api/v1/endpoints/platform.py | 121 ++++++++++++++++-- backend/app/db/platform_store.py | 68 +++++++++- backend/app/modules/compute_gateway/client.py | 4 + backend/app/modules/compute_gateway/sync.py | 3 +- compute/api/main.py | 49 ++++++- compute/engines/llama_factory/inference.py | 25 ++++ frontend/src/api/modules/fineTune.ts | 11 ++ frontend/src/types/index.ts | 8 +- frontend/src/views/eval/EvalCreateView.vue | 18 ++- .../views/eval/create/EvalTaskSetupStep.vue | 20 ++- .../src/views/eval/create/StartEvalStep.vue | 2 +- .../views/inference/InferenceCreateView.vue | 57 ++++++--- frontend/src/views/system/TrainingLogView.vue | 22 ++-- 13 files changed, 360 insertions(+), 48 deletions(-) diff --git a/backend/app/api/v1/endpoints/platform.py b/backend/app/api/v1/endpoints/platform.py index eb3ab1d..7b9ac29 100644 --- a/backend/app/api/v1/endpoints/platform.py +++ b/backend/app/api/v1/endpoints/platform.py @@ -57,6 +57,33 @@ def _select_first_online_node(store: Any) -> dict[str, Any] | None: return None +def _normalize_gpu_indices(payload: dict[str, Any], *, allow_primary: bool = True) -> list[int]: + """Normalize all frontend GPU selection shapes to sorted integer indexes.""" + raw = payload.get("gpu_indices") + if raw is None: + raw = payload.get("gpus") + if raw is None and allow_primary and payload.get("gpu_id") is not None: + raw = [payload.get("gpu_id")] + if raw is None or raw == "": + return [] + if isinstance(raw, str): + raw = [item.strip() for item in raw.split(",") if item.strip()] + if not isinstance(raw, (list, tuple, set)): + raw = [raw] + result: set[int] = set() + for item in raw: + if isinstance(item, str) and ":" in item: + item = item.rsplit(":", 1)[-1] + try: + index = int(item) + except (TypeError, ValueError) as exc: + raise ValueError(f"invalid GPU index: {item}") from exc + if index < 0: + raise ValueError("GPU index must be non-negative") + result.add(index) + return sorted(result) + + async def _wait_for_object_storage() -> None: """Wait for MinIO before starting a resource task.""" settings = get_settings() @@ -1527,10 +1554,10 @@ async def start_fine_tune( if node_id and gpu_indices: if not store.check_gpu_access(current_user["id"], node_id, gpu_indices): raise fail(403, "无权使用所选 GPU，请联系管理员分配") - # 记录创建者 if node_id and not gpu_indices: payload["allowed_gpu_indices"] = store.assigned_gpu_indexes(current_user["id"], node_id) - payload["strict_node_selection"] = bool(node_id) + # 页面明确选择节点时，调度器必须保持节点约束；否则可能落到其它节点。 + payload["strict_node_selection"] = bool(payload.get("compute_node_id") or payload.get("node_id")) payload.setdefault("created_by", current_user.get("id")) try: return ok(await _submit_fine_tune_task(store, payload)) @@ -1661,6 +1688,42 @@ async def fine_tune_diagnostics(task_id: str) -> dict[str, Any]: ) +@router.get("/fine-tune/{task_id}/gpu-status") +async def fine_tune_gpu_status(task_id: str, current_user: dict[str, Any] = Depends(get_current_user)) -> dict[str, Any]: + """Return live GPU metrics for the task's selected node and cards.""" + store = get_platform_store() + try: + task = store.task(task_id) + except KeyError: + raise fail(404, "fine tune task not found") + if not has_resource_access("fine-tune", task_id, current_user, "read"): + raise fail(403, "no permission to access this task") + node = _node_for_task(task) + selected = set(_normalize_gpu_indices({"gpus": task.get("gpus") or []}, allow_primary=False)) + if not node: + return ok({"source": "unavailable", "items": [], "selected_gpus": sorted(selected)}) + try: + if get_settings().compute_mode == "simulator": + live_items = store.gpus() + else: + live_items = await ComputeNodeClient(node["api_base_url"]).gpu_resources() + items = [] + for item in live_items: + index = int(item.get("gpu_index", item.get("id", -1))) + if selected and index not in selected: + continue + items.append({ + **item, + "id": index, + "node_id": node["id"], + "node_code": node.get("code"), + "node_name": node.get("name"), + }) + return ok({"source": "compute", "items": items, "selected_gpus": sorted(selected)}) + except Exception as exc: # noqa: BLE001 - let UI retain last good snapshot + return ok({"source": "unavailable", "items": [], "selected_gpus": sorted(selected), "error": str(exc)}) + + @router.put("/fine-tune/{task_id}") async def update_fine_tune(task_id: str, payload: dict[str, Any] = Body(...), current_user: dict = Depends(get_current_user)) -> dict[str, Any]: if not has_resource_access("fine-tune", task_id, current_user, "write"): @@ -1808,6 +1871,12 @@ async def model_eval_detail(task_id: str, current_user: dict = Depends(get_curre async def model_eval_start(payload: dict[str, Any] = Body(...), current_user: dict = Depends(get_current_user)) -> dict[str, Any]: """Start an evaluation task: submit eval job to compute node.""" store = get_platform_store() + try: + gpu_indices = _normalize_gpu_indices(payload) + except ValueError as exc: + raise fail(400, str(exc)) + if not gpu_indices: + raise fail(400, "请选择至少一张 GPU") # 1. Create eval task record payload.setdefault("created_by", current_user.get("id")) task = store.create_eval_task({**payload, "status": "pending"}) @@ -1910,6 +1979,17 @@ async def model_eval_start(payload: dict[str, Any] = Body(...), current_user: di store.update_eval_task(task["id"], {"status": "failed", "error": message}) return ok({"task_id": task["id"], "status": "failed", "error": message}) + node_gpus = { + int(item.get("id", item.get("gpu_index", -1))): item + for item in store.gpus() + if item.get("node_id") == node["id"] + } + unavailable = [index for index in gpu_indices if node_gpus.get(index, {}).get("status") != "idle"] + if unavailable: + message = f"selected GPU is not idle on compute node {node.get('code')}: {unavailable}" + store.update_eval_task(task["id"], {"status": "failed", "error": message}) + return ok({"task_id": task["id"], "status": "failed", "error": message}) + # 6. Build eval job payload output_dir = f"/data/yg-ft/outputs/{task['id']}" job_payload = { @@ -1923,7 +2003,9 @@ async def model_eval_start(payload: dict[str, Any] = Body(...), current_user: di "output_dir": output_dir, "basic_metrics": payload.get("basic_metrics", {}), "dimension": dimension_cfg, - "gpus": [int(payload.get("gpu_id", 0))], + "gpu_id": gpu_indices[0], + "gpu_indices": gpu_indices, + "gpus": gpu_indices, "temperature": payload.get("temperature", 0.1), "max_new_tokens": payload.get("max_new_tokens", 512), "compute_node_id": node["id"], @@ -2222,6 +2304,15 @@ async def model_compare_load(task_id: str, current_user: dict = Depends(get_curr "model_name_or_path": model_path, "template": item.get("template", "qwen"), } + try: + item_gpu_indices = _normalize_gpu_indices(item) + except ValueError as exc: + loaded_models.append({**item, "status": "error", "error": str(exc)}) + continue + if not item_gpu_indices: + loaded_models.append({**item, "status": "error", "error": "no GPU selected"}) + continue + load_payload["gpu_indices"] = item_gpu_indices if item.get("adapter_path"): load_payload["adapter_name_or_path"] = item["adapter_path"] if get_settings().compute_mode == "simulator": @@ -2230,14 +2321,28 @@ async def model_compare_load(task_id: str, current_user: dict = Depends(get_curr # 只派发：HTTP 响应成功即视为已接受（节点会异步加载），loaded 字段忽略 item_dispatched = False errors = [] - for node in _candidate_online_nodes(store, preferred_node_id): + candidate_nodes = _candidate_online_nodes(store, preferred_node_id) + if preferred_node_id: + candidate_nodes = candidate_nodes[:1] + for node in candidate_nodes: try: + node_gpu_map = { + int(gpu.get("id", gpu.get("gpu_index", -1))): gpu + for gpu in store.gpus() + if gpu.get("node_id") == node["id"] + } + unavailable = [ + index for index in item_gpu_indices + if node_gpu_map.get(index, {}).get("status") != "idle" + ] + if unavailable: + raise RuntimeError(f"selected GPU is not idle on compute node {node.get('code')}: {unavailable}") if get_settings().minio_enabled: await _wait_for_object_storage() client = ComputeNodeClient(node["api_base_url"]) await client.inference_load(load_payload) - store.mark_inference_loaded(node["id"]) - loaded_models.append({**item, "status": "starting", "node_id": node["id"], "node_name": node.get("name")}) + store.mark_inference_loaded(node["id"], item_gpu_indices) + loaded_models.append({**item, "gpu_indices": item_gpu_indices, "gpus": item_gpu_indices, "status": "starting", "node_id": node["id"], "node_name": node.get("name")}) item_dispatched = True break except Exception as exc: # noqa: BLE001 - try next candidate node @@ -2340,7 +2445,7 @@ async def model_chat_local_preload(payload: dict[str, Any] = Body(...)) -> dict[ # 计算节点现在异步加载：HTTP 接受（loading/ready）即视为派发成功 result = await client.inference_load(payload) if result.get("loaded") or result.get("status") in {"loading", "ready"}: - store.mark_inference_loaded(node["id"]) + store.mark_inference_loaded(node["id"], _normalize_gpu_indices(payload)) return ok(result) except Exception as exc: return ok({"loaded": False, "error": str(exc)}) @@ -2420,7 +2525,7 @@ async def model_chat_trained_preload(payload: dict[str, Any] = Body(...), curren # 计算节点现在异步加载：HTTP 接受（loading/ready）即视为派发成功 result = await client.inference_load({**payload, "compute_node_id": node["id"]}) if result.get("loaded") or result.get("status") in {"loading", "ready"}: - store.mark_inference_loaded(node["id"]) + store.mark_inference_loaded(node["id"], _normalize_gpu_indices(payload)) return ok(result) except Exception as exc: return ok({"loaded": False, "error": str(exc)}) diff --git a/backend/app/db/platform_store.py b/backend/app/db/platform_store.py index 2a2f5dd..cea1469 100644 --- a/backend/app/db/platform_store.py +++ b/backend/app/db/platform_store.py @@ -454,15 +454,21 @@ class PlatformStore: self.ensure_seed_data() # Track which compute nodes have an active inference model loaded self._inference_nodes: set[str] = set() + self._inference_gpu_indexes: dict[str, set[int]] = {} self._last_runtime_refresh = 0.0 # ── inference node tracking ──────────────────────────────────── - def mark_inference_loaded(self, node_id: str) -> None: + def mark_inference_loaded(self, node_id: str, gpu_indexes: list[int] | None = None) -> None: self._inference_nodes.add(node_id) + if gpu_indexes is not None: + self._inference_gpu_indexes[node_id] = {int(item) for item in gpu_indexes} + else: + self._inference_gpu_indexes.pop(node_id, None) def mark_inference_unloaded(self, node_id: str) -> None: self._inference_nodes.discard(node_id) + self._inference_gpu_indexes.pop(node_id, None) def is_inference_loaded(self, node_id: str) -> bool: return node_id in self._inference_nodes @@ -2766,7 +2772,35 @@ class PlatformStore: "SELECT gpu_index FROM gpu_allocations WHERE node_id=? AND status IN ('allocated','running')", (node_id,), ).fetchall() - return {int(row["gpu_index"]) for row in rows} + active = {int(row["gpu_index"]) for row in rows} + # Evaluation jobs use the same Compute ProcessManager GPU lock but do + # not have fine-tune allocation rows; derive their selected cards here + # so a training task cannot race onto an evaluation GPU. + for row in conn.execute( + "SELECT payload FROM eval_tasks WHERE status IN ('syncing','queued','running')" + ).fetchall(): + payload = json_loads(row["payload"], {}) + if payload.get("compute_node_id") != node_id: + continue + selected = payload.get("gpu_indices") or payload.get("gpus") + if selected is None and payload.get("gpu_id") is not None: + selected = [payload.get("gpu_id")] + active.update(int(item) for item in selected or []) + # Loaded inference models also reserve only their selected cards. + active.update(self._inference_gpu_indexes.get(node_id, set())) + for row in conn.execute("SELECT payload FROM compare_tasks").fetchall(): + payload = json_loads(row["payload"], {}) + load_status = payload.get("load_status") or {} + if isinstance(load_status, str): + load_status = json_loads(load_status, {}) + for item in load_status.get("loaded_models") or []: + if item.get("node_id") != node_id or item.get("status") not in {"starting", "ready", "running"}: + continue + selected = item.get("gpu_indices") or item.get("gpus") + if selected is None and item.get("gpu_id") is not None: + selected = [item.get("gpu_id")] + active.update(int(gpu) for gpu in selected or []) + return active def _node_gpu_indexes(self, conn: PgConnection, node: dict[str, Any]) -> set[int]: rows = conn.execute("SELECT gpu_index FROM gpus WHERE node_id=?", (node["id"],)).fetchall() @@ -3195,6 +3229,9 @@ class PlatformStore: # 推理模型占用算力节点同样计入：优先从 compare_tasks 持久化状态派生 # （重启后仍准确），并用内存标记兜底（直接 preload 的模型无 compare 记录） inference_node_ids = set(self._inference_nodes) + inference_gpu_indexes: dict[str, set[int]] = { + node_id: set(indexes) for node_id, indexes in self._inference_gpu_indexes.items() + } for ctr in conn.execute("SELECT payload FROM compare_tasks").fetchall(): ls = json_loads(ctr["payload"], {}).get("load_status") or {} if isinstance(ls, str): @@ -3204,8 +3241,13 @@ class PlatformStore: ls = {} for m in ls.get("loaded_models") or []: if m.get("status") in {"ready", "running"} and m.get("node_id"): - inference_node_ids.add(m["node_id"]) - for nid in inference_node_ids: + node_id = m["node_id"] + selected = m.get("gpu_indices") or m.get("gpus") + if selected: + inference_gpu_indexes.setdefault(node_id, set()).update(int(item) for item in selected) + else: + inference_node_ids.add(node_id) + for nid in set(inference_node_ids) | set(inference_gpu_indexes): running_map[nid] = running_map.get(nid, 0) + 1 rows = conn.execute("SELECT * FROM compute_nodes ORDER BY scheduler_weight DESC, code").fetchall() return [ @@ -3434,6 +3476,9 @@ class PlatformStore: # 推理模型占用的节点：优先从 compare_tasks 持久化状态派生（重启后仍准确）， # 内存标记兜底（直接 preload 的模型无 compare 记录） inference_node_ids = set(self._inference_nodes) + inference_gpu_indexes: dict[str, set[int]] = { + node_id: set(indexes) for node_id, indexes in self._inference_gpu_indexes.items() + } for ctr in conn.execute("SELECT payload FROM compare_tasks").fetchall(): ls = json_loads(ctr["payload"], {}).get("load_status") or {} if isinstance(ls, str): @@ -3443,7 +3488,12 @@ class PlatformStore: ls = {} for m in ls.get("loaded_models") or []: if m.get("status") in {"ready", "running"} and m.get("node_id"): - inference_node_ids.add(m["node_id"]) + node_id = m["node_id"] + selected = m.get("gpu_indices") or m.get("gpus") + if selected: + inference_gpu_indexes.setdefault(node_id, set()).update(int(item) for item in selected) + else: + inference_node_ids.add(node_id) items = [] for row in rows: task = next( @@ -3459,7 +3509,10 @@ class PlatformStore: t for t in eval_running if t.get("compute_node_id") == row["node_id"] - and row["gpu_index"] == (int(t["gpu_id"]) if t.get("gpu_id") is not None else -1) + and row["gpu_index"] in { + int(item) + for item in (t.get("gpu_indices") or t.get("gpus") or ([t["gpu_id"]] if t.get("gpu_id") is not None else [])) + } ), None, ) @@ -3468,7 +3521,8 @@ class PlatformStore: eval_task is not None and eval_task.get("status") in {"syncing", "queued"} ) # Also mark GPU as busy if an inference model is loaded on this node - if row["node_id"] in inference_node_ids and not busy: + inference_on_gpu = row["node_id"] in inference_node_ids or row["gpu_index"] in inference_gpu_indexes.get(row["node_id"], set()) + if inference_on_gpu and not busy: busy = True reserved = False memory_used = round(row["memory_total_gb"] * (0.72 if busy else 0.18 if reserved else 0.04), 1) diff --git a/backend/app/modules/compute_gateway/client.py b/backend/app/modules/compute_gateway/client.py index 280d617..4287d5a 100644 --- a/backend/app/modules/compute_gateway/client.py +++ b/backend/app/modules/compute_gateway/client.py @@ -240,6 +240,10 @@ class ComputeNodeClient: async def inference_status(self) -> dict[str, Any]: return await self._request("GET", "/inference/status", timeout=INFERENCE_STATUS_TIMEOUT) + async def gpu_resources(self) -> list[dict[str, Any]]: + """Read live per-GPU metrics from this compute node.""" + return await self.gpus() + async def inference_unload(self) -> dict[str, Any]: return await self._request("POST", "/inference/unload", json_data={}, timeout=INFERENCE_UNLOAD_TIMEOUT) diff --git a/backend/app/modules/compute_gateway/sync.py b/backend/app/modules/compute_gateway/sync.py index 1cac998..e57b775 100644 --- a/backend/app/modules/compute_gateway/sync.py +++ b/backend/app/modules/compute_gateway/sync.py @@ -73,7 +73,8 @@ async def reconcile_inference_loads(store: Any) -> list[dict[str, Any]]: if node_status == "ready": item["status"] = "ready" item.pop("error", None) - store.mark_inference_loaded(node["id"]) + selected_gpus = item.get("gpu_indices") or item.get("gpus") + store.mark_inference_loaded(node["id"], selected_gpus) elif node_status == "error": item["status"] = "error" item["error"] = status.get("error") or "model load failed on compute node" diff --git a/compute/api/main.py b/compute/api/main.py index 6d47c60..2f71703 100644 --- a/compute/api/main.py +++ b/compute/api/main.py @@ -221,6 +221,35 @@ def create_app() -> FastAPI: return fallback_gpu_resources() items: list[dict[str, Any]] = [] + processes_by_uuid: dict[str, list[dict[str, Any]]] = {} + try: + process_result = subprocess.run( + [ + "nvidia-smi", + "--query-compute-apps=gpu_uuid,pid,process_name,used_memory", + "--format=csv,noheader,nounits", + ], + check=True, + capture_output=True, + text=True, + timeout=5, + ) + for process_line in process_result.stdout.splitlines(): + process_parts = [part.strip() for part in process_line.split(",")] + if len(process_parts) < 4: + continue + process_uuid, pid, process_name, used_memory = process_parts[:4] + processes_by_uuid.setdefault(process_uuid, []).append( + { + "pid": int(_safe_float(pid)), + "name": process_name, + "memory_used_gb": round(_safe_float(used_memory) / 1024, 2), + } + ) + except Exception: + # Some driver/runtime combinations do not expose compute-apps; + # utilization and memory metrics remain useful without processes. + pass for line in result.stdout.splitlines(): parts = [part.strip() for part in line.split(",")] if len(parts) < 9: @@ -244,7 +273,7 @@ def create_app() -> FastAPI: "temperature": int(_safe_float(temp)), "power_w": round(_safe_float(power), 1), "power_limit_w": round(_safe_float(power_limit), 1), - "processes": [], + "processes": processes_by_uuid.get(uuid, []), } ) return items @@ -747,6 +776,23 @@ def create_app() -> FastAPI: infer_backend: str (default: "huggingface") infer_dtype: str (default: "auto") """ + requested_gpus = payload.get("gpu_indices") + if requested_gpus is None: + requested_gpus = payload.get("gpus") or [] + try: + requested_gpus = sorted({int(item) for item in requested_gpus}) + except (TypeError, ValueError) as exc: + raise HTTPException(status_code=400, detail=f"invalid GPU selection: {exc}") from exc + if any(item < 0 for item in requested_gpus): + raise HTTPException(status_code=400, detail="GPU index must be non-negative") + if requested_gpus: + known_gpus = {int(item.get("gpu_index", item.get("id", -1))) for item in gpu_resources()} + missing = sorted(set(requested_gpus) - known_gpus) + if missing: + raise HTTPException(status_code=409, detail=f"requested GPU not found: {missing}") + conflict = sorted(set(requested_gpus).intersection(process_manager.locked_gpus())) + if conflict: + raise HTTPException(status_code=409, detail=f"GPU already used by another compute job: {conflict}") session = get_inference_session() result = session.load( model_name_or_path=payload.get("model_name_or_path", ""), @@ -754,6 +800,7 @@ def create_app() -> FastAPI: template=payload.get("template", "qwen"), infer_backend=payload.get("infer_backend", "huggingface"), infer_dtype=payload.get("infer_dtype", "auto"), + gpu_indices=requested_gpus, ) return result diff --git a/compute/engines/llama_factory/inference.py b/compute/engines/llama_factory/inference.py index 72cad82..018ec0b 100644 --- a/compute/engines/llama_factory/inference.py +++ b/compute/engines/llama_factory/inference.py @@ -1,5 +1,6 @@ from __future__ import annotations +import os import threading import time import uuid @@ -36,6 +37,7 @@ class InferenceSession: self._generating_args: dict[str, Any] = {} self._model_name: str = "" self._adapter_path: str = "" + self._gpu_indices: list[int] = [] self._loaded_at: float = 0.0 @property @@ -53,6 +55,7 @@ class InferenceSession: "loaded_at": self._loaded_at, "request_id": self._request_id, "error": self._error, + "gpu_indices": list(self._gpu_indices), } def wait_until_loaded(self, timeout: float | None = None) -> dict[str, Any]: @@ -87,8 +90,12 @@ class InferenceSession: template="qwen", infer_backend="huggingface", infer_dtype="auto", + gpu_indices=None, **kwargs, ) -> dict[str, Any]: + requested_gpus = sorted({int(item) for item in (gpu_indices or [])}) + if any(item < 0 for item in requested_gpus): + return {"loaded": False, "status": "error", "error": "GPU index must be non-negative"} with self._state_lock: if self._status == "loading": # A model is already loading — dedupe, reuse the same request id. @@ -97,6 +104,7 @@ class InferenceSession: self._status = "loading" self._error = "" self._request_id = uuid.uuid4().hex[:12] + self._gpu_indices = requested_gpus self._cancel_requested = False self._load_args = { "model_name_or_path": model_name_or_path, @@ -115,13 +123,21 @@ class InferenceSession: def _load_worker(self) -> None: """Build the ChatModel off the state lock so info() never blocks.""" + with self._state_lock: + requested_gpus = list(self._gpu_indices) model = None tokenizer = None generating_args: dict[str, Any] = {} error = "" + previous_visible_devices = os.environ.get("CUDA_VISIBLE_DEVICES") try: + # Set visibility before LLaMA-Factory/PyTorch initializes CUDA. + if requested_gpus: + os.environ["CUDA_VISIBLE_DEVICES"] = ",".join(str(item) for item in requested_gpus) if self._teardown_old: self._release_model() + with self._state_lock: + self._gpu_indices = requested_gpus from llamafactory.chat import ChatModel from llamafactory.hparams import get_infer_args @@ -138,6 +154,12 @@ class InferenceSession: generating_args = dict(generating_args) except Exception as exc: # noqa: BLE001 - surface load failure via status error = str(exc) + finally: + if requested_gpus: + if previous_visible_devices is None: + os.environ.pop("CUDA_VISIBLE_DEVICES", None) + else: + os.environ["CUDA_VISIBLE_DEVICES"] = previous_visible_devices with self._state_lock: if error: self._model = None @@ -152,6 +174,7 @@ class InferenceSession: self._model = None self._tokenizer = None self._status = "idle" + self._gpu_indices = [] return self._model = model self._tokenizer = tokenizer @@ -189,6 +212,7 @@ class InferenceSession: self._adapter_path = "" self._loaded_at = 0.0 self._error = "" + self._gpu_indices = [] def unload(self) -> dict[str, Any]: with self._state_lock: @@ -208,6 +232,7 @@ class InferenceSession: self._adapter_path = "" self._loaded_at = 0.0 self._error = "" + self._gpu_indices = [] return {"unloaded": True, "status": "idle"} def chat(self, messages, temperature=0.95, top_p=0.7, max_new_tokens=1024, do_sample=True, **kwargs) -> dict[str, Any]: diff --git a/frontend/src/api/modules/fineTune.ts b/frontend/src/api/modules/fineTune.ts index 374ed49..aefc53e 100644 --- a/frontend/src/api/modules/fineTune.ts +++ b/frontend/src/api/modules/fineTune.ts @@ -42,12 +42,23 @@ export interface FineTunePreflightResult { sync_results?: Array> } +export interface FineTuneGpuStatus { + source: string + items: Array> + selected_gpus: number[] + error?: string +} + /** 训练任务列表 */ export const getFineTuneList = () => get('/fine-tune') /** 训练任务详情 */ export const getFineTune = (id: string | number) => get(`/fine-tune/${id}`) +/** 获取任务所在 Compute 节点的实时 GPU 指标 */ +export const getFineTuneGpuStatus = (id: string | number) => + get(`/fine-tune/${id}/gpu-status`) + /** 任务名查重 */ export const checkFineTuneName = (name: string) => get<{ exists: boolean }>('/fine-tune/check-name', { name }) diff --git a/frontend/src/types/index.ts b/frontend/src/types/index.ts index 4c93efa..722b186 100644 --- a/frontend/src/types/index.ts +++ b/frontend/src/types/index.ts @@ -218,6 +218,8 @@ export interface LoadedModel { port?: number node_id?: string node_name?: string + gpu_indices?: number[] + gpus?: number[] error?: string } @@ -239,6 +241,8 @@ export interface CompareModelRef { gpu_id: number node_id?: string node_name?: string + gpu_indices?: number[] + gpus?: number[] source?: string port?: number } @@ -282,7 +286,9 @@ export interface StartEvalPayload { eval_task_name: string eval_type: EvalType model_id: string | number - gpu_id: string | number + gpu_id: string | number | string[] + gpu_indices?: number[] + gpus?: number[] compute_node_id?: string dataset_id: string | number dimension_id: string | number diff --git a/frontend/src/views/eval/EvalCreateView.vue b/frontend/src/views/eval/EvalCreateView.vue index 1e82048..4973197 100644 --- a/frontend/src/views/eval/EvalCreateView.vue +++ b/frontend/src/views/eval/EvalCreateView.vue @@ -39,7 +39,7 @@ const createdDimensionId = ref('') const taskForm = ref({ eval_task_name: '', model_id: '', - gpu_id: '', + gpu_id: [], data_source: 'dataset', dataset_id: '', leaderboard: false, @@ -145,12 +145,24 @@ async function handleSubmit() { const dimensionId = await resolveDimensionId() // GPU 选择为「节点:GPU序号」复合值，解析出节点与 GPU 序号， // 多算力节点时必须把节点信息传给后端，否则会派发到错误的算力节点 - const [gpuNodeId, gpuIndex] = String(taskForm.value.gpu_id).split(':') + const selectedGpuKeys = Array.isArray(taskForm.value.gpu_id) + ? taskForm.value.gpu_id + : [String(taskForm.value.gpu_id)] + const gpuSelections = selectedGpuKeys + .map((key) => { + const [nodeId, gpuIndex] = String(key).split(':') + return { nodeId, gpuIndex: Number(gpuIndex) } + }) + .filter((item) => item.nodeId && Number.isInteger(item.gpuIndex) && item.gpuIndex >= 0) + const gpuNodeId = gpuSelections[0]?.nodeId || '' + const gpuIndices = gpuSelections.map((item) => item.gpuIndex) const evalResult: any = await startEval({ eval_task_name: taskForm.value.eval_task_name, eval_type: 'custom', model_id: taskForm.value.model_id, - gpu_id: Number(gpuIndex) || 0, + gpu_id: gpuIndices[0] ?? 0, + gpu_indices: gpuIndices, + gpus: gpuIndices, compute_node_id: gpuNodeId || '', dataset_id: taskForm.value.data_source === 'dataset' ? taskForm.value.dataset_id : '', dimension_id: dimensionId, diff --git a/frontend/src/views/eval/create/EvalTaskSetupStep.vue b/frontend/src/views/eval/create/EvalTaskSetupStep.vue index 8b595c5..cf49fb7 100644 --- a/frontend/src/views/eval/create/EvalTaskSetupStep.vue +++ b/frontend/src/views/eval/create/EvalTaskSetupStep.vue @@ -6,7 +6,7 @@ import type { DatasetItem, GpuInfo, TrainedModel } from '@/types' export interface EvalTaskSetupDraft { eval_task_name: string model_id: string | number - gpu_id: string | number + gpu_id: string | number | string[] data_source: 'dataset' | 'inference' dataset_id: string | number leaderboard: boolean @@ -23,6 +23,13 @@ defineProps<{ const form = defineModel({ required: true }) const formRef = ref() +function handleGpuChange(value: string | number | string[]) { + const keys = Array.isArray(value) ? value.map(String) : [String(value || '')] + const nodeId = keys[0]?.split(':', 1)[0] + if (!nodeId || !Array.isArray(form.value.gpu_id)) return + form.value.gpu_id = keys.filter((key) => key.split(':', 1)[0] === nodeId) +} + const rules: FormRules = { eval_task_name: [ { required: true, message: '请输入任务名称', trigger: 'blur' }, @@ -101,7 +108,16 @@ defineExpose({ validate }) - + (items: T[], id /** GPU 选择为「节点:GPU序号」复合值，解析并展示为可读标签 */ const gpuLabel = computed(() => { - const key = String(props.task.gpu_id || '') + const key = Array.isArray(props.task.gpu_id) ? String(props.task.gpu_id[0] || '') : String(props.task.gpu_id || '') const gpu = props.gpus.find((g) => `${g.node_id || ''}:${g.id ?? 0}` === key) if (gpu) return `${gpu.node_name || gpu.node_code || '算力节点'} / GPU ${gpu.id ?? 0}` const [nodeId, idx] = key.split(':') diff --git a/frontend/src/views/inference/InferenceCreateView.vue b/frontend/src/views/inference/InferenceCreateView.vue index 4224f17..c1b174e 100644 --- a/frontend/src/views/inference/InferenceCreateView.vue +++ b/frontend/src/views/inference/InferenceCreateView.vue @@ -4,8 +4,7 @@ import { useRouter } from 'vue-router' import { ElMessage, type FormInstance, type FormRules } from 'element-plus' import PageCard from '@/components/PageCard.vue' import { getModelList, getTrainedModels } from '@/api/modules/model' -import { getSystemInfo } from '@/api/modules/system' -import { getComputeNodes, type ComputeNode } from '@/api/modules/compute' +import { getComputeGpus, getComputeNodes, type ComputeNode } from '@/api/modules/compute' import { createCompare, loadCompare } from '@/api/modules/compare' import type { ModelItem, TrainedModel, GpuInfo } from '@/types' @@ -83,8 +82,8 @@ const form = reactive({ description: '', /** 选中的模型 key（单选） */ model_key: '', - /** 使用的 GPU */ - gpu_key: '', + /** 使用的 GPU（同一节点内可多选） */ + gpu_keys: [] as string[], }) const rules: FormRules = { @@ -94,12 +93,26 @@ const rules: FormRules = { /** 当前选中的模型对象 */ const selectedModel = computed(() => modelMap.value[form.model_key]) -const selectedGpu = computed(() => idleGpus.value.find((g) => `${g.node_id || ''}:${g.id ?? 0}` === form.gpu_key)) +const selectedGpus = computed(() => idleGpus.value.filter((gpu) => form.gpu_keys.includes(gpuKey(gpu)))) + +function gpuKey(gpu: GpuInfo) { + return `${gpu.node_id || ''}:${gpu.id ?? 0}` +} + +function handleGpuChange(keys: string[]) { + const nodeId = keys[0]?.split(':', 1)[0] + if (!nodeId) return + const filtered = keys.filter((key) => key.split(':', 1)[0] === nodeId) + if (filtered.length !== keys.length) { + ElMessage.info('一次推理只能使用同一算力节点内的 GPU，已忽略其它节点的选择') + } + form.gpu_keys = filtered +} watch(selectedModel, (model) => { if (!model?.compute_node_id) return const gpu = idleGpus.value.find((item) => item.node_id === model.compute_node_id) - if (gpu) form.gpu_key = `${gpu.node_id || ''}:${gpu.id ?? 0}` + if (gpu) form.gpu_keys = [gpuKey(gpu)] }) async function handleSubmit() { @@ -111,6 +124,10 @@ async function handleSubmit() { ElMessage.warning('请选择模型') return } + if (!selectedGpus.value.length) { + ElMessage.warning('请至少选择一张空闲 GPU') + return + } submitting.value = true startupStatus.value = '正在创建推理任务...' try { @@ -130,9 +147,11 @@ async function handleSubmit() { model_name: m.name, model_path: m.model_path, source: m.source, - gpu_id: selectedGpu.value?.id ?? 0, - node_id: selectedGpu.value?.node_id || m.compute_node_id, - node_name: selectedGpu.value?.node_name || m.compute_node_name, + gpu_id: selectedGpus.value[0]?.id ?? 0, + gpu_indices: selectedGpus.value.map((gpu) => Number(gpu.id ?? 0)), + gpus: selectedGpus.value.map((gpu) => Number(gpu.id ?? 0)), + node_id: selectedGpus.value[0]?.node_id || m.compute_node_id, + node_name: selectedGpus.value[0]?.node_name || m.compute_node_name, }, ], }) @@ -169,17 +188,17 @@ async function loadData() { const [db, trained, sys, nodes] = await Promise.all([ getModelList(), getTrainedModels(), - getSystemInfo(), + getComputeGpus(), getComputeNodes(), ]) dbModels.value = db || [] trainedModels.value = trained?.models || [] - gpus.value = sys?.gpu || [] + gpus.value = (sys || []) as unknown as GpuInfo[] computeNodes.value = nodes || [] // 默认选中第一个空闲 GPU if (idleGpus.value.length > 0) { const firstGpu = idleGpus.value[0] - form.gpu_key = `${firstGpu.node_id || ''}:${firstGpu.id ?? 0}` + form.gpu_keys = [gpuKey(firstGpu)] } } catch { // ignore @@ -228,12 +247,20 @@ onMounted(loadData) - + diff --git a/frontend/src/views/system/TrainingLogView.vue b/frontend/src/views/system/TrainingLogView.vue index 80c946e..35e98f2 100644 --- a/frontend/src/views/system/TrainingLogView.vue +++ b/frontend/src/views/system/TrainingLogView.vue @@ -8,10 +8,9 @@ import TrainingTaskOverview from './training-log/TrainingTaskOverview.vue' import { usePolling } from '@/composables/usePolling' import '@/plugins/echarts-training-log' import { useModelsStore } from '@/stores/models' -import { getFineTune, getFineTuneDiagnostics, getFineTuneLogs, getFineTuneMetrics, type TrainingDiagnostic } from '@/api/modules/fineTune' +import { getFineTune, getFineTuneDiagnostics, getFineTuneGpuStatus, getFineTuneLogs, getFineTuneMetrics, type TrainingDiagnostic } from '@/api/modules/fineTune' import { getTrainingLogFiles, getTrainingLogContent } from '@/api/modules/log' import { getDataset } from '@/api/modules/dataset' -import { getSystemInfo } from '@/api/modules/system' import { TRAIN_TYPE_MAP, TRAIN_METHOD_MAP } from '@/constants' import { buildMetricChartOption, @@ -205,15 +204,20 @@ async function loadDataset(datasetId: string | number) { } } -async function loadGpuStatus() { +async function loadGpuStatus(currentTask: FineTuneTask) { try { - const systemInfo = await getSystemInfo() - gpuPool.value = systemInfo.gpu ?? [] - gpuUpdatedAt.value = new Date() - gpuLoadError.value = '' + const live = await getFineTuneGpuStatus(currentTask.id) + if (live.source === 'compute' && live.items.length) { + gpuPool.value = live.items as unknown as GpuInfo[] + gpuUpdatedAt.value = new Date() + gpuLoadError.value = '' + return + } + gpuPool.value = [] + gpuLoadError.value = live.error || 'Compute 节点暂未返回实时 GPU 指标' } catch { gpuLoadError.value = 'GPU 监控数据暂时不可用' - if (!gpuUpdatedAt.value) gpuPool.value = [] + gpuPool.value = [] } } @@ -306,7 +310,7 @@ async function refreshAll() { const datasetPromise = currentTask.train_dataset_id ? loadDataset(currentTask.train_dataset_id) : Promise.resolve() - await Promise.all([datasetPromise, loadLog(currentTask), loadGpuStatus(), loadDiagnostics(currentTask)]) + await Promise.all([datasetPromise, loadLog(currentTask), loadGpuStatus(currentTask), loadDiagnostics(currentTask)]) await loadMetrics(currentTask) } finally { loading.value = false From 93449b7f0e05c0c8ff1490ede02ee8e928b3a0d0 Mon Sep 17 00:00:00 2001 From: wuyongtao Date: Wed, 19 Aug 2026 11:33:45 +0800 Subject: [PATCH 2/6] =?UTF-8?q?fix(frontend):=20=E4=BF=AE=E5=A4=8D?= =?UTF-8?q?=E6=93=8D=E4=BD=9C=E6=97=A5=E5=BF=97=E4=B8=8E=E6=95=B0=E6=8D=AE?= =?UTF-8?q?=E8=BD=AC=E6=8D=A2=E7=B1=BB=E5=9E=8B=E9=94=99=E8=AF=AF=EF=BC=8C?= =?UTF-8?q?=E4=BF=AE=E6=AD=A3=E6=99=AE=E9=80=9A=E7=94=A8=E6=88=B7=E8=A7=92?= =?UTF-8?q?=E8=89=B2=E5=8F=96=E5=80=BC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - OperationLogView: showDetail 调用补充 OperationLog 类型断言 - DataConvertView: beforeUpload 参数类型改为 UploadRawFile 并直接读取 file.type - UserCreateView: 普通用户角色值 user 修正为 operator Co-Authored-By: Claude --- frontend/src/views/audit/OperationLogView.vue | 2 +- frontend/src/views/data-convert/DataConvertView.vue | 6 +++--- frontend/src/views/system/UserCreateView.vue | 4 ++-- 3 files changed, 6 insertions(+), 6 deletions(-) diff --git a/frontend/src/views/audit/OperationLogView.vue b/frontend/src/views/audit/OperationLogView.vue index 4a0485e..d413718 100644 --- a/frontend/src/views/audit/OperationLogView.vue +++ b/frontend/src/views/audit/OperationLogView.vue @@ -313,7 +313,7 @@ onMounted(() => { diff --git a/frontend/src/views/data-convert/DataConvertView.vue b/frontend/src/views/data-convert/DataConvertView.vue index e57bd0e..2fb28b0 100644 --- a/frontend/src/views/data-convert/DataConvertView.vue +++ b/frontend/src/views/data-convert/DataConvertView.vue @@ -2,7 +2,7 @@ import { onMounted, ref } from 'vue' import { ElMessage, ElMessageBox } from 'element-plus' import { Plus, Delete, Refresh } from '@element-plus/icons-vue' -import type { TagProps, UploadRequestOptions, UploadFile } from 'element-plus' +import type { TagProps, UploadRequestOptions, UploadFile, UploadRawFile } from 'element-plus' import PageCard from '@/components/PageCard.vue' import { getDataConvertTasks, @@ -30,9 +30,9 @@ async function load() { } // 文件上传前的校验（仅校验文件格式） -function beforeUpload(file: UploadFile) { +function beforeUpload(file: UploadRawFile) { // 检查文件类型 - const isJson = file.name.endsWith('.json') || file.raw?.type === 'application/json' + const isJson = file.name.endsWith('.json') || file.type === 'application/json' if (!isJson) { ElMessage.error('只能上传 .json 格式的文件') return false diff --git a/frontend/src/views/system/UserCreateView.vue b/frontend/src/views/system/UserCreateView.vue index c2304d7..b0efe7f 100644 --- a/frontend/src/views/system/UserCreateView.vue +++ b/frontend/src/views/system/UserCreateView.vue @@ -12,7 +12,7 @@ const form = reactive({ username: '', display_name: '', password: 'platform123', - role: 'user', + role: 'operator', status: 'active', permissions: [], }) @@ -45,7 +45,7 @@ async function submit() { - + From f47611020aecab3f0adefb0922c8414f1dbda224 Mon Sep 17 00:00:00 2001 From: caoxiaozhu Date: Wed, 19 Aug 2026 14:21:41 +0800 Subject: [PATCH 3/6] =?UTF-8?q?fix(data=5Fprocess):=20=E4=BF=AE=E5=A4=8D?= =?UTF-8?q?=20Word=20=E5=88=87=E5=88=86=E8=A1=8C=E5=8F=B7=E6=96=AD?= =?UTF-8?q?=E6=A1=A3=E4=B8=8E=E9=A2=84=E8=A7=88=E6=A0=87=E9=A2=98=E7=BC=BA?= =?UTF-8?q?=E5=A4=B1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 正文抽取下钻 SDT 内容控件，目录等内容不再整段丢失 - 切片投影匹配剥离序列化插入的列表自动编号，新增行锚点兜底与游标防回退 - fixed/semantic 切分路径定位失败时保留切片，不再静默丢弃 - Word 预览按段落大纲级别识别标题，未套标题样式的小节正常渲染 - 本地嵌入模型抽为共享单例，供语义分块与质量评分共用 --- .../data_process/algorithms/embedding.py | 24 ++++ .../data_process/algorithms/parsers/office.py | 20 ++- .../modules/data_process/document_chunking.py | 130 ++++++++++++++---- .../modules/data_process/office_preview.py | 23 +++- backend/tests/test_data_process_algorithms.py | 71 ++++++++++ backend/tests/test_document_chunking.py | 81 +++++++++++ 6 files changed, 315 insertions(+), 34 deletions(-) create mode 100644 backend/app/modules/data_process/algorithms/embedding.py diff --git a/backend/app/modules/data_process/algorithms/embedding.py b/backend/app/modules/data_process/algorithms/embedding.py new file mode 100644 index 0000000..eb21d03 --- /dev/null +++ b/backend/app/modules/data_process/algorithms/embedding.py @@ -0,0 +1,24 @@ +"""数据处理算法 - 本地语义嵌入模型共享单例。""" + +from __future__ import annotations + +import os +from functools import lru_cache +from typing import Any + + +@lru_cache(maxsize=1) +def semantic_embedding_model() -> Any: + """加载本地嵌入模型，供语义分块与语义质量评分共用。 + + 模型可在部署环境覆盖；默认模型体积较小且适合中英文语义判断。 + 返回 LlamaIndex BaseEmbedding，通过 ``get_text_embedding`` 使用。 + """ + + from llama_index.embeddings.huggingface import HuggingFaceEmbedding + + return HuggingFaceEmbedding( + model_name=os.getenv("DATA_PROCESS_EMBEDDING_MODEL", "BAAI/bge-small-zh-v1.5"), + device=os.getenv("DATA_PROCESS_EMBEDDING_DEVICE", "cpu"), + trust_remote_code=False, + ) diff --git a/backend/app/modules/data_process/algorithms/parsers/office.py b/backend/app/modules/data_process/algorithms/parsers/office.py index ad3a00b..7cda3de 100644 --- a/backend/app/modules/data_process/algorithms/parsers/office.py +++ b/backend/app/modules/data_process/algorithms/parsers/office.py @@ -7,12 +7,13 @@ import re import unicodedata import zipfile import xml.etree.ElementTree as ET -from collections.abc import Mapping, Sequence +from collections.abc import Iterator, Mapping, Sequence from pathlib import PurePosixPath from typing import Any from urllib.parse import unquote, urlsplit from docx import Document +from docx.oxml.ns import qn from docx.oxml.table import CT_Tbl from docx.oxml.text.paragraph import CT_P from docx.table import Table @@ -121,6 +122,21 @@ def _validate_office_archive(raw: bytes, file_format: TextFormat) -> None: except zipfile.BadZipFile as exc: raise ValueError(f"invalid {file_format.upper()} file: not an Office ZIP package") from exc +def iter_document_blocks(parent: Any) -> Iterator[Any]: + """按文档顺序产出正文段落与表格，并下钻 SDT 内容控件。 + + Word 的目录、复选框等内容控件包在 ``w:sdt`` 元素里，只遍历 body + 直接子级会把这些段落整段丢掉。 + """ + + for child in parent.iterchildren(): + if isinstance(child, (CT_P, CT_Tbl)): + yield child + elif child.tag == qn("w:sdt"): + content = child.find(qn("w:sdtContent")) + if content is not None: + yield from iter_document_blocks(content) + def _extract_docx_text(raw: bytes) -> str: _validate_office_archive(raw, "docx") try: @@ -130,7 +146,7 @@ def _extract_docx_text(raw: bytes) -> str: parts: list[str] = [] total = 0 - for child in document.element.body.iterchildren(): + for child in iter_document_blocks(document.element.body): if isinstance(child, CT_P): total = _append_bounded_text(parts, Paragraph(child, document).text, total) continue diff --git a/backend/app/modules/data_process/document_chunking.py b/backend/app/modules/data_process/document_chunking.py index 830c70a..b36816c 100644 --- a/backend/app/modules/data_process/document_chunking.py +++ b/backend/app/modules/data_process/document_chunking.py @@ -18,12 +18,19 @@ from llama_index.core.base.embeddings.base import BaseEmbedding from llama_index.core.node_parser import SemanticSplitterNodeParser, SentenceSplitter from app.modules.data_process.algorithms import normalize_text +from app.modules.data_process.algorithms.embedding import semantic_embedding_model ChunkMethod = Literal["layout_hybrid", "semantic", "fixed"] _PAGE_FURNITURE = re.compile( r"(?m)^\s*(?:第\s*\d+\s*页\s*共\s*\d+\s*页|[-—–]?\s*\d+\s*[/／]\s*\d+\s*[-—–]?)\s*$" ) +# Docling 的 markdown 序列化会给列表项补上自动编号，而 Word 的编号存放在 +# numbering.xml 中，python-docx 抽取的正文不含这些编号；紧凑匹配前剥掉 +# 行首编号，否则带列表的切片会整体定位失败。 +_LIST_MARKER_PREFIX = re.compile( + r"(?m)^[ \t>]*(?:(?:\d{1,3}[.)])+|\([a-zA-Z0-9]{1,3}\)|[a-zA-Z][.)]|[-*+•·])[ \t]+" +) _COMPACT_CHARACTER = re.compile(r"[\w\u3400-\u4dbf\u4e00-\u9fff]", re.UNICODE) _CONVERTER_LOCK = threading.Lock() @@ -149,18 +156,6 @@ def chunk_fixed_text( return _text_chunks(text, chunk_size=chunk_size, chunk_overlap=chunk_overlap) -@lru_cache(maxsize=1) -def _semantic_embedding_model() -> BaseEmbedding: - # 模型可在部署环境覆盖；默认模型体积较小且适合中英文语义边界判断。 - from llama_index.embeddings.huggingface import HuggingFaceEmbedding - - return HuggingFaceEmbedding( - model_name=os.getenv("DATA_PROCESS_EMBEDDING_MODEL", "BAAI/bge-small-zh-v1.5"), - device=os.getenv("DATA_PROCESS_EMBEDDING_DEVICE", "cpu"), - trust_remote_code=False, - ) - - def chunk_semantic_text( text: str, *, @@ -175,7 +170,7 @@ def chunk_semantic_text( if not normalized: return [] splitter = SemanticSplitterNodeParser.from_defaults( - embed_model=embed_model or _semantic_embedding_model(), + embed_model=embed_model or semantic_embedding_model(), breakpoint_percentile_threshold=breakpoint_percentile_threshold, buffer_size=1, sentence_splitter=_sentence_chunks, @@ -193,6 +188,7 @@ def chunk_semantic_text( if start is None: start = _locate_text(normalized, content, 0) if start is None: + result.append(_unlocated_chunk(content)) continue if len(_tokenizer().encode(content)) <= chunk_size: result.append(_make_text_chunk(normalized, start, start + len(content))) @@ -203,6 +199,7 @@ def chunk_semantic_text( chunk_overlap=chunk_overlap, ): if child.source_start is None or child.source_end is None: + result.append(_unlocated_chunk(child.original_content)) continue result.append( _make_text_chunk( @@ -235,6 +232,7 @@ def _nodes_to_chunks(nodes: list[Any], source_text: str) -> list[DocumentChunk]: if start is None: start = _locate_text(source_text, content, 0) if start is None: + chunks.append(_unlocated_chunk(content)) continue end = start + len(content) chunks.append(_make_text_chunk(source_text, start, end)) @@ -247,6 +245,20 @@ def _locate_text(source: str, content: str, start: int) -> int | None: return position if position >= 0 else None +def _unlocated_chunk(content: str) -> DocumentChunk: + """正文在源文本中定位失败时保底保留切片，只放弃行号信息。""" + + return DocumentChunk( + original_content=content, + contextualized_content=content, + source_start=None, + source_end=None, + source_start_line=None, + source_end_line=None, + token_count=len(_tokenizer().encode(content)), + ) + + def _make_text_chunk(source: str, start: int, end: int) -> DocumentChunk: content = source[start:end] return DocumentChunk( @@ -316,6 +328,14 @@ def _compact_with_offsets(value: str) -> tuple[str, list[int]]: return "".join(compact), offsets +def _expand_to_line_boundaries(source_text: str, start: int, end: int) -> tuple[int, int]: + while start > 0 and source_text[start - 1] not in "\r\n": + start -= 1 + while end < len(source_text) and source_text[end] not in "\r\n": + end += 1 + return start, end + + def _project_layout_span( source_text: str, content: str, @@ -324,21 +344,79 @@ def _project_layout_span( source_offsets: list[int], compact_start: int, ) -> tuple[int | None, int | None, int]: - compact_content, _ = _compact_with_offsets(content) - if len(compact_content) < 4: + for candidate in (content, _LIST_MARKER_PREFIX.sub("", content)): + compact_content, _ = _compact_with_offsets(candidate) + if len(compact_content) < 4: + continue + position = compact_source.find(compact_content, compact_start) + if position < 0: + position = compact_source.find(compact_content) + if position < 0: + continue + start, end = _expand_to_line_boundaries( + source_text, + source_offsets[position], + source_offsets[position + len(compact_content) - 1] + 1, + ) + # 重复内容回退匹配可能命中已消费的更早位置，游标只进不退， + # 避免后续切片跟着错位。 + return start, end, max(compact_start, position + len(compact_content)) + return _project_layout_span_by_anchors( + source_text, + content, + compact_source=compact_source, + source_offsets=source_offsets, + compact_start=compact_start, + ) + + +def _project_layout_span_by_anchors( + source_text: str, + content: str, + *, + compact_source: str, + source_offsets: list[int], + compact_start: int, +) -> tuple[int | None, int | None, int]: + """按行锚点顺序匹配，容忍切片里插入的重复表头等非连续内容。""" + + segments = [ + compact + for compact in ( + _compact_with_offsets(line)[0] + for line in _LIST_MARKER_PREFIX.sub("", content).split("\n") + ) + if len(compact) >= 6 + ] + if not segments: return None, None, compact_start - position = compact_source.find(compact_content, compact_start) - if position < 0: - position = compact_source.find(compact_content) - if position < 0: + total = sum(len(segment) for segment in segments) + + def match_from(cursor: int) -> tuple[list[tuple[int, int]], int]: + matched: list[tuple[int, int]] = [] + position = cursor + for segment in segments: + found = compact_source.find(segment, position) + if found < 0: + continue + matched.append((found, found + len(segment))) + position = found + len(segment) + return matched, sum(end - start for start, end in matched) + + matched, covered = match_from(compact_start) + if covered * 2 < total: + retried, retry_covered = match_from(0) + if retry_covered > covered: + matched, covered = retried, retry_covered + # 覆盖不足一半时宁可不定位，也不能给出错误的行号。 + if not matched or covered * 2 < total: return None, None, compact_start - start = source_offsets[position] - end = source_offsets[position + len(compact_content) - 1] + 1 - while start > 0 and source_text[start - 1] not in "\r\n": - start -= 1 - while end < len(source_text) and source_text[end] not in "\r\n": - end += 1 - return start, end, position + len(compact_content) + start, end = _expand_to_line_boundaries( + source_text, + source_offsets[matched[0][0]], + source_offsets[matched[-1][1] - 1] + 1, + ) + return start, end, max(compact_start, matched[-1][1]) def chunk_layout_document( diff --git a/backend/app/modules/data_process/office_preview.py b/backend/app/modules/data_process/office_preview.py index fcfd0ed..e496b7a 100644 --- a/backend/app/modules/data_process/office_preview.py +++ b/backend/app/modules/data_process/office_preview.py @@ -11,6 +11,7 @@ import re from typing import Any from docx import Document +from docx.oxml.ns import qn from docx.oxml.table import CT_Tbl from docx.oxml.text.paragraph import CT_P from docx.table import Table @@ -27,6 +28,7 @@ from app.modules.data_process.algorithms import ( _xlsx_sheet_merge_ranges, normalize_text, ) +from app.modules.data_process.algorithms.parsers.office import iter_document_blocks MAX_DOCX_PREVIEW_BLOCKS = 2_000 MAX_XLSX_PREVIEW_ROWS = 200 @@ -49,12 +51,21 @@ def _docx_alignment(paragraph: Paragraph) -> str: def _docx_heading_level(paragraph: Paragraph) -> int | None: style = paragraph.style - if style is None: - return None - style_name = str(style.name or "") - style_id = str(style.style_id or "") + style_name = str(style.name or "") if style is not None else "" + style_id = str(style.style_id or "") if style is not None else "" match = re.search(r"(?:heading|标题)\s*([1-6])", f"{style_name} {style_id}", re.IGNORECASE) - return int(match.group(1)) if match else None + if match: + return int(match.group(1)) + # Word 的目录和导航窗格依据大纲级别识别标题；未套标题样式但带 + # outlineLvl 的段落（如手工排版的编号小节）同样是标题。 + outline = paragraph._p.find(f"{qn('w:pPr')}/{qn('w:outlineLvl')}") + if outline is not None: + value = outline.get(qn("w:val")) + if value is not None and value.isdigit(): + level = int(value) + if 0 <= level <= 5: + return level + 1 + return None def build_docx_preview(raw: bytes) -> dict[str, Any]: @@ -84,7 +95,7 @@ def build_docx_preview(raw: bytes) -> dict[str, Any]: has_source_content = True return text, start, source_cursor - for child in document.element.body.iterchildren(): + for child in iter_document_blocks(document.element.body): if rendered_blocks >= MAX_DOCX_PREVIEW_BLOCKS: truncated = True break diff --git a/backend/tests/test_data_process_algorithms.py b/backend/tests/test_data_process_algorithms.py index cda83f5..cff6132 100644 --- a/backend/tests/test_data_process_algorithms.py +++ b/backend/tests/test_data_process_algorithms.py @@ -9,6 +9,8 @@ from decimal import Decimal import pytest from docx import Document +from docx.oxml import parse_xml +from docx.oxml.ns import nsdecls, qn from openpyxl import Workbook from pptx import Presentation from pptx.util import Inches @@ -38,6 +40,7 @@ from app.modules.data_process.algorithms import ( stable_split_assignments, structured_json_dumps, ) +from app.modules.data_process.office_preview import build_docx_preview def _pdf_page_texts(*texts: str) -> tuple[PdfPageText, ...]: @@ -318,6 +321,74 @@ def test_parse_pdf_docx_xlsx_and_pptx() -> None: assert parsed_pptx.records == () +def _docx_with_sdt_bytes() -> bytes: + """构造带 SDT 目录内容控件的 docx，段落顺序为正文、SDT、正文。""" + + document = Document() + document.add_paragraph("正文开头。") + sdt = parse_xml( + "" + "目录条目 第一章 概述" + "" % nsdecls("w") + ) + body = document.element.body + sect_pr = body.find(qn("w:sectPr")) + if sect_pr is not None: + sect_pr.addprevious(sdt) + else: + body.append(sdt) + document.add_paragraph("正文结尾。") + output = io.BytesIO() + document.save(output) + return output.getvalue() + + +def test_docx_extraction_and_preview_include_sdt_content() -> None: + raw = _docx_with_sdt_bytes() + + parsed = parse_text_content(raw, filename="toc.docx") + assert "目录条目 第一章 概述" in parsed.text + assert ( + parsed.text.index("正文开头。") + < parsed.text.index("目录条目 第一章 概述") + < parsed.text.index("正文结尾。") + ) + + preview = build_docx_preview(raw) + paragraph_texts = [ + block["text"] for block in preview["blocks"] if block["type"] == "paragraph" + ] + assert "目录条目 第一章 概述" in paragraph_texts + # 预览偏移必须与正文抽取规则一致，否则前端定位会错位。 + sdt_block = next( + block + for block in preview["blocks"] + if block.get("text") == "目录条目 第一章 概述" + ) + assert parsed.text[sdt_block["source_start"] : sdt_block["source_end"]] == ( + "目录条目 第一章 概述" + ) + + +def test_docx_preview_detects_outline_level_headings() -> None: + """未套标题样式但设了大纲级别的段落（Word 目录按此收录）也按标题渲染。""" + + document = Document() + document.add_heading("一级标题", level=1) + plain = document.add_paragraph("4.2.1 数据管理") + p_pr = plain._p.get_or_add_pPr() + p_pr.append(parse_xml("" % nsdecls("w"))) + document.add_paragraph("普通正文段落。") + output = io.BytesIO() + document.save(output) + + preview = build_docx_preview(output.getvalue()) + blocks = {b["text"]: b for b in preview["blocks"] if b["type"] == "paragraph"} + assert blocks["一级标题"]["heading_level"] == 1 + assert blocks["4.2.1 数据管理"]["heading_level"] == 3 + assert blocks["普通正文段落。"]["heading_level"] is None + + def test_xlsx_record_locators_distinguish_sheets_rows_and_duplicate_records() -> None: workbook = Workbook() first = workbook.active diff --git a/backend/tests/test_document_chunking.py b/backend/tests/test_document_chunking.py index f705104..29afe6d 100644 --- a/backend/tests/test_document_chunking.py +++ b/backend/tests/test_document_chunking.py @@ -9,6 +9,7 @@ from app.modules.data_process.document_chunking import ( DocumentChunk, _compact_with_offsets, _document_converter, + _nodes_to_chunks, _project_layout_span, chunk_fixed_text, chunk_semantic_text, @@ -103,6 +104,86 @@ def test_layout_projection_ignores_layout_whitespace_but_keeps_source_lines() -> assert cursor > 0 +def test_layout_projection_tolerates_list_numbers_inserted_by_serializer() -> None: + # Word 自动编号存放在 numbering.xml，python-docx 抽取的正文没有编号， + # 而 Docling 序列化切片时会补上 "1. " 前缀，投影不能因此失败。 + source = "接入方式说明\n结构化数据接入需要先配置连接地址。\n非结构化接入需要上传文档。" + compact_source, offsets = _compact_with_offsets(source) + start, end, cursor = _project_layout_span( + source, + "1. 结构化数据接入需要先配置连接地址。\n2. 非结构化接入需要上传文档。", + compact_source=compact_source, + source_offsets=offsets, + compact_start=0, + ) + + assert start is not None and end is not None + assert source[start:end] == "结构化数据接入需要先配置连接地址。\n非结构化接入需要上传文档。" + assert cursor > 0 + + +def test_layout_projection_never_moves_cursor_backwards() -> None: + source = "重复段落内容。\n中间正文。\n重复段落内容。" + compact_source, offsets = _compact_with_offsets(source) + # 重复内容回退匹配命中已消费的更早位置时，游标必须保持不退。 + _, _, cursor = _project_layout_span( + source, + "重复段落内容。", + compact_source=compact_source, + source_offsets=offsets, + compact_start=compact_source.index("中间正文"), + ) + assert cursor >= compact_source.index("中间正文") + + +def test_layout_projection_falls_back_to_line_anchors_for_inserted_content() -> None: + # 表格跨切片时 Docling 会在续片中重复表头，正文不再是连续子串； + # 按行锚点匹配仍应定位到表头所在行到末行数据之间的连续区间。 + source = "表头甲\t表头乙\n第一行数据\t说明一\n第二行数据\t说明二" + compact_source, offsets = _compact_with_offsets(source) + start, end, _ = _project_layout_span( + source, + "表头甲 表头乙\n第二行数据 说明二", + compact_source=compact_source, + source_offsets=offsets, + compact_start=0, + ) + + assert start is not None and end is not None + assert source[start:end] == ( + "表头甲\t表头乙\n第一行数据\t说明一\n第二行数据\t说明二" + ) + + +def test_layout_projection_refuses_low_coverage_anchor_match() -> None: + source = "完全无关的正文内容甲。\n完全无关的正文内容乙。" + compact_source, offsets = _compact_with_offsets(source) + start, end, cursor = _project_layout_span( + source, + "找不到的数据行内容\n另一条找不到的数据行内容", + compact_source=compact_source, + source_offsets=offsets, + compact_start=0, + ) + + assert start is None + assert end is None + assert cursor == 0 + + +def test_text_splitter_keeps_chunks_that_cannot_be_located() -> None: + class FakeNode: + def get_content(self) -> str: + return "这段文本在源文本中不存在。" + + chunks = _nodes_to_chunks([FakeNode()], "完全不同的源文本。") + + assert len(chunks) == 1 + assert chunks[0].original_content == "这段文本在源文本中不存在。" + assert chunks[0].source_start is None + assert chunks[0].source_start_line is None + + def test_short_layout_chunk_merges_with_neighbor_and_keeps_page_provenance() -> None: source = "短标题\n这是一段足够长的正文内容，用于测试相邻切片合并。" chunks = [ From 81c2f85c3adca5b087988dc396c660d5e628e5aa Mon Sep 17 00:00:00 2001 From: caoxiaozhu Date: Wed, 19 Aug 2026 14:21:54 +0800 Subject: [PATCH 4/6] =?UTF-8?q?feat(data=5Fprocess):=20=E9=97=AE=E7=AD=94?= =?UTF-8?q?=E5=AF=B9=E6=95=B0=E6=8D=AE=E8=AF=84=E6=B5=8B=E4=BD=93=E7=B3=BB?= =?UTF-8?q?=E4=B8=8E=E8=B4=A8=E9=87=8F=E5=88=86=E9=9B=B7=E8=BE=BE=E5=9B=BE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 三层评测：规则层沿用原五维规则分，语义层用本地 BGE 向量算问答/来源 相关性，评审层复用生成模型按 rubric 打分（忠实度/正确性/清晰度等， 区分 standard/reasoning/dpo 输出类型），任一层失败自动降级 - 组合分 = 规则 35% + 语义 20% + 评审 45%，缺层自动重归一 - 新增 results/evaluate-batch 批量评测接口，镜像批量重生成的并发、 乐观锁与部分成功语义；生成阶段不再展示质量分 - 详情页与结果编辑页新增"数据评测"按钮和批量进度；质量分列悬停弹出 雷达图浮窗（评审 5 维 + 语义 2 维、三层分项、评审理由） - 手动编辑/恢复后重算规则与语义层并丢弃过期评审分，雷达图不再展示 失效数据 --- backend/app/api/v1/endpoints/data_process.py | 244 +++++++++++++- .../data_process/algorithms/quality.py | 85 +++++ .../app/modules/data_process/evaluation.py | 313 ++++++++++++++++++ backend/app/schemas/data_process.py | 20 ++ backend/tests/test_data_process_api.py | 217 ++++++++++++ backend/tests/test_data_process_evaluation.py | 284 ++++++++++++++++ frontend/src/api/modules/dataProcess.ts | 11 + frontend/src/plugins/echarts.ts | 5 +- frontend/src/types/dataProcess.ts | 51 +++ .../data-process/DataProcessCreateView.vue | 16 + .../data-process/DataProcessDetailView.vue | 171 +++++++++- .../create/QualityRadarPopover.vue | 252 ++++++++++++++ .../data-process/create/ResultEditorStep.vue | 121 ++++++- .../src/views/data-process/create/types.ts | 21 +- .../create/useDataProcessEvaluation.ts | 126 +++++++ .../create/useDataProcessGeneration.ts | 13 +- 16 files changed, 1917 insertions(+), 33 deletions(-) create mode 100644 backend/app/modules/data_process/evaluation.py create mode 100644 backend/tests/test_data_process_evaluation.py create mode 100644 frontend/src/views/data-process/create/QualityRadarPopover.vue create mode 100644 frontend/src/views/data-process/create/useDataProcessEvaluation.ts diff --git a/backend/app/api/v1/endpoints/data_process.py b/backend/app/api/v1/endpoints/data_process.py index b706779..2a23cde 100644 --- a/backend/app/api/v1/endpoints/data_process.py +++ b/backend/app/api/v1/endpoints/data_process.py @@ -60,6 +60,10 @@ from app.modules.data_process.document_chunking import ( chunk_semantic_text, merge_short_chunks, ) +from app.modules.data_process.evaluation import ( + evaluate_result_record, + reevaluate_edited_record, +) from app.modules.data_process.generation import generate_model_records from app.modules.data_process.office_preview import ( MAX_XLSX_PREVIEW_ROWS, @@ -96,6 +100,7 @@ from app.schemas.data_process import ( PreviewItemUpdate, ProcessType, PublishRequest, + ResultBatchEvaluateRequest, ResultBatchRegenerateRequest, ResultRegenerateRequest, ResultUpdate, @@ -2039,12 +2044,13 @@ def update_result( or 20 ), ) - quality = score_quality( + # 编辑后内容已变化：重算规则与语义层，旧的评审分不再可信直接丢弃。 + update["quality_score"] = reevaluate_edited_record( merged, - min_output_length=minimum, source_content=source_content, + previous_quality=current.get("quality_score"), + min_output_length=minimum, ) - update["quality_score"] = asdict(quality) result = store.update_result( task_id, result_id, @@ -2089,11 +2095,6 @@ def restore_result( or 20 ), ) - quality = score_quality( - restored, - min_output_length=minimum, - source_content=source_content, - ) restored = store.update_result( task_id, result_id, @@ -2103,7 +2104,12 @@ def restore_result( "output": restored["output"], "chosen": restored["chosen"], "rejected": restored["rejected"], - "quality_score": asdict(quality), + "quality_score": reevaluate_edited_record( + restored, + source_content=source_content, + previous_quality=current.get("quality_score"), + min_output_length=minimum, + ), "expected_updated_at": current.get("updated_at"), }, ) @@ -2266,6 +2272,46 @@ def _safe_regeneration_error(exc: Exception) -> str: return re.sub(r"\s+", " ", str(exc)).strip()[:500] or "result regeneration failed" +def _evaluate_result_in_place( + task_id: str, + current: dict[str, Any], + source_content: str, + config: dict[str, Any], + evaluation_model: dict[str, Any] | None, + store: DataProcessStore, + *, + expected_updated_at: str, + model_client: httpx.Client | None = None, + minimum: int = 20, +) -> dict[str, Any]: + """评测单条结果并落库；复用逐结果互斥锁避免与重生成并发写冲突。""" + + result_id = str(current["id"]) + with _claim_result_regeneration(task_id, result_id): + quality = evaluate_result_record( + { + "instruction": current.get("instruction"), + "input": current.get("input"), + "output": current.get("output"), + "chosen": current.get("chosen"), + "rejected": current.get("rejected"), + }, + source_content=source_content, + model=evaluation_model, + config=config, + client=model_client, + min_output_length=minimum, + ) + return store.update_result( + task_id, + result_id, + { + "quality_score": quality, + "expected_updated_at": expected_updated_at, + }, + ) + + @router.post("/{task_id}/results/regenerate-batch") def regenerate_results_batch( task_id: str, @@ -2439,6 +2485,186 @@ def regenerate_results_batch( ) +@router.post("/{task_id}/results/evaluate-batch") +def evaluate_results_batch( + task_id: str, + payload: ResultBatchEvaluateRequest, + store: DataProcessStore = Depends(get_data_process_store), +) -> dict[str, Any]: + """对一批结果执行三层质量评测（规则+语义+评审），允许部分成功。""" + + started_at = time.perf_counter() + batch_id = new_id("dpeb") + with api_errors(): + task = store.get_task(task_id) + if task.get("status") == "running": + raise ConflictError("data process task is running") + if task.get("output_dataset_id"): + raise InvalidStateError("published results cannot be evaluated") + config = task.get("config") or {} + evaluation_model: dict[str, Any] | None = None + model_id = _value(config, "generation_model_id", "generationModelId", None) + if model_id: + try: + evaluation_model = store.get_generation_model(str(model_id)) + except NotFoundError: + logger.warning( + "data process evaluation model unavailable, judge layer " + "skipped task_id=%s model_id=%s", + task_id, + model_id, + ) + evaluation_config = { + **config, + "output_type": str( + _value(config, "output_type", "outputType", "standard") + ).strip().lower(), + } + minimum = max( + 1, + int(_value(config, "min_output_length", "minOutputLength", 20) or 20), + ) + + prepared: list[tuple[int, dict[str, Any], str, str]] = [] + failures: list[tuple[int, dict[str, str]]] = [] + for index, requested in enumerate(payload.items): + try: + current = store.get_result(task_id, requested.result_id) + if requested.expected_updated_at != str(current.get("updated_at") or ""): + raise ConflictError("data process result was modified by another request") + source_content = "" + preview_id = current.get("preview_item_id") + if preview_id: + preview = store.get_preview_item(task_id, str(preview_id)) + source_content = str( + preview.get("edited_content") + or preview.get("original_content") + or "" + ) + prepared.append( + (index, current, source_content, requested.expected_updated_at) + ) + except ConflictError as exc: + failures.append((index, { + "result_id": requested.result_id, + "code": "conflict", + "message": _safe_regeneration_error(exc), + })) + except (NotFoundError, InvalidStateError) as exc: + failures.append((index, { + "result_id": requested.result_id, + "code": "skipped", + "message": _safe_regeneration_error(exc), + })) + + logger.info( + "data process result batch evaluation started batch_id=%s task_id=%s " + "requested=%s prepared=%s judge_enabled=%s", + batch_id, + task_id, + len(payload.items), + len(prepared), + evaluation_model is not None, + ) + successes: list[tuple[int, dict[str, Any]]] = [] + if prepared: + try: + from app.modules.data_process.algorithms.embedding import ( + semantic_embedding_model, + ) + + semantic_embedding_model() + except Exception: + logger.warning( + "data process semantic embedding unavailable, semantic layer " + "will be skipped batch_id=%s", + batch_id, + ) + request_timeout = _result_regeneration_timeout(config) + model_timeout = httpx.Timeout( + request_timeout, + connect=min(10.0, request_timeout), + ) + model_limits = httpx.Limits( + max_connections=RESULT_REGENERATION_CONCURRENCY, + max_keepalive_connections=RESULT_REGENERATION_CONCURRENCY, + ) + with httpx.Client(timeout=model_timeout, limits=model_limits) as model_client, \ + ThreadPoolExecutor( + max_workers=min(RESULT_REGENERATION_CONCURRENCY, len(prepared)), + thread_name_prefix="data-result-evaluation", + ) as executor: + futures = { + executor.submit( + _evaluate_result_in_place, + task_id, + current, + source_content, + evaluation_config, + evaluation_model, + store, + expected_updated_at=expected_updated_at, + model_client=model_client if evaluation_model else None, + minimum=minimum, + ): (index, str(current["id"]), time.perf_counter()) + for index, current, source_content, expected_updated_at in prepared + } + for future in as_completed(futures): + index, result_id, item_started_at = futures[future] + try: + evaluated = future.result() + successes.append((index, evaluated)) + outcome = "succeeded" + except ConflictError as exc: + outcome = "conflict" + failures.append((index, { + "result_id": result_id, + "code": outcome, + "message": _safe_regeneration_error(exc), + })) + except Exception as exc: + outcome = "evaluation_failed" + failures.append((index, { + "result_id": result_id, + "code": outcome, + "message": _safe_regeneration_error(exc), + })) + logger.info( + "data process result batch evaluation item finished " + "batch_id=%s task_id=%s result_id=%s outcome=%s duration_ms=%.2f", + batch_id, + task_id, + result_id, + outcome, + (time.perf_counter() - item_started_at) * 1000, + ) + + success_items = [item for _, item in sorted(successes, key=lambda pair: pair[0])] + failure_items = [item for _, item in sorted(failures, key=lambda pair: pair[0])] + duration_ms = (time.perf_counter() - started_at) * 1000 + logger.info( + "data process result batch evaluation completed batch_id=%s task_id=%s " + "succeeded=%s failed=%s duration_ms=%.2f", + batch_id, + task_id, + len(success_items), + len(failure_items), + duration_ms, + ) + return ok( + { + "batch_id": batch_id, + "total": len(payload.items), + "succeeded": len(success_items), + "failed": len(failure_items), + "duration_ms": round(duration_ms, 2), + "items": success_items, + "failures": failure_items, + }, + "data process results evaluated", + ) + + @router.post("/{task_id}/results/{result_id}/regenerate") def regenerate_result( task_id: str, diff --git a/backend/app/modules/data_process/algorithms/quality.py b/backend/app/modules/data_process/algorithms/quality.py index 78b2fca..308c244 100644 --- a/backend/app/modules/data_process/algorithms/quality.py +++ b/backend/app/modules/data_process/algorithms/quality.py @@ -4,6 +4,7 @@ from __future__ import annotations import hashlib import json +import math import re import unicodedata from collections import Counter @@ -341,3 +342,87 @@ def score_quality( flags=tuple(flags), fingerprint=fingerprint, ) + + +def _cosine_similarity(left: Sequence[float], right: Sequence[float]) -> float: + if not left or not right or len(left) != len(right): + return 0.0 + dot = math.fsum(a * b for a, b in zip(left, right)) + norm_left = math.sqrt(math.fsum(a * a for a in left)) + norm_right = math.sqrt(math.fsum(b * b for b in right)) + if not norm_left or not norm_right: + return 0.0 + return dot / (norm_left * norm_right) + + +def semantic_quality_scores( + record: Mapping[str, Any], + *, + source_content: str = "", + embed_model: Any = None, +) -> dict[str, Any] | None: + """用本地嵌入向量计算语义相关性（0-100）。 + + 返回 ``question_answer``（问题↔答案）、``answer_source``（答案↔来源， + 无来源时缺省）与 ``overall``；嵌入模型不可用时返回 None 降级，不阻断流程。 + """ + + try: + if embed_model is None: + from .embedding import semantic_embedding_model + + embed_model = semantic_embedding_model() + if embed_model is None: + return None + + question = normalize_text( + " ".join( + str(record.get(field) or "") + for field in ("instruction", "input") + ) + ) + answer = normalize_text( + str(record.get("output") or "") or str(record.get("chosen") or "") + ) + source = normalize_text(source_content) + texts = [text for text in {question, answer, source} if text] + if not texts: + return None + vectors = {text: embed_model.get_text_embedding(text) for text in texts} + except Exception: + return None + + scores: dict[str, Any] = {} + if question and answer: + scores["question_answer"] = round( + 100 * max(0.0, _cosine_similarity(vectors[question], vectors[answer])), 2 + ) + if answer and source: + scores["answer_source"] = round( + 100 * max(0.0, _cosine_similarity(vectors[answer], vectors[source])), 2 + ) + if not scores: + return None + scores["overall"] = round(sum(scores.values()) / len(scores), 2) + return scores + + +def composite_overall( + *, + rule: float | None, + semantic: float | None = None, + judge: float | None = None, +) -> float: + """三层加权组合：规则 35% + 语义 20% + 评审 45%，缺失层自动重归一。""" + + if rule is None: + rule = 0.0 + if judge is not None and semantic is not None: + overall = rule * 0.35 + semantic * 0.20 + judge * 0.45 + elif semantic is not None: + overall = rule * 0.60 + semantic * 0.40 + elif judge is not None: + overall = rule * 0.55 + judge * 0.45 + else: + overall = rule + return round(max(0.0, min(100.0, overall)), 2) diff --git a/backend/app/modules/data_process/evaluation.py b/backend/app/modules/data_process/evaluation.py new file mode 100644 index 0000000..c9051f3 --- /dev/null +++ b/backend/app/modules/data_process/evaluation.py @@ -0,0 +1,313 @@ +"""数据处理 - 生成结果的多层质量评测。 + +三层体系：规则层（确定性规则分）+ 语义层（本地嵌入向量）+ 评审层 +（复用生成模型按 rubric 打分的 LLM-as-judge）。任一层失败自动降级， +评测永远返回可用结果，不阻断调用方流程。 +""" + +from __future__ import annotations + +import logging +from collections.abc import Mapping +from dataclasses import asdict +from datetime import UTC, datetime +from typing import Any + +import httpx + +from .algorithms import normalize_text, score_quality +from .algorithms.quality import composite_overall, semantic_quality_scores +from .generation import ( + ModelGenerationError, + _is_retryable_generation_error, + _json_payload, + _message_content, + chat_completions_url, +) + +logger = logging.getLogger(__name__) + +# 送入评审提示词的来源正文上限，避免超长切片挤占评分输出空间。 +_MAX_JUDGE_SOURCE_CHARS = 6000 + +_JUDGE_DIMENSIONS: dict[str, tuple[str, ...]] = { + "standard": ( + "faithfulness", + "correctness", + "clarity", + "completeness", + "alignment", + ), + "reasoning": ( + "faithfulness", + "correctness", + "clarity", + "completeness", + "alignment", + "reasoning_validity", + ), + "dpo": ( + "clarity", + "chosen_quality", + "rejected_quality", + "preference_reasonableness", + "faithfulness", + ), +} + +_DIMENSION_LABELS: dict[str, str] = { + "faithfulness": "忠实度", + "correctness": "正确性", + "clarity": "问题清晰度", + "completeness": "回答完整性", + "alignment": "指令对齐", + "reasoning_validity": "推理有效性", + "chosen_quality": "chosen 回答质量", + "rejected_quality": "rejected 回答质量", + "preference_reasonableness": "偏好区分合理性", +} + +_DIMENSION_RULES: dict[str, str] = { + "faithfulness": "忠实度：答案的全部陈述是否被参考资料支持，没有编造、没有引入资料之外的信息；未提供参考资料时按答案内部自洽性评估", + "correctness": "正确性：答案中的事实、概念与计算是否正确", + "clarity": "问题清晰度：问题是否清晰、自包含、无歧义，脱离上下文也能理解", + "completeness": "回答完整性：答案是否充分、直接地回应了问题的全部要点", + "alignment": "指令对齐：答案的形式与范围是否符合问题的要求（如格式、语言、范围限定）", + "reasoning_validity": "推理有效性：思维链步骤是否逻辑连贯、无跳步或循环论证，结论是否由推理过程自然得出", + "chosen_quality": "chosen 回答质量：更优回答的正确性、完整性与表述质量", + "rejected_quality": "rejected 回答质量：较差回答是否仍具备基本可读性，使对比训练有意义", + "preference_reasonableness": "偏好区分合理性：chosen 是否明显优于 rejected，且优劣差异与问题直接相关", +} + + +def _judge_system_prompt(output_type: str) -> str: + dimensions = _JUDGE_DIMENSIONS[output_type] + rules = "\n".join(f"- {_DIMENSION_RULES[name]}" for name in dimensions) + scores_schema = ", ".join(f'"{name}": 1-5' for name in dimensions) + return ( + "你是大模型训练数据质量评审员。严格依据用户消息中的【参考资料】评审这条训练数据，逐维度按 1-5 分打分：\n" + f"{rules}\n" + "评分锚点：5 分=完全符合维度描述；3 分=基本符合但有明显不足；1 分=严重不符合。\n" + "忠实度只依据参考资料与公认常识判断，无法得到支持的陈述必须扣分；不要因为答案冗长而加分。\n" + "只输出一个 JSON 对象，不要输出 JSON 之外的任何文字。\n" + '输出格式：{"scores": {' + scores_schema + '}, "reason": "一句话总评", "issues": ["具体问题，没有则为空数组"]}' + ) + + +def _judge_user_prompt(record: Mapping[str, Any], source_content: str) -> str: + source = normalize_text(source_content)[:_MAX_JUDGE_SOURCE_CHARS] or "（无参考资料）" + instruction = normalize_text(str(record.get("instruction") or "")) or "（空）" + input_text = normalize_text(str(record.get("input") or "")) + sections = [f"【参考资料】\n{source}", f"【问题】\n{instruction}"] + if input_text: + sections.append(f"【输入】\n{input_text}") + if record.get("chosen") or record.get("rejected"): + sections.append(f"【更优回答 chosen】\n{normalize_text(str(record.get('chosen') or '')) or '（空）'}") + sections.append(f"【较差回答 rejected】\n{normalize_text(str(record.get('rejected') or '')) or '（空）'}") + else: + output = normalize_text(str(record.get("output") or "")) + sections.append(f"【回答】\n{output or '（空）'}") + return "\n\n".join(sections) + + +def _validated_judge_payload(payload: Any, output_type: str) -> dict[str, Any]: + if not isinstance(payload, Mapping): + raise ModelGenerationError("评审响应不是 JSON 对象") + raw_scores = payload.get("scores") + if not isinstance(raw_scores, Mapping): + raise ModelGenerationError("评审响应缺少 scores 对象") + expected = _JUDGE_DIMENSIONS[output_type] + scores: dict[str, float] = {} + for name in expected: + value = raw_scores.get(name) + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise ModelGenerationError(f"评审响应缺少维度 {name} 的有效分数") + scores[name] = round(max(1.0, min(5.0, float(value))), 1) + issues = payload.get("issues") + if not isinstance(issues, list): + issues = [] + issues = [str(item)[:200] for item in issues if str(item).strip()][:8] + reason = normalize_text(str(payload.get("reason") or ""))[:300] + return { + "scores": scores, + "overall": round(sum(scores.values()) / len(scores) * 20, 2), + "reason": reason, + "issues": issues, + } + + +def _judge_record( + record: Mapping[str, Any], + source_content: str, + *, + model: Mapping[str, Any], + config: Mapping[str, Any], + client: httpx.Client | None, +) -> dict[str, Any] | None: + output_type = str(config.get("output_type") or "standard").strip().lower() + if output_type not in _JUDGE_DIMENSIONS: + output_type = "standard" + endpoint = chat_completions_url(str(model.get("api_url") or "")) + model_name = str(model.get("online_model_name") or model.get("name") or "").strip() + if not model_name: + raise ModelGenerationError("generation model name is required") + temperature = 0.1 + max_tokens = max(256, min(2048, int(config.get("max_tokens", 1024) or 1024))) + timeout = max(1.0, min(120.0, float(config.get("request_timeout_seconds", 60) or 60))) + retries = max(0, min(5, int(config.get("generation_retries", 2) or 2))) + headers = {"Content-Type": "application/json"} + api_key = str(model.get("api_key") or "").strip() + if api_key: + headers["Authorization"] = f"Bearer {api_key}" + + request_payload: dict[str, Any] = { + "model": model_name, + "messages": [ + {"role": "system", "content": _judge_system_prompt(output_type)}, + {"role": "user", "content": _judge_user_prompt(record, source_content)}, + ], + "temperature": temperature, + "max_tokens": max_tokens, + } + if bool(config.get("json_mode", False)): + request_payload["response_format"] = {"type": "json_object"} + + owns_client = client is None + http_client = client or httpx.Client(timeout=timeout) + try: + last_error: Exception | None = None + for _ in range(retries + 1): + try: + response = http_client.post(endpoint, headers=headers, json=request_payload) + response.raise_for_status() + body = response.json() + if not isinstance(body, Mapping): + raise ModelGenerationError("model response body must be a JSON object") + judged = _validated_judge_payload( + _json_payload(_message_content(body)), + output_type, + ) + judged["model"] = model_name + judged["output_type"] = output_type + return judged + except Exception as exc: + last_error = exc + if not _is_retryable_generation_error(exc): + break + raise ModelGenerationError(f"质量评审调用失败: {last_error}") + finally: + if owns_client: + http_client.close() + + +def evaluate_result_record( + record: Mapping[str, Any], + *, + source_content: str = "", + model: Mapping[str, Any] | None = None, + config: Mapping[str, Any] | None = None, + client: httpx.Client | None = None, + embed_model: Any = None, + min_output_length: int = 20, +) -> dict[str, Any]: + """对一条生成结果执行三层评测，返回可直接落库的 quality_score 字典。 + + 规则层字段保持原样平铺（向后兼容既有读取方）；新增 ``semantic``、 + ``judge``、``layers``、``evaluated`` 与组合 ``overall``。 + """ + + config_dict = dict(config or {}) + rule = score_quality( + record, + min_output_length=min_output_length, + source_content=source_content, + ) + quality: dict[str, Any] = asdict(rule) + + semantic = semantic_quality_scores( + record, + source_content=source_content, + embed_model=embed_model, + ) + judge: dict[str, Any] | None = None + if model is not None: + try: + judge = _judge_record( + record, + source_content, + model=model, + config=config_dict, + client=client, + ) + except Exception as exc: + logger.warning( + "data process judge evaluation degraded: %s", + exc, + ) + + layers = { + "rule": rule.overall, + "semantic": semantic.get("overall") if semantic else None, + "judge": judge.get("overall") if judge else None, + } + quality.update( + semantic=semantic, + judge=judge, + layers=layers, + evaluated=True, + evaluated_at=datetime.now(UTC).isoformat(), + overall=composite_overall( + rule=layers["rule"], + semantic=layers["semantic"], + judge=layers["judge"], + ), + ) + return quality + + +def reevaluate_edited_record( + record: Mapping[str, Any], + *, + source_content: str = "", + previous_quality: Mapping[str, Any] | None = None, + embed_model: Any = None, + min_output_length: int = 20, +) -> dict[str, Any]: + """手动编辑/恢复后重算规则与语义层，丢弃已过期的评审层。 + + 编辑会改变内容，旧的评审分不再可信；规则与语义层本地重算零成本。 + ``evaluated`` 标记沿用原值，保证已评测过的结果编辑后仍有可用分数。 + """ + + rule = score_quality( + record, + min_output_length=min_output_length, + source_content=source_content, + ) + quality: dict[str, Any] = asdict(rule) + semantic = semantic_quality_scores( + record, + source_content=source_content, + embed_model=embed_model, + ) + previous = dict(previous_quality or {}) + evaluated = bool(previous.get("evaluated")) + layers = { + "rule": rule.overall, + "semantic": semantic.get("overall") if semantic else None, + "judge": None, + } + quality.update( + semantic=semantic, + judge=None, + layers=layers, + evaluated=evaluated, + evaluated_at=( + datetime.now(UTC).isoformat() if evaluated else None + ), + overall=composite_overall( + rule=layers["rule"], + semantic=layers["semantic"], + ), + ) + return quality diff --git a/backend/app/schemas/data_process.py b/backend/app/schemas/data_process.py index aa3671f..4778ba0 100644 --- a/backend/app/schemas/data_process.py +++ b/backend/app/schemas/data_process.py @@ -384,6 +384,26 @@ class ResultBatchRegenerateRequest(BaseModel): return self +class ResultBatchEvaluateItem(BaseModel): + model_config = ConfigDict(extra="forbid") + + result_id: str = Field(min_length=1, max_length=100) + expected_updated_at: str = Field(min_length=1, max_length=100) + + +class ResultBatchEvaluateRequest(BaseModel): + model_config = ConfigDict(extra="forbid") + + items: list[ResultBatchEvaluateItem] = Field(min_length=1, max_length=50) + + @model_validator(mode="after") + def validate_unique_results(self) -> ResultBatchEvaluateRequest: + result_ids = [item.result_id for item in self.items] + if len(result_ids) != len(set(result_ids)): + raise ValueError("result_id values must be unique") + return self + + class DatasetSplit(BaseModel): model_config = ConfigDict(extra="forbid") diff --git a/backend/tests/test_data_process_api.py b/backend/tests/test_data_process_api.py index 2170935..1e354ac 100644 --- a/backend/tests/test_data_process_api.py +++ b/backend/tests/test_data_process_api.py @@ -1691,6 +1691,223 @@ def test_batch_result_regeneration_rejects_locked_tasks_before_model_call( assert model_calls == 0 +def _prepare_evaluation_task( + client: TestClient, + store: Any, + tmp_path: Path, + *, + config: dict[str, Any] | None = None, +) -> str: + task_id = client.post( + "/modelTF/data-process", + json={ + "name": "数据评测", + "process_type": "structured", + "config": config or {"generation_model_id": "model-1", "output_type": "standard"}, + }, + ).json()["data"]["id"] + store.tasks[task_id].update( + status="completed", + progress=100, + workflow_step="results", + results_confirmed=False, + ) + store.models["model-1"] = { + "id": "model-1", + "online_model_name": "test-model", + "api_url": "https://model.example/v1", + "api_key": "secret", + } + store.previews[task_id] = [ + { + "id": "preview-1", + "status": "original", + "original_content": "申请编号用于唯一标识一笔报销申请。", + "edited_content": "申请编号用于唯一标识一笔报销申请。", + }, + { + "id": "preview-2", + "status": "original", + "original_content": "联系电话用于联系申请人。", + "edited_content": "联系电话用于联系申请人。", + }, + ] + store.results[task_id] = [ + { + "id": "result-1", + "preview_item_id": "preview-1", + "instruction": "申请编号有什么作用？", + "input": "", + "output": "申请编号用于唯一标识一笔报销申请。", + "original_instruction": "申请编号有什么作用？", + "original_input": "", + "original_output": "申请编号用于唯一标识一笔报销申请。", + "status": "valid", + "error": None, + "split": "train", + "quality_score": {}, + "updated_at": "2026-08-19T09:00:00Z", + }, + { + "id": "result-2", + "preview_item_id": "preview-2", + "instruction": "联系电话有什么作用？", + "input": "", + "output": "联系电话用于联系申请人。", + "original_instruction": "联系电话有什么作用？", + "original_input": "", + "original_output": "联系电话用于联系申请人。", + "status": "valid", + "error": None, + "split": "train", + "quality_score": {}, + "updated_at": "2026-08-19T09:00:01Z", + }, + ] + return task_id + + +def test_results_can_be_evaluated_in_batch_with_partial_success( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + client, store, _ = make_client(tmp_path) + task_id = _prepare_evaluation_task(client, store, tmp_path) + evaluation_calls: list[dict[str, Any]] = [] + + def fake_evaluate(record: dict[str, Any], **kwargs: Any) -> dict[str, Any]: + evaluation_calls.append({"record": deepcopy(record), "kwargs": {k: v for k, v in kwargs.items() if k != "client"}}) + return { + "overall": 88.0, + "completeness": 100.0, + "length": 100.0, + "readability": 100.0, + "relevance": 90.0, + "duplicate": 100.0, + "is_valid": True, + "flags": [], + "fingerprint": "fp", + "semantic": {"question_answer": 80.0, "answer_source": 90.0, "overall": 85.0}, + "judge": {"scores": {"faithfulness": 5}, "overall": 90.0}, + "layers": {"rule": 92.0, "semantic": 85.0, "judge": 90.0}, + "evaluated": True, + } + + monkeypatch.setattr(data_process_endpoint, "evaluate_result_record", fake_evaluate) + response = client.post( + f"/modelTF/data-process/{task_id}/results/evaluate-batch", + json={ + "items": [ + {"result_id": "result-1", "expected_updated_at": "2026-08-19T09:00:00Z"}, + # 乐观锁版本不匹配：该条应按冲突失败，另一条仍成功。 + {"result_id": "result-2", "expected_updated_at": "2026-08-18T00:00:00Z"}, + ], + }, + ) + + assert response.status_code == 200 + data = response.json()["data"] + assert data["total"] == 2 + assert data["succeeded"] == 1 + assert data["failed"] == 1 + assert [item["id"] for item in data["items"]] == ["result-1"] + assert data["failures"][0]["result_id"] == "result-2" + assert data["failures"][0]["code"] == "conflict" + + assert len(evaluation_calls) == 1 + assert evaluation_calls[0]["record"]["instruction"] == "申请编号有什么作用？" + assert evaluation_calls[0]["kwargs"]["model"]["online_model_name"] == "test-model" + assert evaluation_calls[0]["kwargs"]["source_content"] == "申请编号用于唯一标识一笔报销申请。" + + stored = store.results[task_id][0]["quality_score"] + assert stored["evaluated"] is True + assert stored["layers"]["judge"] == 90.0 + assert store.results[task_id][1]["quality_score"] == {} + + +def test_evaluation_without_generation_model_skips_judge_layer( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + client, store, _ = make_client(tmp_path) + task_id = _prepare_evaluation_task(client, store, tmp_path, config={"output_type": "standard"}) + seen_models: list[Any] = [] + + def fake_evaluate(record: dict[str, Any], **kwargs: Any) -> dict[str, Any]: + seen_models.append(kwargs.get("model")) + return { + "overall": 70.0, "is_valid": True, "flags": [], + "semantic": None, "judge": None, + "layers": {"rule": 70.0, "semantic": None, "judge": None}, + "evaluated": True, + } + + monkeypatch.setattr(data_process_endpoint, "evaluate_result_record", fake_evaluate) + response = client.post( + f"/modelTF/data-process/{task_id}/results/evaluate-batch", + json={"items": [{"result_id": "result-1", "expected_updated_at": "2026-08-19T09:00:00Z"}]}, + ) + + assert response.status_code == 200 + assert response.json()["data"]["succeeded"] == 1 + # 任务未配置生成模型时，评审层收到的 model 必须是 None。 + assert seen_models == [None] + + +def test_evaluation_rejects_running_task( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + client, store, _ = make_client(tmp_path) + task_id = _prepare_evaluation_task(client, store, tmp_path) + store.tasks[task_id]["status"] = "running" + evaluation_calls = 0 + + def fake_evaluate(*args: Any, **kwargs: Any) -> dict[str, Any]: + nonlocal evaluation_calls + evaluation_calls += 1 + return {"overall": 0, "is_valid": True, "flags": []} + + monkeypatch.setattr(data_process_endpoint, "evaluate_result_record", fake_evaluate) + response = client.post( + f"/modelTF/data-process/{task_id}/results/evaluate-batch", + json={"items": [{"result_id": "result-1", "expected_updated_at": "2026-08-19T09:00:00Z"}]}, + ) + + assert response.status_code == 409 + assert evaluation_calls == 0 + + +def test_result_update_preserves_evaluation_layers_and_drops_stale_judge( + tmp_path: Path, +) -> None: + client, store, _ = make_client(tmp_path) + task_id = _prepare_evaluation_task(client, store, tmp_path) + store.results[task_id][0]["quality_score"] = { + "overall": 90.0, + "is_valid": True, + "flags": [], + "semantic": {"overall": 85.0}, + "judge": {"overall": 92.0}, + "layers": {"rule": 90.0, "semantic": 85.0, "judge": 92.0}, + "evaluated": True, + } + + response = client.put( + f"/modelTF/data-process/{task_id}/results/result-1", + json={"output": "人工修正后的答案：申请编号唯一标识一笔报销申请。"}, + ) + + assert response.status_code == 200 + stored = store.results[task_id][0]["quality_score"] + # 手动编辑后：规则+语义重算，评审分丢弃，evaluated 标记保留。 + assert stored["evaluated"] is True + assert stored["judge"] is None + assert stored["layers"]["judge"] is None + assert stored["layers"]["rule"] is not None + assert stored["overall"] >= 0 + + def test_preview_build_replaces_only_selected_files_and_reports_file_counts( tmp_path: Path, ) -> None: diff --git a/backend/tests/test_data_process_evaluation.py b/backend/tests/test_data_process_evaluation.py new file mode 100644 index 0000000..eb974bf --- /dev/null +++ b/backend/tests/test_data_process_evaluation.py @@ -0,0 +1,284 @@ +"""数据评测模块（三层质量评分）的单元测试。""" + +from __future__ import annotations + +import json +from typing import Any + +import httpx +import pytest + +from app.modules.data_process.algorithms.quality import ( + composite_overall, + semantic_quality_scores, +) +from app.modules.data_process.evaluation import ( + _JUDGE_DIMENSIONS, + _judge_system_prompt, + _validated_judge_payload, + evaluate_result_record, + reevaluate_edited_record, +) +from app.modules.data_process.generation import ModelGenerationError + +RECORD = { + "instruction": "申请编号有什么作用？", + "input": "", + "output": "申请编号用于唯一标识一笔报销申请，便于跟踪审批状态。", +} +SOURCE = "报销系统中，申请编号用于唯一标识一笔报销申请，并支持跟踪审批状态。" + + +class _FakeEmbedModel: + """按关键词返回固定向量，模拟语义嵌入。""" + + def get_text_embedding(self, text: str) -> list[float]: + if "作用" in text or "编号" in text and "？" in text: + return [0.9, 0.1, 0.0] + if "申请编号" in text: + return [0.85, 0.2, 0.0] + return [0.0, 0.1, 0.9] + + +class _FailingEmbedModel: + def get_text_embedding(self, text: str) -> list[float]: + raise RuntimeError("embedding unavailable") + + +class _FakeResponse: + def __init__(self, payload: dict[str, Any]): + self._payload = payload + + def raise_for_status(self) -> None: + return None + + def json(self) -> dict[str, Any]: + return self._payload + + +class _FakeClient: + def __init__(self, content: str): + self._content = content + self.calls: list[dict[str, Any]] = [] + + def post(self, endpoint: str, headers: Any = None, json: Any = None) -> _FakeResponse: + self.calls.append({"endpoint": endpoint, "payload": json}) + return _FakeResponse({ + "choices": [{"message": {"content": self._content}, "finish_reason": "stop"}], + }) + + def close(self) -> None: + return None + + +class _RaisingClient: + def post(self, endpoint: str, headers: Any = None, json: Any = None) -> _FakeResponse: + raise httpx.ConnectError("model endpoint unreachable") + + def close(self) -> None: + return None + + +def _judge_content(scores: dict[str, float], **extra: Any) -> str: + return json.dumps({"scores": scores, "reason": "总体可靠", "issues": [], **extra}) + + +def test_judge_system_prompt_covers_rubric_dimensions() -> None: + standard = _judge_system_prompt("standard") + for name in _JUDGE_DIMENSIONS["standard"]: + assert name in standard + assert "1-5" in standard + + dpo = _judge_system_prompt("dpo") + assert "chosen_quality" in dpo + assert "preference_reasonableness" in dpo + + reasoning = _judge_system_prompt("reasoning") + assert "reasoning_validity" in reasoning + + +def test_validated_judge_payload_converts_scores_to_overall() -> None: + judged = _validated_judge_payload( + { + "scores": { + "faithfulness": 5, + "correctness": 4, + "clarity": 4, + "completeness": 3, + "alignment": 4, + }, + "reason": "答案可靠", + "issues": ["回答略冗长"], + }, + "standard", + ) + + assert judged["overall"] == round((5 + 4 + 4 + 3 + 4) / 5 * 20, 2) + assert judged["issues"] == ["回答略冗长"] + assert judged["reason"] == "答案可靠" + + +def test_validated_judge_payload_clamps_out_of_range_scores() -> None: + judged = _validated_judge_payload( + { + "scores": { + "faithfulness": 9, + "correctness": 4, + "clarity": 4, + "completeness": 0, + "alignment": 4, + }, + }, + "standard", + ) + + assert judged["scores"]["faithfulness"] == 5.0 + assert judged["scores"]["completeness"] == 1.0 + + +@pytest.mark.parametrize( + "scores", + [ + {"faithfulness": 5, "correctness": 4, "clarity": 4, "completeness": 3}, + { + "faithfulness": 5, + "correctness": 4, + "clarity": "high", + "completeness": 3, + "alignment": 4, + }, + ], +) +def test_validated_judge_payload_rejects_incomplete_scores(scores: dict[str, Any]) -> None: + with pytest.raises(ModelGenerationError): + _validated_judge_payload({"scores": scores}, "standard") + + +def test_semantic_quality_scores_uses_cosine_similarity() -> None: + scores = semantic_quality_scores( + RECORD, + source_content=SOURCE, + embed_model=_FakeEmbedModel(), + ) + + assert scores is not None + assert 0 < scores["question_answer"] <= 100 + assert 0 < scores["answer_source"] <= 100 + assert scores["overall"] == round((scores["question_answer"] + scores["answer_source"]) / 2, 2) + + +def test_semantic_quality_scores_degrades_to_none_on_failure() -> None: + assert ( + semantic_quality_scores( + RECORD, + source_content=SOURCE, + embed_model=_FailingEmbedModel(), + ) + is None + ) + + +def test_composite_overall_weights_available_layers() -> None: + assert composite_overall(rule=80, semantic=90, judge=70) == round(80 * 0.35 + 90 * 0.20 + 70 * 0.45, 2) + assert composite_overall(rule=80, semantic=90) == round(80 * 0.6 + 90 * 0.4, 2) + assert composite_overall(rule=80) == 80.0 + assert composite_overall(rule=None, judge=100) == 45.0 + + +def test_evaluate_result_record_combines_three_layers() -> None: + client = _FakeClient( + _judge_content({ + "faithfulness": 5, + "correctness": 4, + "clarity": 5, + "completeness": 4, + "alignment": 5, + }) + ) + quality = evaluate_result_record( + RECORD, + source_content=SOURCE, + model={"api_url": "https://model.example", "online_model_name": "judge-model"}, + config={"output_type": "standard", "generation_retries": 0}, + client=client, + embed_model=_FakeEmbedModel(), + ) + + assert quality["evaluated"] is True + assert quality["judge"] is not None + assert quality["judge"]["model"] == "judge-model" + assert quality["semantic"] is not None + assert quality["layers"]["judge"] == quality["judge"]["overall"] + assert quality["overall"] == composite_overall( + rule=quality["layers"]["rule"], + semantic=quality["layers"]["semantic"], + judge=quality["layers"]["judge"], + ) + # 评审提示词必须携带来源原文作为评分锚点（正文经 NFKC 归一化）。 + user_message = client.calls[0]["payload"]["messages"][1]["content"] + assert "申请编号用于唯一标识一笔报销" in user_message + + +def test_evaluate_result_record_degrades_when_model_fails() -> None: + quality = evaluate_result_record( + RECORD, + source_content=SOURCE, + model={"api_url": "https://model.example", "online_model_name": "judge-model"}, + config={"output_type": "standard", "generation_retries": 0}, + client=_RaisingClient(), + embed_model=_FakeEmbedModel(), + ) + + assert quality["judge"] is None + assert quality["layers"]["judge"] is None + assert quality["semantic"] is not None + assert quality["overall"] == composite_overall( + rule=quality["layers"]["rule"], + semantic=quality["layers"]["semantic"], + ) + + +def test_evaluate_result_record_without_model_runs_two_layers() -> None: + quality = evaluate_result_record( + RECORD, + source_content=SOURCE, + model=None, + embed_model=_FakeEmbedModel(), + ) + + assert quality["judge"] is None + assert quality["evaluated"] is True + assert quality["overall"] == composite_overall( + rule=quality["layers"]["rule"], + semantic=quality["layers"]["semantic"], + ) + + +def test_reevaluate_edited_record_drops_stale_judge() -> None: + previous = { + "evaluated": True, + "judge": {"overall": 90.0}, + } + quality = reevaluate_edited_record( + {**RECORD, "output": "编辑后的新答案内容，用于验证重评逻辑。"}, + source_content=SOURCE, + previous_quality=previous, + embed_model=_FakeEmbedModel(), + ) + + assert quality["evaluated"] is True + assert quality["judge"] is None + assert quality["layers"]["judge"] is None + assert quality["semantic"] is not None + + +def test_reevaluate_edited_record_keeps_unevaluated_state() -> None: + quality = reevaluate_edited_record( + RECORD, + source_content=SOURCE, + previous_quality={}, + embed_model=_FakeEmbedModel(), + ) + + assert quality["evaluated"] is False + assert quality["evaluated_at"] is None diff --git a/frontend/src/api/modules/dataProcess.ts b/frontend/src/api/modules/dataProcess.ts index a9c35e0..71faaff 100644 --- a/frontend/src/api/modules/dataProcess.ts +++ b/frontend/src/api/modules/dataProcess.ts @@ -20,6 +20,8 @@ import type { DataProcessPublishResult, DataProcessQualityScore, DataProcessResult, + DataProcessResultBatchEvaluatePayload, + DataProcessResultBatchEvaluateResult, DataProcessResultBatchRegeneratePayload, DataProcessResultBatchRegenerateResult, DataProcessResultRegeneratePayload, @@ -320,5 +322,14 @@ export const regenerateDataProcessResults = ( { timeout: 240_000 }, ) +export const evaluateDataProcessResults = ( + taskId: string | number, + payload: DataProcessResultBatchEvaluatePayload, +) => post( + `/data-process/${encodeURIComponent(taskId)}/results/evaluate-batch`, + payload, + { timeout: 240_000 }, +) + export const publishDataProcess = (taskId: string | number, payload: DataProcessPublishPayload) => post(`/data-process/${encodeURIComponent(taskId)}/publish`, payload) diff --git a/frontend/src/plugins/echarts.ts b/frontend/src/plugins/echarts.ts index ce1737c..3508f4b 100644 --- a/frontend/src/plugins/echarts.ts +++ b/frontend/src/plugins/echarts.ts @@ -3,18 +3,21 @@ */ import { use } from 'echarts/core' import { CanvasRenderer } from 'echarts/renderers' -import { BarChart, PieChart } from 'echarts/charts' +import { BarChart, PieChart, RadarChart } from 'echarts/charts' import { GridComponent, TooltipComponent, LegendComponent, + RadarComponent, } from 'echarts/components' use([ CanvasRenderer, BarChart, PieChart, + RadarChart, GridComponent, TooltipComponent, LegendComponent, + RadarComponent, ]) diff --git a/frontend/src/types/dataProcess.ts b/frontend/src/types/dataProcess.ts index 921387f..63e86b9 100644 --- a/frontend/src/types/dataProcess.ts +++ b/frontend/src/types/dataProcess.ts @@ -398,6 +398,52 @@ export interface DataProcessResultBatchRegenerateResult { failures: DataProcessResultBatchRegenerateFailure[] } +export interface DataProcessResultBatchEvaluateItem { + result_id: string + expected_updated_at: string +} + +export interface DataProcessResultBatchEvaluatePayload { + items: DataProcessResultBatchEvaluateItem[] +} + +export interface DataProcessResultBatchEvaluateFailure { + result_id: string + code: 'conflict' | 'skipped' | 'evaluation_failed' | 'internal_error' + message: string +} + +export interface DataProcessResultBatchEvaluateResult { + batch_id: string + total: number + succeeded: number + failed: number + duration_ms: number + items: DataProcessResult[] + failures: DataProcessResultBatchEvaluateFailure[] +} + +export interface DataProcessQualitySemantic { + question_answer?: number + answer_source?: number + overall?: number +} + +export interface DataProcessQualityJudge { + scores?: Record + overall?: number + reason?: string + issues?: string[] + model?: string + output_type?: string +} + +export interface DataProcessQualityLayers { + rule?: number | null + semantic?: number | null + judge?: number | null +} + export interface DataProcessQualityScore { overall?: number completeness?: number @@ -408,6 +454,11 @@ export interface DataProcessQualityScore { is_valid?: boolean flags?: string[] fingerprint?: string + semantic?: DataProcessQualitySemantic | null + judge?: DataProcessQualityJudge | null + layers?: DataProcessQualityLayers | null + evaluated?: boolean + evaluated_at?: string | null source_pages?: number[] heading_path?: string[] source_locator?: DataProcessSourceLocator diff --git a/frontend/src/views/data-process/DataProcessCreateView.vue b/frontend/src/views/data-process/DataProcessCreateView.vue index 21951fc..a449f69 100644 --- a/frontend/src/views/data-process/DataProcessCreateView.vue +++ b/frontend/src/views/data-process/DataProcessCreateView.vue @@ -18,6 +18,7 @@ import { previewAffectingOptionsFor, } from './create/dataProcessCreateState' import { useDataProcessGeneration } from './create/useDataProcessGeneration' +import { useDataProcessEvaluation } from './create/useDataProcessEvaluation' import { useDataProcessPreviewBuild } from './create/useDataProcessPreviewBuild' import { useDataProcessRegeneration } from './create/useDataProcessRegeneration' import { createDefaultExternalSource, externalSourcePayload, restoreExternalSourceConfig, sourceConfigForBackend } from './create/externalSourceConfig' @@ -126,6 +127,19 @@ const { outputType: activeOutputType, beforeGenerate: beforeStartGeneration, }) +const { + evaluation, + evaluateAllResults, + resetEvaluation, +} = useDataProcessEvaluation({ + taskId, + results, + selectedResultId, +}) +// 生成结果被重置（重新切分/上传/重新生成配置）时同步清空评测进度。 +watch(results, (items) => { + if (!items.length) resetEvaluation() +}) const { enqueueSourceUpload, sourceUploading } = useDataProcessSourceUpload({ taskId, uploadedFiles, @@ -1156,10 +1170,12 @@ onMounted(() => { :preview-items="previewItems" :regenerating-result-id="regeneratingResultId" :bulk-regeneration="bulkRegeneration" + :evaluation="evaluation" :output-type="activeOutputType" @update:field="updateResultField" @regenerate:all="regenerateAllResults" @regenerate:item="regenerateResult" + @evaluate:all="evaluateAllResults" /> diff --git a/frontend/src/views/data-process/DataProcessDetailView.vue b/frontend/src/views/data-process/DataProcessDetailView.vue index ef0919b..d2bbef3 100644 --- a/frontend/src/views/data-process/DataProcessDetailView.vue +++ b/frontend/src/views/data-process/DataProcessDetailView.vue @@ -5,6 +5,7 @@ import { ElMessage, ElMessageBox } from 'element-plus' import PageCard from '@/components/PageCard.vue' import { usePolling } from '@/composables/usePolling' import { + evaluateDataProcessResults, getDataProcessProgress, getDataProcessResults, getDataProcessTask, @@ -13,6 +14,7 @@ import { restoreDataProcessResult, updateDataProcessResult, } from '@/api/modules/dataProcess' +import QualityRadarPopover from './create/QualityRadarPopover.vue' import type { DataProcessDatasetSplit, DataProcessPublishPayload, @@ -456,13 +458,101 @@ function resultStatusType(status: DataProcessResultStatus) { } function qualityScoreLabel(value: DataProcessResult['quality_score']) { - if (value == null) return '-' + if (value == null || !value.evaluated) return '-' const score = value.overall return Number.isFinite(score) ? Number(score).toFixed(1) : '-' } -function qualityFlagsLabel(value: DataProcessResult['quality_score']) { - return value?.flags?.length ? value.flags.join('、') : '未命中质量规则' +function qualityScoreTone(value: DataProcessResult['quality_score']) { + const score = Number(value?.overall) + if (!value?.evaluated || !Number.isFinite(score)) return '' + return score >= 80 ? 'is-success' : score >= 60 ? 'is-warning' : 'is-danger' +} + +function qualityScoreEvaluated(value: DataProcessResult['quality_score']) { + return Boolean(value?.evaluated && Number.isFinite(Number(value?.overall))) +} + +const evaluationRunning = ref(false) +const evaluationProgress = reactive({ + visible: false, + total: 0, + completed: 0, + succeeded: 0, + failed: 0, +}) +// 与批量重生成一致的分块大小，单批在接口 240 秒超时预算内。 +const EVALUATION_CHUNK_SIZE = 12 +const canEvaluate = computed(() => ( + detail.value?.status === 'completed' && !hasCurrentPublishedDataset.value +)) +const evaluationPercentage = computed(() => ( + evaluationProgress.total + ? Math.round((evaluationProgress.completed / evaluationProgress.total) * 100) + : 0 +)) + +async function loadAllResultIds() { + const first = await getDataProcessResults(taskId.value, { page: 1, page_size: 500 }) + const items = [...first.items] + const pages = Math.ceil(first.total / first.page_size) + for (let page = 2; page <= pages; page += 1) { + const next = await getDataProcessResults(taskId.value, { page, page_size: 500 }) + items.push(...next.items) + } + return items +} + +async function runResultEvaluation() { + if (evaluationRunning.value || !canEvaluate.value) return + evaluationRunning.value = true + Object.assign(evaluationProgress, { + visible: true, + total: 0, + completed: 0, + succeeded: 0, + failed: 0, + }) + try { + const candidates = (await loadAllResultIds()).filter((item) => item.updated_at) + if (!candidates.length) { + ElMessage.info('当前没有可评测的结果') + return + } + evaluationProgress.total = candidates.length + for (let offset = 0; offset < candidates.length; offset += EVALUATION_CHUNK_SIZE) { + const chunk = candidates.slice(offset, offset + EVALUATION_CHUNK_SIZE) + try { + const evaluated = await evaluateDataProcessResults(taskId.value, { + items: chunk.map((item) => ({ + result_id: String(item.id), + expected_updated_at: item.updated_at as string, + })), + }) + evaluationProgress.completed += evaluated.total + evaluationProgress.succeeded += evaluated.succeeded + evaluationProgress.failed += evaluated.failed + } catch { + evaluationProgress.completed = evaluationProgress.total + evaluationProgress.failed += candidates.length - offset + break + } + } + await loadResults() + if (evaluationProgress.failed === 0) { + ElMessage.success(`数据评测完成：成功 ${evaluationProgress.succeeded} 条`) + } else if (evaluationProgress.succeeded > 0) { + ElMessage.warning( + `数据评测完成：成功 ${evaluationProgress.succeeded} 条，失败 ${evaluationProgress.failed} 条`, + ) + } else { + ElMessage.error(`数据评测失败：${evaluationProgress.failed} 条结果未完成评测`) + } + } catch { + ElMessage.error('数据评测中断，已完成的评分保持不变') + } finally { + evaluationRunning.value = false + } } function replaceResult(updated: DataProcessResult) { @@ -824,10 +914,33 @@ onBeforeUnmount(() => { + +