第 45 章

第 45 章：中文数据集构建实验

第 45 章：中文数据集构建实验

本文整理自 Datawhale 开源项目 datawhalechina/jev-cookbook（CC BY-NC-SA 4.0），源文件：main/10_本地模型/notebooks/zh_dataset_construction.ipynb。

本 Notebook 保留原有的任务规范、候选生成和人工审核流程；新增一条基于 ModelScope ShareGPT 中文对话的真实语料构造流程。112 条 zh-pilot-112 试验数据与其训练 Notebook 不会被覆盖。

新增数据是对话决策辅助数据：输入为当前用户轮次及此前对话，标签描述下一步回答策略、主题、是否澄清、是否外部核验和推理深度。它不训练模型复述原始 assistant 回答。所有模型标注只作为待审核候选；投票频率是伪标签代理，不是人工概率校准结果。

from pathlib import Path
import sys

ROOT = next(
    (parent for parent in (Path.cwd(), *Path.cwd().parents)
     if (parent / "laya" / "generate_synthetic_data.py").is_file()),
    None,
)
if ROOT is None:
    raise FileNotFoundError("请从 jev-cookbook 仓库根目录或 10_本地模型/notebooks 打开本 Notebook")
sys.path.insert(0, str(ROOT / "laya"))

EXAMPLE_SPEC = ROOT / "laya" / "data_generation" / "spec.example.json"
GENERATED_DIR = ROOT / "laya" / "data_generation" / "generated"
GENERATED_DIR.mkdir(parents=True, exist_ok=True)
SPEC_PATH = GENERATED_DIR / "my_task_spec.json"
print("仓库根目录:", ROOT)
print("任务规范副本:", SPEC_PATH)
print("生成目录（Git 忽略）:", GENERATED_DIR)

1. 复制任务规范并定义标签规则

任务规范决定模型可以生成什么，以及每道题按什么证据贴标签。先阅读 spec.example.json；它定义了客服部门、是否加急和严重程度三个任务。改用自己的领域时，编辑 Notebook 下一格中的路径副本或直接打开 my_task_spec.json，重点完善 state 可见字段、候选定义、labeling_policy、边界案例和版本号。

示例只作为规则格式参考，不要把真实验证集、个人资料或客户秘密放进规范和示例。

import json
from generate_synthetic_data import load_spec

if not SPEC_PATH.exists():
    template = json.loads(EXAMPLE_SPEC.read_text(encoding="utf-8"))
    SPEC_PATH.write_text(json.dumps(template, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    print("已复制示例规范。请按业务规则修改该文件，再重新运行本单元格。")

spec = load_spec(SPEC_PATH)
print(json.dumps(spec, ensure_ascii=False, indent=2))

标签约定

题型 在任务规范里定义候选 API 返回标签
choice 按固定顺序排列的对象，key 是标签 ID 精确返回其中一个 key
noul 同时定义 false 和 true JSON 布尔值 false / true
score 从低到高排列的候选数组 从 0 开始的整数等级

生成器会把标签转换为候选位置上的整数 y。choice 候选顺序和 score 等级顺序属于标签定义的一部分，不要在规范中随意重排。soft 概率分布不会由生成模型伪造。

# 重新读取并检查规范；字段或标签规则有误时会在这里报出具体问题。
spec = load_spec(SPEC_PATH)
print("任务族:", spec["task_family"])
print("规范版本:", spec["policy_version"])
print("语言:", spec["lang"])
print("题型:", {q["id"]: q["t"] for q in spec["questions"]})
print("人工示例数:", len(spec.get("examples", [])))

2. 配置 API 环境变量和并发参数

先在启动 Jupyter 的操作系统会话中设置 DEEPSEEK_API_KEY，再启动或重启 Notebook kernel。具体设置方式见数据构造指南第 2 节。

默认模型是 deepseek-flash（DeepSeek-V4.1-Flash）；API 使用 Chat Completions JSON Output。可在运行参数中调整模型、输出条数、每请求 batch size 和并发 worker。workers=4 时最多同时等待 4 个 API 请求；生成器按完成的请求打印进度，而不是逐 token 流式输出。

import os
import uuid
from datetime import datetime, timezone

MODEL = "deepseek-flash"
COUNT = 20
BATCH_SIZE = 4
WORKERS = 4
TEMPERATURE = 0.7
MAX_TOKENS = 4096
TIMEOUT_SECONDS = 120
RUN_GENERATION = False  # 确认规则和 API 用量后，手动改为 True

run_stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + "-" + uuid.uuid4().hex[:6]
task_slug = "".join(ch if ch.isalnum() or ch in "-_" else "_" for ch in spec["task_family"]).strip("-_") or "laya-data"
OUTPUT_PATH = GENERATED_DIR / (task_slug + "-" + run_stamp + "-candidate.jsonl")
key_is_configured = bool(os.environ.get("DEEPSEEK_API_KEY"))
print("API key 已配置:", key_is_configured, "（只显示状态，不显示内容）")
print("模型:", MODEL, "样本数:", COUNT, "batch:", BATCH_SIZE, "并发:", WORKERS)
print("候选输出:", OUTPUT_PATH)

3. 并发生成候选 JSONL

第一次建议生成 10–20 条并逐条复核。Notebook 用 python -u 无缓冲运行脚本，因此每批请求完成后能看到进度。失败时，已成功完成的批次已写入候选文件；本单元格重复运行时会自动追加并跳过重复 state。

import subprocess

if not RUN_GENERATION:
    print("已跳过 API 请求。确认 DEEPSEEK_API_KEY、任务规范和用量后，把 RUN_GENERATION 设为 True 并重新运行本单元格。")
elif not key_is_configured:
    raise RuntimeError("DEEPSEEK_API_KEY 未设置；请在启动 Jupyter 的终端配置环境变量并重启 kernel。")
else:
    command = [
        sys.executable, "-u", str(ROOT / "laya" / "generate_synthetic_data.py"),
        "--spec", str(SPEC_PATH),
        "--count", str(COUNT),
        "--batch-size", str(BATCH_SIZE),
        "--workers", str(WORKERS),
        "--model", MODEL,
        "--temperature", str(TEMPERATURE),
        "--max-tokens", str(MAX_TOKENS),
        "--timeout", str(TIMEOUT_SECONDS),
        "--out", str(OUTPUT_PATH),
    ]
    if OUTPUT_PATH.exists():
        command.append("--append")
    print("启动并发生成；命令中不包含 API key。")
    child_env = os.environ.copy()
    child_env["PYTHONIOENCODING"] = "utf-8"
    process = subprocess.Popen(
        command, cwd=ROOT, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
        text=True, encoding="utf-8", errors="replace", bufsize=1, env=child_env,
    )
    for line in process.stdout:
        print(line, end="", flush=True)
    status = process.wait()
    if status:
        raise subprocess.CalledProcessError(status, command)

4. 检查 JSONL、重复项和标签分布

如果上一格被跳过，检查会提示尚无本次输出；生成成功后重新运行本格。候选仍必须逐题审核，统计通过不表示标签正确。

from collections import Counter

if not OUTPUT_PATH.is_file():
    audit_records = []
    print("没有找到本次候选文件。先启用生成单元格，再运行本格。")
else:
    audit_records = [
        json.loads(line)
        for line in OUTPUT_PATH.read_text(encoding="utf-8").splitlines()
        if line.strip()
    ]
    ids = [record["id"] for record in audit_records]
    state_keys = [json.dumps(record["state"], ensure_ascii=False, sort_keys=True) for record in audit_records]
    assert len(ids) == len(set(ids)), "发现重复 id"
    assert len(state_keys) == len(set(state_keys)), "发现重复 state"
    assert all(record["split"] == "train_candidate" for record in audit_records)
    assert all(record["metadata"]["review_status"] == "needs_human_review" for record in audit_records)

    label_counts = {q["id"]: Counter() for q in spec["questions"]}
    qspec_by_id = {q["id"]: q for q in spec["questions"]}
    for record in audit_records:
        assert [q["id"] for q in record["qs"]] == list(qspec_by_id)
        for generated_q in record["qs"]:
            qspec = qspec_by_id[generated_q["id"]]
            y = generated_q["y"]
            assert generated_q["t"] == qspec["t"]
            if qspec["t"] == "choice":
                assert isinstance(y, int) and not isinstance(y, bool) and 0 <= y < len(qspec["crit"])
                label = list(qspec["crit"])[y]
            elif qspec["t"] == "noul":
                assert y in (0, 1) and not isinstance(y, bool)
                label = "true" if y == 1 else "false"
            else:
                assert isinstance(y, int) and not isinstance(y, bool) and 0 <= y < len(qspec["crit"])
                label = str(y)
            label_counts[generated_q["id"]][label] += 1

    audit_summary = {
        "file": str(OUTPUT_PATH),
        "records": len(audit_records),
        "unique_ids": len(set(ids)),
        "unique_states": len(set(state_keys)),
        "review_status": "needs_human_review",
        "label_counts": {qid: dict(counts) for qid, counts in label_counts.items()},
    }
    print(json.dumps(audit_summary, ensure_ascii=False, indent=2))

5. 导出逐题人工审核表

CSV 一行对应一条记录中的一道题，包含 state、规范候选、模型建议标签和证据线索。请审核者填写 decision、corrected_label、reviewer、review_note，并结合原文和规范判断；证据线索不能当作自动通过理由。

import csv

if not audit_records:
    REVIEW_CSV = None
    print("没有候选记录，暂不生成审核表。")
else:
    REVIEW_CSV = OUTPUT_PATH.with_name(OUTPUT_PATH.stem + "-review.csv")
    with REVIEW_CSV.open("w", encoding="utf-8-sig", newline="") as csv_file:
        columns = [
            "record_id", "source_group_id", "task_family", "policy_version", "lang", "state_json",
            "question_id", "question_type", "instructions", "labeling_policy", "options_json", "suggested_label",
            "review_evidence", "decision", "corrected_label", "reviewer", "review_note",
        ]
        writer = csv.DictWriter(csv_file, fieldnames=columns)
        writer.writeheader()
        for record in audit_records:
            evidence = record["metadata"].get("review_evidence", {})
            for question in record["qs"]:
                qspec = qspec_by_id[question["id"]]
                y = question["y"]
                if qspec["t"] == "choice":
                    label = list(qspec["crit"])[y]
                    options = qspec["crit"]
                elif qspec["t"] == "noul":
                    label = "true" if y else "false"
                    options = qspec["crit"]
                else:
                    label = str(y)
                    options = qspec["crit"]
                writer.writerow({
                    "record_id": record["id"],
                    "source_group_id": record["source_group_id"],
                    "task_family": record["task_family"],
                    "policy_version": record["metadata"]["policy_version"],
                    "lang": record["lang"],
                    "state_json": json.dumps(record["state"], ensure_ascii=False),
                    "question_id": question["id"],
                    "question_type": question["t"],
                    "instructions": question["ins"],
                    "labeling_policy": qspec["labeling_policy"],
                    "options_json": json.dumps(options, ensure_ascii=False),
                    "suggested_label": label,
                    "review_evidence": evidence.get(question["id"], ""),
                    "decision": "",
                    "corrected_label": "",
                    "reviewer": "",
                    "review_note": "",
                })
    print("人工审核表:", REVIEW_CSV)
    print("审核完成后保留原 JSONL 和审计信息；不要在本 Notebook 中自动晋升 split 或审核状态。")

6. 审核后再进入训练

  1. 逐题核对标签与 labeling_policy，纠正错误或拒绝样本，并记录审核者、规则版本和理由。
  2. 保留原始 JSONL。把接受的记录复制到新的数据文件中，记录人工审核 metadata，再把 split 标为 train。
  3. 合成样本只用于训练。请用独立、人工审核的真实数据构造 dev、校准、锁定 test 和 OOD 集；不要从同一套合成模板随机切出“验证集”。
  4. 再运行 Laya 微调 Notebook 或阅读微调实操指南。训练器会拒绝未经审核的候选数据。

生成器会拒绝覆盖已有文件。重试同一次生成请保留本 Notebook 内的 OUTPUT_PATH，重新运行生成单元格即可续写；新开 kernel 后请把 OUTPUT_PATH 指向相同候选文件，并在 CLI 指令后加 --append。生成目录已被 Git 忽略，审核前不要将业务数据提交到仓库。

7. 从公开中文对话构造决策辅助数据

这里复用 sharegpt_policy_v2.json 的五类决策问题。每条样本从一段公开对话中选一个 user 轮次，state 只保留当前 user 及此前最多 7 条上下文消息；源对话中紧随其后的 assistant 回复不会进入 state，也不会发送给标注 API。v2 policy 加入了“开放问题可以直接回答”和“澄清策略必须与澄清标记一致”等边界示例。每个原始会话最多产生一条样本，避免同一会话跨 split 泄漏。

默认目标规模为 train 1,200、dev 200、calibration 100、test 400（共 1,900 cases / 9,500 decisions）。敏感标识和 URL 在本地过滤；输出先标记为候选。ModelScope 卡片显示 CC-BY-4.0，同时注明沿用 ShareGPT 许可；社区公开前应核对上游 ShareGPT 条款并完成逐题人工审核。

import subprocess
BUILDER = ROOT / "laya" / "data_generation" / "build_sharegpt_laya.py"
POLICY_PATH = ROOT / "laya" / "data_generation" / "sharegpt_policy_v2.json"
SOURCE_DIR = GENERATED_DIR / "sharegpt_zh_38k"
VERSION_DIR = SOURCE_DIR / "v2"
RAW_PATH = SOURCE_DIR / "raw" / "sharegpt_zh_38K_format.jsonl"
CASES_PATH = VERSION_DIR / "cases.jsonl"
VOTES_PATH = VERSION_DIR / "deepseek_votes.jsonl"
MANIFEST_PATH = VERSION_DIR / "manifest.json"
LAYA_CANDIDATES_PATH = VERSION_DIR / "laya_candidates.jsonl"
REVIEW_CSV_PATH = VERSION_DIR / "human_review.csv"
ENV_FILE = ROOT.parent / "laya" / ".env"  # 只把路径传给脚本；密钥不会显示或写入输出
print("原始语料存在:", RAW_PATH.is_file())
print("112 条 pilot 保留:", (ROOT / "laya" / "experiments" / "zh-pilot-112" / "train-dev.jsonl").is_file())
VERSION_DIR.mkdir(parents=True, exist_ok=True)
print("原始语料目录:", SOURCE_DIR)
print("构造版本目录:", VERSION_DIR)

7.0 获取 ModelScope 原始语料（缺文件时）

本工作区已下载原始文件。其他环境可将 DOWNLOAD_SOURCE_IF_MISSING 设为 True 下载；文件约 147 MB，保存到 Git 忽略目录。

from urllib.request import urlopen
SOURCE_REVISION = "75412fc0a6a262899c6b99bfa35349d323d0c5c3"
SOURCE_URL = f"https://www.modelscope.cn/datasets/AI-ModelScope/sharegpt_gpt4/resolve/{SOURCE_REVISION}/sharegpt_zh_38K_format.jsonl"
DOWNLOAD_SOURCE_IF_MISSING = False
if DOWNLOAD_SOURCE_IF_MISSING and not RAW_PATH.is_file():
    RAW_PATH.parent.mkdir(parents=True, exist_ok=True)
    with urlopen(SOURCE_URL, timeout=60) as response, RAW_PATH.open("wb") as output:
        while True:
            chunk = response.read(1024 * 1024)
            if not chunk:
                break
            output.write(chunk)
print("原始数据就绪:", RAW_PATH.is_file(), "字节数:", RAW_PATH.stat().st_size if RAW_PATH.is_file() else 0)

7.1 本地筛选与分层切分

筛选脚本会校验 JSONL，按固定 seed 选出唯一对话状态，输出 cases.jsonl 和带来源、许可提示、过滤统计的 manifest.json。不打印对话正文。原始语料和生成文件都位于 .gitignore 忽略目录。

RUN_REAL_EXTRACTION = True  # 纯本地处理，不产生 API 用量
if RUN_REAL_EXTRACTION:
    command = [sys.executable, str(BUILDER), "extract", "--raw", str(RAW_PATH),
        "--policy", str(POLICY_PATH), "--out", str(CASES_PATH),
        "--manifest", str(MANIFEST_PATH), "--seed", "42",
        "--counts", "train=1200,dev=200,calibration=100,test=400"]
    completed = subprocess.run(command, cwd=ROOT, check=True, text=True, capture_output=True)
    print(completed.stdout)

7.2 并发生成多轮伪标签

每个 state 用 3 轮独立 API 请求标注五道问题，保留每票标签；人工审核时回看原始 state。合并时将投票频率写入 soft，供 soft-target SFT / RLCD 数据管线实验使用；三票分辨率有限且由同一模型产生，必须标为 deepseek_vote_frequency_proxy，不能据此报告真实校准。API 请求可能产生费用；脚本支持断点续跑，日志不输出原文或 key。

RUN_REAL_ANNOTATION = False  # 需明确检查语料范围后再改 True；此步骤会调用 DeepSeek API
ANNOTATION_BATCH_SIZE = 10
ANNOTATION_WORKERS = 4
ANNOTATION_VOTES = 3
if RUN_REAL_ANNOTATION:
    command = [sys.executable, "-u", str(BUILDER), "annotate",
        "--cases", str(CASES_PATH), "--policy", str(POLICY_PATH),
        "--votes-out", str(VOTES_PATH), "--env-file", str(ENV_FILE),
        "--batch-size", str(ANNOTATION_BATCH_SIZE), "--workers", str(ANNOTATION_WORKERS),
        "--votes", str(ANNOTATION_VOTES)]
    child_env = os.environ.copy()
    child_env["PYTHONIOENCODING"] = "utf-8"
    process = subprocess.Popen(command, cwd=ROOT, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
        text=True, encoding="utf-8", errors="replace", bufsize=1, env=child_env)
    for line in process.stdout:
        print(line, end="", flush=True)
    if process.wait():
        raise RuntimeError("标注未完整结束；修复 API 问题后可复用 votes 文件续跑")
else:
    print("已跳过 API 标注。设置 RUN_REAL_ANNOTATION=True 后可断点续跑。")

7.3 合并为 Laya JSONL、审计并导出审核表

合并要求每条 case 收齐 3 票；y 取多数标签，平票按 policy 中的固定选项顺序处理。split 暂时统一为 train_candidate，原计划用途保存在 metadata.planned_split。人工审核并记录修正、审核者和理由后，再形成真正的 train/dev/calibration/test；未经审核不能直接训练。

if LAYA_CANDIDATES_PATH.is_file():
    print("候选数据已存在，跳过 assemble，避免覆盖现有审核表。")
elif REVIEW_CSV_PATH.is_file():
    raise FileExistsError(
        "审核表已存在，但候选文件缺失。为保护人工复核结果，请先备份审核表，"
        "再将重建产物输出到新的版本目录。"
    )
elif CASES_PATH.is_file() and VOTES_PATH.is_file():
    command = [sys.executable, str(BUILDER), "assemble", "--cases", str(CASES_PATH),
        "--policy", str(POLICY_PATH), "--votes", str(VOTES_PATH),
        "--out", str(LAYA_CANDIDATES_PATH), "--review-csv", str(REVIEW_CSV_PATH),
        "--manifest", str(MANIFEST_PATH), "--required-votes", str(ANNOTATION_VOTES)]
    completed = subprocess.run(command, cwd=ROOT, check=True, text=True, capture_output=True)
    print(completed.stdout)
    audit_cmd = [sys.executable, str(BUILDER), "audit", "--cases", str(CASES_PATH),
        "--policy", str(POLICY_PATH), "--votes", str(VOTES_PATH), "--laya", str(LAYA_CANDIDATES_PATH)]
    print(subprocess.run(audit_cmd, cwd=ROOT, check=True, text=True, capture_output=True).stdout)
    print("逐题人工审核表:", REVIEW_CSV_PATH)
else:
    print("尚无完整 votes 文件；先执行本地筛选与标注。")

7.4 进入三种微调对比前的门槛

  • 保留 zh-pilot-112 原样作为小规模流程回归基线；真实公开语料单独存放。
  • 从审核 CSV 逐题核对输入上下文、标签和证据，必要时纠正 y/soft；记录 reviewer 和 policy 版本。
  • 按 source_group_id 建立实际 split；dev、calibration、test 要人工审核。伪标签 test 只能检查管线，不能支撑模型效果结论。
  • 三种方法可以共享审核后的 Laya JSONL schema；当前仓库只有 Head-only trainer，LoRA-SFT 和 RLCD 的 trainer 仍待实现。RLCD 可读取 soft，但本数据的 vote frequency 只是教师模型代理。报告中应单独标注这一限制。
  • 后续迁移 AutoDL 时，将数据与 manifest 放在实例的持久化数据盘目录，并一并分享来源、许可和审核说明。

7.5 完成人工审核后导出正式 split

审核者需要为每个 case 的五道题填写 decision 和 reviewer；corrected_label 使用候选 key、true/false 或 score 索引。下面的导出器只会晋升五题都已审核、每题都填写审核人的记录；任何空项继续留在候选集中。输出 train-dev.jsonl 可供当前 Head-only trainer 使用，calibration/test 独立写出。

RUN_PROMOTE_REVIEWED = False  # 填完并复核 CSV 后显式改为 True
REVIEWED_DIR = VERSION_DIR / "reviewed"
if RUN_PROMOTE_REVIEWED:
    command = [sys.executable, str(BUILDER), "promote",
        "--candidates", str(LAYA_CANDIDATES_PATH),
        "--review-csv", str(REVIEW_CSV_PATH),
        "--out-dir", str(REVIEWED_DIR)]
    completed = subprocess.run(command, cwd=ROOT, check=True, text=True, capture_output=True)
    print(completed.stdout)
else:
    print("待人工逐题审核；审核前不会晋升为正式训练/评估 split。")
上一章

第 44 章：RLCD 原理与实验优化
下一章

第 46 章：中文 Head 微调实验

📑 Jev Cookbook：System One 判断模型实战教程

1 第 1 章：认识 Jev：模型、上手与场景 2 第 2 章：核心概念总览 3 第 3 章：System One：判断的核心心智模型 4 第 4 章：状态：让判断连续可追溯 5 第 5 章：原语：Choice、Score 与 Noul 6 第 6 章：置信度：让概率可信 7 第 7 章：应用构建：从原语到完整系统 8 第 8 章：架构模式 9 第 9 章：实战指南总览 10 第 10 章：自一致性 · Noul 11 第 11 章：自一致性 · Choice 12 第 12 章：并行提问 13 第 13 章：重排序 14 第 14 章：逐行语义搜索 15 第 15 章：结构恢复 16 第 16 章：函数调用 17 第 17 章：技能推荐 18 第 18 章：实体对齐 19 第 19 章：RAG 段落分类 20 第 20 章：引用核查 21 第 21 章：LLM 防护栏 22 第 22 章：SDE 级联 23 第 23 章：日期抽取 24 第 24 章：预解析值抽取 25 第 25 章：层级分类 26 第 26 章：自动研究特征发现 27 第 27 章：基于置信度的分类 28 第 28 章：智能家居实验 29 第 29 章：模型评测总览 30 第 30 章：模型评测实验 31 第 31 章：Laya vs Jev 对比基准 32 第 32 章：JevBench：LLM 评测体系 33 第 33 章：智能家居应用实战 34 第 34 章：Jev-Mem 研究总览 35 第 35 章：Jev-Mem 缩放实验 36 第 36 章：Jev-Mem 完整走查 37 第 37 章：Agent 集成总览 38 第 38 章：Pi 集成实验 39 第 39 章：DSH 决策协作 40 第 40 章：本地模型总览 41 第 41 章：本地模型介绍与对比 42 第 42 章：中文数据集构建方案 43 第 43 章：微调指南 44 第 44 章：RLCD 原理与实验优化 45 第 45 章：中文数据集构建实验 46 第 46 章：中文 Head 微调实验 47 第 47 章：全量 v2 微调实验 48 第 48 章：知识库
← 返回本书大纲