nemoguardrails.llm.cache.lfu | NVIDIA NeMo Guardrails Library Developer Guide

Least Frequently Used (LFU) cache implementation.

Module Contents

Classes

Name	Description
`DoublyLinkedList`	Doubly linked list to maintain nodes with the same frequency.
`LFUCache`	Least Frequently Used (LFU) Cache implementation.
`LFUNode`	Node for the LFU cache doubly linked list.

Data

log

API

class nemoguardrails.llm.cache.lfu.DoublyLinkedList()

Doubly linked list to maintain nodes with the same frequency.

head

= LFUNode(None, None)

size

= 0

tail

= LFUNode(None, None)

nemoguardrails.llm.cache.lfu.DoublyLinkedList.append(
    node: nemoguardrails.llm.cache.lfu.LFUNode
) -> None

Add node to the end of the list (before tail).

nemoguardrails.llm.cache.lfu.DoublyLinkedList.pop(
    node: typing.Optional[nemoguardrails.llm.cache.lfu.LFUNode] = None
) -> typing.Optional[nemoguardrails.llm.cache.lfu.LFUNode]

Remove and return a node. If no node specified, removes the first node.

class nemoguardrails.llm.cache.lfu.LFUCache(
    maxsize: int,
    track_stats: bool = False,
    stats_logging_interval: typing.Optional[float] = None
)

Bases: CacheInterface

Least Frequently Used (LFU) Cache implementation.

When the cache reaches maxsize, it evicts the least frequently used item. If there are ties in frequency, it evicts the least recently used among them.

_computing

dict[Any, Future] = {}

_lock

= threading.RLock()

freq_map

dict[int, DoublyLinkedList] = {}

key_map

dict[Any, LFUNode] = {}

maxsize

int

Get the maximum size of the cache.

min_freq

= 0

stats

nemoguardrails.llm.cache.lfu.LFUCache._check_and_log_stats() -> None

Check if enough time has passed and log stats if needed.

nemoguardrails.llm.cache.lfu.LFUCache._evict_lfu() -> None

Evict the least frequently used item from the cache.

nemoguardrails.llm.cache.lfu.LFUCache._log_stats() -> None

Log current cache statistics.

nemoguardrails.llm.cache.lfu.LFUCache._update_node_freq(
    node: nemoguardrails.llm.cache.lfu.LFUNode
) -> None

Update the frequency of a node and move it to the appropriate frequency list.

nemoguardrails.llm.cache.lfu.LFUCache.clear() -> None

Clear all items from the cache.

nemoguardrails.llm.cache.lfu.LFUCache.contains(
    key: typing.Any
) -> bool

Check if a key exists in the cache without updating its frequency.

This is more efficient than the default implementation which uses get() and has the side effect of updating frequency counts.

Parameters:

key

Any

The key to check

Returns: bool

True if the key exists in the cache, False otherwise

nemoguardrails.llm.cache.lfu.LFUCache.get(
    key: typing.Any,
    default: typing.Any = None
) -> typing.Any

Get an item from the cache.

Parameters:

key

Any

The key to look up

default

AnyDefaults to None

Value to return if key is not found

Returns: Any

The value associated with the key, or default if not found

nemoguardrails.llm.cache.lfu.LFUCache.get_or_compute(
    key: typing.Any,
    compute_fn: typing.Callable[[], typing.Any],
    default: typing.Any = None
) -> typing.Any

async

Atomically get a value from the cache or compute it if not present.

This method ensures that the compute function is called at most once even in the presence of concurrent requests for the same key.

Parameters:

key

Any

The key to look up

compute_fn

Callable[[], Any]

Async function to compute the value if key is not found

default

AnyDefaults to None

Value to return if compute_fn raises an exception

Returns: Any

The cached value or the computed value

nemoguardrails.llm.cache.lfu.LFUCache.get_stats() -> dict

Get cache statistics.

Returns: dict

Dictionary with cache statistics (if tracking is enabled)

nemoguardrails.llm.cache.lfu.LFUCache.is_empty() -> bool

Check if the cache is empty.

nemoguardrails.llm.cache.lfu.LFUCache.log_stats_now() -> None

Force immediate logging of cache statistics.

nemoguardrails.llm.cache.lfu.LFUCache.put(
    key: typing.Any,
    value: typing.Any
) -> None

Put an item into the cache.

Parameters:

key

Any

The key to store

value

Any

The value to associate with the key

nemoguardrails.llm.cache.lfu.LFUCache.reset_stats() -> None

Reset cache statistics.

nemoguardrails.llm.cache.lfu.LFUCache.size() -> int

Return the current size of the cache.

nemoguardrails.llm.cache.lfu.LFUCache.supports_stats_logging() -> bool

Check if this cache instance supports stats logging.

class nemoguardrails.llm.cache.lfu.LFUNode(
    key: typing.Any,
    value: typing.Any
)

Node for the LFU cache doubly linked list.

accessed_at

= self.created_at

created_at

= time.time()

freq

= 1

Optional[LFUNode] = None

nemoguardrails.llm.cache.lfu.log = logging.getLogger(__name__)