feat(sft): Phase 13 正式 7B SFT/LoRA baseline（launcher + evaluator + merge tool）

script/verl/sft/evaluate_baseline.py

@@ -0,0 +1,254 @@
+"""SFT baseline evaluator: generate with a model and score choice-protocol output.
+
+Runs a model (base HF model or a LoRA-merged HF directory) over a Phase-8 SFT
+parquet split and reports the two-stage choice-protocol metrics using the SHARED
+choice-aware evaluation layer (``agent.evaluation.classification``), so the
+metrics exactly match the frozen parser / contract (no re-implementation).
+
+For each row the prompt is the exported conversation WITHOUT the assistant gold
+(loss target is never shown at inference). Generation is fixed and identical
+for every model: greedy, ``do_sample=False``, ``num_beams=1``, fixed
+``max_new_tokens`` and seed.
+
+Metrics (per the Phase-13 spec; Stage 2 here is scored against the
+PRE-BUILT gold-containing bundle from the parquet, so the "end-to-end"
+metric below is the FACTORIZED / PROXY e2e, not the true pipeline e2e —
+see ``script/verl/sft/evaluate_true_e2e.py`` for the real chained evaluator):
+- stage1: format_valid, contract_valid, recall@5 (GT in the 5 predicted ids)
+- stage2: format_valid, contract_valid, accuracy-overall,
+          accuracy-when-GT-in-candidates (and counts)
+- proxy_e2e: stage1 recall (same source) AND stage2 correct (gold bundle)
+
+Usage (server, in the SFT venv):
+  python -m script.verl.sft.evaluate_baseline \
+    --model-path <hf dir or merged dir> \
+    --data data/sft/pers_info/test.parquet \
+    --registry cfg/task/registry/pers_info.registry.json \
+    --max-new-tokens 128 --seed 42 --report <out.json>
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+from collections import Counter, defaultdict
+from pathlib import Path
+from typing import Any, Iterable
+import sys
+
+from agent.evaluation.classification import (
+    evaluate_stage1_choices,
+    evaluate_stage2_choices,
+)
+from agent.task.contracts import LeafRegistry
+from agent.task.prompt_choices import PromptChoiceRegistry
+
+FULL_CATALOG_REGISTRY = "pers_info"  # evaluator is registry-driven, not hard-coded
+
+
+def aggregate_baseline(records: list[dict[str, Any]]) -> dict[str, Any]:
+    """Aggregate per-row evaluation records into the Phase-13 metric table.
+
+    ``records`` items: {stage, source_id, ground_truth, candidates,
+    format_valid, contract_valid, correct, recalled} (choice-decode already
+    applied by the shared evaluator). Pure function, no transformers/torch.
+    """
+    by_stage: dict[str, list[dict]] = defaultdict(list)
+    for record in records:
+        by_stage[record["stage"]].append(record)
+
+    def summarize(stage_rows: Iterable[dict]) -> dict[str, Any]:
+        rows = list(stage_rows)
+        n = len(rows)
+        format_valid = sum(r["format_valid"] for r in rows) / n if n else 0.0
+        contract_valid = sum(r["contract_valid"] for r in rows) / n if n else 0.0
+        return {
+            "n": n,
+            "format_valid": format_valid,
+            "contract_valid": contract_valid,
+        }
+
+    # --- stage 1 ---
+    s1 = dict(summarize(by_stage["stage1"]))
+    n_s1 = s1["n"]
+    s1["recall_at_5"] = sum(r["recalled"] for r in by_stage["stage1"]) / n_s1 if n_s1 else 0.0
+    s1["recalled_count"] = sum(r["recalled"] for r in by_stage["stage1"])
+
+    # --- stage 2 ---
+    s2 = dict(summarize(by_stage["stage2"]))
+    n_s2 = s2["n"]
+    correct = sum(r["correct"] for r in by_stage["stage2"])
+    s2["accuracy_overall"] = correct / n_s2 if n_s2 else 0.0
+    gt_in = [r for r in by_stage["stage2"] if r["gt_in_candidates"]]
+    s2["n_gt_in_candidates"] = len(gt_in)
+    s2["accuracy_when_gt_in_candidates"] = (
+        sum(r["correct"] for r in gt_in) / len(gt_in) if gt_in else 0.0
+    )
+    s2["correct_count"] = correct
+
+    # --- factorized/proxy e2e: stage1 recall AND stage2 correct (gold bundle) ---
+    by_source: dict[str, dict[str, bool]] = defaultdict(dict)
+    for r in records:
+        by_source[r["source_id"]][r["stage"]] = r
+    e2e_correct = sum(
+        1
+        for per_source in by_source.values()
+        if per_source.get("stage1", {}).get("recalled")
+        and per_source.get("stage2", {}).get("correct")
+    )
+    e2e = {
+        "pairs": len(by_source),
+        "correct": e2e_correct,
+        "correct_rate": e2e_correct / len(by_source) if by_source else 0.0,
+    }
+
+    return {"stage1": s1, "stage2": s2, "proxy_e2e": e2e}
+
+
+def _evaluate_rows(
+    rows: list[dict[str, Any]],
+    registry: LeafRegistry,
+) -> list[dict[str, Any]]:
+    choices = PromptChoiceRegistry.from_registry(registry)
+    records: list[dict[str, Any]] = []
+    for row in rows:
+        stage = row["stage"]
+        gt = row["ground_truth"]
+        source_id = row["source_id"]
+        if stage == "stage1":
+            evaluation = evaluate_stage1_choices(
+                row["completion"], ground_truth=gt, registry=registry, choices=choices
+            )
+            records.append(
+                {
+                    "stage": "stage1",
+                    "source_id": source_id,
+                    "ground_truth": gt,
+                    "candidates": None,
+                    "completion": row["completion"],
+                    "format_valid": evaluation.format_valid,
+                    "contract_valid": evaluation.contract_valid,
+                    "recalled": evaluation.ground_truth_recalled,
+                    "correct": False,
+                    "gt_in_candidates": False,
+                    "prediction": evaluation.prediction,
+                }
+            )
+        else:
+            candidates = list(row["candidates"])
+            evaluation = evaluate_stage2_choices(
+                row["completion"], ground_truth=gt, candidates=candidates, registry=registry
+            )
+            records.append(
+                {
+                    "stage": "stage2",
+                    "source_id": source_id,
+                    "ground_truth": gt,
+                    "candidates": candidates,
+                    "completion": row["completion"],
+                    "format_valid": evaluation.format_valid,
+                    "contract_valid": evaluation.contract_valid,
+                    "correct": evaluation.correct,
+                    "gt_in_candidates": gt in candidates,
+                    "recalled": False,
+                    "prediction": evaluation.prediction,
+                }
+            )
+    return records
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--model-path", required=True, help="HF model dir (base or LoRA-merged)")
+    parser.add_argument("--data", required=True, help="SFT parquet split (e.g. test.parquet)")
+    parser.add_argument("--registry", required=True, help="Leaf registry JSON")
+    parser.add_argument("--max-new-tokens", type=int, default=128)
+    parser.add_argument("--seed", type=int, default=42)
+    parser.add_argument("--report", required=True, help="metrics JSON output path")
+    parser.add_argument("--completions", help="optional path to precomputed completions JSONL; skips generation")
+    args = parser.parse_args(argv)
+
+    import pyarrow.parquet as pq
+
+    registry = LeafRegistry.from_path(args.registry)
+    rows = pq.read_table(args.data).to_pylist()
+
+    generation: dict[str, Any] = {
+        "model_path": args.model_path,
+        "do_sample": False,
+        "num_beams": 1,
+        "max_new_tokens": args.max_new_tokens,
+        "seed": args.seed,
+    }
+
+    if args.completions:
+        records = []
+        # completions JSONL: one {"source_id"/"stage"/"completion"} per eval row
+        with open(args.completions, encoding="utf-8") as handle:
+            for line in handle:
+                if line.strip():
+                    records.append(json.loads(line))
+        # rows are in parquet order -> match completion entries by identity (stage+source_id)
+        by_key = {(r["stage"], r["source_id"]): r for r in records}
+        enriched = []
+        for row in rows:
+            key = (row["stage"], row["source_id"])
+            if key not in by_key:
+                print(f"error: missing completion for {key}", file=sys.stderr)
+                return 2
+            enriched.append({**row, "completion": by_key[key]["completion"]})
+        rows = enriched
+    else:
+        import torch
+        from transformers import AutoModelForCausalLM, AutoTokenizer
+
+        torch.manual_seed(args.seed)
+        tokenizer = AutoTokenizer.from_pretrained(args.model_path)
+        model = AutoModelForCausalLM.from_pretrained(
+            args.model_path, torch_dtype=torch.bfloat16, device_map="auto"
+        )
+        if tokenizer.pad_token_id is None:
+            tokenizer.pad_token_id = tokenizer.eos_token_id
+        model.eval()
+        import tqdm
+
+        for index in tqdm.tqdm(range(len(rows)), desc="generate"):
+            row = rows[index]
+            messages = row["messages"][:2]  # system + user, no assistant gold
+            text = tokenizer.apply_chat_template(
+                messages, tokenize=False, add_generation_prompt=True
+            )
+            inputs = tokenizer(text, return_tensors="pt").to(model.device)
+            with torch.inference_mode():
+                output = model.generate(
+                    **inputs,
+                    do_sample=False,
+                    num_beams=1,
+                    max_new_tokens=args.max_new_tokens,
+                    pad_token_id=tokenizer.eos_token_id,
+                )
+            input_len = inputs["input_ids"].shape[-1]
+            rows[index]["completion"] = tokenizer.decode(
+                output[0][input_len:], skip_special_tokens=True
+            ).strip()
+
+    records = _evaluate_rows(rows, registry)
+    metrics = aggregate_baseline(records)
+
+    report = {
+        "metrics": metrics,
+        "generation": generation,
+        "per_row": records,
+        "registry": str(args.registry),
+        "data": str(args.data),
+    }
+    rendered = json.dumps(report, ensure_ascii=False, indent=2)
+    path = Path(args.report)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(rendered, encoding="utf-8")
+    print(json.dumps(metrics, ensure_ascii=False, indent=2))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

script/verl/sft/merge_lora_checkpoint.py

@@ -0,0 +1,95 @@
+"""Merge a verl LoRA FSDP checkpoint into a standalone HF model directory.
+
+verl saves LoRA checkpoints in peft-compatible state-dict layout
+(``base_model.model....lora_A.default.weight`` + ``lora_train_meta.json``).
+This script rebuilds a PeftModel on the base weights, loads the checkpoint
+state dict, merges LoRA into the base and saves a plain HF directory that any
+evaluator / downstream RL init can load normally (no verl dependency).
+
+Verification: reports how many checkpoint keys were consumed; a LoRA merge
+with the frozen base weights must consume 100% of keys (base + lora).
+
+Usage:
+  python -m script.verl.sft.merge_lora_checkpoint \
+    --checkpoint <verl global_step dir> \
+    --base-model <HF base dir> \
+    --output <merged HF dir>
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import re
+from pathlib import Path
+import sys
+
+import torch
+
+LORA_A = re.compile(r"(.*)\.lora_A\.default\.weight$")
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--checkpoint", required=True, help="verl checkpoint dir (has model_world_size_1_rank_0.pt)")
+    parser.add_argument("--base-model", required=True, help="HF base model dir (same as trained)")
+    parser.add_argument("--output", required=True, help="merged HF output dir")
+    args = parser.parse_args(argv)
+
+    ckpt_dir = Path(args.checkpoint)
+    model_file = ckpt_dir / "model_world_size_1_rank_0.pt"
+    meta_file = ckpt_dir / "lora_train_meta.json"
+    if not model_file.is_file():
+        print(f"error: {model_file} not found", file=sys.stderr)
+        return 2
+
+    sd = torch.load(model_file, map_location="cpu", weights_only=False)
+    keys = list(sd.keys())
+    targets = sorted(
+        {m.group(1).rsplit(".", 1)[-1] for k in keys for m in [LORA_A.match(k)] if m}
+    )
+    if not targets:
+        print("error: no LoRA keys found in checkpoint", file=sys.stderr)
+        return 2
+
+    meta = {}
+    if meta_file.is_file():
+        meta = json.loads(meta_file.read_text(encoding="utf-8"))
+    rank = meta.get("r", 8)
+    alpha = meta.get("lora_alpha", rank)
+
+    from peft import LoraConfig, PeftModel
+    from transformers import AutoModelForCausalLM, AutoTokenizer
+
+    print(f"[merge] targets: {targets}")
+    print(f"[merge] r={rank} alpha={alpha} keys={len(keys)}")
+    base = AutoModelForCausalLM.from_pretrained(args.base_model, torch_dtype=torch.bfloat16)
+    config = LoraConfig(
+        r=rank,
+        lora_alpha=alpha,
+        target_modules=targets,
+        task_type="CAUSAL_LM",
+    )
+    peft = PeftModel(base, config)
+    missing, unexpected = peft.load_state_dict(sd, strict=False)
+    unexpected = [k for k in unexpected if not k.startswith("base_model.model.model")]
+    if unexpected:
+        print(f"warning: {len(unexpected)} unexpected keys (ignored): {unexpected[:5]}")
+    used = len(keys) - len(unexpected)
+    ratio = used / len(keys)
+    print(f"[merge] consumed {used}/{len(keys)} checkpoint keys ({ratio:.1%})")
+    if ratio < 0.99:
+        print("error: checkpoint keys consumed < 99%; aborting", file=sys.stderr)
+        return 1
+    merged = peft.merge_and_unload()
+    out = Path(args.output)
+    out.mkdir(parents=True, exist_ok=True)
+    merged.save_pretrained(out)
+    tokenizer = AutoTokenizer.from_pretrained(args.base_model)
+    tokenizer.save_pretrained(out)
+    print(f"[merge] saved merged model -> {out}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())