Prompt-Tuning 进阶：ICL、CoT 与 PEFT 全景

# -*- coding: utf-8 -*-
"""In-Context Learning 输入拼装器：zero-shot / one-shot / few-shot 一次看全。

纯标准库可跑。跑完你会看到一件事：三种做法的差别**只在送进去的那串文本**，
模型权重、优化器、梯度全都不存在。所谓「学习」，学的是当前这一次输入。
"""

TASK_DESC = "判断下面这句话的情感，只回答「正面」或「负面」。"

# 候选示例池：真实项目里这些来自训练集
POOL = [
    ("菜很难吃，等位还等了两小时。", "负面"),
    ("环境安静舒服，适合聊天。", "正面"),
    ("服务员态度差，叫了三次都没人理。", "负面"),
    ("分量足，价格也实在。", "正面"),
]

QUERY = "这家店服务很贴心。"


def render_demo(sent, label):
    """一个示例渲染成两行：输入行 + 答案行。格式必须与提问行完全一致。"""
    return "句子：%s\n情感：%s" % (sent, label)


def build_prompt(query, demos):
    """把任务描述、若干示例、待预测句子拼成最终输入。

    demos 传空列表就是 zero-shot，传 1 个就是 one-shot，传 N 个就是 few-shot。
    """
    blocks = [TASK_DESC]
    for sent, label in demos:
        blocks.append(render_demo(sent, label))
    # 提问行故意留空，让模型顺着这个格式往下补
    blocks.append("句子：%s\n情感：" % query)
    return "\n\n".join(blocks)


def stat(prompt):
    """粗略统计：字符数、示例块数。真实项目里换成 tokenizer 的 token 数。"""
    return len(prompt), prompt.count("句子：") - 1


def main():
    modes = [
        ("zero-shot 零样本", []),
        ("one-shot  单样本", POOL[:1]),
        ("few-shot  少样本", POOL[:4]),
    ]

    lens = {}
    for name, demos in modes:
        p = build_prompt(QUERY, demos)
        n_char, n_demo = stat(p)
        lens[name] = n_char
        print("=" * 60)
        print("【%s】示例 %d 个 · 输入长度 %d 字符" % (name, n_demo, n_char))
        print("-" * 60)
        print(p)
        print()

    print("=" * 60)
    print("【三种做法的唯一差别】")
    print("  模型权重      : 完全相同，一个字节都没改")
    print("  梯度 / 优化器 : 不存在")
    print("  变的是什么    : 送进去的那串文本的长度和内容")
    print()
    for name in lens:
        print("  %s 输入 %d 字符" % (name, lens[name]))

    # 断言一：示例越多，输入越长——这正是 ICL 的全部成本所在
    assert lens["zero-shot 零样本"] < lens["one-shot  单样本"] < lens["few-shot  少样本"], \
        "示例增加必然带来输入变长，这是 ICL 唯一实打实的代价"

    # 断言二：zero-shot 的输入里不能出现任何已知答案，否则就是泄题
    zero = build_prompt(QUERY, [])
    for _, label in POOL:
        assert ("情感：%s" % label) not in zero, "zero-shot 的输入里不该出现任何带答案的示例"

    # 断言三：示例的渲染格式必须和提问行同构，否则模型不知道该补什么
    demo_line = render_demo(*POOL[0]).split("\n")[0]
    assert demo_line.startswith("句子：") and "句子：%s" % QUERY in zero, \
        "示例的输入行与提问行必须用同一个前缀，格式对不齐模型就不会照着填"

    print()
    print("自检通过：三种形态只是同一条输入的不同长度，没有任何参数被更新。")


if __name__ == "__main__":
    main()

# -*- coding: utf-8 -*-
"""示例的顺序和数量到底改变了什么：把「玄学」量化成可核对的数字。

纯标准库可跑。这个脚本不预测情感，它统计的是**模型看到的输入本身**——
标签分布、最后一个示例的标签、输入长度。这三样是 ICL 里最常被忽略的变量。
"""

POOL = [
    ("菜很难吃，等位还等了两小时。", "负面"),
    ("环境安静舒服，适合聊天。", "正面"),
    ("服务员态度差，叫了三次都没人理。", "负面"),
    ("分量足，价格也实在。", "正面"),
    ("上菜快，味道也稳定。", "正面"),
    ("包间隔音差，隔壁说话听得一清二楚。", "负面"),
]


def label_hist(demos):
    """统计示例里各标签出现了几次。"""
    h = {}
    for _, label in demos:
        h[label] = h.get(label, 0) + 1
    return h


def majority_ratio(demos):
    """占比最高的那个标签占了多少。越接近 1，多数标签偏置越强。"""
    h = label_hist(demos)
    return max(h.values()) / float(len(demos))


def describe(name, demos):
    h = label_hist(demos)
    last = demos[-1][1] if demos else "（无示例）"
    hist_s = " / ".join("%s×%d" % (k, v) for k, v in sorted(h.items()))
    print("%-14s 示例%d个  分布[%s]  末位标签=%s  多数占比=%.2f"
          % (name, len(demos), hist_s, last, majority_ratio(demos)))
    return majority_ratio(demos), last


def main():
    balanced = [POOL[0], POOL[1], POOL[2], POOL[3]]          # 负正负正
    skewed = [POOL[1], POOL[3], POOL[4], POOL[0]]            # 正正正负
    reversed_order = list(reversed(balanced))                # 正负正负

    print("【一、同样 4 个示例，排法不同，模型看到的统计特征就不同】")
    r_bal, last_bal = describe("交替排列", balanced)
    r_skew, last_skew = describe("同类扎堆", skewed)
    r_rev, last_rev = describe("逆序排列", reversed_order)
    print()

    print("【二、两个真实存在的偏置】")
    print("  多数标签偏置：示例里哪个标签多，模型就更爱答哪个。")
    print("    交替排列多数占比 %.2f，同类扎堆多数占比 %.2f —— 后者把模型往「正面」推。"
          % (r_bal, r_skew))
    print("  近因偏置：离提问最近的那个示例影响最大。")
    print("    交替排列末位=%s，逆序排列末位=%s —— 同一批示例，最后一眼看到的标签反了。"
          % (last_bal, last_rev))
    print()

    print("【三、示例数量的代价：长度是线性涨的】")
    print("%-8s %-10s %-10s" % ("示例数", "输入字符", "相对 zero-shot"))
    base = None
    for k in [0, 1, 2, 4, 6]:
        demos = POOL[:k]
        text = "\n\n".join(["句子：%s\n情感：%s" % (s, l) for s, l in demos])
        text += "\n\n句子：这家店服务很贴心。\n情感："
        n = len(text)
        if base is None:
            base = n
        print("%-8d %-10d %-10s" % (k, n, "%.1f×" % (n / float(base))))
    print()
    print("  示例不是免费的：每加一个都要占掉上下文窗口，并按 token 计费。")
    print("  一旦发现 4 个和 8 个效果差不多，就该停在 4 个。")

    # 断言一：同类扎堆确实抬高了多数标签占比
    assert r_skew > r_bal, "同类扎堆应当产生更强的多数标签偏置"

    # 断言二：逆序确实换掉了末位标签——这就是「顺序敏感」的可测量来源
    assert last_rev != last_bal, "顺序一反，离提问最近的那个标签就变了"

    # 断言三：交替排列的标签分布是均衡的（各占一半）
    assert abs(r_bal - 0.5) < 1e-9, "交替排列应当做到标签分布完全均衡"

    # 断言四：示例集合完全相同，只是顺序不同
    assert sorted(balanced) == sorted(reversed_order), \
        "逆序只改顺序不改内容，因此差异只可能来自顺序本身"

    print()
    print("自检通过：顺序与数量都是**可测量的超参数**，不是玄学。")


if __name__ == "__main__":
    main()

# -*- coding: utf-8 -*-
"""CoT 提示结构解析器：把 Few-shot CoT 的示例拆成 <输入, 推理, 答案> 三段。

纯标准库可跑。普通 ICL 的示例是二元组 <input, output>；
Few-shot CoT 把它扩成三元组 <input, CoT, output>——多出来的中间那段就是全部秘密。
这个脚本负责检查：三段齐不齐、顺序对不对、答案是不是真从推理里长出来的。
"""

import re

# 一个写好的 Few-shot CoT 示例块。三行标记必须严格对齐，缺一行就不成立
DEMO = """问：食堂原有 23 个苹果，用掉 20 个，又买了 6 个，现在有几个？
推理：原有 23 个，用掉 20 个，还剩 23 - 20 = 3 个；又买了 6 个，3 + 6 = 9 个。
答：9"""

# 一个只有问答、没有推理的普通 ICL 示例块，用来做对照
DEMO_PLAIN = """问：食堂原有 23 个苹果，用掉 20 个，又买了 6 个，现在有几个？
答：9"""


def parse_demo(block):
    """把一个示例块解析成 (输入, 推理, 答案)。没有推理段就返回 None。"""
    q = re.search(r"^问：(.*)$", block, re.M)
    r = re.search(r"^推理：(.*)$", block, re.M)
    a = re.search(r"^答：(.*)$", block, re.M)
    if not q or not a:
        raise ValueError("示例块必须同时有「问：」和「答：」两行")
    return q.group(1).strip(), (r.group(1).strip() if r else None), a.group(1).strip()


def numbers_in(text):
    """抽出文本里所有整数，用来检查答案有没有凭空冒出来。"""
    return [int(x) for x in re.findall(r"\d+", text)]


def audit(block, name):
    """审一个示例块：三段齐不齐、答案是否出现在推理里。"""
    q, cot, a = parse_demo(block)
    print("【%s】" % name)
    print("  输入 : %s" % q)
    print("  推理 : %s" % (cot if cot else "（没有推理段）"))
    print("  答案 : %s" % a)
    has_cot = cot is not None
    grounded = has_cot and int(a) in numbers_in(cot)
    print("  三段齐全 : %s" % ("是" if has_cot else "否，这是普通 ICL 示例，不是 CoT 示例"))
    print("  答案有据 : %s" % ("是，%s 在推理里算出来过" % a if grounded
                              else "否，答案没有在推理过程中出现"))
    print()
    return has_cot, grounded


def build_fewshot_cot(demos, query):
    """把若干 CoT 示例和待解题拼成最终输入。提问块只到「推理：」为止。"""
    return "\n\n".join(demos + ["问：%s\n推理：" % query])


def main():
    print("【一、解析两种示例，看差在哪一段】")
    has_cot, grounded = audit(DEMO, "Few-shot CoT 示例")
    plain_cot, _ = audit(DEMO_PLAIN, "普通 ICL 示例")

    print("【二、拼出最终输入】")
    query = "书架上有 15 本书，搬走 7 本，又放上 4 本，现在有几本？"
    prompt = build_fewshot_cot([DEMO], query)
    print(prompt)
    print()

    print("【三、这个结构为什么重要】")
    print("  提问块停在「推理：」而不是「答：」——这是在逼模型先写过程再给结论。")
    print("  如果提问块直接给到「答：」，模型会跳过推理直接猜一个数，CoT 就白写了。")
    print()

    # 断言一：CoT 示例必须有推理段，普通示例必须没有
    assert has_cot is True, "CoT 示例必须含推理段"
    assert plain_cot is False, "普通 ICL 示例不应含推理段"

    # 断言二：答案必须能在推理里找到，不能凭空出现
    assert grounded is True, "CoT 示例的答案必须是推理过程算出来的数"

    # 断言三：最终输入必须以「推理：」结尾，把模型逼进先推理的轨道
    assert prompt.rstrip().endswith("推理："), \
        "提问块要停在推理段，直接给到答案段会让模型跳过推理"

    # 断言四：示例里的推理链条本身要算得对
    q, cot, a = parse_demo(DEMO)
    assert 23 - 20 + 6 == int(a), "示例里的算式必须和答案自洽，否则教坏模型"

    print("自检通过：Few-shot CoT = 普通示例 + 中间那段推理，且答案必须由推理产生。")


if __name__ == "__main__":
    main()

# -*- coding: utf-8 -*-
"""Zero-shot CoT 的两阶段提示：一个示例都不给，靠两句固定咒语把推理逼出来。

纯标准库可跑。Zero-shot CoT 不是「加一句话」，是**两次调用**：
第一次让模型写推理，第二次让它从推理里把答案抽出来。
少了第二次，你拿到的是一大段话，不是一个能入库的答案。
"""

TRIGGER = "Let's think step by step"          # 第一阶段：触发推理
EXTRACT = "Therefore, the answer is"          # 第二阶段：抽取答案

QUESTION = "食堂原有 23 个苹果，用掉 20 个，又买了 6 个，现在有几个？"


def stage1(question):
    """第一阶段输入：问题 + 触发语。此时不要求模型给答案。"""
    return "问：%s\n答：%s" % (question, TRIGGER)


def stage2(question, reasoning):
    """第二阶段输入：把第一阶段的原文接回去，再追加抽取语。"""
    return "问：%s\n答：%s%s\n%s" % (question, TRIGGER, reasoning, EXTRACT)


def main():
    print("【第一阶段：触发推理】")
    p1 = stage1(QUESTION)
    print(p1)
    print()
    print("  这一步模型会顺着 %r 往下写出推导过程。" % TRIGGER)
    print("  注意：此时**不要**追问答案，一追问它就会跳过推理。")
    print()

    # 这里用一段占位推理来演示第二阶段怎么拼，真实使用时换成模型第一阶段的输出
    reasoning_placeholder = "（模型在这里写出的推导过程）"

    print("【第二阶段：抽取答案】")
    p2 = stage2(QUESTION, reasoning_placeholder)
    print(p2)
    print()
    print("  第二阶段的输入 = 问题 + 触发语 + 第一阶段输出 + 抽取语。")
    print("  少拼任何一段，模型就失去了上下文，只能重新猜。")
    print()

    print("【和 Few-shot CoT 的分工】")
    rows = [
        ("Few-shot CoT", "要手写带推理的示例", "1 次调用", "示例写得好就稳，但每个任务都要重写示例"),
        ("Zero-shot CoT", "一个示例都不用写", "2 次调用", "省事，但精度通常不如写好的 Few-shot CoT"),
    ]
    print("%-16s %-20s %-10s %s" % ("方法", "要准备什么", "调用次数", "取舍"))
    for a, b, c, d in rows:
        print("%-16s %-20s %-10s %s" % (a, b, c, d))
    print()

    print("【什么时候它会不灵】")
    for line in [
        "模型太小：推理能力本身不存在，加咒语只会多产出一段像模像样的错话。",
        "题目不需要多步推导：单跳的事实问答加了 CoT 反而绕远、更容易跑偏。",
        "只做了第一阶段：拿到一大段推理却没有结构化答案，没法入库也没法评测。",
        "触发语被翻译或改写：换成别的措辞就不再是论文验证过的那句话，效果要重新实测。",
    ]:
        print("  · %s" % line)

    # 断言一：第一阶段必须以触发语收尾，不能带上抽取语
    assert p1.rstrip().endswith(TRIGGER), "第一阶段要停在触发语，别提前索要答案"
    assert EXTRACT not in p1, "抽取语不能出现在第一阶段"

    # 断言二：第二阶段必须完整包含第一阶段的内容
    assert p1 in p2.replace(reasoning_placeholder, ""), \
        "第二阶段必须把第一阶段原样接回去，否则模型看不到自己刚写的推理"

    # 断言三：第二阶段以抽取语收尾
    assert p2.rstrip().endswith(EXTRACT), "第二阶段要停在抽取语上，逼出一个短答案"

    # 断言四：两句咒语一字不改
    assert TRIGGER == "Let's think step by step"
    assert EXTRACT == "Therefore, the answer is"

    print()
    print("自检通过：Zero-shot CoT 是两阶段流程，不是一句咒语。")


if __name__ == "__main__":
    main()

# -*- coding: utf-8 -*-
"""Prompt 与 Instruction：同一个任务的两种问法，差别在「续写」还是「听指令」。

纯标准库可跑。这个脚本把两种问法渲染出来并逐项对比结构特征：
有没有任务说明、有没有给定选项、模型要做的是续写还是选择。
"""

SENT = "带女朋友去了一家餐厅，她吃的很开心。"

# 第一种问法：靠模型的续写能力，把句子往下接完
PROMPT_STYLE = "{x}这家餐厅太____了！"

# 第二种问法：明确写出要做什么、有哪些选项
INSTRUCTION_STYLE = "判断这句话的情感：{x}选项：A=好，B=一般，C=差"

LABEL_WORDS = ["好", "一般", "差"]
OPTIONS = ["A", "B", "C"]


def render(tpl, sent):
    return tpl.format(x=sent)


def features(text):
    """抽出三个可判定的结构特征。"""
    return {
        "含任务说明": any(k in text for k in ["判断", "请", "任务"]),
        "含候选选项": any(("%s=" % o) in text for o in OPTIONS),
        "含待补空位": "____" in text,
    }


def main():
    p = render(PROMPT_STYLE, SENT)
    ins = render(INSTRUCTION_STYLE, SENT)

    print("【一、同一句话，两种问法】")
    print("  Prompt 式      : %s" % p)
    print("  Instruction 式 : %s" % ins)
    print()

    print("【二、结构特征逐项对比】")
    fp, fi = features(p), features(ins)
    keys = ["含任务说明", "含候选选项", "含待补空位"]
    print("  %-12s %-12s %s" % ("特征", "Prompt 式", "Instruction 式"))
    for k in keys:
        print("  %-12s %-14s %s" % (k, "是" if fp[k] else "否", "是" if fi[k] else "否"))
    print()

    print("【三、模型被要求做的事不一样】")
    rows = [
        ("Prompt-Tuning", "激发续写能力", "把空位填上一个词", "填出的词还要过 Verbalizer 才变成类别"),
        ("Instruction-Tuning", "激发理解能力", "读懂指令并选一个选项", "直接吐选项，判别比生成容易"),
    ]
    print("  %-20s %-14s %-18s %s" % ("方法", "激发什么", "模型要做什么", "备注"))
    for a, b, c, d in rows:
        print("  %-20s %-14s %-18s %s" % (a, b, c, d))
    print()

    print("【四、最要紧的一条区别】")
    print("  Prompt-Tuning 在没有额外训练的模型上也能有一定效果。")
    print("  Instruction-Tuning 必须先用大量指令数据训练过，模型才认识这种问法。")
    print("  换句话说：前者是会不会问，后者是模型有没有被教过「听指令」这件事。")
    print()

    print("【五、为什么要为一个任务准备多个指令模板】")
    variants = [
        "判断这句话的情感：{x}选项：A=好，B=一般，C=差",
        "请阅读下面这句话，并从 A/B/C 中选出它的情感：{x}",
        "下面这句话表达的情感是？{x}A=好 B=一般 C=差",
    ]
    for i, v in enumerate(variants, 1):
        print("  指令%d: %s" % (i, render(v, SENT)))
    print("  单个模板的成绩会被措辞运气带偏，所以评测时看的是**一组模板的平均表现**。")

    # 断言一：Instruction 式必须同时含任务说明与候选选项
    assert fi["含任务说明"] and fi["含候选选项"], "Instruction 必须写明要做什么、可选哪些"
    # 断言二：Prompt 式靠的是空位续写，不给选项
    assert fp["含待补空位"] and not fp["含候选选项"], "Prompt 式靠续写，不提供候选选项"
    # 断言三：两种问法的原句必须一字不差，差异只能来自问法本身
    assert SENT in p and SENT in ins, "对比的前提是原句完全相同"
    # 断言四：多模板必须真的互不相同
    assert len(set(variants)) == len(variants), "指令模板不能重复，否则平均表现没有意义"

    print()
    print("自检通过：Prompt 教模型接话，Instruction 教模型听话。")


if __name__ == "__main__":
    main()

# -*- coding: utf-8 -*-
"""PEFT 参数量计算器：同一个 base model，四种做法各要训练多少参数、存盘多大。

纯标准库可跑。数字全部由公式算出，不是抄来的经验值——
把 CFG 换成自己的模型再跑一遍，就是你自己的选型依据。
"""

# 一个 7B 量级 decoder-only 模型的常见配置
CFG = {
    "name": "7B decoder-only",
    "hidden": 4096,       # d_model
    "layers": 32,
    "ff": 11008,          # FFN 中间层
    "vocab": 32000,
    "heads": 32,
}

# 四种 PEFT 的超参数
HP = {
    "soft_prompt_len": 20,     # Prompt Tuning：只在输入层拼的伪标记个数
    "prefix_len": 10,          # Prefix-Tuning：每层前缀长度
    "adapter_bottleneck": 64,  # Adapter：降维后的维度
    "lora_rank": 8,            # LoRA：低秩支路的秩 r
    "lora_targets": 2,         # LoRA 挂在每层的几个投影上（常见做法：q_proj 与 v_proj）
}


def base_params(cfg):
    """base model 主体参数量（embedding + N 层 transformer），用于算占比。"""
    h, ff, L = cfg["hidden"], cfg["ff"], cfg["layers"]
    emb = cfg["vocab"] * h
    attn = 4 * h * h                       # Q K V O 四个投影
    mlp = 3 * h * ff                       # 门控 MLP 的三个矩阵
    per_layer = attn + mlp
    return emb + L * per_layer


def p_prompt_tuning(cfg, hp):
    """Prompt Tuning：只在输入 embedding 前拼一段向量，与层数无关。"""
    return hp["soft_prompt_len"] * cfg["hidden"]


def p_prefix_tuning(cfg, hp):
    """Prefix-Tuning：每一层的 key 和 value 各挂一段前缀，所以要乘 2 再乘层数。"""
    return 2 * hp["prefix_len"] * cfg["hidden"] * cfg["layers"]


def p_adapter(cfg, hp):
    """Adapter：降维 + 升维两个线性层（含偏置），每层插 2 个（attn 后与 FFN 后）。"""
    h, b = cfg["hidden"], hp["adapter_bottleneck"]
    one = (h * b + b) + (b * h + h)
    return one * 2 * cfg["layers"]


def p_lora(cfg, hp):
    """LoRA：每个目标矩阵旁边并一条 A(h×r) + B(r×h) 的支路，无偏置。"""
    h, r = cfg["hidden"], hp["lora_rank"]
    one = 2 * h * r
    return one * hp["lora_targets"] * cfg["layers"]


def mb(n_params, bytes_per=2):
    """按 fp16（每参数 2 字节）估存盘体积，单位 MB。"""
    return n_params * bytes_per / 1024.0 / 1024.0


def main():
    total = base_params(CFG)
    print("base model: %s  主体参数量 %s" % (CFG["name"], "{:,}".format(total)))
    print("（fp16 全量存盘约 %.0f MB）" % mb(total))
    print()

    rows = [
        ("Prompt Tuning", p_prompt_tuning(CFG, HP), "只改输入层", "soft_prompt_len=%d" % HP["soft_prompt_len"]),
        ("Prefix-Tuning", p_prefix_tuning(CFG, HP), "每层前缀", "prefix_len=%d" % HP["prefix_len"]),
        ("LoRA", p_lora(CFG, HP), "权重旁支", "r=%d, 目标%d个/层" % (HP["lora_rank"], HP["lora_targets"])),
        ("Adapter", p_adapter(CFG, HP), "层内插模块", "bottleneck=%d" % HP["adapter_bottleneck"]),
    ]
    rows.sort(key=lambda x: x[1])

    print("%-16s %14s %9s %11s  %-12s %s" % ("方法", "可训参数", "占比", "存盘(fp16)", "改哪一层", "超参"))
    for name, n, where, hp_s in rows:
        print("%-16s %14s %8.4f%% %10.2f MB  %-12s %s"
              % (name, "{:,}".format(n), 100.0 * n / total, mb(n), where, hp_s))
    print()

    print("【同一方法内部，超参怎么影响参数量】")
    print("  LoRA 的 r 翻倍，可训参数就翻倍（线性）：")
    for r in [4, 8, 16, 32, 64]:
        hp = dict(HP); hp["lora_rank"] = r
        n = p_lora(CFG, hp)
        print("    r=%-3d -> %12s 个 (%.4f%%, %.2f MB)" % (r, "{:,}".format(n), 100.0 * n / total, mb(n)))
    print()
    print("  Adapter 的 bottleneck 翻倍，可训参数也基本翻倍：")
    for b in [8, 16, 32, 64, 128]:
        hp = dict(HP); hp["adapter_bottleneck"] = b
        n = p_adapter(CFG, hp)
        print("    b=%-3d -> %12s 个 (%.4f%%, %.2f MB)" % (b, "{:,}".format(n), 100.0 * n / total, mb(n)))
    print()
    print("  Prompt Tuning 的长度翻倍也只是线性涨，但基数极小：")
    for k in [10, 20, 50, 100]:
        hp = dict(HP); hp["soft_prompt_len"] = k
        n = p_prompt_tuning(CFG, hp)
        print("    len=%-4d -> %12s 个 (%.6f%%, %.4f MB)" % (k, "{:,}".format(n), 100.0 * n / total, mb(n)))
    print()

    print("【一句话读懂这张表】")
    print("  四种做法的可训参数都落在主体的 1% 以内，但彼此差着两个数量级：")
    print("  Prompt Tuning 最省（只碰输入层），Adapter 最贵（每层插两个模块）。")
    print("  省不等于好——参数越少，能表达的下游变化越有限，小模型上尤其明显。")

    # 断言一：四种方法都必须远低于全参微调
    for name, n, _, _ in rows:
        assert n < total * 0.01, "%s 的可训参数应当低于主体的 1%%，否则不配叫参数高效" % name

    # 断言二：与层数的关系——Prompt Tuning 是唯一不随层数增长的
    cfg2 = dict(CFG); cfg2["layers"] = CFG["layers"] * 2
    assert p_prompt_tuning(cfg2, HP) == p_prompt_tuning(CFG, HP), \
        "Prompt Tuning 只改输入层，层数翻倍不应改变它的参数量"
    for fn in (p_prefix_tuning, p_adapter, p_lora):
        assert fn(cfg2, HP) == 2 * fn(CFG, HP), \
            "%s 是逐层挂的，层数翻倍参数量必须翻倍" % fn.__name__

    # 断言三：量级排序（本组超参下）
    assert p_prompt_tuning(CFG, HP) < p_prefix_tuning(CFG, HP) < p_lora(CFG, HP) < p_adapter(CFG, HP), \
        "本组超参下的量级排序：Prompt Tuning < Prefix < LoRA < Adapter"

    # 断言四：LoRA 对 r 是严格线性的
    hp_a = dict(HP); hp_a["lora_rank"] = 8
    hp_b = dict(HP); hp_b["lora_rank"] = 16
    assert p_lora(CFG, hp_b) == 2 * p_lora(CFG, hp_a), "LoRA 参数量对 r 线性"

    print()
    print("自检通过：四种 PEFT 的参数量全部由公式确定，量级差异可复算。")


if __name__ == "__main__":
    main()

# -*- coding: utf-8 -*-
"""LoRA 的旁支是怎么算的：手算一遍 W + A@B，看清「为什么训练开始时输出不变」。

纯标准库可跑，矩阵用嵌套 list 手写，维度缩到 6×6 方便肉眼核对。
三件事跑成断言：① B 初始化为 0 时旁支输出恒为 0；
② 旁支的参数量是 2·d·r，远小于 d·d；③ 训练完可以合并回 W，推理不多一次矩阵乘。
"""

import random

D = 6      # 演示用的隐层维度（真实模型是 4096 这种量级）
R = 2      # LoRA 的秩 r


def zeros(rows, cols):
    return [[0.0] * cols for _ in range(rows)]


def randmat(rows, cols, seed):
    rnd = random.Random(seed)
    return [[round(rnd.uniform(-1, 1), 3) for _ in range(cols)] for _ in range(rows)]


def matmul(X, Y):
    n, k, m = len(X), len(Y), len(Y[0])
    out = zeros(n, m)
    for i in range(n):
        for j in range(m):
            s = 0.0
            for t in range(k):
                s += X[i][t] * Y[t][j]
            out[i][j] = s
    return out


def matadd(X, Y):
    return [[X[i][j] + Y[i][j] for j in range(len(X[0]))] for i in range(len(X))]


def allclose(X, Y, tol=1e-9):
    for i in range(len(X)):
        for j in range(len(X[0])):
            if abs(X[i][j] - Y[i][j]) > tol:
                return False
    return True


def show(name, M):
    print("  %s  (%d×%d)" % (name, len(M), len(M[0])))
    for row in M[:3]:
        print("    [" + ", ".join("%7.3f" % v for v in row) + (" ...]" if len(row) > 3 else "]"))
    if len(M) > 3:
        print("    ...")


def main():
    # 预训练权重，冻结不动
    W = randmat(D, D, seed=7)
    # LoRA 旁支：A 随机初始化，B 全零初始化（论文里的做法）
    A = randmat(D, R, seed=11)
    B = zeros(R, D)
    x = [[round(random.Random(3).uniform(-1, 1), 3) for _ in range(D)]]   # 一条输入，1×D

    print("【一、三个矩阵的形状】")
    show("W（冻结）", W)
    show("A（可训练，d×r）", A)
    show("B（可训练，r×d，初始化为 0）", B)
    print()

    print("【二、训练刚开始时：旁支输出恒为 0】")
    base = matmul(x, W)                 # 原始通路
    delta = matmul(matmul(x, A), B)     # 旁支通路 x@A@B
    merged = matadd(base, delta)
    show("x @ W", base)
    show("x @ A @ B", delta)
    show("两者相加", merged)
    print("  B 是零矩阵，所以 x@A@B 必然是零向量 —— 加上它等于什么都没加。")
    print("  这就是 LoRA 能安全接到任何预训练模型上的原因：**接上去的那一刻，模型行为完全不变**。")
    print()

    print("【三、训练之后：B 不再是 0，旁支开始起作用】")
    B_trained = randmat(R, D, seed=13)
    delta2 = matmul(matmul(x, A), B_trained)
    out2 = matadd(base, delta2)
    show("训练后的 x @ A @ B", delta2)
    show("最终输出", out2)
    print()

    print("【四、合并：把旁支折进权重，推理时不多算一次】")
    W_merged = matadd(W, matmul(A, B_trained))      # W' = W + A@B
    out_merged = matmul(x, W_merged)
    show("W' = W + A@B", W_merged)
    print("  用 W' 直接算 x@W'，结果应当与「原通路 + 旁支」逐元素相同。")
    print()

    print("【五、参数账】")
    full = D * D
    lora = D * R + R * D
    print("  原权重 W      : %d × %d = %d 个参数（冻结，不训练）" % (D, D, full))
    print("  LoRA 支路 A+B : %d×%d + %d×%d = %d 个参数（训练这些）" % (D, R, R, D, lora))
    print("  比例          : %.1f%%" % (100.0 * lora / full))
    print("  维度放大到 d=4096、r=8：%d 个 vs %d 个，比例 %.3f%%"
          % (2 * 4096 * 8, 4096 * 4096, 100.0 * (2 * 4096 * 8) / (4096 * 4096)))
    print("  秩 r 越小越省，但能表达的更新方向也越少 —— r 是效果与成本的旋钮。")

    # 断言一：B 为零时旁支输出必须是零向量
    assert allclose(delta, zeros(1, D)), "B 初始化为 0 时，旁支输出必须恒为 0"
    # 断言二：因此接上 LoRA 的初始输出与原模型完全一致
    assert allclose(merged, base), "刚接上 LoRA 时，模型输出不允许发生任何变化"
    # 断言三：合并回权重后结果等价，推理不需要额外的矩阵乘
    assert allclose(out_merged, out2), "W+A@B 合并后必须与分开算的结果一致"
    # 断言四：旁支参数量是 2·d·r，且在 r < d/2 时严格小于原权重
    assert lora == 2 * D * R and lora < full, "LoRA 参数量应为 2·d·r 且小于 d·d"

    print()
    print("自检通过：零初始化保证无损接入，低秩保证省，合并保证推理不变慢。")


if __name__ == "__main__":
    main()

# -*- coding: utf-8 -*-
"""选型判据：手头这个任务，到底该用 Prompt Engineering、ICL、CoT 还是 PEFT。

纯标准库可跑。把四个问题答成 True / False，脚本给出建议路线并说明理由。
判据不是拍脑袋来的：每一条都对应「要不要梯度、要不要标注数据、要不要多步推理」。
"""

# 四个决定性问题
QUESTIONS = [
    ("has_labeled_data", "手上有没有成百上千条标注数据？"),
    ("can_train", "有没有 GPU 和训练环境，能不能跑梯度回传？"),
    ("needs_reasoning", "任务要不要多步推理（算术、逻辑、多跳问答）？"),
    ("many_tasks", "同一个 base model 要不要同时服务很多个任务？"),
]


def decide(has_labeled_data, can_train, needs_reasoning, many_tasks):
    """返回 (推荐方法, 理由)。顺序就是判断优先级。"""
    if not can_train:
        # 没法训练，一切落在「改便条」这一侧
        if needs_reasoning:
            return ("CoT（Few-shot CoT 优先）",
                    "不能训练就只能改输入；任务要多步推理，普通示例不够，示例里必须带推理过程")
        if has_labeled_data:
            return ("In-Context Learning（few-shot）",
                    "不能训练但有现成的标注，挑几条当示例塞进输入，零成本拿到任务格式")
        return ("Prompt Engineering",
                "不能训练、也没有可用示例，只剩把任务描述写清楚这一条路")

    # 能训练，进入 PEFT 侧
    if not has_labeled_data:
        return ("先用 ICL / CoT 攒数据，再谈训练",
                "能训练但没有标注数据，梯度无从算起；先用改便条的办法跑起来并积累样本")
    if many_tasks:
        return ("PEFT（LoRA 优先，任务极多时看 Prompt Tuning）",
                "多任务共用一个冻结底座，每个任务只存一小份增量；LoRA 还能合并回权重不增推理延迟")
    return ("PEFT（LoRA）",
            "有数据也能训练，单任务下 LoRA 是效果与成本最平衡的默认选项")


CASES = [
    ("线上客服意图分类，标注 5000 条，有 GPU，只此一个任务", True, True, False, False),
    ("同一底座要服务 20 个部门的不同任务", True, True, False, True),
    ("只有 API 调用权限，拿不到权重，要做财报数字核算", False, False, True, False),
    ("只有 API 权限，有 200 条历史工单可当示例", True, False, False, False),
    ("只有 API 权限，任务是把一段话改写得更礼貌", False, False, False, False),
    ("有 GPU，但一条标注都没有", False, True, False, False),
]


def main():
    print("【四个决定性问题】")
    for key, q in QUESTIONS:
        print("  · %s" % q)
    print()

    print("【六个场景走一遍】")
    for name, a, b, c, d in CASES:
        method, why = decide(a, b, c, d)
        print("-" * 68)
        print("场景：%s" % name)
        print("  有标注=%s 能训练=%s 需推理=%s 多任务=%s" % (a, b, c, d))
        print("  建议 → %s" % method)
        print("  理由   %s" % why)
    print("-" * 68)
    print()

    print("【这棵树的骨架其实只有一句话】")
    print("  能不能跑梯度，决定你在「改便条」还是「改耳机」这一侧；")
    print("  跨过这条线之后，才轮到比较 LoRA、Prefix、Adapter 谁更合适。")

    # 断言一：不能训练时，绝不会推荐任何需要梯度的方案
    for a in (True, False):
        for c in (True, False):
            for d in (True, False):
                m, _ = decide(a, False, c, d)
                assert "PEFT" not in m, "不能训练时不允许推荐 PEFT"
    # 断言二：需要多步推理且不能训练时，必须走 CoT
    m, _ = decide(False, False, True, False)
    assert m.startswith("CoT"), "不能训练 + 要推理，答案只能是 CoT"
    # 断言三：能训练但没数据时，不能直接上 PEFT
    m, _ = decide(False, True, False, False)
    assert "PEFT" not in m, "没有标注数据就没有梯度可算，不该推荐 PEFT"
    # 断言四：有数据 + 能训练，一定落到 PEFT 侧
    m, _ = decide(True, True, False, False)
    assert "PEFT" in m, "有数据且能训练时，PEFT 是默认选项"

    print()
    print("自检通过：四个问题把方法空间切干净了，没有互相重叠的分支。")


if __name__ == "__main__":
    main()

# -*- coding: utf-8 -*-
"""ICL + CoT 提示骨架：改五处 TODO 就能套到自己的任务上。

纯标准库，先在本地把「示例选得对不对、格式对不对齐、推理段在不在」验证通过，
再接真实模型。骨架自带的检查会在启动时就把常见错误炸出来。
"""


class PromptBuilder:
    # TODO 1：换成你的任务说明。写清楚做什么、输出什么格式，不要写成寒暄
    TASK_DESC = "判断下面这句话的情感，只回答「正面」或「负面」。"

    # TODO 2：换成你的示例池。注意标签要均衡，别一边倒
    POOL = [
        ("菜很难吃，等位还等了两小时。", "负面"),
        ("环境安静舒服，适合聊天。", "正面"),
        ("服务员态度差，叫了三次都没人理。", "负面"),
        ("分量足，价格也实在。", "正面"),
    ]

    # TODO 3：字段名要和示例、提问完全一致，改一个字就得全改
    IN_FIELD = "句子"
    OUT_FIELD = "情感"
    COT_FIELD = "推理"

    # TODO 4：需要多步推理时设为 True，示例里必须带推理过程
    USE_COT = False

    # Zero-shot CoT 的两句咒语，原样使用，改写就不再是论文验证过的那句
    TRIGGER = "Let's think step by step"
    EXTRACT = "Therefore, the answer is"

    def __init__(self, n_shot=2):
        self.n_shot = n_shot
        self._check_pool()

    def _check_pool(self):
        """启动即检查，别把错误留到指标里。"""
        assert self.POOL, "示例池不能为空"
        hist = {}
        for _, label in self.POOL:
            hist[label] = hist.get(label, 0) + 1
        assert len(hist) >= 2, "示例池里至少要有两个不同的标签，否则模型只会复读多数类"
        most, least = max(hist.values()), min(hist.values())
        assert most <= 2 * least, "标签分布过于失衡（%s），模型会被多数标签偏置带走" % hist
        assert self.n_shot <= len(self.POOL), "要的示例数超过了示例池的大小"

    def pick_demos(self):
        """TODO 5：换成你的选例策略。

        这里用的是最朴素的「按标签轮流取」，保证送进去的示例标签均衡。
        进阶做法是按与待预测句子的相似度检索，但先把均衡这条守住。
        """
        by_label = {}
        for item in self.POOL:
            by_label.setdefault(item[1], []).append(item)
        picked, labels = [], sorted(by_label)
        i = 0
        while len(picked) < self.n_shot:
            bucket = by_label[labels[i % len(labels)]]
            idx = i // len(labels)
            if idx < len(bucket):
                picked.append(bucket[idx])
            i += 1
            if i > 1000:
                break
        return picked[:self.n_shot]

    def render_demo(self, sent, label, cot=None):
        lines = ["%s：%s" % (self.IN_FIELD, sent)]
        if self.USE_COT:
            assert cot, "开了 USE_COT 就必须给每个示例写出推理过程，否则退化成普通 ICL"
            lines.append("%s：%s" % (self.COT_FIELD, cot))
        lines.append("%s：%s" % (self.OUT_FIELD, label))
        return "\n".join(lines)

    def build(self, query, cots=None):
        """拼出最终输入。提问块故意留空，让模型顺着格式往下写。"""
        blocks = [self.TASK_DESC]
        demos = self.pick_demos()
        for i, (sent, label) in enumerate(demos):
            cot = cots[i] if (self.USE_COT and cots) else None
            blocks.append(self.render_demo(sent, label, cot))
        tail_field = self.COT_FIELD if self.USE_COT else self.OUT_FIELD
        blocks.append("%s：%s\n%s：" % (self.IN_FIELD, query, tail_field))
        text = "\n\n".join(blocks)
        self._check_prompt(text, query)
        return text

    def _check_prompt(self, text, query):
        """三条只要踩中就一定出问题的检查。"""
        assert text.rstrip().endswith("："), "提问块要停在字段名后面，别把答案位也填上"
        assert query in text, "待预测句子必须出现在最终输入里"
        assert text.count("%s：" % self.IN_FIELD) == self.n_shot + 1, \
            "输入行数对不上示例数，多半是字段名写岔了"

    def build_zero_shot_cot(self, query):
        """不给任何示例，靠两阶段咒语逼出推理。返回第一阶段输入。"""
        return "%s\n%s：%s\n%s：%s" % (
            self.TASK_DESC, self.IN_FIELD, query, self.OUT_FIELD, self.TRIGGER)

    def build_extract(self, stage1_text, reasoning):
        """第二阶段：把第一阶段原文接回去再要答案。少拼一段模型就失去上下文。"""
        return "%s%s\n%s" % (stage1_text, reasoning, self.EXTRACT)


def _demo():
    q = "这家店服务很贴心。"

    b = PromptBuilder(n_shot=2)
    print("【Few-shot（无推理段）】")
    print(b.build(q))
    print()

    print("【Zero-shot CoT 第一阶段】")
    s1 = b.build_zero_shot_cot(q)
    print(s1)
    print()
    print("【Zero-shot CoT 第二阶段】")
    print(b.build_extract(s1, "（这里放模型第一阶段写出的推理）"))
    print()

    demos = b.pick_demos()
    labels = [l for _, l in demos]
    assert len(set(labels)) == len(set(l for _, l in PromptBuilder.POOL)), \
        "选例策略要保证每个标签都被覆盖到"
    print("骨架自检通过：选出的 %d 个示例标签为 %s，覆盖均衡。" % (len(demos), labels))


if __name__ == "__main__":
    _demo()