custom per_user_vector_store_retriever can't get configurable fields get filled if they get defined already in other runnable

[shaojun (shaojun)]

hi.
I created a basic RunnableWithMessageHistory which has configurable fields: user_id and conversation_id, all set with is_shared to true.
Also, I need to add a per_user_vector_store_retriever in chain for RAG, the retriever itself relies on configurable field: user_id as well.

so the chain now is like:

class PerUserVectorStoreRetriever(VectorStoreRetriever):
    """A custom runnable that for retrieve data from a vector store for the given user.
    """
    user_id: str
    meta_data: Dict[str, Any]

    async def _aget_relevant_documents(
        self, query: str, *, run_manager: AsyncCallbackManagerForRetrieverRun
    ) -> List[Document]:
        if query is None or query == "":
            raise HTTPException(
                status_code=400, detail="query is required.")
        if self.user_id is None or self.user_id == "":
            raise HTTPException(
                status_code=400, detail="User id is required.")
        self.meta_data.update({"user_id": self.user_id})
        self.search_kwargs.update({"filter": self.meta_data})
        return await super()._aget_relevant_documents(query, run_manager=run_manager)

def _get_per_user_vector_store_retriever_(self, disable_configurable_user_id: bool = False) -> VectorStoreRetriever:
    """
    disable_configurable_user_id is used for avoid duplicate define of user_id field.
    """
    retriever = PerUserVectorStoreRetriever(
        # below parameters will be fill with concrete values via configurable_fields
        user_id="",
        meta_data={},
        vectorstore=PineconeVectorDbHelper().vectorstore
    )
    if not disable_configurable_user_id:
        retriever = retriever.configurable_fields(
            user_id=ConfigurableField(
                id="user_id",
                name="User ID",
                annotation=str,
                description="Current user id",
            ),
            meta_data=ConfigurableField(
                id="vector_store_meta_data",
                name="VectorStore meta data",
                annotation=dict,
                description="when retrieving, it used for filter the searching scope",
                is_shared=True,
            ))
    else:
        retriever = retriever.configurable_fields(
            meta_data=ConfigurableField(
                id="vector_store_meta_data",
                name="VectorStore meta data",
                annotation=dict,
                description="when retrieving, it used for filter the searching scope",
                is_shared=True,
            ))
    return retriever
chain = RunnablePassthrough.assign(
            context=itemgetter("human_input") |
            _get_per_user_vector_store_retriever_(disable_configurable_user_id=True) | (
                lambda x: "\n".join([f"{doc.page_content }" for doc in x]))
        ) | chain_of_RunnableWithMessageHistory

the problem is, in above code, the user_id in class PerUserVectorStoreRetriever are always empty, while user_id is working good in get_chat_history(user_id: str, conversation_id: str).

Later, by a testing, If I don't use the RunnableWithMessageHistory, then all good, the user_id and meta_data can get filled in retriever.
could you help?

[Eugene Yurtsev (eyurtsev)]

Likely need to propagate methods for configurable fields in the PerUserVectorStoreRetriever retriever

    def configurable_fields(
        self, **kwargs: AnyConfigurableField
    ) -> RunnableSerializable[Input, Output]:
        """Configure particular runnable fields at runtime.
        return self.child_runnable.configurable_fields(**kwargs)

[shaojun (shaojun)]

you mean simply add that function in my PerUserVectorStoreRetriever like this:

class PerUserVectorStoreRetriever(VectorStoreRetriever):
    """A custom runnable that for retrieve data from a vector store for the given user.
    """
    user_id: str
    meta_data: Dict[str, Any]

    async def _aget_relevant_documents(
        self, query: str, *, run_manager: AsyncCallbackManagerForRetrieverRun
    ) -> List[Document]:
        if query is None or query == "":
            raise HTTPException(
                status_code=400, detail="query is required.")
        if self.user_id is None or self.user_id == "":
            raise HTTPException(
                status_code=400, detail="User id is required.")
        self.meta_data.update({"user_id": self.user_id})
        self.search_kwargs.update({"filter": self.meta_data})
        return await super()._aget_relevant_documents(query, run_manager=run_manager)
    def configurable_fields(
            self, **kwargs: AnyConfigurableField
        ) -> RunnableSerializable[Input, Output]:
        """Configure particular runnable fields at runtime."""

        return self.child_runnable.configurable_fields(**kwargs)

but there's a exception:

Exception has occurred: AttributeError (note: full exception trace is shown but execution is paused at: _run_module_as_main)
'PerUserVectorStoreRetriever' object has no attribute 'child_runnable'

[shaojun (shaojun)]

Eugene Yurtsev (@eyurtsev) could you take a look?