Class SequenceDiskCache

examples/server/server.py:12302–12613

Directory-backed cache for serialized llama.cpp sequence state.

class SequenceDiskCache(SequenceCache):
    """Directory-backed cache for serialized llama.cpp sequence state."""

    @dataclass
    class Entry:
        entry_id: int
        path: Path
        tokens: Tuple[int, ...]
        size_bytes: int
        has_prompt_logits: bool
        last_accessed: float

    @dataclass(frozen=True)
    class Header:
        tokens: Tuple[int, ...]
        has_prompt_logits: bool

    @dataclass
    class Payload:
        tokens: List[int]
        state_bytes: np.ndarray
        prompt_logits: Optional[np.ndarray]

    FORMAT_VERSION = "1"

    TENSOR_TOKENS = "tokens"
    TENSOR_STATE = "state"
    TENSOR_PROMPT_LOGITS = "prompt_logits"

    METADATA_FORMAT = "sequence_cache_format"
    METADATA_COMPATIBILITY_KEY = "compatibility_key"

    def __init__(
        self,
        *,
        path: Union[str, Path],
        max_bytes: int,
        compatibility_key: str,
        min_tokens: int = 128,
    ) -> None:
        self.path = Path(path)
        self.path.mkdir(parents=True, exist_ok=True)
        self.max_bytes = max(0, int(max_bytes))
        self.min_tokens = max(1, int(min_tokens))
        self.compatibility_key = compatibility_key
        self._metadata = {
            self.METADATA_FORMAT: self.FORMAT_VERSION,
            self.METADATA_COMPATIBILITY_KEY: compatibility_key,
        }
        self._trie = RadixTrie()
        self._entries_by_id: Dict[int, SequenceDiskCache.Entry] = {}
        self._entries_by_tokens: Dict[Tuple[int, ...], SequenceDiskCache.Entry] = {}
        self._next_entry_id = 0
        self._size_bytes = 0
        self._load_entries()
        self._evict_if_needed()

    @staticmethod