(
self,
tokens: Sequence[int],
state_bytes: np.ndarray,
prompt_logits: Optional[np.ndarray],
)
| 12585 | ) |
| 12586 | |
| 12587 | def save( |
| 12588 | self, |
| 12589 | tokens: Sequence[int], |
| 12590 | state_bytes: np.ndarray, |
| 12591 | prompt_logits: Optional[np.ndarray], |
| 12592 | ) -> None: |
| 12593 | if len(tokens) < self.min_tokens or self.max_bytes <= 0: |
| 12594 | return |
| 12595 | state = np.asarray(state_bytes, dtype=np.uint8) |
| 12596 | if state.size <= 0: |
| 12597 | return |
| 12598 | state = np.ascontiguousarray(state).copy() |
| 12599 | entry_tokens = tuple(int(token) for token in tokens) |
| 12600 | tensors: Dict[str, np.ndarray] = { |
| 12601 | self.TENSOR_TOKENS: np.asarray(entry_tokens, dtype=np.int32), |
| 12602 | self.TENSOR_STATE: state, |
| 12603 | } |
| 12604 | if prompt_logits is not None: |
| 12605 | tensors[self.TENSOR_PROMPT_LOGITS] = np.asarray(prompt_logits, dtype=np.float32) |
| 12606 | path = self.path / f"{uuid.uuid4().hex}.safetensors" |
| 12607 | self._write_entry(path, tensors) |
| 12608 | self._add_entry( |
| 12609 | path=path, |
| 12610 | tokens=entry_tokens, |
| 12611 | has_prompt_logits=prompt_logits is not None, |
| 12612 | ) |
| 12613 | self._evict_if_needed() |
| 12614 | |
| 12615 | |
| 12616 | class MemoryPolicy(abc.ABC): |
nothing calls this directly
no test coverage detected