Skip to content

Cache

Bases: ABC

used to store AutoLabeling results, allowing for interrupted labeling runs to be continued from a save point without the need to restart from the beginning. Any custom Cache classes should extend from BaseCache.

Source code in src/autolabel/cache/base.py 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
class BaseCache(ABC):
    """used to store AutoLabeling results, allowing for interrupted labeling runs to be continued from a save point without the need to restart from the beginning. Any custom Cache classes should extend from BaseCache."""

    def __init__(self) -> None:
        super().__init__()

    @abstractmethod
    def initialize():
        """initialize the cache. Must be implemented by classes derived from BaseCache."""
        pass

    @abstractmethod
    def lookup(self, entry):
        """abstract method to retrieve a cached entry. Must be implemented by classes derived from BaseCache."""
        pass

    @abstractmethod
    def update(self, entry):
        """abstract method to update the cache with a new entry. Must be implemented by classes derived from BaseCache."""
        pass

    @abstractmethod
    def clear(self) -> None:
        """abstract method to clear the cache. Must be implemented by classes derived from BaseCache."""
        pass

clear() abstractmethod

abstract method to clear the cache. Must be implemented by classes derived from BaseCache.

Source code in src/autolabel/cache/base.py 
27
28
29
30
@abstractmethod
def clear(self) -> None:
    """abstract method to clear the cache. Must be implemented by classes derived from BaseCache."""
    pass

initialize() abstractmethod

initialize the cache. Must be implemented by classes derived from BaseCache.

Source code in src/autolabel/cache/base.py 
12
13
14
15
@abstractmethod
def initialize():
    """initialize the cache. Must be implemented by classes derived from BaseCache."""
    pass

lookup(entry) abstractmethod

abstract method to retrieve a cached entry. Must be implemented by classes derived from BaseCache.

Source code in src/autolabel/cache/base.py 
17
18
19
20
@abstractmethod
def lookup(self, entry):
    """abstract method to retrieve a cached entry. Must be implemented by classes derived from BaseCache."""
    pass

update(entry) abstractmethod

abstract method to update the cache with a new entry. Must be implemented by classes derived from BaseCache.

Source code in src/autolabel/cache/base.py 
22
23
24
25
@abstractmethod
def update(self, entry):
    """abstract method to update the cache with a new entry. Must be implemented by classes derived from BaseCache."""
    pass

rendering: show_root_heading: yes show_root_full_path: no

Bases: BaseCache

A cache system implemented with SQL Alchemy

Source code in src/autolabel/cache/sqlalchemy_generation_cache.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
class SQLAlchemyGenerationCache(BaseCache):
    """A cache system implemented with SQL Alchemy"""

    def __init__(self):
        self.engine = None
        self.base = Base
        self.session = None

    def initialize(self):
        self.engine = create_db_engine()
        self.base.metadata.create_all(self.engine)
        self.session = sessionmaker(bind=self.engine)()

    def lookup(
        self, entry: GenerationCacheEntry
    ) -> List[Union[Generation, ChatGeneration]]:
        """Retrieves an entry from the Cache. Returns an empty list [] if not found.
        Args:
            entry: GenerationCacheEntry we wish to retrieve from the Cache
        Returns:
            result: A list of langchain Generation objects, containing the results of the labeling run for this GenerationCacheEntry. Empty list [] if not found.
        """
        cache_entry = GenerationCacheEntryModel.get(self.session, entry)
        if cache_entry is None:
            logger.debug("Cache miss")
            return []

        logger.debug("Cache hit")
        return cache_entry.generations

    def update(self, entry: GenerationCacheEntry) -> None:
        """Inserts the provided GenerationCacheEntry into the Cache, overriding it if it already exists
        Args:
            entry: GenerationCacheEntry we wish to put into the Cache
        """
        GenerationCacheEntryModel.insert(self.session, entry)

    def clear(self) -> None:
        """Clears the entire Cache"""
        GenerationCacheEntryModel.clear(self.session)

clear()

Clears the entire Cache

Source code in src/autolabel/cache/sqlalchemy_generation_cache.py 
52
53
54
def clear(self) -> None:
    """Clears the entire Cache"""
    GenerationCacheEntryModel.clear(self.session)

lookup(entry)

Retrieves an entry from the Cache. Returns an empty list [] if not found. Args: entry: GenerationCacheEntry we wish to retrieve from the Cache Returns: result: A list of langchain Generation objects, containing the results of the labeling run for this GenerationCacheEntry. Empty list [] if not found.

Source code in src/autolabel/cache/sqlalchemy_generation_cache.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
def lookup(
    self, entry: GenerationCacheEntry
) -> List[Union[Generation, ChatGeneration]]:
    """Retrieves an entry from the Cache. Returns an empty list [] if not found.
    Args:
        entry: GenerationCacheEntry we wish to retrieve from the Cache
    Returns:
        result: A list of langchain Generation objects, containing the results of the labeling run for this GenerationCacheEntry. Empty list [] if not found.
    """
    cache_entry = GenerationCacheEntryModel.get(self.session, entry)
    if cache_entry is None:
        logger.debug("Cache miss")
        return []

    logger.debug("Cache hit")
    return cache_entry.generations

update(entry)

Inserts the provided GenerationCacheEntry into the Cache, overriding it if it already exists Args: entry: GenerationCacheEntry we wish to put into the Cache

Source code in src/autolabel/cache/sqlalchemy_generation_cache.py 
45
46
47
48
49
50
def update(self, entry: GenerationCacheEntry) -> None:
    """Inserts the provided GenerationCacheEntry into the Cache, overriding it if it already exists
    Args:
        entry: GenerationCacheEntry we wish to put into the Cache
    """
    GenerationCacheEntryModel.insert(self.session, entry)

rendering: show_root_heading: yes show_root_full_path: no

Bases: BaseCache

A cache system implemented with SQL Alchemy for storing the output of transforms. This cache system is used to avoid re-computing the output of transforms that have already been computed. This currently stores the input and the outputs of the transform. Caching is based on the transform name, params and input.

Source code in src/autolabel/cache/sqlalchemy_transform_cache.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
class SQLAlchemyTransformCache(BaseCache):
    """
    A cache system implemented with SQL Alchemy for storing the output of transforms.
    This cache system is used to avoid re-computing the output of transforms that have already been computed.
    This currently stores the input and the outputs of the transform.
    Caching is based on the transform name, params and input.
    """

    def __init__(self):
        self.engine = None
        self.base = Base
        self.session = None

    def initialize(self):
        self.engine = create_db_engine()
        self.base.metadata.create_all(self.engine)
        self.session = sessionmaker(bind=self.engine)()

    def lookup(self, entry: TransformCacheEntry) -> Optional[Dict[str, Any]]:
        """Retrieves an entry from the Cache. Returns None if not found.
        Args:
            entry: TransformCacheEntry we wish to retrieve from the Cache
        Returns:
            result: The output of the transform for this input. None if not found.
        """
        cache_entry = TransformCacheEntryModel.get(self.session, entry)
        if cache_entry is None:
            return None

        return cache_entry.output

    def update(self, entry: TransformCacheEntry) -> None:
        """Inserts the provided TransformCacheEntry into the Cache, overriding it if it already exists
        Args:
            entry: TransformCacheEntry we wish to put into the Cache
        """
        TransformCacheEntryModel.insert(self.session, entry)

    def clear(self, use_ttl: bool = True) -> None:
        """Clears the entire Cache based on ttl"""
        TransformCacheEntryModel.clear(self.session, use_ttl=use_ttl)

clear(use_ttl=True)

Clears the entire Cache based on ttl

Source code in src/autolabel/cache/sqlalchemy_transform_cache.py 
52
53
54
def clear(self, use_ttl: bool = True) -> None:
    """Clears the entire Cache based on ttl"""
    TransformCacheEntryModel.clear(self.session, use_ttl=use_ttl)

lookup(entry)

Retrieves an entry from the Cache. Returns None if not found. Args: entry: TransformCacheEntry we wish to retrieve from the Cache Returns: result: The output of the transform for this input. None if not found.

Source code in src/autolabel/cache/sqlalchemy_transform_cache.py 
32
33
34
35
36
37
38
39
40
41
42
43
def lookup(self, entry: TransformCacheEntry) -> Optional[Dict[str, Any]]:
    """Retrieves an entry from the Cache. Returns None if not found.
    Args:
        entry: TransformCacheEntry we wish to retrieve from the Cache
    Returns:
        result: The output of the transform for this input. None if not found.
    """
    cache_entry = TransformCacheEntryModel.get(self.session, entry)
    if cache_entry is None:
        return None

    return cache_entry.output

update(entry)

Inserts the provided TransformCacheEntry into the Cache, overriding it if it already exists Args: entry: TransformCacheEntry we wish to put into the Cache

Source code in src/autolabel/cache/sqlalchemy_transform_cache.py 
45
46
47
48
49
50
def update(self, entry: TransformCacheEntry) -> None:
    """Inserts the provided TransformCacheEntry into the Cache, overriding it if it already exists
    Args:
        entry: TransformCacheEntry we wish to put into the Cache
    """
    TransformCacheEntryModel.insert(self.session, entry)

rendering: show_root_heading: yes show_root_full_path: no