import pandas as pd from ast import literal_eval from sentence_transformers import SentenceTransformer from sklearn.metrics.pairwise import cosine_similarity import numpy as np import warnings from typing import Any, Dict, List, Optional from pydantic import BaseModel, Field import os import pprint import tiktoken from tqdm import tqdm from langchain_experimental.sql import SQLDatabaseChain from langchain.callbacks.manager import CallbackManagerForChainRun from langchain.chains.llm import LLMChain from langchain.prompts.prompt import PromptTemplate from langchain.tools.sql_database.prompt import QUERY_CHECKER import pandas as pd from sqlalchemy.schema import CreateTable, CreateColumn from sqlalchemy.types import NullType from sqlalchemy import MetaData, Table, create_engine, inspect, select, text from sqlalchemy.sql.expression import func, select from langchain.callbacks.manager import CallbackManagerForChainRun from langchain.chains.base import Chain from langchain.chains.llm import LLMChain from langchain.chains.sql_database.prompt import DECIDER_PROMPT, PROMPT, SQL_PROMPTS from langchain.prompts.prompt import PromptTemplate from langchain.schema import BasePromptTemplate from langchain.schema.language_model import BaseLanguageModel from langchain.utilities.sql_database import SQLDatabase from langchain_experimental.pydantic_v1 import Extra, Field, root_validator emb_model = SentenceTransformer("all-MiniLM-L6-v2") class EmbeddingsSearch: def __init__(self, metadata_df, emb_model): self.model = emb_model self.metadata_df = metadata_df self.embeddings = self.model.encode(self.metadata_df['final_metadata'].tolist()) def __call__(self, text: str, topk: int = 5): q_emb = self.model.encode([text]) distances = cosine_similarity(q_emb, self.embeddings) idx = np.flip(distances.argsort())[0] distances.sort() distances = np.flip(distances)[0] results = pd.DataFrame() results['idx'] = idx.tolist()[:topk] results['distances'] = distances.tolist()[:topk] results['table'] = [ self.metadata_df.loc[i, "table"] for i in results['idx'] ] return results #xls = pd.ExcelFile('SmartClever table explanations updated.xlsx') #metadata_df = pd.DataFrame() #i = 0 #sheet_to_df_map = {} #for k, sheet_name in enumerate(xls.sheet_names): # if k > 0: # sheet_to_df_map[sheet_name.strip()] = xls.parse(sheet_name, header=None) # sheet_to_df_map[sheet_name.strip()].columns = sheet_to_df_map[sheet_name.strip()].iloc[1] # sheet_to_df_map[sheet_name.strip()] = sheet_to_df_map[sheet_name.strip()].iloc[:1].fillna('') # sheet_to_df_map[sheet_name.strip()]['metadata'] = sheet_to_df_map[sheet_name.strip()].apply(lambda x: \ # ". ".join([x[col] for col in sheet_to_df_map[sheet_name.strip()].columns]), axis=1) # metadata_df.loc[i, "table"] = sheet_name.strip() # metadata_df.loc[i, "desc"] = sheet_to_df_map[sheet_name.strip()]['metadata'].iloc[0] # # i += 1 #metadata_df2 = xls.parse('Table explanations',header=1).dropna(axis=0,how='all').dropna(axis=1,how='all') #metadata_df2.columns = ['table','metadata'] #metadata_df2.table = metadata_df2.table.apply(lambda x: x.strip()) #metadata_df = pd.merge(metadata_df, metadata_df2, how='inner') xls = pd.ExcelFile('SmartClever table explanations_V5.xlsx') metadata_df = pd.DataFrame() i = 0 sheet_to_df_map = {} for k, sheet_name in enumerate(xls.sheet_names): if k > 0: sheet_to_df_map[sheet_name.strip()] = xls.parse(sheet_name, header=None) sheet_to_df_map[sheet_name.strip()].columns = sheet_to_df_map[sheet_name.strip()].iloc[1] sheet_to_df_map[sheet_name.strip()] = sheet_to_df_map[sheet_name.strip()].iloc[:1].fillna('') sheet_to_df_map[sheet_name.strip()]['metadata'] = sheet_to_df_map[sheet_name.strip()].apply(lambda x: \ ". ".join([x[col] for col in sheet_to_df_map[sheet_name.strip()].columns]), axis=1) metadata_df.loc[i, "table"] = sheet_name.strip() metadata_df.loc[i, "desc"] = sheet_to_df_map[sheet_name.strip()]['metadata'].iloc[0] i += 1 metadata_df2 = xls.parse('Table explanations',header=1).dropna(axis=0,how='all').dropna(axis=1,how='all') metadata_df2.columns = ['table','nickname','metadata'] metadata_df2.table = metadata_df2.table.apply(lambda x: x.strip()) metadata_df = pd.merge(metadata_df, metadata_df2, how='inner') table_desc = pd.read_csv("table_desc.csv", lineterminator='\n') table_desc.columns = ['table','desc'] metadata_df = metadata_df.drop(['desc'], axis=1) metadata_df = pd.merge(metadata_df, table_desc, how='inner') metadata_df['final_metadata'] = metadata_df.apply(lambda x: x["desc"] + "\n" + x['metadata'], axis=1) #metadata_df.loc[metadata_df.table == 'History_All_Skus_Availability', 'table'] = 'TBL_History_All_Skus_Availability' #metadata_df.loc[metadata_df.table == 'daily_inventory', 'table'] = 'TBL_DAILY_INVENTORY' #metadata_df.loc[metadata_df.table == 'HISTORY_OpenOrderShortage', 'table'] = 'TBL_HISTORY_OpenOrderShortage' metadata_df.loc[metadata_df.table == 'daily_inventory', 'table'] = 'DAILY_INVENTORY' table_search = EmbeddingsSearch(metadata_df=metadata_df, emb_model=emb_model) def extract_question_type(llm, query): sys_prompt = """ You are an AI assistant that determines if a user provided question can be answered from the given tables. The metadata of the tables are provided here - {}. \ If the question can be answered return yes. If the question is a generic one and cannot be answered using these tables, return no. Note that any question specific to families, commodities, products, forecasts, SKUs can be related to the tables, so return yes.""".format(metadata_df[['table','metadata']].to_string()) messages = [ ("system", sys_prompt), ("human", query), ] output = llm.invoke(messages) pred = output.content return pred def extract_table_name(llm, query): messages = [ ( "system", """ You are an AI assistant that determines the most relevant table name given a user query. Following is the metadata information you need to use to determine the most relevant table.\ {}.""".format(metadata_df[['table','metadata']].to_string()), ), ("human", query), ] output = llm.invoke(messages) pred = output.content tables = [] for table in metadata_df.table.unique(): if table in pred: tables.append(table) return tables def extract_question_list(llm, query): sys_prompt = """You are an AI assistant that determines if multiple questions are stacked in a single question and split the question into sub questions and return a list of them. Make sure the response is a valid Python list. If the question is a single question and return the original question. Please do not add any additional text, only return the final response.""" messages = [ ( "system", sys_prompt, ), ("human", query), ] output = llm.invoke(messages) pred = output.content try: return literal_eval(pred) except: return query def translate_to_english(llm, user_query): sys_prompt = """ You are an AI assistant that translates a text to English. \ Do not generate any irrelavant text, only return the translation.""" messages = [ ( "system", sys_prompt, ), ("human", user_query), ] output = llm.invoke(messages) pred = output.content return pred def translate(llm, user_query, to_translate): sys_prompt = """ You are an AI assistant that determines the language of the following text - {} and translate the user provided text in that language. \ Do not generate any irrelavant text, only return the translation.""".format(user_query) messages = [ ( "system", sys_prompt, ), ("human", to_translate), ] output = llm.invoke(messages) pred = output.content return pred def num_tokens_from_string(string: str, encoding_name: str) -> int: encoding = tiktoken.get_encoding(encoding_name) num_tokens = len(encoding.encode(string)) return num_tokens def clean_sql(s: str) -> str: #s = s.replace("SQL:","").strip() #s = s.replace("Let's execute these queries step-by-step to get the final answer.","").strip() s = s.replace("```sql", "") for symb in ["'", '"']: if s.startswith(symb) and s.endswith(symb): s = s[1:-1] for symb in [";"]: if s.endswith(symb): s = s[:-1] s = s.replace("```", "") s = s.replace("\n", " ") s = s.replace("\t", " ") if "LIMIT 1" in s: s = s.replace("LIMIT 1","").strip() s = s.replace("SELECT","SELECT TOP 1") if s.endswith("TOP 1"): s = s.replace("TOP 1","").strip() s = s.replace("SELECT","SELECT TOP 1") s = s.split("SQLQuery:")[-1].strip() return s def get_metadata_info(metadata_df, table_names): str = "" for table in table_names: try: str += "The following metadata is for the table " + table + "\n" #str += metadata_df[metadata_df.table == table].final_metadata.iloc[0] str += metadata_df[metadata_df.table == table].desc.iloc[0] except: pass return str class SQLDatabaseChainPatched(SQLDatabaseChain): intermediate_steps: List[Any] = Field(default_factory=list) llms: Dict[str, Any] = Field(default_factory=dict) def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.return_intermediate_steps = True self.intermediate_steps = [] def set_llms(self, llms): self.llms = llms print("Set llms") print(self.llms) def prepare_llm(self, inputs, chain, replace_llm: bool = False): # this function is used to monkey path llm in case num tokens is above max num tokens for a small # 4k model # after llm call we need to call `revert_to_small_model` function to revert to small 4k model # get number of tokens in the input prompt selected_inputs = {k: inputs[k] for k in chain.prompt.input_variables} prompt = chain.prompt.format_prompt(**selected_inputs) #print ("==================================") #print (prompt) #print ("==================================") # https://stackoverflow.com/questions/75804599/openai-api-how-do-i-count-tokens-before-i-send-an-api-request n_tokens = num_tokens_from_string(string=prompt.text, encoding_name='cl100k_base') print(f"N tokens in input: {n_tokens}") if replace_llm: max_tokens_small_model = 8000 if n_tokens > max_tokens_small_model * 0.9: chain.llm = self.llms['16k'] print("Using large model") return chain, n_tokens def revert_to_small_model(self, chain): chain.llm = self.llms['4k'] print("Reverted model to 4k") return chain def _call( self, inputs: Dict[str, Any], run_manager: Optional[CallbackManagerForChainRun] = None, ) -> Dict[str, Any]: _run_manager = run_manager or CallbackManagerForChainRun.get_noop_manager() #print ("===============") #print ("input key", self.input_key) #print ("===============") orig_question = inputs[self.input_key] history = inputs['history'].copy() history.reverse() #inputs[self.input_key] = translate_to_english(self.llms['4k'], inputs[self.input_key]) input_text = f"{inputs[self.input_key]} \nHistory: {history} \nSQLQuery:" _run_manager.on_text(input_text, verbose=self.verbose) # If not present, then defaults to None which is all tables. table_names_to_use = inputs.get("table_names_to_use") table_info = self.database.get_table_info(table_names=table_names_to_use) table_info += get_metadata_info(metadata_df, table_names_to_use) llm_inputs = { "input": input_text, "history": history, "top_k": str(self.top_k), "dialect": self.database.dialect, "table_info": table_info, "stop": ["\nSQLResult:"], } if self.memory is not None: for k in self.memory.memory_variables: llm_inputs[k] = inputs[k] self.intermediate_steps = {} # remove table info due to large size self.intermediate_steps['llm_inputs'] = {} for k, v in llm_inputs.items(): if k not in ['table_info']: self.intermediate_steps['llm_inputs'][k] = v # list to store estimated num of tokens self.intermediate_steps['n_tokens_list'] = [] input_text_bkp = input_text try: # get sql self.llm_chain, n_tokens1 = self.prepare_llm(llm_inputs, chain=self.llm_chain) # self.intermediate_steps['n_tokens_list'].append(n_tokens1) sql_cmd = self.llm_chain.predict( callbacks=_run_manager.get_child(), **llm_inputs, ).strip() # self.llm_chain = self.revert_to_small_model(chain=self.llm_chain) self.intermediate_steps['sql_cmd_unchecked'] = sql_cmd self.intermediate_steps['sql_cmd'] = clean_sql(sql_cmd) # run sql sql_data = self.database._execute(self.intermediate_steps['sql_cmd'], fetch='all') self.intermediate_steps['sql_data'] = sql_data # provide human answer input_text += f"{sql_cmd}\nSQLResult: {str(sql_data)}\nAnswer:" llm_inputs["input"] = input_text self.llm_chain, n_tokens3 = self.prepare_llm(llm_inputs, chain=self.llm_chain) # self.intermediate_steps['n_tokens_list'].append(n_tokens3) final_result = self.llm_chain.predict( callbacks=_run_manager.get_child(), **llm_inputs, ).strip() # self.llm_chain = self.revert_to_small_model(chain=self.llm_chain) self.intermediate_steps['result'] = final_result # provide explanation input_text += f"{final_result}\nExplanation:" llm_inputs["input"] = input_text self.llm_chain, n_tokens4 = self.prepare_llm(llm_inputs, chain=self.llm_chain) # self.intermediate_steps['n_tokens_list'].append(n_tokens3) explanation = self.llm_chain.predict( callbacks=_run_manager.get_child(), **llm_inputs, ).strip() # self.llm_chain = self.revert_to_small_model(chain=self.llm_chain) self.intermediate_steps['query_explanation'] = explanation #if 'result' in self.intermediate_steps: # self.intermediate_steps['translated_result'] = translate(self.llms['4k'], orig_question, self.intermediate_steps['result']) except: try: sql_data_new = sql_data[-20:] + sql_data[:20] input_text = input_text_bkp + f"{sql_cmd}\nSQLResult: {str(sql_data_new)}\nAnswer:" llm_inputs["input"] = input_text self.llm_chain, n_tokens3 = self.prepare_llm(llm_inputs, chain=self.llm_chain) # self.intermediate_steps['n_tokens_list'].append(n_tokens3) final_result = self.llm_chain.predict( callbacks=_run_manager.get_child(), **llm_inputs, ).strip() # self.llm_chain = self.revert_to_small_model(chain=self.llm_chain) self.intermediate_steps['result'] = final_result # provide explanation input_text += f"{final_result}\nExplanation:" llm_inputs["input"] = input_text self.llm_chain, n_tokens4 = self.prepare_llm(llm_inputs, chain=self.llm_chain) # self.intermediate_steps['n_tokens_list'].append(n_tokens3) explanation = self.llm_chain.predict( callbacks=_run_manager.get_child(), **llm_inputs, ).strip() # self.llm_chain = self.revert_to_small_model(chain=self.llm_chain) self.intermediate_steps['query_explanation'] = explanation #if 'result' in self.intermediate_steps: # self.intermediate_steps['translated_result'] = translate(self.llms['4k'], orig_question, self.intermediate_steps['result']) except Exception as exc: # Append intermediate steps to exception, to aid in logging and later # improvement of few shot prompt seeds #exc.intermediate_steps = self.intermediate_steps # type: ignore #raise exc self.intermediate_steps['result'] = "I don't know the answer for this." #self.intermediate_steps['translated_result'] = "I don't know the answer for this."