from __future__ import annotations import json import httpx import pytest 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: assert chat_completions_url("www.caoxiaozhu.com") == ( "https://www.caoxiaozhu.com/v1/chat/completions" ) assert chat_completions_url("https://model.example/v1") == ( "https://model.example/v1/chat/completions" ) complete = "https://model.example/openai/v1/chat/completions" assert chat_completions_url(complete) == complete def test_generate_model_records_uses_prompt_auth_and_stable_split() -> None: requests: list[httpx.Request] = [] progress_updates: list[tuple[int, int]] = [] def handler(request: httpx.Request) -> httpx.Response: requests.append(request) payload = json.loads(request.content) assert payload["model"] == "qwen-plus" assert payload["response_format"] == {"type": "json_object"} assert "客户反馈页面加载慢" in payload["messages"][1]["content"] assert "你正在生成标准监督微调问答数据" in payload["messages"][0]["content"] assert "禁止输出分析、推理过程" in payload["messages"][0]["content"] return httpx.Response( 200, json={ "choices": [ { "message": { "content": json.dumps( { "items": [ { "instruction": "请生成简洁客服回复", "input": "客户反馈页面加载慢", "output": "已收到反馈,我们正在排查。", } ] }, ensure_ascii=False, ) } } ] }, ) client = httpx.Client(transport=httpx.MockTransport(handler)) records = generate_model_records( [{"id": "preview-1", "edited_content": "客户反馈页面加载慢"}], model={ "name": "Qwen", "online_model_name": "qwen-plus", "api_url": "model.example", "api_key": "test-secret", }, config={ "generation_prompt": "请处理:{{ content }}", "json_mode": True, "temperature": 0.2, "max_tokens": 512, }, task_id="task-1", split={"train": 100, "validation": 0, "test": 0}, qa_pairs_per_item=1, client=client, on_progress=lambda processed, total: progress_updates.append((processed, total)), ) assert len(records) == 1 assert records[0]["status"] == "valid" assert records[0]["split"] == "train" assert requests[0].headers["Authorization"] == "Bearer test-secret" 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]] = [] def handler(request: httpx.Request) -> httpx.Response: payload = json.loads(request.content) requests.append(payload) return httpx.Response( 200, json={ "choices": [{ "finish_reason": "stop", "message": { "reasoning_content": "模型内部思考不应混入业务 JSON", "content": json.dumps({ "items": [{ "instruction": "申请编号有什么作用?", "reasoning": "来源说明它用于标识报销申请。", "answer": "它用于唯一标识一笔报销申请。", }], }, ensure_ascii=False), }, }], "output_sensitive": False, "base_resp": {"status_code": 0, "status_msg": ""}, }, ) records = generate_model_records( [{"id": "preview-minimax", "edited_content": "申请编号用于标识报销申请。"}], model={ "name": "MiniMax", "online_model_name": "MiniMax-M3", "api_url": "https://api.minimaxi.com/v1", }, config={ "output_type": "reasoning", "json_mode": True, "max_tokens": 1024, "generation_retries": 0, }, task_id="task-minimax", 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 len(requests) == 1 assert requests[0]["reasoning_split"] is True assert requests[0]["max_completion_tokens"] >= 4096 assert "max_tokens" not in requests[0] assert "response_format" not in requests[0] def test_minimax_m3_keeps_larger_configured_completion_budget() -> None: requests: list[dict[str, object]] = [] def handler(request: httpx.Request) -> httpx.Response: requests.append(json.loads(request.content)) return httpx.Response( 200, json={ "choices": [{ "message": { "content": json.dumps({ "items": [{ "instruction": "问题", "output": "这是满足测试要求的完整答案。", }], }, ensure_ascii=False), }, }], }, ) generate_model_records( [{"id": "preview-minimax-budget", "edited_content": "来源正文"}], model={ "online_model_name": "MiniMax-M3", "api_url": "https://api.minimax.io/v1", }, config={"max_tokens": 8192, "generation_retries": 0}, task_id="task-minimax-budget", split={"train": 100, "validation": 0, "test": 0}, qa_pairs_per_item=1, client=httpx.Client(transport=httpx.MockTransport(handler)), ) assert requests[0]["max_completion_tokens"] == 8192 def test_minimax_m3_name_on_custom_proxy_keeps_generic_openai_parameters() -> None: requests: list[dict[str, object]] = [] def handler(request: httpx.Request) -> httpx.Response: requests.append(json.loads(request.content)) return httpx.Response( 200, json={ "choices": [{ "message": { "content": json.dumps({ "items": [{ "instruction": "问题", "output": "这是代理服务返回的完整答案。", }], }, ensure_ascii=False), }, }], }, ) generate_model_records( [{"id": "preview-minimax-proxy", "edited_content": "来源正文"}], model={ "online_model_name": "MiniMax-M3", "api_url": "https://model-proxy.example/v1", }, config={"max_tokens": 1024, "json_mode": True, "generation_retries": 0}, task_id="task-minimax-proxy", split={"train": 100, "validation": 0, "test": 0}, qa_pairs_per_item=1, client=httpx.Client(transport=httpx.MockTransport(handler)), ) assert requests[0]["max_tokens"] == 1024 assert requests[0]["response_format"] == {"type": "json_object"} assert "reasoning_split" not in requests[0] assert "max_completion_tokens" not in requests[0] def test_generate_model_records_extracts_json_surrounded_by_model_explanation() -> None: content = "模型结果如下:\n```json\n" + json.dumps( { "items": [{ "instruction": "字段有什么作用?", "output": "该字段用于唯一标识记录。", }], }, ensure_ascii=False, ) + "\n```\n生成完毕。" client = httpx.Client( transport=httpx.MockTransport( lambda _: httpx.Response( 200, json={"choices": [{"message": {"content": content}}]}, ) ) ) records = generate_model_records( [{"id": "preview-explanation", "edited_content": "字段用于唯一标识记录。"}], model={"name": "model", "api_url": "https://model.example/v1"}, config={"generation_retries": 0}, task_id="task-explanation", split={"train": 100, "validation": 0, "test": 0}, qa_pairs_per_item=1, client=client, ) assert records[0]["status"] == "valid" assert records[0]["output"] == "该字段用于唯一标识记录。" def test_generate_model_records_reports_token_truncation_instead_of_json_error() -> None: client = httpx.Client( transport=httpx.MockTransport( lambda _: httpx.Response( 200, json={ "choices": [{ "finish_reason": "length", "message": {"content": ""}, }], "output_sensitive": False, }, ) ) ) records = generate_model_records( [{"id": "preview-truncated", "edited_content": "来源正文"}], model={"name": "model", "api_url": "https://model.example/v1"}, config={"generation_retries": 0}, task_id="task-truncated", split={"train": 100, "validation": 0, "test": 0}, qa_pairs_per_item=1, client=client, ) assert records[0]["status"] == "invalid" assert "Token" in records[0]["error"] assert "截断" in records[0]["error"] def test_token_truncation_is_not_retried_even_when_json_looks_complete() -> None: request_count = 0 content = json.dumps({ "items": [{ "instruction": "问题", "output": "表面完整但服务端已声明截断。", }], }, ensure_ascii=False) def handler(_: httpx.Request) -> httpx.Response: nonlocal request_count request_count += 1 return httpx.Response( 200, json={ "choices": [{ "finish_reason": "length", "message": {"content": content}, }], }, ) records = generate_model_records( [{"id": "preview-length", "edited_content": "来源正文"}], model={"name": "model", "api_url": "https://model.example/v1"}, config={"generation_retries": 5}, task_id="task-length", split={"train": 100, "validation": 0, "test": 0}, qa_pairs_per_item=1, client=httpx.Client(transport=httpx.MockTransport(handler)), ) assert request_count == 1 assert records[0]["status"] == "invalid" assert "finish_reason=length" in records[0]["error"] def test_sensitive_model_response_is_not_retried_or_saved() -> None: request_count = 0 def handler(_: httpx.Request) -> httpx.Response: nonlocal request_count request_count += 1 return httpx.Response( 200, json={ "choices": [{ "finish_reason": "stop", "message": {"content": "{}"}, }], "output_sensitive": True, "base_resp": {"status_code": 1027, "status_msg": "output sensitive"}, }, ) records = generate_model_records( [{"id": "preview-sensitive", "edited_content": "来源正文"}], model={"name": "model", "api_url": "https://model.example/v1"}, config={"generation_retries": 5}, task_id="task-sensitive", split={"train": 100, "validation": 0, "test": 0}, qa_pairs_per_item=1, client=httpx.Client(transport=httpx.MockTransport(handler)), ) assert request_count == 1 assert records[0]["status"] == "invalid" assert "安全拦截" in records[0]["error"] assert "1027" in records[0]["error"] def test_empty_model_content_can_retry_then_succeed() -> None: request_count = 0 def handler(_: httpx.Request) -> httpx.Response: nonlocal request_count request_count += 1 if request_count == 1: return httpx.Response( 200, json={"choices": [{"finish_reason": "stop", "message": {"content": ""}}]}, ) return httpx.Response( 200, json={ "choices": [{ "finish_reason": "stop", "message": { "content": json.dumps({ "items": [{ "instruction": "问题", "output": "第二次请求返回了完整答案。", }], }, ensure_ascii=False), }, }], }, ) records = generate_model_records( [{"id": "preview-empty-retry", "edited_content": "来源正文"}], model={"name": "model", "api_url": "https://model.example/v1"}, config={"generation_retries": 1}, task_id="task-empty-retry", split={"train": 100, "validation": 0, "test": 0}, qa_pairs_per_item=1, client=httpx.Client(transport=httpx.MockTransport(handler)), ) assert request_count == 2 assert records[0]["status"] == "valid" def test_multiple_top_level_json_documents_are_rejected_as_ambiguous() -> None: first = json.dumps({ "items": [{"instruction": "问题一", "output": "答案一"}], }, ensure_ascii=False) second = json.dumps({ "items": [{"instruction": "问题二", "output": "答案二"}], }, ensure_ascii=False) client = httpx.Client( transport=httpx.MockTransport( lambda _: httpx.Response( 200, json={"choices": [{"message": {"content": f"{first}\n{second}"}}]}, ) ) ) records = generate_model_records( [{"id": "preview-ambiguous", "edited_content": "来源正文"}], model={"name": "model", "api_url": "https://model.example/v1"}, config={"generation_retries": 0}, task_id="task-ambiguous", split={"train": 100, "validation": 0, "test": 0}, qa_pairs_per_item=1, client=client, ) assert records[0]["status"] == "invalid" assert "多个 JSON" in records[0]["error"] 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 assert "推理详细程度为“普通”" in system_prompt assert "系统会在保存时统一组装" in system_prompt content = "模型接口自己的分析" + 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"] == ( "\n先识别两项费用,再计算 30 + 20。\n\n合计 50 元。" ) def test_generate_model_records_uses_detailed_reasoning_instruction() -> None: def handler(request: httpx.Request) -> httpx.Response: system_prompt = json.loads(request.content)["messages"][0]["content"] assert "推理详细程度为“详细”" in system_prompt assert "完整展开问题条件、来源依据、中间计算或推导" in system_prompt return httpx.Response( 200, json={ "choices": [ { "message": { "content": json.dumps( { "items": [ { "instruction": "计算报销总额", "reasoning": "条件为交通费 30 元和餐费 20 元。分别核对后相加,30 + 20 = 50。", "answer": "报销总额为 50 元。", } ] }, ensure_ascii=False, ) } } ] }, ) records = generate_model_records( [{"id": "preview-detailed", "edited_content": "交通费 30 元,餐费 20 元"}], model={"name": "model", "api_url": "https://model.example/v1"}, config={"output_type": "reasoning", "reasoning_detail": "detailed"}, task_id="task-detailed", 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 "分别核对后相加" in records[0]["output"] 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": "不应保存的分析最终答案", } ] }, 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( lambda _: httpx.Response(200, json={"choices": [{"message": {"content": "not-json"}}]}) ) ) records = generate_model_records( [{"id": "preview-1", "edited_content": "来源正文"}], model={"name": "model", "api_url": "https://model.example/v1"}, config={"generation_retries": 1}, task_id="task-1", split={"train": 80, "validation": 10, "test": 10}, qa_pairs_per_item=1, client=client, ) 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"] def test_generate_model_records_does_not_retry_non_retryable_http_errors() -> None: request_count = 0 def handler(_: httpx.Request) -> httpx.Response: nonlocal request_count request_count += 1 return httpx.Response(401, json={"error": {"message": "unauthorized"}}) records = generate_model_records( [{"id": "preview-auth", "edited_content": "来源内容"}], model={ "api_url": "https://model.example/v1", "online_model_name": "test-model", "api_key": "invalid", }, config={"generation_retries": 5}, task_id="task-auth", split={"train": 100, "validation": 0, "test": 0}, qa_pairs_per_item=1, client=httpx.Client(transport=httpx.MockTransport(handler)), ) assert request_count == 1 assert records[0]["status"] == "invalid" assert "401" in records[0]["error"] @pytest.mark.parametrize("status_code", [408, 425, 429, 500]) def test_generate_model_records_retries_retryable_http_statuses( status_code: int, ) -> None: request_count = 0 def handler(_: httpx.Request) -> httpx.Response: nonlocal request_count request_count += 1 if request_count == 1: return httpx.Response(status_code) return httpx.Response( 200, json={ "choices": [{ "message": { "content": json.dumps({ "items": [{ "instruction": "来源内容是什么?", "output": "这是用于验证可重试错误的来源内容。", }], }, ensure_ascii=False), }, }], }, ) records = generate_model_records( [{"id": "preview-retryable", "edited_content": "来源内容"}], model={"name": "model", "api_url": "https://model.example/v1"}, config={"generation_retries": 1}, task_id="task-retryable", split={"train": 100, "validation": 0, "test": 0}, qa_pairs_per_item=1, client=httpx.Client(transport=httpx.MockTransport(handler)), ) assert request_count == 2 assert records[0]["status"] == "valid" def test_generate_model_records_retries_transient_network_errors() -> None: request_count = 0 def handler(request: httpx.Request) -> httpx.Response: nonlocal request_count request_count += 1 if request_count == 1: raise httpx.ConnectError("temporary connection failure", request=request) return httpx.Response( 200, json={ "choices": [{ "message": { "content": json.dumps({ "items": [{ "instruction": "网络恢复了吗?", "output": "临时连接错误后,第二次模型请求已经成功。", }], }, ensure_ascii=False), }, }], }, ) records = generate_model_records( [{"id": "preview-network", "edited_content": "网络重试来源"}], model={"name": "model", "api_url": "https://model.example/v1"}, config={"generation_retries": 1}, task_id="task-network", split={"train": 100, "validation": 0, "test": 0}, qa_pairs_per_item=1, client=httpx.Client(transport=httpx.MockTransport(handler)), ) assert request_count == 2 assert records[0]["status"] == "valid" @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, ) 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, ) def test_generate_model_records_rejects_unknown_reasoning_detail() -> None: with pytest.raises(ModelGenerationError, match="reasoning_detail"): generate_model_records( [], model={"name": "model", "api_url": "https://model.example/v1"}, config={"output_type": "reasoning", "reasoning_detail": "verbose"}, task_id="task-invalid-reasoning-detail", split={"train": 100, "validation": 0, "test": 0}, qa_pairs_per_item=1, )