first

app.py

@@ -0,0 +1,55 @@
+import gradio as gr
+from utils import Bot
+
+# Define the initialization function for the bot
+def initialize_bot(openai_api_key, table_descriptions, text_documents):
+    global bot
+    bot = Bot(openai_api_key=openai_api_key, table_descriptions=table_descriptions, text_documents=text_documents, verbose=True)
+
+# Define the function to generate the bot's response
+def generate_response(question):
+    response = bot(question)
+    return response
+
+# Define the interface components
+openai_api_key_input = gr.inputs.Textbox(label="OpenAI API Key")
+table_descriptions_input = gr.inputs.Textbox(label="Table Descriptions")
+text_documents_input = gr.inputs.Textbox(label="Text Documents")
+
+submit_button = gr.outputs.Button(label="Submit")
+
+question_input = gr.inputs.Textbox(label="Ask a question")
+submit_question_button = gr.outputs.Button(label="Submit")
+
+# Define the interface layout
+input_components = [
+    openai_api_key_input,
+    table_descriptions_input,
+    text_documents_input,
+    submit_button,
+]
+
+output_components = [
+    question_input,
+    submit_question_button,
+]
+
+interface = gr.Interface(
+    initialize_bot,
+    input_components,
+    show_input=False,
+    outputs=output_components,
+    title="Chat with your Bot",
+)
+
+# Define the callback function for the submit button
+def submit_callback(openai_api_key, table_descriptions, text_documents):
+    initialize_bot(openai_api_key, table_descriptions, text_documents)
+
+# Define the callback function for the submit question button
+def submit_question_callback(question):
+    response = generate_response(question)
+    return response
+
+# Add the callbacks to the interface
+interface.launch(inline=False, submit_button=submit_callback, submit_question_button=submit_question_callback)

utils/__init__.py

@@ -0,0 +1 @@
+from .bot import Bot

utils/bot.py

@@ -0,0 +1,168 @@
+import langchain
+from langchain.agents import create_csv_agent
+from langchain.schema import HumanMessage
+from langchain.chat_models import ChatOpenAI
+from langchain.embeddings import OpenAIEmbeddings
+from langchain.vectorstores import Chroma
+from typing import List, Dict
+from langchain.agents import AgentType
+from langchain.chains.conversation.memory import ConversationBufferWindowMemory
+
+
+class Bot:
+
+    def __init__(
+            self,
+            openai_api_key: str,
+            table_descriptions: List[Dict[str, any]],
+            text_documents: List[langchain.schema.Document],
+            verbose: bool = False
+    ):
+        self.verbose = verbose
+        self.table_descriptions = table_descriptions
+
+        self.llm = ChatOpenAI(
+            openai_api_key=openai_api_key,
+            temperature=0,
+            model_name="gpt-3.5-turbo"
+        )
+
+        embeddings = OpenAIEmbeddings(openai_api_key=openai_api_key)
+        vector_store = Chroma.from_documents(text_documents, embeddings)
+        self.text_retriever = langchain.chains.RetrievalQAWithSourcesChain.from_chain_type(
+            llm=self.llm,
+            chain_type='stuff',
+            retriever=vector_store.as_retriever()
+        )
+        self.text_search_tool = langchain.agents.Tool(
+            func=self._text_search,
+            description="Use this tool when searching for text information",
+            name="search text information"
+        )
+
+    def __call__(
+            self,
+            question: str
+    ):
+        self.tools = []
+        self.tools.append(self.text_search_tool)
+        table = self._define_appropriate_table(question)
+        if table != "None of the tables":
+            number = int(table[table.find('№')+1:])
+            table_description = [x for x in self.table_descriptions if x['number'] == number][0]
+            table_path = table_description['path']
+
+            self.csv_agent = create_csv_agent(
+                llm=self.llm,
+                path=table_path,
+                verbose=self.verbose
+            )
+
+            self._init_tabular_search_tool(table_description)
+            self.tools.append(self.tabular_search_tool)
+
+        self._init_chatbot()
+        print(table)
+        response = self.agent(question)
+        return response
+
+    def _init_chatbot(self):
+
+        conversational_memory = ConversationBufferWindowMemory(
+            memory_key='chat_history',
+            k=5,
+            return_messages=True
+        )
+
+        self.agent = langchain.agents.initialize_agent(
+            agent=AgentType.CHAT_CONVERSATIONAL_REACT_DESCRIPTION,
+            tools=self.tools,
+            llm=self.llm,
+            verbose=self.verbose,
+            max_iterations=5,
+            early_stopping_method='generate',
+            memory=conversational_memory
+        )
+        sys_msg = (
+            "You are an expert summarizer and deliverer of information. "
+            "Yet, the reason you are so intelligent is that you make complex "
+            "information incredibly simple to understand. It's actually rather incredible."
+            "When users ask information you refer to the relevant tools."
+            "if one of the tools helped you with only a part of the necessary information, you must "
+            "try to find the missing information using another tool"
+            "if you can't find the information using the provided tools, you MUST "
+            "say 'I don't know'. Don't try to make up an answer."
+        )
+        prompt = self.agent.agent.create_prompt(
+            system_message=sys_msg,
+            tools=self.tools
+        )
+        self.agent.agent.llm_chain.prompt = prompt
+
+    def _text_search(
+            self,
+            query: str
+    ) -> str:
+        query = self.text_retriever.prep_inputs(query)
+        res = self.text_retriever(query)['answer']
+        return res
+
+    def _tabular_search(
+            self,
+            query: str
+    ) -> str:
+        res = self.csv_agent.run(query)
+        return res
+
+    def _init_tabular_search_tool(
+            self,
+            table_description: Dict[str, any]
+    ) -> None:
+
+        columns = table_description["columns"]
+        columns = '"' + '", "'.join(columns) + '"'
+        tittle = table_description["tittle"]
+
+        description = f"""
+            Use this tool when searching for tabular information.
+            With this tool you could get access to table.
+            This table tittle is "{tittle}" and the names of the columns in this table: {columns} 
+        """
+
+        self.tabular_search_tool = langchain.agents.Tool(
+            func=self._tabular_search,
+            description=description,
+            name="search tabular information"
+        )
+
+    def _define_appropriate_table(
+            self,
+            question: str
+    ) -> str:
+        ''' Определяет по описаниям таблиц в какой из них может содержаться ответ на во��рос.
+        Возвращает номер таблицы по шаблону "Table №1" или "None of the tables" '''
+
+        message = 'I have list of table descriptions: \n'
+        k = 0
+        for description in self.table_descriptions:
+            k += 1
+            number = description["number"]
+            columns = description["columns"]
+            columns = '"' + '", "'.join(columns) + '"'
+            tittle = description["tittle"]
+            str_description = f"""  {k}) description for Table №{number}:
+            a) table consist of columns with names: {columns}; 
+            b) table tittle: {tittle}.\n"""
+            message += str_description
+
+        question = f""" How do you think, which table can help answer the question: "{question}" .
+        Your answer MUST be specific, 
+        for example if you think that Table №2 can help answer the question, you MUST just write  "Table №2". 
+        If you think that none of the tables can help answer the question just write "None of the tables"
+        Don't include to answer information about your thinking.
+        """
+        message += question
+
+        res = self.llm([HumanMessage(content=message)])
+        return res.content[:-1]
+