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)[源代码]

基类：object

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

参数:
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.

参数:

  • 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.

返回类型:

CacheEntry

classmethod from_dict(data, exclude=None)[源代码]

Creates a CacheEntry from a dictionary.

参数:

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

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

返回类型:

CacheEntry

to_dict()[源代码]

Converts the CacheEntry to a dictionary.

返回:

A dictionary representation of the CacheEntry.

返回类型:

Dict[str, Any]

clone(**updates)[源代码]

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

参数:

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

返回:

A new CacheEntry instance with updated attributes.

返回类型:

CacheEntry

annotate(expected=..., metadata=None)[源代码]

Annotates the CacheEntry with expected output and metadata.

参数:

  • 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.

返回:

A new CacheEntry instance with the annotation.

返回类型:

CacheEntry

property annotated: bool

Checks if the CacheEntry has been annotated with expected output.

返回:

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

返回类型:

bool

__init__(func, inputs=<factory>, output=Ellipsis, expected=Ellipsis, metadata=<factory>, _key=None)
参数:
返回类型:

None

class ahvn.cache.BaseCache(exclude=None, *args, **kwargs)[源代码]

基类：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.

参数:

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.

备注

  • 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.

参数:

  • 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.

参数:

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.

参数:

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

__contains__(key)[源代码]

Checks if a cache entry exists for the given key.

参数:

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

返回:

True if the cache entry exists, False otherwise.

返回类型:

bool

exists(func, **kwargs)[源代码]

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

参数:

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

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

返回:

True if the cache entry exists, False otherwise.

返回类型:

bool

__getitem__(key)[源代码]

Retrieves a cache entry for the given function and inputs.

参数:

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

返回:

The cached entry if found. Otherwise Ellipsis.

返回类型:

CacheEntry

retrieve(func, **kwargs)[源代码]

Retrieves a cached entry for the given function and inputs.

参数:

  • 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.

返回:

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

返回类型:

Any

get(func, **kwargs)[源代码]

Gets a cached value for the given function and inputs.

参数:

  • 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.

返回:

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

返回类型:

Any

__setitem__(key, value)[源代码]

Sets a cache entry for the given function and inputs.

参数:
set(func, output=..., expected=..., metadata=None, **kwargs)[源代码]

Sets a cached value for the given function and inputs.

参数:

  • 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.

参数:

entry (CacheEntry) -- The cache entry to add.

annotate(func, expected=..., metadata=None, **kwargs)[源代码]

Annotates a cached value for the given function and inputs.

参数:

  • 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.

抛出:

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.

参数:

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.

参数:

  • 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.

参数:

entry (CacheEntry) -- The cache entry to remove.

__iter__()[源代码]

Iterates over the cache entries.

生成器:

CacheEntry -- The CacheEntry objects in the cache.

返回类型:

Generator[CacheEntry, None, None]

pop()[源代码]

Pops an arbitrary cache entry from the cache.

返回:

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

返回类型:

Optional[CacheEntry]

popall()[源代码]

Pops all cache entries from the cache.

返回:

A list of all popped CacheEntry objects.

返回类型:

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.

返回类型:

Callable

参数:
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(...): ...

参数:

  • func (Callable) -- The function to cache.

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

返回:

The wrapped function with batch caching.

返回类型:

Callable

class ahvn.cache.NoCache(exclude=None, *args, **kwargs)[源代码]

基类：BaseCache

An implementation of BaseCache that does not cache any data.

参数:

exclude (Iterable[str] | None)

__init__(exclude=None, *args, **kwargs)[源代码]

Initialization.

参数:

  • 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)[源代码]

基类：BaseCache

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

参数:
__init__(path, size_limit=int(32e9), exclude=None, *args, **kwargs)[源代码]

Initialization.

参数:

  • 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.

返回类型:

None

class ahvn.cache.JsonCache(path, exclude=None, *args, **kwargs)[源代码]

基类：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.

参数:
__init__(path, exclude=None, *args, **kwargs)[源代码]

Initialization.

参数:

  • 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)[源代码]

基类：BaseCache

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

参数:

exclude (Iterable[str] | None)

__init__(exclude=None, *args, **kwargs)[源代码]

Initialization.

参数:

  • 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)[源代码]

基类：BaseCache

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

参数:
__init__(callbacks=None, feeds=None, exclude=None, *args, **kwargs)[源代码]

Initialization.

参数:

  • 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.

参数:

  • 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.

返回:

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

返回类型:

Any

Submodules