ZGCM-1 数据处理（data-process）全流程详解

1 · 导读：两份材料各自说什么

用户问的是“如何操作 data-process 过程”。要回答这个问题，必须先把两份材料的分工讲清楚，否则很容易误读成“代码就是论文的实现”或“论文描述的就是这段代码”。事实不是这样，二者是互补但边界明确的：

1.1 两句话读懂整体目标

• 输入：任意来源的 JSONL 记录（网页 / 代码 / PDF-OCR / 数学 / 指令 / 工具轨迹）。
• 输出：带 decision / reasons / flags / buckets / metrics / record_sha256 的清洗后 JSONL + summary.json + manifest.json，可直接交给下游分片与训练。
• 哲学：共享管道只提供“护栏”，不为所有数据集写死一套规则；由 AI/工程师为每个数据集生成一个 profile，插进同一套决策与审计接口（论文 §6.2 的 L3 “Conditional Autonomy”正是这一形态）。

2 · 数据阶段全景（论文规模 × 代码入口）

论文把数据旅程分为三段：General Pre-Training（两个 data stage）、Mid-Training（三个 context stage）、Post-Training（SFT，其后 RL）。代码用三个 stage 目录一一对应，另加一个通用 CLI。

2.1 Mid-Training 候选池与三段采样

论文附录 A.5 给出：先做池级去重（pool-wide deduplication）与数学错误过滤，得到候选池约 5.3B 条 / 2.86T tokens，再采样成三段累积式 context 混合。代码里这条链的“采样器”是 mix.prefix_sample_mix，“候选池 → 三段”的长度划分器是 midtrain/buckets.assign_length_bucket。

3 · 代码总体架构与执行链

3.1 目录职责

data-process/
├── pyproject.toml                  # 4 个 console script 入口
├── src/zgcm_data_pipeline/
│   ├── models.py                   # Stage / DataCategory / Decision 枚举 + Sample 记录
│   ├── core.py                     # Step / Pipeline / PipelineContext（流式引擎）
│   ├── adapters.py                 # DatasetAdapter：原始字段 → 规范 Sample
│   ├── quality.py                  # QualityPolicy + apply_quality_policy
│   ├── dedup.py                    # simhash64 / hamming_distance / NearDeduper
│   ├── io.py                       # JSONL(.gz/.zst) 读写
│   ├── pipelines.py                # 按类别装配：类别算子 + 公共前后缀
│   ├── training.py                 # token 化/打包/loss mask/配额/分片 接口
│   ├── cli.py                      # 通用类别 CLI
│   ├── common/                     # 稳定门面：把上述能力重新导出为“可复用算子”
│   ├── steps/                      # 6 类类别算子 + common.py（公共前后缀）
│   └── stages/
│       ├── pretrain/               # 3 源族 + gharchive/finepdfs/olmocr profile
│       ├── midtrain/               # 6 类别 + buckets + mix + 3 profile
│       └── posttrain/              # 消息/工具格式 + packing 门面 + 3 profile
├── examples/                       # 最小 JSONL/JSON 输入
├── scripts/smoke_test.py           # 四个 CLI 的端到端冒烟
└── tests/                          # unittest：管道、profile、阶段装配

3.2 执行链（一次 run 的完整生命周期）

4 · 数据模型与流式引擎

4.1 三个枚举（models.py）

4.2 Sample：唯一的规范记录

@dataclass(slots=True)
class Sample:
    dataset: str                      # 数据集名
    source_id: str                    # 源内唯一 ID（缺失时用 "dataset:index" 兜底）
    stage: Stage
    category: DataCategory
    data: dict[str, Any]              # 原始字段 + 规范化字段（不丢源记录）
    text_fields: tuple[str, ...]      # 哪些字段构成“训练文本”
    decision: Decision = KEEP
    reasons: list[str] = []           # 决策理由（去重、保序）
    flags: set[str] = set()           # 审计标注（问题标记）
    buckets: dict[str, str] = {}      # 分桶标签（长度、语言、任务、质量…）
    metrics: dict[str, float] = {}    # 数值指标（分数、token 数、轮次…）

三个方法决定了它的行为：

4.3 PipelineContext / Step / Pipeline（core.py）

@dataclass(slots=True)
class PipelineContext:
    config: dict[str, object]       # 所有阈值/策略都从这里读
    counters: Counter[str]          # 每步 seen / keep / review / drop 计数
    seen_hashes: set[str]           # 精确去重状态（跨整轮）
    near_dedup_state: object | None # 惰性创建的 NearDeduper

@dataclass(frozen=True, slots=True)
class Step:
    name: str; fn: StepFn; description: str
    def __call__(self, sample, context):
        before = sample.decision
        sample = self.fn(sample, context)
        context.count("step." + self.name + ".seen")
        if sample.decision != before:
            context.count("step." + self.name + "." + sample.decision.value)
        return sample

4.4 DatasetAdapter：原始字段 → 规范记录

@dataclass(frozen=True, slots=True)
class DatasetAdapter:
    dataset: str
    stage: Stage
    category: DataCategory
    id_field: str = "id"
    text_fields: tuple[str, ...] = ("text",)
    field_map: Mapping[str, str] | None = None   # 源字段名 → 规范字段名
    hooks: tuple[Hook, ...] = ()                 # 数据集专属修复，先于共享管道

    def adapt(self, raw, index=0):
        mapped = dict(raw)
        if self.field_map:
            for src, canonical in self.field_map.items():
                if src in mapped and canonical not in mapped:
                    mapped[canonical] = mapped[src]      # 不覆盖已有规范字段
        source_id = str(mapped.get(self.id_field) or self.dataset + ":" + str(index))
        sample = Sample(...)
        for hook in self.hooks:      # 按顺序执行 hook
            sample = hook(sample)
        return sample

适配器是论文所说“AI 为每个数据集生成独立 adapter/profile”的落点：字段映射与数据专属修复都在这里，而共享的 schema / 质量 / 去重 / 长度 / manifest 行为保持复用。

5 · 处理契约：14 步规范顺序

5.1 装配公式

pipelines.build_pipeline 是整份代码里最需要理解的一行：

steps = category_steps[:1] + COMMON_PREFIX + category_steps[1:] + COMMON_SUFFIX

5.2 完整步骤序列（以 instruction 为例）

5.3 输出记录契约

每条落盘记录至少包含（README 与 release_record 共同定义）：

{
  "dataset": "example",
  "source_id": "row-1",
  "stage": "midtrain",
  "category": "instruction",
  "decision": "keep",          // keep | review | drop
  "reasons": [],               // 决策理由（如 quality_score_below_drop_threshold）
  "flags": [],                 // 审计标注（如 missing_license, possible_secret）
  "buckets": {},               // 长度 / 质量 / 语言 / 任务… 分桶
  "metrics": {},               // quality_score / estimated_tokens / turns …
  "data": {},                  // 源记录 + 规范化字段
  "record_sha256": "..."       // 对 data 排序 JSON 的 sha256
}

默认丢弃记录不写出，但计入 summary；用 --keep-dropped 可保留以便排查。这正是论文“AI 检查数据组成与分布统计，若不符预设标准则自动迭代脚本”所依赖的可观测面。

6 · 通用规范化与完整性算子

这些算子在 steps/common.py，是所有类别共享的“公共前后缀”。

6.1 normalize_text_value（文本规范化）

value = unicodedata.normalize("NFKC", value.replace("\r\n","\n").replace("\r","\n"))
value = _CONTROL.sub("", value)                      # 去掉控制字符
value = "\n".join(_WHITESPACE.sub(" ", line).strip() for line in value.splitlines())
value = _BLANK_LINES.sub("\n\n", value).strip()      # 3+ 连续空行压成 2

6.2 validate_basic_integrity

注意：过短是丢弃，过长是送审——超长记录通常不是噪声，而是需要单独处理的特殊样本。

6.3 validate_provenance（来源与许可）

• 若 require_license=True 且记录无 license（也会查 metadata.license）→ flags += missing_license，REVIEW。
• 若 data.source 缺失 → flags += missing_source，REVIEW。
• 开启方式：CLI 的 --require-license，或配置文件里的 require_license。

这对应论文“来源、修订、许可证、校验和的登记”（docs/dataset_category_inventory.md 的 cross-category 清单），也是可复现 release 的硬要求。

6.4 scan_sensitive_content（敏感内容）

_SECRET = re.compile(r"(?i)(?:api[_-]?key|secret|password|access[_-]?token)\s*[:=]\s*[\"']?[A-Za-z0-9_\-/.]{12,}")
# 命中 → flags += possible_secret, REVIEW
# 另支持 config["sensitive_terms"] 列表，大小写不敏感子串匹配 → flags += sensitive_term, REVIEW

6.5 exact_deduplicate（精确去重）

normalized = normalize_text_value(sample.text()).casefold()
digest = sha256(normalized)
sample.data.setdefault("text_sha256", digest)
if digest in context.seen_hashes: mark(DROP, "exact_duplicate")
else: context.seen_hashes.add(digest)

去重键是规范化 + 小写化后的文本，因此对大小写、全半角、换行差异免疫。状态存在 PipelineContext.seen_hashes，一次 run 内跨所有记录生效。

6.6 estimate_tokens_and_length_bucket

• 若 config["token_estimator"] 是可调用对象 → 用它；否则回退到 max(1, (len(text)+3)//4)（约 4 字符/token 的可复现估计）。
• 写入 metrics["estimated_tokens"]。
• 默认边界 (2048, 8192, 32768, 65536)，标签形如 1_2048、2049_8192、32769_65536、gt_65536；边界会被排序去重。

6.7 finalize_quality_bucket 与 release_record

# finalize_quality_bucket
bucket = "main" if KEEP else "review" if REVIEW else "drop"
sample.buckets.setdefault("quality", bucket)

# release_record（落盘前最后一步）
result = sample.as_dict()
result["record_sha256"] = sha256(json.dumps(result["data"], ensure_ascii=False, sort_keys=True, default=str))

7 · 去重算法：精确 + SimHash 近似

7.1 simhash64：确定性 64 位指纹

def simhash64(text, *, ngram=5):
    tokens = _TOKEN.findall(text.casefold())        # \w+（Unicode）
    if not tokens: return 0
    width = max(1, int(ngram))
    features = (" ".join(tokens[i:i+width]) for i in range(max(1, len(tokens)-width+1)))
    weights = [0]*64
    for feature in features:
        digest = blake2b(feature.encode(), digest_size=8)  # 64 位哈希
        for bit in range(64):
            weights[bit] += 1 if digest & (1<<bit) else -1
    result = 0
    for bit, w in enumerate(weights):
        if w >= 0: result |= (1<<bit)
    return result

算法要点：把文本切成 词 5-gram；每个 gram 用 BLAKE2b-64 得到 64 位；对每一位做 ±1 投票（位为 1 记 +1，否则 −1）；最终按符号定该位（≥0 取 1）。相同文本必然得到相同指纹，相似文本得到汉明距离很小的指纹。

7.2 NearDeduper：带索引的流式近似去重（附交互演示）

朴素做法是“新指纹与历史全部两两比较”，复杂度 O(n²)。这里用 band 索引把候选集缩到极小：

threshold, bands = 3, 4                 # 不变量：bands 必须整除 64，且 threshold < bands
width = 64 // bands                     # 16 位/段
values = [(fp >> (band*width)) & mask for band in range(bands)]

# check_and_add(text)：先算指纹 → 取 4 段 → 只在“共享任一段”的候选里比汉明距离
candidates = set().union(*[indexes[b].get(values[b], ()) for b in range(bands)])
duplicate = any(hamming_distance(fp, fingerprints[i]) <= threshold for i in candidates)
# 无论是否重复，都把新指纹登记进 4 个 band 索引

配置：near_dedup_threshold（默认 3）、near_dedup_bands（默认 4）、near_dedup_ngram（默认 5）。近似去重默认关闭（near_dedup=False），因为它对阈值敏感、必须按类别校准。开启方式：examples/quality-policies.json 里写 "near_dedup": true。

8 · 质量策略（阈值、硬错误、分档）

quality.py 把“质量判断”抽象成一份可按数据源或类别配置的策略，而不写死任何数据集。

8.1 QualityPolicy 字段

8.2 策略解析优先级

key = data["source_key"] or data["source"] or dataset or category
policy = policies[key] or policies[category] or policies["default"]

即：源特定 → 类别 → default。这让同一类别下的不同来源可以有不同的严格度。

8.3 应用逻辑

score = _number(data[score_field])      # bool 被显式排除；字符串会尝试转 float
if score is None:
    if require_score: flags.add("missing_quality_score"); mark(REVIEW, "missing_quality_score")
else:
    metrics["quality_score"] = score
    if review_min is not None and score < review_min: mark(DROP,   "quality_score_below_drop_threshold")
    elif keep_min is not None and score < keep_min:   mark(REVIEW, "quality_score_below_keep_threshold")
if hard_error_field and hard_error_max is not None:
    he = _number(data[hard_error_field])
    if he is not None:
        metrics["hard_error_score"] = he
        if he > hard_error_max: mark(DROP, "hard_error_score_exceeded")

8.4 示例配置（examples/quality-policies.json）

{
  "near_dedup": true,
  "near_dedup_threshold": 3,
  "quality_policies": {
    "web":  { "score_field": "quality_mean", "keep_min": 4.0, "review_min": 3.0, "require_score": false },
    "math": { "score_field": "quality_score", "keep_min": 0.8, "review_min": 0.5,
              "hard_error_field": "hard_error_score", "hard_error_max": 3 }
  }
}

9 · Pretrain 阶段详解

入口：stages/pretrain/cli.py，装配在 pipeline.py。支持三个源族，装配顺序与通用管道不同：

steps = dataset_steps(profile, source) + source_steps + COMMON_PREFIX + COMMON_SUFFIX
#           ① 数据集档案（可选）      ② 源族归一化       ③④ 公共前后缀

9.1 三个源族（source_cleaning.py）

9.2 Profile：gharchive（代码源，最完整的一套过滤）

reject_path_reason 是逐条路径的分类器，返回第一个命中的原因：

此外还有内容层过滤：\x00（前 8192 字符内）→ DROP gharchive_binary_nul；前 2000 字符命中 GENERATED_MARKERS（auto-generated / autogenerated / code generated / do not edit / do not modify / generated by）→ DROP gharchive_generated_file；is_minified（长度 ≥2000 且最长行 >200k，或 <10 行且平均行 >1000）→ DROP gharchive_minified_or_long_line。

• Notebook 处理：normalize_notebook 只保留 code/markdown cell 的 source，丢弃 execution outputs 与 metadata，并用 # code cell / <!-- markdown cell --> 标记。
• 语言识别：LANGUAGE_BY_EXTENSION（约 30 种后缀）+ SPECIAL_NAMES（dockerfile / makefile / cmakelists.txt / rakefile / gemfile）。
• 出处指纹：metadata.blob_id = "sha256:<text sha256>"，并写回 path/language，buckets["gharchive_bucket"]="core"。

9.3 Profile：finepdfs（PDF/OCR 语言 + 重叠 + 质量）

另生成 data.finepdfs_dedup_key：优先 id:<id>，否则用 url/dump/file_path/offset 组成的 provenance 键，再否则用文本 sha256；buckets["finepdfs_policy"]="relaxed_lid085_mh2_p08" 把“用的是哪版策略”写进记录，便于追溯。

9.4 Profile：olmocr（科学 PDF 压缩比过滤）

ratio = len(zlib.compress(text.encode("utf-8"), 6)) / len(raw)   # 0.0 若空
# ratio < olmocr_min_compression_ratio(0.08) → DROP olmocr_low_compression_ratio
# edu_score < olmocr_min_edu_score(0.6)      → DROP olmocr_low_edu_score
#                                               （非法值 → REVIEW olmocr_invalid_edu_score）
# buckets["olmocr_policy"] = "compression_ratio_ge_0p08"

10 · Midtrain 阶段详解

入口：stages/midtrain/cli.py；类别 = code / web / instruction / agentic / math / reasoning。装配：

steps = profile_steps + category_pipeline.steps        # profile 前置，类别管道已含公共前后缀

10.1 长度分桶（buckets.py）——论文三段 context 的落点

def assign_length_bucket(token_count, boundaries=(16_384, 65_536, 262_144)):
    for upper in boundaries:
        if value <= upper: return "B" + str(upper // 1024)   # B16 / B64 / B256
    return "gt_" + str(lower // 1024)                        # gt_256

10.2 前缀采样（mix.py）——互斥分段混合

DEFAULT_ELIGIBLE = {
    "mix16":  ("B16",),
    "mix64":  ("B16", "B64"),
    "mix256": ("B16", "B64", "B256"),
}

prefix_sample_mix 按 mix16 → mix64 → mix256 的规范顺序处理；每段先从“允许的基础桶”里剔除已被前序 mix 选走的 ID，再用 blake2b(seed:final_name) 稳定排序取前 target 个。性质：

• 前缀可用：短样本可进入更长的 mix（mix64 含 B16，mix256 含 B16/B64）。
• 互斥：任一 ID 只出现在一个最终 mix 里（防止同一数据被重复计入不同 context 阶段）。
• 可复现：默认 seed "zgcm-midtrain"；候选不足会抛错，绝不静默少采。

10.3 三个 Midtrain Profile

web_knowledge（类别 web）

nemotron_math（类别 math）

• record_drop_key 归一出行级谱系键 (source, row_index)（支持 source_rel/source_file/source_path 与 row_index/file_row_index）。
• normalize_drop_keys 接受字符串（"src:12" 或 tab 分隔）、序列、映射三种配置形式。
• 硬错误分来自 hard_error_score / math_hard_error_score / error_risk_score，默认阈值 4.0；命中配置键或分数 ≥ 阈值 → DROP nemotron_math_strict_hard_error；两者都缺 → REVIEW nemotron_math_missing_hard_error_signal。
• buckets["hard_error_risk"] = "high" | "accepted"。

agent_coding（类别 agentic）

• 粗估 token：rough_tokens = chars / 3.6；桶 le16k / gt16k_le64k / gt64k_le256k / gt256k。
• 去重优先级：agent_coding_same_repo_multiissue_pack=0 < agent_coding_trace64k_packed_full=1 < agent_coding_nonzerohero_singletrace=2 < 其它=3（dedup_priority）。
• 近似去重采样文本：超 36000 字符时取“头 12k + 中 12k + 尾 12k”拼接后再哈希（near_dedup_sample_sha256），避免只比前缀。
• 写 text_chars / rough_tokens_chars_div_3_6 / length_bucket / dedup_text_sha256 / pre_rendered_agent_trace=True / training_tags / completion_status / source_trace_keys。
• 完成状态不在 {completed_success, success, passed} → REVIEW agent_coding_unverified_completion。

11 · Posttrain（SFT）阶段详解

入口：stages/posttrain/cli.py；类别 = instruction / agentic / math / reasoning / code / web。装配：

steps = profile_steps + POSTTRAIN_FORMAT_STEPS + category_pipeline.steps
#        ① 数据集档案       ② 消息/工具格式规范化      ③ 类别管道（含公共前后缀）

11.1 消息与工具格式（format.py）

_ROLE_ALIASES = {"human":"user", "gpt":"assistant", "model":"assistant", "function":"tool"}

def split_think_content(content):
    m = _THINK.search(content)          # 匹配 think 标签包裹的推理块
    if not m: return None, content.strip()
    reasoning = m.group(1).strip()
    visible   = _THINK.sub("", content).strip()
    return reasoning or None, visible

• normalize_messages：角色别名归一、非字符串 content 转字符串、把 assistant 的第一个 think 块抽到 reasoning_content，并记录 metrics["turns"]。
• normalize_tool_calls：若 call 里嵌套 function{name,arguments} 则上提 name/arguments；arguments 非字符串则转字符串；不强制任何厂商 schema。

11.2 Profile：dolci_think（思考数据，8 个来源白名单）

parse_assistant 对 assistant 内容做严格解析，返回 (reasoning, answer, error)：无完整 think 块 → missing_complete_think；think 为空 → empty_think；可见回答为空 → empty_visible_answer；出现多个 think 块 → multiple_think_blocks。此外：

• TEMPLATE_TOKEN_RE 命中外部模板 token → strict 时 DROP dolci_think_template_token，否则 REVIEW。
• has_repetition：≥120 个词且 5-gram 重复率 > 20% → DROP dolci_think_repetition。
• 渲染 token 粗估（词 / 汉字计数 + 每个 assistant +4）超 dolci_think_max_rough_tokens(32768) → DROP dolci_think_over_length；超 16000 → REVIEW dolci_think_long_record。

11.3 Profile：dolci_tooluse（legacy 工具调用 → 原生 tools/messages）

1. load_functions：把 system 里的 functions（JSON 字符串或列表）转成 {"type":"function","function":{...}}。
2. strip_legacy_system：剥离 “You are a helpful function-calling AI assistant…” 这段 legacy 系统提示；若残留 <functions>/<function_calls> 则整段清空。
3. split_calls：用一个带引号 / 转义状态机的扫描器把 weather(city='Beijing') 这类文本按顶层括号切分；parse_arguments 用 ast.parse("f(" + text + ")", mode="eval") 解析实参（位置参数 → argN，关键字参数 → 原名）。
4. 角色映射：system 的 functions 上提到顶层 tools；environment → tool；带 function_calls 的 assistant → 带 tool_calls 的 assistant。
5. validate_dolci_tooluse_record：校验角色集合、system 必须在最前、assistant 前一条必须是 user/tool、tool 前一条必须是 assistant/tool、assistant 不能既无 content 又无 tool_calls、每个 call 必须有 name + dict arguments。

11.4 Profile：ultradata（消息修复）

• FOREIGN_TOKENS（<|end_header_id|>、<|start_header_id|>、<|model|>、<|messages|>、<|im_start|>、<|im_end|>、<|eot_id|>）出现在 content / reasoning_content / reasoning → DROP ultradata_foreign_chat_template_token。
• 空的 system 消息移除（flags += ultradata_removed_empty_system）。
• repair_visible_assistant_content：若 assistant 的可见 content 里漏了 think 结束标签，取最后一个结束标签之后的内容作为可见答案；修完仍为空 → DROP ultradata_empty_after_think_repair。
• reasoning 字段里的 think 标签剥离（记 flag）。

12 · 训练侧接口：配额、分片、打包、loss mask

这些是“清洗完成到可训练”的桥梁，都在 training.py（并通过 common/ 与 posttrain/packing.py 重新导出）。它们是与 tokenizer 无关的纯算法接口。

12.1 负载均衡与确定性分片

12.2 混合配额与选择

def allocate_mix_quotas(weights, total_samples):
    exact = {k: total*w/sum(w)}
    quotas = {k: floor(v)}; remaining = total - sum(quotas)
    for k in sorted(k, key=(-(exact[k]-quotas[k]), k))[:remaining]: quotas[k] += 1
    return quotas              # 最大余数法，保证配额之和精确等于 total

def select_mixed_sample_ids(source_to_ids, quotas, seed="zgcm"):
    for source: ranked = sort(ids, key=blake2b(seed + ":" + source)); take ranked[:quota]
    selected.sort(key=blake2b(seed + ":mix"))       # 确定性交织

配额不足会抛 ValueError（source X has N but quota is M），杜绝“悄悄少训”。这一层直接对应论文 §3.1.1 “before each source is sampled according to the final data recipe”。

12.3 token 化样本、loss mask 与打包

IGNORE_INDEX = -100

@dataclass(frozen=True, slots=True)
class TokenizedSample:
    tokens: tuple[int, ...]
    source_id: str
    targets: tuple[int, ...] | None = None   # None = 纯预训练；有值 = SFT

def validate_tokenized_sample(s):
    if not s.tokens: raise ValueError("tokenized sample is empty")
    if s.targets is not None:
        if len(s.tokens) != len(s.targets): raise ValueError("tokens and targets must have the same length")
        if not any(t != IGNORE_INDEX for t in s.targets): raise ValueError("SFT sample has no trainable target token")

def build_loss_mask(targets):                # 0 = 不计损失，1 = 计损失
    return tuple(0 if t == IGNORE_INDEX else 1 for t in targets)

def pack_tokenized_samples(samples, sequence_length, allow_overlength=False):
    # 1) 校验每条；超长进 overlength；不允许则跳过
    # 2) accepted 按 (-length, source_id) 排序（first-fit-decreasing）
    # 3) free 列表 + bisect 找“剩余空间足够”的 bin；否则新开 bin
    # 4) 追加 tokens/targets，记录 cu_seqlens（累积边界），更新 free
    # 5) 禁止在一次调用里混装 pretrain 与 SFT 样本
    return bins, overlength

13 · 六类类别算子逐条拆解

13.1 Code（steps/code.py）

13.2 Web（steps/web.py）

• normalize_web_schema：question+answer 都是字符串 → text_fields=(question,answer)、subtype=web_qa；否则 subtype=web_text。
• filter_web_boilerplate：模板词（cookie policy / accept all cookies / privacy policy / sign in / subscribe now）命中 ≥ web_boilerplate_limit(4) → REVIEW web_boilerplate；非空行去重率 < 0.45 → DROP repetitive_web_text。
• validate_web_qa：question <4 或 answer 空 → DROP incomplete_qa_pair；answer 命中 unknown / n/a / i don't know / 无法回答 / 不知道 → DROP unknown_answer；grounded=False → REVIEW ungrounded_answer。
• bucket_web：topic 分桶；quality_score ≥0.8 / ≥0.5 → high / medium / low。

13.3 Instruction（steps/instruction.py）

• normalize_instruction_schema：instruction→prompt、output→response；messages 列表则拼成 role: content 的 text。
• validate_instruction_pair：messages 缺 user/assistant → DROP incomplete_instruction_messages；prompt/response 缺失 → DROP incomplete_instruction_pair；两者完全相同 → DROP prompt_response_identical。
• filter_instruction_templates：response 命中 as an ai language model / i cannot assist with / here is the requested response → REVIEW generic_model_template。
• bucket_instruction：task / risk / turns（>2 条 messages 记 multi）。

13.4 Agentic（steps/agentic.py）

• _messages：从 messages 或 trajectory 取字典列表。
• normalize_agentic_schema：无 text 时把 messages 拼成 text；有 text 但无 messages 且非 pre-rendered → REVIEW unstructured_agent_trace；metrics["turns"]；subtype=sample_type。
• validate_role_order：首角色非 system/user → REVIEW invalid_first_role；空角色 → DROP missing_message_role。
• validate_tool_call_pairs（pre-rendered 跳过）：统计 tool_calls 与 observations，配对未闭合 → REVIEW unpaired_tool_call，零调用 → REVIEW no_tool_call。
• detect_trajectory_outcome：失败标记（exit due to / permission denied / tool error / timed out）与成功标记（submitted / tests passed / task completed / success）；buckets["outcome"]=verified/failed/weak；failed 且未 verified → REVIEW。
• bucket_agentic：tool_family 由 tool_names/target_tools 排序去重拼接。

13.5 Math（steps/math.py）

• 字段：problem→question、response→solution；text_fields=(question,solution)。
• 缺题 / 解 → DROP missing_problem_or_solution；出现图片依赖（<img / 图片后缀 / includegraphics / see figure）但无 image_text → REVIEW missing_multimodal_context。
• 解 <32 字符 → REVIEW solution_too_short；无最终答案信号（final answer / answer: / therefore / thus / boxed / 答案 / 所以）且无 answer 字段 → REVIEW missing_final_answer_signal。
• verified True/False/None → buckets verified/failed/unknown；False → REVIEW verifier_failed。
• 分桶：form（textbook/qa）、difficulty、domain。

13.6 Reasoning（steps/reasoning.py）

• 字段：prompt→problem、response→reasoning、final_answer 可选。
• 控制残留（外部 special token、[system prompt]、api_error）→ REVIEW control_or_prompt_artifact。
• reasoning 缺失 → DROP missing_reasoning；<64 字符或无推理信号 → REVIEW weak_reasoning_signal；无最终答案 → REVIEW missing_final_answer。
• 分桶：origin（teacher_rewrite/raw）、domain、verified；source_row_hash → lineage_hash（保留谱系）。

14 · CLI 与运行方式

14.1 四个入口

共同选参：--dataset-profile、--id-field、--text-fields、--config、--require-license、--manifest、--summary、--keep-dropped。

14.2 三个可运行示例

# ① 通用 instruction（无 profile）
PYTHONPATH=src python3 -m zgcm_data_pipeline \
  --input examples/instruction.jsonl \
  --output outputs/instruction.cleaned.jsonl \
  --dataset example_instruction --stage midtrain --category instruction \
  --text-fields instruction,output \
  --config examples/quality-policies.json \
  --manifest outputs/instruction.manifest.json

# ② Pretrain code + gharchive 档案
PYTHONPATH=src python3 -m zgcm_data_pipeline.stages.pretrain.cli \
  --input examples/datasets/gharchive.jsonl \
  --output outputs/gharchive.cleaned.jsonl \
  --dataset gharchive --source code --dataset-profile gharchive

# ③ Posttrain agentic + dolci_tooluse 档案
PYTHONPATH=src python3 -m zgcm_data_pipeline.stages.posttrain.cli \
  --input examples/datasets/dolci_tooluse.jsonl \
  --output outputs/dolci_tooluse.cleaned.jsonl \
  --dataset dolci_tooluse --category agentic --dataset-profile dolci_tooluse

14.3 示例数据的预期行为

14.4 本地验证

PYTHONPATH=src python3 -m unittest discover -s tests -v
python3 -m compileall -q src
python3 scripts/smoke_test.py      # 跑四个 CLI，校验 manifest.records == 输出行数

15 · 论文 ↔ 代码 对照总表

16 · 边界、限制与谁负责什么

16.1 代码明确“不做”的事（README 与各 stage README 均声明）

16.2 需要留意的地方（工程视角）

• token 是估计值：未注入 tokenizer 时 (len+3)//4 只是可复现的近似；长度桶在真实 tokenizer 下可能位移。
• 近似去重默认关闭：必须按类别校准阈值；band/threshold 不满足 64 % bands == 0 且 threshold < bands 会在构造时报错。
• 去重作用域是单次 run：跨 run / 跨阶段的去重由外部组织。
• review_min 语义：名字像“送审线”，实际是丢弃线（below_drop_threshold）；配置时不要按字面理解。
• 决策单调：一旦 DROP 不会再被后续算子“救回”；若希望某步先于质量过滤，应调整步骤顺序而非期待覆盖。
• 两套长度桶并存：公共后缀 1_2048… 与 Midtrain B16/B64/B256 用途不同，不要混用标签。

17 · 复现实操清单

17.1 五分钟跑通

1. 进入 data-process/；确认 Python ≥ 3.10。
2. PYTHONPATH=src python3 -m unittest discover -s tests -v —— 跑通全部单元测试。
3. python3 scripts/smoke_test.py —— 四个 CLI 端到端 + manifest 计数一致性。
4. 按 §14.2 的三条命令跑通三个示例，检查输出 JSONL 的 decision/reasons/buckets/metrics。

17.2 新增一个数据集 profile 的步骤

1. 写一个 clean_<name>_record(sample, context) -> Sample，只做该数据集特有的字段修复 / 过滤 / 指标，并返回 Step(...) 元组。
2. 在该 stage 的 datasets/__init__.py 注册 DATASET_PROFILES["<name>"] = <Profile>(<required source/category>, STEPS)。
3. 用 --dataset-profile <name> 运行；源族 / 类别不匹配会立即报错（严格校验）。
4. 把阈值开放为 context.config 键（如 gharchive_max_file_bytes），以便不改代码调参。
5. 补一个 examples/datasets/<name>.jsonl 并加一条 unittest，锁定预期决策。