Vector Data Store

Modules containing data store implementations to be used in Gen AI applications.

ChromaVectorDataStore(collection_name, embedding=None, client_type=ChromaClientType.MEMORY, persist_directory=None, host=None, port=None, headers=None, num_candidates=DEFAULT_NUM_CANDIDATES, **kwargs)

Bases: BaseVectorDataStore, CacheCompatibleMixin

Datastore for interacting with ChromaDB.

This class provides methods to interact with ChromaDB for vector storage and retrieval using the langchain-chroma integration.

Attributes:

Name	Type	Description
vector_store	Chroma	The langchain Chroma vector store instance.
collection_name	str	The name of the ChromaDB collection to use.
num_candidates	int	The maximum number of candidates to consider during search.

Initialize the ChromaDB vector data store with langchain-chroma.

Parameters:

Name	Type	Description	Default
collection_name	str	Name of the collection to use in ChromaDB.	required
embedding	BaseEMInvoker | Embeddings | None	The embedding model to perform vectorization. Defaults to None.	None
client_type	ChromaClientType	Type of ChromaDB client to use. Defaults to ChromaClientType.MEMORY.	MEMORY
persist_directory	str | None	Directory to persist vector store data. Required for PERSISTENT client type. Defaults to None.	None
host	str | None	Host address for ChromaDB server. Required for HTTP client type. Defaults to None.	None
port	int | None	Port for ChromaDB server. Required for HTTP client type. Defaults to None.	None
headers	dict | None	Headers for ChromaDB server. Used for HTTP client type. Defaults to None.	None
num_candidates	int	Maximum number of candidates to consider during search. Defaults to DEFAULT_NUM_CANDIDATES.	DEFAULT_NUM_CANDIDATES
**kwargs	Any	Additional parameters for Chroma initialization.	{}

Note

num_candidates (int, optional): This constant affects the maximum number of results to consider during the search. Index with more documents would need a higher value for the whole documents to be considered during search. This happens due to a bug with Chroma's search algorithm as discussed in this issue: [3] https://github.com/langchain-ai/langchain/issues/1946

add_chunks(chunks, **kwargs) async

Add chunks to the vector data store.

Parameters:

Name	Type	Description	Default
chunks	Chunk | list[Chunk]	A single chunk or list of chunks to add.	required
**kwargs		Additional keyword arguments for the add operation.	{}

Returns:

Type	Description
list[str]	list[str]: List of IDs of the added chunks.

clear() async

Clear all entries in the storage.

Raises:

Type	Description
NotImplementedError	Currently, app-level eviction is not supported for ChromaVectorDataStore.

delete_chunks(where=None, where_document=None, **kwargs) async

Delete chunks from the vector data store.

Parameters:

Name	Type	Description	Default
where	Where | None	A Where type dict used to filter the deletion by metadata. E.g. {"source": "mydoc"}. Defaults to None.	None
where_document	WhereDocument | None	A WhereDocument type dict used to filter the deletion by the document content. E.g. {$contains: {"text": "hello"}}. Defaults to None.	None
**kwargs	Any	Additional keyword arguments for the delete operation.	{}

Note

If no filter criteria is provided, all chunks in the collection will be deleted. Please use with caution.

delete_chunks_by_ids(ids, **kwargs) async

Delete chunks from the vector data store by IDs.

Parameters:

Name	Type	Description	Default
ids	str | list[str]	A single ID or a list of IDs to delete.	required
**kwargs	Any	Additional keyword arguments.	{}

Note

If no IDs are provided, no chunks will be deleted.

delete_entries_by_key(key, metadata=None) async

Delete entries by key.

Parameters:

Name	Type	Description	Default
key	str	The key to delete entries for.	required
metadata	dict[str, Any] | None	Optional metadata filter to apply to the search. For example, {"key": "value"}. Defaults to None.	None

Raises:

Type	Description
NotImplementedError	Currently, app-level eviction is not supported for ChromaVectorDataStore.

delete_expired_entries(now, max_size=10000) async

Delete expired entries (for TTL eviction).

Parameters:

Name	Type	Description	Default
now	datetime	The current datetime for comparison.	required
max_size	int	The maximum number of entries to return. Defaults to 10000.	10000

Raises:

Type	Description
NotImplementedError	Currently, app-level eviction is not supported for ChromaVectorDataStore.

delete_least_frequently_used_entries(num_entries) async

Delete least frequently used entries (for LFU eviction).

Parameters:

Name	Type	Description	Default
num_entries	int	Number of entries to return.	required

Raises:

Type	Description
NotImplementedError	Currently, app-level eviction is not supported for ChromaVectorDataStore.

delete_least_recently_used_entries(num_entries) async

Delete least recently used entries (for LRU eviction).

Parameters:

Name	Type	Description	Default
num_entries	int	Number of entries to return.	required

Raises:

Type	Description
NotImplementedError	Currently, app-level eviction is not supported for ChromaVectorDataStore.

exact_match(key, metadata=None) async

Find chunks that exactly match the given key.

This method searches for documents with the exact original_key in metadata.

Parameters:

Name	Type	Description	Default
key	str	The key to match.	required
metadata	dict[str, Any] | None	Optional metadata filter to apply to the search. For example, {"key": "value"}. Defaults to None.	None

Returns:

Name	Type	Description
Any	Any | None	The value stored with the exact key match, or None if no match is found.

fuzzy_match(key, max_distance=2, metadata=None) async

Find chunks that approximately match the given key using fuzzy matching.

Parameters:

Name	Type	Description	Default
key	str	The key to match.	required
max_distance	int	Maximum allowed Levenshtein distance for fuzzy matching. Higher values are more lenient. Defaults to 2.	2
metadata	dict[str, Any] | None	Optional metadata filter to apply to the search. For example, {"key": "value"}. Defaults to None.	None

Returns:

Name	Type	Description
Any	Any | None	The value with the closest fuzzy match to the key, or None if no match meets the threshold.

get_size() async

Returns the total number of vectors in the index.

If the index is not initialized returns 0.

Returns:

Name	Type	Description
int	int	The total number of vectors.

query(query, top_k=DEFAULT_TOP_K, retrieval_params=None) async

Query the vector data store for similar chunks with similarity scores.

Parameters:

Name	Type	Description	Default
query	str	The query string to find similar chunks for.	required
top_k	int	Maximum number of results to return. Defaults to DEFAULT_TOP_K.	DEFAULT_TOP_K
retrieval_params	dict[str, Any] | None	Additional parameters for retrieval. - filter (Where, optional): A Where type dict used to filter the retrieval by the metadata keys. E.g. {"$and": [{"color" : "red"}, {"price": {"$gte": 4.20}]}}. - where_document (WhereDocument, optional): A WhereDocument type dict used to filter the retrieval by the document content. E.g. {$contains: {"text": "hello"}}. Defaults to None.	None

Returns:

Type	Description
list[Chunk]	list[Chunk]: A list of Chunk objects matching the query, with similarity scores.

query_by_field(retrieval_params, limit=None, **kwargs) async

Retrieve documents that match specific metadata constraints.

This method filters and returns stored chunks based on metadata values rather than vector similarity. It is particularly useful for structured lookups, such as retrieving all chunks from a certain source, tagged with a specific label, or authored by a particular user.

Unlike semantic search methods, query_by_field operates purely on metadata fields associated with each document, allowing precise filtering based on key-value pairs.

Parameters:

Name	Type	Description	Default
retrieval_params	dict[str, Any]	A dictionary defining filter criteria. Common keys include: - filter (dict): A dictionary of metadata field conditions. - where_document (dict, optional): Conditions based on document content.	required
limit	int | None	The maximum number of results to return. If None, all matching documents will be returned.	None
**kwargs		Additional arguments to support datastore-specific behavior or filtering logic.	{}

Returns:

Type	Description
list[Chunk]	list[Chunk]: A list of Chunk objects that satisfy the metadata criteria.

Raises:

Type	Description
NotImplementedError	If not implemented in the subclass.

query_by_id(id) async

Retrieve chunks by their IDs.

Parameters:

Name	Type	Description	Default
id	str | list[str]	A single ID or a list of IDs to retrieve.	required

Returns:

Type	Description
list[Chunk]	list[Chunk]: A list of retrieved Chunk objects.

query_by_vector(vector, top_k=DEFAULT_TOP_K, min_similarity=0.8, retrieval_params=None) async

Search for documents that are similar to a given vector.

Parameters:

Name	Type	Description	Default
vector	list[float]	The query embedding vector to compare against stored vectors.	required
top_k	int	The number of top results to return. Defaults to DEFAULT_TOP_K.	DEFAULT_TOP_K
min_similarity	float	Minimum similarity score for vector similarity.	0.8
retrieval_params	dict | None	Filter parameters to narrow the search: - filter (Where): Metadata-based filter. - where_document (WhereDocument): Content-based filter. Defaults to None.	None

Returns:

Type	Description
list[Chunk]	list[Chunk]: A list of Chunk objects with similarity scores based on the input vector.

semantic_match(key, min_similarity=0.2, metadata=None) async

Find chunks that semantically match the given key using vector similarity.

Parameters:

Name	Type	Description	Default
key	str	The key to match.	required
min_similarity	float	Minimum similarity score for semantic matching (higher values are more strict). Ranges from 0 to 1. Defaults to 0.8.	0.2
metadata	dict[str, Any] | None	Optional metadata filter to apply to the search. For example, {"key": "value"}. Defaults to None.	None

Returns:

Name	Type	Description
Any	Any | None	The semantically closest value, or None if no match meets the min_similarity.

ElasticsearchVectorDataStore(index_name, embedding=None, connection=None, url=None, cloud_id=None, user=None, api_key=None, password=None, vector_query_field='vector', query_field='text', distance_strategy=None, strategy=None, request_timeout=DEFAULT_REQUEST_TIMEOUT)

Bases: BaseVectorDataStore, CacheCompatibleMixin

DataStore for interacting with Elasticsearch.

This class provides methods for executing queries and retrieving documents from Elasticsearch. It relies on the LangChain's ElasticsearchStore for vector operations and the underlying Elasticsearch client management.

Attributes:

Name	Type	Description
vector_store	ElasticsearchStore	The ElasticsearchStore instance for vector operations.
sync_vector_store	ElasticsearchStore	The ElasticsearchStore instance for sync operations.
index_name	str	The name of the Elasticsearch index.
logger	Logger	The logger object.

Initializes an instance of the ElasticsearchVectorDataStore class.

Parameters:

Name	Type	Description	Default
index_name	str	The name of the Elasticsearch index.	required
embedding	BaseEMInvoker | Embeddings | None	The embedding model to perform vectorization. Defaults to None.	None
connection	Any | None	The Elasticsearch connection object. Defaults to None.	None
url	str | None	The URL of the Elasticsearch server. Defaults to None.	None
cloud_id	str | None	The cloud ID of the Elasticsearch cluster. Defaults to None.	None
user	str | None	The username for authentication. Defaults to None.	None
api_key	str | None	The API key for authentication. Defaults to None.	None
password	str | None	The password for authentication. Defaults to None.	None
vector_query_field	str	The field name for vector queries. Defaults to "vector".	'vector'
query_field	str	The field name for text queries. Defaults to "text".	'text'
distance_strategy	str | None	The distance strategy for retrieval. Defaults to None.	None
strategy	Any | None	The retrieval strategy for retrieval. Defaults to None, in which case DenseVectorStrategy() is used.	None
request_timeout	int	The request timeout. Defaults to DEFAULT_REQUEST_TIMEOUT.	DEFAULT_REQUEST_TIMEOUT

Raises:

Type	Description
TypeError	If embedding is not an instance of BaseEMInvoker or Embeddings.

add_chunks(chunk, **kwargs) async

Adds a chunk or a list of chunks to the data store.

Parameters:

Name	Type	Description	Default
chunk	Chunk | list[Chunk]	The chunk or list of chunks to add.	required
kwargs	Any	Additional keyword arguments.	{}

Returns:

Type	Description
list[str]	list[str]: A list of unique identifiers (IDs) assigned to the added chunks.

add_embeddings(text_embeddings, metadatas=None, ids=None, **kwargs) async

Adds text embeddings to the data store.

Parameters:

Name	Type	Description	Default
text_embeddings	list[tuple[str, list[float]]]	Pairs of string and embedding to add to the store.	required
metadatas	list[dict]	Optional list of metadatas associated with the texts. Defaults to None.	None
ids	list[str]	Optional list of unique IDs. Defaults to None.	None
kwargs	Any	Additional keyword arguments.	{}

Returns:

Type	Description
list[str]	list[str]: A list of unique identifiers (IDs) assigned to the added embeddings.

autocomplete(query, field, size=20, fuzzy_tolerance=1, min_prefix_length=3, filter_query=None) async

Provides suggestions based on a prefix query for a specific field.

Parameters:

Name	Type	Description	Default
query	str	The query string.	required
field	str	The field name for autocomplete.	required
size	int	The number of suggestions to retrieve. Defaults to 20.	20
fuzzy_tolerance	int	The level of fuzziness for suggestions. Defaults to 1.	1
min_prefix_length	int	The minimum prefix length to trigger fuzzy matching. Defaults to 3.	3
filter_query	dict[str, Any] | None	The filter query. Defaults to None.	None

Returns:

Type	Description
list[str]	list[str]: A list of suggestions.

autosuggest(query, search_fields, autocomplete_field, size=20, min_length=3, filter_query=None) async

Generates suggestions across multiple fields using a multi_match query to broaden the search criteria.

Parameters:

Name	Type	Description	Default
query	str	The query string.	required
search_fields	list[str]	The fields to search for.	required
autocomplete_field	str	The field name for autocomplete.	required
size	int	The number of suggestions to retrieve. Defaults to 20.	20
min_length	int	The minimum length of the query. Defaults to 3.	3
filter_query	dict[str, Any] | None	The filter query. Defaults to None.	None

Returns:

Type	Description
list[str]	list[str]: A list of suggestions.

bm25_query(query, top_k=DEFAULT_TOP_K, search_fields=None, filter=None, metadata=None, k1=None, b=None) async

Queries the Elasticsearch data store using BM25 algorithm for keyword-based search.

Parameters:

Name	Type	Description	Default
query	str	The query string.	required
top_k	int	The number of top results to retrieve. Defaults to DEFAULT_TOP_K.	DEFAULT_TOP_K
search_fields	list[str] | None	The fields to search in. If None, defaults to ["text"]. For multiple fields, uses multi_match query. Defaults to None.	None
filter	dict[str, Any] | None	Optional metadata filter to apply to the search. For example, {"category": "AI", "source": ["doc1", "doc2"]}. Defaults to None.	None
metadata	dict[str, Any] | None	Optional metadata filter to apply to the search. DEPRECATED: Use filter parameter instead. Will be removed in a future version. For example, {"category": "AI", "source": ["doc1", "doc2"]}. Defaults to None.	None
k1	float | None	BM25 parameter controlling term frequency saturation. Higher values mean term frequency has more impact before diminishing returns. Typical values: 1.2-2.0. If None, uses Elasticsearch default (~1.2). Defaults to None.	None
b	float | None	BM25 parameter controlling document length normalization. 0.0 = no length normalization, 1.0 = full normalization. Typical values: 0.75. If None, uses Elasticsearch default (~0.75). Defaults to None.	None

Example

# Basic BM25 query on the 'text' field
results = await data_store.bm25_query("machine learning")

# BM25 query on specific fields with a custom top_k
results = await data_store.bm25_query(
    "natural language",
    top_k=5,
    search_fields=["title", "abstract"]
)

# BM25 query with filter
results = await data_store.bm25_query(
    "deep learning",
    filter={"category": "AI", "status": "published"}
)

# BM25 query with metadata filtering (deprecated)
results = await data_store.bm25_query(
    "deep learning",
    metadata={"category": "AI", "status": "published"}
)

# BM25 query with custom BM25 parameters for more aggressive term frequency weighting
results = await data_store.bm25_query(
    "artificial intelligence",
    k1=2.0,
    b=0.5
)

# BM25 query with both search fields and BM25 tuning
results = await data_store.bm25_query(
    "data science applications",
    search_fields=["content", "tags"],
    filter={"author_id": "user123", "publication_year": [2022, 2023]},
    k1=1.5,
    b=0.9
)

Returns:

Type	Description
list[Chunk]	list[Chunk]: A list of Chunk objects representing the retrieved documents.

clear() async

Clear all entries in the storage.

Raises:

Type	Description
NotImplementedError	Currently, app-level eviction is not supported for ElasticsearchVectorDataStore.

delete_chunks(query, **kwargs) async

Deletes chunks from the data store based on a query.

Parameters:

Name	Type	Description	Default
query	dict[str, Any]	Query to match documents for deletion.	required
kwargs	Any	Additional keyword arguments.	{}

Returns:

Type	Description
None	None

delete_chunks_by_ids(ids, **kwargs) async

Deletes chunks from the data store based on IDs.

Parameters:

Name	Type	Description	Default
ids	str | list[str]	A single ID or a list of IDs to delete.	required
kwargs	Any	Additional keyword arguments.	{}

delete_entries_by_key(key, metadata=None) async

Delete entries by key.

Example

key = "key-1"
metadata = {"id": "id-1"}
await delete_entries_by_key(key, metadata)

This will delete the entry with the key "key-1" by filtering by the metadata "id": "id-1".

Parameters:

Name	Type	Description	Default
key	str | list[str]	The key or list of keys to delete entries for.	required
metadata	dict[str, Any] | None	Optional metadata filter to apply to the search. Defaults to None.	None

delete_expired_entries(now, max_size=10000) async

Delete expired entries (for TTL eviction).

Parameters:

Name	Type	Description	Default
now	datetime	The current datetime for comparison.	required
max_size	int	The maximum number of entries to return. Defaults to 10000.	10000

Returns:

Type	Description
None	None

delete_least_frequently_used_entries(num_entries) async

Delete least frequently used entries (for LFU eviction).

Parameters:

Name	Type	Description	Default
num_entries	int	Number of entries to return.	required

Returns:

Type	Description
None	None

delete_least_recently_used_entries(num_entries) async

Delete least recently used entries (for LRU eviction).

Parameters:

Name	Type	Description	Default
num_entries	int	Number of entries to return.	required

Returns:

Type	Description
None	None

exact_match(key, metadata=None) async

Find chunks that exactly match the given key.

Parameters:

Name	Type	Description	Default
key	str	The key to match.	required
metadata	dict[str, Any] | None	Optional metadata filter to apply to the search. For example, {"key": "value"}. Defaults to None.	None

Returns:

Name	Type	Description
Any	Any | None	The value stored with the exact key, or None if no match is found.

fuzzy_match(key, max_distance=2, metadata=None) async

Find chunks that approximately match the given key using fuzzy matching.

Parameters:

Name	Type	Description	Default
key	str	The key to match.	required
max_distance	int	The maximum distance for fuzzy matching. Defaults to 2. Ranges from 0 to 2. Higher values are more lenient.	2
metadata	dict[str, Any] | None	Optional metadata filter to apply to the search. For example, {"key": "value"}. Defaults to None.	None

Returns:

Name	Type	Description
Any	Any | None	The value with the closest fuzzy match, or None if no match meets the threshold.

get_size() async

Returns the total number of vectors in the index.

If the index is not initialized returns 0.

Returns:

Name	Type	Description
int	int	The total number of vectors.

query(query, top_k=DEFAULT_TOP_K, retrieval_params=None) async

Queries the Elasticsearch data store and includes similarity scores.

Parameters:

Name	Type	Description	Default
query	str	The query string.	required
top_k	int	The number of top results to retrieve. Defaults to DEFAULT_TOP_K.	DEFAULT_TOP_K
retrieval_params	dict[str, Any] | None	Additional retrieval parameters. Defaults to None.	None

Returns:

Type	Description
list[Chunk]	list[Chunk]: A list of Chunk objects representing the retrieved documents with similarity scores.

query_by_field(retrieval_params, limit=None, **kwargs) async

Retrieve documents that match specific metadata constraints.

This method filters and returns stored chunks based on metadata values rather than vector similarity. It is particularly useful for structured lookups, such as retrieving all chunks from a certain source, tagged with a specific label, or authored by a particular user.

Unlike semantic search methods, query_by_field operates purely on metadata fields associated with each document, allowing precise filtering based on key-value pairs.

Expected

Returns a list of Chunk objects matching the metadata query.

Parameters:

Name	Type	Description	Default
retrieval_params	dict	Must contain a filter key with an Elasticsearch DSL query.	required
limit	int	Maximum number of results to return.	None
**kwargs		Additional arguments for the Elasticsearch search call.	{}

Returns:

Type	Description
list[Chunk]	list[Chunk]: The filtered results as Chunk objects.

query_by_id(id_) async

Queries the data store by ID and returns a list of Chunk objects.

Parameters:

Name	Type	Description	Default
id_	str | list[str]	The ID of the document to query.	required

Returns:

Type	Description
list[Chunk]	A list of Chunk objects representing the queried documents.

Note

This method not implement yet. Because the ElasticsearchStore still not implement the get_by_ids method yet.

query_by_vector(vector, top_k=DEFAULT_TOP_K, min_similarity=0.8, retrieval_params=None) async

Search for documents that are similar to a given vector.

Parameters:

Name	Type	Description	Default
vector	list[float]	The query embedding vector to compare against stored vectors.	required
top_k	int	The number of top results to return. Defaults to DEFAULT_TOP_K.	DEFAULT_TOP_K
min_similarity	float	Minimum similarity score for vector similarity.	0.8
retrieval_params	dict | None	Filter parameters to narrow the search: - filter (Where): Metadata-based filter. - where_document (WhereDocument): Content-based filter. Defaults to None.	None

Returns:

Type	Description
list[Chunk]	list[Chunk]: A list of Chunk objects with similarity scores based on the input vector.

semantic_match(key, min_similarity=0.8, metadata=None) async

Find chunks that semantically match the given key using vector similarity.

Parameters:

Name	Type	Description	Default
key	str	The key to match.	required
min_similarity	float	Minimum similarity score for semantic matching (higher values are more strict). Ranges from 0 to 1. Defaults to 0.8.	0.8
metadata	dict[str, Any] | None	Optional metadata filter to apply to the search. For example, {"key": "value"}. Defaults to None.	None

Returns:

Name	Type	Description
Any	Any | None	The semantically closest value, or None if no match meets the min_similarity threshold.

shingles(query, field, size=20, min_length=3, max_length=30, filter_query=None) async

Searches using shingles for prefix and fuzzy matching.

Parameters:

Name	Type	Description	Default
query	str	The query string.	required
field	str	The field name for autocomplete.	required
size	int	The number of suggestions to retrieve. Defaults to 20.	20
min_length	int	The minimum length of the query. Queries shorter than this limit will return an empty list. Defaults to 3.	3
max_length	int	The maximum length of the query. Queries exceeding this limit will return an empty list. Defaults to 30.	30
filter_query	dict[str, Any] | None	The filter query. Defaults to None.	None

Returns:

Type	Description
list[str]	list[str]: A list of suggestions.

InMemoryVectorDataStore(embedding=None)

Bases: BaseVectorDataStore

In-memory vector data store implementation.

This class provides a simple in-memory implementation of the BaseVectorDataStore that stores vectors and metadata in memory. It's primarily intended for testing purposes and does not require any external services.

Attributes:

Name	Type	Description
store	dict[str, dict[str, Any]]	Dictionary storing documents with their vectors and metadata. Each entry has keys: 'id', 'vector', 'text', 'metadata'.
embedding	BaseEMInvoker | None	Optional embedding model for vectorization.

Initialize the in-memory vector data store.

Parameters:

Name	Type	Description	Default
embedding	BaseEMInvoker | None	The embedding model to perform vectorization. Defaults to None, in which case vectors must be provided manually when adding chunks.	None

add_chunks(chunk, vector=None) async

Adds a chunk or a list of chunks in the data store.

Example

await store.add_chunks(
    [Chunk(id="1", content="AI contains machine learning", metadata={"topic": "AI"}),
    Chunk(id="2", content="AI in 2025", metadata={"topic": "AI"}),
])

Args: chunk (Chunk | list[Chunk]): A single chunk or a list of chunks to index. vector (Vector | list[Vector] | None, optional): A manual vector specification. Defaults to None, in which case the embedding model will be used. The vector length must match the embedding size used by the store. **kwargs: Additional keyword arguments.

Returns:

Type	Description
list[str]	list[str]: A list of unique identifiers (IDs) assigned to the added chunks.

Raises:

Type	Description
ValueError	If the number of chunks and vectors are not the same.
ValueError	If no embedding model is provided and no vector is specified.

clear() async

Clear all entries in the storage.

delete_chunks(retrieval_params=None) async

Deletes chunks from the data store by filter criteria.

Example

sample_chunks = [
    Chunk(id="1", content="AI is a topic", metadata={"topic": "AI"}),
    Chunk(id="2", content="Deep learning is a topic", metadata={"topic": "Deep Learning"}),
]
await store.add_chunks(sample_chunks)
await store.delete_chunks(retrieval_params={"topic": "AI"})

Parameters:

Name	Type	Description	Default
retrieval_params	dict[str, Any] | None	A dictionary with metadata field names as keys and their expected values. Defaults to None, in which case no operation is performed (no-op).	None

delete_chunks_by_ids(ids, **kwargs) async

Deletes a chunk or a list of chunks from the data store by their IDs.

Example

await store.delete_chunks_by_ids(["1", "2"])

Parameters:

Name	Type	Description	Default
ids	str | list[str]	A single ID or a list of IDs to delete.	required
**kwargs	Any	Additional keyword arguments (currently unused).	{}

get_size() async

Return the number of items in the data store.

Returns:

Name	Type	Description
int	int	The number of items in the data store.

query(query, top_k=DEFAULT_TOP_K, retrieval_params=None) async

Executes a query on the data store using semantic similarity.

Example

chunks = await store.add_chunks(
    [
        Chunk(id="1", content="AI contains machine learning", metadata={"topic": "AI"}),
        Chunk(id="2", content="AI in 2025", metadata={"topic": "AI"}),
    ]
)
await store.query(query="AI and machine learning", retrieval_params={"topic": "AI"})

Parameters:

Name	Type	Description	Default
query	str	The query string to execute.	required
top_k	int	The maximum number of results to return. Defaults to DEFAULT_TOP_K.	DEFAULT_TOP_K
retrieval_params	dict[str, Any] | None	Additional parameters for the query.	None

Returns:

Type	Description
list[Chunk]	list[Chunk]: A list of query results with similarity scores.

Raises:

Type	Description
ValueError	If no embedding model is provided.

query_by_field(retrieval_params, limit=None, **kwargs) async

Retrieve documents that match specific metadata constraints.

Example

sample_chunks = [
    Chunk(id="1", content="AI is a topic", metadata={"topic": "AI"}),
    Chunk(id="2", content="Deep learning is a topic", metadata={"topic": "Deep Learning"}),
]
await store.add_chunks(sample_chunks)
await store.query_by_field({"topic": "AI"})

Parameters:

Name	Type	Description	Default
retrieval_params	dict[str, Any]	A dictionary with metadata field names as keys and their expected values.	required
limit	int | None	The maximum number of results to return. Defaults to None, in which case all matching documents will be returned.	None
**kwargs		Additional arguments (currently unused).	{}

Returns:

Type	Description
list[Chunk]	list[Chunk]: A list of Chunk objects that satisfy the metadata criteria.

query_by_id(id_) async

Retrieves chunks by their IDs.

Example

chunks = await store.add_chunks(
    [Chunk(id="1", content="AI contains machine learning", metadata={"topic": "AI"}),
    Chunk(id="2", content="AI in 2025", metadata={"topic": "AI"}),
])
await store.query_by_id(["1", "2"])

Parameters:

Name	Type	Description	Default
id_	str | list[str]	A single ID or a list of IDs to retrieve.	required

Returns:

Type	Description
list[Chunk]	list[Chunk]: A list of retrieved chunks.

query_by_vector(vector, top_k=DEFAULT_TOP_K, min_similarity=0.8, retrieval_params=None) async

Search for documents that are similar to a given vector.

Example

chunks = await store.add_chunks(
    [
        Chunk(id="1", content="AI contains machine learning", metadata={"topic": "AI"}),
        Chunk(id="2", content="AI in 2025", metadata={"topic": "AI"}),
    ]
)
query_vector = await embedding.invoke("AI and machine learning")
await store.query_by_vector(query_vector, retrieval_params={"topic": "AI"})

Args: vector (Vector): The query embedding vector to compare against stored vectors. top_k (int, optional): The number of top results to return. Defaults to DEFAULT_TOP_K. min_similarity (float, optional): Minimum similarity score for vector similarity. Defaults to 0.8. retrieval_params (dict[str, Any] | None, optional): Filter parameters to narrow the search.

Returns:

Type	Description
list[Chunk]	list[Chunk]: A list of Chunk objects with similarity scores based on the input vector.

RedisVectorDataStore(index_name, url=None, client=None, embedding=None, additional_filter_fields=None)

Bases: BaseVectorDataStore, CacheCompatibleMixin

Vector data store implementation that uses Redis with RedisVL for vector search.

This class provides methods to interact with Redis for vector storage and retrieval using Redis Vector Search capabilities via RedisVL and langchain-redis.

Attributes:

Name	Type	Description
redis_url	str	URL for Redis connection.
index_name	str	Name for the vector index.
search_index	SearchIndex	RedisVL SearchIndex instance.
cache_store	SemanticCache	RedisVL SemanticCache instance.
_logger	Logger	Logger instance.

Initialize Redis vector store using RedisVL and langchain-redis.

Parameters:

Name	Type	Description	Default
index_name	str	Name of the index to use.	required
url	str	URL for Redis connection.	None
client	Redis | None	Redis client to use for vectorization.	None
embedding	BaseEMInvoker | None	Embedding function to use for vectorization. Defaults to None. If None, the default embedding model (redis/langcache-embed-v1) will be used.	None
additional_filter_fields	list[dict[str, Any]] | None	Additional filterable fields to add to the index. For example, to add entry_id as a filterable field, pass [{"name": "entry_id", "type": "text"}]. Defaults to None.	None

Notes

Besides the additional_filter_fields, the class will automatically create default filterable fields: 1. prompt: TEXT (default from redisvl). 2. response: TEXT (default from redisvl). 3. prompt_vector: VECTOR (default from redisvl). 4. chunk_id: TEXT (default additional_filter_fields).

Raises:

Type	Description
TypeError	If embedding is not an instance of BaseEMInvoker.

add_chunks(chunks, **kwargs) async

Add chunks to the vector store.

Parameters:

Name	Type	Description	Default
chunks	Chunk | list[Chunk]	A single chunk or a list of chunks to add	required
**kwargs		Additional parameters for adding chunks	{}

Returns:

Type	Description
list[str]	list[str]: List of IDs of the added chunks

clear() async

Clear all entries in the storage.

delete_chunks(query, **kwargs) async

Delete chunks from the vector store by filter/query. Not supported for Redis backend.

Parameters:

Name	Type	Description	Default
query	str	The query to delete chunks by. For example, "user_*" would match keys like "user_1", "user_2", etc.	required
**kwargs	Any	Additional keyword arguments.	{}

delete_chunks_by_ids(ids, **kwargs) async

Delete chunks from the vector store by their IDs.

Parameters:

Name	Type	Description	Default
ids	str | list[str]	A single ID or a list of IDs to delete.	required
**kwargs	Any	Additional keyword arguments.	{}

delete_entries_by_key(key, metadata=None) async

Delete entries by key.

Parameters:

Name	Type	Description	Default
key	str | list[str]	The key or list of keys to delete entries for.	required
metadata	dict[str, Any] | None	Optional metadata filter to apply to the search. For example, {"key": "value"}. Defaults to None.	None

delete_expired_entries(now, max_size=10000) async

Delete expired entries (for TTL eviction).

Parameters:

Name	Type	Description	Default
now	datetime	The current datetime for comparison.	required
max_size	int	The maximum number of entries to return. Defaults to 10000.	10000

Raises:

Type	Description
NotImplementedError	Currently, app-level eviction is not supported for RedisVectorDataStore.

delete_least_frequently_used_entries(num_entries) async

Delete least frequently used entries (for LFU eviction).

Parameters:

Name	Type	Description	Default
num_entries	int	Number of entries to return.	required

Raises:

Type	Description
NotImplementedError	Currently, app-level eviction is not supported for RedisVectorDataStore.

delete_least_recently_used_entries(num_entries) async

Delete least recently used entries (for LRU eviction).

Parameters:

Name	Type	Description	Default
num_entries	int	Number of entries to return.	required

Raises:

Type	Description
NotImplementedError	Currently, app-level eviction is not supported for RedisVectorDataStore.

exact_match(key, metadata=None) async

Find chunks that exactly match the given prompt.

Parameters:

Name	Type	Description	Default
key	str	The prompt to match.	required
metadata	dict[str, Any] | None	Optional metadata filter to apply to the search. For example, {"key": "value"}. Defaults to None.	None

Returns:

Name	Type	Description
Any	Any | None	The value stored with the matching prompt, or None if no match is found.

fuzzy_match(key, max_distance=2, metadata=None) async

Find chunks that approximately match the given key using fuzzy matching.

Parameters:

Name	Type	Description	Default
key	str	The key to match	required
max_distance	int	Maximum allowed distance for fuzzy matching (higher values allow for more differences). Maximum is 3. Defaults to 2.	2
metadata	dict[str, Any] | None	Optional metadata filter to apply to the search. For example, {"key": "value"}. Defaults to None.	None

Note

Maximum fuzzy distance is 3. This is a limitation of the Redis Vector Search and the Redis Search module. See [5] for more details.

Returns:

Name	Type	Description
Any	Any | None	The value with the closest fuzzy match, or None if no match is found

get_size() async

Returns the total number of vectors in the index.

If the index is not initialized returns 0.

Returns:

Name	Type	Description
int	int	The total number of vectors.

query(query, top_k=DEFAULT_TOP_K, retrieval_params=None) async

Search for semantically similar documents which returns similarity scores.

Parameters:

Name	Type	Description	Default
query	str	The query text to search for.	required
top_k	int	Number of top results to return.	DEFAULT_TOP_K
retrieval_params	dict[str, Any] | None	Additional parameters for the query such as: - filter: Redis filter expression to narrow results following RedisVL FilterExpression.	None

Returns:

Type	Description
list[Chunk]	list[Chunk]: List of chunks semantically similar to the query

query_by_id(id_) async

Retrieve chunks by their IDs.

Parameters:

Name	Type	Description	Default
id_	str | list[str]	A single ID or list of chunk IDs to retrieve	required

Returns:

Type	Description
list[Chunk]	list[Chunk]: List of retrieved chunks

semantic_match(key, min_similarity=0.8, metadata=None) async

Find chunks that semantically match the given key using vector similarity.

This method compares the vector embedding of the search key with vector embeddings of stored keys to find semantically similar matches.

Parameters:

Name	Type	Description	Default
key	str	The key to match	required
min_similarity	float	Minimum similarity score for semantic matching (higher values are more strict). Ranges from 0 to 1. Defaults to 0.8.	0.8
metadata	dict[str, Any] | None	Optional metadata filter to apply to the search. For example, {"key": "value"}. Defaults to None.	None

Returns:

Name	Type	Description
Any	Any | None	The semantically closest value, or None if no match meets the threshold