ahvn.klengine.mongo_engine module

MongoDB-based KL engine implementation.

class ahvn.klengine.mongo_engine.MongoKLEngine(storage, inplace=True, include=None, exclude=None, filters=None, name=None, condition=None, encoder=None, embedder=False, *args, **kwargs)

基类：BaseKLEngine

A MongoDB-based search KLEngine implementation that provides multiple search interfaces.

This class extends BaseKLEngine with specialized search methods: - MongoDB query language (MQL) search through _search_mql method - Vector similarity + filter hybrid search through _search_vector method (requires encoder/embedder) - Default search delegates to vector search

The engine supports two modes: - inplace=True: Search directly on storage MongoDB collection (storage must be MongoKLStore) - inplace=False: Create separate MongoDB collection for indexing with custom field selection

Vector Search Support: - Requires encoder and embedder to be configured - Uses MongoDB's $vectorSearch for efficient similarity search - Combines vector search with MQL filters for hybrid retrieval

Search Methods:

_search_mql(mql, include, *args, **kwargs): Raw MQL query search. _search_vector(query, topk, fetchk, include, **kwargs): Execute hybrid vector + filter search. _search = _search_vector: Alias for _search_vector for default search behavior.

Abstract Methods (inherited from BaseKLEngine):

_upsert(kl): Insert or update a KL in the engine. _remove(key): Remove a KL from the engine by its key (id). _clear(): Clear all KLs from the engine.

参数:

• storage (BaseKLStore)

• inplace (bool)

• include (List[str] | None)

• exclude (List[str] | None)

• filters (dict | None)

• name (str | None)

• condition (Callable | None)

• encoder (Callable | None)

• embedder (Callable | None)

inplace: bool = True

__init__(storage, inplace=True, include=None, exclude=None, filters=None, name=None, condition=None, encoder=None, embedder=False, *args, **kwargs)

Initialize the MongoKLEngine.

参数:

• storage (BaseKLStore) -- Attach MongoKLEngine to a KLStore. For inplace=True, must be a MongoKLStore instance. For inplace=False, can be any KLStore.

• inplace (bool) -- If True, search directly on storage MongoDB collection; if False, create a separate MongoDB collection for indexing.

• include (if inplace=False) -- List of BaseUKF field names to include. If None, includes all fields. Default is None.

• exclude (if inplace=False) -- List of BaseUKF field names to exclude. If None, excludes no fields. Default is None. Notice that exclude is applied after include, so if a field is in both, it will be excluded. It is recommended to use only one of include or exclude.

• filters (Optional[dict]) -- Global filters that will be applied to all searches.

• name (Optional[str]) -- Name of the KLEngine instance. If None, defaults to "{storage.name}_mongo_idx" for non-inplace mode.

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

• encoder (Optional[Callable]) -- Optional encoder function for converting queries/KLs to text. If None, a default provider will be used.

• embedder (Optional[Callable]) -- Optional embedder function for converting encoded text to vectors. If False, vector search is not available. If None, a default embedder will be used.

• *args -- Additional positional arguments passed to MongoKLEngine.

• **kwargs -- Additional keyword arguments passed to MongoKLEngine.

recoverable: bool = True

k_encode(kl)

Encode a KL object to text using k_encoder.

返回类型:

str

参数:

kl (Any)

k_embed(encoded_kl)

Generate embedding from encoded KL text using k_embedder.

返回类型:

List[float]

参数:

encoded_kl (str)

batch_k_encode(kls)

Batch encode KL objects to text using k_encoder.

返回类型:

List[str]

参数:

kls (Iterable[Any])

batch_k_embed(encoded_kls)

Batch generate embeddings from encoded KL texts using k_embedder.

返回类型:

List[List[float]]

参数:

encoded_kls (List[str])

q_encode(query)

Encode a query to text using q_encoder.

返回类型:

str

参数:

query (Any)

batch_q_encode(queries)

Batch encode queries to text using q_encoder.

返回类型:

List[str]

参数:

queries (Iterable[Any])

q_embed(encoded_query)

Generate embedding from encoded query text using q_embedder.

返回类型:

List[float]

参数:

encoded_query (str)

batch_q_embed(encoded_queries)

Batch generate embeddings from encoded query texts using q_embedder.

返回类型:

List[List[float]]

参数:

encoded_queries (List[str])

k_encode_embed(obj)

Encode an object and generate its embedding.

参数:

obj (Any) -- Object to encode and embed.

返回类型:

Tuple[str, List[float]]

返回:

Tuple of (encoded_text, embedding).

batch_k_encode_embed(objs)

Batch encode objects and generate their embeddings.

参数:

objs (Iterable[Any]) -- Iterable of objects to encode and embed.

返回类型:

List[Tuple[str, List[float]]]

返回:

List of tuples (encoded_text, embedding) for each object.

q_encode_embed(query)

Encode a query and generate its embedding.

参数:

query (Any) -- Query to encode and embed.

返回类型:

Tuple[str, List[float]]

返回:

Tuple of (encoded_text, embedding).

batch_q_encode_embed(queries)

Batch encode queries and generate their embeddings.

参数:

queries (Iterable[Any]) -- Iterable of queries to encode and embed.

返回类型:

List[Tuple[str, List[float]]]

返回:

List of tuples (encoded_text, embedding) for each query.

close()

Closes the engine.