Why does my custom Langflow Elasticsearch vectorstore report an empty variable during build?

I'm encountering an issue with a custom Elasticsearch vectorstore component in Langflow. I copied the code from the existing elasticsearch.py file into a new file and removed some unused functions for my purposes. The component appears in the UI and can be added to the flow, but while building, the part of the file custom_component.py that calls get_variable raises an error: "variable not found". Specifically, when printing the arguments passed to variable_service.get_variable, I found that the "name" argument is empty.

Below is the custom vectorstore code (with code formatting preserved):

PYTHONCopied
from typing import Any

from langchain.schema import Document
from langchain_elasticsearch import ElasticsearchStore
from loguru import logger
from elasticsearch import Elasticsearch

from langflow.base.vectorstores.model import LCVectorStoreComponent, check_cached_vector_store
from langflow.io import (
    DataInput,
    DropdownInput,
    FloatInput,
    HandleInput,
    IntInput,
    MultilineInput,
    SecretStrInput,
    StrInput,
)
from langflow.schema import Data


class ElasticsearchCustomVectorStoreComponent(LCVectorStoreComponent):
    """Elasticsearch custom vector store with with advanced, customizable search capabilities."""

    display_name: str = "Elasticsearch Custom"
    description: str = "Elasticsearch Custom Vector Store with with advanced, customizable search capabilities."
    documentation = "https://python.langchain.com/docs/integrations/vectorstores/elasticsearch"
    name = "ElasticsearchCustom"
    icon = "ElasticsearchStore"

    inputs = [
        StrInput(
            name="elasticsearch_url",
            display_name="Elasticsearch URL",
            value="http://localhost:9200",
            info="URL for self-managed Elasticsearch deployments (e.g., http://localhost:9200). "
            "Do not use with Elastic Cloud deployments, use Elastic Cloud ID instead.",
        ),
        SecretStrInput(
            name="cloud_id",
            display_name="Elastic Cloud ID",
            value="",
            info="Use this for Elastic Cloud deployments. Do not use together with 'Elasticsearch URL'.",
        ),
        StrInput(
            name="index_name",
            display_name="Index Name",
            value="langflow",
            info="The index name where the vectors will be stored in Elasticsearch cluster.",
        ),
        MultilineInput(
            name="search_input",
            display_name="Search Input",
            info="Enter a search query. Leave empty to retrieve all documents.",
        ),
        StrInput(
            name="username",
            display_name="Username",
            value="",
            advanced=False,
            info=(
                "Elasticsearch username (e.g., 'elastic'). "
                "Required for both local and Elastic Cloud setups unless API keys are used."
            ),
        ),
        SecretStrInput(
            name="password",
            display_name="Password",
            value="",
            advanced=False,
            info=(
                "Elasticsearch password for the specified user. "
                "Required for both local and Elastic Cloud setups unless API keys are used."
            ),
        ),
        HandleInput(
            name="embedding",
            display_name="Embedding",
            input_types=["Embeddings"],
        ),
        DropdownInput(
            name="search_type",
            display_name="Search Type",
            options=["similarity", "mmr"],
            value="similarity",
            advanced=True,
        ),
        IntInput(
            name="number_of_results",
            display_name="Number of Results",
            info="Number of results to return.",
            advanced=True,
            value=4,
        ),
        FloatInput(
            name="search_score_threshold",
            display_name="Search Score Threshold",
            info="Minimum similarity score threshold for search results.",
            value=0.0,
            advanced=True,
        ),
        SecretStrInput(
            name="api_key",
            display_name="Elastic API Key",
            value="",
            advanced=True,
            info="API Key for Elastic Cloud authentication. If used, 'username' and 'password' are not required.",
        ),
    ]

    @check_cached_vector_store
    def build_vector_store(self) -> ElasticsearchStore:
        print("inside build_vector_store")
        print("%"*55)
        """Builds the Elasticsearch Vector Store object."""
        if self.cloud_id and self.elasticsearch_url:
            msg = (
                "Both 'cloud_id' and 'elasticsearch_url' provided. "
                "Please use only one based on your deployment (Cloud or Local)."
            )
            raise ValueError(msg)

        es_params = {
            "index_name": self.index_name,
            "embedding": self.embedding,
            "es_user": self.username or None,
            "es_password": self.password or None,
        }

        if self.cloud_id:
            es_params["es_cloud_id"] = self.cloud_id
        else:
            es_params["es_url"] = self.elasticsearch_url

        if self.api_key:
            es_params["api_key"] = self.api_key
            
        elastic_client = Elasticsearch(hosts=[es_params["es_url"]], basic_auth=(es_params["es_user"], es_params["es_password"]), verify_certs=False)
        elasticsearch = ElasticsearchStore(index_name = es_params["index_name"], embedding=es_params["embedding"],es_connection = elastic_client) #(**es_params)

        return elasticsearch

    def search(self, query: str | None = None) -> list[dict[str, Any]]:
        """Search for similar documents in the vector store or retrieve all documents if no query is provided."""
        vector_store = self.build_vector_store()
        search_kwargs = {
            "k": self.number_of_results,
            "score_threshold": self.search_score_threshold,
        }

        if query:
            search_type = self.search_type.lower()
            if search_type not in {"similarity", "mmr"}:
                msg = f"Invalid search type: {self.search_type}"
                logger.error(msg)
                raise ValueError(msg)
            try:
                if search_type == "similarity":
                    print("%"*55)
                    results = vector_store.similarity_search_with_score(query, **search_kwargs)
                    print(results)
                    print("%"*55)
                elif search_type == "mmr":
                    results = vector_store.max_marginal_relevance_search(query, **search_kwargs)
            except Exception as e:
                msg = (
                    "Error occurred while querying the Elasticsearch VectorStore,"
                    " there is no Data into the VectorStore."
                )
                logger.exception(msg)
                raise ValueError(msg) from e
            return [
                {"page_content": doc.page_content, "metadata": doc.metadata, "score": score} for doc, score in results
            ]
        results = self.get_all_documents(vector_store, **search_kwargs)
        return [{"page_content": doc.page_content, "metadata": doc.metadata, "score": score} for doc, score in results]

    def get_all_documents(self, vector_store: ElasticsearchStore, **kwargs) -> list[tuple[Document, float]]:
        """Retrieve all documents from the vector store."""
        client = vector_store.client
        index_name = self.index_name

        query = {
            "query": {"match_all": {}},
            "size": kwargs.get("k", self.number_of_results),
        }

        response = client.search(index=index_name, body=query)

        results = []
        for hit in response["hits"]["hits"]:
            doc = Document(
                page_content=hit["_source"].get("text", ""),
                metadata=hit["_source"].get("metadata", {}),
            )
            score = hit["_score"]
            results.append((doc, score))

        return results

    def search_documents(self) -> list[Data]:
        """Search for documents in the vector store based on the search input.

        If no search input is provided, retrieve all documents.
        """
        results = self.search(self.search_input)
        retrieved_data = [
            Data(
                text=result["page_content"],
                file_path=result["metadata"].get("file_path", ""),
            )
            for result in results
        ]
        self.status = retrieved_data
        return retrieved_data

    def get_retriever_kwargs(self):
        """Get the keyword arguments for the retriever."""
        return {
            "search_type": self.search_type.lower(),
            "search_kwargs": {
                "k": self.number_of_results,
                "score_threshold": self.search_score_threshold,
            },
        }

The error appears to occur during building when this part in custom_component.py is executed, as shown below:

PYTHONCopied
  def get_variable(name: str, field: str):
            if hasattr(self, "_user_id") and not self.user_id:
                msg = f"User id is not set for {self.__class__.__name__}"
                raise ValueError(msg)
            variable_service = get_variable_service()  # Get service instance
            # Retrieve and decrypt the variable by name for the current user
            with session_scope() as session:
                user_id = self.user_id or ""
                return variable_service.get_variable(user_id=user_id, name=name, field=field, session=session)
        return get_variable

What am I missing? I've confirmed that all required input properties are properly set via the UI. Any insights on how to ensure that the proper "name" property is passed to get_variable would be much appreciated.

Python Version

3.12

Flow File

document rag.json