Utils

Utils#

This module contains useful functions that are used throughout the codebase.

Cache#

class flexrag.utils.persistent_cache.FIFOPersistentCache(maxsize=None, cache_path=None)[源代码]#
popitem()[源代码]#

This method should be implemented by subclasses.

class flexrag.utils.persistent_cache.LFUPersistentCache(maxsize=None, cache_path=None)[源代码]#

The LFUPersistentCache evicts the least frequently used item from the cache when the cache is full.

This implementation employs a Counter to keep track of the frequency of access. However, the frequency will not be persisted to disk. Thus, the frequency will be reset when the cache is loaded from disk.

popitem()[源代码]#

This method should be implemented by subclasses.

class flexrag.utils.persistent_cache.LRUPersistentCache(maxsize=None, cache_path=None)[源代码]#

The LRUPersistentCache evicts the least recently used item from the cache when the cache is full.

This implementation employs an OrderedDict to keep track of the order of access. However, the order will not be persisted to disk. Thus, the order will be reset when the cache is loaded from disk.

popitem()[源代码]#

This method should be implemented by subclasses.

class flexrag.utils.persistent_cache.PersistentCacheBase(maxsize=None, cache_path=None)[源代码]#

The base class for PersistentCache.

The PersistentCache is a cache that can be persisted to disk, and provide a simple interface like a dictionary. The subclasses should implement the popitem method, which decides which item to evict from the cache when the cache is full.

cache(func)[源代码]#

Decorator to cache the result of a function. The arguments of the function should be hashable.

For example:

from flexrag.utils import LRUPersistentCache

cache = LRUPersistentCache()

@cache.cache
def expensive_function(x):
    # Some expensive computation
    return x * 2
abstract popitem()[源代码]#

This method should be implemented by subclasses.

reduce_size(size=None)[源代码]#

Reduce the size of the cache to the specified size.

param size: The size to reduce to. If None, use the self.maxsize. type size: int return: None rtype: None

class flexrag.utils.persistent_cache.RandomPersistentCache(maxsize=None, cache_path=None)[源代码]#

The RandomPersistentCache evicts a random item from the cache when the cache is full.

In this implementation, the evict order is determined by the __iter__ method of the backend.

popitem()[源代码]#

This method should be implemented by subclasses.

Other Utils#

class flexrag.utils.Register(register_name=None, allow_load_from_repo=False)[源代码]#
get(key, default=None)[源代码]#

Get the item dict by name.

参数:

  • key (str) -- The name of the item.

  • default (Any) -- The default value to return, defaults to None.

返回:

The item dict containing the item, main_name, short_names, and config_class.

返回类型:

dict

get_item(key)[源代码]#

Get the item by name.

参数:

key (str) -- The name of the item.

返回:

The item.

返回类型:

Any

load(config, **kwargs)[源代码]#

Load the item(s) from the generated config.

参数:

  • config (DictConfig) -- The config generated by make_config method.

  • kwargs (Any) -- The additional arguments to pass to the item(s).

抛出:

ValueError -- If the item type is invalid.

返回:

The loaded item(s).

返回类型:

RegistedType | list[RegistedType]

property mainnames#

Get the main names of the registered items.

make_config(allow_multiple=False, default=None, config_name=None)[源代码]#

Make a config class for the registered items.

参数:

  • allow_multiple (bool, optional) -- Whether to allow multiple items to be selected, defaults to False.

  • default (Optional[str], optional) -- The default item to select, defaults to None.

  • config_name (str, optional) -- The name of the config class, defaults to None.

返回:

The config class.

返回类型:

dataclass

property names#

Get the names of the registered items.

property shortnames#

Get the short names of the registered items.

squeeze(config_instance)[源代码]#

Convert the nused fields to None.

Common Dataclass#

This module provides several pre-defined dataclasses that are commonly used in the project.

class flexrag.utils.dataclasses.Context(context_id=None, data=<factory>, source=None, meta_data=<factory>)[源代码]#

The dataclass for retrieved context.

参数:

  • context_id (Optional[str]) -- The unique identifier of the context. Default: None.

  • data (dict) -- The context data. Default: {}.

  • source (Optional[str]) -- The source of the retrieved data. Default: None.

  • meta_data (dict) -- The metadata of the context. Default: {}.

dump(path)#

Dump the dataclass to a YAML file.

dumps()#

Dump the dataclass to a YAML string.

classmethod load(path)#

Load the dataclass from a YAML file.

classmethod loads(s)#

Load the dataclass from a YAML string.

class flexrag.utils.dataclasses.RetrievedContext(context_id=None, data=<factory>, source=None, meta_data=<factory>, retriever='', query='', score=0.0)[源代码]#

The dataclass for retrieved context.

参数:

  • retriever (str) -- The name of the retriever. Required.

  • query (str) -- The query for retrieval. Required.

  • score (float) -- The relevance score of the retrieved data. Default: 0.0.

dump(path)#

Dump the dataclass to a YAML file.

dumps()#

Dump the dataclass to a YAML string.

classmethod load(path)#

Load the dataclass from a YAML file.

classmethod loads(s)#

Load the dataclass from a YAML string.