autogen.agentchat.contrib.vectordb.mongodb.MongoDBAtlasVectorDB

MongoDBAtlasVectorDB

MongoDBAtlasVectorDB(
    connection_string: str = '',
    database_name: str = 'vector_db',
    embedding_function: Callable[..., Any] | None = None,
    collection_name: str = None,
    index_name: str = 'vector_index',
    overwrite: bool = False,
    wait_until_index_ready: float | None = None,
    wait_until_document_ready: float | None = None
)

A Collection object for MongoDB.
Initialize the vector database.

Parameters:

Name	Description
connection_string	Type: str Default: ”
database_name	Type: str Default: ‘vector_db’
embedding_function	Type: Callable[…, Any] | None Default: None
collection_name	Type: str Default: None
index_name	Type: str Default: ‘vector_index’
overwrite	Type: bool Default: False
wait_until_index_ready	Type: float | None Default: None
wait_until_document_ready	Type: float | None Default: None

Class Attributes

active_collection

embedding_function

type

Instance Methods

create_collection

create_collection(
    self,
    collection_name: str,
    overwrite: bool = False,
    get_or_create: bool = True
) -> Collection

Create a collection in the vector database and create a vector search index in the collection.

Parameters:

Name	Description
collection_name	str	The name of the collection. Type: str
overwrite	bool	Whether to overwrite the collection if it exists. Default is False. Type: bool Default: False
get_or_create	bool	Whether to get or create the collection. Default is True Type: bool Default: True

create_index_if_not_exists

create_index_if_not_exists(
    self,
    index_name: str = 'vector_index',
    collection: Collection = None
) -> 

Creates a vector search index on the specified collection in MongoDB.

Parameters:

Name	Description
index_name	Type: str Default: ‘vector_index’
collection	The MongoDB collection to create the index on. Defaults to None. Type: Collection Default: None

create_vector_search_index

create_vector_search_index(
    self,
    collection: Collection,
    index_name: str | None = 'vector_index',
    similarity: Literal['euclidean', 'cosine', 'dotProduct'] = 'cosine'
) -> 

Create a vector search index in the collection.

Parameters:

Name	Description
collection	An existing Collection in the Atlas Database. Type: Collection
index_name	Vector Search Index name. Type: str | None Default: ‘vector_index’
similarity	Algorithm used for measuring vector similarity. Type: Literal[‘euclidean’, ‘cosine’, ‘dotProduct’] Default: ‘cosine’

delete_collection

delete_collection(self, collection_name: str) -> None

Delete the collection from the vector database.

Parameters:

Name	Description
collection_name	str	The name of the collection. Type: str

delete_docs

delete_docs(
    self,
    ids: list[str | int],
    collection_name: str = None,
    **kwargs
) -> 

Delete documents from the collection of the vector database.

Parameters:

Name	Description
ids	List[ItemID]	A list of document ids. Each id is a typed ItemID. Type: list[str | int]
collection_name	str	The name of the collection. Default is None. Type: str Default: None
**kwargs

get_collection

get_collection(self, collection_name: str = None) -> Collection

Get the collection from the vector database.

Parameters:

Name	Description
collection_name	str	The name of the collection. Default is None. If None, return the current active collection. Type: str Default: None

Returns:

Type	Description
Collection	Collection | The collection object.

get_docs_by_ids

get_docs_by_ids(
    self,
    ids: list[str | int] = None,
    collection_name: str = None,
    include: list[str] = None,
    **kwargs
) -> list[Document]

Retrieve documents from the collection of the vector database based on the ids.

Parameters:

Name	Description
ids	List[ItemID]	A list of document ids. If None, will return all the documents. Default is None. Type: list[str | int] Default: None
collection_name	str	The name of the collection. Default is None. Type: str Default: None
include	List[str]	The fields to include. If None, will include [“metadata”, “content”], ids will always be included. Basically, use include to choose whether to include embedding and metadata Type: list[str] Default: None
**kwargs

Returns:

Type	Description
list[Document]	List[Document] | The results.

insert_docs

insert_docs(
    self,
    docs: list[Document],
    collection_name: str = None,
    upsert: bool = False,
    batch_size: int = 100000,
    **kwargs: Any
) -> None

Insert Documents and Vector Embeddings into the collection of the vector database.
For large numbers of Documents, insertion is performed in batches.

Parameters:

Name	Description
docs	List[Document]	A list of documents. Each document is a TypedDict Document. Type: list[Document]
collection_name	str	The name of the collection. Default is None. Type: str Default: None
upsert	bool	Whether to update the document if it exists. Default is False. Type: bool Default: False
batch_size	Number of documents to be inserted in each batch Type: int Default: 100000
**kwargs	Type: Any

list_collections

list_collections(self) -> 

List the collections in the vector database.
Returns:
List[str] | The list of collections.

retrieve_docs

retrieve_docs(
    self,
    queries: list[str],
    collection_name: str = None,
    n_results: int = 10,
    distance_threshold: float = -1,
    **kwargs: Any
) -> list[list[tuple[Document, float]]]

Retrieve documents from the collection of the vector database based on the queries.

Parameters:

Name	Description
queries	List[str]	A list of queries. Each query is a string. Type: list[str]
collection_name	str	The name of the collection. Default is None. Type: str Default: None
n_results	int	The number of relevant documents to return. Default is 10. Type: int Default: 10
distance_threshold	float	The threshold for the distance score, only distance smaller than it will be returned. Don’t filter with it if 0. Default is -1. Type: float Default: -1
**kwargs	Type: Any

Returns:

Type	Description
list[list[tuple[Document, float]]]	QueryResults | For each query string, a list of nearest documents and their scores.

update_docs

update_docs(
    self,
    docs: list[Document],
    collection_name: str = None,
    **kwargs: Any
) -> None

Update documents, including their embeddings, in the Collection.
Optionally allow upsert as kwarg.
Uses deepcopy to avoid changing docs.

Parameters:

Name	Description
docs	List[Document]	A list of documents. Type: list[Document]
collection_name	str	The name of the collection. Default is None. Type: str Default: None
**kwargs	Type: Any