Llamaindex and langchain support for Rag based chatbot

[saitej123]

Please provide integration with llamaindex and langchain

[evrardt]

+1 so we can use flowise or autogen, could not be so hard using flask or fast-api : https://anderfernandez.com/en/blog/how-to-create-api-python/

[tebohomanyeli]

I have something that works with llama-index==0.9.48, but I feel that it opens unnecessary empty chats, although it works fine.

Custom_HugChat_LLM.py

import os
from dotenv import load_dotenv
from typing import Optional, Any
from hugchat.login import Login
from hugchat.message import Message
from hugchat.hugchat import ChatBot, Model
from requests.cookies import RequestsCookieJar
from llama_index.llms.base import llm_completion_callback
from llama_index.llms import CustomLLM, CompletionResponse, CompletionResponseGen,LLMMetadata


class HugChatLLM(CustomLLM):
    chosen_llm_model: int = 0


    @property
    def metadata(self) -> LLMMetadata:
        """Get LLM metadata."""
        bot = self.setup_chatbot_in_one_step()
        metadata = self.get_model_metadata(chatbot=bot, selected_llm_model= self.chosen_llm_model)

        return LLMMetadata(
            context_window  =   metadata.get('context_window'),
            num_output      =   metadata.get('num_output'),
            model_name      =   metadata.get('model_name')
        )

    @llm_completion_callback()
    def complete(self, prompt: str, **kwargs: Any) -> CompletionResponse:

        bot = self.setup_chatbot_in_one_step()

        response = self.chat_normal(bot, prompt)
        return CompletionResponse(text=response)
    

    @llm_completion_callback()
    def stream_complete(
        self, prompt: str, **kwargs: Any
    ) -> CompletionResponseGen:
        
        bot = self.setup_chatbot_in_one_step()
        response = ""
        stream = self.chat_stream(chatbot=bot, message=prompt)
        for token in stream:
            if token == "<EOS>":  # Check for end of stream token
                break  # Exit the loop when end of stream is reached
            response += token
            yield CompletionResponse(text=response, delta=token)
 
    #---CUSTOM METHODS----------------------------------------------------------
    #---Find Logins from .env file
    def get_hugging_chat_logins(self):
        load_dotenv()
        email = os.getenv('HF_EMAIL')
        password = os.getenv('HF_PASSWORD')
        return email, password
    

    #---Connect to Api
    def connection(self,hf_email:str, hf_password:str, persist: bool = True, persist_dir: str = "./private_connection_data") -> RequestsCookieJar:
        connect = Login(hf_email,hf_password)

        if persist:
            connect.saveCookiesToDir(persist_dir)

        return connect.login()
    
    
    #---Create Chatbot
    def get_models_data(self, chatbot: ChatBot, selected_llm_model:Optional[int]=None):
        models: list[Model] = chatbot.get_available_llm_models()
        models_data = []
        for model in models:
            model_data = {
                'name': str(model.name),
                'description': str(model.description),
                'parameters': dict(model.parameters),
                'model_url': str(model.modelUrl),
                'website_url': str(model.websiteUrl),
                'preprompt': str(model.preprompt),
                'prompt_examples': list(model.promptExamples)
            }
            models_data.append(model_data)
        
        # Return based on selected_llm_model
        if selected_llm_model is not None:
            if 0 <= selected_llm_model < len(models_data):
                return models_data[selected_llm_model]
            else:
                return {'message': 'Selected LLM model index is out of range.'}   
                    
        return models_data
    

    def get_model_metadata(self, chatbot: ChatBot, selected_llm_model:int):
        # Call get_models_data to retrieve model data
        model_data = self.get_models_data(chatbot, selected_llm_model)

        if isinstance(model_data, dict) and 'message' in model_data:
            # If an error message is returned, return it as is
            return model_data

        # Extract the required data
        context_window = model_data.get('parameters', {}).get('truncate')
        num_output = model_data.get('parameters', {}).get('max_new_tokens')
        model_name = model_data.get('name')

        # Construct and return the metadata dictionary
        metadata = {
            'context_window': context_window,
            'num_output': num_output,
            'model_name': model_name
        }
        return metadata


    def create_chatbot(self, connection:RequestsCookieJar, select_llm_model:int=0,  chatbot_system_prompt="", debug:bool=False ):
        
        bot = ChatBot(cookies=connection.get_dict(),default_llm=select_llm_model, system_prompt=chatbot_system_prompt) 
        
        if debug:
            print(f"All Available Models: \n{[model['name'] for model in self.get_models_data(bot)]} \n")
            print(f"Selected Model: \n{self.get_models_data(bot)[select_llm_model]} \n")

        return bot
    

    def setup_chatbot_in_one_step(self):
        email, password = self.get_hugging_chat_logins()
        api_connection = self.connection(email, password)
        chatbot:ChatBot = self.create_chatbot(connection=api_connection, select_llm_model=0)
        return chatbot
    
    
    #---Send Message and Get Response
    def chat_normal(self, chatbot :ChatBot , message:str):
        message_object:Message = chatbot.query(message, stream=False)
        
        final_message:str = message_object.wait_until_done()
        return final_message


    def chat_stream(self, chatbot :ChatBot, message):
        for response in chatbot.query(message, stream=True):
            if response is not None:
                token = response['token']
                if token is not None:
                    yield token
            else:
                yield "<EOS>"  # Yield a special token for "end of stream"

    #-------------------------------------------------------------CUSTOM METHOD