wdoc.utils.customs package

Submodules

wdoc.utils.customs.binary_faiss_vectorstore module

Inherit from FAISS vectorstore from langchain but using binary embeddings from faiss.

Source: https://python.langchain.com/api_reference/_modules/langchain_community/vectorstores/faiss.html#FAISS facebookresearch/faiss

class wdoc.utils.customs.binary_faiss_vectorstore.BinaryFAISS(embedding_function: Callable[[str], list[int]] | Embeddings, index: Any, docstore: Docstore, index_to_docstore_id: dict[int, str], relevance_score_fn: Callable[[float], float] | None = None, normalize_L2: bool = False, distance_strategy: DistanceStrategy = DistanceStrategy.EUCLIDEAN_DISTANCE)

Bases: CompressedFAISS

FAISS vector store integration for binary embeddings.

This subclass is specifically designed for binary embeddings that use Hamming distance for similarity calculations. Binary embeddings are represented as uint8 arrays where each bit represents a feature.

Key differences from FAISS: - Uses binary FAISS indices (IndexBinaryFlat) - Only supports Hamming distance strategy - Does not support L2 normalization (incompatible with binary) - Uses Hamming distance for relevance scoring - Embeddings must be binary (uint8 arrays)

Setup:

Install langchain_community and faiss-cpu python packages.

pip install -qU langchain_community faiss-cpu

Key init args — indexing params:

embedding_function: Embeddings

Embedding function that produces binary embeddings.

Key init args — client params:

index: Any

Binary FAISS index to use (e.g., IndexBinaryFlat).

docstore: Docstore

Docstore to use.

index_to_docstore_id: Dict[int, str]

Mapping of index to docstore id.

Instantiate:

import faiss
from langchain_community.vectorstores import BinaryFAISS
from langchain_community.docstore.in_memory import InMemoryDocstore
from langchain_openai import OpenAIEmbeddings

# Assuming binary embeddings of 128 bits (16 bytes)
index = faiss.IndexBinaryFlat(128)

vector_store = BinaryFAISS(
    embedding_function=binary_embeddings,
    index=index,
    docstore=InMemoryDocstore(),
    index_to_docstore_id={}
)

async classmethod afrom_texts(texts: list[str], embedding: Embeddings, metadatas: list[dict] | None = None, ids: list[str] | None = None, **kwargs: Any) → BinaryFAISS

Construct BinaryFAISS from raw documents with binary embeddings asynchronously.

classmethod from_texts(texts: list[str], embedding: Embeddings, metadatas: list[dict] | None = None, ids: list[str] | None = None, **kwargs: Any) → BinaryFAISS

Construct BinaryFAISS from raw documents with binary embeddings.

__init__(embedding_function: Callable[[str], list[int]] | Embeddings, index: Any, docstore: Docstore, index_to_docstore_id: dict[int, str], relevance_score_fn: Callable[[float], float] | None = None, normalize_L2: bool = False, distance_strategy: DistanceStrategy = DistanceStrategy.EUCLIDEAN_DISTANCE)

Initialize BinaryFAISS with necessary components.

Parameters:

• embedding_function – Function or Embeddings object that produces binary embeddings

• index – Binary FAISS index (must be a binary index type)

• docstore – Document storage

• index_to_docstore_id – Mapping from index to docstore IDs

• relevance_score_fn – Optional custom relevance scoring function

• normalize_L2 – Must be False for binary embeddings

• distance_strategy – Must be compatible with binary embeddings

Raises:

ValueError – If incompatible options are specified for binary embeddings

max_marginal_relevance_search_with_score_by_vector(embedding: IsEqual[dtype('uint8')]]], *, k: int = 4, fetch_k: int = 20, lambda_mult: float = 0.5, filter: Callable | dict[str, Any] | None = None) → list[tuple[Document, float]]

Return docs selected using maximal marginal relevance for binary embeddings.

Note: In BinaryFAISS, we return floats instead of the original int, just in case it would break langchain.

Note: MMR for binary embeddings uses Hamming distance calculations.

async new_aembedding_function(texts: list[str]) → IsEqual[dtype('uint8')]]]

Override to convert embeddings to binary for async operations

new_embedding_function(texts: list[str]) → IsEqual[dtype('uint8')]]]

Override to convert embeddings to binary

similarity_search_with_score_by_vector(embedding: IsEqual[dtype('uint8')]]], k: int = 4, filter: Callable | dict[str, Any] | None = None, fetch_k: int = 20, **kwargs: Any) → list[tuple[Document, float]]

Return docs most similar to binary embedding vector.

Note: In BinaryFAISS, we return floats instead of the original int, just in case it would break langchain.

Parameters:

• embedding – Binary embedding vector to look up documents similar to.

• k – Number of Documents to return. Defaults to 4.

• filter – Filter by metadata. Defaults to None.

• fetch_k – Number of Documents to fetch before filtering. Defaults to 20.

• **kwargs – Additional arguments including score_threshold.

Returns:

List of documents most similar to the query and Hamming distance in float for each. Lower score represents more similarity.

class wdoc.utils.customs.binary_faiss_vectorstore.CompressedFAISS(embedding_function: Callable[[str], List[float]] | Embeddings, index: Any, docstore: Docstore, index_to_docstore_id: Dict[int, str], relevance_score_fn: Callable[[float], float] | None = None, normalize_L2: bool = False, distance_strategy: DistanceStrategy = DistanceStrategy.EUCLIDEAN_DISTANCE)

Bases: FAISS

FAISS vector store with compressed storage.

This subclass adds zlib compression to the save_local and load_local methods to reduce storage size of the docstore and index mappings.

classmethod load_local(folder_path: str, embeddings: Embeddings, index_name: str = 'index', *, allow_dangerous_deserialization: bool = False, **kwargs: Any) → CompressedFAISS

Load FAISS index, docstore, and index_to_docstore_id from disk with decompression.

Parameters:

• folder_path – folder path to load index, docstore, and index_to_docstore_id from.

• embeddings – Embeddings to use when generating queries

• index_name – for saving with a specific index file name

• allow_dangerous_deserialization – whether to allow deserialization of the data which involves loading a pickle file. Pickle files can be modified by malicious actors to deliver a malicious payload that results in execution of arbitrary code on your machine.

save_local(folder_path: str, index_name: str = 'index') → None

Save FAISS index, docstore, and index_to_docstore_id to disk with compression.

Parameters:

• folder_path – folder path to save index, docstore, and index_to_docstore_id to.

• index_name – for saving with a specific index file name

wdoc.utils.customs.compressed_embeddings_cacher module

source : https://api.python.langchain.com/en/latest/_modules/langchain/storage/file_system.html#LocalFileStore

This is basically the exact same code but with added compression

wdoc.utils.customs.fix_llm_caching module

source : https://api.python.langchain.com/en/latest/_modules/langchain_community/cache.html#InMemoryCache

This workaround is to solve this: langchain-ai/langchain#22389 Create a caching class that looks like it’s just in memory but actually saves to sql

class wdoc.utils.customs.fix_llm_caching.SQLiteCacheFixed(database_path: str | Path, expiration_days: int | None = 0, verbose: bool = False)

Bases: BaseCache

Cache that stores things in memory using SQLiteDict.

async aclear() → None

Clear cache.

async alookup(prompt: str, llm_string: str) → Any

Look up based on prompt and llm_string.

async aupdate(prompt: str, llm_string: str, return_val: Any) → None

Update cache based on prompt and llm_string.

clear() → None

Clear cache that can take additional keyword arguments.

lookup(prompt: str, llm_string: str) → Any

Look up based on prompt and llm_string.

update(prompt: str, llm_string: str, return_val: Any) → None

Update cache based on prompt and llm_string.

wdoc.utils.customs.litellm_embeddings module

Custom embeddings to use litellm. This allows using for example “ollama/bge-m3” as a model name. Source: https://python.langchain.com/docs/how_to/custom_embeddings/

class wdoc.utils.customs.litellm_embeddings.LiteLLMEmbeddings(model: str, dimensions: int | None, api_base: str | None, private: bool, **embed_kwargs)

Bases: Embeddings

Litellm embedding model integration.

embed_documents(texts: List[str]) → List[List[float]]

Embed search docs.

embed_query(text: str) → List[float]

Embed query text.

Module contents