File size: 3,874 Bytes
a5b21df
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
90
91
92
93
94
95
96
97
98
99
100
101
102
103
# Copyright 2026 The vLLM Semantic Router Authors.
# SPDX-License-Identifier: Apache-2.0
"""Decision 1.0 configuration for 🤗 Transformers (``trust_remote_code=True``).

The repository's root ``config.json`` is its Decision file map
(``decision_format: vllm-sr-decision``): model name, runtime family and the
paths of the backbone, tokenizer, decision weights and calibration, plus
``model_type``, ``architectures``, ``auto_map`` and ``custom_pipelines``.
"""

from __future__ import annotations

from pathlib import PurePosixPath
from typing import Any

try:
    from transformers import PreTrainedConfig
except ImportError:  # Transformers 4
    from transformers import PretrainedConfig as PreTrainedConfig

FAMILIES = ("vela-encoder", "qwen3.5-decision")
DESCRIPTOR_KEYS = (
    "decision_format",
    "format_version",
    "model_name",
    "runtime_family",
    "model_config",
    "backbone",
    "tokenizer",
    "decision_weights",
    "calibration",
)


def _relative(path: Any) -> str:
    if not isinstance(path, str) or not path or "\\" in path:
        raise ValueError("config.json names a file with an invalid path")
    parts = PurePosixPath(path)
    if parts.is_absolute() or ".." in parts.parts or "." in parts.parts:
        raise ValueError("config.json names a file outside the repository")
    return path


class Decision1Config(PreTrainedConfig):
    model_type = "decision1"

    def __init__(
        self,
        decision_format: str = "vllm-sr-decision",
        format_version: int = 1,
        model_name: str | None = None,
        runtime_family: str | None = None,
        model_config: str | None = None,
        backbone: dict[str, Any] | None = None,
        tokenizer: dict[str, Any] | None = None,
        decision_weights: dict[str, str] | None = None,
        calibration: dict[str, Any] | None = None,
        **kwargs: Any,
    ):
        self.decision_format = decision_format
        self.format_version = format_version
        self.model_name = model_name
        self.runtime_family = runtime_family
        self.model_config = model_config
        self.backbone = backbone
        self.tokenizer = tokenizer
        self.decision_weights = decision_weights
        if calibration is not None:
            self.calibration = calibration
        super().__init__(**kwargs)

    def descriptor(self) -> dict[str, Any]:
        """The Decision file map of ``config.json``."""
        if self.decision_format != "vllm-sr-decision" or self.format_version != 1:
            raise ValueError("config.json is not a Decision 1.0 file map")
        if self.runtime_family not in FAMILIES:
            raise ValueError(
                f"Unsupported Decision runtime family: {self.runtime_family!r}"
            )
        descriptor = {
            key: getattr(self, key)
            for key in DESCRIPTOR_KEYS
            if getattr(self, key, None) is not None
        }
        for key in ("model_config", "backbone", "tokenizer", "decision_weights"):
            if key not in descriptor:
                raise ValueError(f"config.json does not name {key}")
        return descriptor

    def files(self) -> list[str]:
        """Every repository file that inference reads."""
        descriptor = self.descriptor()
        backbone, tokenizer = descriptor["backbone"], descriptor["tokenizer"]
        names = [descriptor["model_config"], backbone["config"], *backbone["weights"]]
        if backbone.get("index"):
            names.append(backbone["index"])
        names += [value for value in tokenizer.values() if isinstance(value, str)]
        names += list(descriptor["decision_weights"].values())
        calibration = descriptor.get("calibration") or {}
        if calibration.get("temperature_file"):
            names.append(calibration["temperature_file"])
        return sorted({_relative(name) for name in names})