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

feat(data-process): 支持思维链输出类型

Browse Source
This commit is contained in:
caoxiaozhu
2026-07-27 13:08:44 +08:00
parent ecafb7eb13
commit de2e8952b5
14 changed files with 443 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -556,12 +556,16 @@ def _run_generation(
if task["process_type"] == "unstructured"
else _value(config, "qa_pairs_per_row", "qaPairsPerRow", 1)
)
output_type = str(
_value(config, "output_type", "outputType", "standard")
).strip().lower()
if generation_model:
runtime_config = {
**config,
"generation_prompt": _value(
config, "generation_prompt", "generationPrompt", ""
),
"output_type": output_type,
"max_tokens": _value(config, "max_tokens", "maxTokens", 1024),
"json_mode": _value(config, "json_mode", "jsonMode", False),
}
@@ -583,6 +587,8 @@ def _run_generation(
qa_pairs_per_item=int(pairs or 1),
on_progress=report_progress,
)
elif output_type == "reasoning":
raise InvalidStateError("思维链输出必须配置可用的数据生成模型")
else:
generated = generate_standard_records(
preview_items,

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

@@ -22,6 +22,11 @@ class ModelGenerationError(ValueError):
"""模型配置、响应或调用失败。"""
OUTPUT_TYPE_STANDARD = "standard"
OUTPUT_TYPE_REASONING = "reasoning"
SUPPORTED_OUTPUT_TYPES = {OUTPUT_TYPE_STANDARD, OUTPUT_TYPE_REASONING}
def chat_completions_url(value: str) -> str:
"""把域名、基础 URL 或完整地址统一为 chat completions 地址。"""
@@ -52,7 +57,9 @@ def _message_content(payload: Mapping[str, Any]) -> str:
try:
content = payload["choices"][0]["message"]["content"]
except (KeyError, IndexError, TypeError) as exc:
raise ModelGenerationError("model response does not contain choices[0].message.content") from exc
raise ModelGenerationError(
"model response does not contain choices[0].message.content"
) from exc
if isinstance(content, str):
return content
if isinstance(content, list):
@@ -67,7 +74,14 @@ def _message_content(payload: Mapping[str, Any]) -> str:
def _json_payload(content: str) -> Any:
cleaned = re.sub(r"<think>[\s\S]*?</think>", "", content, flags=re.IGNORECASE).strip()
# 只移除模型在 JSON 之前自行输出的思考过程，不能破坏 JSON 字段中的训练内容。
cleaned = re.sub(
r"^\s*<think>[\s\S]*?</think>\s*",
"",
content,
count=1,
flags=re.IGNORECASE,
).strip()
fenced = re.fullmatch(r"```(?:json)?\s*([\s\S]*?)\s*```", cleaned, flags=re.IGNORECASE)
if fenced:
cleaned = fenced.group(1).strip()
@@ -107,19 +121,27 @@ def _prompt_messages(
*,
start_index: int,
total_count: int,
output_type: str,
) -> list[dict[str, str]]:
end_index = start_index + count - 1
if output_type == OUTPUT_TYPE_REASONING:
schema = '{"items":[{"instruction":"...","input":"...","reasoning":"...","answer":"..."}]}'
output_rule = (
"instruction、reasoning 和 answer 均不得为空；reasoning 必须是基于来源内容、"
"可核对且简洁的推理步骤，answer 只写最终答案。"
"这是思维链输出模式，即使其他提示语要求省略分析，也不得省略 reasoning。"
"不要自行添加 <think> 标签，系统会在保存时统一组装。"
)
else:
schema = '{"items":[{"instruction":"...","input":"...","output":"..."}]}'
output_rule = "instruction 和 output 不得为空，不要输出分析过程。"
schema_instruction = (
f"必须只返回 JSON 对象，格式为 {{\"items\":[{{\"instruction\":\"...\","
f"\"input\":\"...\",\"output\":\"...\"}}]}}；items 必须包含 {count} 条。"
f"必须只返回 JSON 对象，格式为 {schema}；items 必须包含 {count} 条。"
f"这是总计 {total_count} 条中的第 {start_index}-{end_index} 条，"
"各条必须使用不同的提问角度和表述，避免重复。"
"instruction 和 output 不得为空，不要输出 Markdown 代码围栏或分析过程。"
)
base_prompt = (
normalize_text(prompt)
or "请根据来源内容生成可用于监督微调的问答数据。"
f"{output_rule}不要输出 Markdown 代码围栏或 JSON 之外的说明。"
)
base_prompt = normalize_text(prompt) or "请根据来源内容生成可用于监督微调的问答数据。"
if "{{ content }}" in base_prompt:
user_prompt = base_prompt.replace("{{ content }}", content)
return [
@@ -150,9 +172,10 @@ def generate_model_records(
"""
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}]"
)
raise ModelGenerationError(f"qa_pairs_per_item must be in [1, {MAX_QA_PAIRS_PER_ITEM}]")
output_type = str(config.get("output_type") or OUTPUT_TYPE_STANDARD).strip().lower()
if output_type not in SUPPORTED_OUTPUT_TYPES:
raise ModelGenerationError(f"output_type must be one of {sorted(SUPPORTED_OUTPUT_TYPES)}")
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:
@@ -193,6 +216,7 @@ def generate_model_records(
batch_count,
start_index=batch_start,
total_count=qa_pairs_per_item,
output_type=output_type,
),
"temperature": temperature,
"max_tokens": max_tokens,
@@ -212,12 +236,8 @@ def generate_model_records(
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))
)
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: "
@@ -266,19 +286,56 @@ def generate_model_records(
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 ""
if output_type == OUTPUT_TYPE_REASONING:
reasoning = normalize_text(
re.sub(
r"</?think>",
"",
str(value.get("reasoning") or value.get("analysis") or ""),
flags=re.IGNORECASE,
)
)
)
answer = normalize_text(
re.sub(
r"</?think>",
"",
str(
value.get("answer")
or value.get("final_answer")
or value.get("output")
or ""
),
flags=re.IGNORECASE,
)
)
output = (
f"<think>\n{reasoning}\n</think>\n{answer}"
if reasoning and answer
else answer or (f"<think>\n{reasoning}\n</think>" if reasoning else "")
)
valid = bool(instruction and reasoning and answer)
missing_error = "model result is missing instruction, reasoning or answer"
else:
output = normalize_text(
str(
value.get("output")
or value.get("answer")
or value.get("response")
or ""
)
)
output = normalize_text(
re.sub(
r"<think>[\s\S]*?(?:</think>|$)",
"",
output,
flags=re.IGNORECASE,
)
)
valid = bool(instruction and output)
missing_error = "model result is missing instruction or output"
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)
result_id = f"result_{hashlib.sha256(raw_id.encode()).hexdigest()[:16]}"
results.append(
{
"id": result_id,
@@ -290,11 +347,7 @@ def generate_model_records(
"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"
),
"error": (None if valid else missing_error),
"split": "train",
}
)

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

@@ -2,6 +2,7 @@ from __future__ import annotations
import hashlib
import json
import re
import uuid
from collections.abc import Iterator, Sequence
from contextlib import contextmanager
@@ -87,6 +88,31 @@ def _json_value(value: Any, default: Any) -> Any:
return default
def _task_output_type(task: dict[str, Any]) -> str:
config = _json_value(task.get("config"), {})
if not isinstance(config, dict):
return "standard"
return str(config.get("output_type") or config.get("outputType") or "standard")
def _reasoning_output_is_valid(value: Any) -> bool:
match = re.fullmatch(
r"\s*<think>\s*(?P<reasoning>[\s\S]*?)\s*</think>\s*(?P<answer>[\s\S]+?)\s*",
str(value or ""),
flags=re.IGNORECASE,
)
return bool(
match
and match.group("reasoning").strip()
and match.group("answer").strip()
and all(
tag not in part.lower()
for tag in ("<think", "</think")
for part in (match.group("reasoning"), match.group("answer"))
)
)
def _preview_config_value(config: dict[str, Any], key: str, default: Any) -> Any:
if key in config:
return config[key]
@@ -1565,10 +1591,13 @@ class DataProcessStore:
raise ConflictError("data process result was modified by another request")
merged = {**current, **values}
quality = payload.get("quality_score") or {}
hard_valid = bool(
str(merged.get("instruction") or "").strip()
and str(merged.get("output") or "").strip()
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"
or _reasoning_output_is_valid(merged.get("output"))
)
hard_valid = instruction_valid and output_valid and reasoning_valid
quality_valid = bool(quality.get("is_valid", hard_valid))
changed = any(
str(merged.get(field) or "")
@@ -1580,8 +1609,16 @@ class DataProcessStore:
)
values["status"] = status
flags = quality.get("flags") if isinstance(quality, dict) else None
format_error = (
"思维链输出必须包含非空的 <think>...</think> 推理过程和最终答案"
if instruction_valid and output_valid and not reasoning_valid
else "Instruction 和 Output 不能为空"
if not instruction_valid or not output_valid
else None
)
values["error"] = ", ".join(str(flag) for flag in flags or []) or (
"quality validation failed" if status == "invalid" else None
format_error
or ("quality validation failed" if status == "invalid" else None)
)
values["updated_at"] = utcnow()
assignments = ", ".join(f"{key}=%s" for key in values)
@@ -1671,6 +1708,10 @@ class DataProcessStore:
if row["status"] == "invalid"
or not str(row.get("instruction") or "").strip()
or not str(row.get("output") or "").strip()
or (
_task_output_type(task) == "reasoning"
and not _reasoning_output_is_valid(row.get("output"))
)
)
if invalid_count:
raise InvalidStateError(f"task contains {invalid_count} invalid results")
@@ -1731,6 +1772,7 @@ class DataProcessStore:
"source": "data_process",
"storage_backend": "database",
"source_task_id": task_id,
"output_type": _task_output_type(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",

34
backend/tests/test_data_process_api.py
View File

@@ -1405,6 +1405,40 @@ def test_stale_generation_worker_cannot_overwrite_new_run(monkeypatch: Any) -> N
assert store.tasks[task_id]["status"] == "running"
def test_reasoning_output_requires_generation_model() -> None:
store = FakeDataProcessStore()
task = store.create_task(
{
"name": "思维链模型校验",
"process_type": "structured",
"config": {"output_type": "reasoning"},
}
)
task_id = task["id"]
store.replace_preview_items(
task_id,
[
{
"source_file_id": None,
"original_content": "需要推理的来源内容",
"edited_content": "需要推理的来源内容",
"status": "manual",
}
],
)
started = store.start_generation(task_id, replace_existing=True)
data_process_endpoint._run_generation(
store,
task_id,
started["generation_run_id"],
)
failed = store.get_task(task_id)
assert failed["status"] == "failed"
assert failed["failure_reason"] == "思维链输出必须配置可用的数据生成模型"
def test_result_status_cannot_be_forged_by_client(tmp_path: Path) -> None:
client, _, _ = make_client(tmp_path)
task_id = client.post(

125
backend/tests/test_data_process_generation.py
View File

@@ -86,6 +86,119 @@ def test_generate_model_records_uses_prompt_auth_and_stable_split() -> None:
assert progress_updates == [(1, 1)]
def test_generate_model_records_builds_reasoning_output_with_think_tags() -> None:
def handler(request: httpx.Request) -> httpx.Response:
payload = json.loads(request.content)
system_prompt = payload["messages"][0]["content"]
assert '"reasoning":"...","answer":"..."' in system_prompt
assert "系统会在保存时统一组装" in system_prompt
content = "<think>模型接口自己的分析</think>" + json.dumps(
{
"items": [
{
"instruction": "计算两项费用合计",
"input": "交通费 30 元，餐费 20 元",
"reasoning": "先识别两项费用，再计算 30 + 20。",
"answer": "合计 50 元。",
}
]
},
ensure_ascii=False,
)
return httpx.Response(
200,
json={"choices": [{"message": {"content": content}}]},
)
records = generate_model_records(
[{"id": "preview-reasoning", "edited_content": "交通费 30 元，餐费 20 元"}],
model={"name": "model", "api_url": "https://model.example/v1"},
config={"output_type": "reasoning"},
task_id="task-reasoning",
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]["output"] == (
"<think>\n先识别两项费用,再计算 30 + 20。\n</think>\n合计 50 元。"
)
def test_generate_model_records_marks_reasoning_without_reasoning_field_invalid() -> None:
response = {
"choices": [
{
"message": {
"content": json.dumps(
{
"items": [
{
"instruction": "问题",
"answer": "只有最终答案",
}
]
},
ensure_ascii=False,
)
}
}
]
}
client = httpx.Client(
transport=httpx.MockTransport(lambda _: httpx.Response(200, json=response))
)
records = generate_model_records(
[{"id": "preview-missing-reasoning", "edited_content": "来源正文"}],
model={"name": "model", "api_url": "https://model.example/v1"},
config={"output_type": "reasoning"},
task_id="task-missing-reasoning",
split={"train": 100, "validation": 0, "test": 0},
qa_pairs_per_item=1,
client=client,
)
assert records[0]["status"] == "invalid"
assert records[0]["output"] == "只有最终答案"
assert "reasoning" in records[0]["error"]
def test_standard_output_removes_model_think_block() -> None:
content = json.dumps(
{
"items": [
{
"instruction": "问题",
"output": "<think>不应保存的分析</think>最终答案",
}
]
},
ensure_ascii=False,
)
client = httpx.Client(
transport=httpx.MockTransport(
lambda _: httpx.Response(
200,
json={"choices": [{"message": {"content": content}}]},
)
)
)
records = generate_model_records(
[{"id": "preview-standard", "edited_content": "来源正文"}],
model={"name": "model", "api_url": "https://model.example/v1"},
config={"output_type": "standard"},
task_id="task-standard",
split={"train": 100, "validation": 0, "test": 0},
qa_pairs_per_item=1,
client=client,
)
assert records[0]["output"] == "最终答案"
def test_generate_model_records_keeps_partial_failure_for_manual_repair() -> None:
client = httpx.Client(
transport=httpx.MockTransport(
@@ -267,3 +380,15 @@ def test_generate_model_records_rejects_out_of_range_count(
split={"train": 100, "validation": 0, "test": 0},
qa_pairs_per_item=qa_pairs_per_item,
)
def test_generate_model_records_rejects_unknown_output_type() -> None:
with pytest.raises(ModelGenerationError, match="output_type"):
generate_model_records(
[],
model={"name": "model", "api_url": "https://model.example/v1"},
config={"output_type": "unknown"},
task_id="task-invalid-output",
split={"train": 100, "validation": 0, "test": 0},
qa_pairs_per_item=1,
)

71
backend/tests/test_data_process_store.py
View File

@@ -15,6 +15,7 @@ from app.modules.data_process.store import (
InvalidStateError,
_decode_row,
_preview_config_changed,
_reasoning_output_is_valid,
_source_storage_descriptor,
)
@@ -114,14 +115,25 @@ class _PublishConnection:
)
if normalized.startswith("INSERT INTO dataset_records"):
self.records.append(
{"dataset_id": params[1], "line_no": params[4], "split": params[5]}
{
"dataset_id": params[1],
"line_no": params[4],
"split": params[5],
"output": params[8],
"raw": json.loads(params[9]),
}
)
return _Result()
class _PublishStore(DataProcessStore):
def __init__(self, conn: _PublishConnection):
def __init__(
self,
conn: _PublishConnection,
task_config: dict[str, Any] | None = None,
):
self._conn = conn
self._task_config = task_config or {}
@contextmanager
def connect(self) -> Iterator[_PublishConnection]:
@@ -135,7 +147,7 @@ class _PublishStore(DataProcessStore):
"id": task_id,
"status": "completed",
"description": "",
"config": {},
"config": self._task_config,
"output_dataset_id": train_dataset and train_dataset["id"],
}
@@ -967,7 +979,11 @@ def test_publish_creates_three_independent_datasets_with_exact_counts() -> None:
"status": "valid",
"instruction": f"问题 {index}",
"input": "",
"output": f"答案 {index}",
"output": (
"<think>\n先读取制度条款。\n</think>\n答案 0"
if index == 0
else f"答案 {index}"
),
"preview_item_id": f"preview-{index}",
}
for index in range(28)
@@ -993,6 +1009,10 @@ def test_publish_creates_three_independent_datasets_with_exact_counts() -> None:
item["id"] for item in conn.datasets
}
assert len(conn.records) == 28
reasoning_record = next(
item for item in conn.records if item["output"].startswith("<think>")
)
assert reasoning_record["raw"]["output"] == reasoning_record["output"]
assert published["dataset"]["type"] == "train"
assert len(published["datasets"]) == 3
assert published["split_counts"] == {"train": 22, "validation": 3, "test": 3}
@@ -1078,6 +1098,49 @@ def test_publish_keeps_all_three_datasets_when_a_small_split_is_empty() -> None:
assert len(conn.files) == 3
@pytest.mark.parametrize(
("output", "expected"),
[
("<think>\n推理步骤\n</think>\n最终答案", True),
("<think></think>\n最终答案", False),
("<think>只有推理</think>", False),
("没有标签的最终答案", False),
("<think>外层<think>嵌套</think></think>答案", False),
],
)
def test_reasoning_output_validator_requires_one_complete_pair(
output: str,
expected: bool,
) -> None:
assert _reasoning_output_is_valid(output) is expected
def test_publish_rejects_invalid_reasoning_output_format() -> None:
conn = _PublishConnection(
[
{
"id": "result-reasoning-invalid",
"status": "valid",
"instruction": "需要推理的问题",
"input": "",
"output": "只有最终答案",
"preview_item_id": "preview-reasoning-invalid",
}
]
)
with pytest.raises(InvalidStateError, match="1 invalid results"):
_PublishStore(conn, {"output_type": "reasoning"}).publish(
"task-reasoning-invalid",
{
"dataset_name": "无效思维链",
"storage_type": "local",
"format": "alpaca_jsonl",
"split": {"train": 80, "validation": 10, "test": 10},
},
)
def test_source_storage_descriptor_accepts_owned_local_and_legacy_db_references() -> None:
task_id = "dpt_task"
source_file_id = "dpsf_source"