Buckets:
| """Shared autonomous interleaving grammar helpers.""" | |
| from __future__ import annotations | |
| import re | |
| import torch | |
| _IMAGE_SCAFFOLD = re.compile( | |
| r"^<image_gen> generate image ([1-9]\d*) ([1-9]\d*)<\|vision_start\|>$" | |
| ) | |
| def parse_image_scaffold(text: str) -> tuple[int, int] | None: | |
| """Parse a complete student-delimited image scaffold without exceptions.""" | |
| match = _IMAGE_SCAFFOLD.fullmatch(text) | |
| if match is None: | |
| return None | |
| try: | |
| height, width = match.groups() | |
| return int(height), int(width) | |
| except ValueError: | |
| # Python rejects pathologically long decimal strings. Such model | |
| # output is an invalid scaffold, not a rollout-process failure. | |
| return None | |
| def mrope_safe_target_carrier_ids(input_ids: torch.Tensor) -> torch.Tensor: | |
| """Clone a target carrier and close a terminal vision_start for M-RoPE.""" | |
| if input_ids.ndim != 1 or not input_ids.numel(): | |
| raise ValueError("target carrier input IDs must be one non-empty sequence") | |
| from uniar import SPECIAL_TOKEN_IDS | |
| result = input_ids.clone() | |
| if int(result[-1]) == SPECIAL_TOKEN_IDS["<|vision_start|>"]: | |
| # Its own next-token logit comes from the preceding state, so changing | |
| # only this terminal carrier token is causally inert. | |
| result[-1] = SPECIAL_TOKEN_IDS["<|im_end|>"] | |
| return result | |
| __all__ = ["mrope_safe_target_carrier_ids", "parse_image_scaffold"] | |
Xet Storage Details
- Size:
- 1.45 kB
- Xet hash:
- 43d7c3a9575d786e08ecc24fa9b16c39a55a1ac8b391e9e937a592e58e538f08
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.