File size: 2,803 Bytes
a4b0c5b
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
from __future__ import annotations

import importlib.util
from pathlib import Path

import pytest

pytest.importorskip("torch")
pytest.importorskip("onnxruntime")
pytest.importorskip("transformers")

from vons.models import ModelConfig, build_model

_SPEC = importlib.util.spec_from_file_location(
    "verify_onnx_bundle", Path(__file__).parents[1] / "tools/verify_onnx_bundle.py"
)
assert _SPEC is not None and _SPEC.loader is not None
_MODULE = importlib.util.module_from_spec(_SPEC)
_SPEC.loader.exec_module(_MODULE)
_model_config_from_checkpoint = _MODULE._model_config_from_checkpoint


def test_token_conditioned_checkpoint_config_reconstructs_model_state() -> None:
    expected = ModelConfig(
        hidden_size=256,
        latent_size=64,
        diffusion_steps=64,
        inference_steps=8,
        abstain_threshold=0.55,
        diffusion_conditioning="token_cross_attention",
        attention_heads=4,
    )
    trained_head = build_model("diffusion", option_count=32, config=expected)
    checkpoint = {
        "backend": "diffusion",
        "hidden_size": 256,
        "config": {
            "diffusion_conditioning": "token_cross_attention",
            "latent_size": 64,
            "diffusion_steps": 64,
            "inference_steps": 8,
            "abstain_threshold": 0.55,
            "attention_heads": 4,
        },
    }

    reconstructed = _model_config_from_checkpoint(checkpoint)
    verifier_head = build_model("diffusion", option_count=32, config=reconstructed)
    verifier_head.load_state_dict(trained_head.state_dict())

    assert reconstructed == expected


def test_token_conditioned_checkpoint_uses_training_defaults() -> None:
    checkpoint = {
        "backend": "diffusion",
        "hidden_size": 256,
        "config": {"diffusion_conditioning": "token_cross_attention"},
    }

    reconstructed = _model_config_from_checkpoint(checkpoint)

    assert reconstructed.latent_size == 64
    assert reconstructed.diffusion_steps == 64
    assert reconstructed.inference_steps == 8
    assert reconstructed.diffusion_conditioning == "token_cross_attention"


def test_pooled_checkpoint_without_schedule_fields_uses_export_defaults() -> None:
    expected = ModelConfig(
        hidden_size=256,
        latent_size=32,
        diffusion_steps=32,
        inference_steps=4,
        diffusion_conditioning="pooled",
    )
    trained_head = build_model("diffusion", option_count=32, config=expected)
    checkpoint = {
        "backend": "diffusion",
        "hidden_size": 256,
        "config": {},
    }

    reconstructed = _model_config_from_checkpoint(checkpoint)
    verifier_head = build_model("diffusion", option_count=32, config=reconstructed)
    verifier_head.load_state_dict(trained_head.state_dict())

    assert reconstructed == expected