from __future__ import annotations
import logging
from typing import Any, Callable, Dict, Iterable, List, Optional, Tuple, Union
import numpy as np
try:
import deeplake
from deeplake import VectorStore as DeepLakeVectorStore
from deeplake.core.fast_forwarding import version_compare
from deeplake.util.exceptions import SampleExtendError
_DEEPLAKE_INSTALLED = True
except ImportError:
_DEEPLAKE_INSTALLED = False
from langchain_core.documents import Document
from langchain_core.embeddings import Embeddings
from langchain_core.vectorstores import VectorStore
from langchain_community.vectorstores.utils import maximal_marginal_relevance
logger = logging.getLogger(__name__)
[docs]class DeepLake(VectorStore):
"""`Activeloop Deep Lake` 向量存储。
我们集成了deeplake的相似性搜索和过滤功能，用于快速原型设计。
现在，它支持用于生产用例的Tensor Query Language (TQL)，
能处理超过十亿行数据。
为什么选择Deep Lake？
- 不仅存储嵌入向量，还存储具有版本控制的原始数据。
- 无服务器，不需要另一个服务，并且可以与主要云提供商（如S3、GCS等）一起使用。
- 不仅仅是一个多模态向量存储。您可以使用数据集来微调自己的LLM模型。
要使用，您应该安装``deeplake`` python包。
示例：
.. code-block:: python
from langchain_community.vectorstores import DeepLake
from langchain_community.embeddings.openai import OpenAIEmbeddings
embeddings = OpenAIEmbeddings()
vectorstore = DeepLake("langchain_store", embeddings.embed_query)"""
_LANGCHAIN_DEFAULT_DEEPLAKE_PATH = "./deeplake/"
_valid_search_kwargs = ["lambda_mult"]
[docs] def __init__(
self,
dataset_path: str = _LANGCHAIN_DEFAULT_DEEPLAKE_PATH,
token: Optional[str] = None,
embedding: Optional[Embeddings] = None,
embedding_function: Optional[Embeddings] = None,
read_only: bool = False,
ingestion_batch_size: int = 1024,
num_workers: int = 0,
verbose: bool = True,
exec_option: Optional[str] = None,
runtime: Optional[Dict] = None,
index_params: Optional[Dict[str, Union[int, str]]] = None,
**kwargs: Any,
) -> None:
"""创建一个空的DeepLakeVectorStore或加载一个现有的。
DeepLakeVectorStore位于指定的“path”处。
示例：
>>> # 使用默认张量创建一个向量存储
>>> deeplake_vectorstore = DeepLake(
... path = <path_for_storing_Data>,
... )
>>>
>>> # 在Deep Lake托管的张量数据库中创建一个向量存储
>>> data = DeepLake(
... path = "hub://org_id/dataset_name",
... runtime = {"tensor_db": True},
... )
参数：
dataset_path (str)：用于存储到Deep Lake Vector Store的完整路径。可以是：
- 形式为“hub://org_id/dataset_name”的Deep Lake云路径。需要在Deep Lake中注册。
- 形式为“s3://bucketname/path/to/dataset”的s3路径。需要环境中的凭据或传递给creds参数。
- 形式为“./path/to/dataset”或“~/path/to/dataset”或“path/to/dataset”的本地文件系统路径。
- 形式为“mem://path/to/dataset”的内存路径，不保存数据集，而是将其保存在内存中。仅用于测试，因为它不会持久保存。默认为_LANGCHAIN_DEFAULT_DEEPLAKE_PATH。
token (str, 可选)：Activeloop令牌，用于获取路径处数据集的凭据，如果是Deep Lake数据集。令牌通常是自动生成的。可选。
embedding (Embeddings, 可选)：用于转换文档或查询的函数。可选。
embedding_function (Embeddings, 可选)：用于转换文档或查询的函数。可选。已弃用：为了向后兼容性而保留此参数。
read_only (bool)：以只读模式打开数据集。默认为False。
ingestion_batch_size (int)：在数据摄取期间，数据被分成批次。批处理大小是每个批次的大小。默认为1024。
num_workers (int)：数据摄取期间要使用的工作程序数量。默认为0。
verbose (bool)：在每次操作后打印数据集摘要。默认为True。
exec_option (str, 可选)：搜索执行的默认方法。可以是“auto”、“python”、“compute_engine”或“tensor_db”。默认为“auto”。如果为None，则设置为“auto”。
- “auto”- 根据Vector Store的存储位置选择最佳执行方法。这是默认选项。
- “python” - 在客户端上运行的纯Python实现，可用于存储在任何位置的数据。警告：不建议在大型数据集中使用此选项，因为可能会导致内存问题。
- “compute_engine” - 在客户端上运行的Deep Lake Compute Engine的高性能C++实现，可用于存储在Deep Lake中或连接到Deep Lake的任何数据。不能与内存或本地数据集一起使用。
- “tensor_db” - 高性能且完全托管的Managed Tensor Database，负责存储和查询执行。仅适用于存储在Deep Lake托管数据库中的数据。通过在数据集创建期间指定runtime = {"tensor_db": True}来将数据集存储在此数据库中。
runtime (Dict, 可选)：在Deep Lake的托管张量数据库中创建Vector Store的参数。加载现有Vector Store时不适用。要在托管张量数据库中创建Vector Store，请设置`runtime = {"tensor_db": True}`。
index_params (Optional[Dict[str, Union[int, str]]], 可选)：包含将要创建的向量索引信息的字典。默认为None，将利用“deeplake.constants”中的“DEFAULT_VECTORSTORE_INDEX_PARAMS”。
指定的键-值将覆盖默认值。
- threshold: 数据集大小超过此阈值时将为嵌入张量创建索引。当阈值值设置为-1时，将关闭索引创建。默认为-1，关闭索引。
- distance_metric: 此键指定创建向量数据库（VDB）索引时计算向量之间距离的方法。可以是与DistanceType枚举成员对应的字符串，或字符串值本身。
- 如果未提供值，则默认为“L2”。
- “L2”对应于DistanceType.L2_NORM。
- “COS”对应于DistanceType.COSINE_SIMILARITY。
- additional_params: 用于微调索引的其他参数。
**kwargs：其他可选关键字参数。
Raises:
ValueError: 如果不满足某些条件。
"""
self.ingestion_batch_size = ingestion_batch_size
self.num_workers = num_workers
self.verbose = verbose
if _DEEPLAKE_INSTALLED is False:
raise ImportError(
"Could not import deeplake python package. "
"Please install it with `pip install deeplake[enterprise]`."
)
if (
runtime == {"tensor_db": True}
and version_compare(deeplake.__version__, "3.6.7") == -1
):
raise ImportError(
"To use tensor_db option you need to update deeplake to `3.6.7` or "
"higher. "
f"Currently installed deeplake version is {deeplake.__version__}. "
)
self.dataset_path = dataset_path
if embedding_function:
logger.warning(
"Using embedding function is deprecated and will be removed "
"in the future. Please use embedding instead."
)
self.vectorstore = DeepLakeVectorStore(
path=self.dataset_path,
embedding_function=embedding_function or embedding,
read_only=read_only,
token=token,
exec_option=exec_option,
verbose=verbose,
runtime=runtime,
index_params=index_params,
**kwargs,
)
self._embedding_function = embedding_function or embedding
self._id_tensor_name = "ids" if "ids" in self.vectorstore.tensors() else "id"
@property
def embeddings(self) -> Optional[Embeddings]:
return self._embedding_function
[docs] def add_texts(
self,
texts: Iterable[str],
metadatas: Optional[List[dict]] = None,
ids: Optional[List[str]] = None,
**kwargs: Any,
) -> List[str]:
"""运行更多文本通过嵌入并添加到向量存储。
示例：
>>> ids = deeplake_vectorstore.add_texts(
... texts = <list_of_texts>,
... metadatas = <list_of_metadata_jsons>,
... ids = <list_of_ids>,
... )
参数：
texts (Iterable[str]): 要添加到向量存储的文本。
metadatas (Optional[List[dict]], optional): 可选的元数据列表。
ids (Optional[List[str]], optional): 可选的ID列表。
embedding_function (Optional[Embeddings], optional): 用于将文本转换为嵌入的嵌入函数。
**kwargs (Any): 通过此方法传递的任何其他关键字参数不受支持。
返回：
List[str]: 添加的文本的ID列表。
"""
self._validate_kwargs(kwargs, "add_texts")
kwargs = {}
if ids:
if self._id_tensor_name == "ids": # for backwards compatibility
kwargs["ids"] = ids
else:
kwargs["id"] = ids
if metadatas is None:
metadatas = [{}] * len(list(texts))
if not isinstance(texts, list):
texts = list(texts)
if texts is None:
raise ValueError("`texts` parameter shouldn't be None.")
elif len(texts) == 0:
raise ValueError("`texts` parameter shouldn't be empty.")
try:
return self.vectorstore.add(
text=texts,
metadata=metadatas,
embedding_data=texts,
embedding_tensor="embedding",
embedding_function=self._embedding_function.embed_documents, # type: ignore
return_ids=True,
**kwargs,
)
except SampleExtendError as e:
if "Failed to append a sample to the tensor 'metadata'" in str(e):
msg = (
"**Hint: You might be using invalid type of argument in "
"document loader (e.g. 'pathlib.PosixPath' instead of 'str')"
)
raise ValueError(e.args[0] + "\n\n" + msg)
else:
raise e
def _search_tql(
self,
tql: Optional[str],
exec_option: Optional[str] = None,
**kwargs: Any,
) -> List[Document]:
"""执行tql_search的函数。
参数：
tql（str）：用于直接评估的TQL查询字符串。
仅适用于`compute_engine`和`tensor_db`。
exec_option（str，可选）：支持3种搜索方式。
可以是"python"，"compute_engine"或"tensor_db"。默认为"python"。
- ``python`` - 客户端的纯Python实现。
警告：由于潜在的内存问题，不建议用于大型数据集。
- ``compute_engine`` - 用于客户端的Deep Lake Compute Engine的C++实现。
不适用于内存中或本地数据集。
- ``tensor_db`` - 托管的管理型张量数据库，用于存储和查询执行。
仅适用于Deep Lake托管数据库中的数据。
在数据集创建期间使用runtime = {"db_engine": True}。
return_score（bool）：返回带有文档的分数。默认为False。
返回：
Tuple[List[Document], List[Tuple[Document, float]]] - 两个列表的元组。
第一个列表包含文档，第二个列表包含文档和浮点分数的元组。
Raises：
ValueError：如果return_score为True但未满足某些条件。
"""
result = self.vectorstore.search(
query=tql,
exec_option=exec_option,
)
metadatas = result["metadata"]
texts = result["text"]
docs = [
Document(
page_content=text,
metadata=metadata,
)
for text, metadata in zip(texts, metadatas)
]
if kwargs:
unsupported_argument = next(iter(kwargs))
if kwargs[unsupported_argument] is not False:
raise ValueError(
f"specifying {unsupported_argument} is "
"not supported with tql search."
)
return docs
def _search(
self,
query: Optional[str] = None,
embedding: Optional[Union[List[float], np.ndarray]] = None,
embedding_function: Optional[Callable] = None,
k: int = 4,
distance_metric: Optional[str] = None,
use_maximal_marginal_relevance: bool = False,
fetch_k: Optional[int] = 20,
filter: Optional[Union[Dict, Callable]] = None,
return_score: bool = False,
exec_option: Optional[str] = None,
deep_memory: bool = False,
**kwargs: Any,
) -> Any[List[Document], List[Tuple[Document, float]]]:
"""返回与查询类似的文档。
参数:
query (str, optional): 要查找相似文档的文本。
embedding (Union[List[float], np.ndarray], optional): 查询的嵌入。
embedding_function (Callable, optional): 将 `query` 转换为嵌入的函数。
k (int): 要返回的文档数量。
distance_metric (Optional[str], optional): `L2` 表示欧几里德距离，`L1` 表示核范数，`max` 表示L-无穷距离，`cos` 表示余弦相似度，'dot' 表示点积。
filter (Union[Dict, Callable], optional): 在嵌入搜索之前的额外过滤器。
- ``Dict`` - 在htype json张量上进行键-值搜索，基于AND基础（样本必须满足所有键-值过滤器才为True）
Dict = {"tensor_name_1": {"key": value},
"tensor_name_2": {"key": value}}
- ``Function`` - 与 `deeplake.filter` 兼容的任何函数。
use_maximal_marginal_relevance (bool): 使用最大边际相关性。
fetch_k (int): MMR算法的文档数量。
return_score (bool): 返回得分。
exec_option (str, optional): 支持3种执行搜索的方式。
可以是 "python"，"compute_engine" 或 "tensor_db"。
- ``python`` - 用于客户端的纯Python实现。
警告：不建议用于大型数据集。
- ``compute_engine`` - 用于客户端的Deep Lake Compute Engine的C++实现。不适用于内存中或本地数据集。
- ``tensor_db`` - 托管的管理张量数据库，用于存储和查询执行。仅适用于Deep Lake托管数据库中的数据。
在数据集创建期间使用 runtime = {"db_engine": True}。
deep_memory (bool): 是否使用Deep Memory模型来改善搜索结果。如果未在Vector Store初始化中指定 deep_memory，则默认为False。如果为True，则距离度量设置为 "deepmemory_distance"，表示模型训练时使用的度量。使用Deep Memory模型执行搜索。如果为False，则距离度量设置为 "COS" 或用户指定的任何距离度量。
**kwargs: 其他关键字参数。
返回:
按指定距离度量的文档列表，如果 return_score 为True，则返回 (Document, score) 元组。
Raises:
ValueError: 如果未同时指定 `embedding` 和 `embedding_function`。
"""
if kwargs.get("tql_query"):
logger.warning("`tql_query` is deprecated. Please use `tql` instead.")
kwargs["tql"] = kwargs.pop("tql_query")
if kwargs.get("tql"):
return self._search_tql(
tql=kwargs["tql"],
exec_option=exec_option,
return_score=return_score,
embedding=embedding,
embedding_function=embedding_function,
distance_metric=distance_metric,
use_maximal_marginal_relevance=use_maximal_marginal_relevance,
filter=filter,
)
self._validate_kwargs(kwargs, "search")
if embedding_function:
if isinstance(embedding_function, Embeddings):
_embedding_function = embedding_function.embed_query
else:
_embedding_function = embedding_function
elif self._embedding_function:
_embedding_function = self._embedding_function.embed_query
else:
_embedding_function = None
if embedding is None:
if _embedding_function is None:
raise ValueError(
"Either `embedding` or `embedding_function` needs to be"
" specified."
)
embedding = _embedding_function(query) if query else None
if isinstance(embedding, list):
embedding = np.array(embedding, dtype=np.float32)
if len(embedding.shape) > 1:
embedding = embedding[0]
result = self.vectorstore.search(
embedding=embedding,
k=fetch_k if use_maximal_marginal_relevance else k,
distance_metric=distance_metric,
filter=filter,
exec_option=exec_option,
return_tensors=["embedding", "metadata", "text", self._id_tensor_name],
deep_memory=deep_memory,
)
scores = result["score"]
embeddings = result["embedding"]
metadatas = result["metadata"]
texts = result["text"]
if use_maximal_marginal_relevance:
lambda_mult = kwargs.get("lambda_mult", 0.5)
indices = maximal_marginal_relevance( # type: ignore
embedding, # type: ignore
embeddings,
k=min(k, len(texts)),
lambda_mult=lambda_mult,
)
scores = [scores[i] for i in indices]
texts = [texts[i] for i in indices]
metadatas = [metadatas[i] for i in indices]
docs = [
Document(
page_content=text,
metadata=metadata,
)
for text, metadata in zip(texts, metadatas)
]
if return_score:
if not isinstance(scores, list):
scores = [scores]
return [(doc, score) for doc, score in zip(docs, scores)]
return docs
[docs] def similarity_search(
self,
query: str,
k: int = 4,
**kwargs: Any,
) -> List[Document]:
"""返回与查询最相似的文档。
示例：
>>> # 使用嵌入进行搜索
>>> data = vector_store.similarity_search(
... query=<your_query>,
... k=<num_items>,
... exec_option=<preferred_exec_option>,
... )
>>> # 运行tql搜索：
>>> data = vector_store.similarity_search(
... query=None,
... tql="SELECT * WHERE id == <id>",
... exec_option="compute_engine",
... )
参数：
k (int): 要返回的文档数量。默认为4。
query (str): 要查找相似文档的文本。
**kwargs: 其他关键字参数包括：
embedding (Callable): 要使用的嵌入函数。默认为None。
distance_metric (str): 'L2'表示欧几里德距离，'L1'表示核范数，'max'
表示L-无穷范数，'cos'表示余弦相似度，'dot'表示点积。
默认为'L2'。
filter (Union[Dict, Callable], optional): 在嵌入搜索之前的额外过滤器。
- Dict: 在htype json张量上进行键-值搜索，
（样本必须满足所有键-值过滤器）
Dict = {"tensor_1": {"key": value}, "tensor_2": {"key": value}}
- Function: 与`deeplake.filter`兼容。
默认为None。
exec_option (str): 支持3种执行搜索的方式。
'python'、'compute_engine'或'tensor_db'。默认为'python'。
- 'python': 用于客户端的纯Python实现。
警告：不推荐用于大型数据集。
- 'compute_engine': 用于客户端的计算引擎的C++实现。不适用于内存中或本地数据集。
- 'tensor_db': 用于存储和查询的托管张量数据库。
仅适用于Deep Lake托管数据库中的数据。
在数据集创建期间使用`runtime = {"db_engine": True}`。
deep_memory (bool): 是否使用Deep Memory模型来改善搜索结果。
如果在Vector Store初始化中未指定deep_memory，则默认为False。
如果为True，则距离度量设置为"deepmemory_distance"，表示模型训练时使用的度量。
使用Deep Memory模型执行搜索。如果为False，则距离度量设置为"COS"或
用户指定的任何距离度量。
返回：
List[Document]: 与查询向量最相似的文档列表。
"""
return self._search(
query=query,
k=k,
use_maximal_marginal_relevance=False,
return_score=False,
**kwargs,
)
[docs] def similarity_search_by_vector(
self,
embedding: Union[List[float], np.ndarray],
k: int = 4,
**kwargs: Any,
) -> List[Document]:
"""返回与嵌入向量最相似的文档。
示例：
>>> # 使用嵌入进行搜索
>>> data = vector_store.similarity_search_by_vector(
... embedding=<your_embedding>,
... k=<num_items_to_return>,
... exec_option=<preferred_exec_option>,
... )
参数：
embedding (Union[List[float], np.ndarray]):
要查找相似文档的嵌入。
k (int): 要返回的文档数量。默认为4。
**kwargs: 包括以下附加关键字参数：
filter (Union[Dict, Callable], optional):
在嵌入搜索之前的额外过滤器。
- ``Dict`` - 在htype json张量上进行键-值搜索。如果所有键-值过滤器都满足，则为True。
Dict = {"tensor_name_1": {"key": value},
"tensor_name_2": {"key": value}}
- ``Function`` - 与`deeplake.filter`兼容的任何函数。
默认为None。
exec_option (str): 搜索执行选项包括
"python"、"compute_engine"或"tensor_db"。默认为"python"。
- "python" - 在客户端上运行的纯Python实现。可用于存储在任何位置的数据。警告：不建议在大型数据集上使用此选项，可能会出现内存问题。
- "compute_engine" - 高性能的Deep Lake Compute Engine的C++实现。在客户端上运行，可用于存储在Deep Lake中或连接到Deep Lake的任何数据。不能与内存中或本地数据集一起使用。
- "tensor_db" - 高性能、完全托管的托管张量数据库。负责存储和查询执行。仅适用于存储在Deep Lake托管数据库中的数据。
要将数据集存储在此数据库中，请在数据集创建时指定`runtime = {"db_engine": True}`。
distance_metric (str): `L2`表示欧几里德距离，`L1`表示核范数，`max`表示L-无穷距离，`cos`表示余弦相似度，'dot'表示点积。默认为`L2`。
deep_memory (bool): 是否使用Deep Memory模型来改善搜索结果。如果在Vector Store初始化中未指定deep_memory，则默认为False。如果为True，则距离度量设置为"deepmemory_distance"，表示模型训练时使用的度量。使用Deep Memory模型执行搜索。如果为False，则距离度量设置为"COS"或用户指定的任何距离度量。
返回：
List[Document]: 与查询向量最相似的文档列表。
"""
return self._search(
embedding=embedding,
k=k,
use_maximal_marginal_relevance=False,
return_score=False,
**kwargs,
)
[docs] def similarity_search_with_score(
self,
query: str,
k: int = 4,
**kwargs: Any,
) -> List[Tuple[Document, float]]:
"""运行带有返回距离的Deep Lake相似性搜索。
示例：
>>> data = vector_store.similarity_search_with_score(
... query=<your_query>,
... embedding=<your_embedding_function>
... k=<number_of_items_to_return>,
... exec_option=<preferred_exec_option>,
... )
参数：
query (str)：要搜索的查询文本。
k (int)：要返回的结果数量。默认为4。
**kwargs：其他关键字参数。其中一些参数包括：
distance_metric：`L2`表示欧氏距离，`L1`表示核范数，`max`表示极限距离，`cos`表示余弦相似度，'dot'表示点积。默认为`L2`。
filter (Optional[Dict[str, str]])：按元数据进行过滤。默认为None。
embedding_function (Callable)：要使用的嵌入函数。默认为None。
exec_option (str)：DeepLakeVectorStore支持3种执行搜索的方式。可以是"python"、"compute_engine"或"tensor_db"。默认为"python"。
- "python" - 在客户端上运行的纯Python实现。可用于存储在任何位置的数据。警告：不建议在大型数据集上使用此选项，因为可能会出现内存问题。
- "compute_engine" - Deep Lake Compute Engine的高性能C++实现。在客户端上运行，可用于存储在Deep Lake中或连接到Deep Lake的任何数据。不能与内存中或本地数据集一起使用。
- "tensor_db" - 高性能、完全托管的管理型张量数据库。负责存储和查询执行。仅适用于存储在Deep Lake托管数据库中的数据。要将数据集存储在此数据库中，请在数据集创建时指定 `runtime = {"db_engine": True}`。
deep_memory (bool)：是否使用Deep Memory模型来改善搜索结果。如果在Vector Store初始化中未指定deep_memory，则默认为False。如果为True，则距离度量设置为"deepmemory_distance"，表示模型训练时使用的度量。使用Deep Memory模型执行搜索。如果为False，则距离度量设置为"COS"或用户指定的任何距离度量。
返回：
List[Tuple[Document, float]]：与查询文本最相似的文档列表，带有浮点数距离。
"""
return self._search(
query=query,
k=k,
return_score=True,
**kwargs,
)
[docs] def max_marginal_relevance_search_by_vector(
self,
embedding: List[float],
k: int = 4,
fetch_k: int = 20,
lambda_mult: float = 0.5,
exec_option: Optional[str] = None,
**kwargs: Any,
) -> List[Document]:
"""返回使用最大边际相关性选择的文档。最大边际相关性优化了与查询的相似性和所选文档之间的多样性。
示例：
>>> data = vector_store.max_marginal_relevance_search_by_vector(
... embedding=<your_embedding>,
... fetch_k=<elements_to_fetch_before_mmr_search>,
... k=<number_of_items_to_return>,
... exec_option=<preferred_exec_option>,
... )
参数：
embedding: 要查找类似文档的嵌入。
k: 要返回的文档数量。默认为4。
fetch_k: MMR算法中要获取的文档数量。
lambda_mult: 0到1之间的数字，确定多样性程度。
0表示最大多样性，1表示最小多样性。默认为0.5。
exec_option (str): DeepLakeVectorStore支持3种搜索方式。
可以是"python"、"compute_engine"或"tensor_db"。默认为"python"。
- "python" - 在客户端上运行的纯Python实现。
可用于存储在任何位置的数据。警告：不建议在大型数据集上使用此选项，可能会出现内存问题。
- "compute_engine" - Deep Lake Compute Engine的高性能C++实现。
在客户端上运行，可用于存储在Deep Lake中或连接到Deep Lake的任何数据。不能用于内存中或本地数据集。
- "tensor_db" - 高性能、完全托管的Managed Tensor Database。
负责存储和查询执行。仅适用于存储在Deep Lake托管数据库中的数据。要将数据集存储在此数据库中，请在数据集创建时指定`runtime = {"db_engine": True}`。
deep_memory (bool): 是否使用Deep Memory模型来改善搜索结果。如果在Vector Store初始化中未指定deep_memory，则默认为False。
如果为True，则将距离度量设置为"deepmemory_distance"，表示使用模型训练的度量。使用Deep Memory模型进行搜索。
如果为False，则将距离度量设置为"COS"或用户指定的任何距离度量。
**kwargs: 其他关键字参数。
返回：
List[Documents] - 文档列表。
"""
return self._search(
embedding=embedding,
k=k,
fetch_k=fetch_k,
use_maximal_marginal_relevance=True,
lambda_mult=lambda_mult,
exec_option=exec_option,
**kwargs,
)
[docs] def max_marginal_relevance_search(
self,
query: str,
k: int = 4,
fetch_k: int = 20,
lambda_mult: float = 0.5,
exec_option: Optional[str] = None,
**kwargs: Any,
) -> List[Document]:
"""返回使用最大边际相关性选择的文档。
最大边际相关性优化了与查询的相似性和所选文档之间的多样性。
示例：
>>> # 使用嵌入进行搜索
>>> data = vector_store.max_marginal_relevance_search(
... query = <query_to_search>,
... embedding_function = <embedding_function_for_query>,
... k = <number_of_items_to_return>,
... exec_option = <preferred_exec_option>,
... )
参数：
query：要查找相似文档的文本。
k：要返回的文档数量。默认为4。
fetch_k：MMR算法的文档数量。
lambda_mult：介于0和1之间的值。0对应于最大多样性，1对应于最小多样性。默认为0.5。
exec_option（str）：支持3种执行搜索的方式。
- "python" - 在客户端上运行的纯Python实现。可用于存储在任何位置的数据。警告：不建议在大型数据集上使用此选项，可能会出现内存问题。
- "compute_engine" - Deep Lake Compute Engine的高性能C++实现。在客户端上运行，可用于存储在Deep Lake中或连接到Deep Lake的任何数据。不能与内存中或本地数据集一起使用。
- "tensor_db" - 高性能的完全托管的Managed Tensor Database。负责存储和查询执行。仅适用于存储在Deep Lake托管数据库中的数据。要将数据集存储在此数据库中，请在数据集创建期间指定`runtime = {"db_engine": True}`。
deep_memory（bool）：是否使用Deep Memory模型来改善搜索结果。如果在Vector Store初始化中未指定deep_memory，则默认为False。如果为True，则距离度量设置为"deepmemory_distance"，表示使用模型训练的度量。使用Deep Memory模型执行搜索。如果为False，则距离度量设置为"COS"或用户指定的任何距离度量。
**kwargs：其他关键字参数
返回：
通过最大边际相关性选择的文档列表。
引发：
ValueError：当MRR搜索开启但未指定嵌入函数时。
"""
embedding_function = kwargs.get("embedding") or self._embedding_function
if embedding_function is None:
raise ValueError(
"For MMR search, you must specify an embedding function on"
" `creation` or during add call."
)
return self._search(
query=query,
k=k,
fetch_k=fetch_k,
use_maximal_marginal_relevance=True,
lambda_mult=lambda_mult,
exec_option=exec_option,
embedding_function=embedding_function, # type: ignore
**kwargs,
)
[docs] @classmethod
def from_texts(
cls,
texts: List[str],
embedding: Optional[Embeddings] = None,
metadatas: Optional[List[dict]] = None,
ids: Optional[List[str]] = None,
dataset_path: str = _LANGCHAIN_DEFAULT_DEEPLAKE_PATH,
**kwargs: Any,
) -> DeepLake:
"""从原始文档中创建一个Deep Lake数据集。
如果指定了dataset_path，则数据集将持久化在该位置，否则默认在`./deeplake`。
示例：
>>> # 使用嵌入进行搜索
>>> vector_store = DeepLake.from_texts(
... texts = <要嵌入的文本>,
... embedding_function = <用于查询的嵌入函数>,
... k = <要返回的项目数量>,
... exec_option = <首选执行选项>,
)
参数：
dataset_path (str): - 数据集的完整路径。可以是：
- Deep Lake云路径的形式为``hub://username/dataset_name``。
要写入Deep Lake云数据集，
确保您已登录到Deep Lake
（从命令行使用'activeloop login'）
- AWS S3路径的形式为``s3://bucketname/path/to/dataset``。
环境中需要凭据
- Google Cloud Storage路径的形式为
``gcs://bucketname/path/to/dataset`` 环境中需要凭据
- 本地文件系统路径的形式为``./path/to/dataset``或
``~/path/to/dataset``或``path/to/dataset``。
- 内存路径的形式为``mem://path/to/dataset``，不会
保存数据集，而是将其保存在内存中。
仅应用于测试，因为它不会持久化。
texts (List[Document]): 要添加的文档列表。
embedding (Optional[Embeddings]): 嵌入函数。默认为None。
注意，在其他地方，它被称为embedding_function。
metadatas (Optional[List[dict]]): 元数据列表。默认为None。
ids (Optional[List[str]]): 文档ID列表。默认为None。
**kwargs: 附加关键字参数。
返回：
DeepLake: Deep Lake数据集。
"""
deeplake_dataset = cls(dataset_path=dataset_path, embedding=embedding, **kwargs)
deeplake_dataset.add_texts(
texts=texts,
metadatas=metadatas,
ids=ids,
)
return deeplake_dataset
[docs] def delete(self, ids: Optional[List[str]] = None, **kwargs: Any) -> bool:
"""删除数据集中的实体。
参数：
ids（可选[List[str] ]，可选）：要删除的document_ids。
默认为None。
**kwargs：子类可能使用的其他关键字参数。
- filter（可选[Dict[str, str] ]，可选）：要删除的筛选条件。
- delete_all（可选[bool]，可选）：是否删除整个数据集。
返回：
bool：删除操作是否成功。
"""
filter = kwargs.get("filter")
delete_all = kwargs.get("delete_all")
self.vectorstore.delete(ids=ids, filter=filter, delete_all=delete_all)
return True
[docs] @classmethod
def force_delete_by_path(cls, path: str) -> None:
"""通过路径强制删除数据集。
参数：
path（str）：要删除的数据集的路径。
引发：
ValueError：如果未安装deeplake。
"""
try:
import deeplake
except ImportError:
raise ImportError(
"Could not import deeplake python package. "
"Please install it with `pip install deeplake`."
)
deeplake.delete(path, large_ok=True, force=True)
[docs] def delete_dataset(self) -> None:
"""删除集合。"""
self.delete(delete_all=True)
[docs] def ds(self) -> Any:
logger.warning(
"this method is deprecated and will be removed, "
"better to use `db.vectorstore.dataset` instead."
)
return self.vectorstore.dataset
@classmethod
def _validate_kwargs(cls, kwargs, method_name): # type: ignore[no-untyped-def]
if kwargs:
valid_items = cls._get_valid_args(method_name)
unsupported_items = cls._get_unsupported_items(kwargs, valid_items)
if unsupported_items:
raise TypeError(
f"`{unsupported_items}` are not a valid "
f"argument to {method_name} method"
)
@classmethod
def _get_valid_args(cls, method_name): # type: ignore[no-untyped-def]
if method_name == "search":
return cls._valid_search_kwargs
else:
return []
@staticmethod
def _get_unsupported_items(kwargs, valid_items): # type: ignore[no-untyped-def]
kwargs = {k: v for k, v in kwargs.items() if k not in valid_items}
unsupported_items = None
if kwargs:
unsupported_items = "`, `".join(set(kwargs.keys()))
return unsupported_items