from __future__ import annotations
import uuid
from typing import TYPE_CHECKING, Any, Dict, Iterable, List, Optional, Tuple, Type
from langchain_core.documents import Document
from langchain_core.embeddings import Embeddings
from langchain_core.utils import get_from_env
from langchain_core.vectorstores import VectorStore
if TYPE_CHECKING:
from meilisearch import Client
def _create_client(
client: Optional[Client] = None,
url: Optional[str] = None,
api_key: Optional[str] = None,
) -> Client:
try:
import meilisearch
except ImportError:
raise ImportError(
"Could not import meilisearch python package. "
"Please install it with `pip install meilisearch`."
)
if not client:
url = url or get_from_env("url", "MEILI_HTTP_ADDR")
try:
api_key = api_key or get_from_env("api_key", "MEILI_MASTER_KEY")
except Exception:
pass
client = meilisearch.Client(url=url, api_key=api_key)
elif not isinstance(client, meilisearch.Client):
raise ValueError(
f"client should be an instance of meilisearch.Client, "
f"got {type(client)}"
)
try:
client.version()
except ValueError as e:
raise ValueError(f"Failed to connect to Meilisearch: {e}")
return client
[docs]class Meilisearch(VectorStore):
"""`Meilisearch`向量存储。
要使用此功能，您需要安装`meilisearch` python包，并运行一个Meilisearch实例。
要了解更多关于Meilisearch Python的信息，请参考深入的Meilisearch Python文档：https://meilisearch.github.io/meilisearch-python/。
查看以下文档以了解如何运行Meilisearch实例：
https://www.meilisearch.com/docs/learn/getting_started/quick_start。
示例：
.. code-block:: python
from langchain_community.vectorstores import Meilisearch
from langchain_community.embeddings.openai import OpenAIEmbeddings
import meilisearch
# api_key是可选的；如果您的meilisearch实例需要，请提供
client = meilisearch.Client(url='http://127.0.0.1:7700', api_key='***')
embeddings = OpenAIEmbeddings()
embedders = {
"theEmbedderName": {
"source": "userProvided",
"dimensions": "1536"
}
}
vectorstore = Meilisearch(
embedding=embeddings,
embedders=embedders,
client=client,
index_name='langchain_demo',
text_key='text')"""
[docs] def __init__(
self,
embedding: Embeddings,
client: Optional[Client] = None,
url: Optional[str] = None,
api_key: Optional[str] = None,
index_name: str = "langchain-demo",
text_key: str = "text",
metadata_key: str = "metadata",
*,
embedders: Optional[Dict[str, Any]] = None,
):
"""使用Meilisearch客户端进行初始化。"""
client = _create_client(client=client, url=url, api_key=api_key)
self._client = client
self._index_name = index_name
self._embedding = embedding
self._text_key = text_key
self._metadata_key = metadata_key
self._embedders = embedders
self._embedders_settings = self._client.index(
str(self._index_name)
).update_embedders(embedders)
[docs] def add_texts(
self,
texts: Iterable[str],
metadatas: Optional[List[dict]] = None,
ids: Optional[List[str]] = None,
embedder_name: Optional[str] = "default",
**kwargs: Any,
) -> List[str]:
"""运行更多文本通过嵌入并将它们添加到向量存储中。
参数:
texts (Iterable[str]): 要添加到向量存储中的字符串/文本的可迭代对象。
embedder_name: 嵌入器的名称。默认为"default"。
metadatas (Optional[List[dict]]): 可选的元数据列表。
默认为None。
ids Optional[List[str]]: 可选的ID列表。
默认为None。
返回:
List[str]: 已添加到向量存储中的文本的ID列表。
"""
texts = list(texts)
# Embed and create the documents
docs = []
if ids is None:
ids = [uuid.uuid4().hex for _ in texts]
if metadatas is None:
metadatas = [{} for _ in texts]
embedding_vectors = self._embedding.embed_documents(texts)
for i, text in enumerate(texts):
id = ids[i]
metadata = metadatas[i]
metadata[self._text_key] = text
embedding = embedding_vectors[i]
docs.append(
{
"id": id,
"_vectors": {f"{embedder_name}": embedding},
f"{self._metadata_key}": metadata,
}
)
# Send to Meilisearch
self._client.index(str(self._index_name)).add_documents(docs)
return ids
[docs] def similarity_search(
self,
query: str,
k: int = 4,
filter: Optional[Dict[str, str]] = None,
embedder_name: Optional[str] = "default",
**kwargs: Any,
) -> List[Document]:
"""返回与查询最相似的MeiliSearch文档。
参数:
query (str): 要查找相似文档的查询文本。
embedder_name: 要使用的嵌入器的名称。默认为"default"。
k (int): 要返回的文档数量。默认为4。
filter (Optional[Dict[str, str]]): 按元数据过滤。
默认为None。
返回:
List[Document]: 查询最相似的文档列表
每个文档的文本和分数。
"""
docs_and_scores = self.similarity_search_with_score(
query=query,
embedder_name=embedder_name,
k=k,
filter=filter,
kwargs=kwargs,
)
return [doc for doc, _ in docs_and_scores]
[docs] def similarity_search_with_score(
self,
query: str,
k: int = 4,
filter: Optional[Dict[str, str]] = None,
embedder_name: Optional[str] = "default",
**kwargs: Any,
) -> List[Tuple[Document, float]]:
"""返回与查询最相似的meilisearch文档，以及分数。
参数：
query (str): 要查找相似文档的查询文本。
embedder_name: 要使用的嵌入器的名称。默认为"default"。
k (int): 要返回的文档数量。默认为4。
filter (Optional[Dict[str, str]]): 按元数据筛选。
默认为None。
返回：
List[Document]: 与查询最相似的文档列表
每个文档的文本和分数。
"""
_query = self._embedding.embed_query(query)
docs = self.similarity_search_by_vector_with_scores(
embedding=_query,
embedder_name=embedder_name,
k=k,
filter=filter,
kwargs=kwargs,
)
return docs
[docs] def similarity_search_by_vector_with_scores(
self,
embedding: List[float],
embedder_name: Optional[str] = "default",
k: int = 4,
filter: Optional[Dict[str, Any]] = None,
**kwargs: Any,
) -> List[Tuple[Document, float]]:
"""返回与嵌入向量最相似的MeiliSearch文档。
参数：
embedding（List[float]）：要查找相似文档的嵌入向量。
embedder_name：要使用的嵌入器的名称。默认为"default"。
k（int）：要返回的文档数量。默认为4。
filter（Optional[Dict[str, str]]）：按元数据筛选。默认为None。
返回：
List[Document]：与查询向量最相似的文档列表，以及每个文档的分数。
"""
docs = []
results = self._client.index(str(self._index_name)).search(
"",
{
"vector": embedding,
"hybrid": {"semanticRatio": 1.0, "embedder": embedder_name},
"limit": k,
"filter": filter,
},
)
for result in results["hits"]:
metadata = result[self._metadata_key]
if self._text_key in metadata:
text = metadata.pop(self._text_key)
semantic_score = result["_semanticScore"]
docs.append(
(Document(page_content=text, metadata=metadata), semantic_score)
)
return docs
[docs] def similarity_search_by_vector(
self,
embedding: List[float],
k: int = 4,
filter: Optional[Dict[str, str]] = None,
embedder_name: Optional[str] = "default",
**kwargs: Any,
) -> List[Document]:
"""返回与嵌入向量最相似的MeiliSearch文档。
参数：
embedding（List[float]）：要查找相似文档的嵌入向量。
embedder_name：要使用的嵌入器的名称。默认为"default"。
k（int）：要返回的文档数量。默认为4。
filter（Optional[Dict[str, str]]）：按元数据筛选。默认为None。
返回：
List[Document]：与查询向量最相似的文档列表，以及每个文档的分数。
"""
docs = self.similarity_search_by_vector_with_scores(
embedding=embedding,
embedder_name=embedder_name,
k=k,
filter=filter,
kwargs=kwargs,
)
return [doc for doc, _ in docs]
[docs] @classmethod
def from_texts(
cls: Type[Meilisearch],
texts: List[str],
embedding: Embeddings,
metadatas: Optional[List[dict]] = None,
client: Optional[Client] = None,
url: Optional[str] = None,
api_key: Optional[str] = None,
index_name: str = "langchain-demo",
ids: Optional[List[str]] = None,
text_key: Optional[str] = "text",
metadata_key: Optional[str] = "metadata",
embedders: Dict[str, Any] = {},
embedder_name: Optional[str] = "default",
**kwargs: Any,
) -> Meilisearch:
"""从原始文档构建Meilisearch包装器。
这是一个用户友好的接口，可以：
1. 嵌入文档。
2. 将文档添加到提供的Meilisearch索引中。
这旨在是一个快速入门的方式。
示例：
.. code-block:: python
from langchain_community.vectorstores import Meilisearch
from langchain_community.embeddings import OpenAIEmbeddings
import meilisearch
# 环境应该是在Meilisearch控制台中API密钥旁边指定的那个
client = meilisearch.Client(url='http://127.0.0.1:7700', api_key='***')
embedding = OpenAIEmbeddings()
embedders: Embedders index setting.
embedder_name: 嵌入器的名称。默认为"default"。
docsearch = Meilisearch.from_texts(
client=client,
embedding=embedding,
)
"""
client = _create_client(client=client, url=url, api_key=api_key)
vectorstore = cls(
embedding=embedding,
embedders=embedders,
client=client,
index_name=index_name,
)
vectorstore.add_texts(
texts=texts,
embedder_name=embedder_name,
metadatas=metadatas,
ids=ids,
text_key=text_key,
metadata_key=metadata_key,
)
return vectorstore