ai llm训练数据合成说明

一、推理服务

使用llamacpp做本地推理服务，使用gguf加gpu加速。

模型使用Jackrong/Qwen3.5-9B-Claude-4.6-Opus-Reasoning-Distilled-GGUF

llama-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; charset=utf-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(name="ExamGenerator") as pipeline: load_dataset = LoadDataFromDicts( name="load_instructions", data=[ {"page": "迁移学习（TL）是机器学习（ML）中的一种技术，它涉及将知识从一个模型迁移到另一个模型。关键技术包括：1）特征提取——使用预训练模型中学到的表示；2）微调——调整预训练模型的全部或部分层；3）预训练模型——在大型数据集（如ImageNet）上训练的模型，如ResNet、VGG。迁移学习的优点包括减少训练时间、提高性能和降低数据需求。"}, {"page": "深度学习是机器学习的一个子集，它使用具有多层的神经网络。关键架构包括：1）卷积神经网络（CNN）——用于图像识别；2）循环神经网络（RNN）——用于序列数据；3）Transformer——用于自然语言处理。应用领域包括计算机视觉、语音识别和自动驾驶。"}, ], output_mappings={"page": "instruction"}, ) # 核心优化：使用 structured_output 自动处理格式，无需正则 text_generation = TextGeneration( name="exam_generation", system_prompt=SYSTEM_PROMPT, template="文档内容：\n{{ instruction }}\n\n请按要求生成JSON：", llm=OpenAILLM( model="glm-5-fp8", base_url="http://192.168.5.149:8080/v1", api_key="EMPTY", # 很多兼容 OpenAI 的本地服务支持 response_format generation_kwargs={"response_format": {"type": "json_object"}} ), input_batch_size=1, num_generations=1, ) 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_cache=False, ) # 导出数据：由于使用了 structured_output，item["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", encoding="utf-8") as f: json.dump(output_data, f, ensure_ascii=False, indent=2) 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）——用于序列数据；3）Transformer——用于自然语言处理。应用领域包括计算机视觉、语音识别和自动驾驶。", "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(r'1[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', encoding='utf-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', encoding='utf-8') as f: for entry in sft_data: f.write(json.dumps(entry, ensure_ascii=False) + "\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_url=os.environ["ARGILLA_API_URL"], api_key=os.environ["ARGILLA_API_KEY"] ) # 3. 定义数据集 (v2.0 使用 rg.Dataset) # 如果数据集已存在，可以使用 client.datasets("exam_dataset_v1") 获取 dataset = rg.Dataset( name="exam_dataset_v1", settings=rg.Settings( fields=[ rg.TextField(name="instruction"), rg.TextField(name="question"), rg.TextField(name="answer"), rg.TextField(name="topic"), ], questions=[ rg.LabelQuestion( name="difficulty_check", title="难度是否准确？", labels=["简单", "中等", "困难"] ), rg.RatingQuestion( name="quality_score", title="数据质量评分", values=[1, 2, 3, 4, 5] ), rg.LabelQuestion( name="task_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", encoding="utf-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 'UseSVE=0' Error: Could not create the Java Virtual Machine. Error: A fatal exception has occurred. Program will exit.

原因：Unrecognized VM option 'UseSVE=0'，是 Argilla 官方示例docker-compose.yaml里给 Elasticsearch 加了-XX:UseSVE=0这个 JVM 选项，但你的 JVM / 架构不支持这个参数，导致 ES 容器直接起不来，所以 Argilla 也连不上 ES、UI 也不会正常可用。

解决方法：把elasticsearch组件ES_JAVA_OPTS和CLI_JAVA_OPTS里的-XX:UseSVE=0删掉，再重启容器。

environment: - ES_JAVA_OPTS=-Xms512m -Xmx512m -XX:UseSVE=0 - CLI_JAVA_OPTS=-XX:UseSVE=0

environment: - 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末端加入强类型校验Schema

Validation。任何不符合预定义结构的数据应被视为“脏数据”并直接过滤，而不是试图对其进行复杂的正则修复。

• 工程化过滤策略：清洗不应仅针对特殊符号，更要涉及“业务过滤”。例如：长度分布过滤，剔除过短或过长的无效样本、事实覆盖过滤，剔除文档外信息、以及隐私脱敏。
• Markdown 与 噪声处理：LLM 倾向于添加 Markdown 格式包装，清洗逻辑应作为 Pipeline的标配组件，专门负责去除干扰字符，确保数据进入训练框架前是纯净的JSON/JSONL。

3、 数据分析与处理过程：从“数据”到“资产”

• 利用元数据实现“数据驱动”的决策：一旦具备了主题、难度等元数据，数据分析就不再是简单的“看一眼”，而是能够计算出各分类下的量值占比。这能精准识别数据“洼地”如：计算机视觉领域缺少困难题目，从而指引下一步的补充生成Data

Augmentation。

• 人工辅助校验 (Human-in-the-loop)：Argilla

等工具的价值在于通过抽检和校准来闭环合成流程。当分析发现自动标注的“难度”与人类感知不一致时，应反向优化合成阶段的System Prompt，而非仅在后期修补。

• SFT 的标准化适配：在数据处理末端，必须将嵌套的结构化数据如文档+QA列表扁平化（Flatten）为标准化的 messages

格式。这是训练框架可移植性的保障，也是实现“多领域微调”或“课程学习”的基础。

• 在 Argilla 中标记为“低质量”的数据后，不要直接丢弃。将这些数据导出，分析其共同特征,比如是否都出现在某个特定的 Topic 下，然后回过头去优化 gensftqa.py 中的 SYSTEM_PROMPT。每次微调前，你可以只导出在 Argilla 中被人工标记为 quality_score >= 4 的数据，确保训练出的模型性能最大化。

核心结论

高质量的 SFT数据集不是“清洗出来的”，而是“设计出来的”。通过在合成阶段强制结构化与元数据标注，在清洗阶段进行强校验过滤，在分析阶段进行可视化监控，可以形成一个从“原始文档”到“黄金训练集”的闭环系统，极大地降低微调过程中的“试错成本”并提升模型的泛化与逻辑能力。