From 9428c6b785bde79f035bb8cf65bf2cfee851cc19 Mon Sep 17 00:00:00 2001 From: caoxiaozhu Date: Mon, 27 Jul 2026 09:11:51 +0800 Subject: [PATCH] =?UTF-8?q?feat(data-process):=20=E6=94=AF=E6=8C=81?= =?UTF-8?q?=E5=8D=95=E9=A1=B9=E7=94=9F=E6=88=90=E4=BA=94=E5=8D=81=E6=9D=A1?= =?UTF-8?q?=E6=95=B0=E6=8D=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../app/modules/data_process/algorithms.py | 16 +- backend/app/modules/data_process/constants.py | 7 + .../app/modules/data_process/generation.py | 217 +++++++++++------- backend/app/schemas/data_process.py | 12 +- backend/tests/test_data_process_algorithms.py | 24 +- backend/tests/test_data_process_api.py | 41 ++++ backend/tests/test_data_process_generation.py | 169 +++++++++++++- 7 files changed, 400 insertions(+), 86 deletions(-) create mode 100644 backend/app/modules/data_process/constants.py diff --git a/backend/app/modules/data_process/algorithms.py b/backend/app/modules/data_process/algorithms.py index 51d9a3b..07d4b38 100644 --- a/backend/app/modules/data_process/algorithms.py +++ b/backend/app/modules/data_process/algorithms.py @@ -34,6 +34,8 @@ from openpyxl.utils.cell import range_boundaries from pptx import Presentation from pypdf import PdfReader +from app.modules.data_process.constants import MAX_QA_PAIRS_PER_ITEM + TextFormat = Literal[ "json", "jsonl", @@ -2659,8 +2661,10 @@ def generate_standard_records( LLM 响应解析后的统一落库步骤。 """ - if not 1 <= qa_pairs_per_item <= 5: - raise ValueError("qa_pairs_per_item must be in [1, 5]") + if not 1 <= qa_pairs_per_item <= MAX_QA_PAIRS_PER_ITEM: + raise ValueError( + f"qa_pairs_per_item must be in [1, {MAX_QA_PAIRS_PER_ITEM}]" + ) prefixes = ( "请结合实际情况说明：", "请用通俗易懂的方式说明：", @@ -2677,7 +2681,13 @@ def generate_standard_records( variant_instruction = instruction if variant_index: if semantic_enrichment: - variant_instruction = f"{prefixes[variant_index]}{instruction}" + prefix = prefixes[variant_index % len(prefixes)] + if variant_index >= len(prefixes): + prefix = ( + f"{prefix.removesuffix('：')}" + f"（问法 {variant_index + 1}）：" + ) + variant_instruction = f"{prefix}{instruction}" else: variant_instruction = f"{instruction}(问法 {variant_index + 1})" raw_id = f"{preview_id}:{variant_index + 1}" diff --git a/backend/app/modules/data_process/constants.py b/backend/app/modules/data_process/constants.py new file mode 100644 index 0000000..9752443 --- /dev/null +++ b/backend/app/modules/data_process/constants.py @@ -0,0 +1,7 @@ +"""数据处理模块的共享限制。""" + +MAX_QA_PAIRS_PER_ITEM = 50 +MODEL_GENERATION_BATCH_SIZE = 10 + + +__all__ = ["MAX_QA_PAIRS_PER_ITEM", "MODEL_GENERATION_BATCH_SIZE"] diff --git a/backend/app/modules/data_process/generation.py b/backend/app/modules/data_process/generation.py index 6de4adc..1c1e14f 100644 --- a/backend/app/modules/data_process/generation.py +++ b/backend/app/modules/data_process/generation.py @@ -12,6 +12,10 @@ from urllib.parse import urlsplit, urlunsplit import httpx from app.modules.data_process.algorithms import normalize_text, stable_split_assignments +from app.modules.data_process.constants import ( + MAX_QA_PAIRS_PER_ITEM, + MODEL_GENERATION_BATCH_SIZE, +) class ModelGenerationError(ValueError): @@ -96,10 +100,20 @@ def _result_items(payload: Any) -> list[Mapping[str, Any]]: return items -def _prompt_messages(prompt: str, content: str, count: int) -> list[dict[str, str]]: +def _prompt_messages( + prompt: str, + content: str, + count: int, + *, + start_index: int, + total_count: int, +) -> list[dict[str, str]]: + end_index = start_index + count - 1 schema_instruction = ( f"必须只返回 JSON 对象，格式为 {{\"items\":[{{\"instruction\":\"...\"," f"\"input\":\"...\",\"output\":\"...\"}}]}}；items 必须包含 {count} 条。" + f"这是总计 {total_count} 条中的第 {start_index}-{end_index} 条，" + "各条必须使用不同的提问角度和表述，避免重复。" "instruction 和 output 不得为空，不要输出 Markdown 代码围栏或分析过程。" ) base_prompt = ( @@ -131,11 +145,14 @@ def generate_model_records( ) -> list[dict[str, Any]]: """调用 OpenAI 兼容接口，将预览切片生成标准训练记录。 - 单条调用失败会产生可人工修复的 invalid 结果，不会丢弃整批任务。 + 每个切片按安全批次调用模型；失败批次会产生一条可人工修复的 + invalid 结果，已经成功的批次不会丢失。 """ - if not 1 <= qa_pairs_per_item <= 5: - raise ModelGenerationError("qa_pairs_per_item must be in [1, 5]") + if not 1 <= qa_pairs_per_item <= MAX_QA_PAIRS_PER_ITEM: + raise ModelGenerationError( + f"qa_pairs_per_item must be in [1, {MAX_QA_PAIRS_PER_ITEM}]" + ) 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: @@ -161,84 +178,126 @@ def generate_model_records( content = normalize_text( str(item.get("edited_content") or item.get("original_content") or "") ) - request_payload: dict[str, Any] = { - "model": model_name, - "messages": _prompt_messages( - str(config.get("generation_prompt") or ""), - content, - qa_pairs_per_item, - ), - "temperature": temperature, - "max_tokens": max_tokens, - } - if bool(config.get("json_mode", False)): - request_payload["response_format"] = {"type": "json_object"} - - last_error: Exception | None = None - generated_items: list[Mapping[str, Any]] | 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") - generated_items = _result_items(_json_payload(_message_content(body))) - break - except (httpx.HTTPError, json.JSONDecodeError, ModelGenerationError) as exc: - last_error = exc - - if generated_items is None: - error_message = str(last_error or "model generation failed")[:2000] - result_id = f"result_{hashlib.sha256(f'{preview_id}:error'.encode()).hexdigest()[:16]}" - results.append( - { - "id": result_id, - "preview_item_id": preview_id, - "instruction": "模型生成失败，请人工补充", - "input": content, - "output": "", - "original_instruction": "模型生成失败，请人工补充", - "original_input": content, - "original_output": "", - "status": "invalid", - "error": error_message, - "split": "train", - } + for batch_offset in range(0, qa_pairs_per_item, MODEL_GENERATION_BATCH_SIZE): + batch_count = min( + MODEL_GENERATION_BATCH_SIZE, + qa_pairs_per_item - batch_offset, ) - if on_progress: - on_progress(item_index + 1, total_items) - continue + batch_start = batch_offset + 1 + batch_end = batch_offset + batch_count + request_payload: dict[str, Any] = { + "model": model_name, + "messages": _prompt_messages( + str(config.get("generation_prompt") or ""), + content, + batch_count, + start_index=batch_start, + total_count=qa_pairs_per_item, + ), + "temperature": temperature, + "max_tokens": max_tokens, + } + if bool(config.get("json_mode", False)): + request_payload["response_format"] = {"type": "json_object"} - for variant_index, value in enumerate(generated_items[:qa_pairs_per_item]): - instruction = normalize_text(str(value.get("instruction") or value.get("question") or "")) - input_text = normalize_text(str(value.get("input") or value.get("context") or "")) - output = normalize_text( - str( - value.get("output") - or value.get("answer") - or value.get("response") - or "" + last_error: Exception | None = None + generated_items: list[Mapping[str, Any]] | 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" + ) + candidate_items = _result_items( + _json_payload(_message_content(body)) + ) + if len(candidate_items) < batch_count: + raise ModelGenerationError( + "model response contains fewer result objects than requested: " + f"expected {batch_count}, got {len(candidate_items)}" + ) + generated_items = candidate_items + break + except ( + httpx.HTTPError, + json.JSONDecodeError, + ModelGenerationError, + ) as exc: + last_error = exc + + if generated_items is None: + error_message = str(last_error or "model generation failed")[:2000] + failure_instruction = ( + f"模型生成失败，请人工补充（第 {batch_start}-{batch_end} 条）" + ) + result_id = ( + "result_" + f"{hashlib.sha256(f'{preview_id}:error:{batch_start}'.encode()).hexdigest()[:16]}" + ) + results.append( + { + "id": result_id, + "preview_item_id": preview_id, + "instruction": failure_instruction, + "input": content, + "output": "", + "original_instruction": failure_instruction, + "original_input": content, + "original_output": "", + "status": "invalid", + "error": error_message, + "split": "train", + } + ) + continue + + for batch_index, value in enumerate(generated_items[:batch_count]): + variant_index = batch_offset + batch_index + instruction = normalize_text( + str(value.get("instruction") or value.get("question") or "") + ) + input_text = normalize_text( + str(value.get("input") or value.get("context") or "") + ) + output = normalize_text( + str( + value.get("output") + or value.get("answer") + or value.get("response") + or "" + ) + ) + raw_id = f"{preview_id}:{variant_index + 1}:{instruction}:{output}" + result_id = ( + f"result_{hashlib.sha256(raw_id.encode()).hexdigest()[:16]}" + ) + valid = bool(instruction and output) + results.append( + { + "id": result_id, + "preview_item_id": preview_id, + "instruction": instruction, + "input": input_text, + "output": output, + "original_instruction": instruction, + "original_input": input_text, + "original_output": output, + "status": "valid" if valid else "invalid", + "error": ( + None + if valid + else "model result is missing instruction or output" + ), + "split": "train", + } ) - ) - raw_id = f"{preview_id}:{variant_index + 1}:{instruction}:{output}" - result_id = f"result_{hashlib.sha256(raw_id.encode()).hexdigest()[:16]}" - valid = bool(instruction and output) - results.append( - { - "id": result_id, - "preview_item_id": preview_id, - "instruction": instruction, - "input": input_text, - "output": output, - "original_instruction": instruction, - "original_input": input_text, - "original_output": output, - "status": "valid" if valid else "invalid", - "error": None if valid else "model result is missing instruction or output", - "split": "train", - } - ) if on_progress: on_progress(item_index + 1, total_items) finally: diff --git a/backend/app/schemas/data_process.py b/backend/app/schemas/data_process.py index ce2a03a..b79f75e 100644 --- a/backend/app/schemas/data_process.py +++ b/backend/app/schemas/data_process.py @@ -5,6 +5,8 @@ from typing import Any, Literal from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator +from app.modules.data_process.constants import MAX_QA_PAIRS_PER_ITEM + def _config_value(config: dict[str, Any], snake_name: str, camel_name: str, default: Any) -> Any: if snake_name in config: @@ -89,8 +91,14 @@ def _validate_process_config(config: dict[str, Any]) -> None: pairs = _config_value(config, snake_name, camel_name, None) if pairs is None: continue - if isinstance(pairs, bool) or not isinstance(pairs, int) or not 1 <= pairs <= 5: - raise ValueError(f"{snake_name} must be an integer in [1, 5]") + if ( + isinstance(pairs, bool) + or not isinstance(pairs, int) + or not 1 <= pairs <= MAX_QA_PAIRS_PER_ITEM + ): + raise ValueError( + f"{snake_name} must be an integer in [1, {MAX_QA_PAIRS_PER_ITEM}]" + ) class DataProcessStatus(StrEnum): diff --git a/backend/tests/test_data_process_algorithms.py b/backend/tests/test_data_process_algorithms.py index e091f53..1fa8520 100644 --- a/backend/tests/test_data_process_algorithms.py +++ b/backend/tests/test_data_process_algorithms.py @@ -21,7 +21,6 @@ from app.modules.data_process.algorithms import ( detect_document_structure, detect_pdf_document_noise, detect_text_format, - estimate_token_count, extract_pdf_page_texts, extract_structured_records, generate_standard_records, @@ -748,3 +747,26 @@ def test_generate_standard_records_supports_json_qa_and_stable_variants() -> Non split={"train": 100, "validation": 0, "test": 0}, split_seed="task-1", ) + + +def test_generate_standard_records_supports_fifty_unique_semantic_variants() -> None: + records = generate_standard_records( + [{"id": "preview-50", "edited_content": "问：如何操作？\n答：按步骤操作。"}], + qa_pairs_per_item=50, + semantic_enrichment=True, + split={"train": 100, "validation": 0, "test": 0}, + split_seed="task-50", + ) + + assert len(records) == 50 + assert len({record["id"] for record in records}) == 50 + assert len({record["instruction"] for record in records}) == 50 + assert all(record["status"] == "valid" for record in records) + + +@pytest.mark.parametrize("qa_pairs_per_item", [0, 51]) +def test_generate_standard_records_rejects_out_of_range_count( + qa_pairs_per_item: int, +) -> None: + with pytest.raises(ValueError, match=r"\[1, 50\]"): + generate_standard_records([], qa_pairs_per_item=qa_pairs_per_item) diff --git a/backend/tests/test_data_process_api.py b/backend/tests/test_data_process_api.py index 1c7ea3a..ea506dd 100644 --- a/backend/tests/test_data_process_api.py +++ b/backend/tests/test_data_process_api.py @@ -945,6 +945,47 @@ def test_config_validation_and_stop_state(tmp_path: Path) -> None: assert stopped.json()["data"]["status"] == "stopped" +@pytest.mark.parametrize("config_key", ["qa_pairs_per_row", "qa_pairs_per_chunk"]) +@pytest.mark.parametrize("count", [1, 50]) +def test_qa_pair_config_accepts_supported_boundaries( + tmp_path: Path, + config_key: str, + count: int, +) -> None: + client, _, _ = make_client(tmp_path) + response = client.post( + "/modelTF/data-process", + json={ + "name": "问答数量边界", + "process_type": "unstructured", + "config": {config_key: count}, + }, + ) + + assert response.status_code == 200 + + +@pytest.mark.parametrize("config_key", ["qa_pairs_per_row", "qa_pairs_per_chunk"]) +@pytest.mark.parametrize("count", [0, 51]) +def test_qa_pair_config_rejects_out_of_range_boundaries( + tmp_path: Path, + config_key: str, + count: int, +) -> None: + client, _, _ = make_client(tmp_path) + response = client.post( + "/modelTF/data-process", + json={ + "name": "问答数量越界", + "process_type": "unstructured", + "config": {config_key: count}, + }, + ) + + assert response.status_code == 422 + assert "[1, 50]" in response.text + + def test_upload_batch_is_atomic_and_empty_files_are_rejected(tmp_path: Path) -> None: client, store, storage = make_client(tmp_path) task_id = client.post( diff --git a/backend/tests/test_data_process_generation.py b/backend/tests/test_data_process_generation.py index ef3fa19..2961bcf 100644 --- a/backend/tests/test_data_process_generation.py +++ b/backend/tests/test_data_process_generation.py @@ -3,8 +3,13 @@ from __future__ import annotations import json import httpx +import pytest -from app.modules.data_process.generation import chat_completions_url, generate_model_records +from app.modules.data_process.generation import ( + ModelGenerationError, + chat_completions_url, + generate_model_records, +) def test_chat_completions_url_accepts_host_base_and_complete_url() -> None: @@ -100,3 +105,165 @@ def test_generate_model_records_keeps_partial_failure_for_manual_repair() -> Non assert len(records) == 1 assert records[0]["status"] == "invalid" assert records[0]["error"] + + +def test_generate_model_records_batches_fifty_results_with_unique_ids() -> None: + requests: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + requests.append(request) + batch_start = (len(requests) - 1) * 10 + 1 + batch_end = batch_start + 9 + payload = json.loads(request.content) + system_prompt = payload["messages"][0]["content"] + assert "items 必须包含 10 条" in system_prompt + assert f"第 {batch_start}-{batch_end} 条" in system_prompt + return httpx.Response( + 200, + json={ + "choices": [ + { + "message": { + "content": json.dumps( + { + "items": [ + { + "instruction": "同一问题", + "input": "来源正文", + "output": "同一答案", + } + for _ in range(batch_start, batch_end + 1) + ] + }, + ensure_ascii=False, + ) + } + } + ] + }, + ) + + records = generate_model_records( + [{"id": "preview-50", "edited_content": "来源正文"}], + model={"name": "model", "api_url": "https://model.example/v1"}, + config={}, + task_id="task-50", + split={"train": 100, "validation": 0, "test": 0}, + qa_pairs_per_item=50, + client=httpx.Client(transport=httpx.MockTransport(handler)), + ) + + assert len(requests) == 5 + assert len(records) == 50 + assert len({record["id"] for record in records}) == 50 + assert {record["instruction"] for record in records} == {"同一问题"} + assert all(record["status"] == "valid" for record in records) + + +def test_generate_model_records_preserves_successful_batches_when_one_fails() -> None: + request_count = 0 + + def handler(_: httpx.Request) -> httpx.Response: + nonlocal request_count + request_count += 1 + if request_count == 2: + return httpx.Response(500) + return httpx.Response( + 200, + json={ + "choices": [ + { + "message": { + "content": json.dumps( + { + "items": [ + { + "instruction": f"问题 {index}", + "output": f"答案 {index}", + } + for index in range(1, 11) + ] + }, + ensure_ascii=False, + ) + } + } + ] + }, + ) + + records = generate_model_records( + [{"id": "preview-partial", "edited_content": "来源正文"}], + model={"name": "model", "api_url": "https://model.example/v1"}, + config={"generation_retries": 0}, + task_id="task-partial", + split={"train": 100, "validation": 0, "test": 0}, + qa_pairs_per_item=20, + client=httpx.Client(transport=httpx.MockTransport(handler)), + ) + + assert len(records) == 11 + assert sum(record["status"] == "valid" for record in records) == 10 + failed = next(record for record in records if record["status"] == "invalid") + assert "第 11-20 条" in failed["instruction"] + assert len({record["id"] for record in records}) == len(records) + + +def test_generate_model_records_retries_short_batch_then_marks_it_invalid() -> None: + request_count = 0 + + def handler(_: httpx.Request) -> httpx.Response: + nonlocal request_count + request_count += 1 + return httpx.Response( + 200, + json={ + "choices": [ + { + "message": { + "content": json.dumps( + { + "items": [ + { + "instruction": "只有一条", + "output": "不足本批要求数量", + } + ] + }, + ensure_ascii=False, + ) + } + } + ] + }, + ) + + records = generate_model_records( + [{"id": "preview-short", "edited_content": "来源正文"}], + model={"name": "model", "api_url": "https://model.example/v1"}, + config={"generation_retries": 1}, + task_id="task-short", + split={"train": 100, "validation": 0, "test": 0}, + qa_pairs_per_item=10, + client=httpx.Client(transport=httpx.MockTransport(handler)), + ) + + assert request_count == 2 + assert len(records) == 1 + assert records[0]["status"] == "invalid" + assert "expected 10, got 1" in records[0]["error"] + + +@pytest.mark.parametrize("qa_pairs_per_item", [0, 51]) +def test_generate_model_records_rejects_out_of_range_count( + qa_pairs_per_item: int, +) -> None: + with pytest.raises(ModelGenerationError, match=r"\[1, 50\]"): + generate_model_records( + [], + model={"name": "model", "api_url": "https://model.example/v1"}, + config={}, + task_id="task-invalid", + split={"train": 100, "validation": 0, "test": 0}, + qa_pairs_per_item=qa_pairs_per_item, + )