Mongodb

MongoDBAtlasVectorSearch

Bases: BasePydanticVectorStore

MongoDB Atlas Vector Store.

要使用，您应该同时安装以下内容： - 安装了pymongo python包 - 与具有Atlas Vector Search索引的MongoDB Atlas Cluster相关联的连接字符串

要开始，请访问Atlas快速入门。

创建存储后，请确保在Atlas GUI中启用索引。

请参考文档，以获取有关如何定义Atlas Vector Search索引的更多详细信息。您可以将索引命名为{ATLAS_VECTOR_SEARCH_INDEX_NAME}，并在命名空间{DB_NAME}.{COLLECTION_NAME}上创建索引。最后，在MongoDB Atlas的JSON编辑器中编写以下定义：

{
    "name": "index_name",
    "type": "vectorSearch",
    "fields":[
        {
        "type": "vector",
        "path": "embedding",
        "numDimensions": 1536,
        "similarity": "cosine"
        }
    ]
}

示例

pip install llama-index-vector-stores-mongodb

import pymongo
from llama_index.vector_stores.mongodb import MongoDBAtlasVectorSearch

# 确保您具有具有适当凭据的MongoDB URI
mongo_uri = "mongodb+srv://<username>:<password>@<host>?retryWrites=true&w=majority"
mongodb_client = pymongo.MongoClient(mongo_uri)

# 创建MongoDBAtlasVectorSearch的实例
vector_store = MongoDBAtlasVectorSearch(mongodb_client)

Source code in llama_index/vector_stores/mongodb/base.py

class MongoDBAtlasVectorSearch(BasePydanticVectorStore):
    """MongoDB Atlas Vector Store.

    要使用，您应该同时安装以下内容：
    - 安装了``pymongo`` python包
    - 与具有Atlas Vector Search索引的MongoDB Atlas Cluster相关联的连接字符串

    要开始，请访问[Atlas快速入门](https://www.mongodb.com/docs/atlas/getting-started/)。

    创建存储后，请确保在Atlas GUI中启用索引。

    请参考[文档](https://www.mongodb.com/docs/atlas/atlas-vector-search/create-index/)，以获取有关如何定义Atlas Vector Search索引的更多详细信息。您可以将索引命名为{ATLAS_VECTOR_SEARCH_INDEX_NAME}，并在命名空间{DB_NAME}.{COLLECTION_NAME}上创建索引。最后，在MongoDB Atlas的JSON编辑器中编写以下定义：

    ```
    {
        "name": "index_name",
        "type": "vectorSearch",
        "fields":[
            {
            "type": "vector",
            "path": "embedding",
            "numDimensions": 1536,
            "similarity": "cosine"
            }
        ]
    }
    ```

    示例:
        `pip install llama-index-vector-stores-mongodb`

        ```python
        import pymongo
        from llama_index.vector_stores.mongodb import MongoDBAtlasVectorSearch

        # 确保您具有具有适当凭据的MongoDB URI
        mongo_uri = "mongodb+srv://<username>:<password>@<host>?retryWrites=true&w=majority"
        mongodb_client = pymongo.MongoClient(mongo_uri)

        # 创建MongoDBAtlasVectorSearch的实例
        vector_store = MongoDBAtlasVectorSearch(mongodb_client)
        ```"""

    stores_text: bool = True
    flat_metadata: bool = True

    _mongodb_client: Any = PrivateAttr()
    _collection: Any = PrivateAttr()
    _index_name: str = PrivateAttr()
    _embedding_key: str = PrivateAttr()
    _id_key: str = PrivateAttr()
    _text_key: str = PrivateAttr()
    _metadata_key: str = PrivateAttr()
    _insert_kwargs: Dict = PrivateAttr()

    def __init__(
        self,
        mongodb_client: Optional[Any] = None,
        db_name: str = "default_db",
        collection_name: str = "default_collection",
        index_name: str = "default",
        id_key: str = "_id",
        embedding_key: str = "embedding",
        text_key: str = "text",
        metadata_key: str = "metadata",
        insert_kwargs: Optional[Dict] = None,
        **kwargs: Any,
    ) -> None:
        """初始化向量存储。

Args:
    mongodb_client：一个MongoDB客户端。
    db_name：一个MongoDB数据库名称。
    collection_name：一个MongoDB集合名称。
    index_name：一个MongoDB Atlas Vector Search索引名称。
    id_key：用作id的数据字段。
    embedding_key：将包含每个文档的嵌入的MongoDB字段。
    text_key：将包含每个文档的文本的MongoDB字段。
    metadata_key：将包含每个文档的元数据的MongoDB字段。
    insert_kwargs：在`insert`期间使用的kwargs。
"""
        if mongodb_client is not None:
            self._mongodb_client = cast(MongoClient, mongodb_client)
        else:
            if "MONGODB_URI" not in os.environ:
                raise ValueError(
                    "Must specify MONGODB_URI via env variable "
                    "if not directly passing in client."
                )
            self._mongodb_client = MongoClient(
                os.environ["MONGODB_URI"],
                driver=DriverInfo(name="llama-index", version=version("llama-index")),
            )

        self._collection = self._mongodb_client[db_name][collection_name]
        self._index_name = index_name
        self._embedding_key = embedding_key
        self._id_key = id_key
        self._text_key = text_key
        self._metadata_key = metadata_key
        self._insert_kwargs = insert_kwargs or {}

        super().__init__()

    def add(
        self,
        nodes: List[BaseNode],
        **add_kwargs: Any,
    ) -> List[str]:
        """将节点添加到索引中。

Args:
    节点：List[BaseNode]：带有嵌入的节点列表

Returns:
    成功添加节点的id列表。
"""
        ids = []
        data_to_insert = []
        for node in nodes:
            metadata = node_to_metadata_dict(
                node, remove_text=True, flat_metadata=self.flat_metadata
            )

            entry = {
                self._id_key: node.node_id,
                self._embedding_key: node.get_embedding(),
                self._text_key: node.get_content(metadata_mode=MetadataMode.NONE) or "",
                self._metadata_key: metadata,
            }
            data_to_insert.append(entry)
            ids.append(node.node_id)
        logger.debug("Inserting data into MongoDB: %s", data_to_insert)
        insert_result = self._collection.insert_many(
            data_to_insert, **self._insert_kwargs
        )
        logger.debug("Result of insert: %s", insert_result)
        return ids

    def delete(self, ref_doc_id: str, **delete_kwargs: Any) -> None:
        """使用ref_doc_id删除节点。

Args:
    ref_doc_id（str）：要删除的文档的doc_id。
"""
        # delete by filtering on the doc_id metadata
        self._collection.delete_many(
            filter={self._metadata_key + ".ref_doc_id": ref_doc_id}, **delete_kwargs
        )

    @property
    def client(self) -> Any:
        """返回 MongoDB 客户端。"""
        return self._mongodb_client

    def _query(self, query: VectorStoreQuery) -> VectorStoreQueryResult:
        params: Dict[str, Any] = {
            "queryVector": query.query_embedding,
            "path": self._embedding_key,
            "numCandidates": query.similarity_top_k * 10,
            "limit": query.similarity_top_k,
            "index": self._index_name,
        }
        if query.filters:
            params["filter"] = _to_mongodb_filter(query.filters)

        query_field = {"$vectorSearch": params}

        pipeline = [
            query_field,
            {
                "$project": {
                    "score": {"$meta": "vectorSearchScore"},
                    self._embedding_key: 0,
                }
            },
        ]
        logger.debug("Running query pipeline: %s", pipeline)
        cursor = self._collection.aggregate(pipeline)  # type: ignore
        top_k_nodes = []
        top_k_ids = []
        top_k_scores = []
        for res in cursor:
            text = res.pop(self._text_key)
            score = res.pop("score")
            id = res.pop(self._id_key)
            metadata_dict = res.pop(self._metadata_key)

            try:
                node = metadata_dict_to_node(metadata_dict)
                node.set_content(text)
            except Exception:
                # NOTE: deprecated legacy logic for backward compatibility
                metadata, node_info, relationships = legacy_metadata_dict_to_node(
                    metadata_dict
                )

                node = TextNode(
                    text=text,
                    id_=id,
                    metadata=metadata,
                    start_char_idx=node_info.get("start", None),
                    end_char_idx=node_info.get("end", None),
                    relationships=relationships,
                )

            top_k_ids.append(id)
            top_k_nodes.append(node)
            top_k_scores.append(score)
        result = VectorStoreQueryResult(
            nodes=top_k_nodes, similarities=top_k_scores, ids=top_k_ids
        )
        logger.debug("Result of query: %s", result)
        return result

    def query(self, query: VectorStoreQuery, **kwargs: Any) -> VectorStoreQueryResult:
        """查询前k个最相似节点的索引。

Args:
    query: 一个VectorStoreQuery对象。

Returns:
    包含查询结果的VectorStoreQueryResult。
"""
        return self._query(query)

client property

client: Any

返回 MongoDB 客户端。

add

add(nodes: List[BaseNode], **add_kwargs: Any) -> List[str]

将节点添加到索引中。

Returns:

Type	Description
List[str]	成功添加节点的id列表。

Source code in llama_index/vector_stores/mongodb/base.py

    def add(
        self,
        nodes: List[BaseNode],
        **add_kwargs: Any,
    ) -> List[str]:
        """将节点添加到索引中。

Args:
    节点：List[BaseNode]：带有嵌入的节点列表

Returns:
    成功添加节点的id列表。
"""
        ids = []
        data_to_insert = []
        for node in nodes:
            metadata = node_to_metadata_dict(
                node, remove_text=True, flat_metadata=self.flat_metadata
            )

            entry = {
                self._id_key: node.node_id,
                self._embedding_key: node.get_embedding(),
                self._text_key: node.get_content(metadata_mode=MetadataMode.NONE) or "",
                self._metadata_key: metadata,
            }
            data_to_insert.append(entry)
            ids.append(node.node_id)
        logger.debug("Inserting data into MongoDB: %s", data_to_insert)
        insert_result = self._collection.insert_many(
            data_to_insert, **self._insert_kwargs
        )
        logger.debug("Result of insert: %s", insert_result)
        return ids

delete

delete(ref_doc_id: str, **delete_kwargs: Any) -> None

使用ref_doc_id删除节点。

Source code in llama_index/vector_stores/mongodb/base.py

    def delete(self, ref_doc_id: str, **delete_kwargs: Any) -> None:
        """使用ref_doc_id删除节点。

Args:
    ref_doc_id（str）：要删除的文档的doc_id。
"""
        # delete by filtering on the doc_id metadata
        self._collection.delete_many(
            filter={self._metadata_key + ".ref_doc_id": ref_doc_id}, **delete_kwargs
        )

query

query(
    query: VectorStoreQuery, **kwargs: Any
) -> VectorStoreQueryResult

查询前k个最相似节点的索引。

Parameters:

Name	Type	Description	Default
query	VectorStoreQuery	一个VectorStoreQuery对象。	required

Returns:

Type	Description
VectorStoreQueryResult	包含查询结果的VectorStoreQueryResult。

Source code in llama_index/vector_stores/mongodb/base.py

    def query(self, query: VectorStoreQuery, **kwargs: Any) -> VectorStoreQueryResult:
        """查询前k个最相似节点的索引。

Args:
    query: 一个VectorStoreQuery对象。

Returns:
    包含查询结果的VectorStoreQueryResult。
"""
        return self._query(query)