feat(data-process): 支持单项生成五十条数据

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:
+    if not 1 <= qa_pairs_per_item <= MAX_QA_PAIRS_PER_ITEM:
-        raise ValueError("qa_pairs_per_item must be in [1, 5]")
+        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}"

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"]

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:
+    if not 1 <= qa_pairs_per_item <= MAX_QA_PAIRS_PER_ITEM:
-        raise ModelGenerationError("qa_pairs_per_item must be in [1, 5]")
+        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] = {
+            for batch_offset in range(0, qa_pairs_per_item, MODEL_GENERATION_BATCH_SIZE):
-                "model": model_name,
+                batch_count = min(
-                "messages": _prompt_messages(
+                    MODEL_GENERATION_BATCH_SIZE,
-                    str(config.get("generation_prompt") or ""),
+                    qa_pairs_per_item - batch_offset,
-                    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",
-                    }
                 )
-                if on_progress:
+                batch_start = batch_offset + 1
-                    on_progress(item_index + 1, total_items)
+                batch_end = batch_offset + batch_count
-                continue
+                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]):
+                last_error: Exception | None = None
-                instruction = normalize_text(str(value.get("instruction") or value.get("question") or ""))
+                generated_items: list[Mapping[str, Any]] | None = None
-                input_text = normalize_text(str(value.get("input") or value.get("context") or ""))
+                for _ in range(retries + 1):
-                output = normalize_text(
+                    try:
-                    str(
+                        response = http_client.post(
-                        value.get("output")
+                            endpoint,
-                        or value.get("answer")
+                            headers=headers,
-                        or value.get("response")
+                            json=request_payload,
-                        or ""
+                        )
+                        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:

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:
+        if (
-            raise ValueError(f"{snake_name} must be an integer in [1, 5]")
+            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):

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)

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(

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,
+        )