How to add the params to add_routes

[kientv]

I want to support multi chatbot with single langserve routes statement.
How to add prompt for each chatbot chain at runtime.
I see that a answer_chain need to create before add_routes.

answer_chain = get_chain(llm: str, chatbot_name: str, prompt: str)
add_routes(app, answer_chain, path="/chat/:bot_name", input_type=ChatRequest, config_keys=["metadata"])

[eyurtsev]

The way to do this now is via configurable runnables

chain.invoke(..., {'configurable': {'chatbot_name': 'name1'}) 

I'll expose the underlying API handler in a bit to give users more freedom to do custom things like path parameters (but with a bit more effort too)

[eyurtsev]

For reference here are configurable chain examples:

https://github.com/langchain-ai/langserve/blob/main/examples/configurable_chain/server.py

https://github.com/langchain-ai/langserve/blob/main/examples/configurable_retrieval/server.py

[kientv]

Seem that It's can not solve my problem. Because it's need to define vector store first (Hardcode).

vectorstore1 = FAISS.from_texts(
    ["cats like fish", "dogs like sticks"], embedding=OpenAIEmbeddings()
)

vectorstore2 = FAISS.from_texts(["x_n+1=a * xn * (1-xn)"], embedding=OpenAIEmbeddings())

...

    def as_retriever(self) -> BaseRetriever:
        if self.collection_name == "index1":
            return vectorstore1.as_retriever()
        elif self.collection_name == "index2":
            return vectorstore2.as_retriever()
        else:
            raise NotImplementedError(
                f"No retriever for collection {self.collection_name}"
            )

To let user create a chatbot by themselve following these steps:

1. Create chatbot_name (same collection_name for that chatbot)
2. Chatbot setting will be store in database (Fields: chatbot_name, LLM (openai, gemini,...), response_template (prompt))
3. Langserve will serve API with path look like: https://example.com/chat/:chatbot_name/...

How can I solve it in langserve?

Thank you for your support

[kientv]

For reference here are configurable chain examples:

https://github.com/langchain-ai/langserve/blob/main/examples/configurable_chain/server.py

https://github.com/langchain-ai/langserve/blob/main/examples/configurable_retrieval/server.py

There are some problems with the configuration at runtime:

1. Stream (SSE) is not working.
2. vectorstore take much time to reply question

[kientv]

@eyurtsev Could you please suggest any idea or workaround solution to solve the problem that the openai stream is not working with invoke function?

Thank you.

[eyurtsev]

Hi @kientv , sorry missed the extra questions here!

The underlying API Handler has complete flexibility:

https://github.com/langchain-ai/langserve/blob/main/examples/api_handler_examples/server.py

With that you can define path parameters.

Stream (SSE) is not working.

Will check but I suspect it's not due to run time configuration. The root cause of almost all streaming problems is that one or more runnables used in the chain do not provide support streaming.

vectorstore take much time to reply question

This is not likely to be caused by run time configuration. You'll need to provide more context for me to be able to help with response time of vectorstores.

Seem that It's can not solve my problem. Because it's need to define vector store first (Hardcode).

The attached examples are examples that show how to use configurable runnables, they need to be adapted for your use case.

What's your use case? Are you doing chat over your docs with vectorstores? Or is the goal to keep track of chat history (in which case you need a backend to store the chat history).

[kientv]

Thank you

This is my code that following your suggestion. Seem that It need to init vectorstore for every request, and not support chat stream.

class RetrieverChain(RunnableSerializable[str, List[Document]]):
    collection_name: str
    def invoke(
        self, input: str, config: Optional[RunnableConfig] = None
    ) -> List[Document]:
        
        if not chatbots[self.collection_name]:
            raise NotImplementedError(
                f"No chatbot for collection {self.collection_name}"
            )

        if self.collection_name not in ChainPool or ChainPool[self.collection_name] == None:
            llm = get_ai_chat_model(chatbots[self.collection_name]["llm"]);
            embeddings = get_embeddings(self.collection_name)
            vectorstore = UnderlyingVectorStore(self.collection_name, embeddings, 1000, 3)
            retriever = vectorstore.as_retriever()
            # Add chain to pool
            ChainPool[self.collection_name] = create_chain(
                llm, 
                retriever, 
                chatbots[self.collection_name]["response_template"]
            )
            print(f"Added chain {self.collection_name} to pool")

        return ChainPool[self.collection_name].invoke(input, config=config)

[eyurtsev]

1. Why are you using a vectorstore? What are you storing in it? -- with some context i may be able to help with some other suggestions. Generally it's OK to initialize on every request, but it's not OK to do expensive work in the initialization. Initialization is just filling in some config parameters, so should be in microseconds range. If needed connections can be cached and re-used to make things even faster to init.
2. If you're using this vanilla for streaming, instead of overriding invoke you can override .stream

class RetrieverChain(RunnableSerializable[str, List[Document]]):
    collection_name: str
    def invoke(  # <-- Change to stream
        self, input: str, config: Optional[RunnableConfig] = None
    ) -> List[Document]:
        
        if not chatbots[self.collection_name]:
            raise NotImplementedError(
                f"No chatbot for collection {self.collection_name}"
            )

        if self.collection_name not in ChainPool or ChainPool[self.collection_name] == None:
            llm = get_ai_chat_model(chatbots[self.collection_name]["llm"]);
            embeddings = get_embeddings(self.collection_name)
            vectorstore = UnderlyingVectorStore(self.collection_name, embeddings, 1000, 3)
            retriever = vectorstore.as_retriever()
            # Add chain to pool
            ChainPool[self.collection_name] = create_chain(
                llm, 
                retriever, 
                chatbots[self.collection_name]["response_template"]
            )
            print(f"Added chain {self.collection_name} to pool")

        return ChainPool[self.collection_name].invoke(input, config=config) # <-- Change to stream

[kientv]

Thanks, but It's not working after changing 'invoke' to 'stream':

TypeError: Can't instantiate abstract class RetrieverChain with abstract method invoke

[eyurtsev]

It's an abstract method so you need to leave some implementation in place even if it's not used:

class RetrieverChain(RunnableSerializable[str, List[Document]]):
    collection_name: str
    def invoke(  
        self, input: str, config: Optional[RunnableConfig] = None
    ) -> List[Document]:
      # You can replace this with an actual implementation once you confirm that you can get streaming to work
      raise NotImplementedError('This is not implemented to make sure that `asream` is invoked.')
        

    async def astream( # <-- This needs to be async to work with langserve, since the server is async by default
        self, input: str, config: Optional[RunnableConfig] = None
    ) -> AsyncIterator[?? whatever output type from the chain is ??]:
        
        if not chatbots[self.collection_name]:
            raise NotImplementedError(
                f"No chatbot for collection {self.collection_name}"
            )

        if self.collection_name not in ChainPool or ChainPool[self.collection_name] == None:
            llm = get_ai_chat_model(chatbots[self.collection_name]["llm"]);
            embeddings = get_embeddings(self.collection_name)
            vectorstore = UnderlyingVectorStore(self.collection_name, embeddings, 1000, 3)
            retriever = vectorstore.as_retriever()
            # Add chain to pool
            ChainPool[self.collection_name] = create_chain(
                llm, 
                retriever, 
                chatbots[self.collection_name]["response_template"]
            )
            print(f"Added chain {self.collection_name} to pool")

         async for chunk in ChainPool[self.collection_name].astream(input, config=config):
            yield chunk

This will only work as long as your chain itself supports astream

[kientv]

Thank you

astream work well.

The underlying API Handler has complete flexibility:

https://github.com/langchain-ai/langserve/blob/main/examples/api_handler_examples/server.py

ImportError: cannot import name 'APIHandler' from 'langserve'

[eyurtsev]

Update langserve to a more recent version to get API handler