Error in custom query

[sumitbindra]

I have this custom query functions.

# This function converts the survey yaml file to a vector store that can be used as context with the SQL

class LoadQuestions(CustomQueryComponent):

  def __init__(self):
      super().__init__()
      self.__dict__['_index'] = None
      self.__dict__['_dir_name'] = None
  
  
  def _run_component(self, **kwargs) -> Dict[str, Any]:
      query_str = kwargs.get('query_str')
      table_name = kwargs.get('table_name')
      if query_str is None:
          raise ValueError("query_str cannot be None")
  
      self._dir_name = table_name.replace("_responses", "") if table_name else "default"
      PERSIST_DIR = f"./surveys/{self._dir_name}/RAGFiles/quesindex"
  
      try:
          if not os.path.exists(PERSIST_DIR):
              reader = SimpleDirectoryReader(f"./surveys/{self._dir_name}")
              documents = reader.load_data()
              self._index = VectorStoreIndex.from_documents(documents)
              self._index.storage_context.persist(persist_dir=PERSIST_DIR)
          else:
              storage_context = StorageContext.from_defaults(persist_dir=PERSIST_DIR)
              self._index = cast(VectorStoreIndex, load_index_from_storage(storage_context))
  
          # Retrieve relevant nodes for context
          vector_retriever = self._index.as_retriever(similarity_top_k=3)
          relevant_nodes = vector_retriever.retrieve(query_str)
  
          question_context = ""
          if len(relevant_nodes) > 0:
              question_context = "\nHere are questions that are relevant to this query\n"
              for node in relevant_nodes:
                  question_context += str(node.get_content()) + "\n"
  
          return {"quesindex": question_context}
  
      except Exception as e:
          print(f"Error in LoadQuestions: {str(e)}")
          return {"quesindex": ""}
  
  def get_query_engine(self):
      if self._index is None:
          raise ValueError("Index has not been initialized. Run _run_component first.")
  
      class QuestionQueryEngine(CustomQueryEngine):
          def __init__(self, _index):
              self._index = _index
  
          def custom_query(self, query_str: str):
              # Retrieve relevant nodes
              retriever = self._index.as_retriever(similarity_top_k=3)
              nodes = retriever.retrieve(query_str)
  
              # Prepare context from retrieved nodes
              context = "\n".join([node.get_content() for node in nodes])
  
              # Prepare prompt for the LLM
              prompt = f"Based on the following survey questions:\n\n{context}\n\nPlease answer the following question: {query_str}"
  
              # Query LLM
              #llm = OpenAI(temperature=0, model="gpt-3.5-turbo")
              llm_response = llm.complete(prompt)
  
              # Create and return Response object
              return str(llm_response)
  
      return QuestionQueryEngine(self._index)
  
  @property
  def _input_keys(self):
      return {"query_str", "table_name"}
  
  @property
  def _optional_input_keys(self):
      return set()
  
  @property
  def _output_keys(self):
      return {"quesindex"}

Why am I getting the error:

File "/home/runner/surveyanalysis/llama.py", line 141, in _run_component
self._dir_name = table_name.replace("_responses", "") if table_name else "default"
File "/home/runner/surveyanalysis/.pythonlibs/lib/python3.10/site-packages/pydantic/v1/main.py", line 357, in setattr
raise ValueError(f'"{self.class.name}" object has no field "{name}"')
ValueError: "LoadQuestions" object has no field "_dir_name"

[dosubot[bot]]

The error you're encountering is due to the fact that the LoadQuestions class does not have a field named _dir_name when you try to set it. This is because Pydantic models (which CustomQueryComponent likely inherits from) enforce strict field definitions.

To resolve this, you need to define _dir_name and _index as fields in the LoadQuestions class. You can do this by using Pydantic's Field class or by defining them in the class body. Here's how you can modify your class:

from pydantic import Field
from typing import Any, Dict, Optional, cast
import os

class LoadQuestions(CustomQueryComponent):
    _index: Optional[VectorStoreIndex] = Field(default=None, exclude=True)
    _dir_name: Optional[str] = Field(default=None, exclude=True)

    def __init__(self):
        super().__init__()

    def _run_component(self, **kwargs) -> Dict[str, Any]:
        query_str = kwargs.get('query_str')
        table_name = kwargs.get('table_name')
        if query_str is None:
            raise ValueError("query_str cannot be None")

        self._dir_name = table_name.replace("_responses", "") if table_name else "default"
        PERSIST_DIR = f"./surveys/{self._dir_name}/RAGFiles/quesindex"

        try:
            if not os.path.exists(PERSIST_DIR):
                reader = SimpleDirectoryReader(f"./surveys/{self._dir_name}")
                documents = reader.load_data()
                self._index = VectorStoreIndex.from_documents(documents)
                self._index.storage_context.persist(persist_dir=PERSIST_DIR)
            else:
                storage_context = StorageContext.from_defaults(persist_dir=PERSIST_DIR)
                self._index = cast(VectorStoreIndex, load_index_from_storage(storage_context))

            # Retrieve relevant nodes for context
            vector_retriever = self._index.as_retriever(similarity_top_k=3)
            relevant_nodes = vector_retriever.retrieve(query_str)

            question_context = ""
            if len(relevant_nodes) > 0:
                question_context = "\nHere are questions that are relevant to this query\n"
                for node in relevant_nodes:
                    question_context += str(node.get_content()) + "\n"

            return {"quesindex": question_context}

        except Exception as e:
            print(f"Error in LoadQuestions: {str(e)}")
            return {"quesindex": ""}

    def get_query_engine(self):
        if self._index is None:
            raise ValueError("Index has not been initialized. Run _run_component first.")

        class QuestionQueryEngine(CustomQueryEngine):
            def __init__(self, _index):
                self._index = _index

            def custom_query(self, query_str: str):
                # Retrieve relevant nodes
                retriever = self._index.as_retriever(similarity_top_k=3)
                nodes = retriever.retrieve(query_str)

                # Prepare context from retrieved nodes
                context = "\n".join([node.get_content() for node in nodes])

                # Prepare prompt for the LLM
                prompt = f"Based on the following survey questions:\n\n{context}\n\nPlease answer the following question: {query_str}"

                # Query LLM
                #llm = OpenAI(temperature=0, model="gpt-3.5-turbo")
                llm_response = llm.complete(prompt)

                # Create and return Response object
                return str(llm_response)

        return QuestionQueryEngine(self._index)

    @property
    def _input_keys(self):
        return {"query_str", "table_name"}

    @property
    def _optional_input_keys(self):
        return set()

    @property
    def _output_keys(self):
        return {"quesindex"}

By defining _index and _dir_name as fields in the class, you ensure that they are recognized by Pydantic and can be set without raising an error. This should resolve the ValueError you are encountering ^[1].

To continue talking to Dosu, mention @dosu.

[sumitbindra]

I am trying to access it like this in a function:

# Load questions index
   ques_retriever = LoadQuestions()
   ques_retriever._run_component(query_str=query_str, table_name=table_name)

# Now you can get the custom query engine
   custom_query_engine = ques_retriever.get_query_engine()

# Use the custom query engine
       response = custom_query_engine.query(query_str)

I made the changes you suggested but still getting the error

[dosubot[bot]]

The error you're encountering might be due to the way Pydantic handles attribute setting. Instead of using Pydantic's Field, you can define _index and _dir_name as regular class attributes. Here's an updated version of your class:

from typing import Any, Dict, Optional, cast
import os

class LoadQuestions(CustomQueryComponent):
    def __init__(self):
        super().__init__()
        self._index: Optional[VectorStoreIndex] = None
        self._dir_name: Optional[str] = None

    def _run_component(self, **kwargs) -> Dict[str, Any]:
        query_str = kwargs.get('query_str')
        table_name = kwargs.get('table_name')
        if query_str is None:
            raise ValueError("query_str cannot be None")

        self._dir_name = table_name.replace("_responses", "") if table_name else "default"
        PERSIST_DIR = f"./surveys/{self._dir_name}/RAGFiles/quesindex"

        try:
            if not os.path.exists(PERSIST_DIR):
                reader = SimpleDirectoryReader(f"./surveys/{self._dir_name}")
                documents = reader.load_data()
                self._index = VectorStoreIndex.from_documents(documents)
                self._index.storage_context.persist(persist_dir=PERSIST_DIR)
            else:
                storage_context = StorageContext.from_defaults(persist_dir=PERSIST_DIR)
                self._index = cast(VectorStoreIndex, load_index_from_storage(storage_context))

            # Retrieve relevant nodes for context
            vector_retriever = self._index.as_retriever(similarity_top_k=3)
            relevant_nodes = vector_retriever.retrieve(query_str)

            question_context = ""
            if len(relevant_nodes) > 0:
                question_context = "\nHere are questions that are relevant to this query\n"
                for node in relevant_nodes:
                    question_context += str(node.get_content()) + "\n"

            return {"quesindex": question_context}

        except Exception as e:
            print(f"Error in LoadQuestions: {str(e)}")
            return {"quesindex": ""}

    def get_query_engine(self):
        if self._index is None:
            raise ValueError("Index has not been initialized. Run _run_component first.")

        class QuestionQueryEngine(CustomQueryEngine):
            def __init__(self, _index):
                self._index = _index

            def custom_query(self, query_str: str):
                # Retrieve relevant nodes
                retriever = self._index.as_retriever(similarity_top_k=3)
                nodes = retriever.retrieve(query_str)

                # Prepare context from retrieved nodes
                context = "\n".join([node.get_content() for node in nodes])

                # Prepare prompt for the LLM
                prompt = f"Based on the following survey questions:\n\n{context}\n\nPlease answer the following question: {query_str}"

                # Query LLM
                #llm = OpenAI(temperature=0, model="gpt-3.5-turbo")
                llm_response = llm.complete(prompt)

                # Create and return Response object
                return str(llm_response)

        return QuestionQueryEngine(self._index)

    @property
    def _input_keys(self):
        return {"query_str", "table_name"}

    @property
    def _optional_input_keys(self):
        return set()

    @property
    def _output_keys(self):
        return {"quesindex"}

This should resolve the ValueError you are encountering. If the error persists, please ensure that the CustomQueryComponent class does not have any additional constraints or requirements that might be causing the issue.

To continue talking to Dosu, mention @dosu.

[sumitbindra]

here is how custom query engine is implemented in llama index. Do you see a reson for the error I continue to get?

"""Custom query engine."""

from abc import abstractmethod
from typing import Union

from llama_index.core.base.base_query_engine import BaseQueryEngine
from llama_index.core.base.response.schema import RESPONSE_TYPE, Response
from llama_index.core.bridge.pydantic import BaseModel, Field, ConfigDict
from llama_index.core.callbacks.base import CallbackManager
from llama_index.core.prompts.mixin import PromptMixinType
from llama_index.core.schema import QueryBundle, QueryType

STR_OR_RESPONSE_TYPE = Union[RESPONSE_TYPE, str]


class CustomQueryEngine(BaseModel, BaseQueryEngine):
    """Custom query engine.

    Subclasses can define additional attributes as Pydantic fields.
    Subclasses must implement the `custom_query` method, which takes a query string
    and returns either a Response object or a string as output.

    They can optionally implement the `acustom_query` method for async support.

    """

    model_config = ConfigDict(arbitrary_types_allowed=True)
    callback_manager: CallbackManager = Field(
        default_factory=lambda: CallbackManager([]), exclude=True
    )

    def _get_prompt_modules(self) -> PromptMixinType:
        """Get prompt sub-modules."""
        return {}

    def query(self, str_or_query_bundle: QueryType) -> RESPONSE_TYPE:
        with self.callback_manager.as_trace("query"):
            # if query bundle, just run the query
            if isinstance(str_or_query_bundle, QueryBundle):
                query_str = str_or_query_bundle.query_str
            else:
                query_str = str_or_query_bundle
            raw_response = self.custom_query(query_str)
            return (
                Response(raw_response)
                if isinstance(raw_response, str)
                else raw_response
            )

    async def aquery(self, str_or_query_bundle: QueryType) -> RESPONSE_TYPE:
        with self.callback_manager.as_trace("query"):
            if isinstance(str_or_query_bundle, QueryBundle):
                query_str = str_or_query_bundle.query_str
            else:
                query_str = str_or_query_bundle
            raw_response = await self.acustom_query(query_str)
            return (
                Response(raw_response)
                if isinstance(raw_response, str)
                else raw_response
            )

    @abstractmethod
    def custom_query(self, query_str: str) -> STR_OR_RESPONSE_TYPE:
        """Run a custom query."""

    async def acustom_query(self, query_str: str) -> STR_OR_RESPONSE_TYPE:
        """Run a custom query asynchronously."""
        # by default, just run the synchronous version
        return self.custom_query(query_str)

    def _query(self, query_bundle: QueryBundle) -> RESPONSE_TYPE:
        raise NotImplementedError("This query engine does not support _query.")

    async def _aquery(self, query_bundle: QueryBundle) -> RESPONSE_TYPE:
        raise NotImplementedError("This query engine does not support _aquery.")

[dosubot[bot]]

The ValueError you are encountering is likely due to the way Pydantic handles attribute setting in the CustomQueryComponent class, which LoadQuestions inherits from. Pydantic models enforce strict field definitions, and attributes not defined as fields cannot be set dynamically.

To resolve this, you should define _index and _dir_name as regular class attributes instead of using Pydantic's Field. Here is the updated version of your LoadQuestions class:

from typing import Any, Dict, Optional, cast
import os

class LoadQuestions(CustomQueryComponent):
    def __init__(self):
        super().__init__()
        self._index: Optional[VectorStoreIndex] = None
        self._dir_name: Optional[str] = None

    def _run_component(self, **kwargs) -> Dict[str, Any]:
        query_str = kwargs.get('query_str')
        table_name = kwargs.get('table_name')
        if query_str is None:
            raise ValueError("query_str cannot be None")

        self._dir_name = table_name.replace("_responses", "") if table_name else "default"
        PERSIST_DIR = f"./surveys/{self._dir_name}/RAGFiles/quesindex"

        try:
            if not os.path.exists(PERSIST_DIR):
                reader = SimpleDirectoryReader(f"./surveys/{self._dir_name}")
                documents = reader.load_data()
                self._index = VectorStoreIndex.from_documents(documents)
                self._index.storage_context.persist(persist_dir=PERSIST_DIR)
            else:
                storage_context = StorageContext.from_defaults(persist_dir=PERSIST_DIR)
                self._index = cast(VectorStoreIndex, load_index_from_storage(storage_context))

            # Retrieve relevant nodes for context
            vector_retriever = self._index.as_retriever(similarity_top_k=3)
            relevant_nodes = vector_retriever.retrieve(query_str)

            question_context = ""
            if len(relevant_nodes) > 0:
                question_context = "\nHere are questions that are relevant to this query\n"
                for node in relevant_nodes:
                    question_context += str(node.get_content()) + "\n"

            return {"quesindex": question_context}

        except Exception as e:
            print(f"Error in LoadQuestions: {str(e)}")
            return {"quesindex": ""}

    def get_query_engine(self):
        if self._index is None:
            raise ValueError("Index has not been initialized. Run _run_component first.")

        class QuestionQueryEngine(CustomQueryEngine):
            def __init__(self, _index):
                self._index = _index

            def custom_query(self, query_str: str):
                # Retrieve relevant nodes
                retriever = self._index.as_retriever(similarity_top_k=3)
                nodes = retriever.retrieve(query_str)

                # Prepare context from retrieved nodes
                context = "\n".join([node.get_content() for node in nodes])

                # Prepare prompt for the LLM
                prompt = f"Based on the following survey questions:\n\n{context}\n\nPlease answer the following question: {query_str}"

                # Query LLM
                #llm = OpenAI(temperature=0, model="gpt-3.5-turbo")
                llm_response = llm.complete(prompt)

                # Create and return Response object
                return str(llm_response)

        return QuestionQueryEngine(self._index)

    @property
    def _input_keys(self):
        return {"query_str", "table_name"}

    @property
    def _optional_input_keys(self):
        return set()

    @property
    def _output_keys(self):
        return {"quesindex"}

This should resolve the ValueError you are encountering. If the error persists, ensure that the CustomQueryComponent class does not have any additional constraints or requirements that might be causing the issue.

To continue talking to Dosu, mention @dosu.