from __future__ import annotations
import logging
from typing import Any, Dict, List, Optional
from langchain_core.embeddings import Embeddings
from langchain_community.vectorstores.milvus import Milvus
logger = logging.getLogger(__name__)
[docs]class Zilliz(Milvus):
"""`Zilliz` 向量存储。
您需要安装 `pymilvus` 并运行 Zilliz 数据库。
请参阅以下文档以了解如何运行 Zilliz 实例：
https://docs.zilliz.com/docs/create-cluster
如果使用 L2/IP 度量，强烈建议对数据进行归一化。
参数:
embedding_function (Embeddings): 用于嵌入文本的函数。
collection_name (str): 要使用的 Zilliz 集合。默认为 "LangChainCollection"。
connection_args (Optional[dict[str, any]]): 用于此类的连接参数以字典形式提供。
consistency_level (str): 用于集合的一致性级别。默认为 "Session"。
index_params (Optional[dict]): 要使用的索引参数。根据服务的不同，默认为 HNSW/AUTOINDEX。
search_params (Optional[dict]): 要使用的搜索参数。默认为索引的默认值。
drop_old (Optional[bool]): 是否删除当前集合。默认为 False。
auto_id (bool): 是否为主键启用自动 id。默认为 False。
如果为 False，则需要提供文本 id（小于 65535 字节的字符串）。
如果为 True，则 Milvus 将生成唯一整数作为主键。
用于此类的连接参数以字典形式提供，以下是一些选项:
address (str): Zilliz 实例的实际地址。示例地址："localhost:19530"
uri (str): Zilliz 实例的 uri。示例 uri："https://in03-ba4234asae.api.gcp-us-west1.zillizcloud.com"
host (str): Zilliz 实例的主机。默认为 "localhost"，
如果只提供端口，则 PyMilvus 将填充默认主机。
port (str/int): Zilliz 实例的端口。默认为 19530，
如果只提供主机，则 PyMilvus 将填充默认端口。
user (str): 连接到 Zilliz 实例的用户。如果提供了用户和密码，我们将在每个 RPC 调用中添加相关的标头。
password (str): 提供用户时需要。与用户对应的密码。
token (str): API 密钥，用作用户和密码的替代品，用于无服务器集群。
secure (bool): 默认为 false。如果设置为 true，则将启用 tls。
client_key_path (str): 如果使用 tls 双向认证，需要编写 client.key 路径。
client_pem_path (str): 如果使用 tls 双向认证，需要编写 client.pem 路径。
ca_pem_path (str): 如果使用 tls 双向认证，需要编写 ca.pem 路径。
server_pem_path (str): 如果使用 tls 单向认证，需要编写 server.pem 路径。
server_name (str): 如果使用 tls，需要编写通用名称。
示例:
.. code-block:: python
from langchain_community.vectorstores import Zilliz
from langchain_community.embeddings import OpenAIEmbeddings
embedding = OpenAIEmbeddings()
# 连接到 Zilliz 实例
milvus_store = Milvus(
embedding_function = embedding,
collection_name = "LangChainCollection",
connection_args = {
"uri": "https://in03-ba4234asae.api.gcp-us-west1.zillizcloud.com",
"user": "temp",
"password": "temp",
"token": "temp", # API 密钥作为用户和密码的替代品
"secure": True
}
drop_old: True,
)
引发:
ValueError: 如果未安装 pymilvus python 包。"""
def _create_index(self) -> None:
"""在集合上创建一个索引"""
from pymilvus import Collection, MilvusException
if isinstance(self.col, Collection) and self._get_index() is None:
try:
# If no index params, use a default AutoIndex based one
if self.index_params is None:
self.index_params = {
"metric_type": "L2",
"index_type": "AUTOINDEX",
"params": {},
}
try:
self.col.create_index(
self._vector_field,
index_params=self.index_params,
using=self.alias,
)
# If default did not work, most likely Milvus self-hosted
except MilvusException:
# Use HNSW based index
self.index_params = {
"metric_type": "L2",
"index_type": "HNSW",
"params": {"M": 8, "efConstruction": 64},
}
self.col.create_index(
self._vector_field,
index_params=self.index_params,
using=self.alias,
)
logger.debug(
"Successfully created an index on collection: %s",
self.collection_name,
)
except MilvusException as e:
logger.error(
"Failed to create an index on collection: %s", self.collection_name
)
raise e
[docs] @classmethod
def from_texts(
cls,
texts: List[str],
embedding: Embeddings,
metadatas: Optional[List[dict]] = None,
collection_name: str = "LangChainCollection",
connection_args: Optional[Dict[str, Any]] = None,
consistency_level: str = "Session",
index_params: Optional[dict] = None,
search_params: Optional[dict] = None,
drop_old: bool = False,
*,
ids: Optional[List[str]] = None,
auto_id: bool = False,
**kwargs: Any,
) -> Zilliz:
"""创建一个Zilliz集合，使用HNSW对其进行索引，并插入数据。
参数:
texts (List[str]): 文本数据。
embedding (Embeddings): 嵌入函数。
metadatas (Optional[List[dict]]): 如果存在，每个文本的元数据。默认为None。
collection_name (str, optional): 要使用的集合名称。默认为"LangChainCollection"。
connection_args (dict[str, Any], optional): 要使用的连接参数。默认为DEFAULT_MILVUS_CONNECTION。
consistency_level (str, optional): 要使用的一致性级别。默认为"Session"。
index_params (Optional[dict], optional): 要使用的索引参数。默认为None。
search_params (Optional[dict], optional): 要使用的搜索参数。默认为None。
drop_old (Optional[bool], optional): 如果存在，是否删除该名称的集合。默认为False。
ids (Optional[List[str]]): 文本id列表。
auto_id (bool): 是否为主键启用自动id。默认为False。如果为False，则需要提供文本id（小于65535字节的字符串）。如果为True，Milvus将生成唯一整数作为主键。
返回:
Zilliz: Zilliz向量存储器
"""
vector_db = cls(
embedding_function=embedding,
collection_name=collection_name,
connection_args=connection_args or {},
consistency_level=consistency_level,
index_params=index_params,
search_params=search_params,
drop_old=drop_old,
auto_id=auto_id,
**kwargs,
)
vector_db.add_texts(texts=texts, metadatas=metadatas, ids=ids)
return vector_db