ahvn.cache package

Cache backends for AgentHeaven.

Includes in-memory, on-disk, and JSON-file caches, plus a no-op cache.

class ahvn.cache.CacheEntry(func, inputs=<factory>, output=Ellipsis, expected=Ellipsis, metadata=<factory>, _key=None)

Bases: object

A universal object to store cache entries, containing the function (name), inputs, output, and optional metadata.

Parameters:

• func (str)

• inputs (Dict[str, Any])

• output (Any)

• expected (Any)

• metadata (Dict[str, Any] | None)

• _key (int)

func: str

inputs: Dict[str, Any]

output: Any = Ellipsis

expected: Any = Ellipsis

metadata: Dict[str, Any] | None

property key

property value

classmethod from_args(func='', output=..., expected=..., metadata=None, exclude=None, **inputs)

Creates a CacheEntry from function arguments.

Parameters:

• func (Union[Callable, str]) – The function or its name to be cached.

• output (Any) – The output of the function.

• expected (Any) – The expected output of the function.

• metadata (Optional[Dict[str, Any]]) – Optional metadata to store with the cache entry

• exclude (Optional[Iterable[str]]) – Keys to exclude from inputs.

• **inputs (Any) – Arbitrary keyword arguments representing the inputs to the function.

Return type:

CacheEntry

classmethod from_dict(data, exclude=None)

Creates a CacheEntry from a dictionary.

Parameters:

• data (Dict[str, Any]) – The dictionary containing the cache entry data.

• exclude (Optional[Iterable[str]]) – Keys to exclude from inputs.

Return type:

CacheEntry

to_dict()

Converts the CacheEntry to a dictionary.

Returns:

A dictionary representation of the CacheEntry.

Return type:

Dict[str, Any]

clone(**updates)

Creates a clone of the CacheEntry with optional updates to its attributes.

Parameters:

**updates – Arbitrary keyword arguments to update the CacheEntry attributes.

Returns:

A new CacheEntry instance with updated attributes.

Return type:

CacheEntry

annotate(expected=..., metadata=None)

Annotates the CacheEntry with expected output and metadata.

Parameters:

• expected (Any) – The expected output of the function. If omitted, will use the actual output as annotation.

• metadata (Optional[Dict[str, Any]]) – Optional metadata to store with the cache entry.

Returns:

A new CacheEntry instance with the annotation.

Return type:

CacheEntry

property annotated: bool

Checks if the CacheEntry has been annotated with expected output.

Returns:

True if the CacheEntry has an expected output, False otherwise.

Return type:

bool

__init__(func, inputs=<factory>, output=Ellipsis, expected=Ellipsis, metadata=<factory>, _key=None)

Parameters:

• func (str)

• inputs (Dict[str, Any])

• output (Any)

• expected (Any)

• metadata (Dict[str, Any] | None)

• _key (int)

Return type:

None

class ahvn.cache.BaseCache(exclude=None, *args, **kwargs)

Bases: ABC

An abstract base class for cache implementations.

The class provides memoize and batch_memoize decorators to cache function results. The class requires subclasses to implement storage and retrieval methods. The caching supports both synchronous and asynchronous functions.

Parameters:

exclude (Iterable[str] | None)

_cache

A dictionary to store cache entries.

Type:

Dict[str, CacheEntry]

Abstract Methods:

_get(key, default): Retrieves a dict (CacheEntry) from the cache by its key. Use Ellipsis as default to raise KeyError if not found (to avoid collisions with functions returning None). _set(key, value): Sets a cache entry in the cache. _remove(key): Removes a cache entry from the cache by its key. _itervalues(): Returns an iterator over the values in the cache. _clear(): Clears the cache.

Optional Override Methods:

_has(key): Determines if a cache entry exists for the given key. __len__(): Returns the number of entries in the cache.

Notes

• The default implementations of __len__, _has, and batch operations are not optimized and may be slow for large stores. Subclasses are encouraged to override these for efficiency.

• It is worth noticing that only the __getitem__ (implemented by _get) and add (implemented by _set) are used during memoize.

__init__(exclude=None, *args, **kwargs)

Initialization.

Parameters:

• exclude (Optional[Iterable[str]]) – Keys to exclude from inputs when creating cache entries.

• *args – Additional positional arguments.

• **kwargs – Additional keyword arguments.

set_exclude(exclude=None)

Set the keys to exclude from inputs when creating cache entries.

Parameters:

exclude (Optional[Iterable[str]]) – Keys to exclude from inputs when creating cache entries.

add_exclude(exclude=None)

Add keys to exclude from inputs when creating cache entries.

Parameters:

exclude (Optional[Iterable[str]]) – Keys to add to the exclusion list.

__contains__(key)

Checks if a cache entry exists for the given key.

Parameters:

key (Union[int, CacheEntry]) – The key or CacheEntry to check in the cache.

Returns:

True if the cache entry exists, False otherwise.

Return type:

bool

exists(func, **kwargs)

Checks if a cache entry exists for the given function and inputs.

Parameters:

• func (Union[Callable, str]) – The function or its name to check in the cache.

• **kwargs – Arbitrary keyword arguments representing the inputs to the function.

Returns:

True if the cache entry exists, False otherwise.

Return type:

bool

__getitem__(key)

Retrieves a cache entry for the given function and inputs.

Parameters:

key (Union[int, str, CacheEntry]) – The key or CacheEntry to retrieve.

Returns:

The cached entry if found. Otherwise Ellipsis.

Return type:

CacheEntry

retrieve(func, **kwargs)

Retrieves a cached entry for the given function and inputs.

Parameters:

• func (Union[Callable, str]) – The function or its name to retrieve the cached value for.

• **kwargs – Arbitrary keyword arguments representing the inputs to the function.

Returns:

The cached output if found, otherwise Ellipsis (to avoid collisions with functions returning None).

Return type:

Any

get(func, **kwargs)

Gets a cached value for the given function and inputs.

Parameters:

• func (Union[Callable, str]) – The function or its name to retrieve the cached value for.

• **kwargs – Arbitrary keyword arguments representing the inputs to the function.

Returns:

The cached output if found, otherwise Ellipsis (to avoid collisions with functions returning None).

Return type:

Any

__setitem__(key, value)

Sets a cache entry for the given function and inputs.

Parameters:

• key (Union[int, str, CacheEntry]) – The key or CacheEntry to set.

• value (Union[Dict[str, Any], CacheEntry]) – The value to cache.

set(func, output=..., expected=..., metadata=None, **kwargs)

Sets a cached value for the given function and inputs.

Parameters:

• func (Union[Callable, str]) – The function or its name to cache the value for.

• output (Any) – The output to cache.

• expected (Any) – The expected output of the function.

• metadata (Optional[Dict[str, Any]]) – Optional metadata to store with the cache entry.

• **kwargs – Arbitrary keyword arguments representing the inputs to the function.

add(entry)

Sets a cache entry by directly adding it to the cache.

Parameters:

entry (CacheEntry) – The cache entry to add.

annotate(func, expected=..., metadata=None, **kwargs)

Annotates a cached value for the given function and inputs.

Parameters:

• func (Union[Callable, str]) – The function or its name to annotate the cached value for.

• expected (Any) – The expected output of the function. If the cache entry exists, its expected will be modified, but its output will remain unchanged. Otherwise, a new entry with output and expected both set as the annotation will be created. If set to … and the cache entry exists, it will use the actual output as the annotation. If set to … but the cache entry does not exist, raise an error.

• metadata (Optional[Dict[str, Any]]) – Optional metadata to store with the cache entry.

• **kwargs – Arbitrary keyword arguments representing the inputs to the function.

Raises:

ValueError – If the original cache entry does not exist and expected is not provided.

__delitem__(key)

Deletes a cache entry for the given function and inputs.

Parameters:

key (Union[int, str, CacheEntry]) – The key or CacheEntry of the cache entry to delete.

unset(func, **kwargs)

Deletes a cache entry for the given function and inputs.

Parameters:

• func (Union[Callable, str]) – The function or its name to delete the cache entry for.

• **kwargs – Arbitrary keyword arguments representing the inputs to the function.

remove(entry)

Deletes a cache entry by directly removing it from the cache.

Parameters:

entry (CacheEntry) – The cache entry to remove.

__iter__()

Iterates over the cache entries.

Yields:

CacheEntry – The CacheEntry objects in the cache.

Return type:

Generator[CacheEntry, None, None]

pop()

Pops an arbitrary cache entry from the cache.

Returns:

The popped CacheEntry if the cache is not empty, otherwise None.

Return type:

Optional[CacheEntry]

popall()

Pops all cache entries from the cache.

Returns:

A list of all popped CacheEntry objects.

Return type:

List[CacheEntry]

clear()

Clears the cache.

close()

Closes the cache, if applicable.

flush(**kwargs)

Flushes the cache, if applicable.

memoize(func=None, *, name=None)

Decorator (or decorator factory) to cache the output of a function based on its inputs.

Return type:

Callable

Parameters:

• func (Callable | None)

• name (str | None)

Usage:

@cache.memoize def f(…): …

@cache.memoize(name=”xxx”) def g(…): …

When name is provided, it is used as the function identifier in the cache key. Otherwise, the wrapped function’s name is used.

batch_memoize(func=None, *, name=None)

Decorator (or decorator factory) to cache the output of a function based on its inputs in batch mode.

Usage:

@cache.batch_memoize def f(…): …

@cache.batch_memoize(name=”xxx”) def g(…): …

Parameters:

• func (Callable) – The function to cache.

• name (Optional[str]) – Optional name for the function in the cache key.

Returns:

The wrapped function with batch caching.

Return type:

Callable

class ahvn.cache.NoCache(exclude=None, *args, **kwargs)

Bases: BaseCache

An implementation of BaseCache that does not cache any data.

Parameters:

exclude (Iterable[str] | None)

__init__(exclude=None, *args, **kwargs)

Initialization.

Parameters:

• exclude (Optional[Iterable[str]]) – Keys to exclude from inputs when creating cache entries.

• *args – Additional positional arguments.

• **kwargs – Additional keyword arguments.

class ahvn.cache.DiskCache(path, size_limit=int(32e9), exclude=None, *args, **kwargs)

Bases: BaseCache

An implementation of BaseCache that stores data on disk using diskcache.

Parameters:

• path (str)

• size_limit (int)

• exclude (Iterable[str] | None)

__init__(path, size_limit=int(32e9), exclude=None, *args, **kwargs)

Initialization.

Parameters:

• path (str) – Path to the directory where cache files will be stored.

• size_limit (int) – Maximum size of the cache in bytes. Defaults to 32e9 (32 GB).

• exclude (Optional[Iterable[str]]) – Keys to exclude from inputs when creating cache entries.

• *args – Additional positional arguments.

• **kwargs – Additional keyword arguments.

close()

Closes the cache.

Return type:

None

class ahvn.cache.JsonCache(path, exclude=None, *args, **kwargs)

Bases: BaseCache

An implementation of BaseCache that stores data in JSON files in a specified directory. Each item key:value is stored in a separate JSON file named after the key, with values serialized as JSON.

Parameters:

• path (str)

• exclude (Iterable[str] | None)

__init__(path, exclude=None, *args, **kwargs)

Initialization.

Parameters:

• path (str) – Path to the directory where JSON files will be stored.

• exclude (Optional[Iterable[str]]) – Keys to exclude from inputs when creating cache entries.

• *args – Additional positional arguments.

• **kwargs – Additional keyword arguments.

class ahvn.cache.InMemCache(exclude=None, *args, **kwargs)

Bases: BaseCache

An implementation of BaseCache that stores data in memory as Python dictionaries.

Parameters:

exclude (Iterable[str] | None)

__init__(exclude=None, *args, **kwargs)

Initialization.

Parameters:

• exclude (Optional[Iterable[str]]) – Keys to exclude from inputs when creating cache entries.

• *args – Additional positional arguments.

• **kwargs – Additional keyword arguments.

close()

Closes the cache, if applicable.

class ahvn.cache.CallbackCache(callbacks=None, feeds=None, exclude=None, *args, **kwargs)

Bases: BaseCache

An implementation of BaseCache that does not cache any data, but calls callbacks on set, and feeds on get.

Parameters:

• callbacks (Iterable[Callable[[int, Dict[str, Any]], None]] | None)

• feeds (Iterable[Callable[[Callable, Any], None]] | None)

• exclude (Iterable[str] | None)

__init__(callbacks=None, feeds=None, exclude=None, *args, **kwargs)

Initialization.

Parameters:

• callbacks (Optional[Iterable[Callable[[int, Dict[str, Any]], None]]]) – List of callback functions to call on set. Each callback function must has API callback(key: int, value: Dict[str, Any]), which handles a cache set event.

• feeds (Optional[Iterable[Callable[[Union[Callable, str], Any], None]]]) – List of feed functions to call on get. Each feed function must have API feed(func: Union[Callable, str], **kwargs), which handles a cache get event. The kwargs are the input to the function. Notice that feeds must be ordered: the first feed function with a non-Ellipsis return value will be used.

• exclude (Optional[Iterable[str]]) – Keys to exclude from inputs when creating cache entries.

• *args – Additional positional arguments.

• **kwargs – Additional keyword arguments.

get(func, **kwargs)

Retrieves a cached value for the given function and inputs.

Parameters:

• func (Union[Callable, str]) – The function or its name to retrieve the cached value for. Notice that for CallbackCache, when all feed functions return …, the function will be called: # (deprecated) If the func is callable, it will be called with the provided keyword arguments. # (deprecated) Otherwise, it will NOT be called. For better stability, it is recommend to use a default feed function that can handle missing values.

• **kwargs – Arbitrary keyword arguments representing the inputs to the function.

Returns:

The cached output if found, otherwise Ellipsis (to avoid collisions with functions returning None).

Return type:

Any

Submodules

• ahvn.cache.base module
  • CacheEntry
    • CacheEntry.func
    • CacheEntry.inputs
    • CacheEntry.output
    • CacheEntry.expected
    • CacheEntry.metadata
    • CacheEntry.key
    • CacheEntry.value
    • CacheEntry.from_args()
    • CacheEntry.from_dict()
    • CacheEntry.to_dict()
    • CacheEntry.clone()
    • CacheEntry.annotate()
    • CacheEntry.annotated
    • CacheEntry.__init__()
  • BaseCache
    • BaseCache._cache
    • BaseCache.__init__()
    • BaseCache.set_exclude()
    • BaseCache.add_exclude()
    • BaseCache.__contains__()
    • BaseCache.exists()
    • BaseCache.__getitem__()
    • BaseCache.retrieve()
    • BaseCache.get()
    • BaseCache.__setitem__()
    • BaseCache.set()
    • BaseCache.add()
    • BaseCache.annotate()
    • BaseCache.__delitem__()
    • BaseCache.unset()
    • BaseCache.remove()
    • BaseCache.__iter__()
    • BaseCache.pop()
    • BaseCache.popall()
    • BaseCache.clear()
    • BaseCache.close()
    • BaseCache.flush()
    • BaseCache.memoize()
    • BaseCache.batch_memoize()
• ahvn.cache.callback_cache module
  • CallbackCache
    • CallbackCache.__init__()
    • CallbackCache.get()
• ahvn.cache.db_cache module
  • DatabaseCache
    • DatabaseCache.__init__()
    • DatabaseCache.__len__()
    • DatabaseCache.close()
• ahvn.cache.disk_cache module
  • DiskCache
    • DiskCache.__init__()
    • DiskCache.close()
• ahvn.cache.in_mem_cache module
  • InMemCache
    • InMemCache.__init__()
    • InMemCache.close()
• ahvn.cache.json_cache module
  • JsonCache
    • JsonCache.__init__()
• ahvn.cache.mongo_cache module
  • MongoCache
    • MongoCache.__init__()
    • MongoCache.__len__()
    • MongoCache.close()
• ahvn.cache.no_cache module
  • NoCache
    • NoCache.__init__()