"""随机微型网络：只核对视觉前缀、回答掩码和连接层梯度，不展示看图能力。"""

import torch
from torch import nn
from torch.nn import functional as F


class TinyLanguageModel(nn.Module):
    def __init__(self):
        super().__init__()
        self.embedding = nn.Embedding(8, 16)
        self.position = nn.Embedding(16, 16)
        self.block = nn.TransformerEncoderLayer(16, 2, 32, dropout=0,
                    batch_first=True, norm_first=True, activation="gelu")
        self.head = nn.Linear(16, 8)

    def forward(self, embeddings):
        length = embeddings.shape[1]
        x = embeddings + self.position(torch.arange(length))[None]
        causal = torch.ones(length, length, dtype=torch.bool).triu(1)
        return self.head(self.block(x, src_mask=causal))


def pack(visual, language_model, answers):
    batch = len(visual)
    # 词元 ID：BOS=1，问题=2，回答开始=3，红=4，EOS=5，蓝=6。
    before = torch.ones(batch, 1, dtype=torch.long)
    after = torch.stack([torch.full_like(answers, 2), torch.full_like(answers, 3),
                         answers, torch.full_like(answers, 5)], dim=1)
    embeddings = torch.cat([language_model.embedding(before), visual,
                            language_model.embedding(after)], dim=1)
    labels = torch.full(embeddings.shape[:2], -100, dtype=torch.long)
    labels[:, -2] = answers
    labels[:, -1] = 5
    return embeddings, labels


def answer_loss(logits, labels):
    # 当前位置预测下一个词元；最后一个位置的分数不参加这次损失。
    return F.cross_entropy(logits[:, :-1].reshape(-1, logits.shape[-1]),
                           labels[:, 1:].reshape(-1), ignore_index=-100)


def experiment():
    torch.set_num_threads(1)
    torch.manual_seed(2026)
    torch.backends.mha.set_fastpath_enabled(False)
    vision = nn.Conv2d(3, 8, 4, stride=4)
    language = TinyLanguageModel().eval()
    bridge = nn.Linear(8, 16)
    vision.requires_grad_(False)
    language.requires_grad_(False)
    images = torch.zeros(2, 3, 8, 8)
    images[0, 0] = 1.; images[1, 2] = 1.
    answers = torch.tensor([4, 6])
    with torch.no_grad():
        features = vision(images).flatten(2).transpose(1, 2)
    visual = bridge(features)
    visual.retain_grad()
    embeddings, labels = pack(visual, language, answers)
    logits = language(embeddings)
    loss = answer_loss(logits, labels)
    loss.backward()
    gradient = bridge.weight.grad.norm().item()
    visual_gradient = visual.grad.norm().item()
    frozen_clean = all(p.grad is None for p in [*vision.parameters(), *language.parameters()])
    before = loss.item()
    torch.optim.SGD(bridge.parameters(), lr=.2).step()
    with torch.no_grad():
        embeddings, _ = pack(bridge(features), language, answers)
        after = answer_loss(language(embeddings), labels).item()
    return {"features": list(features.shape), "visual": list(visual.shape),
            "sequence": list(embeddings.shape), "supervised": int((labels != -100).sum()),
            "labels": labels.tolist(), "before": before, "after": after,
            "bridge_gradient": gradient, "visual_gradient": visual_gradient,
            "frozen_clean": frozen_clean}


def main():
    result = experiment()
    print("随机微型网络：这里只核对连接方式与梯度，不测看图能力。")
    print(f"视觉特征 {result['features']} → 连接后 {result['visual']}")
    print(f"拼接序列 {result['sequence']}；监督位置共 {result['supervised']} 个（含 EOS）。")
    print("标签：", result["labels"])
    print(f"损失：{result['before']:.6f} → {result['after']:.6f}")
    print(f"连接层梯度范数 {result['bridge_gradient']:.6f}；视觉位置梯度范数 {result['visual_gradient']:.6f}")
    print("冻结参数没有梯度：", result["frozen_clean"])


if __name__ == "__main__":
    main()

"""第 41 章：固定小型视觉语言模型的真实看图记录。默认只复核随书结果。"""

from argparse import ArgumentParser
from hashlib import sha256
from pathlib import Path
import json
import platform
import time

from vlm_cases import CASES, IMAGE_DIR, score

MODEL = "HuggingFaceTB/SmolVLM-500M-Instruct"
REVISION = "a7da5b986cb59b408707209984f360a5f4ad7e47"
REPORT = Path(__file__).with_name("data") / "visual-language.json"
MAX_NEW_TOKENS = 32


def digest(path):
    result = sha256()
    with Path(path).open("rb") as stream:
        for block in iter(lambda: stream.read(1024*1024), b""):
            result.update(block)
    return result.hexdigest()


def run(output, local_only=False):
    import torch
    import transformers
    import PIL
    from PIL import Image
    from huggingface_hub import snapshot_download
    from transformers import AutoProcessor, AutoModelForImageTextToText

    torch.set_num_threads(4)
    torch.manual_seed(2026)
    snapshot = Path(snapshot_download(MODEL, revision=REVISION,
        local_files_only=local_only,
        allow_patterns=["*.json", "*.txt", "model.safetensors", "README.md"]))
    processor = AutoProcessor.from_pretrained(snapshot, trust_remote_code=False, use_fast=False)
    # 固定为单幅、最长边 512；不使用仓库默认的多块高分辨率设置。
    processor.image_processor.size = {"longest_edge": 512}
    processor.image_processor.do_image_splitting = False
    model = AutoModelForImageTextToText.from_pretrained(snapshot, dtype=torch.float32,
        attn_implementation="eager", trust_remote_code=False).eval()
    report = {
        "complete": False, "model": MODEL, "revision": REVISION,
        "source_sha256": digest(__file__), "cases_sha256": digest(Path(__file__).with_name("vlm_cases.py")),
        "weights_sha256": digest(snapshot/"model.safetensors"),
        "weights_bytes": (snapshot/"model.safetensors").stat().st_size,
        "environment": {"python": platform.python_version(), "torch": torch.__version__,
                        "transformers": transformers.__version__, "pillow": PIL.__version__,
                        "platform": platform.platform(), "device": "cpu", "dtype": "float32", "threads": 4},
        "config": {"max_new_tokens": MAX_NEW_TOKENS, "do_sample": False,
                   "attention": "eager", "image_size": 512, "image_splitting": False,
                   "image_seq_len": processor.image_seq_len,
                   "image_token_id": model.config.image_token_id,
                   "vision_patch_size": model.config.vision_config.patch_size,
                   "vision_hidden_size": model.config.vision_config.hidden_size,
                   "scale_factor": model.config.scale_factor,
                   "text_hidden_size": model.config.text_config.hidden_size,
                   "eos_token_id": model.generation_config.eos_token_id},
        "processor": processor.image_processor.to_dict(),
        "rows": [],
    }
    output.parent.mkdir(parents=True, exist_ok=True)
    observed = []
    handle = model.model.connector.register_forward_hook(
        lambda module, args, result: observed.append(list(result.shape)))
    try:
        for case in CASES:
            image_path = IMAGE_DIR / case["image"]
            with Image.open(image_path) as source:
                image = source.convert("RGB")
            messages = [{"role": "user", "content": [
                {"type": "image"}, {"type": "text", "text": case["question"]}]}]
            prompt = processor.apply_chat_template(messages, add_generation_prompt=True, tokenize=False)
            inputs = processor(text=prompt, images=[image], return_tensors="pt")
            length = inputs.input_ids.shape[1]
            observed.clear()
            started = time.perf_counter()
            with torch.inference_mode():
                generated = model.generate(**inputs, max_new_tokens=MAX_NEW_TOKENS, do_sample=False,
                                           use_cache=True)
            elapsed = time.perf_counter()-started
            token_ids = generated[0, length:].tolist()
            answer = processor.decode(token_ids, skip_special_tokens=True)
            eos = model.generation_config.eos_token_id
            eos_ids = [eos] if isinstance(eos, int) else eos
            row = {**case, "image_sha256": digest(image_path), "image_size": list(image.size),
                   "prompt": prompt, "input_ids": inputs.input_ids[0].tolist(),
                   "input_tokens": length,
                   "visual_tokens": int((inputs.input_ids == model.config.image_token_id).sum()),
                   "pixel_values_shape": list(inputs.pixel_values.shape),
                   "valid_pixels": int(inputs.pixel_attention_mask.sum()),
                   "connector_shapes": list(observed),
                   "output_ids": token_ids, "answer": answer, "seconds": elapsed,
                   "stop_reason": "eos" if token_ids and token_ids[-1] in eos_ids else "limit",
                   "score": score(case, answer)}
            report["rows"].append(row)
            output.write_text(json.dumps(report, ensure_ascii=False, indent=2)+"\n")
            print(f"{case['id']}: {answer!r}；{row['score']}", flush=True)
    finally:
        handle.remove()
    report["complete"] = True
    output.write_text(json.dumps(report, ensure_ascii=False, indent=2)+"\n")
    return report


def checked_report(path=REPORT):
    report = json.loads(path.read_text())
    assert report["complete"] and report["model"] == MODEL and report["revision"] == REVISION
    assert report["source_sha256"] == digest(__file__)
    assert report["cases_sha256"] == digest(Path(__file__).with_name("vlm_cases.py"))
    assert report["config"]["max_new_tokens"] == MAX_NEW_TOKENS
    assert len(report["rows"]) == len(CASES)
    for row, case in zip(report["rows"], CASES):
        assert all(row[key] == value for key, value in case.items())
        assert row["image_sha256"] == digest(IMAGE_DIR/case["image"])
        assert row["score"] == score(case, row["answer"])
        assert row["input_tokens"] == len(row["input_ids"])
        assert row["visual_tokens"] == row["input_ids"].count(report["config"]["image_token_id"])
        assert 0 < len(row["output_ids"]) <= MAX_NEW_TOKENS
        assert row["seconds"] > 0
    return report


def print_summary(report):
    print("随书保存的真实图文回答；默认不下载、不加载模型：")
    for row in report["rows"]:
        print(f"{row['id']}: {row['answer']!r}；{row['score']}")
    print("固定 11 道自编题，单次贪心生成；不是通用视觉能力基准。")


def main():
    parser = ArgumentParser(description=__doc__)
    parser.add_argument("--run", action="store_true", help="实际加载权重并重新生成")
    parser.add_argument("--out", type=Path, help="新实验的保存路径")
    parser.add_argument("--local-only", action="store_true", help="只使用本机模型缓存")
    args = parser.parse_args()
    if args.run:
        if args.out is None:
            parser.error("--run 需要 --out，避免无意覆盖随书记录")
        run(args.out, args.local_only)
    else:
        print_summary(checked_report())


if __name__ == "__main__":
    main()

1. 先有 $32\times32=1024$ 个位置，分组后为 $8\times8=64$ 个。
2. 一组包含 16 个向量，拼接后为 $16\times768=12288$ 维。无偏置投影有 $12288\times960=11796480$ 个权重。
3. 自回归模型在当前位置预测下一位置；“回答开始”预测第一个回答词元，第一个回答词元再预测 EOS。
4. 冻结参数只禁止更新这些参数，输入到输出的导数仍可用于连接层训练；no_grad 会让这段前向不被自动微分记录，从而切断该路径。
5. 改变位置 7、8 的输入，不应改变位置 0～6 的输出。测试时关闭 dropout 等随机因素，并保持其他输入相同。
6. 分别为 1、0。无效坐标可能没有定义正面积矩形，直接代入可能得到无意义数值；格式失败与合法框不重叠也代表不同的问题。
7. 记录全部结果，并保证各次对话独立。增加对照以后，结论仍应限定在实际测试的图形、布局与模型设置上。
8. 能检查模型对直接数字标签的依赖；单次成功仍不能证明它理解任意图表、单位或多系列关系。
9. 预先写好每个问题的正确字段与答案，不把答案放进提示。遮挡后若不可读，应把“无法从图中确定”作为需要检查的行为，而不是用原来的隐藏答案评分。
10. 首先确认处理器展开的图像位置与模板一致，只在目标回答和结束标记上监督。微调后的形式更整齐，不代表字段内容更准确，两个方面要分开比较。