How to instantiate a retriever for a vector database containing multiple collections in langchain

[Alok1191]

Checked other resources

• I added a very descriptive title to this question.
• I searched the LangChain documentation with the integrated search.
• I used the GitHub search to find a similar question and didn't find it.

Commit to Help

• I commit to help with one of those options 👆

Example Code

loader = WebBaseLoader(
    web_paths=("https://lilianweng.github.io/posts/2023-06-23-agent/",),
    bs_kwargs=dict(
        parse_only=bs4.SoupStrainer(
            class_=("post-content", "post-title", "post-header")
        )
    ),
)

docs = loader.load()
embedding_ = SentenceTransformerEmbeddings(model_name="all-MiniLM-L6-v2")
text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
splits = text_splitter.split_documents(docs)
vector_store = Chroma.from_documents(documents=splits, embedding=embedding_)
retriever = vector_store.as_retriever()

Description

I am trying to create an application in which I have a vector database with multiple containers and I want the langchain retriever to be get the relevant text after looping through all the containers
As of now the vectorstore variable is associated with a single container

System Info

langchain_core: 0.2.11
langchain: 0.2.6
langchain_community: 0.2.6
langsmith: 0.1.83
langchain_chroma: 0.1.2
langchain_text_splitters: 0.2.0
langchainhub: 0.1.20
langgraph: 0.1.5

[dosubot[bot]]

Hey @Alok1191! I'm here to assist you with any bugs, questions, or contributions. Let's tackle this issue together!

To instantiate a retriever for a vector database containing multiple collections in LangChain, you can modify your setup to handle multiple collections. Here is an example of how you can achieve this:

1. Define a function to get the vector store for each collection:

from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import OpenAIEmbeddings
from pathlib import Path

def get_vectorstore(collection_name: str):
    """Get the vectorstore used for this example."""
    return Chroma(
        collection_name=collection_name,
        persist_directory=str(Path(__file__).parent.parent / f"chroma_db_{collection_name}"),
        embedding_function=OpenAIEmbeddings(),
    )

2. Define a function to create a multi-vector retriever:

from langchain.retrievers.multi_vector import MultiVectorRetriever
from langchain_community.storage import RedisStore

def get_multi_vector_retriever(docstore_id_key: str, collection_names: list):
    """Create the composed retriever object."""
    vectorstores = [get_vectorstore(name) for name in collection_names]
    store = get_docstore()
    return MultiVectorRetriever(
        vectorstores=vectorstores,
        byte_store=store,
        id_key=docstore_id_key,
    )

def get_docstore():
    """Get the metadata store used for this example."""
    return RedisStore(
        redis_url=format_redis_conn_from_env(),
        namespace="multi_vector_retriever_metadata"
    )

def format_redis_conn_from_env() -> str:
    redis_url = os.getenv("REDIS_URL", None)
    if redis_url:
        return redis_url
    else:
        using_ssl = get_boolean_env_var("REDIS_SSL", False)
        start = "rediss://" if using_ssl else "redis://"

        # if using RBAC
        password = os.getenv("REDIS_PASSWORD", None)
        username = os.getenv("REDIS_USERNAME", "default")
        if password is not None:
            start += f"{username}:{password}@"

        host = os.getenv("REDIS_HOST", "localhost")
        port = int(os.getenv("REDIS_PORT", 6379))

        return start + f"{host}:{port}"

def get_boolean_env_var(var_name, default_value=False):
    """Retrieve the boolean value of an environment variable."""
    true_values = {"true", "1", "t", "y", "yes"}
    false_values = {"false", "0", "f", "n", "no"}

    value = os.getenv(var_name, "").lower()

    if value in true_values:
        return True
    elif value in false_values:
        return False
    else:
        return default_value

3. Instantiate the multi-vector retriever with your collection names:

collection_names = ["collection1", "collection2", "collection3"]  # Add your collection names here
retriever = get_multi_vector_retriever(docstore_id_key="doc_id", collection_names=collection_names)

This setup allows you to create a retriever that can handle multiple collections within the vector database by looping through all the collections and retrieving the relevant text ^[1][2][3].

To continue talking to Dosu, mention @dosu.

[Alok1191]

@dosu
i have implemented this
from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import OpenAIEmbeddings
from pathlib import Path
from langchain.storage import InMemoryByteStore
from langchain_community.embeddings.sentence_transformer import SentenceTransformerEmbeddings
from langchain.retrievers.multi_vector import MultiVectorRetriever
embedding_ = SentenceTransformerEmbeddings(model_name="all-MiniLM-L6-v2")
def get_vectorstore(collection_name: str):
"""Get the vectorstore used for this example."""
return Chroma(
collection_name=collection_name,
embedding_function=embedding_,
)

a = get_vectorstore("collection1")

id_key = "id1"

store = InMemoryByteStore()
def get_multi_vector_retriever(collection_names: list):
"""Create the composed retriever object."""
vectorstores = [get_vectorstore(name) for name in collection_names]
return MultiVectorRetriever(
vectorstores=vectorstores,
byte_store=store,
id_key=id_key,
)
a = get_multi_vector_retriever(["collection1","collection2"])
but i get this error

raise validation_error
pydantic.v1.error_wrappers.ValidationError: 1 validation error for MultiVectorRetriever
vectorstore
field required (type=value_error.missing)

[dosubot[bot]]

The error you're encountering indicates that the MultiVectorRetriever class expects a single vectorstore field, not a list of vector stores. To handle multiple collections, you need to create a custom retriever that can manage multiple vector stores internally.

Here's how you can modify your code to create a custom retriever that loops through multiple collections:

1. Define a custom retriever class:

from langchain.retrievers.base import BaseRetriever
from langchain_core.callbacks import CallbackManagerForRetrieverRun
from typing import List, Dict
from langchain_core.document import Document

class MultiCollectionRetriever(BaseRetriever):
    def __init__(self, vectorstores: List[Chroma], id_key: str):
        self.vectorstores = vectorstores
        self.id_key = id_key

    def _get_relevant_documents(self, query: str, *, run_manager: CallbackManagerForRetrieverRun) -> List[Document]:
        all_docs = []
        for vectorstore in self.vectorstores:
            docs = vectorstore.similarity_search(query)
            all_docs.extend(docs)
        return all_docs

    async def _aget_relevant_documents(self, query: str, *, run_manager: CallbackManagerForRetrieverRun) -> List[Document]:
        all_docs = []
        for vectorstore in self.vectorstores:
            docs = await vectorstore.asimilarity_search(query)
            all_docs.extend(docs)
        return all_docs

2. Update your get_multi_vector_retriever function to use the custom retriever:

def get_multi_vector_retriever(collection_names: list):
    """Create the composed retriever object."""
    vectorstores = [get_vectorstore(name) for name in collection_names]
    return MultiCollectionRetriever(
        vectorstores=vectorstores,
        id_key=id_key,
    )

3. Instantiate the custom retriever with your collection names:

a = get_multi_vector_retriever(["collection1", "collection2"])

This custom retriever will loop through all the vector stores and retrieve relevant documents from each collection. This should resolve the ValidationError you encountered ^[1][2].

To continue talking to Dosu, mention @dosu.