How can I create a CustomRetriever with more parameters?

[levalencia]

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

import time
import os
from dotenv import load_dotenv
from langchain.vectorstores import VectorStore
from langchain.schema import Document
from langchain.schema import BaseRetriever
from langchain_core.callbacks import CallbackManagerForRetrieverRun
from utils.telemetry import log_custom_metric
from config.config import NUMBER_OF_CHUNKS_TO_RETURN
from typing import List

# Code from: https://redis.com/blog/build-ecommerce-chatbot-with-redis/
class MyUserRetrieverPolicyLCEL(BaseRetriever):

    vectorstore: VectorStore
    username: str
    instrumentation_key: str  # Added to log metrics
    session_id: str  # Added to log metrics

    def __init__(self, vectorstore: VectorStore, username: str, instrumentation_key: str, session_id: str):
        super().__init__()
        self.vectorstore = vectorstore
        self.username = username
        self.instrumentation_key = instrumentation_key
        self.session_id = session_id
        
    def clean_metadata(self, doc):
        """
        Cleans the metadata of a document.

        Parameters:
            doc (object): The document object.

        Returns:
            dict: A dictionary containing the cleaned metadata.

        """
        metadata = doc.metadata

        return {
            "file_id": metadata["title"], 
            "source": metadata["title"] + "_page=" + str(int(metadata["chunk_id"].split("_")[-1])+1), 
            "page_number": str(int(metadata["chunk_id"].split("_")[-1])+1), 
            "document_title": metadata["document_title_result"] ,
            "document_owning_department": metadata["document_owning_department_result"],
            "manual" : metadata["manual_result"]
        }

               
    def _get_relevant_documents(        self, query: str, *, run_manager: CallbackManagerForRetrieverRun
        ) -> List[Document]:
        """
        Retrieves relevant documents based on a given query.

        Args:
            query (str): The query to search for relevant documents.

        Returns:
            list: A list of relevant documents.

        """
        docs = []
        is_match_filter = ""
        load_dotenv()
        admins = os.getenv('ADMINS', '')
        admins_list = admins.split(',')
        is_admin = self.username.split('@')[0] in admins_list

        if not is_admin:
            is_match_filter = f"search.ismatch('{self.username.split('@')[0]}', 'usernames_result')"

        # Start timing the retrieval process
        retrieval_start_time = time.time()
        # print("\n\n----------------QUERY------------------\n\n", query)
        try:
            for doc in self.vectorstore.similarity_search(query, search_type="hybrid", k=NUMBER_OF_CHUNKS_TO_RETURN, filters=is_match_filter):
                cleaned_metadata = self.clean_metadata(doc)
                docs.append(Document(
                    page_content=doc.page_content,
                    metadata=cleaned_metadata))
        finally:
                # End timing and log the retrieval time
                retrieval_end_time = time.time()
                log_custom_metric(self.instrumentation_key, 'retriever_execution_time', retrieval_end_time - retrieval_start_time, self.session_id)

        return docs

Description

I am trying to create a new custom retriever, but inside the get relevant documents I want to measure the search time, thats why I need to pass the session_id and the instrumentation_key, however it fails on the line super().init()

with this errlr
vectorstore
field required (type=value_error.missing)
username
field required (type=value_error.missing)
instrumentation_key
field required (type=value_error.missing)
session_id
field required (type=value_error.missing)

System Info

Package Information

langchain_core: 0.2.21
langchain: 0.2.9
langchain_community: 0.2.7
langsmith: 0.1.92
langchain_openai: 0.1.17
langchain_text_splitters: 0.2.2
langchainhub: 0.1.20

[dosubot[bot]]

To create a CustomRetriever with more parameters in LangChain, you need to ensure that all required fields are provided when initializing the BaseRetriever. The error "field required (type=value_error.missing)" indicates that some required fields are missing during initialization.

Here is an example of how to properly initialize your custom retriever with the required parameters:

import time
import os
from dotenv import load_dotenv
from langchain.vectorstores import VectorStore
from langchain.schema import Document
from langchain.schema import BaseRetriever
from langchain_core.callbacks import CallbackManagerForRetrieverRun
from utils.telemetry import log_custom_metric
from config.config import NUMBER_OF_CHUNKS_TO_RETURN
from typing import List

class MyUserRetrieverPolicyLCEL(BaseRetriever):

    vectorstore: VectorStore
    username: str
    instrumentation_key: str  # Added to log metrics
    session_id: str  # Added to log metrics

    def __init__(self, vectorstore: VectorStore, username: str, instrumentation_key: str, session_id: str):
        super().__init__()
        self.vectorstore = vectorstore
        self.username = username
        self.instrumentation_key = instrumentation_key
        self.session_id = session_id
        
    def clean_metadata(self, doc):
        """
        Cleans the metadata of a document.

        Parameters:
            doc (object): The document object.

        Returns:
            dict: A dictionary containing the cleaned metadata.

        """
        metadata = doc.metadata

        return {
            "file_id": metadata["title"], 
            "source": metadata["title"] + "_page=" + str(int(metadata["chunk_id"].split("_")[-1])+1), 
            "page_number": str(int(metadata["chunk_id"].split("_")[-1])+1), 
            "document_title": metadata["document_title_result"],
            "document_owning_department": metadata["document_owning_department_result"],
            "manual" : metadata["manual_result"]
        }

    def _get_relevant_documents(self, query: str, *, run_manager: CallbackManagerForRetrieverRun) -> List[Document]:
        """
        Retrieves relevant documents based on a given query.

        Args:
            query (str): The query to search for relevant documents.

        Returns:
            list: A list of relevant documents.

        """
        docs = []
        is_match_filter = ""
        load_dotenv()
        admins = os.getenv('ADMINS', '')
        admins_list = admins.split(',')
        is_admin = self.username.split('@')[0] in admins_list

        if not is_admin:
            is_match_filter = f"search.ismatch('{self.username.split('@')[0]}', 'usernames_result')"

        # Start timing the retrieval process
        retrieval_start_time = time.time()
        try:
            for doc in self.vectorstore.similarity_search(query, search_type="hybrid", k=NUMBER_OF_CHUNKS_TO_RETURN, filters=is_match_filter):
                cleaned_metadata = self.clean_metadata(doc)
                docs.append(Document(
                    page_content=doc.page_content,
                    metadata=cleaned_metadata))
        finally:
            # End timing and log the retrieval time
            retrieval_end_time = time.time()
            log_custom_metric(self.instrumentation_key, 'retriever_execution_time', retrieval_end_time - retrieval_start_time, self.session_id)

        return docs

# Example initialization
vectorstore = VectorStore()  # Replace with actual VectorStore instance
username = "user@example.com"
instrumentation_key = "your_instrumentation_key"
session_id = "your_session_id"

retriever = MyUserRetrieverPolicyLCEL(
    vectorstore=vectorstore,
    username=username,
    instrumentation_key=instrumentation_key,
    session_id=session_id
)

Make sure to replace the placeholder instances (VectorStore, username, instrumentation_key, session_id) with actual instances and values. The vectorstore, username, instrumentation_key, and session_id are required fields, and you need to provide them to avoid the "field required" error ^[1].

To continue talking to Dosu, mention @dosu.

[dosubot[bot]]

The error you're encountering suggests that the object is missing some internal attributes typically managed by Pydantic's model initialization. To fix this, you need to ensure that the SelfQueryRetriever class is properly initialized with the required attributes.

Here's how you can properly initialize the SelfQueryRetriever class:

from langchain_core.vectorstores import VectorStore
from langchain_core.language_models import BaseLanguageModel
from langchain.chains.query_constructor.schema import AttributeInfo
from langchain.retrievers.self_query.base import SelfQueryRetriever

# Assuming you have instances of llm, vectorstore, document_contents, and metadata_field_info
llm = ...  # Your language model instance
vectorstore = ...  # Your vector store instance
document_contents = "..."  # Your document contents
metadata_field_info = [...]  # Your metadata field info

# Initialize the SelfQueryRetriever
retriever = SelfQueryRetriever.from_llm(
    llm=llm,
    vectorstore=vectorstore,
    document_contents=document_contents,
    metadata_field_info=metadata_field_info
)

# Now you can use the retriever instance

Make sure that the vectorstore instance you are passing is a valid instance of a class that inherits from VectorStore. The from_llm class method handles the initialization and ensures that all required attributes are set correctly ^[1].

To continue talking to Dosu, mention @dosu.

[levalencia]

@dosu I am not using SelfQueryRetriever

I am using a CustomRetriever inheriting from BaseRetriever.

import time
import os
from dotenv import load_dotenv
from langchain.vectorstores import VectorStore
from langchain.schema import Document
from langchain.schema import BaseRetriever
from langchain_core.callbacks import CallbackManagerForRetrieverRun
from utils.telemetry import log_custom_metric
from config.config import NUMBER_OF_CHUNKS_TO_RETURN
from typing import List

Code from: https://redis.com/blog/build-ecommerce-chatbot-with-redis/

class CustomUserRetrieverPolicyLCEL(BaseRetriever):

vectorstore: VectorStore
username: str
instrumentation_key: str  # Added to log metrics
session_id: str  # Added to log metrics

def __init__(self, vectorstore: VectorStore, username: str, instrumentation_key: str, session_id: str):
    #super().__init__()
    self.vectorstore = vectorstore
    self.username = username
    self.instrumentation_key = instrumentation_key
    self.session_id = session_id
    
def clean_metadata(self, doc):
    """
    Cleans the metadata of a document.

    Parameters:
        doc (object): The document object.

    Returns:
        dict: A dictionary containing the cleaned metadata.

    """
    metadata = doc.metadata

    return {
        "file_id": metadata["title"], 
        "source": metadata["title"] + "_page=" + str(int(metadata["chunk_id"].split("_")[-1])+1), 
        "page_number": str(int(metadata["chunk_id"].split("_")[-1])+1), 
        "document_title": metadata["document_title_result"] ,
        "document_owning_department": metadata["document_owning_department_result"],
        "manual" : metadata["manual_result"]
    }

           
def _get_relevant_documents(        self, query: str, *, run_manager: CallbackManagerForRetrieverRun
    ) -> List[Document]:
    """
    Retrieves relevant documents based on a given query.

    Args:
        query (str): The query to search for relevant documents.

    Returns:
        list: A list of relevant documents.

    """
    docs = []
    is_match_filter = ""
    load_dotenv()
    admins = os.getenv('ADMINS', '')
    admins_list = admins.split(',')
    is_admin = self.username.split('@')[0] in admins_list

    if not is_admin:
        is_match_filter = f"search.ismatch('{self.username.split('@')[0]}', 'usernames_result')"

    # Start timing the retrieval process
    retrieval_start_time = time.time()
    # print("\n\n----------------QUERY------------------\n\n", query)
    try:
        for doc in self.vectorstore.similarity_search(query, search_type="hybrid", k=NUMBER_OF_CHUNKS_TO_RETURN, filters=is_match_filter):
            cleaned_metadata = self.clean_metadata(doc)
            docs.append(Document(
                page_content=doc.page_content,
                metadata=cleaned_metadata))
    finally:
            # End timing and log the retrieval time
            retrieval_end_time = time.time()
            log_custom_metric(self.instrumentation_key, 'retriever_execution_time', retrieval_end_time - retrieval_start_time, self.session_id)

    return docs

[dosubot[bot]]

To fix the error "'CustomUserRetrieverPolicyLCEL' object has no attribute 'fields_set'" when using a CustomRetriever inheriting from BaseRetriever, you need to ensure that your custom retriever class properly initializes its attributes. The error suggests that the __fields_set__ attribute, which is typically managed by Pydantic models, is missing. This can happen if your custom retriever class is not correctly set up as a Pydantic model.

Here is an example of how you can define a custom retriever class that inherits from BaseRetriever and ensures proper initialization:

import time
import os
from dotenv import load_dotenv
from langchain.vectorstores import VectorStore
from langchain.schema import Document
from langchain.schema import BaseRetriever
from langchain_core.callbacks import CallbackManagerForRetrieverRun
from utils.telemetry import log_custom_metric
from config.config import NUMBER_OF_CHUNKS_TO_RETURN
from typing import List

class CustomUserRetrieverPolicyLCEL(BaseRetriever):
    vectorstore: VectorStore
    username: str
    instrumentation_key: str  # Added to log metrics
    session_id: str  # Added to log metrics

    def __init__(self, vectorstore: VectorStore, username: str, instrumentation_key: str, session_id: str):
        self.vectorstore = vectorstore
        self.username = username
        self.instrumentation_key = instrumentation_key
        self.session_id = session_id
        super().__init__()

    def clean_metadata(self, doc):
        """
        Cleans the metadata of a document.

        Parameters:
            doc (object): The document object.

        Returns:
            dict: A dictionary containing the cleaned metadata.
        """
        metadata = doc.metadata

        return {
            "file_id": metadata["title"], 
            "source": metadata["title"] + "_page=" + str(int(metadata["chunk_id"].split("_")[-1])+1), 
            "page_number": str(int(metadata["chunk_id"].split("_")[-1])+1), 
            "document_title": metadata["document_title_result"],
            "document_owning_department": metadata["document_owning_department_result"],
            "manual": metadata["manual_result"]
        }

    def _get_relevant_documents(self, query: str, *, run_manager: CallbackManagerForRetrieverRun) -> List[Document]:
        """
        Retrieves relevant documents based on a given query.

        Args:
            query (str): The query to search for relevant documents.

        Returns:
            list: A list of relevant documents.
        """
        docs = []
        is_match_filter = ""
        load_dotenv()
        admins = os.getenv('ADMINS', '')
        admins_list = admins.split(',')
        is_admin = self.username.split('@')[0] in admins_list

        if not is_admin:
            is_match_filter = f"search.ismatch('{self.username.split('@')[0]}', 'usernames_result')"

        # Start timing the retrieval process
        retrieval_start_time = time.time()
        try:
            for doc in self.vectorstore.similarity_search(query, search_type="hybrid", k=NUMBER_OF_CHUNKS_TO_RETURN, filters=is_match_filter):
                cleaned_metadata = self.clean_metadata(doc)
                docs.append(Document(
                    page_content=doc.page_content,
                    metadata=cleaned_metadata))
        finally:
            # End timing and log the retrieval time
            retrieval_end_time = time.time()
            log_custom_metric(self.instrumentation_key, 'retriever_execution_time', retrieval_end_time - retrieval_start_time, self.session_id)

        return docs

# Example initialization
vectorstore = VectorStore()  # Replace with actual VectorStore instance
username = "user@example.com"
instrumentation_key = "your_instrumentation_key"
session_id = "your_session_id"

retriever = CustomUserRetrieverPolicyLCEL(
    vectorstore=vectorstore,
    username=username,
    instrumentation_key=instrumentation_key,
    session_id=session_id
)

In this example, the __init__ method is explicitly defined to initialize the vectorstore, username, instrumentation_key, and session_id attributes, and super().__init__() is called to ensure that the base class is properly initialized. This should resolve the issue with the missing __fields_set__ attribute ^[1].

To continue talking to Dosu, mention @dosu.

[levalencia]

@dosu I think the error is due to the VectorStore type,

vector_store = setup_vector_store(embeddings, access_token_keyvault, access_token_search)

def setup_vector_store(embeddings, access_token_keyvault, access_token_search):
"""
Set up the vector store.

Args:
    embeddings (AzureOpenAIEmbeddings): Instance of AzureOpenAIEmbeddings.

Returns:
    AzureSearch: Instance of AzureSearch.

"""
try:
    SHD_SEARCH_SERVICE_ENDPOINT = config.SHD_SEARCH_SERVICE_ENDPOINT(access_token_keyvault)
    
    return AzureSearch(
        azure_search_endpoint=SHD_SEARCH_SERVICE_ENDPOINT,
        azure_search_key=None,
        azure_ad_access_token = access_token_search,
        index_name="index-policy",
        embedding_function=embeddings.embed_query,
        semantic_configuration_name="vector-policy-qa-semantic-configuration"
    )
except Exception as e:
    logging.error(f"Search exception: {e}")
    raise e
    
    as you can see this comes from AzureSearch.
    
    but      the VectorStore in the retriever is from langchain
    
    import time

import os
from dotenv import load_dotenv
from langchain.vectorstores import VectorStore
from utils.AzureSearch import AzureSearch
from langchain.schema import Document
from langchain.schema import BaseRetriever
from langchain_core.callbacks import CallbackManagerForRetrieverRun
from utils.telemetry import log_custom_metric
from config.config import NUMBER_OF_CHUNKS_TO_RETURN
from typing import List

Code from: https://redis.com/blog/build-ecommerce-chatbot-with-redis/

class CustomUserRetrieverPolicyLCEL(BaseRetriever):

"""
ArgenxUserRetriever class extends BaseRetriever and is designed for retrieving relevant documents
based on a user query using hybrid similarity search with a VectorStore.

Attributes:
- vectorstore (VectorStore): The VectorStore instance used for similarity search.
- username (str): The username associated with the documents, used for personalized retrieval.

Methods:
- clean_metadata(self, doc): Cleans the metadata of a document, extracting relevant information for display.
- get_relevant_documents(self, query): Retrieves relevant documents based on a user query using hybrid similarity search.

Example:
retriever = ArgenxRetriever(vectorstore=vector_store, username="john_doe")
relevant_docs = retriever.get_relevant_documents("How does photosynthesis work?")
for doc in relevant_docs:
    print(doc.metadata["Title"], doc.page_content)
"""

vectorstore: VectorStore
username: str
instrumentation_key: str  # Added to log metrics
session_id: str  # Added to log metrics

def __init__(self, vectorstore: VectorStore, username: str, instrumentation_key: str, session_id: str):
    #self.vectorstore = vectorstore
    #self.username = username
    #self.instrumentation_key = instrumentation_key
    #self.session_id = session_id
    #super().__init__()
    
def clean_metadata(self, doc):
    """
    Cleans the metadata of a document.

    Parameters:
        doc (object): The document object.

    Returns:
        dict: A dictionary containing the cleaned metadata.

    """
    metadata = doc.metadata

    return {
        "file_id": metadata["title"], 
        "source": metadata["title"] + "_page=" + str(int(metadata["chunk_id"].split("_")[-1])+1), 
        "page_number": str(int(metadata["chunk_id"].split("_")[-1])+1), 
        "document_title": metadata["document_title_result"] ,
        "document_owning_department": metadata["document_owning_department_result"],
        "manual" : metadata["manual_result"]
    }

           
def _get_relevant_documents(        self, query: str, *, run_manager: CallbackManagerForRetrieverRun
    ) -> List[Document]:
    """
    Retrieves relevant documents based on a given query.

    Args:
        query (str): The query to search for relevant documents.

    Returns:
        list: A list of relevant documents.

    """
    docs = []
    is_match_filter = ""
    load_dotenv()
    admins = os.getenv('ADMINS', '')
    admins_list = admins.split(',')
    is_admin = self.username.split('@')[0] in admins_list

    if not is_admin:
        is_match_filter = f"search.ismatch('{self.username.split('@')[0]}', 'usernames_result')"

    # Start timing the retrieval process
    retrieval_start_time = time.time()
    # print("\n\n----------------QUERY------------------\n\n", query)
    try:
        for doc in self.vectorstore.similarity_search(query, search_type="hybrid", k=NUMBER_OF_CHUNKS_TO_RETURN, filters=is_match_filter):
            cleaned_metadata = self.clean_metadata(doc)
            docs.append(Document(
                page_content=doc.page_content,
                metadata=cleaned_metadata))
    finally:
            # End timing and log the retrieval time
            retrieval_end_time = time.time()
            log_custom_metric(self.instrumentation_key, 'retriever_execution_time', retrieval_end_time - retrieval_start_time, self.session_id)

    return docs

[dosubot[bot]]

The issue you're encountering is likely due to a mismatch between the VectorStore type expected by your CustomUserRetrieverPolicyLCEL class and the actual type being used, which is AzureSearch. The AzureSearch class should be compatible with the VectorStore interface.

To resolve this, ensure that AzureSearch is correctly imported and used as a VectorStore. Here is an updated version of your code:

import time
import os
from dotenv import load_dotenv
from langchain.vectorstores import VectorStore
from utils.AzureSearch import AzureSearch
from langchain.schema import Document
from langchain.schema import BaseRetriever
from langchain_core.callbacks import CallbackManagerForRetrieverRun
from utils.telemetry import log_custom_metric
from config.config import NUMBER_OF_CHUNKS_TO_RETURN
from typing import List

class CustomUserRetrieverPolicyLCEL(BaseRetriever):
    vectorstore: VectorStore
    username: str
    instrumentation_key: str  # Added to log metrics
    session_id: str  # Added to log metrics

    def __init__(self, vectorstore: VectorStore, username: str, instrumentation_key: str, session_id: str):
        self.vectorstore = vectorstore
        self.username = username
        self.instrumentation_key = instrumentation_key
        self.session_id = session_id
        super().__init__()

    def clean_metadata(self, doc):
        metadata = doc.metadata
        return {
            "file_id": metadata["title"], 
            "source": metadata["title"] + "_page=" + str(int(metadata["chunk_id"].split("_")[-1])+1), 
            "page_number": str(int(metadata["chunk_id"].split("_")[-1])+1), 
            "document_title": metadata["document_title_result"],
            "document_owning_department": metadata["document_owning_department_result"],
            "manual": metadata["manual_result"]
        }

    def _get_relevant_documents(self, query: str, *, run_manager: CallbackManagerForRetrieverRun) -> List[Document]:
        docs = []
        is_match_filter = ""
        load_dotenv()
        admins = os.getenv('ADMINS', '')
        admins_list = admins.split(',')
        is_admin = self.username.split('@')[0] in admins_list

        if not is_admin:
            is_match_filter = f"search.ismatch('{self.username.split('@')[0]}', 'usernames_result')"

        retrieval_start_time = time.time()
        try:
            for doc in self.vectorstore.similarity_search(query, search_type="hybrid", k=NUMBER_OF_CHUNKS_TO_RETURN, filters=is_match_filter):
                cleaned_metadata = self.clean_metadata(doc)
                docs.append(Document(
                    page_content=doc.page_content,
                    metadata=cleaned_metadata))
        finally:
            retrieval_end_time = time.time()
            log_custom_metric(self.instrumentation_key, 'retriever_execution_time', retrieval_end_time - retrieval_start_time, self.session_id)

        return docs

# Example initialization
def setup_vector_store(embeddings, access_token_keyvault, access_token_search):
    SHD_SEARCH_SERVICE_ENDPOINT = config.SHD_SEARCH_SERVICE_ENDPOINT(access_token_keyvault)
    return AzureSearch(
        azure_search_endpoint=SHD_SEARCH_SERVICE_ENDPOINT,
        azure_search_key=None,
        azure_ad_access_token=access_token_search,
        index_name="index-policy",
        embedding_function=embeddings.embed_query,
        semantic_configuration_name="vector-policy-qa-semantic-configuration"
    )

vectorstore = setup_vector_store(embeddings, access_token_keyvault, access_token_search)
username = "user@example.com"
instrumentation_key = "your_instrumentation_key"
session_id = "your_session_id"

retriever = CustomUserRetrieverPolicyLCEL(
    vectorstore=vectorstore,
    username=username,
    instrumentation_key=instrumentation_key,
    session_id=session_id
)

Ensure that AzureSearch is correctly implemented and compatible with the VectorStore interface. This should resolve the attribute error and allow your custom retriever to function correctly ^[1].

To continue talking to Dosu, mention @dosu.