Support to Custom Chains

[rogalvil]

I have a custom chain for Conversational Retrieval With Source Chain because currently not exist support on langchian

class ConversationalRetrievalWithSourcesChain(ConversationalRetrievalChain):
    """Chain for chatting with sources over documents."""

    sources_output_key: str = "sources"  #: :meta private:

    @property
    def output_keys(self) -> List[str]:
        """Return the output keys.

        :meta private:
        """
        _output_keys = [self.output_key]
        _output_keys.append(self.sources_output_key)
        if self.return_source_documents:
            _output_keys = _output_keys + ["source_documents"]
        return _output_keys

    def _call(
        self,
        inputs: Dict[str, Any],
        run_manager: Optional[CallbackManagerForChainRun] = None,
    ) -> Dict[str, Any]:
        result = super()._call(
            inputs=inputs,
            run_manager=run_manager,
        )
        answer = result[self.output_key]
        if re.search(r"SOURCES:\s", answer):
            answer, sources = re.split(r"SOURCES:\s", answer)
        else:
            sources = ""
        result[self.output_key] = answer
        result[self.sources_output_key] = sources
        return result

    async def _acall(
        self,
        inputs: Dict[str, Any],
        run_manager: Optional[AsyncCallbackManagerForChainRun] = None,
    ) -> Dict[str, Any]:
        result = await super()._acall(
            inputs=inputs,
            run_manager=run_manager,
        )
        answer = result[self.output_key]
        if re.search(r"SOURCES:\s", answer):
            answer, sources = re.split(r"SOURCES:\s", answer)
        else:
            sources = ""
        result[self.output_key] = answer
        result[self.sources_output_key] = sources
        return result

I love a new feat: auto-detect callback

But now I receive a this kind of errors

Any idea to include my custom chain on auto-detect callback

[ajndkr]

glad you like the new feature!

We really need a docs page for this project 😅

I see 2 approaches here:

Approach 1:

You define a new callback handler called AsyncConversationalRetrievalWithSourcesChainStreamingCallback and register it as follows:

@register_streaming_callback("ConversationalRetrievalWithSourcesChain")
class AsyncConversationalRetrievalWithSourcesChainStreamingCallback(
    AsyncConversationalRetrievalChainStreamingCallback
):
    """AsyncStreamingResponseCallback handler for ConversationalRetrievalWithSourcesChain."""

    pass

Approach 2:

We update fastapi_async_langchain to auto-detect using chain class name (as we do now) but if the chain type is not supported, we search for the inherit chain class name. This could work for your use-case but I sense there might be bugs down this road as some custom chains will simply inherit from Chain and it might break the application.

I think approach 1 is a better solution. Try it out and let me know how it goes!

[ajndkr]

@rogalvil fyi: https://lanarky.readthedocs.io/en/latest/advanced/custom_callbacks.html

feel free to make suggestions via pr.

[rogalvil]

@ajndkr Cool, thanks