DeepSeek 实战：思维链数据生产线

# 拿到的两段文本
think  = response.json()['choices'][0]['message']['reasoning_content']   # 念白
result = response.json()['choices'][0]['message']['content']             # 菜品

# 拼成一条 output：标记 + 念白 + 标记 + 三个换行 + 结论
data["output"] = "<think>\n" + think + "\n</think>\n\n\n" + result

# 拼出来的实际长相（省略号处是真实的推理文字）
<think>
好的，用户让我给出“wander”这个词的三个近义词。首先，我需要回忆一下这个词的中文释义……
总结下来，三个近义词应该是“游荡”、“闲逛”和“徘徊”。
</think>


“wander”（漫游、闲逛、徘徊）的三个近义词包括：
1. 游荡：指不规则地走动，多用于形容人在空闲时随意游走。
2. 闲逛：指随意地在某个场所走动，多指在不特定的目的下漫无目的地走动。
3. 徘徊：指在某个地方来回走动，常用于描述在某一地点停留或思考。

{"instruction": "夫妻共同财产的范围在法律中有哪些明确列举？", "input": "", "output": "<think>\n用户问的是夫妻共同财产的范围……需要先区分法定共同财产与约定财产制，再逐项列举……\n</think>\n\n\n夫妻共同财产主要包括：一、工资、奖金、劳务报酬；二、生产、经营、投资的收益……"}
{"instruction": "一方擅自将婚后共同存款赠与他人的行为是否有效？", "input": "", "output": "<think>\n这条要落到无权处分与善意取得上……共同共有意味着单方处分需要另一方同意……\n</think>\n\n\n原则上无效。婚后存款属于夫妻共同财产，一方未经另一方同意的大额无偿赠与……"}
{"instruction": "请给出“wander”这个词的三个近义词？", "input": "", "output": "<think>\n好的，用户让我给出“wander”这个词的三个近义词。首先，我需要回忆一下这个词的中文释义……\n</think>\n\n\n“wander”的三个近义词包括：1. 游荡；2. 闲逛；3. 徘徊。"}

{
  "chat-train": {
    "file_name": "train.jsonl"
  }
}

# -*- coding: utf-8 -*-
"""调用 DeepSeek-R1 拿「思考 + 结论」两段文本，拼成可训练的 output。

要点只有三条：
  1. R1 的返回里 reasoning_content 与 content 是两个独立字段，少取一个就丢掉思维链；
  2. 两段用 <think>...</think> 包起来再接结论，格式固定，学生模型最吃这一套；
  3. key 只从环境变量读，绝不写进源码。

运行前：export SILICONFLOW_API_KEY=你的key
"""
import json
import os
import time

import requests

URL = "https://api.siliconflow.cn/v1/chat/completions"
MODEL = "deepseek-ai/DeepSeek-R1"

# 温度是这一步唯一需要反复权衡的参数：
#   高 -> 答案多样，但跑偏、胡说的概率同步上升
#   低 -> 答案稳定，但成千上万条语料彼此雷同，学生模型学不到泛化
# 0.6 是「有变化但不放飞」的经验取值。
TEMPERATURE = 0.6
MAX_TOKENS = 15000
RETRY = 3          # 网络抖动、网关 5xx 都靠它兜底
RETRY_SLEEP = 2    # 秒，指数退避的基数


def _api_key() -> str:
    key = os.environ.get("SILICONFLOW_API_KEY")
    if not key:
        raise RuntimeError("环境变量 SILICONFLOW_API_KEY 没设置")
    return key


def call_server(prompt, model=MODEL, temperature=TEMPERATURE):
    """返回 (success, msg, think, result)。

    think   -> reasoning_content，模型的推理过程
    result  -> content，模型给用户看的结论
    """
    headers = {
        "Authorization": "Bearer " + _api_key(),
        "Content-Type": "application/json",
    }
    payload = {
        "model": model,
        "messages": [{"role": "user", "content": prompt}],
        "temperature": temperature,
        "max_tokens": MAX_TOKENS,
    }

    msg = "OK"
    for attempt in range(1, RETRY + 1):
        try:
            resp = requests.post(URL, json=payload, headers=headers, timeout=600)
            resp.raise_for_status()
            message = resp.json()["choices"][0]["message"]
            # 关键：两个字段都要取。只取 content 等于把老师的解题过程扔了。
            think = message.get("reasoning_content") or ""
            result = message.get("content") or ""
            if not result:
                raise ValueError("content 为空")
            return True, "OK", think, result
        except Exception as exc:                      # noqa: BLE001
            msg = "第 %d 次调用异常: %s" % (attempt, exc)
            if attempt < RETRY:
                time.sleep(RETRY_SLEEP * attempt)
    return False, msg, "", ""


def build_output(think, result):
    """把思考和结论拼成训练语料的 output 字段。

    格式写死成 <think>\\n…\\n</think> 空两行再接结论：
    大模型对格式性强的结构学得又快又稳，推理时也会照着这个壳子吐。
    """
    return "<think>\n" + think + "\n</think>\n\n\n" + result


def ask_batch(src_path, dst_path):
    """把 jsonl 里每条 instruction 送给 R1，回填 output 后写回 jsonl。"""
    ok = fail = 0
    with open(src_path, "r", encoding="utf-8") as src, \
            open(dst_path, "a", encoding="utf-8") as dst:
        for line in src:
            line = line.strip()
            if not line:
                continue
            data = json.loads(line)
            success, msg, think, result = call_server(data["instruction"])
            if not success:
                fail += 1
                print("失败", data.get("id"), msg)
                continue
            data["output"] = build_output(think, result)
            dst.write(json.dumps(data, ensure_ascii=False) + "\n")
            dst.flush()          # 跑几小时的任务，不 flush 一崩就全没了
            ok += 1
            print("入库", data.get("id"), "思考%d字 结论%d字" % (len(think), len(result)))
    print("完成 成功 %d 条，失败 %d 条" % (ok, fail))


if __name__ == "__main__":
    question = "请给出“wander”这个词的三个近义词？"
    success, msg, think, result = call_server(question)
    print(success, msg)
    print("思考部分：" + think)
    print("结论部分：" + result)
    print("拼好的 output 前 80 字：" + build_output(think, result)[:80])

True OK

思考部分：
好的，用户让我给出“wander”这个词的三个近义词。首先，我需要回忆一下这个词的中文释义。
Wander 通常指的是漫游、闲逛，或者不规则地走动。接下来，我需要想三个相关的中文词，
可能需要考虑不同的侧重点。

首先想到的是“漫游”，因为“漫游”和“走动”相关，但可能不够精确。然后是“游荡”，这个词更强调
不规律地走动，符合“wander”的意思。第三个可能需要更正式一点，比如“徘徊”，用来形容在某个
地方来回走动，可能更符合“wander”的场景，比如游园或者散步。

不过，用户可能需要的是更自然的词汇，所以“游荡”可能更贴近。但需要确认是否有其他更常见的
词。比如“闲逛”也可以，但“游荡”更强调不规则的走动，可能更准确。另外，“徘徊”也可以，但
“游荡”可能更常用。

再想想有没有其他可能性。比如“流连忘返”，但这个词更多是形容时间上的停留，不如“游荡”贴切。
所以可能“游荡”和“闲逛”是合适的。但用户可能需要三个不同的词，所以可能需要调整。

总结下来，三个近义词应该是“游荡”、“闲逛”和“徘徊”。这样既涵盖了不规则走动的方面，
也包括随意走动的情况，可能满足用户的需求。

结论部分：
“wander”（漫游、闲逛、徘徊）的三个近义词包括：
1. 游荡：指不规则地走动，多用于形容人在空闲时随意游走。
2. 闲逛：指随意地在某个场所走动，多指在不特定的目的下漫无目的地走动。
3. 徘徊：指在某个地方来回走动，常用于描述在某一地点停留或思考。

这三个词都表达了“wander”中的“不规则走动”这一核心含义。

# -*- coding: utf-8 -*-
"""用 Qwen2.5-72B-Instruct 给「问答对」打 0~9 分，低分样本直接淘汰。

三个必须守住的设定：
  1. 换一个模型打分。让 R1 给自己的答案打分，等于让考生自己批卷。
  2. temperature 调到 0.2。打分要的是可复现，不是创意。
  3. max_tokens 压到 10。输出长度对吞吐量的影响极大，这里只需要一个数字。

运行前：export SILICONFLOW_API_KEY=你的key
"""
import json
import os
import re

import requests

URL = "https://api.siliconflow.cn/v1/chat/completions"
JUDGE_MODEL = "Qwen/Qwen2.5-72B-Instruct"

# 打分 prompt 的四要素，缺一条分数就开始飘：
#   ① 身份限定：把模型摆到「质量评估专家」的位置上
#   ② 任务说明：0 分什么样、9 分什么样，说死
#   ③ 输出格式限定：只允许输出分数，不许写理由
#   ④ 强格式包裹：【问题】【答案】把两段料分开，模型不会读串
PROMPT_TMPL = (
    "你是一位问答对的质量评估专家，请对以下问答对的质量作出评估，"
    "分数从0到9，0表示答案和问题无关，9表示【答案】非常好地回答了【问题】，"
    "只允许输出分数，不允许输出其他任何字符。\n"
    "【问题】\n{question}\n\n"
    "【答案】\n{answer}\n"
)

KEEP_THRESHOLD = 7   # 只留高分样本


def _api_key() -> str:
    key = os.environ.get("SILICONFLOW_API_KEY")
    if not key:
        raise RuntimeError("环境变量 SILICONFLOW_API_KEY 没设置")
    return key


def score_one(question, answer):
    """返回 (success, msg, score)，score 为 -1 表示没拿到有效分数。"""
    payload = {
        "model": JUDGE_MODEL,
        "messages": [{
            "role": "user",
            "content": PROMPT_TMPL.format(question=question, answer=answer),
        }],
        "temperature": 0.2,
        "max_tokens": 10,
    }
    headers = {
        "Authorization": "Bearer " + _api_key(),
        "Content-Type": "application/json",
    }
    try:
        resp = requests.post(URL, json=payload, headers=headers, timeout=120)
        resp.raise_for_status()
        text = resp.json()["choices"][0]["message"]["content"]
        # 即使限定了输出格式，也要防一手模型多吐一个句号或换行
        m = re.search(r"\d", text)
        if not m:
            return False, "未解析到分数: %r" % text, -1
        return True, "OK", int(m.group())
    except Exception as exc:                          # noqa: BLE001
        return False, "API调用异常: %s" % exc, -1


def filter_by_score(src_path, keep_path, drop_path, threshold=KEEP_THRESHOLD):
    """逐条打分，高分写 keep_path，低分写 drop_path 留档备查。"""
    kept = dropped = broken = 0
    with open(src_path, "r", encoding="utf-8") as src, \
            open(keep_path, "w", encoding="utf-8") as keep, \
            open(drop_path, "w", encoding="utf-8") as drop:
        for line in src:
            line = line.strip()
            if not line:
                continue
            data = json.loads(line)
            success, msg, score = score_one(data["instruction"], data["output"])
            if not success:
                broken += 1
                print("打分失败", data.get("id"), msg)
                continue
            data["score"] = score
            target, counter = (keep, "kept") if score >= threshold else (drop, "dropped")
            target.write(json.dumps(data, ensure_ascii=False) + "\n")
            if counter == "kept":
                kept += 1
            else:
                dropped += 1
    print("保留 %d 条，淘汰 %d 条，打分失败 %d 条" % (kept, dropped, broken))


def reliability(gen_acc=0.80, judge_acc=0.90):
    """两道关口串起来之后的整体准确率估算。

    生成答案是复杂任务，假设 80% 正确；打分是简单任务，假设 90% 正确。
    一条错答案要混进最终数据集，必须「生成错了」且「打分也没看出来」，
    所以留下来的样本估算准确率 = 1 - (1-0.80) x (1-0.90) = 98%。
    这是建立在两个假设之上的估算值，不是实测结果。
    """
    return 1 - (1 - gen_acc) * (1 - judge_acc)


if __name__ == "__main__":
    print("可靠性估算：%.2f%%" % (reliability() * 100))

可靠性估算：98.00%

# -*- coding: utf-8 -*-
"""统计语料长度分布，按累计数量的拐点决定截断阈值。

纯标准库，可直接跑。不带参数时用内置的真实分布做演示。

思路：
  1. 按 1000 字一档统计样本数量；
  2. 算出每一档的累计占比；
  3. 找「再往后放宽一档，只能多收不到 delta 比例样本」的那一档 —— 就是拐点；
  4. 阈值定在拐点，超长样本直接丢。
"""
import json
import sys

BUCKET = 1000

# 一批 15000 余条思维链语料的真实长度分布（区间上界 -> 样本数）
SAMPLE_DIST = [
    (1000, 616), (2000, 4817), (3000, 6660), (4000, 1703), (5000, 439),
    (6000, 228), (7000, 237), (8000, 154), (9000, 118), (10000, 83),
    (11000, 42), (12000, 42), (13000, 33), (14000, 23), (15000, 27),
    (16000, 16), (17000, 13), (18000, 13), (19000, 10), (20000, 7),
    (21000, 10), (22000, 5), (23000, 6), (24000, 3), (25000, 1),
    (26000, 3), (27000, 0), (28000, 6), (29000, 0), (30000, 1),
    (31000, 2), (32000, 0), (33000, 0), (34000, 0), (35000, 1), (36000, 1),
]


def build_dist(path, field="output"):
    """从 jsonl 里统计长度分布，返回 [(区间上界, 数量), ...]。"""
    counter = {}
    with open(path, "r", encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            data = json.loads(line)
            n = len(data.get(field, ""))
            upper = (n // BUCKET + 1) * BUCKET
            counter[upper] = counter.get(upper, 0) + 1
    return sorted(counter.items())


def find_knee(dist, delta=0.005):
    """拐点：某一档之后，每档新增样本占比都低于 delta，就在这里切。"""
    total = sum(c for _u, c in dist)
    acc = 0
    for upper, count in dist:
        acc += count
        if count / total < delta:
            return upper, acc / total
    return dist[-1][0], 1.0


def report(dist, delta=0.005):
    total = sum(c for _u, c in dist)
    knee, knee_ratio = find_knee(dist, delta)
    acc = 0
    print("%-14s %8s %8s %10s" % ("长度区间", "样本数", "占比", "累计占比"))
    for upper, count in dist:
        acc += count
        mark = "  <== 拐点" if upper == knee else ""
        print("%-14s %8d %7.2f%% %9.2f%%%s"
              % ("%d~%d" % (upper - BUCKET, upper), count,
                 count / total * 100, acc / total * 100, mark))
    print()
    print("样本总量：%d" % total)
    print("建议截断阈值：%d 字" % knee)
    print("保留样本占比：%.2f%%，丢弃 %d 条"
          % (knee_ratio * 100, total - round(knee_ratio * total)))
    return knee


def cut(src_path, dst_path, threshold, field="output"):
    kept = dropped = 0
    with open(src_path, "r", encoding="utf-8") as src, \
            open(dst_path, "w", encoding="utf-8") as dst:
        for line in src:
            line = line.strip()
            if not line:
                continue
            data = json.loads(line)
            if len(data.get(field, "")) <= threshold:
                dst.write(json.dumps(data, ensure_ascii=False) + "\n")
                kept += 1
            else:
                dropped += 1
    print("保留 %d 条，丢弃 %d 条" % (kept, dropped))


if __name__ == "__main__":
    if len(sys.argv) > 1:
        report(build_dist(sys.argv[1]))
    else:
        report(SAMPLE_DIST)

长度区间                样本数       占比       累计占比
0~1000              616    4.02%      4.02%
1000~2000          4817   31.44%     35.46%
2000~3000          6660   43.47%     78.94%
3000~4000          1703   11.12%     90.05%
4000~5000           439    2.87%     92.92%
5000~6000           228    1.49%     94.41%
6000~7000           237    1.55%     95.95%
7000~8000           154    1.01%     96.96%
8000~9000           118    0.77%     97.73%
9000~10000           83    0.54%     98.27%
10000~11000          42    0.27%     98.54%  <== 拐点
11000~12000          42    0.27%     98.82%
12000~13000          33    0.22%     99.03%
13000~14000          23    0.15%     99.18%
14000~15000          27    0.18%     99.36%
15000~16000          16    0.10%     99.46%
16000~17000          13    0.08%     99.55%
17000~18000          13    0.08%     99.63%
18000~19000          10    0.07%     99.70%
19000~20000           7    0.05%     99.75%
20000~21000          10    0.07%     99.81%
21000~22000           5    0.03%     99.84%
22000~23000           6    0.04%     99.88%
23000~24000           3    0.02%     99.90%
24000~25000           1    0.01%     99.91%
25000~26000           3    0.02%     99.93%
26000~27000           0    0.00%     99.93%
27000~28000           6    0.04%     99.97%
28000~29000           0    0.00%     99.97%
29000~30000           1    0.01%     99.97%
30000~31000           2    0.01%     99.99%
31000~32000           0    0.00%     99.99%
32000~33000           0    0.00%     99.99%
33000~34000           0    0.00%     99.99%
34000~35000           1    0.01%     99.99%
35000~36000           1    0.01%    100.00%

样本总量：15320
建议截断阈值：11000 字
保留样本占比：98.54%，丢弃 223 条

# -*- coding: utf-8 -*-
"""生产者消费者模式：批量跑数的骨架，可直接在本机跑通。

这里用一个会 sleep、偶尔抛异常的模拟任务代替真实 API 调用，
换成 call_deepseek.call_server 就是生产版本。

四个工程细节都在代码里：
  1. 崩溃可续跑   —— 先读结果文件，把已完成的 id 收进 finishedSet
  2. 队列限长     —— qsize() 超过上限就让生产者歇一秒，内存不是无限的
  3. 大文件逐行读 —— 不用 readlines()，几十 GB 也不会撑爆内存
  4. 写一条 flush 一条 —— 进程被杀也只丢最后一条

用法：python3 producer_consumer.py            # 自带演示数据，跑完自动退出
"""
import json
import os
import queue
import random
import threading
import time

TASK_QUEUE_LIMIT = 20    # 内存里的队列，放多了没意义
CONSUMER_COUNT = 10      # 调 API 是 IO 密集型，用线程；数量看服务端吞吐
RETRY = 3

task_queue = queue.Queue()
result_queue = queue.Queue()
DONE = object()          # 毒丸：生产者放完任务后塞进去，通知消费者收工


def call_server_simulate(prompt):
    """模拟远程调用：随机耗时，遇到以「能」开头的 prompt 触发异常走重试。"""
    time.sleep(random.uniform(0.05, 0.2))
    msg = "OK"
    for attempt in range(1, RETRY + 1):
        try:
            if prompt.startswith("能"):
                1 / 0                       # 故意造一个异常，检验重试分支
            return True, "OK", "reasoning_content of " + prompt, "content of " + prompt
        except Exception as exc:            # noqa: BLE001
            msg = "第 %d 次调用异常: %s" % (attempt, exc)
    return False, msg, "", ""


def produce_task(src_path, dst_path):
    """读源文件，跳过已完成样本，把任务塞进队列。"""
    finished = set()
    if os.path.exists(dst_path):
        with open(dst_path, "r", encoding="utf-8") as done_file:
            for line in done_file:
                line = line.strip()
                if not line:
                    continue
                data = json.loads(line)
                if "id" in data:
                    finished.add(data["id"])
    if finished:
        print("检测到已完成 %d 条，本轮跳过" % len(finished))

    with open(src_path, "r", encoding="utf-8") as src:
        for line in src:                     # 逐行读，内存友好
            line = line.strip()
            if not line:
                continue
            data = json.loads(line)
            if data["id"] in finished:
                continue
            while task_queue.qsize() >= TASK_QUEUE_LIMIT:
                time.sleep(1)                # 队列满了就等，别把内存撑爆
            task_queue.put(data)

    for _ in range(CONSUMER_COUNT):
        task_queue.put(DONE)


def consume_task(index):
    """从任务队列领任务，把结果丢进结果队列。"""
    while True:
        data = task_queue.get(block=True)
        if data is DONE:
            result_queue.put(DONE)
            return
        success, msg, think, result = call_server_simulate(data["instruction"])
        result_queue.put((data, success, msg, think, result))


def collect(dst_path):
    """唯一的写盘口，顺带充当进度条。"""
    ok = fail = 0
    finished_consumers = 0
    with open(dst_path, "a", encoding="utf-8") as target:
        while finished_consumers < CONSUMER_COUNT:
            item = result_queue.get(block=True)
            if item is DONE:
                finished_consumers += 1
                continue
            data, success, msg, think, result = item
            if success:
                data["output"] = "<think>\n" + think + "\n</think>\n\n\n" + result
                target.write(json.dumps(data, ensure_ascii=False) + "\n")
                target.flush()               # 写一条落一条
                ok += 1
            else:
                fail += 1
                print("结果错误 id=%s %s" % (data["id"], msg))
    print("入库 %d 条，丢弃 %d 条" % (ok, fail))


def ask_batch_mt(src_path, dst_path, consumer_count=CONSUMER_COUNT):
    producer = threading.Thread(target=produce_task, args=(src_path, dst_path))
    producer.start()
    for i in range(consumer_count):
        threading.Thread(target=consume_task, args=(i,), daemon=True).start()
    collect(dst_path)
    producer.join()


def _make_demo(src_path):
    """造 30 条演示数据，其中 5 条以「能」开头，用来触发失败分支。"""
    with open(src_path, "w", encoding="utf-8") as f:
        for i in range(30):
            head = "能否" if i % 6 == 0 else "请说明"
            f.write(json.dumps(
                {"id": i, "instruction": "%s第 %d 个问题" % (head, i), "input": ""},
                ensure_ascii=False) + "\n")


if __name__ == "__main__":
    src = "/tmp/cot_demo_query.jsonl"
    dst = "/tmp/cot_demo_answer.jsonl"
    _make_demo(src)
    if os.path.exists(dst):
        os.remove(dst)
    start = time.time()
    ask_batch_mt(src, dst)
    print("耗时 %.2f 秒" % (time.time() - start))
    print("---- 断点续跑：原样再跑一次 ----")
    ask_batch_mt(src, dst)

结果错误 id=6 第 3 次调用异常: division by zero
结果错误 id=0 第 3 次调用异常: division by zero
结果错误 id=12 第 3 次调用异常: division by zero
结果错误 id=18 第 3 次调用异常: division by zero
结果错误 id=24 第 3 次调用异常: division by zero
入库 25 条，丢弃 5 条
耗时 1.20 秒
---- 断点续跑：原样再跑一次 ----
检测到已完成 25 条，本轮跳过
结果错误 id=18 第 3 次调用异常: division by zero
结果错误 id=0 第 3 次调用异常: division by zero
结果错误 id=12 第 3 次调用异常: division by zero
结果错误 id=24 第 3 次调用异常: division by zero
结果错误 id=6 第 3 次调用异常: division by zero
入库 0 条，丢弃 5 条

===============================================================================
问题生产三段式 prompt 模板
把 {{...}} 换成你自己的领域内容即可复用。
三段的顺序不能调换：先划分子领域定配额，再按类型批量出题，最后换模型打分筛选。
===============================================================================


-------------------------------------------------------------------------------
【第一段】子领域划分 —— 先把配额表定下来
产出：一张三列表格（子领域名称 / 语料数量 / 详细说明）
用途：第二段的 prompt 直接吃这张表；同时它就是数据集的结构比例，去重和截断时
      都要照着它保持比例。
-------------------------------------------------------------------------------
你是一位 {领域，例如「中国现行婚姻法」} 方面的专家，我需要整理一份
{领域} 方面的训练语料，共 {总条数，例如 5000} 条，需要考虑
{领域} 的哪几个子领域，各自语料的数量是多少？用一个表格输出结果，
每一行是一个子领域，第一列是子领域的名称，第二列是语料的数量，
第三列是对这个子领域的详细说明。不允许输出其他字符。


-------------------------------------------------------------------------------
【第二段】按类型批量生成问题
产出：每行一道题的纯文本，可直接按行切成 jsonl
关键：把上一段拿到的「类别清单 + 各类别详细说明」整段塞进来，模型才知道
      「夫妻权利义务」到底该出哪些题；不塞说明，出的题会大面积跑到别的类别去。
      循环这一段，每次只换【本次类别】，就能把配额一格一格填满。
-------------------------------------------------------------------------------
你是一位 {领域} 方面的专家，根据【特定类别】的定义，针对其中
{本次类别，例如「夫妻权利义务」} 方面的规定，向学生提供
{每批题量，例如 20} 道练习题，你能提出哪些问题？
每个问题单独一行，只允许输出问题本身，不允许输出任何其他字符。

【特定类别】
{第一段产出的类别名称数组，例如
["结婚登记与条件","夫妻权利义务","离婚程序与条件","财产分割"]}

各类别详细说明如下：
{逐行贴第一段表格的第三列，格式为「类别名称：详细说明」}


-------------------------------------------------------------------------------
【第三段】问题检查 —— 换一个高质量模型打分
产出：每行一个 0~9 的整数，与输入文本逐行对应
关键：换模型。出题的模型来判自己出的题是否切题，等于考生自己批卷。
      输出格式限死成纯数字，把 max_tokens 压到个位数，吞吐立刻上去。
      低分的题直接删掉，删的时候注意各子领域之间的比例不要被删失衡。
-------------------------------------------------------------------------------
你是一位 {领域} 方面的专家，如果要把【输入文本】分类到【特定类别】中，
那么输入文本属于 {本次类别} 是否准确？给出一个准确性评分，
该评分是 0 到 9 之间的正整数，其中 0 表示完全错误，9 表示非常准确。
【输入文本】中的每一行是一条独立数据，需要单独判定。
输出时每条数据的结果单独输出一行，只允许输出准确性评分，不允许输出其他任何字符。

【特定类别】
{同第二段的类别数组}

各类别详细说明如下：
{同第二段的类别说明}

【输入文本】
{第二段生成的问题，每行一条，一次别塞太多，20~50 行为宜}


===============================================================================
接下来的三件事（都不在 prompt 里，靠代码做）
  1. 去重：simhash 算指纹，指纹汉明距离小于阈值判为重复；量大时先分桶再比对。
     删重复样本时同样要保持各子领域的比例。
  2. 取答案：把筛过的问题送教师模型，见 call_deepseek.py。
  3. 打分筛选：把「问答对」整体再评一次，见 score_answers.py。
===============================================================================

data/
└── chatData/                     <- 训练命令里的 --dataset_dir 指向这里
    ├── dataset_info.json         <- 必须有，把数据集名映射到文件名
    ├── train.jsonl               <- 终稿：一行一条 {"instruction","input","output"}
    ├── queryData.jsonl           <- 中间产物：只有题，没有答案
    ├── answerData.jsonl          <- 中间产物：题 + 未过滤的答案
    └── dropped.jsonl             <- 中间产物：被打分淘汰的样本，留档备查

dataset_info.json 的内容：
{
  "chat-train": {                 <- 训练命令里的 --dataset 用这个名字
    "file_name": "train.jsonl"    <- 实际读的文件
  }
}