ai llm训练数据合成说明

一、推理服务使用llamacpp做本地推理服务使用gguf加gpu加速。模型使用Jackrong/Qwen3.5-9B-Claude-4.6-Opus-Reasoning-Distilled-GGUFllama-server.exe -m .\Qwen3.5-9B.Q4_K_M.gguf -ngl 99 -c 4096 --host 0.0.0.0 --port 8080 --parallel 4 -np 4 -cb参数 全称 说明--host 监听地址。默认 127.0.0.1仅本机访问设为 0.0.0.0 允许局域网/外网访问--port 监听端口。默认 8080-c --ctx-size 上下文长度。影响单次对话能处理的最大 token 数-np --parallel 同时处理的请求数并发槽位。需要配合 -cb 使用-cb --cont-batching 连续批处理。允许多个请求共享同一个批次大幅提升并发性能强烈建议开启--parallel 同 -np设置并发数量-ngl --gpu-layers GPU 卸载层数99 表示全部卸载-sm --split-mode 多卡模式layer / row-ts --tensor-split 多卡显存分配比例请求测试llamacpp 服务$body { messages ( { role user content 人工智能的发展历史 } ) temperature 0.7 max_tokens 512 } | ConvertTo-Json -Depth 3 $bytes [System.Text.Encoding]::UTF8.GetBytes($body) $response Invoke-RestMethod -Uri http://localhost:8080/v1/chat/completions -Method Post -ContentType application/json; charsetutf-8 -Body $bytes Write-Host 模型思考过程 -ForegroundColor Cyan Write-Host $response.choices[0].message.reasoning_content Write-Host n 模型最终回复 -ForegroundColor Green Write-Host $response.choices[0].message.content使用powershell指令确实比linux shell麻烦很多不熟悉用起来太坑人。curl http://192.168.5.142:8080/v1/chat/completions -H Content-Type: application/json -d { messages: [ { role: system, content: 你是一个直接的AI助手请直接给出答案不要输出思考过程。 }, { role: user, content: 人工智能的发展历史? } ], temperature: 0.7, max_tokens: 1024 }二、数据合成处理distilabel做数据合成代码示例import json import re from typing import List, Literal from pydantic import BaseModel, Field from distilabel.models import OpenAILLM from distilabel.pipeline import Pipeline from distilabel.steps import LoadDataFromDicts from distilabel.steps.tasks import TextGeneration # 1. 定义带有元数据的结构化模型 class ExamQuestion(BaseModel): question: str Field(..., description问题内容) answer: str Field(..., description正确答案必须完全基于文档) distractors: List[str] Field(..., description干扰项列表) difficulty: Literal[简单, 中等, 困难] Field(..., description评估问题的难度) topic: str Field(..., description问题所属的一级领域) task_type: Literal[定义解释, 原理分析, 场景应用, 技术比较] Field(..., description任务类型) is_grounded: bool Field(..., description答案是否完全包含在文档中) class ExamQuestions(BaseModel): exam: List[ExamQuestion] SYSTEM_PROMPT \ 你是一名专业命题老师。根据文档生成高质量考试题。 输出必须是合法的 JSON 对象不要包含 Markdown 标记。 结构如下 { exam:[ { question: 问题, answer: 正确答案, distractors: [错误1, 错误2, 错误3], difficulty: 中等, topic: 领域名, task_type: 定义解释, is_grounded: true } ] } .strip() # 3. 更新 Pipeline with Pipeline(nameExamGenerator) as pipeline: load_dataset LoadDataFromDicts( nameload_instructions, data[ {page: 迁移学习TL是机器学习ML中的一种技术它涉及将知识从一个模型迁移到另一个模型。关键技术包括1特征提取——使用预训练模型中学到的表示2微调——调整预训练模型的全部或部分层3预训练模型——在大型数据集如ImageNet上训练的模型如ResNet、VGG。迁移学习的优点包括减少训练时间、提高性能和降低数据需求。}, {page: 深度学习是机器学习的一个子集它使用具有多层的神经网络。关键架构包括1卷积神经网络CNN——用于图像识别2循环神经网络RNN——用于序列数据3Transformer——用于自然语言处理。应用领域包括计算机视觉、语音识别和自动驾驶。}, ], output_mappings{page: instruction}, ) # 核心优化使用 structured_output 自动处理格式无需正则 text_generation TextGeneration( nameexam_generation, system_promptSYSTEM_PROMPT, template文档内容\n{{ instruction }}\n\n请按要求生成JSON, llmOpenAILLM( modelglm-5-fp8, base_urlhttp://192.168.5.149:8080/v1, api_keyEMPTY, # 很多兼容 OpenAI 的本地服务支持 response_format generation_kwargs{response_format: {type: json_object}} ), input_batch_size1, num_generations1, ) load_dataset text_generation if __name__ __main__: distiset pipeline.run( parameters{ text_generation.name: { llm: {generation_kwargs: {max_new_tokens: 4096, temperature: 0.2}} } }, use_cacheFalse, ) # 导出数据由于使用了 structured_outputitem[generation] 直接就是字典对象 dataset distiset[default][train] output_data [] for idx, item in enumerate(dataset): try: raw_gen item.get(generation, ) # 如果是空的或过短直接跳过 if not raw_gen or len(raw_gen.strip()) 20: print(f数据解析跳过: 第 {idx} 条为空或内容异常) continue # 清理 Markdown 和多余字符 clean_gen re.sub(r^json\s*|\s*$, , raw_gen.strip()) # 解析 JSON data_dict json.loads(clean_gen) # 校验格式 exam_data ExamQuestions(**data_dict) output_data.append({ instruction: item[instruction], qa_list:[q.model_dump() for q in exam_data.exam], model: item.get(model_name, glm-5-fp8) }) print(f成功处理第 {idx} 条) except Exception as e: print(f解析失败: {e} | 数据: {raw_gen[:100]}) with open(data/optimized_exam_with_meta.json, w, encodingutf-8) as f: json.dump(output_data, f, ensure_asciiFalse, indent2) print(数据合成与元数据标注完成已保存至 data/optimized_exam_with_meta.json)合成数据示例[ { instruction: 迁移学习TL是机器学习ML中的一种技术它涉及将知识从一个模型迁移到另一个模型。关键技术包括1特征提取——使用预训练模型中学到的表示2微调——调整预训练模型的全部或部分层3预训练模型——在大型数据集如ImageNet上训练的模型如ResNet、VGG。迁移学习的优点包括减少训练时间、提高性能和降低数据需求。, qa_list: [ { question: 根据文档内容迁移学习中的预训练模型主要是在什么类型的数据集上训练的, answer: 大型数据集如ImageNet, distractors: [ 小型数据集, 无标签数据集, 随机生成的数据集 ], difficulty: 中等, topic: 机器学习/迁移学习, task_type: 定义解释, is_grounded: true } ], model: glm-5 }, { instruction: 深度学习是机器学习的一个子集它使用具有多层的神经网络。关键架构包括1卷积神经网络CNN——用于图像识别2循环神经网络RNN——用于序列数据3Transformer——用于自然语言处理。应用领域包括计算机视觉、语音识别和自动驾驶。, qa_list: [ { question: 根据文档内容深度学习与机器学习之间的关系是什么, answer: 深度学习是机器学习的一个子集, distractors: [ 机器学习是深度学习的子集, 深度学习与机器学习是完全无关的领域, 深度学习与机器学习是同一个概念 ], difficulty: 中等, topic: 深度学习基础概念, task_type: 定义解释, is_grounded: true } ], model: glm-5 } ]三、数据做清洗、处理注意需要此处代码比不能直接处理上面带元数据的合成文件是之前给简单合成问答对数据使用的。import json import re from typing import List # 1. 简单的特殊符号与隐私脱敏清理逻辑 def clean_text(text: str) - str: # 替换常见的全角转半角如果需要或清除不可见字符 text text.replace(\u200b, ).replace(\xa0, ) # 简单脱敏例如将形如 13800000000 的替换为 [PHONE] text re.sub(r1[3-9]\d{9}, [PHONE], text) # 去除多余的重复词简单示例连续出现的相同词组 text re.sub(r(你好){2,}, 你好, text) return text.strip() # 2. 长度统计阈值 MIN_TOKENS 5 MAX_TOKENS 1024 # 根据你的模型上下文设定 def get_token_len(text: str) - int: # 这里使用简单的字符长度估算如果需要精确可以使用 tiktoken return len(text) def process_data(input_file: str, output_file: str): with open(input_file, r, encodingutf-8) as f: data json.load(f) sft_data [] stats {total: 0, filtered: 0} for entry in data: instruction clean_text(entry[instruction]) for qa in entry[qa_list]: stats[total] 1 question clean_text(qa[question]) answer clean_text(qa[answer]) # 3. 长度分布过滤 if not (MIN_TOKENS get_token_len(instruction question) MAX_TOKENS): stats[filtered] 1 continue # 5. 适配 SFT 问答对格式 (ChatML/Messages 格式) sft_entry { messages:[ {role: system, content: 你是一名专业助手请根据提供的参考文档回答问题。}, {role: user, content: f参考文档{instruction}\n\n问题{question}}, {role: assistant, content: answer} ] } sft_data.append(sft_entry) # 导出数据 with open(output_file, w, encodingutf-8) as f: for entry in sft_data: f.write(json.dumps(entry, ensure_asciiFalse) \n) print(f处理完成) print(f原始样本数: {stats[total]}, 过滤后样本数: {len(sft_data)}, 过滤数: {stats[filtered]}) if __name__ __main__: process_data(../distidata/data/optimized_exam.json, ../distidata/data/sft_train.jsonl)四、数据人工处理argilla做数据人工处理服务使用本地docker启动。操作说明https://docs.argilla.io/latest/getting_started/how-to-deploy-argilla-with-docker/本地需要安装sdk 客户端用与加载数据示例import argilla as rg import json import os # 1. 设置连接参数 os.environ[ARGILLA_API_URL] http://localhost:6900 os.environ[ARGILLA_API_KEY] argilla.apikey # 2. 初始化客户端 (v2.0 推荐方式) client rg.Argilla( api_urlos.environ[ARGILLA_API_URL], api_keyos.environ[ARGILLA_API_KEY] ) # 3. 定义数据集 (v2.0 使用 rg.Dataset) # 如果数据集已存在可以使用 client.datasets(exam_dataset_v1) 获取 dataset rg.Dataset( nameexam_dataset_v1, settingsrg.Settings( fields[ rg.TextField(nameinstruction), rg.TextField(namequestion), rg.TextField(nameanswer), rg.TextField(nametopic), ], questions[ rg.LabelQuestion( namedifficulty_check, title难度是否准确, labels[简单, 中等, 困难] ), rg.RatingQuestion( namequality_score, title数据质量评分, values[1, 2, 3, 4, 5] ), rg.LabelQuestion( nametask_type, title任务类型分类, labels[定义解释, 原理分析, 场景应用, 技术比较] ) ] ) ) # 4. 创建或加载数据集 try: dataset.create() print(数据集已创建) except rg._exceptions._api.ConflictError: dataset client.datasets(exam_dataset_v1) print(数据集已存在直接加载) # 5. 读取数据并构建记录 with open(D:/tool/distilabel/distidata/data/optimized_exam_with_meta.json, r, encodingutf-8) as f: data json.load(f) records [] for entry in data: for qa in entry[qa_list]: # v2.0 直接使用 Dataset.records.log 来上传数据 records.append({ instruction: entry[instruction], question: qa[question], answer: qa[answer], topic: qa[topic], difficulty_check: qa[difficulty], # 直接在 record 中填入响应值 task_type: qa[task_type] }) # 6. 上传记录 dataset.records.log(records) print(数据已成功导入 Argilla v2 环境)在windows上遇到问题是启动server服务es服务没有启动报错Unrecognized VM option UseSVE0 Error: Could not create the Java Virtual Machine. Error: A fatal exception has occurred. Program will exit.原因Unrecognized VM option UseSVE0是 Argilla 官方示例docker-compose.yaml里给 Elasticsearch 加了-XX:UseSVE0这个 JVM 选项但你的 JVM / 架构不支持这个参数导致 ES 容器直接起不来所以 Argilla 也连不上 ES、UI 也不会正常可用。解决方法把elasticsearch组件ES_JAVA_OPTS和CLI_JAVA_OPTS里的-XX:UseSVE0删掉再重启容器。environment: - ES_JAVA_OPTS-Xms512m -Xmx512m -XX:UseSVE0 - CLI_JAVA_OPTS-XX:UseSVE0environment: - ES_JAVA_OPTS-Xms512m -Xmx512m - CLI_JAVA_OPTS登录页面的username、password在docker-compose文件里apikey也在其中。五、原始数据合成处理过程总结1、 合成数据阶段从“生成”到“构建”结构化输出是核心不要寄希望于模型的 Prompt 约束“自觉性”。在合成阶段必须通过 Schema定义Pydantic或结构化协议强制模型输出格式。这能从根源上减少后续清洗的负担。元数据注入即正义合成数据不应仅仅包含问题和答案。在合成阶段就通过 Prompt 引导模型自动生成元数据Metadata如领域主题、难度等级、任务类型、事实性标记是实现后续精细化分析的前提。防御性Prompt设计针对合成任务System Prompt应始终包含“事实一致性约束”禁止幻觉要求模型回答必须完全基于给定的文档上下文而非利用模型自身的知识库。2、 数据清洗阶段防御性处理与稳健性后置校验Post-Validation是最后一道防线无论模型生成多么规范都必须在Pipeline末端加入强类型校验SchemaValidation。任何不符合预定义结构的数据应被视为“脏数据”并直接过滤而不是试图对其进行复杂的正则修复。工程化过滤策略清洗不应仅针对特殊符号更要涉及“业务过滤”。例如长度分布过滤剔除过短或过长的无效样本、事实覆盖过滤剔除文档外信息、以及隐私脱敏。Markdown 与 噪声处理LLM 倾向于添加 Markdown 格式包装清洗逻辑应作为 Pipeline的标配组件专门负责去除干扰字符确保数据进入训练框架前是纯净的JSON/JSONL。3、 数据分析与处理过程从“数据”到“资产”利用元数据实现“数据驱动”的决策一旦具备了主题、难度等元数据数据分析就不再是简单的“看一眼”而是能够计算出各分类下的量值占比。这能精准识别数据“洼地”如计算机视觉领域缺少困难题目从而指引下一步的补充生成DataAugmentation。人工辅助校验 (Human-in-the-loop)Argilla等工具的价值在于通过抽检和校准来闭环合成流程。当分析发现自动标注的“难度”与人类感知不一致时应反向优化合成阶段的System Prompt而非仅在后期修补。SFT 的标准化适配在数据处理末端必须将嵌套的结构化数据如文档QA列表扁平化Flatten为标准化的 messages格式。这是训练框架可移植性的保障也是实现“多领域微调”或“课程学习”的基础。在 Argilla 中标记为“低质量”的数据后不要直接丢弃。将这些数据导出分析其共同特征,比如是否都出现在某个特定的 Topic 下然后回过头去优化 gensftqa.py 中的 SYSTEM_PROMPT。每次微调前你可以只导出在 Argilla 中被人工标记为 quality_score 4 的数据确保训练出的模型性能最大化。核心结论高质量的 SFT数据集不是“清洗出来的”而是“设计出来的”。通过在合成阶段强制结构化与元数据标注在清洗阶段进行强校验过滤在分析阶段进行可视化监控可以形成一个从“原始文档”到“黄金训练集”的闭环系统极大地降低微调过程中的“试错成本”并提升模型的泛化与逻辑能力。