【案例】大健康行业智能问诊系统 · 语料与数据集构建

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""把「一问一答」的原始语料，改造成带指令的三元组数据集。

原始语料只有问题和答案两列，模型看到的是「这句后面该接哪句」。
加上 instruction 之后，模型学到的是「在这个任务下，这句后面该接哪句」——
同一个底座就能同时承载分诊引导、科室推荐、术语解释几种任务，
而不是把它们混成一锅。

同时在这一步做清洗：去重、长度过滤、答非所问过滤。
清洗必须在划分训练/验证集之前做完，否则重复样本会横跨两边，
验证集分数虚高，而你还以为模型学得很好。

纯标准库，可直接运行。
"""
import hashlib
import json
import os
import re

from corpus_stats import load_pairs
from deidentify import deidentify

HERE = os.path.dirname(os.path.abspath(__file__))
DATA = os.environ.get(
    "MEDICAL_CORPUS",
    os.path.join(HERE, "..", "..", "src", "health-consult-data", "medical_sample.txt"))

# 按问题的说法，判定它属于哪一类任务，配上对应的指令。
# 不写死成一条指令，是因为同一个底座要承载多种任务；
# 也不逐条人工标，用可维护的规则表来分派。
TASK_RULES = [
    (re.compile(r"(就诊科室|挂什么科|看哪个科|哪个科室)"),
     "根据描述判断应当就诊的科室，并说明判断依据。"),
    (re.compile(r"(临床表现|症状有些什么|有什么症状|表现有些什么)"),
     "列出该疾病的常见临床表现。"),
    (re.compile(r"(辅助治疗|治疗有些什么|怎么治疗|如何治疗|手术治疗)"),
     "介绍该疾病常用的治疗手段，供了解，不替代医生处方。"),
    (re.compile(r"(检查|化验|影像学)"),
     "说明该情况通常需要做哪些检查项目。"),
    (re.compile(r"(发病部位|好发部位|病因|发病原因)"),
     "说明该疾病的发病部位或成因。"),
]
DEFAULT_INSTRUCTION = "根据患者的描述给出健康咨询答复，并在需要时提示就医。"

# 明显不该进训练集的答案
BAD_ANSWER = re.compile(r"^(无|暂无|不详|没有|null|none)$", re.I)


def classify(question):
    """给一条问题配一条指令。"""
    for pat, instruction in TASK_RULES:
        if pat.search(question):
            return instruction
    return DEFAULT_INSTRUCTION


def normalize(text):
    """统一空白与全半角标点，让去重能真正去掉「看起来不一样其实一样」的。"""
    text = text.replace("\u3000", " ").strip()
    text = re.sub(r"\s+", " ", text)
    # 原始语料里中英文逗号句号混用，归一化后再算指纹
    for a, b in ((",", "，"), (";", "；"), ("?", "？"), ("!", "！")):
        text = text.replace(a, b)
    return text


def fingerprint(question):
    """只按问题算指纹：同一个问题配不同答案，也算重复，只留一条。"""
    return hashlib.md5(normalize(question).encode("utf-8")).hexdigest()


def clean(pairs, min_q=4, min_a=6, max_total=400):
    """返回 (保留的样本, 各类丢弃原因的计数)。"""
    kept, dropped = [], {}
    seen = set()

    def drop(reason):
        dropped[reason] = dropped.get(reason, 0) + 1

    for q, a in pairs:
        q, a = normalize(q), normalize(a)

        # 原始语料里有「小孩发烧又吐怎么办？，无」这类把占位符写进问题的
        q = re.sub(r"[，,]\s*无$", "", q)

        if len(q) < min_q:
            drop("问题过短"); continue
        if len(a) < min_a:
            drop("答案过短"); continue
        if BAD_ANSWER.match(a):
            drop("答案是占位符"); continue
        if len(q) + len(a) > max_total:
            drop("超长"); continue
        if normalize(q) == normalize(a):
            drop("问答相同"); continue

        fp = fingerprint(q)
        if fp in seen:
            drop("重复问题"); continue
        seen.add(fp)

        kept.append((q, a))
    return kept, dropped


def to_records(pairs):
    """产出最终的三元组。

    input 放患者原话，instruction 放任务要求——
    这样换任务时只改 instruction，不用重做数据。
    """
    records = []
    for q, a in pairs:
        q_safe, _ = deidentify(q)
        a_safe, _ = deidentify(a)
        records.append({
            "instruction": classify(q_safe),
            "input": q_safe,
            "output": a_safe,
        })
    return records


def dump_jsonl(records, path):
    with open(path, "w", encoding="utf-8") as f:
        for r in records:
            f.write(json.dumps(r, ensure_ascii=False) + "\n")
    return os.path.getsize(path)


if __name__ == "__main__":
    pairs, broken = load_pairs(DATA)
    print("读入 %d 条，格式异常 %d 段" % (len(pairs), len(broken)))

    kept, dropped = clean(pairs)
    print("\n一、清洗结果")
    print("  保留 %d 条（%.1f%%）" % (len(kept), 100.0 * len(kept) / len(pairs)))
    for reason, n in sorted(dropped.items(), key=lambda kv: -kv[1]):
        print("  丢弃 %-10s %d 条" % (reason, n))

    records = to_records(kept)

    print("\n二、指令分派情况")
    dist = {}
    for r in records:
        dist[r["instruction"]] = dist.get(r["instruction"], 0) + 1
    for ins, n in sorted(dist.items(), key=lambda kv: -kv[1]):
        print("  %3d 条  %s" % (n, ins))

    print("\n三、抽两条最终样本")
    for r in records[:2]:
        print("  " + json.dumps(r, ensure_ascii=False)[:150] + "…")

    out = os.path.join(HERE, "medical_sft.jsonl")
    size = dump_jsonl(records, out)
    print("\n已写出 %s（%d 字节，%d 行）" % (os.path.basename(out), size, len(records)))

    # 同一份输入跑两次必须得到同样的结果，否则划分数据集时会串
    again, _ = clean(pairs)
    assert [q for q, _ in again] == [q for q, _ in kept], "清洗过程不是确定性的"
    print("断言通过：清洗过程可复现")

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Dataset + collate_fn：把三元组变成模型真正吃的张量。

两个职责要分清，混在一起是这一步最常见的结构性错误：
  Dataset.__getitem__  只管「第 index 条怎么变成一串 id」，不管长度对齐
  collate_fn           只管「这一批怎么补齐成矩形」，不管单条怎么编码

分开的好处是换 batch_size 不用改编码逻辑，换模板不用改补齐逻辑。
"""
import json
import os

import torch
from torch.nn.utils.rnn import pad_sequence
from torch.utils.data import DataLoader, Dataset
from transformers import AutoTokenizer

HERE = os.path.dirname(os.path.abspath(__file__))
VOCAB = os.environ.get("VOCAB_PATH", "vocab/vocab.txt")
MAX_LEN = int(os.environ.get("MAX_LEN", "352"))

PROMPT_TEMPLATE = (
    "### 指令:\n{instruction}\n\n"
    "### 输入:\n{input}\n\n"
    "### 回复:\n"
)


def load_jsonl(path):
    with open(path, "r", encoding="utf-8") as f:
        return [json.loads(line) for line in f if line.strip()]


class MedicalSFTDataset(Dataset):
    """一条样本 → input_ids + labels。

    labels 与 input_ids 等长，但提示词那一段被置成 -100。
    交叉熵损失会跳过 -100 的位置，所以模型只为「答复」部分负责，
    不会把「### 指令:」这种模板字样也当成要学的目标。
    """

    def __init__(self, records, tokenizer, max_len=MAX_LEN):
        super().__init__()
        self.records = records
        self.tok = tokenizer
        self.max_len = max_len

    def __len__(self):
        return len(self.records)

    def __getitem__(self, index):
        r = self.records[index]
        prompt = PROMPT_TEMPLATE.format(
            instruction=r["instruction"], input=r.get("input", ""))

        # 分别编码，才能知道提示词占了多少个 token。
        # 直接编码拼接好的整句再按字符数去切，中文会切错位置。
        prompt_ids = self.tok.encode(prompt, add_special_tokens=False)
        answer_ids = self.tok.encode(r["output"], add_special_tokens=False)

        # 收尾放一个 [SEP]，推理时模型生成到它就知道该停了
        ids = [self.tok.cls_token_id] + prompt_ids + answer_ids + [self.tok.sep_token_id]
        # 开头的 [CLS] 也属于提示词部分，所以屏蔽长度要 +1
        n_mask = 1 + len(prompt_ids)

        labels = [-100] * n_mask + ids[n_mask:]

        # 截断要同时截 ids 和 labels，两边必须等长
        ids, labels = ids[:self.max_len], labels[:self.max_len]

        return {"input_ids": ids, "labels": labels}


def make_collate(pad_id):
    """按批内最长长度补齐。

    注意两个补齐值不一样：
      input_ids 补 pad_id —— 它要能被 embedding 查表，不能是负数
      labels    补 -100   —— 它要能被损失函数跳过
    补反了不会报错，但会让模型努力去预测一堆填充符。
    """
    def collate(batch):
        ids = [torch.tensor(b["input_ids"], dtype=torch.long) for b in batch]
        labs = [torch.tensor(b["labels"], dtype=torch.long) for b in batch]

        input_ids = pad_sequence(ids, batch_first=True, padding_value=pad_id)
        labels = pad_sequence(labs, batch_first=True, padding_value=-100)
        # 注意力掩码：让模型别去注意补出来的位置
        attention_mask = (input_ids != pad_id).long()

        return {"input_ids": input_ids,
                "labels": labels,
                "attention_mask": attention_mask}
    return collate


def get_dataloaders(train_path, valid_path, batch_size=4):
    tok = AutoTokenizer.from_pretrained(VOCAB)
    pad_id = tok.pad_token_id if tok.pad_token_id is not None else 0

    train_ds = MedicalSFTDataset(load_jsonl(train_path), tok)
    valid_ds = MedicalSFTDataset(load_jsonl(valid_path), tok)
    collate = make_collate(pad_id)

    train_loader = DataLoader(train_ds, batch_size=batch_size, shuffle=True,
                              collate_fn=collate, drop_last=False)
    # 验证集不打乱：每轮评估要在同一个顺序上比，分数才可比
    valid_loader = DataLoader(valid_ds, batch_size=batch_size, shuffle=False,
                              collate_fn=collate, drop_last=False)
    return train_loader, valid_loader, tok


if __name__ == "__main__":
    train_loader, valid_loader, tok = get_dataloaders(
        os.path.join(HERE, "medical_train.jsonl"),
        os.path.join(HERE, "medical_valid.jsonl"))

    print("训练集 %d 批，验证集 %d 批" % (len(train_loader), len(valid_loader)))

    batch = next(iter(train_loader))
    print("\ninput_ids  %s" % (tuple(batch["input_ids"].shape),))
    print("labels     %s" % (tuple(batch["labels"].shape),))
    print("attn_mask  %s" % (tuple(batch["attention_mask"].shape),))

    # 三个张量形状必须完全一致，不一致说明补齐逻辑写歪了
    assert batch["input_ids"].shape == batch["labels"].shape

    # 把第一条的屏蔽情况还原出来看，这是验证标签对不对最直接的办法
    ids, labs = batch["input_ids"][0], batch["labels"][0]
    kept = [i for i, v in enumerate(labs.tolist()) if v != -100]
    print("\n第一条：总长 %d，参与损失 %d 个 token" % (len(ids), len(kept)))
    print("参与损失的内容：%r"
          % tok.decode([ids[i].item() for i in kept]).replace(" ", ""))

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""先量一遍语料，再决定 max_len、batch_size 和要不要截断。

顺序不能反过来：很多人先把 max_len 拍成 512 再开训，
结果要么绝大多数样本被大量 padding 撑着白烧算力，
要么长尾样本被拦腰截断、答案只学到一半。

原始语料的格式是「一行问题、一行答案，段与段之间空一行」。
纯标准库，可直接运行。
"""
import os
import statistics

DATA = os.environ.get(
    "MEDICAL_CORPUS",
    os.path.join(os.path.dirname(os.path.abspath(__file__)),
                 "..", "..", "src", "health-consult-data", "medical_sample.txt"))


def load_pairs(path):
    """读成 [(问题, 答案), ...]，顺带把不成对的段落挑出来。"""
    with open(path, "r", encoding="utf-8") as f:
        raw = f.read()

    pairs, broken = [], []
    for block in raw.split("\n\n"):
        block = block.strip("\n")
        if not block.strip():
            continue
        lines = block.split("\n")
        if len(lines) == 2 and lines[0].strip() and lines[1].strip():
            pairs.append((lines[0].strip(), lines[1].strip()))
        else:
            broken.append(block)
    return pairs, broken


def describe(name, values):
    """打印一列数字的分布。看分布，不要只看平均值。"""
    s = sorted(values)
    p = lambda q: s[min(len(s) - 1, int(len(s) * q))]
    print("  %-8s 条数=%-5d 平均=%-7.1f 中位=%-6d p90=%-6d p99=%-6d 最大=%d"
          % (name, len(s), statistics.mean(s), statistics.median(s),
             p(0.90), p(0.99), s[-1]))
    return p(0.90), s[-1]


def suggest_max_len(pairs, char_to_token=1.1, reserve=8):
    """从字符长度推一个 max_len 建议值。

    中文经 BertTokenizer 基本是一字一 token，但数字、英文和标点会多出一些，
    所以留一个换算系数；reserve 是给 [CLS]/[SEP] 这类特殊符号留的位置。
    取 p99 而不是最大值：为了极少数超长样本把所有样本都撑大不划算。
    """
    totals = sorted(len(q) + len(a) for q, a in pairs)
    p99 = totals[min(len(totals) - 1, int(len(totals) * 0.99))]
    raw = int(p99 * char_to_token) + reserve
    # 对齐到 32 的倍数，方便硬件对齐，也省得每次改数据就换一个怪数字
    return ((raw + 31) // 32) * 32


def padding_waste(pairs, max_len):
    """按「批内补齐到最长」估算 padding 浪费比例。

    这是选 max_len 之外的第二个决策：要不要按长度排序分桶。
    """
    lens = [len(q) + len(a) for q, a in pairs]
    lens = [min(x, max_len) for x in lens]
    batched_cost = 0
    real = sum(lens)
    for i in range(0, len(lens), 8):          # 假设 batch_size=8
        chunk = lens[i:i + 8]
        batched_cost += max(chunk) * len(chunk)
    return real, batched_cost


if __name__ == "__main__":
    pairs, broken = load_pairs(DATA)
    print("语料：%s" % os.path.normpath(DATA))
    print("成对样本 %d 条，格式异常 %d 段\n" % (len(pairs), len(broken)))

    print("一、长度分布（按字符数）")
    describe("问题", [len(q) for q, _ in pairs])
    describe("答案", [len(a) for _, a in pairs])
    describe("合计", [len(q) + len(a) for q, a in pairs])

    max_len = suggest_max_len(pairs)
    print("\n二、max_len 建议值：%d" % max_len)
    over = sum(1 for q, a in pairs if len(q) + len(a) > max_len)
    print("  按这个值，会被截断的样本有 %d 条（%.1f%%）"
          % (over, 100.0 * over / len(pairs)))

    real, cost = padding_waste(pairs, max_len)
    print("\n三、padding 浪费（batch_size=8，批内补齐到最长）")
    print("  真实 token 数 %d，实际算了 %d 个位置" % (real, cost))
    print("  浪费比例 %.1f%%" % (100.0 * (cost - real) / cost))
    print("  → 浪费超过 30% 时值得按长度排序分桶，让长短样本各自成批。")

    print("\n四、抽三条看看内容长什么样")
    for q, a in pairs[:3]:
        print("  问：%s" % (q[:40] + ("…" if len(q) > 40 else "")))
        print("  答：%s\n" % (a[:40] + ("…" if len(a) > 40 else "")))

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""脱敏：把语料里能指向具体某个人的东西抹掉，再进训练集。

为什么必须在「进训练集之前」做：模型会把训练数据里的字面串背下来，
一旦某条记录里带着真实手机号被反复训练，它就有可能在别的对话里被复述出来。
训练完再想把它从权重里摘掉，没有可靠办法——只能重训。
所以脱敏是一道单向闸门，宁可多抹，不可漏放。

纯标准库，可直接运行。
"""
import re

# 每条规则：(名字, 正则, 替换成什么)
# 替换成占位符而不是直接删掉，是为了让句子结构保持完整，
# 否则「请联系」后面凭空少一截，模型会学出半截话。
RULES = [
    ("身份证", re.compile(r"\b\d{17}[\dXx]\b|\b\d{15}\b"), "[身份证]"),
    ("手机号", re.compile(r"\b1[3-9]\d{9}\b"), "[手机号]"),
    ("座机", re.compile(r"\b0\d{2,3}-?\d{7,8}\b"), "[电话]"),
    ("邮箱", re.compile(r"\b[\w.+-]+@[\w-]+\.[\w.]+\b"), "[邮箱]"),
    ("银行卡", re.compile(r"\b\d{16,19}\b"), "[卡号]"),
    ("住院号", re.compile(r"(住院号|门诊号|病历号|就诊卡号)[:：]?\s*\w+"), r"\1[已脱敏]"),
    ("详细地址", re.compile(r"[\u4e00-\u9fa5]{2,8}(省|市|区|县)"
                            r"[\u4e00-\u9fa5\d]{0,20}?(路|街|巷|号|小区|村)"
                            r"[\u4e00-\u9fa5\d]{0,10}"), "[地址]"),
    ("姓名称谓", re.compile(r"(患者|病人|家属)?[\u4e00-\u9fa5]{1,2}"
                            r"(先生|女士|小姐|同志)"), "[某某]"),
]

# 医学上有意义、绝对不能当成敏感信息抹掉的数字模式。
# 这是脱敏最容易过度的地方：把化验值、剂量、尺寸一起抹了，语料就废了。
KEEP = [
    re.compile(r"\d+(\.\d+)?\s*(mm|cm|ml|mg|g|kg|μg|ug|IU|mmol/L|mg/dL|℃|%)", re.I),
    re.compile(r"\d+\s*[×*xX]\s*\d+\s*(mm|cm)?", re.I),
    re.compile(r"\d+\s*(岁|个月|天|周|年|次/分|小时)"),
]


def protect(text):
    """先把要保留的医学数值挖出来换成哨兵，免得被后面的规则误伤。"""
    kept = []

    def stash(m):
        kept.append(m.group(0))
        return "\x00%d\x00" % (len(kept) - 1)

    for pat in KEEP:
        text = pat.sub(stash, text)
    return text, kept


def restore(text, kept):
    for i, v in enumerate(kept):
        text = text.replace("\x00%d\x00" % i, v)
    return text


def deidentify(text):
    """返回 (脱敏后的文本, 命中了哪些规则)。"""
    text, kept = protect(text)

    hits = []
    for name, pat, repl in RULES:
        text, n = pat.subn(repl, text)
        if n:
            hits.append("%s×%d" % (name, n))

    return restore(text, kept), hits


def audit(pairs):
    """批量脱敏并统计，返回 (脱敏后的 pairs, 命中计数)。"""
    out, counter = [], {}
    for q, a in pairs:
        q2, h1 = deidentify(q)
        a2, h2 = deidentify(a)
        for h in h1 + h2:
            name = h.split("×")[0]
            counter[name] = counter.get(name, 0) + int(h.split("×")[1])
        out.append((q2, a2))
    return out, counter


if __name__ == "__main__":
    cases = [
        "患者张先生，手机 13812345678，住院号：ZY20231201，"
        "家住浙江省杭州市西湖区文三路 100 号。",
        "小孩五岁，脐周淋巴结肿大，原来 9.3mm*3.4mm 现在 10mm*5mm，怎么办？",
        "空腹血糖 7.8 mmol/L，体温 38.5℃，服用二甲双胍 500mg 每日两次。",
        "有问题联系 [email protected] 或 021-58776677。",
        "身份证 330106199001011234，卡号 6222021234567890123。",
    ]

    print("一、逐条脱敏效果\n")
    for c in cases:
        out, hits = deidentify(c)
        print("  原文：%s" % c)
        print("  脱敏：%s" % out)
        print("  命中：%s\n" % ("、".join(hits) if hits else "无"))

    print("二、必须保住的医学数值（这几条不能被抹）")
    med = cases[1] + cases[2]
    out, _ = deidentify(med)
    for token in ("9.3mm", "10mm", "5mm", "7.8 mmol/L", "38.5℃", "500mg", "五岁"):
        status = "✅ 保住" if token in out else "❌ 被误伤"
        print("  %-12s %s" % (token, status))

    print("\n三、一条铁律")
    print("  脱敏跑完要人工抽检，不能只看规则命中数。")
    print("  规则只能覆盖有固定格式的东西；自由文本里写的")
    print("  「我是三病区 6 床老李的女儿」这类，正则抓不住，")
    print("  得靠抽检加人工复核补上。")

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""划分训练集与验证集，并检查有没有泄漏。

验证集存在的唯一意义，是给出一个模型「没见过的题」上的分数。
一旦训练集里的样本以任何形式出现在验证集里，这个分数就失去意义——
更糟的是它会偏高，让你以为模型很行，上线才发现不行。

三种泄漏，按隐蔽程度从低到高：
  ① 完全重复    同一条被抄了两份，一份进训一份进验
  ② 问题重复    同一个问题配了两个不同答案
  ③ 近似重复    只差几个字，正则去重抓不到

纯标准库，可直接运行。
"""
import hashlib
import json
import os
import random
import re

HERE = os.path.dirname(os.path.abspath(__file__))
SRC = os.path.join(HERE, "medical_sft.jsonl")


def load_jsonl(path):
    with open(path, "r", encoding="utf-8") as f:
        return [json.loads(line) for line in f if line.strip()]


def norm(text):
    text = re.sub(r"\s+", "", text)
    return re.sub(r"[，。？！、；：,.?!;:]", "", text)


def shingles(text, n=4):
    """把一句话切成长度为 n 的字符窗口集合，用来算近似重复。"""
    t = norm(text)
    if len(t) <= n:
        return {t}
    return {t[i:i + n] for i in range(len(t) - n + 1)}


def jaccard(a, b):
    """两个集合的重合度。1 表示完全一样，0 表示毫无重合。"""
    if not a or not b:
        return 0.0
    return len(a & b) / len(a | b)


def split(records, valid_ratio=0.1, seed=42):
    """按「问题指纹」分组后再划分，保证同一个问题不会横跨两边。"""
    groups = {}
    for r in records:
        key = hashlib.md5(norm(r["input"]).encode("utf-8")).hexdigest()
        groups.setdefault(key, []).append(r)

    keys = sorted(groups)                 # 先排序，保证结果可复现
    random.Random(seed).shuffle(keys)

    n_valid = max(1, int(len(keys) * valid_ratio))
    valid_keys = set(keys[:n_valid])

    train = [r for k in keys if k not in valid_keys for r in groups[k]]
    valid = [r for k in keys if k in valid_keys for r in groups[k]]
    return train, valid, len(groups)


def check_leak(train, valid, threshold=0.8):
    """返回泄漏清单。threshold 是近似重复的判定阈值。"""
    exact, near = [], []

    train_fp = {hashlib.md5(norm(r["input"]).encode("utf-8")).hexdigest(): r
                for r in train}
    for r in valid:
        fp = hashlib.md5(norm(r["input"]).encode("utf-8")).hexdigest()
        if fp in train_fp:
            exact.append(r["input"])

    # 近似重复要两两比，数据量大时先按长度分桶再比，这里数据小就直接比
    train_sh = [(r["input"], shingles(r["input"])) for r in train]
    for r in valid:
        sh = shingles(r["input"])
        for q, tsh in train_sh:
            score = jaccard(sh, tsh)
            if score >= threshold:
                near.append((r["input"], q, score))
                break
    return exact, near


if __name__ == "__main__":
    if not os.path.exists(SRC):
        raise SystemExit("先跑 build_instruction_set.py 生成 %s" % SRC)

    records = load_jsonl(SRC)
    train, valid, n_groups = split(records)

    print("一、划分结果")
    print("  总样本 %d，去重后问题组 %d 个" % (len(records), n_groups))
    print("  训练集 %d 条，验证集 %d 条（%.1f%%）"
          % (len(train), len(valid), 100.0 * len(valid) / len(records)))

    exact, near = check_leak(train, valid)
    print("\n二、泄漏检查")
    print("  完全重复：%d 条" % len(exact))
    print("  近似重复（Jaccard ≥ 0.8）：%d 条" % len(near))
    for v, t, s in near[:3]:
        print("    %.2f  验证集：%s" % (s, v[:30]))
        print("          训练集：%s" % t[:30])

    assert not exact, "存在完全重复的泄漏，验证集分数不可信"
    print("\n  断言通过：训练集与验证集之间没有完全重复")

    print("\n三、换个随机种子，划分必须不同但依然无泄漏")
    t2, v2, _ = split(records, seed=7)
    same = set(r["input"] for r in valid) == set(r["input"] for r in v2)
    print("  seed=42 与 seed=7 的验证集是否相同：%s" % ("是" if same else "否"))
    e2, _ = check_leak(t2, v2)
    print("  seed=7 的完全重复数：%d" % len(e2))

    print("\n四、写出")
    for name, part in (("train", train), ("valid", valid)):
        p = os.path.join(HERE, "medical_%s.jsonl" % name)
        with open(p, "w", encoding="utf-8") as f:
            for r in part:
                f.write(json.dumps(r, ensure_ascii=False) + "\n")
        print("  %s  %d 行  %d 字节"
              % (os.path.basename(p), len(part), os.path.getsize(p)))

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""可复制改造的数据管道骨架：原始语料 → 可训练的 jsonl。

六道工序，每道都留了一个「跑完必须看一眼」的检查点。
数据管道出错不会报错，只会让训练结果莫名其妙地差，
所以每一步都要产出一个能当场核对的数字。

用法：
    export RAW_CORPUS=<原始语料路径>
    python3 skeleton_data_pipeline.py
"""
import json
import os
import re

RAW_CORPUS = os.environ.get("RAW_CORPUS", "TODO-填原始语料路径.txt")
OUT_DIR = os.environ.get("OUT_DIR", "data")

# TODO: 按你的任务改。instruction 描述任务，不描述具体这一条的内容
DEFAULT_INSTRUCTION = "TODO-填这个任务的指令"

# TODO: 按你的领域补敏感信息规则
DEID_RULES = [
    (re.compile(r"\b1[3-9]\d{9}\b"), "[手机号]"),
    (re.compile(r"\b\d{17}[\dXx]\b"), "[身份证]"),
]


def step1_load(path):
    """① 读入原始语料，统一成 [(问题, 答案), ...]。"""
    with open(path, "r", encoding="utf-8") as f:
        raw = f.read()

    pairs = []
    for block in raw.split("\n\n"):
        lines = [x.strip() for x in block.strip().split("\n") if x.strip()]
        if len(lines) == 2:
            pairs.append(tuple(lines))

    print("① 读入 %d 条成对样本" % len(pairs))
    assert pairs, "一条都没读到，检查语料的分隔方式是不是空行"
    return pairs


def step2_deidentify(pairs):
    """② 脱敏。这一步是单向闸门，必须在进训练集之前完成。"""
    out, hit = [], 0
    for q, a in pairs:
        for pat, repl in DEID_RULES:
            q, n1 = pat.subn(repl, q)
            a, n2 = pat.subn(repl, a)
            hit += n1 + n2
        out.append((q, a))

    print("② 脱敏命中 %d 处" % hit)
    print("   ⚠ 命中数为 0 不代表没有敏感信息，要人工抽检 30 条确认")
    return out


def step3_clean(pairs, min_q=4, min_a=6):
    """③ 清洗：去掉短的、空的、重复的。"""
    kept, seen, dropped = [], set(), {}

    for q, a in pairs:
        reason = None
        if len(q) < min_q:
            reason = "问题过短"
        elif len(a) < min_a:
            reason = "答案过短"
        elif q in seen:
            reason = "重复"

        if reason:
            dropped[reason] = dropped.get(reason, 0) + 1
            continue
        seen.add(q)
        kept.append((q, a))

    print("③ 清洗后保留 %d 条（%.1f%%），丢弃明细 %s"
          % (len(kept), 100.0 * len(kept) / max(1, len(pairs)), dropped))
    return kept


def step4_to_records(pairs):
    """④ 组装三元组。"""
    records = [{"instruction": DEFAULT_INSTRUCTION, "input": q, "output": a}
               for q, a in pairs]
    print("④ 组装 %d 条三元组，抽一条：" % len(records))
    print("   " + json.dumps(records[0], ensure_ascii=False)[:120] + "…")
    return records


def step5_length_check(records, max_len):
    """⑤ 量长度，确认 max_len 选得合不合适。"""
    totals = sorted(len(r["instruction"]) + len(r["input"]) + len(r["output"])
                    for r in records)
    p99 = totals[min(len(totals) - 1, int(len(totals) * 0.99))]
    over = sum(1 for t in totals if t > max_len)

    print("⑤ 长度 中位=%d p99=%d 最大=%d；max_len=%d 会截断 %d 条（%.1f%%）"
          % (totals[len(totals) // 2], p99, totals[-1], max_len,
             over, 100.0 * over / len(totals)))
    if over > len(totals) * 0.05:
        print("   ⚠ 截断比例超过 5%，要么调大 max_len，要么回头看是不是有异常长样本")
    return p99


def step6_split(records, valid_ratio=0.1, seed=42):
    """⑥ 划分。按问题去重后再分，避免泄漏。"""
    import random

    by_q = {}
    for r in records:
        by_q.setdefault(r["input"], []).append(r)

    keys = sorted(by_q)
    random.Random(seed).shuffle(keys)
    n_valid = max(1, int(len(keys) * valid_ratio))

    valid = [r for k in keys[:n_valid] for r in by_q[k]]
    train = [r for k in keys[n_valid:] for r in by_q[k]]

    overlap = set(r["input"] for r in train) & set(r["input"] for r in valid)
    print("⑥ 训练 %d / 验证 %d，重叠问题 %d 个" % (len(train), len(valid), len(overlap)))
    assert not overlap, "训练集和验证集有重叠，验证分数不可信"
    return train, valid


def dump(records, path):
    os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
    with open(path, "w", encoding="utf-8") as f:
        for r in records:
            f.write(json.dumps(r, ensure_ascii=False) + "\n")
    print("   写出 %s（%d 行，%d 字节）"
          % (path, len(records), os.path.getsize(path)))


def main():
    pairs = step1_load(RAW_CORPUS)
    pairs = step2_deidentify(pairs)
    pairs = step3_clean(pairs)
    records = step4_to_records(pairs)
    step5_length_check(records, max_len=352)
    train, valid = step6_split(records)

    dump(train, os.path.join(OUT_DIR, "train.jsonl"))
    dump(valid, os.path.join(OUT_DIR, "valid.jsonl"))
    print("\n完成。开训之前最后一件事：人工读 20 条 train.jsonl，")
    print("确认 instruction 描述的确实是你要模型学的那件事。")


if __name__ == "__main__":
    main()