3.1 按单备料：面向 AI 的数据需求分析

3.1 按单备料：面向 AI 的数据需求分析

本节摘要：数据需求分析是把模型侧的一句话翻译成采集侧任务单的过程，分四步——定任务、定字段、定量级、定分布。翻译质量的检验标准只有一个：下游按这份单子备的料，能否直接对上训练与评测的口径。

质检流水线的第一道工序在流水线之前。3.1 之所以排在所有处理工序前面，是因为字段与量纲一旦定错，后面每一道工序都在放大错误。这一节承接第 2 章的产出（能抓到数据），回答"到底该抓什么、要多少、长什么样"。

一句需求的三次翻译

"我们要训练一个懂法律的问答模型。"——这句话要经过三次翻译才能进调度室。第一次翻译定任务：问答模型？检索增强还是微调？前者要知识库语料，后者要指令对，路线不同、备料完全不同。第二次翻译定字段：微调路线需要"问题、参考答案、依据条款"三元组；检索路线需要"段落原文、来源、生效日期"。第三次翻译定量级与分布：十万条民事、两万条刑事、时间跨度覆盖近十年修法周期——分布要求写在纸上，采集才不会扎堆在容易抓的热门板块。

四步法画成图，就是备料单的骨架：

图 3-1 需求分析四步法与产出物

把四步写成代码

备料单用代码承载，才谈得上被下游程序消费。一个可以直接抄走的结构：

from dataclasses import dataclass, field

@dataclass
class FieldSpec:
    name: str            # 字段名
    dtype: str           # 类型：text / number / categorical / datetime
    required: bool       # 是否必填（决定 3.3 节的校验规则）
    unit: str = ""       # 量纲：如 万元、条、秒；文本字段留空
    example: str = ""    # 示例值，给标注与清洗同事对齐认知

@dataclass
class DataRequirement:
    task: str                              # 任务类型：finetune_qa / rag_kb / cls
    fields: list                           # 字段规格列表
    total: int                             # 目标总量
    source_quota: dict = field(default_factory=dict)   # 来源配比
    distribution: dict = field(default_factory=dict)   # 分布约束

req = DataRequirement(
    task="finetune_qa",
    fields=[
        FieldSpec("question", "text", required=True, example="劳动仲裁的时效是几年"),
        FieldSpec("answer", "text", required=True, example="一年，自权利人知道或应当知道之日起算"),
        FieldSpec("legal_basis", "text", required=False, example="劳动争议调解仲裁法第二十七条"),
        FieldSpec("publish_date", "datetime", required=True, unit="ISO日期"),
    ],
    total=50000,
    source_quota={"司法公开文书": 0.5, "法律咨询社区": 0.3, "律所专栏": 0.2},
    distribution={"民事": 0.6, "刑事": 0.15, "行政": 0.15, "劳动": 0.1,
                  "时间跨度": "2016至2026", "问题长度": "10至200字为主"},
)
print("必填字段:", [f.name for f in req.fields if f.required])
# 输出：必填字段: ['question', 'answer', 'publish_date']
print("来源配比合计:", sum(req.source_quota.values()))
# 输出：来源配比合计: 1.0（配比必须闭合为1，这是需求单的自检项）
​

第二个实践要点：量级要折算成请求预算。五万条成品数据，按司法文书平均每页两条相关记录、清洗损耗百分之二十算，需要抓约三十一万页。这类换算决定工期与成本，写进备料单而不是留在脑子里：

def estimate_page_budget(target_rows: int, rows_per_page: float,
                         clean_loss: float = 0.2, dedup_loss: float = 0.1) -> int:
    """从成品量倒推要抓的页数：清洗与去重的损耗都要计入"""
    survive = (1 - clean_loss) * (1 - dedup_loss)
    pages = target_rows / (rows_per_page * survive)
    return int(pages) + 1

pages = estimate_page_budget(50000, rows_per_page=2.0)
print("需要抓取页数:", pages)          # 输出：需要抓取页数: 34723
print("按日上限2000页:", f"{pages / 2000:.0f} 天")   # 输出：按日上限2000页: 17 天
​

十七天工期在开工前算出来，还可以谈；抓完才发现，就只能压清洗工期了——而压清洗工期的后果，正是 1.4 节质量刻度上的红灯。

分布约束的落地检查

四步里最容易被略过的是第四步，补一个落地办法：切片对齐。把评测集按类别、时间、长度切成若干片，逐片检查备料单的分布约束是否覆盖。民事占六成的训练集配一个全是刑事的评测集，指标再高也是假象。这段检查逻辑在 3.3 节清洗完成后还会再跑一次——备料时查"计划覆盖"，入库后查"实际达成"。

排错：备料单返工的三种典型信号

翻译质量要等下游暴露出来才知道，三种信号提示返工。信号一：清洗车间大量记录卡在同一个字段——多半是 dtype 定错，比如把带"万余元"字样的金额当成 number 收，正确处置是改字段规格（金额改为 text 或定义归一化规则），而不是让清洗工位加特判，特判只会把同一处错误复制到每个环节。信号二：标注组反复来问同一个字段怎么填——说明 example 留空或给得太抽象，补三五个真实样例即可，这一步的成本远低于整批标注返工。信号三：需求方中途加字段——先算增量成本，回答"这个字段要不要重抓历史页面、折成多少页多少天"，把无限接单变成有报价的变更。备料单是活文档，但每次变更都要留版本记录，让下游随时能对上"这批数据按哪一版单子备的"。

备料单定稿，下一道工序是拿着它去评审供货源：网上数据源那么多，哪些配得上这张单子。