YG-Soft/YG_FT
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 新增外部数据源拉取与 DPO 输出格式支持

Browse Source 
- 支持从 PostgreSQL 数据库拉取结构化数据作为训练来源
- 新增 DPO (Direct Preference Optimization) 输出类型
- 支持 chosen/rejected 字段的编辑、校验和发布
- 完善数据预处理切分逻辑和元数据管理
- 移除 OCR 扫描 PDF 功能，保持基础文本解析能力

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
caoxiaozhu
2026-08-11 14:17:45 +08:00
parent f809825a7d
commit 5f6e7523cf
26 changed files with 927 additions and 144 deletions
Download Patch File Download Diff File Expand all files Collapse all files

65
backend/app/api/v1/endpoints/data_process.py
View File

@@ -16,7 +16,7 @@ from dataclasses import asdict
from pathlib import Path
from threading import BoundedSemaphore, Lock
from typing import Any, Literal
from urllib.parse import quote, urlsplit
from urllib.parse import parse_qs, quote, urlsplit
import httpx
import psycopg
@@ -676,8 +676,10 @@ def _run_generation(
qa_pairs_per_item=int(pairs or 1),
on_progress=report_progress,
)
elif output_type == "reasoning":
raise InvalidStateError("思维链输出必须配置可用的数据生成模型")
elif output_type in {"reasoning", "dpo"}:
if output_type == "reasoning":
raise InvalidStateError("思维链输出必须配置可用的数据生成模型")
raise InvalidStateError("DPO 输出必须配置可用的数据生成模型")
else:
generated = generate_standard_records(
preview_items,
@@ -1077,7 +1079,10 @@ async def upload_source_files(
with api_errors():
task = store.get_task(task_id)
process_type = str(task["process_type"])
if process_type == "external":
source_mode = str(
_value(task.get("config") or {}, "source_mode", "sourceMode", "local")
)
if process_type == "external" or source_mode == "external":
raise InvalidStateError(
"external tasks must import data through the external source endpoint"
)
@@ -1375,6 +1380,10 @@ def _external_postgres_connection(payload: ExternalSourceRequest) -> psycopg.Con
raise fail(501, f"external data source type is not supported: {payload.type}")
if parsed_url.username or parsed_url.password:
raise fail(400, "database credentials must use the account and password fields")
if set(parse_qs(parsed_url.query)) & {
"password", "secret", "token", "api_key", "user", "username"
}:
raise fail(400, "database URL query must not contain credentials")
if payload.auth_mode not in {"none", "basic"}:
raise fail(400, "PostgreSQL supports only none or basic authentication")
if payload.auth_mode == "basic" and not payload.username:
@@ -1413,10 +1422,14 @@ def _external_postgres_connection(payload: ExternalSourceRequest) -> psycopg.Con
"set DATA_PROCESS_ALLOW_PRIVATE_EXTERNAL_DB=true only in a trusted deployment",
)
kwargs: dict[str, Any] = {
"connect_timeout": 5,
"connect_timeout": payload.connect_timeout_seconds,
"row_factory": dict_row,
"application_name": "yg-ft-data-process-readonly",
"options": "-c default_transaction_read_only=on -c statement_timeout=30000",
"options": (
"-c default_transaction_read_only=on "
f"-c statement_timeout={payload.statement_timeout_seconds * 1000}"
),
"sslmode": payload.ssl_mode,
}
if payload.auth_mode == "basic" and payload.username:
kwargs["user"] = payload.username
@@ -1425,6 +1438,17 @@ def _external_postgres_connection(payload: ExternalSourceRequest) -> psycopg.Con
return psycopg.connect(payload.url, **kwargs)
def _assert_external_source_task(task: dict[str, Any]) -> None:
if str(task.get("process_type")) == "external":
return
config = task.get("config") or {}
source_mode = str(_value(config, "source_mode", "sourceMode", "local"))
if str(task.get("process_type")) != "structured" or source_mode != "external":
raise InvalidStateError(
"external source access requires a structured task with source_mode=external"
)
@router.post("/{task_id}/external/test")
def test_external_source(
task_id: str,
@@ -1433,10 +1457,7 @@ def test_external_source(
) -> dict[str, Any]:
with api_errors():
task = store.get_task(task_id)
if str(task.get("process_type")) != "external":
raise InvalidStateError(
"external source access requires an external data processing task"
)
_assert_external_source_task(task)
try:
with _external_postgres_connection(payload) as conn:
conn.execute("SELECT 1 AS ok").fetchone()
@@ -1462,12 +1483,14 @@ def pull_external_source(
raise fail(400, "a read-only SELECT or WITH query is required for external pull")
with api_errors():
task = store.get_task(task_id)
if str(task.get("process_type")) != "external":
raise InvalidStateError("external pull requires an external data processing task")
_assert_external_source_task(task)
try:
with _external_postgres_connection(payload) as conn:
conn.execute("SET TRANSACTION READ ONLY")
conn.execute("SET LOCAL statement_timeout = '30s'")
conn.execute(
"SELECT set_config('statement_timeout', %s, true)",
(f"{payload.statement_timeout_seconds}s",),
)
cursor = conn.execute(query)
rows: list[dict[str, Any]] = []
content_parts: list[str] = []
@@ -1518,6 +1541,9 @@ def pull_external_source(
"external_type": payload.type,
"external_host": urlsplit(payload.url).hostname,
"external_limit": payload.limit,
"external_ssl_mode": payload.ssl_mode,
"external_connect_timeout_seconds": payload.connect_timeout_seconds,
"external_statement_timeout_seconds": payload.statement_timeout_seconds,
},
}
],
@@ -1566,7 +1592,10 @@ def _prepare_preview_items(
for index, source in enumerate(sources):
source_format = str(source.get("file_format") or "").lower()
needs_structured_xlsx = not is_unstructured and source_format == "xlsx"
needs_layout_raw = is_unstructured and chunk_method == "layout_hybrid"
needs_layout_raw = (
is_unstructured
and chunk_method == "layout_hybrid"
)
needs_pdf_noise = (
is_unstructured
and not needs_layout_raw
@@ -2038,6 +2067,8 @@ def restore_result(
"instruction": current.get("original_instruction") or current.get("instruction") or "",
"input": current.get("original_input") or current.get("input") or "",
"output": current.get("original_output") or current.get("output") or "",
"chosen": current.get("original_chosen") or current.get("chosen") or "",
"rejected": current.get("original_rejected") or current.get("rejected") or "",
}
preview_id = current.get("preview_item_id")
source_content = ""
@@ -2070,6 +2101,8 @@ def restore_result(
"instruction": restored["instruction"],
"input": restored["input"],
"output": restored["output"],
"chosen": restored["chosen"],
"rejected": restored["rejected"],
"quality_score": asdict(quality),
"expected_updated_at": current.get("updated_at"),
},
@@ -2141,13 +2174,15 @@ def _generate_result_replacement(
).strip().lower()
previous_instruction = str(current.get("instruction") or "")[:1000]
previous_output = str(current.get("output") or "")[:1000]
previous_rejected = str(current.get("rejected") or "")[:1000]
base_prompt = str(
_value(config, "generation_prompt", "generationPrompt", "") or ""
)
regeneration_instruction = (
"这是一次失败结果的重新生成。请使用新的提问角度和表达，"
"不要复述旧结果。旧问题："
f"{previous_instruction or '无'}；旧答案：{previous_output or '无'}。"
f"{previous_instruction or '无'}；旧优选答案：{previous_output or '无'}；"
f"旧拒选答案：{previous_rejected or '无'}。"
)
runtime_config = {
**config,

9
backend/app/db/sql/002_data_process.sql
View File

@@ -268,9 +268,13 @@ CREATE TABLE IF NOT EXISTS data_process_results (
instruction TEXT NOT NULL,
input TEXT NOT NULL DEFAULT '',
output TEXT NOT NULL,
chosen TEXT NOT NULL DEFAULT '',
rejected TEXT NOT NULL DEFAULT '',
original_instruction TEXT,
original_input TEXT,
original_output TEXT,
original_chosen TEXT,
original_rejected TEXT,
status VARCHAR(20) NOT NULL DEFAULT 'valid'
CHECK (status IN ('valid', 'modified', 'invalid')),
error TEXT,
@@ -280,6 +284,11 @@ CREATE TABLE IF NOT EXISTS data_process_results (
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
ALTER TABLE data_process_results ADD COLUMN IF NOT EXISTS chosen TEXT NOT NULL DEFAULT '';
ALTER TABLE data_process_results ADD COLUMN IF NOT EXISTS rejected TEXT NOT NULL DEFAULT '';
ALTER TABLE data_process_results ADD COLUMN IF NOT EXISTS original_chosen TEXT;
ALTER TABLE data_process_results ADD COLUMN IF NOT EXISTS original_rejected TEXT;
CREATE INDEX IF NOT EXISTS idx_data_process_results_task_status
ON data_process_results(task_id, status, id);
CREATE INDEX IF NOT EXISTS idx_data_process_results_task_split

57
backend/app/modules/data_process/document_chunking.py
View File

@@ -57,7 +57,62 @@ def _sentence_chunks(text: str) -> list[str]:
@lru_cache(maxsize=1)
def _tokenizer() -> tiktoken.Encoding:
return tiktoken.get_encoding("cl100k_base")
"""加载 cl100k_base 编码器，优先在线下载，失败时使用本地缓存以支持离线环境。"""
import os
import base64
# 先设置缓存目录环境变量
offline_cache = os.path.expanduser("~/.cache/tiktoken")
os.environ.setdefault("TIKTOKEN_CACHE_DIR", offline_cache)
try:
# 尝试标准方式加载
return tiktoken.get_encoding("cl100k_base")
except Exception:
# 如果失败，尝试手动从本地文件构造
try:
from pathlib import Path
local_file = Path(offline_cache) / "9b5ad71b2ce5302211f9c61530b329a4922fc6a4"
if not local_file.exists():
# 尝试另一个可能的文件名
local_file = Path(offline_cache) / "cl100k_base.tiktoken"
if local_file.exists():
# 读取 BPE 文件内容
with open(local_file, "rb") as f:
contents = f.read()
# 解析 BPE 文件
mergeable_ranks = {}
for line in contents.splitlines():
if line:
token, rank = line.split()
mergeable_ranks[base64.b64decode(token)] = int(rank)
# 构造 Encoding 对象
import tiktoken.core
return tiktoken.core.Encoding(
name="cl100k_base",
pat_str=r"""'(?i:[sdmt]|ll|ve|re)|[^\r\n\p{L}\p{N}]?+\p{L}+|\p{N}{1,3}| ?[^\s\p{L}\p{N}]++[\r\n]*|\s*[\r\n]|\s+(?!\S)|\s+""",
mergeable_ranks=mergeable_ranks,
special_tokens={
"<|endoftext|>": 100257,
"<|fim_prefix|>": 100258,
"<|fim_middle|>": 100259,
"<|fim_suffix|>": 100260,
"<|endofprompt|>": 100276,
},
)
except Exception:
pass
raise RuntimeError(
f"无法加载 cl100k_base 编码器\n"
f"请确保以下任一条件满足:\n"
f"1. 服务器可以访问网络\n"
f"2. 本地存在缓存文件: {offline_cache}/9b5ad71b2ce5302211f9c61530b329a4922fc6a4"
)
def _text_chunks(

BIN
backend/app/modules/data_process/document_chunking.txt Normal file
View File

Binary file not shown.

64
backend/app/modules/data_process/generation.py
View File

@@ -29,7 +29,12 @@ class _TerminalModelGenerationError(ModelGenerationError):
OUTPUT_TYPE_STANDARD = "standard"
OUTPUT_TYPE_REASONING = "reasoning"
SUPPORTED_OUTPUT_TYPES = {OUTPUT_TYPE_STANDARD, OUTPUT_TYPE_REASONING}
OUTPUT_TYPE_DPO = "dpo"
SUPPORTED_OUTPUT_TYPES = {
OUTPUT_TYPE_STANDARD,
OUTPUT_TYPE_REASONING,
OUTPUT_TYPE_DPO,
}
REASONING_DETAIL_NORMAL = "normal"
REASONING_DETAIL_DETAILED = "detailed"
SUPPORTED_REASONING_DETAILS = {
@@ -285,6 +290,18 @@ def _prompt_messages(
"这是思维链输出模式，即使其他提示语要求省略分析，也不得省略 reasoning。"
"不要自行添加 <think> 标签，系统会在保存时统一组装。"
)
elif output_type == OUTPUT_TYPE_DPO:
schema = (
'{"items":[{"instruction":"...","input":"...",'
'"chosen":"...","rejected":"..."}]}'
)
output_rule = (
"你正在生成用于直接偏好优化（DPO）的成对偏好数据。"
"instruction、chosen 和 rejected 均不得为空；chosen 必须是忠于来源、"
"准确完整的优选回答，rejected 必须是表面合理但存在明确质量缺陷的拒选回答。"
"两者不得相同；rejected 不得包含违法危险内容，也不得用空白、乱码或无关文本凑数。"
"不要输出分析过程或 <think> 标签。"
)
else:
schema = '{"items":[{"instruction":"...","input":"...","output":"..."}]}'
output_rule = (
@@ -461,9 +478,13 @@ def generate_model_records(
"instruction": failure_instruction,
"input": content,
"output": "",
"chosen": "",
"rejected": "",
"original_instruction": failure_instruction,
"original_input": content,
"original_output": "",
"original_chosen": "",
"original_rejected": "",
"status": "invalid",
"error": error_message,
"split": "train",
@@ -479,6 +500,8 @@ def generate_model_records(
input_text = normalize_text(
str(value.get("input") or value.get("context") or "")
)
chosen = ""
rejected = ""
if output_type == OUTPUT_TYPE_REASONING:
reasoning = normalize_text(
re.sub(
@@ -508,6 +531,36 @@ def generate_model_records(
)
valid = bool(instruction and reasoning and answer)
missing_error = "model result is missing instruction, reasoning or answer"
elif output_type == OUTPUT_TYPE_DPO:
chosen = normalize_text(str(value.get("chosen") or ""))
rejected = normalize_text(str(value.get("rejected") or ""))
chosen = normalize_text(
re.sub(
r"<think>[\s\S]*?(?:</think>|$)",
"",
chosen,
flags=re.IGNORECASE,
)
)
rejected = normalize_text(
re.sub(
r"<think>[\s\S]*?(?:</think>|$)",
"",
rejected,
flags=re.IGNORECASE,
)
)
output = chosen
valid = bool(
instruction
and chosen
and rejected
and chosen.strip() != rejected.strip()
)
missing_error = (
"model result is missing instruction, chosen or rejected, "
"or chosen equals rejected"
)
else:
output = normalize_text(
str(
@@ -527,7 +580,10 @@ def generate_model_records(
)
valid = bool(instruction and output)
missing_error = "model result is missing instruction or output"
raw_id = f"{preview_id}:{variant_index + 1}:{instruction}:{output}"
raw_id = (
f"{preview_id}:{variant_index + 1}:{instruction}:"
f"{output}:{rejected}"
)
result_id = f"result_{hashlib.sha256(raw_id.encode()).hexdigest()[:16]}"
results.append(
{
@@ -536,9 +592,13 @@ def generate_model_records(
"instruction": instruction,
"input": input_text,
"output": output,
"chosen": chosen,
"rejected": rejected,
"original_instruction": instruction,
"original_input": input_text,
"original_output": output,
"original_chosen": chosen,
"original_rejected": rejected,
"status": "valid" if valid else "invalid",
"error": (None if valid else missing_error),
"split": "train",

112
backend/app/modules/data_process/store.py
View File

@@ -140,6 +140,12 @@ def _reasoning_output_is_valid(value: Any) -> bool:
)
def _dpo_fields_are_valid(row: dict[str, Any]) -> bool:
chosen = str(row.get("chosen") or "").strip()
rejected = str(row.get("rejected") or "").strip()
return bool(chosen and rejected and chosen != rejected)
def _preview_config_value(config: dict[str, Any], key: str, default: Any) -> Any:
if key in config:
return config[key]
@@ -846,7 +852,11 @@ class DataProcessStore:
)
instruction = str(record.get("instruction") or raw.get("instruction") or "")
input_text = str(record.get("input") or raw.get("input") or "")
output = str(record.get("output") or raw.get("output") or "")
chosen = str(raw.get("chosen") or "")
rejected = str(raw.get("rejected") or "")
output = str(
record.get("output") or raw.get("output") or chosen or ""
)
split = str(record.get("split") or raw.get("split") or "") or None
status = str(record.get("status") or "valid")
if status not in {"valid", "modified", "invalid"}:
@@ -856,10 +866,11 @@ class DataProcessStore:
"""
INSERT INTO data_process_results
(id, task_id, preview_item_id, instruction, input, output,
original_instruction, original_input, original_output, status,
chosen, rejected, original_instruction, original_input,
original_output, original_chosen, original_rejected, status,
error, split, quality_score, created_at, updated_at)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s,
NULL, %s, '{}', %s, %s)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s,
%s, %s, NULL, %s, '{}', %s, %s)
""",
(
result_id,
@@ -868,9 +879,13 @@ class DataProcessStore:
instruction,
input_text,
output,
chosen,
rejected,
instruction,
input_text,
output,
chosen,
rejected,
status,
split,
created_at,
@@ -2024,9 +2039,11 @@ class DataProcessStore:
"""
INSERT INTO data_process_results
(id, task_id, preview_item_id, instruction, input, output,
original_instruction, original_input, original_output, status, error,
chosen, rejected, original_instruction, original_input,
original_output, original_chosen, original_rejected, status, error,
split, quality_score, created_at, updated_at)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s,
%s, %s, %s, %s, %s, %s, %s)
""",
(
result.get("id") or new_id("dpr"),
@@ -2035,9 +2052,13 @@ class DataProcessStore:
result.get("instruction") or "",
result.get("input") or "",
result.get("output") or "",
result.get("chosen") or "",
result.get("rejected") or "",
result.get("original_instruction", result.get("instruction") or ""),
result.get("original_input", result.get("input") or ""),
result.get("original_output", result.get("output") or ""),
result.get("original_chosen", result.get("chosen") or ""),
result.get("original_rejected", result.get("rejected") or ""),
result.get("status") or "valid",
result.get("error"),
result.get("split"),
@@ -2121,7 +2142,7 @@ class DataProcessStore:
rows = conn.execute(
"""
SELECT status, instruction, output
SELECT status, instruction, output, chosen, rejected
FROM data_process_results
WHERE task_id=%s
""",
@@ -2139,6 +2160,10 @@ class DataProcessStore:
_task_output_type(task) == "reasoning"
and not _reasoning_output_is_valid(row.get("output"))
)
or (
_task_output_type(task) == "dpo"
and not _dpo_fields_are_valid(row)
)
)
if invalid_count:
raise InvalidStateError(
@@ -2210,7 +2235,9 @@ class DataProcessStore:
def update_result(
self, task_id: str, result_id: str, payload: dict[str, Any]
) -> dict[str, Any]:
allowed = {"instruction", "input", "output", "quality_score"}
allowed = {
"instruction", "input", "output", "chosen", "rejected", "quality_score"
}
values = {key: value for key, value in payload.items() if key in allowed}
if "quality_score" in values:
values["quality_score"] = json_dumps(values["quality_score"])
@@ -2232,20 +2259,28 @@ class DataProcessStore:
current_updated_at = _serialize_value(current.get("updated_at"))
if expected_updated_at and expected_updated_at != current_updated_at:
raise ConflictError("data process result was modified by another request")
output_type = _task_output_type(task)
if output_type == "dpo" and "chosen" in values:
values["output"] = values["chosen"]
merged = {**current, **values}
quality = payload.get("quality_score") or {}
instruction_valid = bool(str(merged.get("instruction") or "").strip())
output_valid = bool(str(merged.get("output") or "").strip())
reasoning_valid = (
_task_output_type(task) != "reasoning"
output_type != "reasoning"
or _reasoning_output_is_valid(merged.get("output"))
)
hard_valid = instruction_valid and output_valid and reasoning_valid
dpo_valid = output_type != "dpo" or _dpo_fields_are_valid(merged)
hard_valid = instruction_valid and output_valid and reasoning_valid and dpo_valid
quality_valid = bool(quality.get("is_valid", hard_valid))
changed = any(
str(merged.get(field) or "")
!= str(merged.get(f"original_{field}") or "")
for field in ("instruction", "input", "output")
for field in (
("instruction", "input", "chosen", "rejected")
if output_type == "dpo"
else ("instruction", "input", "output")
)
)
status = "invalid" if not hard_valid or not quality_valid else (
"modified" if changed else "valid"
@@ -2255,6 +2290,8 @@ class DataProcessStore:
format_error = (
"思维链输出必须包含非空的 <think>...</think> 推理过程和最终答案"
if instruction_valid and output_valid and not reasoning_valid
else "DPO 输出必须包含不同的非空 Chosen 和 Rejected 回答"
if instruction_valid and not dpo_valid
else "Instruction 和 Output 不能为空"
if not instruction_valid or not output_valid
else None
@@ -2318,18 +2355,23 @@ class DataProcessStore:
instruction = str(replacement.get("instruction") or "").strip()
input_text = str(replacement.get("input") or "").strip()
output = str(replacement.get("output") or "").strip()
chosen = str(replacement.get("chosen") or "").strip()
rejected = str(replacement.get("rejected") or "").strip()
quality_score = replacement.get("quality_score") or {}
if not instruction or not output or not bool(quality_score.get("is_valid")):
raise InvalidStateError("regenerated result did not pass quality validation")
if _task_output_type(task) == "reasoning" and not _reasoning_output_is_valid(output):
raise InvalidStateError("regenerated reasoning result has an invalid output format")
if _task_output_type(task) == "dpo" and not _dpo_fields_are_valid(replacement):
raise InvalidStateError("regenerated DPO result has invalid preference fields")
now = utcnow()
row = conn.execute(
"""
UPDATE data_process_results
SET instruction=%s, input=%s, output=%s,
SET instruction=%s, input=%s, output=%s, chosen=%s, rejected=%s,
original_instruction=%s, original_input=%s, original_output=%s,
original_chosen=%s, original_rejected=%s,
status='valid', error=NULL, quality_score=%s, updated_at=%s
WHERE id=%s AND task_id=%s
RETURNING *
@@ -2338,9 +2380,13 @@ class DataProcessStore:
instruction,
input_text,
output,
chosen,
rejected,
instruction,
input_text,
output,
chosen,
rejected,
json_dumps(quality_score),
now,
result_id,
@@ -2434,6 +2480,10 @@ class DataProcessStore:
_task_output_type(task) == "reasoning"
and not _reasoning_output_is_valid(row.get("output"))
)
or (
_task_output_type(task) == "dpo"
and not _dpo_fields_are_valid(row)
)
)
if invalid_count:
raise InvalidStateError(f"task contains {invalid_count} invalid results")
@@ -2449,15 +2499,27 @@ class DataProcessStore:
requested_split,
seed=task_id,
)
records = [
{
"instruction": row["instruction"],
"input": row["input"],
"output": row["output"],
"split": assignment,
}
for row, assignment in zip(rows, assignments, strict=True)
]
if _task_output_type(task) == "dpo":
records = [
{
"instruction": row["instruction"],
"input": row["input"],
"chosen": row["chosen"],
"rejected": row["rejected"],
"split": assignment,
}
for row, assignment in zip(rows, assignments, strict=True)
]
else:
records = [
{
"instruction": row["instruction"],
"input": row["input"],
"output": row["output"],
"split": assignment,
}
for row, assignment in zip(rows, assignments, strict=True)
]
split_order = ("train", "validation", "test")
split_counts = {
split_name: assignments.count(split_name) for split_name in split_order
@@ -2498,7 +2560,11 @@ class DataProcessStore:
"reasoning_detail": _task_reasoning_detail(task),
"source_file_ids": [item["id"] for item in self._source_ids(conn, task_id)],
"source_result_ids": source_result_ids,
"format": payload.get("format") or "alpaca_jsonl",
"format": (
"dpo"
if _task_output_type(task) == "dpo"
else payload.get("format") or "alpaca_jsonl"
),
"split": requested_split,
}
@@ -2741,7 +2807,7 @@ class DataProcessStore:
record["split"],
record["instruction"],
record["input"],
record["output"],
record.get("output") or record.get("chosen") or "",
json_dumps(
{
**record,

33
backend/app/schemas/data_process.py
View File

@@ -2,6 +2,7 @@ from __future__ import annotations
from enum import StrEnum
from typing import Any, Literal
from urllib.parse import parse_qs, urlsplit
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
@@ -15,6 +16,31 @@ def _config_value(config: dict[str, Any], snake_name: str, camel_name: str, defa
def _validate_process_config(config: dict[str, Any]) -> None:
output_type = _config_value(config, "output_type", "outputType", "standard")
if output_type not in {"standard", "reasoning", "dpo"}:
raise ValueError("output_type must be one of: standard, reasoning, dpo")
source_mode = _config_value(config, "source_mode", "sourceMode", "local")
if source_mode not in {"local", "external"}:
raise ValueError("source_mode must be one of: local, external")
external_source = _config_value(config, "external_source", "externalSource", None)
if external_source is not None:
if not isinstance(external_source, dict):
raise ValueError("external_source must be an object")
if any(
key.lower() in {"password", "secret", "token", "api_key"}
for key in external_source
):
raise ValueError("external_source must not persist credentials")
external_url = str(external_source.get("url") or "").strip()
if external_url:
parsed_external_url = urlsplit(external_url)
sensitive_query_keys = {"password", "secret", "token", "api_key", "user", "username"}
if parsed_external_url.username or parsed_external_url.password or (
set(parse_qs(parsed_external_url.query)) & sensitive_query_keys
):
raise ValueError("external_source URL must not contain credentials")
chunk_method = _config_value(config, "chunk_method", "chunkMethod", "layout_hybrid")
if not isinstance(chunk_method, str) or chunk_method not in {
"layout_hybrid",
@@ -303,6 +329,9 @@ class ExternalSourceRequest(BaseModel):
username: str | None = Field(default=None, max_length=150)
password: str | None = Field(default=None, max_length=500)
limit: int = Field(default=1000, ge=1, le=100_000)
connect_timeout_seconds: int = Field(default=5, ge=1, le=30)
statement_timeout_seconds: int = Field(default=30, ge=1, le=300)
ssl_mode: Literal["disable", "prefer", "require", "verify-ca", "verify-full"] = "prefer"
class ExternalPullRequest(ExternalSourceRequest):
@@ -324,6 +353,8 @@ class ResultUpdate(BaseModel):
instruction: str | None = None
input: str | None = None
output: str | None = None
chosen: str | None = None
rejected: str | None = None
expected_updated_at: str | None = None
@@ -374,7 +405,7 @@ class PublishRequest(BaseModel):
dataset_type: Literal["train", "test", "eval", "val", "other"] = "train"
storage_type: Literal["local"] = "local"
split: DatasetSplit = Field(default_factory=DatasetSplit)
format: Literal["alpaca_jsonl", "jsonl"] = "alpaca_jsonl"
format: Literal["alpaca_jsonl", "jsonl", "dpo"] = "alpaca_jsonl"
description: str = ""
@field_validator("dataset_name")

37
backend/tests/test_data_process_api.py
View File

@@ -1840,6 +1840,43 @@ def test_external_source_never_returns_fake_success(tmp_path: Path) -> None:
assert response.json()["detail"]["code"] == 501
def test_external_source_mode_belongs_to_step_three_structured_task(tmp_path: Path) -> None:
client, _, _ = make_client(tmp_path)
local_task_id = client.post(
"/modelTF/data-process",
json={
"name": "本地结构化任务",
"process_type": "structured",
"config": {"source_mode": "local"},
},
).json()["data"]["id"]
rejected = client.post(
f"/modelTF/data-process/{local_task_id}/external/test",
json={"type": "mysql", "url": "mysql://db.example/test"},
)
assert rejected.status_code == 409
external_task_id = client.post(
"/modelTF/data-process",
json={
"name": "外部结构化任务",
"process_type": "structured",
"config": {"source_mode": "external"},
},
).json()["data"]["id"]
accepted_as_external = client.post(
f"/modelTF/data-process/{external_task_id}/external/test",
json={"type": "mysql", "url": "mysql://db.example/test"},
)
assert accepted_as_external.status_code == 501
local_upload = client.post(
f"/modelTF/data-process/{external_task_id}/source-files",
files={"files": ("records.jsonl", b'{"id":1}\n', "application/jsonl")},
)
assert local_upload.status_code == 409
def test_regenerate_endpoint_prepares_an_existing_published_task(tmp_path: Path) -> None:
client, store, _ = make_client(tmp_path)
task_id = client.post(

68
backend/tests/test_data_process_generation.py
View File

@@ -88,6 +88,74 @@ def test_generate_model_records_uses_prompt_auth_and_stable_split() -> None:
assert progress_updates == [(1, 1)]
def test_generate_model_records_builds_native_dpo_pair() -> None:
def handler(request: httpx.Request) -> httpx.Response:
payload = json.loads(request.content)
system_prompt = payload["messages"][0]["content"]
assert '"chosen"' in system_prompt
assert '"rejected"' in system_prompt
assert "直接偏好优化" in system_prompt
return httpx.Response(
200,
json={
"choices": [{
"message": {
"content": json.dumps({
"items": [{
"instruction": "系统如何处理扫描 PDF？",
"input": "",
"chosen": "仅在没有文本层时调用 OCR，并保留页码。",
"rejected": "所有 PDF 都重复执行 OCR。",
}],
}, ensure_ascii=False),
},
}],
},
)
records = generate_model_records(
[{"id": "preview-dpo", "edited_content": "扫描 PDF 缺少文本层时执行 OCR。"}],
model={"name": "model", "api_url": "https://model.example/v1"},
config={"output_type": "dpo", "generation_retries": 0},
task_id="task-dpo",
split={"train": 100, "validation": 0, "test": 0},
qa_pairs_per_item=1,
client=httpx.Client(transport=httpx.MockTransport(handler)),
)
assert records[0]["status"] == "valid"
assert records[0]["chosen"] == "仅在没有文本层时调用 OCR,并保留页码。"
assert records[0]["rejected"] == "所有 PDF 都重复执行 OCR。"
assert records[0]["output"] == records[0]["chosen"]
def test_generate_model_records_rejects_equal_dpo_pair() -> None:
def handler(_request: httpx.Request) -> httpx.Response:
return httpx.Response(
200,
json={"choices": [{"message": {"content": json.dumps({
"items": [{
"instruction": "问题",
"chosen": "相同回答",
"rejected": "相同回答",
}],
}, ensure_ascii=False)}}]},
)
records = generate_model_records(
[{"id": "preview-dpo-invalid", "edited_content": "来源"}],
model={"name": "model", "api_url": "https://model.example/v1"},
config={"output_type": "dpo", "generation_retries": 0},
task_id="task-dpo-invalid",
split={"train": 100, "validation": 0, "test": 0},
qa_pairs_per_item=1,
client=httpx.Client(transport=httpx.MockTransport(handler)),
)
assert records[0]["status"] == "invalid"
assert "chosen equals rejected" in records[0]["error"]
def test_minimax_m3_uses_split_reasoning_and_completion_token_budget() -> None:
requests: list[dict[str, object]] = []

4
backend/tests/test_data_process_migration.py
View File

@@ -37,6 +37,10 @@ def test_runtime_migration_fails_fast_on_incompatible_schema() -> None:
for value in ("idle", "queued", "running", "completed", "failed", "cancelled"):
assert f"'{value}'" in sql
assert "CREATE TABLE IF NOT EXISTS data_process_results" in sql
assert "ADD COLUMN IF NOT EXISTS chosen TEXT NOT NULL DEFAULT ''" in sql
assert "ADD COLUMN IF NOT EXISTS rejected TEXT NOT NULL DEFAULT ''" in sql
assert "ADD COLUMN IF NOT EXISTS original_chosen TEXT" in sql
assert "ADD COLUMN IF NOT EXISTS original_rejected TEXT" in sql
assert sql.count("BEGIN;") == 1
assert sql.rstrip().endswith("COMMIT;")

33
backend/tests/test_data_process_store.py
View File

@@ -1335,6 +1335,39 @@ def test_publish_rejects_invalid_reasoning_output_format() -> None:
)
def test_publish_dpo_writes_chosen_and_rejected_jsonl() -> None:
conn = _PublishConnection(
[
{
"id": "result-dpo",
"status": "valid",
"instruction": "如何处理扫描 PDF？",
"input": "",
"output": "仅在无文本层时执行 OCR。",
"chosen": "仅在无文本层时执行 OCR。",
"rejected": "所有 PDF 都执行 OCR。",
"preview_item_id": "preview-dpo",
}
]
)
published = _PublishStore(conn, {"output_type": "dpo"}).publish(
"task-dpo",
{
"dataset_name": "偏好数据",
"storage_type": "local",
"format": "dpo",
"split": {"train": 100, "validation": 0, "test": 0},
},
)
record = conn.records[0]["raw"]
assert record["chosen"] == "仅在无文本层时执行 OCR。"
assert record["rejected"] == "所有 PDF 都执行 OCR。"
assert "output" not in record
assert published["datasets"][0]["metadata"]["format"] == "dpo"
def test_source_storage_descriptor_accepts_owned_local_and_legacy_db_references() -> None:
task_id = "dpt_task"
source_file_id = "dpsf_source"