第07章 - 大模型微调

# SFT 样本拼接与 loss mask 的核心逻辑（框架无关，PyTorch 写法）
# 假设 tokenizer 已定义；chat 模板以所用模型官方 template 为准
prompt_ids = tokenizer.apply_chat_template(messages[:-1], add_generation_prompt=True)
answer_ids = tokenizer(messages[-1]["content"], add_special_tokens=False).input_ids
answer_ids = answer_ids + [tokenizer.eos_token_id]

input_ids = prompt_ids + answer_ids
labels = [-100] * len(prompt_ids) + answer_ids   # 关键：prompt 部分 mask 成 -100

# 训练时不要把 prompt 的 loss 算进去：
# 否则模型会去学"用户会怎么提问"，浪费容量，还会让 loss 曲线看起来更平滑
loss = torch.nn.functional.cross_entropy(
    logits[:, :-1, :].reshape(-1, logits.size(-1)).float(),
    labels_tensor[:, 1:].reshape(-1),
    ignore_index=-100,
)

# data_audit.py —— 训练前数据体检：长度分布、近似去重、格式校验、标签泄漏
import json, hashlib, re, sys
from collections import Counter
from difflib import SequenceMatcher

def norm(text: str) -> str:
    return re.sub(r"\s+", " ", text.strip().lower())

def shingles(text: str, k: int = 5) -> set:
    t = norm(text)
    return {t[i:i + k] for i in range(max(0, len(t) - k + 1))}

def jaccard(a: set, b: set) -> float:
    return len(a & b) / max(1, len(a | b))

def audit(path: str, max_chars: int = 4096):
    rows, seen_hash, lens, role_err, empty = [], set(), [], 0, 0
    with open(path, encoding="utf-8") as f:
        for lineno, line in enumerate(f, 1):
            line = line.strip()
            if not line:
                continue
            try:
                obj = json.loads(line)
            except json.JSONDecodeError as e:
                print(f"[格式错误] 第 {lineno} 行: {e}")
                continue
            msgs = obj.get("messages") or obj.get("conversations") or []
            if not msgs:
                role_err += 1
                continue
            joined = "\n".join(str(m.get("content", "")) for m in msgs)
            if not joined.strip():
                empty += 1
                continue
            h = hashlib.md5(norm(joined).encode()).hexdigest()
            if h in seen_hash:
                print(f"[完全重复] 第 {lineno} 行")
            seen_hash.add(h)
            lens.append(len(joined))
            rows.append((lineno, joined))
    # 近似重复：采样对比，避免 O(n^2) 全量比较
    step = max(1, len(rows) // 400)
    sample = rows[::step]
    sh = [(ln, shingles(t)) for ln, t in sample]
    dup = 0
    for i in range(len(sh)):
        for j in range(i + 1, len(sh)):
            if jaccard(sh[i][1], sh[j][1]) > 0.85:
                dup += 1
    lens.sort()
    p = lambda q: lens[int(len(lens) * q)] if lens else 0
    print(f"样本数={len(rows)} 空样本={empty} 角色错误={role_err}")
    print(f"长度分位: p50={p(0.5)} p90={p(0.9)} p99={p(0.99)} max={lens[-1] if lens else 0}")
    if lens and p(0.99) > max_chars:
        print(f"[警告] p99 超过 {max_chars} 字符，训练前需截断，否则 padding 会拖慢一半吞吐")
    print(f"近似重复（采样 {len(sh)} 条，阈值 0.85）: {dup} 对")
    if empty or role_err:
        sys.exit(1)

if __name__ == "__main__":
    audit(sys.argv[1])

# make_masks.py —— 用矢量面 + 影像生成 (image, mask, class_name) 训练样本
# 依赖：rasterio、geopandas、numpy、Pillow；按瓦片生成，直接用 512×512 滑窗切
import json
import numpy as np
import rasterio
from rasterio.features import rasterize
from rasterio.windows import Window
import geopandas as gpd
from PIL import Image

CLASS_MAP = {                      # 内部口径 -> (类别 id, 概念提示词)
    "building": (1, "a building rooftop"),
    "road": (2, "a paved road surface"),
    "solar_panel": (3, "a solar photovoltaic panel array"),
    "stockpile": (4, "a construction material stockpile"),
}
TILE = 512
STRIDE = 384                       # 重叠 25%，保证目标不被切断
MIN_PIXELS = 64                    # 掩码像素数低于此值的瓦片丢弃（噪声标注）

def build(imagery: str, vectors: str, layer: str, out_dir: str, index: str):
    gdf = gpd.read_file(vectors, layer=layer)
    with rasterio.open(imagery) as src:
        gdf = gdf.to_crs(src.crs)                      # 关键：先对齐坐标系
        shapes = []
        for _, row in gdf.iterrows():
            cls = row.get("category") or row.get("type")
            if cls not in CLASS_MAP:
                continue
            shapes.append((row.geometry, CLASS_MAP[cls][0]))
        # 整图栅格化，再按窗口切，避免每块重复算几何
        full_mask = rasterize(
            shapes, out_shape=(src.height, src.width),
            transform=src.transform, fill=0, dtype="uint8", all_touched=False,
        )
        full_mask = np.where(full_mask == 255, 0, full_mask)   # 防御：无类别值归 0
        records = []
        n = 0
        for row_off in range(0, src.height - TILE + 1, STRIDE):
            for col_off in range(0, src.width - TILE + 1, STRIDE):
                win = Window(col_off, row_off, TILE, TILE)
                img = src.read([1, 2, 3], window=win)
                if img.shape != (3, TILE, TILE):
                    continue
                msk = full_mask[row_off:row_off + TILE, col_off:col_off + TILE]
                present = [cid for cid in CLASS_MAP.values() if (msk == cid[0]).sum() >= MIN_PIXELS]
                present = [c for c in present if (msk == c[0]).sum() >= MIN_PIXELS]
                if not present:
                    continue                                # 全背景瓦片不进训练集
                tid = f"tile_{n:06d}"
                Image.fromarray(np.transpose(img, (1, 2, 0))).save(f"{out_dir}/{tid}.png")
                Image.fromarray(msk, mode="L").save(f"{out_dir}/{tid}_mask.png")
                for cid, prompt in present:
                    records.append({
                        "image": f"{tid}.png",
                        "mask": f"{tid}_mask.png",
                        "class_id": cid,
                        "prompt": prompt,
                        "n_pixels": int((msk == cid).sum()),
                    })
                n += 1
    with open(index, "w", encoding="utf-8") as f:
        for r in records:
            f.write(json.dumps(r, ensure_ascii=False) + "\n")
    print(f"生成瓦片 {n} 张，样本 {len(records)} 条")

if __name__ == "__main__":
    build("data/ortho_2024.tif", "data/vectors.gpkg", "facilities",
          "data/tiles", "data/train_index.jsonl")

# train_sam3_lora.py —— 训练骨架（伪代码级：模型加载与前向签名以官方实现为准）
import torch
import torch.nn.functional as F

def focal_loss(logits, target, alpha=0.25, gamma=2.0):
    # logits: (B, 1, H, W)；target: (B, 1, H, W) 取值 {0,1}，忽略区为 -1
    valid = (target >= 0).float()
    p = torch.sigmoid(logits)
    ce = F.binary_cross_entropy_with_logits(logits, target.clamp(0, 1), reduction="none")
    p_t = p * target + (1 - p) * (1 - target)
    loss = ce * ((1 - p_t) ** gamma)
    loss = loss * (alpha * target + (1 - alpha) * (1 - target))
    return (loss * valid).sum() / valid.sum().clamp(min=1.0)

def dice_loss(logits, target, eps=1e-6):
    valid = (target >= 0).float()
    p = torch.sigmoid(logits) * valid
    t = target.clamp(0, 1) * valid
    inter = (p * t).sum(dim=(1, 2, 3))
    union = p.sum(dim=(1, 2, 3)) + t.sum(dim=(1, 2, 3))
    return 1 - ((2 * inter + eps) / (union + eps)).mean()

def build_optimizer(model, lr=1e-4, weight_decay=0.01):
    trainable = []
    for name, p in model.named_parameters():
        # 冻结策略：image encoder 全部冻结；decoder/prompt encoder 可训
        is_encoder = "image_encoder" in name
        is_lora = "lora_" in name
        p.requires_grad = (not is_encoder) or is_lora
        if p.requires_grad:
            trainable.append(p)
    n = sum(p.numel() for p in trainable)
    print(f"可训练参数 {n/1e6:.2f} M / 总参数 {sum(p.numel() for p in model.parameters())/1e6:.1f} M")
    return torch.optim.AdamW(trainable, lr=lr, weight_decay=weight_decay)

def train_step(model, batch, weight_focal=20.0, weight_dice=1.0):
    # batch: image, prompt(点/框/概念词), gt_mask
    pred = model(image=batch["image"], prompt=batch["prompt"], multimask_output=True)
    loss_total = 0.0
    for logits in pred["masks"]:            # 多候选输出要全部参与监督或取最优候选后监督
        loss_total = loss_total + weight_focal * focal_loss(logits, batch["gt_mask"]) \
                                 + weight_dice * dice_loss(logits, batch["gt_mask"])
    return loss_total / len(pred["masks"])

# 典型超参（SAM 系微调，需按数据量微调）
CONFIG = {
    "lr": 1e-4,              # decoder-only 可到 3e-4；带 LoRA 时降到 1e-4
    "weight_decay": 0.01,
    "batch_size": 4,         # 512×512、24 GB 卡的安全值
    "grad_accum": 4,
    "epochs": 20,
    "warmup_ratio": 0.05,
    "scheduler": "cosine",
    "bf16": True,
    "grad_checkpoint": False,  # encoder 冻结后通常不必要
}

# eval_seg.py —— per-class IoU、BF-score、按尺寸分层的召回率
import numpy as np
from scipy import ndimage
from skimage.segmentation import find_boundaries

def boundary_f1(pred: np.ndarray, gt: np.ndarray, tol: int = 2) -> float:
    pb, gb = find_boundaries(pred, mode="inner"), find_boundaries(gt, mode="inner")
    if pb.sum() == 0 and gb.sum() == 0:
        return 1.0
    if pb.sum() == 0 or gb.sum() == 0:
        return 0.0
    dt_gt = ndimage.distance_transform_edt(~gb)
    dt_pred = ndimage.distance_transform_edt(~pb)
    precision = float((dt_gt[pb] <= tol).mean())
    recall = float((dt_pred[gb] <= tol).mean())
    return 0.0 if precision + recall == 0 else 2 * precision * recall / (precision + recall)

def size_bucket(area_px: int) -> str:
    if area_px < 32 * 32:
        return "small"
    if area_px < 96 * 96:
        return "medium"
    return "large"

def evaluate(pairs, class_ids=(1, 2, 3, 4)):
    """pairs: [(pred_mask, gt_mask), ...]，掩码已在同一坐标系、同一尺寸"""
    per_class, buckets = {}, {"small": [0, 0], "medium": [0, 0], "large": [0, 0]}
    for cid in class_ids:
        inter = union = 0
        bfs = []
        for pred, gt in pairs:
            p, g = (pred == cid), (gt == cid)
            inter += int((p & g).sum())
            union += int((p | g).sum())
            if g.any():
                bfs.append(boundary_f1(p, g))
                for inst in ndimage.label(g)[0].nonzero()[0][1:]:
                    pass
        iou = inter / union if union else float("nan")
        per_class[cid] = {"iou": round(iou, 4), "bf": round(float(np.mean(bfs)) if bfs else 0.0, 4)}
    # 小/中/大目标召回：按 GT 连通域面积分桶
    for pred, gt in pairs:
        for cid in class_ids:
            g = (gt == cid)
            if not g.any():
                continue
            labeled, n = ndimage.label(g)
            for idx in range(1, n + 1):
                inst = labeled == idx
                area = int(inst.sum())
                b = size_bucket(area)
                buckets[b][1] += 1
                if ((pred == cid) & inst).sum() / area >= 0.5:   # 命中阈值 0.5
                    buckets[b][0] += 1
    recall = {k: round(v[0] / v[1], 4) if v[1] else None for k, v in buckets.items()}
    miou = float(np.nanmean([per_class[c]["iou"] for c in class_ids]))
    return {"per_class": per_class, "mIoU": round(miou, 4), "recall_by_size": recall}

if __name__ == "__main__":
    import pickle, json
    with open("preds.pkl", "rb") as f:
        pairs = pickle.load(f)
    print(json.dumps(evaluate(pairs), ensure_ascii=False, indent=2))

# mask2polygon.py —— 掩码转矢量：轮廓提取、简化、平滑、面积过滤
import json
import numpy as np
import rasterio
from rasterio.features import shapes
from shapely.geometry import shape, mapping
from shapely.ops import unary_union
from shapely.validation import make_valid

def mask_to_polygons(mask: np.ndarray, transform, class_id: int,
                     min_area_m2: float = 30.0, simplify_px: float = 0.8,
                     smooth_round: int = 1):
    """mask: uint8 类别掩码；transform: rasterio 仿射变换（用于面积与坐标）"""
    binary = (mask == class_id).astype("uint8")
    geoms = []
    for geom, val in shapes(binary, transform=transform):
        if val != 1:
            continue
        g = make_valid(shape(geom))
        # 简化单位跟着栅格分辨率走：像素大小 × 容差，避免固定经纬度值
        tol = abs(transform.a) * simplify_px
        if smooth_round > 0:
            g = g.buffer(tol * 0.5, join_style=1).buffer(-tol * 0.5, join_style=1)
        g = g.simplify(tol, preserve_topology=True)
        if g.is_empty or g.area < min_area_m2:
            continue
        geoms.append(g)
    merged = unary_union(geoms) if geoms else None
    return [] if merged is None else [
        {"type": "Feature", "properties": {"class_id": class_id},
         "geometry": mapping(g)} for g in getattr(merged, "geoms", [merged])
    ]

def run(mask_path: str, out_geojson: str, class_ids=(1, 2, 3, 4)):
    with rasterio.open(mask_path) as src:
        mask = src.read(1)
        feats = []
        for cid in class_ids:
            # 面积阈值按像元换算：pixel_area = |a·e - b·d|，米制投影下即平方米
            t = src.transform
            px_area = abs(t.a * t.e - t.b * t.d)
            feats += mask_to_polygons(mask, t, cid, min_area_m2=30.0 / px_area * px_area)
    with open(out_geojson, "w", encoding="utf-8") as f:
        json.dump({"type": "FeatureCollection", "features": feats}, f, ensure_ascii=False)
    print(f"输出要素 {len(feats)} 个 -> {out_geojson}")

if __name__ == "__main__":
    run("pred_mask.tif", "pred_polygons.geojson")

# 训练前确认 target_modules 是否命中：打印名字里含 merger/projector 的模块
from transformers import AutoModelForImageTextToText
model = AutoModelForImageTextToText.from_pretrained("Qwen3-VL-2B-Instruct", torch_dtype="bfloat16")
hits = [n for n, _ in model.named_modules() if any(k in n.lower() for k in ("merger", "projector", "visual"))]
print("\n".join(hits[:40]))
print(f"命中 {len(hits)} 个模块，请从中挑出投影/融合层写入 target_modules")

# eval_vlm.py —— 字段抽取型评测：严格 JSON 解析 + 逐字段比对 + badcase 导出
import json

FIELD_SPEC = {"count": int, "items": list}      # 按你的任务改
FENCE = chr(96) * 3                              # 三个反引号，避免正则转义麻烦

def parse_strict(text: str):
    """剥掉代码围栏后严格解析；失败返回 None"""
    t = text.strip()
    if t.startswith(FENCE):
        t = t[len(FENCE):]
    if t.rstrip().endswith(FENCE):
        t = t.rstrip()[:-len(FENCE)]
    t = t.strip()
    try:
        return json.loads(t)
    except json.JSONDecodeError:
        i, j = t.find("{"), t.rfind("}")
        if i >= 0 and j > i:
            try:
                return json.loads(t[i:j + 1])
            except json.JSONDecodeError:
                return None
        return None

def score_one(pred_text: str, gold: dict) -> dict:
    obj = parse_strict(pred_text)
    if obj is None:
        return {"parsable": 0, "field_hit": 0, "field_total": len(gold), "exact": 0}
    hit = sum(1 for k, v in gold.items() if obj.get(k) == v)
    return {"parsable": 1, "field_hit": hit, "field_total": len(gold),
            "exact": int(hit == len(gold))}

def report(records):
    n = len(records)
    if not n:
        return {"parsable_rate": 0.0, "field_acc": 0.0, "exact_rate": 0.0}
    p = sum(r["parsable"] for r in records) / n
    fh = sum(r["field_hit"] for r in records)
    ft = sum(r["field_total"] for r in records) or 1
    return {"parsable_rate": round(p, 4), "field_acc": round(fh / ft, 4),
            "exact_rate": round(sum(r["exact"] for r in records) / n, 4)}

if __name__ == "__main__":
    rows = [json.loads(line) for line in open("eval_preds.jsonl", encoding="utf-8")]
    res = [score_one(r["pred"], r["gold"]) for r in rows]
    print(json.dumps(report(res), ensure_ascii=False))
    bad = [r for r, s in zip(rows, res) if not s["parsable"] or s["exact"] == 0]
    with open("badcase.jsonl", "w", encoding="utf-8") as f:
        for b in bad[:200]:
            f.write(json.dumps(b, ensure_ascii=False) + "\n")
    print(f"badcase {len(bad)} 条，前 200 条已写出，交人工复核")

# distill_label.py —— 用教师模型批量标注，落库 + 重试 + 解析校验
import json, time, hashlib, os
from concurrent.futures import ThreadPoolExecutor, as_completed

PROMPT_TMPL = open("prompts/teacher_label_v3.txt", encoding="utf-8").read()

def call_teacher(client, image_path: str, model: str, retries: int = 3) -> str:
    for attempt in range(retries):
        try:
            resp = client.chat.completions.create(
                model=model,
                messages=[{"role": "user", "content": [
                    {"type": "image_url", "image_url": {"url": f"file://{os.path.abspath(image_path)}"}},
                    {"type": "text", "text": PROMPT_TMPL},
                ]}],
                temperature=0.2,          # 标注任务不要 0，0 容易在边界样本上过度自信
                max_tokens=1024,
                timeout=120,
            )
            return resp.choices[0].message.content
        except Exception as e:
            if attempt == retries - 1:
                return f"__ERROR__:{type(e).__name__}"
            time.sleep(2 ** attempt)     # 指数退避
    return "__ERROR__"

def parse_and_check(text: str):
    """返回 (ok, obj_or_reason)；规则校验在这里一次做完"""
    if text.startswith("__ERROR__"):
        return False, "api_error"
    body = text.strip().strip(chr(96) * 3).strip()
    try:
        obj = json.loads(body)
    except json.JSONDecodeError:
        return False, "json_error"
    if not isinstance(obj, dict) or "count" not in obj:
        return False, "schema_error"
    if isinstance(obj["count"], int) and obj["count"] < 0:
        return False, "negative_count"
    return True, obj

def run(items, out_path, model, workers=8):
    client = None   # 按你用的 SDK 初始化
    keep = 0
    with open(out_path, "w", encoding="utf-8") as f, ThreadPoolExecutor(workers) as ex:
        futs = {ex.submit(call_teacher, client, it["image"], model): it for it in items}
        for fu in as_completed(futs):
            it = futs[fu]
            raw = fu.result()
            ok, payload = parse_and_check(raw)
            rec = {"image": it["image"], "scene": it.get("scene"),
                   "ok": ok, "raw": raw if not ok else None}
            if ok:
                rec["label"] = payload
                keep += 1
            f.write(json.dumps(rec, ensure_ascii=False) + "\n")
    print(f"成功 {keep}/{len(items)}，失败样本已标 ok=false，需重跑或人工处理")

if __name__ == "__main__":
    items = [json.loads(l) for l in open("data/unlabeled.jsonl", encoding="utf-8")]
    run(items, "data/teacher_labels.jsonl", "teacher-vl-large")

# reward_fn.py —— GRPO 的可验证奖励：可解析 + 字段齐全 + 值正确，三层叠加
def reward(pred_text: str, gold: dict) -> float:
    obj = parse_strict(pred_text)            # 复用 7.3.5 的严格解析
    if obj is None:
        return -1.0                          # 不可解析给负分，防止模型学会输出空对象
    score = 0.2                              # 能给出合法 JSON 对象
    score += 0.3 * float(set(gold) <= set(obj))   # 字段齐全
    hit = sum(1 for k, v in gold.items() if obj.get(k) == v)
    score += 0.5 * hit / max(1, len(gold))   # 字段值正确率
    return score

# seed_all.py —— 一次设全，避免"只设了一半"的随机性
import os, random
import numpy as np
import torch

def seed_all(seed: int = 42, deterministic: bool = True):
    os.environ["PYTHONHASHSEED"] = str(seed)
    random.seed(seed)
    np.random.seed(seed)
    torch.manual_seed(seed)
    torch.cuda.manual_seed_all(seed)
    if deterministic:
        torch.backends.cudnn.deterministic = True
        torch.backends.cudnn.benchmark = False
    return seed