ahvn.klstore.base module

class ahvn.klstore.base.BaseKLStore(name=None, condition=None, *args, **kwargs)

基类：ToolRegistry, ABC

An abstract base class for KLStore implementations.

This class provides a protocol for storing and managing Knowledge items, each identified by a unique id attribute, typically using a BaseUKF instance as the storage unit. The class defines the required interface for storage, retrieval, insertion, batch operations, and removal of KL items, as well as iteration and clearing of the store.

Subclasses must implement the abstract methods for storage and retrieval by id, and may override batch and existence-checking methods for performance optimization.

Abstract Methods:

_get(key, default): Retrieve a KL (BaseUKF) by its key (id). Should return default (Ellipsis by default) if not found. _upsert(kl): Insert or update a KL in the store. _remove(key): Remove a KL from the store by its key (id). _itervalues(): Return an iterator over all KLs in the store. _clear(): Clear all KLs from the store.

Optional Override Methods:

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

备注

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

参数:

• name (str)

• condition (Callable | None)

__init__(name=None, condition=None, *args, **kwargs)

Initialization.

参数:

• name (Optional[str]) -- Name of the KLStore instance. If None, defaults to "default".

• condition (Optional[Callable]) -- Optional upsert/insert condition to apply to the KLStore. KLs that do not satisfy the condition will be ignored. If None, all KLs are accepted.

• *args -- Additional positional arguments.

• **kwargs -- Additional keyword arguments.

name: str

__contains__(key)

Checks if a KL exists for the given key.

参数:

key (Union[int, str, BaseUKF]) -- The key or BaseUKF instance to check.

返回:

True if the KL exists, False otherwise.

返回类型:

bool

exists(key)

Checks if a KL exists for the given key.

参数:

key (Union[int, str, BaseUKF]) -- The key or BaseUKF instance to check.

返回:

True if the KL exists, False otherwise.

返回类型:

bool

__getitem__(key)

Retrieve a KL by its key.

参数:

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

返回:

The retrieved KL instance if found. Otherwise Ellipsis.

返回类型:

BaseUKF

get(key, default=...)

Retrieves a KL by its key.

参数:

• key (Union[int, str, BaseUKF]) -- The key or BaseUKF instance to retrieve.

• default (Any) -- The default value to return if the KL is not found.

返回:

The retrieved KL instance if found. Otherwise default.

返回类型:

BaseUKF

batch_get(keys, default=..., progress=None)

Retrieves multiple KLs by their keys. The default batch get is not optimized nor parallelized. It is recommended to override _batch_get for better performance.

参数:

• keys (Iterable[Union[int, str, BaseUKF]]) -- The keys or BaseUKF instances to retrieve.

• default (Any) -- The default value to return if a KL is not found.

• progress (Type[Progress])

返回:

A list of retrieved KL instances. Missing KLs are replaced with default.

返回类型:

list

upsert(kl, **kwargs)

Upsert a KL.

参数:

• kl (BaseUKF) -- The KL to upsert.

• **kwargs -- Additional keyword arguments.

insert(kl, **kwargs)

Insert a KL.

参数:

• kl (BaseUKF) -- The KL to insert.

• kwargs -- Additional keyword arguments.

batch_upsert(kls, progress=None, **kwargs)

Upsert multiple KLs. The default batch upsert is not optimized nor parallelized. It is recommended to override this method for better performance.

参数:

• kls (Iterable[BaseUKF]) -- The KLs to upsert.

• progress (Type[Progress]) -- Progress class for reporting. None for silent, TqdmProgress for terminal.

• **kwargs -- Additional keyword arguments.

batch_insert(kls, progress=None, **kwargs)

Insert multiple KLs. The default batch insert first checks for existing keys and then batch upserts. When overriding batch_upsert, batch insert is automatically optimized. Nevertheless, the existence check uses _has, which is by default not optimized. It is recommended to override batch_insert or _has for better performance.

参数:

• kls (Iterable[BaseUKF]) -- The KLs to insert.

• progress (Type[Progress]) -- Progress class for reporting. None for silent, TqdmProgress for terminal.

• **kwargs -- Additional keyword arguments.

__delitem__(key, **kwargs)

Removes a KL from the store.

参数:

• key (Union[int, str, BaseUKF]) -- The key or BaseUKF instance to remove.

• **kwargs -- Additional keyword arguments.

remove(key, conditioned=True, **kwargs)

Removes a KL from the store.

参数:

• key (Union[int, str, BaseUKF]) -- The key or BaseUKF instance to remove.

• conditioned (bool) -- Remove only if the KL satisfies the store's condition. Default is True. Notice that conditioned removal only applies when passing a BaseUKF instance as key.

• **kwargs -- Additional keyword arguments.

batch_remove(kls, conditioned=True, progress=None, **kwargs)

Removes multiple KLs from the store. The default batch remove is not optimized nor parallelized. It is recommended to override this method for better performance.

参数:

• kls (Iterable[Union[int, str, BaseUKF]]) -- The keys or BaseUKF instances to remove.

• conditioned (bool) -- Remove only if the KLs satisfy the store's condition. Default is True. Notice that conditioned removal only applies when passing BaseUKF instances in kls.

• progress (Type[Progress]) -- Progress class for reporting. None for silent, TqdmProgress for terminal.

• **kwargs -- Additional keyword arguments.

__iter__()

Iterates over the stored KLs.

生成器:

BaseUKF -- The stored KLs in the KLStore.

返回类型:

Generator[BaseUKF, None, None]

batch_iter(batch_size=None, **kwargs)

Iterates over the stored KLs in batches.

参数:

• batch_size (Optional[int]) -- The size of each batch. If None, use the default batch size from configuration (512). If <= 0, yields all KLs in a single batch.

• **kwargs -- Additional keyword arguments.

clear()

Clears the store.

close()

Closes the store, if applicable.

flush()

Flushes the store, if applicable.