Buckets:
| """First generated-state interventions for RGB-visible OPD audits.""" | |
| from __future__ import annotations | |
| import hashlib | |
| import torch | |
| from .contracts import SampledRolloutV2 | |
| def first_generated_visual( | |
| rollout: SampledRolloutV2, | |
| ) -> tuple[torch.Tensor, tuple[int, int]]: | |
| for index, (sequence, codes) in enumerate( | |
| zip( | |
| rollout.visual_sequence_spans, | |
| rollout.visual_code_spans, | |
| strict=True, | |
| ) | |
| ): | |
| if sequence.start < rollout.prefix_length: | |
| continue | |
| grid = rollout.image_grid_thw[index] | |
| if int(grid[0]) != 1 or int(grid[1]) % 2 or int(grid[2]) % 2: | |
| raise ValueError("generated visual grid is not a merged 2-D image") | |
| return ( | |
| rollout.sampled_visual_codes[codes.start : codes.end].detach().clone(), | |
| (int(grid[1]) // 2, int(grid[2]) // 2), | |
| ) | |
| raise ValueError("rollout contains no generated visual state") | |
| def visual_control_codes( | |
| codes: torch.Tensor, *, sample_id: str, flip_probability: float = 0.02 | |
| ) -> dict[str, torch.Tensor]: | |
| if not 0.0 <= flip_probability <= 1.0: | |
| raise ValueError("control flip probability must be in [0, 1]") | |
| seed = int.from_bytes(hashlib.sha256(sample_id.encode()).digest()[:8], "big") | |
| generator = torch.Generator(device=codes.device).manual_seed( | |
| seed & ((1 << 63) - 1) | |
| ) | |
| permutation = torch.randperm( | |
| codes.shape[0], generator=generator, device=codes.device | |
| ) | |
| flips = ( | |
| torch.rand(codes.shape, generator=generator, device=codes.device) | |
| < flip_probability | |
| ) | |
| return { | |
| "null": torch.zeros_like(codes), | |
| "shuffle": codes[permutation].clone(), | |
| "noise": codes ^ flips.to(codes.dtype), | |
| } | |
| class FirstVisualOverrideBackend: | |
| """Delegate generation but replace exactly the first hard visual sample.""" | |
| def __init__(self, backend, codes: torch.Tensor, grid: tuple[int, int]): | |
| self.backend = backend | |
| self.codes = codes.detach().clone() | |
| self.grid = grid | |
| self.used = False | |
| def prepare(self, sample_index, item): | |
| return self.backend.prepare(sample_index, item) | |
| def set_sampling_seed(self, seed): | |
| setter = getattr(self.backend, "set_sampling_seed", None) | |
| if setter is not None: | |
| setter(seed) | |
| def generate_text(self, state, *, phase): | |
| return self.backend.generate_text(state, phase=phase) | |
| def generate_visual(self, state, *, grid): | |
| if not self.used: | |
| if grid != self.grid or self.codes.shape[0] != grid[0] * grid[1]: | |
| raise ValueError("first-visual intervention grid differs from policy") | |
| self.used = True | |
| return self.codes.to(state.visual_codes) | |
| return self.backend.generate_visual(state, grid=grid) | |
| __all__ = [ | |
| "FirstVisualOverrideBackend", | |
| "first_generated_visual", | |
| "visual_control_codes", | |
| ] | |
Xet Storage Details
- Size:
- 2.96 kB
- Xet hash:
- c6c59afb16525a6fc718e91cdf8e95852d6232edf273aea5aeaca15daaccde36
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.