decision-studio / deploy /runtime /fetch_pins.py
Xunzhuo's picture HuaminChen's picture
Decision 2.0 Studio: open foundation decision models, archived (replays recorded results; see Decision 3.0)
53d0a86
Raw History Blame Contribute Delete
1.86 kB
"""Download and verify pinned Decision 2.0 packages into a Hugging Face cache.
Runs inside the runtime image with network access and the host's Hugging Face
token file mounted read-only; the serving containers then start offline from
the same cache. Verification is the runtime's own: every file listed in the
package manifest (and an adapter package's pinned base) must match its digest,
and the model identity must match the manifest. No weights are loaded.
python3 fetch_pins.py --cache-dir /hf-cache vllm-sr/Decision-2.0-Kai-0.6B@<40-hex> ...
"""
from __future__ import annotations
import argparse
import re
import sys
from vllm_sr_runtime.families.decision2.family import Decision2Family
from vllm_sr_runtime.plugins.base import RegistryOptions
from vllm_sr_runtime.registry.resolve import resolve
PIN = re.compile(r"(?P<repo>[A-Za-z0-9][\w.-]*/[\w.-]+)@(?P<revision>[0-9a-f]{40})\Z")
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
parser.add_argument("pins", nargs="+", help="repository@40-hex revision")
parser.add_argument("--cache-dir", required=True)
args = parser.parse_args(argv)
family = Decision2Family(RegistryOptions(cache_dir=args.cache_dir))
for pin in args.pins:
match = PIN.fullmatch(pin)
if match is None:
parser.error(f"not a repository@40-hex pin: {pin}")
ref = resolve(match["repo"], revision=match["revision"], cache_dir=args.cache_dir)
package = family.verify(ref)
print(
f"{match['repo']}@{match['revision']} "
f"manifest_sha256={package.manifest_sha256} model_sha256={package.model_sha256} "
f"max_input_tokens={package.max_input_tokens}",
flush=True,
)
return 0
if __name__ == "__main__":
sys.exit(main())