手把手把 5 节点流程编译进 0.5B 权重——Subterranean Agent 实操陪读篇

主篇里我在末段做了一个诚实的提醒：论文的 462× 是纸面数字，“下一步打算挑一个 5–10 个节点的小流程，跑 200 次对比 in-context Claude Sonnet 4.5”。这篇陪读篇就是把那个下一步铺开——但把规模缩到普通读者能跟着做的版本：5 节点日报归档流程、200 条合成对话、Qwen2.5-0.5B、一张消费级显卡（或一台 M-series Mac）、约 $2 的 Claude API 花费。目标不是复现论文的 8B 效果，是让你亲手把「过程编译进权重」这个抽象概念摸出手感。

先摆一个诚实的边界

主篇论文用的是 Qwen3-8B + 8×A100，普通读者复现不了。 这篇把每一处都往下压：

项目	论文 (Dennis 2026)	本篇	压缩理由
基座模型	Qwen3-8B	Qwen2.5-0.5B-Instruct	8GB VRAM 或 Mac 统一内存能 fit
硬件	8×A100 (80GB)	1×RTX 3090/4090 或 M-series 16GB+	消费级门槛
微调方式	全量参数（FP + DeepSpeed ZeRO-3）	全量参数（bf16, single GPU）	论文明确说 LoRA 失败，坚持全量
训练数据	2,700–6,264 条	200 条	5 节点比 55 节点简单 12 倍，样本可以按比例缩
流程规模	14–55 节点	5 节点，1 决策枢纽	一屏能看完
评测	n=200, LLM-as-judge	n=20, LLM-as-judge	手动能过一遍的规模
总花费	~$50–80 一次编译	~$2 (API 合成) + 免费（本地训练）	单杯咖啡钱

能跨过的与不能跨过的： 0.5B 模型能学会「按顺序问问题、在决策枢纽走对边」这类结构性行为，但学不会广度知识（比如”上周三我写过什么”）——那需要更大的基座。5 节点的流程结构简单到 0.5B 完全够，这是本篇挑这个规模的核心理由。

另一个诚实提醒：我没有实机跑完全流程。 下面所有代码是根据 transformers/trl/anthropic 官方文档写的（版本 transformers 4.44+, trl 0.11+, anthropic 0.34+），数字是硬件规格+手算估计。跑通请回来提 issue 纠正——这是博客证据分级里的 B 档证据（有依据但未实测），不是 A 档（实测复现）。

第 0 步：环境与硬件

Python 环境（用 uv 或 conda 隔离）：

# uv 版本（推荐，快）
uv venv --python 3.11
source .venv/bin/activate
uv pip install "torch>=2.4" "transformers>=4.44" "trl>=0.11" "datasets>=2.20" \
               "accelerate>=0.34" "anthropic>=0.34" "peft>=0.12"

# 或 conda
conda create -n subterranean python=3.11
conda activate subterranean
pip install torch transformers trl datasets accelerate anthropic peft

硬件要求（手算过一遍）：

Qwen2.5-0.5B 全量微调 bf16 的显存账本：

• 模型参数 494M × 2 bytes = 1.0 GB
• 梯度 494M × 2 bytes = 1.0 GB
• Adam 优化器状态 (m + v, fp32) = 494M × 8 bytes = 4.0 GB
• 激活值（batch=4, seq=1024, hidden=896）约 1.5 GB
• 合计 ≈ 7.5 GB

所以 8GB 显存的 GPU（RTX 3060 8GB、RTX 3070 8GB）勉强能跑；RTX 3090/4090 24GB 或 M-series 16GB+ 统一内存宽松。

API key（用来调 Claude 合成数据）：

export ANTHROPIC_API_KEY=sk-ant-...

第 1 步：定义流程图

主篇里说 procedure 表达为有向图 F = (N, E, n₀, T)。5 节点版本（日报归档助手）：

flowchart TD
    A[greet 开场问候<br>请用户粘贴日报] --> B[collect 用户提交日报文本]
    B --> C[classify 分类判断<br>work/study/personal]
    C --> D{分类是否明确?}
    D -- 明确 --> E[confirm 请求归档确认]
    D -- 模糊 --> C
    E --> F{用户是否确认?}
    F -- go --> G[done_ok 归档完成]
    F -- cancel --> H[done_cancel 已取消]

代码表达（flowchart.py）：

# flowchart.py
NODES = {
    "greet": {
        "role": "assistant",
        "template": "打招呼，简介功能，请用户粘贴今天的日报文本。",
    },
    "collect": {
        "role": "user",
        "template": "用户粘贴一段 100-300 字的日报文本，风格随机（技术工作/学习笔记/生活流水）。",
    },
    "classify": {
        "role": "assistant",
        "template": "读日报内容，判断分类倾向（work/study/personal），询问用户确认。语言要自然。",
    },
    "classify_answer": {
        "role": "user",
        "template": "用户确认分类，或指出分类不对给出正确的，或者模糊回答（如'都有一点'）。",
    },
    "confirm": {
        "role": "assistant",
        "template": "复述归档路径 /{category}/{YYYY-MM-DD}/，请用户回复 'go' 确认或 'cancel' 取消。",
    },
    "user_confirm": {
        "role": "user",
        "template": "用户回复 go 或 cancel（约 70% go）。",
    },
    "done_ok": {
        "role": "assistant",
        "template": "告诉用户归档已完成，询问是否还需要其他帮助。",
    },
    "done_cancel": {
        "role": "assistant",
        "template": "确认已取消，无内容归档，礼貌结束。",
    },
}

EDGES = [
    ("greet", "collect"),
    ("collect", "classify"),
    ("classify", "classify_answer"),
    ("classify_answer", "confirm", "clear"),      # 明确边
    ("classify_answer", "classify", "ambiguous"), # 模糊回环
    ("confirm", "user_confirm"),
    ("user_confirm", "done_ok", "go"),
    ("user_confirm", "done_cancel", "cancel"),
]

START = "greet"
TERMINALS = {"done_ok", "done_cancel"}

手算：直接路径 7 步（greet→collect→classify→classify_answer→confirm→user_confirm→done_ok/cancel），带 1 次模糊回环 9 步，带 2 次回环 11 步。核心变化点两个：分类是否清楚、是否确认。合成时需要覆盖所有 4 种组合（清楚/模糊 × go/cancel）。

第 2 步：合成对话生成

调 Claude Sonnet 4.5 按流程图生成 200 条对话（synthesize.py）：

# synthesize.py
import os, json, random, itertools
from anthropic import Anthropic
from flowchart import NODES, EDGES, START, TERMINALS

client = Anthropic()

def sample_path(seed: int) -> list[str]:
    """从起点走到终止节点，在决策枢纽随机选边。"""
    random.seed(seed)
    path = [START]
    current = START
    while current not in TERMINALS:
        options = [e for e in EDGES if e[0] == current]
        # 决策枢纽处按概率选（模糊回环给 25%，取消给 30%）
        if current == "classify_answer":
            weights = [0.75, 0.25]  # clear vs ambiguous
        elif current == "user_confirm":
            weights = [0.70, 0.30]  # go vs cancel
        else:
            weights = [1.0]
        next_edge = random.choices(options, weights=weights)[0]
        current = next_edge[1]
        path.append(current)
        if len(path) > 15:  # 防死循环
            break
    return path

def generate_conversation(path: list[str], seed: int) -> list[dict]:
    """让 Claude 沿路径生成自然对话。"""
    steps = [
        f"[{i}] node={n} role={NODES[n]['role']} intent={NODES[n]['template']}"
        for i, n in enumerate(path)
    ]
    prompt = f"""你要为一个日报归档助手生成一段自然的多轮对话。

# 流程约束
{chr(10).join(steps)}

# 生成规则
- 严格按上面的节点顺序生成，一个节点对应一次发言
- assistant 节点由助手说，user 节点由用户说
- 语言要自然，避免模板化措辞；不要显式说出节点编号
- 用户日报文本要贴合分类（work 提到 PR/bug/会议；study 提到论文/课程；personal 提到运动/家务）

# 输出格式
仅输出严格的 JSON 数组，元素形如 {{"role":"assistant","content":"..."}}，共 {len(path)} 条。不要输出任何其他解释文本。
"""
    resp = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=2048,
        messages=[{"role": "user", "content": prompt}],
    )
    text = resp.content[0].text.strip()
    # 容错剥壳
    if text.startswith("```"):
        text = text.split("```")[1].lstrip("json").strip()
    return json.loads(text)

def main(n_conversations: int = 200, seed_base: int = 42):
    with open("train.jsonl", "w") as f:
        for i in range(n_conversations):
            seed = seed_base + i
            path = sample_path(seed)
            try:
                msgs = generate_conversation(path, seed)
                f.write(json.dumps({"messages": msgs}, ensure_ascii=False) + "\n")
                if (i + 1) % 10 == 0:
                    print(f"generated {i+1}/{n_conversations}")
            except Exception as e:
                print(f"skip {i}: {e}")

if __name__ == "__main__":
    main()

手算成本：

• 输入 ~350 tokens × 200 次 × $3/M = **$0.21**
• 输出 ~500 tokens × 200 次 × $15/M = **$1.50**
• 合计 ≈ $1.71

手算时间：串行时每次调用 3–5 秒，200 次 = 10–16 分钟。想快 5×，把上面加个 asyncio.gather+AsyncAnthropic，并发 10 就够（免费账户 RPM 有限，别开太狂）。

跑完检查 train.jsonl：应该有 200 行，每行是一个 {"messages": [...]} 对话。抽 3 条肉眼过一遍，确认对话自然度和路径分布——尤其看 4 种终态组合（清楚→go、清楚→cancel、模糊→go、模糊→cancel）是否都覆盖。

第 3 步：全量微调 Qwen2.5-0.5B

train.py：

# train.py
import torch
from datasets import load_dataset
from transformers import AutoModelForCausalLM, AutoTokenizer, TrainingArguments
from trl import SFTTrainer, SFTConfig

MODEL = "Qwen/Qwen2.5-0.5B-Instruct"

tokenizer = AutoTokenizer.from_pretrained(MODEL)
if tokenizer.pad_token is None:
    tokenizer.pad_token = tokenizer.eos_token

model = AutoModelForCausalLM.from_pretrained(
    MODEL,
    torch_dtype=torch.bfloat16,
    device_map="auto",
)

dataset = load_dataset("json", data_files="train.jsonl", split="train")

def format_chat(example):
    """把 messages 列表用 chat template 拼成训练文本。"""
    text = tokenizer.apply_chat_template(
        example["messages"],
        tokenize=False,
        add_generation_prompt=False,
    )
    return {"text": text}

dataset = dataset.map(format_chat, remove_columns=["messages"])

config = SFTConfig(
    output_dir="./subterranean-0.5b",
    num_train_epochs=20,
    per_device_train_batch_size=4,
    gradient_accumulation_steps=4,   # effective batch = 16
    learning_rate=2e-5,               # 论文原参
    lr_scheduler_type="cosine",
    warmup_ratio=0.1,
    bf16=True,
    logging_steps=5,
    save_strategy="epoch",
    save_total_limit=2,
    max_seq_length=1024,
    dataset_text_field="text",
    report_to="none",
)

trainer = SFTTrainer(
    model=model,
    tokenizer=tokenizer,
    train_dataset=dataset,
    args=config,
)

trainer.train()
trainer.save_model("./subterranean-0.5b-final")
tokenizer.save_pretrained("./subterranean-0.5b-final")

手算训练时长：

• 200 样本 × 20 epochs / effective batch 16 = 250 优化步
• RTX 4090：约 0.3 s/step → ~1.5 分钟
• RTX 3090：约 0.5 s/step → ~2 分钟
• M2 Pro (mps)：约 3 s/step → ~12 分钟（记得改 device_map=None，让 accelerate 自动挑 mps）

为什么不用 LoRA：这里必须坚持全量微调。论文 §2 明确写：“系统研究 LoRA rank 16–128，在 procedural 任务上都够不着全量微调”。原因是过程内化要求修改模型的隐式状态跟踪行为——这比风格调整更深，rank 有限的低秩更新装不下。想省显存请上量化（bnb 8-bit optimizer）或换更小基座，不要退化成 LoRA。

训练监控自检：

• 初始 loss：应该在 2.5–4 之间。如果 >6，检查 chat template 是否拼错
• 收敛 loss：跑完 20 epochs 应该在 0.3–0.6 之间。<0.1 说明过拟合，需要更多样化的合成数据（seed 分布再散一点）
• 训练曲线：正常应是平滑单调下降 + 末段平台。如果反复震荡，把 warmup_ratio 调到 0.15

第 4 步：部署 + 对照 Claude in-context

eval.py：

# eval.py
import os, json, time
import torch
from anthropic import Anthropic
from transformers import AutoModelForCausalLM, AutoTokenizer
from flowchart import NODES, EDGES

# ---- 编译版本 ----
COMPILED = "./subterranean-0.5b-final"
tokenizer = AutoTokenizer.from_pretrained(COMPILED)
model = AutoModelForCausalLM.from_pretrained(COMPILED, torch_dtype=torch.bfloat16, device_map="auto")

def run_compiled(user_turns: list[str]) -> tuple[list[dict], float]:
    """交替地把用户话喂给编译版模型,让它生成 assistant 回复。"""
    messages = []
    t0 = time.time()
    for u in user_turns:
        messages.append({"role": "user", "content": u})
        inputs = tokenizer.apply_chat_template(
            messages, add_generation_prompt=True, return_tensors="pt"
        ).to(model.device)
        out = model.generate(inputs, max_new_tokens=200, do_sample=False, temperature=0.0)
        reply = tokenizer.decode(out[0][inputs.shape[1]:], skip_special_tokens=True)
        messages.append({"role": "assistant", "content": reply})
    return messages, time.time() - t0

# ---- Claude in-context 对照版 ----
client = Anthropic()
BASELINE_SYSTEM = f"""你是一个日报归档助手。严格按以下流程图对话：

节点定义：
{json.dumps(NODES, ensure_ascii=False, indent=2)}

边定义：
{json.dumps(EDGES, ensure_ascii=False, indent=2)}

从 greet 开始，遇到决策枢纽按语义判断走哪条边，直到到达 done_ok 或 done_cancel。
"""

def run_baseline(user_turns: list[str]) -> tuple[list[dict], float]:
    messages = []
    t0 = time.time()
    for u in user_turns:
        messages.append({"role": "user", "content": u})
        resp = client.messages.create(
            model="claude-sonnet-4-5",
            system=BASELINE_SYSTEM,
            max_tokens=200,
            messages=messages,
        )
        messages.append({"role": "assistant", "content": resp.content[0].text})
    return messages, time.time() - t0

# ---- 20 条测试对话 ----
TEST_CASES = [
    # 清楚→go 类
    ["今天关掉了 3 个 P2 bug,合了一个 auth 相关的 PR", "对", "go"],
    ["读完了 Attention is all you need,笔记做了 30 分钟", "确认 study", "go"],
    # 清楚→cancel
    ["下午跑了 5km,晚上做了饭", "personal 对", "cancel"],
    # 模糊→go
    ["白天 code review,晚上看了论文", "都有一点吧", "算 work", "go"],
    # ... 再加 16 条覆盖各种边界
]

# ---- LLM-as-judge 打分 ----
def judge(compiled_out, baseline_out, test_case):
    prompt = f"""对比两个日报归档助手的对话表现,针对同一条用户输入序列。
用户输入: {test_case}

助手 A(编译版):
{json.dumps(compiled_out, ensure_ascii=False, indent=2)}

助手 B(Claude in-context):
{json.dumps(baseline_out, ensure_ascii=False, indent=2)}

从 5 个维度打分(1-5),仅输出 JSON:
task_success, info_accuracy, consistency, graceful_handling, naturalness
每个维度形如 {{"A": 分, "B": 分}}
"""
    resp = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=400,
        messages=[{"role": "user", "content": prompt}],
    )
    return json.loads(resp.content[0].text.strip("`json\n "))

# ---- 主循环 ----
if __name__ == "__main__":
    results = []
    for tc in TEST_CASES:
        a, ta = run_compiled(tc)
        b, tb = run_baseline(tc)
        s = judge(a, b, tc)
        results.append({"case": tc, "compiled_time": ta, "baseline_time": tb, "scores": s})
    with open("eval_results.json", "w") as f:
        json.dump(results, f, ensure_ascii=False, indent=2)
    # 简单聚合
    for dim in ["task_success", "info_accuracy", "consistency", "graceful_handling", "naturalness"]:
        avg_a = sum(r["scores"][dim]["A"] for r in results) / len(results)
        avg_b = sum(r["scores"][dim]["B"] for r in results) / len(results)
        print(f"{dim}: 编译版 {avg_a:.2f} | Claude in-context {avg_b:.2f}")

第 5 步：期待看到的对照表

跑完 eval.py 后应该得到类似（这些是预期数字，不是实测）：

维度	0.5B 编译版	Claude 4.5 in-context	差距解释
Task Success (1-5)	3.8–4.3	4.7–4.9	0.5B 偶尔在决策枢纽走错;in-context 强
Info Accuracy	3.5–4.0	4.8–4.9	0.5B 复述用户信息偶尔漏字段;这是 0.5B 广度知识不足
Consistency	4.0–4.5	4.9–5.0	编译版有时前后措辞矛盾
Graceful Handling	3.0–4.0	4.5–4.8	用户越界时,0.5B 更容易死板
Naturalness	4.0–4.5	4.7–4.9	训练数据来自 Claude,继承了一定自然度

对照成本（每次 3-4 轮完整对话）：

• 编译版：M-series 本地免费；A100 云上 $2.5/hr ÷ 3600s × 3s = **$0.002**
• Claude in-context：约 1000 输入 tokens × $3/M + 300 输出 ×$15/M = $0.008
• 比值约 4×

为什么这里只有 4× 而不是主篇的 462×？ 主篇的 462× 是 55 节点保险流程——system prompt 塞进 55 节点的 procedure 就要 ~10K tokens，编译版是 0 procedure tokens。5 节点流程本身占用少（~1K tokens），所以 in-context 的 token 开销没那么可怕，差距被压缩到个位数倍。流程越复杂，编译版的相对优势越大——这是论文表 6 里 IC/Sub 比值从 128× 涨到 462× 的机制。

第 6 步：12 项自检清单

复现前打印一份逐条勾：

环境

• Python 3.10+，PyTorch 2.4+，transformers ≥ 4.44，trl ≥ 0.11
• ANTHROPIC_API_KEY 已导出，from anthropic import Anthropic; Anthropic().messages.create(...) 能跑通
• GPU 显存 ≥ 8GB 或 Mac 统一内存 ≥ 16GB

合成数据

• train.jsonl 200 行，每行是 {"messages": [...]}
• 抽 3 条肉眼看：语言自然、路径完整、终态到 done_ok 或 done_cancel
• 4 种组合（清楚/模糊 × go/cancel）都至少各有 20 条

训练

• 初始 loss ∈ [2.5, 4]，收敛 loss ∈ [0.3, 0.6]
• 训练曲线平滑单调下降，没有反复震荡
• 保存的模型加载后 model.config.vocab_size 与 tokenizer 匹配

评测

• 20 条测试对话覆盖 4 种终态组合
• 编译版每次生成延迟 ≤ 5s（M-series）或 ≤ 1s（4090）
• Claude judge 打分 JSON 能被 json.loads 解析

如果任一条不过，回上一步定位。

第 7 步：常见坑与解法

坑 1：SFTTrainer 报 packing 或 max_seq_length 相关警告。 trl 0.11+ 把这些字段从 TrainingArguments 挪到 SFTConfig，用 SFTConfig 别用 TrainingArguments。

坑 2：Loss 不下降或 NaN。 检查 bf16 支持（Ampere 及以上）、check chat template 是否正确拼上系统 token 和 <|im_start|>/<|im_end|>；打印一条格式化后的样本确认结构。

坑 3：编译版生成了不该有的内容（幻觉出用户没提的字段）。 这是 0.5B 广度不足的经典失败——参考主篇里论文的观察，8B 版把 info accuracy 从 82% 拉到 87%，进一步扩容才补得齐。这不是训练问题，是容量问题。

坑 4：合成对话里 Claude 偶尔”违反”节点顺序（比如提前告知归档路径）。 生成 prompt 里把「严格按节点顺序」改成「必须逐节点、不许提前进入下一节点」，或者用少样本示例（把 2 条正确对话贴进 prompt）。

坑 5：M-series 上 device_map="auto" 卡在 CPU。 换成 device_map=None，手动 .to("mps")，或者用 accelerate 的 --multi_gpu 变体让它挑设备。

坑 6：想更省显存。 用 bitsandbytes 的 8-bit AdamW：SFTConfig(..., optim="adamw_8bit")——把 Adam 状态从 4GB 砍到 1GB，代价是训练略慢。不要换 LoRA。

第 8 步：怎么从 0.5B 升到 8B

如果你想真正接近论文效果，规模需要放大：

项目	本篇	论文规模	升级方式
基座	Qwen2.5-0.5B	Qwen3-8B	换权重 + 显存至少 24GB（fp16 全量约 22GB）
硬件	1×消费卡	8×A100 80GB	云上 4-8× A100/H100 + DeepSpeed ZeRO-3
数据	200 条	2,700–6,264 条	合成规模 × 15，API 花费 ~$25
流程	5 节点	14–55 节点	换真实业务流程（客服/结算/审核）
训练时长	分钟级	数小时	8×A100 上 8B 大约 3-5 小时
总花费	~$2	~$60–80（含云 GPU）	见论文 §5

盈亏平衡点还是主篇那句：500 次会话之内。你自己场景的流程一年跑得动 500 次就够本。

参考

• 主篇：过程知识住在哪里？——Agent 编排的第四把交椅
• 论文：Compiling Agentic Workflows into LLM Weights（arXiv 2605.22502）
• 相关实操：本地跑通一个真实 PEFT LoRA、nanoGPT Shakespeare 字符级 LM
• 官方文档：trl SFTTrainer、Anthropic Python SDK、Qwen2.5-0.5B-Instruct

一个诚实的收尾

再重复一遍：这篇的代码是根据官方文档写的、数字是手算估计，我没有实机跑通全流程。如果你跑通了，欢迎在 GitHub 上开 issue 告诉我：合成对话 200 条实际花了多少 API 钱、0.5B 训练 loss 曲线实际长什么样、20 条测试对话的 judge 打分对照跟我预测的差多少。这份陪读的价值不在纸面数字，在你跑一遍之后手上多出的那份手感——把「过程知识住进权重」这个抽象概念，用一次实机训练把它固化下来。