研究毕业实验：小模型适配与 Agent 改进报告

import copy
import json
import statistics
from pathlib import Path
import torch
from torch import nn
from torch.nn import functional as F

torch.set_num_threads(1)
torch.use_deterministic_algorithms(True)
def dataset(seed, count, prefix):
    generator = torch.Generator().manual_seed(seed)
    x = torch.randn(count, 8, generator=generator)
    y = (x[:, 0]+.7*x[:, 1]-.5*x[:, 2] > 0).long()
    return x, y, [f"{prefix}-{i}" for i in range(count)]
train = dataset(1000, 128, "train")
valid = dataset(2000, 96, "valid")
test = dataset(3000, 128, "test")
assert not (set(train[2]) & set(valid[2]) or set(train[2]) & set(test[2]) or set(valid[2]) & set(test[2]))
base = torch.randn(2, 8, generator=torch.Generator().manual_seed(99))*.15

class Policy(nn.Module):
    def __init__(self, mode):
        super().__init__()
        self.mode = mode
        self.register_buffer("base", base.clone())  # 冻结数据，不交给优化器。
        if mode == "full":
            self.weight = nn.Parameter(base.clone())
        elif mode.startswith("lora"):
            self.a = nn.Parameter(torch.randn(1, 8)*.1)
            self.b = nn.Parameter(torch.zeros(2, 1))  # 初始增量为零。
    def forward(self, x):
        if self.mode == "lora_without_feature":
            x = x.clone()
            x[:, 2] = 0  # 只消融输入信息，环境正确答案保持原定义。
        if self.mode == "full":
            return F.linear(x, self.weight)
        scores = F.linear(x, self.base)
        if self.mode.startswith("lora"):
            scores = scores + F.linear(F.linear(x, self.a), self.b)
        return scores

def run(mode, seed):
    torch.manual_seed(seed)
    model = Policy(mode)
    params = list(model.parameters())
    if mode == "lora":
        assert torch.equal(model(train[0]), F.linear(train[0], base))
        assert sum(p.numel() for p in params) == 10
    if params:
        optimizer = torch.optim.Adam(params, lr=.05)
        generator = torch.Generator().manual_seed(seed+10000)
        best_loss, best_state = float("inf"), None
        for step in range(120):
            indices = torch.randperm(len(train[1]), generator=generator)[:32]
            optimizer.zero_grad()
            loss = F.cross_entropy(model(train[0][indices]), train[1][indices])
            assert torch.isfinite(loss)
            loss.backward()
            optimizer.step()
            if (step+1) % 20 == 0:
                with torch.no_grad():
                    score = F.cross_entropy(model(valid[0]), valid[1]).item()
                if score < best_loss:
                    best_loss, best_state = score, copy.deepcopy(model.state_dict())
        model.load_state_dict(best_state)  # 选择规则完全不读取测试标签。
    assert torch.equal(model.base, base)
    with torch.no_grad():
        actions = model(test[0]).argmax(-1)
    assert set(actions.tolist()) <= {0, 1}
    successes = actions.eq(test[1]).to(torch.int).tolist()
    return {"mode": mode, "seed": seed, "trainable_parameters": sum(p.numel() for p in params),
            "success_rate": statistics.mean(successes),
            "tasks": [{"id": i, "action": a, "success": s} for i, a, s in zip(test[2], actions.tolist(), successes)]}

modes = ["frozen", "full", "lora", "lora_without_feature"]
results = [run(mode, seed) for mode in modes for seed in range(5)]
assert len(results) == 20 and all(len(r["tasks"]) == 128 for r in results)
assert {r["trainable_parameters"] for r in results if r["mode"] == "full"} == {16}
assert all(0 <= r["success_rate"] <= 1 for r in results)
summary = {mode: {"mean": statistics.mean(r["success_rate"] for r in results if r["mode"] == mode),
                   "seed_std": statistics.stdev(r["success_rate"] for r in results if r["mode"] == mode)} for mode in modes}
report = {"torch": torch.__version__, "scope": "synthetic one-step CPU policy; not LLM or multi-turn benchmark",
          "protocol": {"train": 128, "valid": 96, "test": 128, "steps": 120, "batch": 32,
                       "checkpoint_rule": "minimum validation loss every 20 steps", "seeds": list(range(5))},
          "summary": summary, "runs": results}
Path("report.json").write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
print(json.dumps(summary, indent=2))
print("PASS: 20 runs, 128 tasks each; report.json saved; no significance claim")