utils.py

# important classes
import json
import hnswlib
import numpy as np
from openai import OpenAI
from sentence_transformers import SentenceTransformer, CrossEncoder
import streamlit as st
# It formats the output by printing it in color.

# Defined arguments

model_name_tr = "gpt-3.5-turbo"  # model name
model_name_ta = "gpt-4-1106-preview"
tool_list_path = './tools.json' # list of tools path[ ]
example_path = './examples.json'  # list of examples path
zero_shot = 0
no_of_examples = 2

# ---------------------------------------------------some constants ----------------------------------------------------------------------

EF = 100  # EF
K = 3  # top k number
COSINE_THRESHOLD = 0.3  # cosine threshold
biencoder = SentenceTransformer("BAAI/bge-large-en-v1.5", device="cpu")
cross_encoder = CrossEncoder(
    "cross-encoder/ms-marco-MiniLM-L-12-v2", max_length=512, device="cpu"
)
client = OpenAI(api_key=st.secrets["API_KEY"], timeout=60, max_retries=2)

class color:
    PURPLE = "\033[1;35;48m"
    CYAN = "\033[1;36;48m"
    BOLD = "\033[1;37;48m"
    BLUE = "\033[1;34;48m"
    GREEN = "\033[1;32;48m"
    YELLOW = "\033[1;33;48m"
    RED = "\033[1;31;48m"
    BLACK = "\033[1;30;48m"
    UNDERLINE = "\033[4;37;48m"
    END = "\033[1;37;0m"

class Tools:
    '''
      A class to represent tools.
      ...
      Attributes
      ----------
      - tools: list
        a list of tools (json format as described above)

      - examples: list
        a list of examples (json format as described above)

      Methods
      -------
      - check_json()
          checks whether the tool added is in the defined JSON schema

      - build_index()
          assigns index values to the examples and their respective tool calls to create an index list.

      - add_tool()
          adds new tool to the index list.

      - add_example()
          adds the new example in the example pool and modifies the index list.

      - modify_example()
          modifies an existing example present in the example list.

      - update_tool()
          updates an exisiting tool in the tool list.

      - replenish_examples()
          on deletion of tool replenish the examples for the tools where the num_examples < threshold

      - delete_tool()
          deletes the tool from the index list and example list.

      - similarity_retriever()
          creates the hsnw index for the example queries

    '''
    def __init__(self, tools, examples):
      # assuming the self.tools to be a dict of tools with tool name as the key and tool info as the value.
      self.tools = tools
      self.examples = examples
      self.index = {}
      self.build_index()
      self.th = 2
      self.query_embeddings = []
      self.queries = []
      self.search_index = None
      self.similarity_retriever()

    def check_json(self,tool_json):
      keys = ['argument_list', 'title', 'tool_description', 'tool_name']
      argument_keys = ['argument_description', 'argument_name', 'argument_type', 'example']
      if type(tool_json) != type({}):
        raise Exception('Given tool json is not a dictionary')
      if keys != sorted(list(tool_json.keys())):
        raise Exception(
            """Keys don't match
              Expected Keys: {}
              Given Keys: {}
            """.format(keys, sorted(list(tool_json.keys())))
        )
      if type(tool_json['argument_list']) != type([]):
          raise Exception('Given argument list is not a list')
      for idx, arg in enumerate(tool_json['argument_list']):
        if type(arg) != type({}):
          raise Exception(f'Argument at the index: {idx} is not a dictionary')
        if argument_keys != sorted(list(arg.keys())):
          raise Exception(
            """Keys don't match at {}
              Expected Keys: {}
              Given Keys: {}
            """.format(idx, argument_keys, sorted(list(arg.keys())))
          )

    def build_index(self):
        '''
        Indexes all the examples with their respective tool calls.
        For example: {'<tool_name>': {'num_examples': <no. of examples containing tool_name>, 'indices': <list of example indices containing respective tool_name>}}
        Parameters
        ----------
          None

        Returns
        ----------
          None

        Modifies
        ----------
          self.index
        '''
        for tool in self.tools:
            self.index[tool] = {'num_examples': 0, 'indices': []}
        for i, example in enumerate(self.examples):
            for tool in example['answer']:
                self.index[tool['tool_name']]['num_examples'] += 1
                self.index[tool['tool_name']]['indices'].append(i)

    def add_tool(self, tool):
        '''
        On addition of any new tool, this function gets called to add the new tool in self.index.

        Parameters
        ----------
          - tool : json (format described above)

        Returns
        ----------
          None

        Modifies
        ----------
          self.index
        '''
        # self.check_json(tool)
        tool_name = tool['tool_name']
        if tool_name in self.tools.keys():
          if tool==self.tools[tool_name]:
            print(color.YELLOW+'Already Exists!'+color.END)
          else:
            self.update_tool(tool)
            print(color.YELLOW+f"[WARNING] You tried adding a tool that already exists, so updating the tool '{tool_name}'"+color.END)
        else:
          self.tools[tool_name] = tool
          self.index[tool_name] = {'num_examples': 0, 'indices': []}
          self.add_example(tool, self.th)
          print(color.GREEN+f"[SUCCESS] Added the tool '{tool_name}'"+color.END)

    def add_example(self, tool, no_of_eg):
        '''
        On addition of new tool or replenishing examples on deletion, this function gets called to add the new example in self.examples and modifies index list accordingly.

        Parameters
        ----------
          - tool : json (format described above)
          - no_of_eg : int

        Returns
        ----------
          None

        Modifies
        ----------
          self.examples
          self.index
        '''
        tool_name = tool['tool_name']
        try:
          ex = self.examples[:3]
        except:
          ex = self.examples
        message = create_prompt_for_new_example(list(self.tools.values()), tool, ex, n=no_of_eg)
        res = client_conn(message, model_name_tr)
        res = res.choices[0].message.content
        try:
          new_examples = json.loads(res)
        except:
          new_examples = get_parsed_json(res)
        for example in new_examples:
          tool_calls = example['answer']
          for tool in tool_calls:
            new_tool_name = tool['tool_name']
            self.index[new_tool_name]['num_examples'] += 1
            self.index[new_tool_name]['indices'].append(len(self.examples))
          self.examples.append(example)
        self.similarity_retriever()

    def modify_example(self, tool):
        '''
        On modification of a tool, this function gets called to modify the examples where the tool is used.

        Parameters
        ----------
          None

        Returns
        ----------
          None

        Modifies
        ----------
          self.examples
        '''
        tool_name = tool["tool_name"]
        indices = self.index[tool_name]["indices"]
        relevant_examples = []
        for ind in indices:
            relevant_examples.append(self.examples[ind])
        message = create_prompt_for_modified_example(
            list(self.tools.values()), tool, relevant_examples
        )
        res = client_conn(message, model_name_tr)
        res = res.choices[0].message.content
        try:
          examples = json.loads(res)
        except:
          examples = get_parsed_json(res)
        for i in range(len(examples)):
            self.examples[indices[i]] = examples[i]

        print(color.GREEN + f"[SUCCESS] Modified Examples!" + color.END)
        self.similarity_retriever()

    def update_tool(self, tool):
        '''
        On modification of a tool, this function gets called update the tool.

        Parameters
        ----------
          tool : json (format described above)

        Returns
        -------
          None

        Modifies
        --------
          self.tools
        '''
        # self.check_json(tool)
        tool_name = tool['tool_name']
        if tool_name in self.tools.keys():
          self.modify_example(tool)
          self.tools[tool_name] = tool
          print(color.GREEN+f"[SUCCESS] Updated the tool '{tool_name}'"+color.END)
        else:
          print(color.YELLOW+f"[WARNING] You tried updating a tool that does not exist, so added the tool '{tool_name}'"+color.END)
          self.add_tool(tool)

    def replenish_examples(self):
        '''
        On deletion of any new tool if number of examples for other tools goes below threshold, this function gets called to create examples.

        Parameters
        ----------
          None

        Returns
        -------
          None

        Modifies
        --------
          None
        '''
        for i,tool in enumerate(self.index):
          if ((self.index[tool]['num_examples']<self.th) and (len(self.tools)>4)):
            self.add_example(self.tools[tool], self.th-self.index[tool]['num_examples'])

    def delete_tool(self, tool_name):
        '''
        On deletion of any new tool, this function gets called to delete the new tool in self.index and remove the respective examples.

        Parameters
        ----------
          - tool_name : str

        Returns
        -------
          None

        Modifies
        --------
          self.index
          self.examples
        '''
        if tool_name not in self.tools.keys():
          print(color.RED+"[ERROR] You tried deleting a tool that does not exist."+color.END)
        else:
          indices = self.index[tool_name]['indices']
          del self.index[tool_name]
          for idx, tool in enumerate(list(self.tools.values())):
              if tool['tool_name'] == tool_name:
                  del self.tools[tool_name]
                  break
          if len(indices) != 0:
              for idx, index in enumerate(indices):
                  del self.examples[index-idx]
              self.index = {}
              self.build_index()
          print(color.GREEN+f"[SUCCESS] Deleted the tool '{tool_name}'"+color.END)
          self.replenish_examples()
          self.similarity_retriever()

    def similarity_retriever(self):
        '''
        On addition of any new example, this function gets called to create the new hnsw index for queries.

        Parameters
        ----------
          None

        Returns
        -------
          None

        Modifies
        --------
          self.query_embeddings
          self.queries
          self.search_index
        '''
        self.query_embeddings, self.queries = create_example_query_embeddings(self.examples)
        self.search_index = create_hnsw_index(np.array(self.query_embeddings))

# ------------------------------------------important functions------------------------------------------------------------


def read_file(path):
    """
    parameters
    ----------
    - path: str
        path of the file to read in json.

    returns
    ---------
    - file: object
        the json file object.
    """

    with open(path, "r") as f:
        file = json.load(f)
    return file


def create_query_embedding(query):
    """
    Encodes the query to get its embedding.

    parameters
    ---------
    - query: str

    returns
    ---------
    - embedding: numpy array
      embedding of the query.

    """
    embedding = biencoder.encode([query], normalize_embeddings=True)[0]
    return embedding


def create_example_query_embeddings(examples):
    """
    creates query embeddings and saves it.

      parameters
      ----------
      - examples: list
          List of examples in json format.

      returns
      ---------
      - query_embeddings: list
          a list of query embeddings.

      - answers: list
          a list of answers corresponding to each query.

      - queries: list
          a list of queries.

    """
    query_embeddings = []
    answers = []
    queries = []
    for example in examples:
        query = example["query"]
        answer = example["answer"]
        queries.append(query)
        query_embedding = create_query_embedding(query)
        query_embeddings.append(query_embedding)
        answers.append(answer)
    np.save("query_embeddings.npy", query_embeddings)
    return query_embeddings, queries


def create_hnsw_index(embedding, M=16, efC=100):
    """
    creates the HNSW index.

    parameters
    ----------
    - embedding: list
      query embedding

    - M: int
      default = 16

    - efc: int
      default = 100

    returns
    ----------
    - index: object
    """
    embeddings = embedding
    num_dim = embeddings.shape[1]
    ids = np.arange(embeddings.shape[0])
    index = hnswlib.Index(space="ip", dim=num_dim)
    index.init_index(max_elements=embeddings.shape[0], ef_construction=efC, M=M)
    index.add_items(embeddings, ids)
    return index


def find_nearest_neighbors(query_embedding, queries, search_index):
    """
    Finds the k nearest neighbors using cosine similarity.
    parameters
    ----------
    - query_embedding: list
      the query embedding whose similarity check has to be made.

    - queries: list
      a list of queries in the examples.

    - search_index: object

    returns
    ----------
    - query_list: list
      a list of similar queries.
    """
    search_index.set_ef(EF)
    labels, distances = search_index.knn_query(
        query_embedding, k=K
    )  # Find the k-nearest neighbors for the query embedding
    labels = [
        label
        for label, distance in zip(labels[0], distances[0])
        if (1 - distance) >= COSINE_THRESHOLD
    ]
    query_list = [queries[i] for i in labels]
    return query_list


def rerank_queries_with_cross_encoder(query, chunks):
    """
    Sorts the chunks based on their scores in descending order.
    parameters
    ---------
      - query: str
      - chunks: list
          the list of similar chunks
    returns
    ---------
      - sorted_chunks: list
          the response after reranking.
    """
    pairs = [(query, chunk) for chunk in chunks]
    scores = cross_encoder.predict(pairs)
    sorted_chunks = [chunk for _, chunk in sorted(zip(scores, chunks), reverse=True)]
    return sorted_chunks


def rerank_queries_with_cross_encoder(query, chunks):
    """
    Sorts the chunks based on their scores in descending order.
    parameters
    ---------
      - query: str
      - chunks: list
          the list of similar chunks
    returns
    ---------
      - sorted_chunks: list
          the response after reranking.
    """
    pairs = [(query, chunk) for chunk in chunks]
    scores = cross_encoder.predict(pairs)
    sorted_chunks = [chunk for _, chunk in sorted(zip(scores, chunks), reverse=True)]
    return sorted_chunks


# -------------------------------------------fetch topk examples--------------------------------------------------------------


def get_index(queries, topk_queries):
    """
    Fetches the topk_indices given topk queries.

    parameters:
      - queries: list
          A list of all queries.
      - topk_queries: list
          A list of topk queries.

    returns:
      - index: list of int
    """
    index = []
    for i in topk_queries:
        index.append(queries.index(i))
    return index


def get_topk_examples(topk_index, examples):
    """
    Fetches topk examples from the examples given indices of topk examples.
    parameters
    ----------
    - topk_index: int
        list of indexes of topk tools.

    - examples: list
        list of all examples.

    returns
    ----------
    - res: list
        list of topk examples.
    """

    res = []
    for index in topk_index:
        res.append(examples[index])
    return res


def get_topk_given_query(query, queries, search_index, examples):
    '''
    fetches topk examples given the queries.
    parameters
    ----------
      - query: str
          user query

      - queries: list
          a list of all queries.

      - search_index:
          index object

      - examples: list
          a list of all examples.
    returns
    -----------
      - topk_examples: list
          a list of topk examples.
    '''
    query_embedding = create_query_embedding(query)
    topk_queries = find_nearest_neighbors(query_embedding, queries, search_index)
    ranked_topk_queries = rerank_queries_with_cross_encoder(query, topk_queries)
    topk_indices = get_index(queries,ranked_topk_queries)
    topk_examples = get_topk_examples(topk_indices, examples)
    return topk_examples


# ---------------------------------------------------prompt generation functions-------------------------------------------------
def client_conn(message, model_name):
    '''
      Generates an API call instant.
      parameters
      ----------
        - message: list
            A list of conversation between ChatGPT and user.

      returns
      ----------
        - completion: object
            An API instant.
    '''
    # please don't change the timeout as example generation is time consuming
    client = OpenAI(api_key=API_KEY, timeout=120, max_retries=2)
    completion = client.chat.completions.create(
        model=model_name,
        messages=message,
        temperature=0.2
    )
    return completion

def create_prompt_for_query(user_query, examples, tools):
    message = [
        {
            "role": "system",
            "content": "The following is a friendly conversation between a human and an AI. The AI is professional and parses user input to several tasks. If the AI does not know the answer to a question, it truthfully says it does not know. The AI will be provided with a set of tools their descriptions and the argument in them. Here is the list of tools:"+ json.dumps(tools) + " \n Provide the answer in the exact format as given in the following examples. \nExamples"
        }
    ]
    for example in examples:
        query = example['query']
        answer = example['answer']
        user_prompt = "Query: "+ str(query)
        assistant_prompt = str(answer)
        message.append(
        {
            "role" : "user",
            "content": user_prompt
        })
        message.append(
        {
            "role" : "assistant",
            "content": assistant_prompt
        })

    message.append(
    {
        "role":"user",
        "content":"Use the above tools to learn how to use the tool on any query. Analyse how to parse the query and extract the correct information and place in the argument name and value. Use all the required tools and arguments in correct order of its calling based on the query and your learning from all the examples. Do not assume any value, you can take the value from query or the previous called tool as shown in the examples. Also focus on the allowed values argument present in tool definition."
    })
    message.append({
        "role" : "user",
        "content" : "Now its your task to respond to the user queries in the same format as that in the above examples which is json."
    })
    message.append({
        "role" : "user",
        "content" : "Query: "+ user_query
    })
    message.append(
    {
        "role":"system",
        "content": "Generate the answer in a json format only. Enclose the strings in double quotes"
    })

    return message


def create_prompt_zero_shot(user_query, tools):
    '''
      Creates the prompt given user query, single examples and multi examples.
      parameters
      ----------
        - user_query: str
          the user query.

        - single_example: list
          the list of single tool examples.

        - multi_example: list
          the list of multi tiik examples.

      returns
      ----------
        - message: list
          the list of user prompts.
    '''
    message = [
        {
            "role": "system",
            "content": "You are a intelligent AI agent specialized in giving the tool responses given a dictionary of tools. Here is the dictionary of tools: "+ json.dumps(tools)
        }
    ]
    message.append({
        "role" : "user",
        "content" : '''
        Now its your task to respond to the user queries in the format given below
        FORMAT:[{"tool_name": "...", "arguments": [{"argument_name": "...", "argument_value": ... (depending on the argument_type)}, ...]}, ...]
        To reference the value of the ith tool in the chain, use $$PREV[i] as argument value. i = 0, 1, .. j-1; j = current tool’s index in the array If the query could not be answered with the given set of tools, output an empty list instead.
        Output in the JSON format
        '''
    })
    message.append({
        "role" : "user",
        "content" : "Query: "+ user_query
    })
    return message


def create_prompt_for_modified_example(old_tools, modified_tool, relevant_examples):
  message = [
      {
          "role":"system",
          "content":"You are an intelligent AI Agent specialized in modifying the old data and generating the new relevant data."
      }
  ]
  message.append({
      "role":"user",
      "content":"Given a list of old tools : " + json.dumps(old_tools) + "Let us say that I modified the tool" + "'" + modified_tool['tool_name'] + "'" + "to be" + json.dumps(modified_tool)+"""
      Now your task is to modify the following examples where this tool was used according to its new definition keeping in mind the new schema of json mentioned above.
      """ + json.dumps(relevant_examples)
  })
  message.append({
      "role":"system",
      "content":"Your response should be in json format only with the strings enclosed in double quotes ready to go in json.loads"
  })
  return message
# ---------------------------------------------------------------------------------------------------------------------------


def create_prompt_for_new_example(old_tools, new_tool, examples, n=no_of_examples):
  message = [
      {
          "role":"system",
          "content":"You are an intelligent AI Agent specialized in generating the new relevant data."
      }
  ]
  message.append({
      "role":"user",
      "content":"Given a list of old tools : " + json.dumps(old_tools) + "and a new tool" + "'" + new_tool['tool_name'] + "'" + " to be " + json.dumps(new_tool)+"""
      Now your task is to create """+str(n)+""" examples of the usage of the new tool along with any of the tools from the old tool list, similar to the following example:.
      """ + json.dumps(examples)
  })
  message.append({
      "role":"system",
      "content":"Your response should be in json format with the strings enclosed in double quotes. Note that To reference the value of the ith tool in the chain, use $$PREV[i] as argument value. i = 0, 1, .. j-1; j = current tool’s index in the array If the query could not be answered with the given set of tools, output an empty list instead."
  })
  return message
# -------------------------------------------------retriever---------------------------------------------------------------------
def create_tool_str(tools):
  tool_str = ''
  for tool in tools:
      tool_str += f"Tool: {tool['tool_name']}, Desc: {tool['tool_description']}\n"
  return tool_str

def tool_retriever_prompt(tool_str,user_query):
    message=[
      {"role": "system", "content": "You are an intelligent assistant. Please help the user below."},
      {"role": "user", "content": f'''You are given the following set of tools:\n {tool_str} \n Can you please figure out which tools the query "{user_query}" will require to solve, out of these tools? Please return only the tool names inside []. If it does not need any tool, return an empty list'''}]
    return message


def tool_retriever(tools, user_query):
  tool_str = create_tool_str(tools)
  message = tool_retriever_prompt(tool_str, user_query)
  output = client_conn(message, model_name_tr)
  output = output.choices[0].message.content
  list_delim=output[output.find('[')+1:output.find(']')]
  tools_retrieved = set()
  for i in list_delim.split(','):
    if (i.strip()!=''):
      tools_retrieved.add(i.strip().replace('\'','').replace('\"',''))
  return tools_retrieved


def final_tools(tools, tool_names):
  tool = []
  for tool_name in tool_names:
    tool.append(tools[tool_name])
  for tool_check in list(tools.values()):
    if not tool_check["argument_list"]:
      tool.append(tool_check)
  return tool


# -------------------------------------------------------bonus section prompts---------------------------------------------------------------
def create_prompt_for_query_bonus(user_query, examples, tools):
    message = [
        {
            "role": "system",
            "content": f"""The following is a friendly conversation between a human and an AI.
             The AI is professional and parses user input to several tasks. If the AI does not
             know the answer to a question, it truthfully says it does not know. The AI will be
              provided with a set of tools their descriptions and the argument in them. Here is
               the list of tools:"+ {json.dumps(tools)} + "\n Provide the answer in the
                exact format as given in the following examples. \nExamples """
        }
    ]
    for example in examples:
        query = example['query']
        answer = example['answer']
        user_prompt = "Query: "+ str(query)
        assistant_prompt = str(answer)
        message.append(
        {
            "role" : "user",
            "content": user_prompt
        })
        message.append(
        {
            "role" : "assistant",
            "content": assistant_prompt
        })

    message.append(
    {
        "role":"user",
        "content":"Use the above tools to learn how to use the tool on any query. Analyse how to parse the query and extract the correct information and place in the argument name and value. Use all the required tools and arguments in correct order of its calling based on the query and your learning from all the examples. Do not assume any value, you can take the value from query or the previous called tool as shown in the examples. Also focus on the allowed values argument present in tool definition."
    })
    message.append(
    {
        "role":"user",
        "content":f"After producing the list of tools, analyze the query and figure out whether it requires the combination of tool outputs via mathematical operations, iterations, conditional logic etc. or not. In case it does, use the lambda function to produce the required results. Examples of such queries are given below: \n "
    })

    message.append(
        {
            "role":"user",
            "content":f"Find all tasks created by user 'USER-321' and check if there are more than 10 such tasks"
        }
    )
    message.append(
        {
            "role":"assistant",
            "content":"""
            "answer": [
        {
          "tool_name": "search_object_by_name",
          "arguments": [
            {
              "argument_name": "query",
              "argument_value": "USER-321"
            }
          ]
        },
        {
          "tool_name": "works_list",
          "arguments": [
            {
              "argument_name": "created_by",
              "argument_value": [
                "$$PREV[0]"
              ]
            },
            {
              "argument_name": "type",
              "argument_value": [
                "task"
              ]
            }
          ]},
          {
          "tool_name": "lambda",
          "arguments": [
            {
              "argument_name": "expression",
              "argument_value": "lambda $$PREV[1]: True if len($$PREV[1]) > 10 else False"
            }
          ]
        }
      ]
            """
        }
    )

    message.append({
        "role" : "user",
        "content" : "Now its your task to respond to the user queries in the same format as that in the above examples which is json. Use the lambda function only when necessary."
    })
    message.append({
        "role" : "user",
        "content" : "Query: "+ user_query
    })
    message.append(
    {
        "role":"system",
        "content": "Generate the answer in a json format only. Enclose the strings in double quotes"
    })

    return message



# -------------------------------------------------miscellaneous functions------------------------------------------------------------
def create_tool_dict(tools):
    tool_dict = {}
    for i in tools:
        tool_dict[i["tool_name"]] = i
    return tool_dict

def get_parsed_json(text_parsed):
  ans_list = []
  json_str = text_parsed.split('```json')
  for i in json_str:
    if "```" in i:
      json_data = i.split("```")[0].strip()
      if json_data:  # Check if the JSON data is not empty
        json_obj = json.loads(json_data)
        if type(json_obj)==list:
          ans_list.extend(json_obj)
        else:
          ans_list.append(json_obj)
  return ans_list
# -------------------------------------------------postprocessing---------------------------------------------------------------------
def get_json(pred):
  '''
  parameters
  -----------------
  pred: str
    Answer predicted by the LLM as a string

  returns
  -----------------
  json_pred: list
    List of dictionaries that represents the input string as a json
  '''

  try:
    # Tries to find ```json ``` type json format
    return json.loads(pred[pred.find('```json'):-1*("".join(reversed(pred)).find('```')+1)])
  except:
    # Tries to find first instance of '[' from the left and first instance of ']' from the right, and converts all in between ito a json.
    try:
      return json.loads(pred[pred.find('['):-1*("".join(reversed(pred)).find(']')+1)] + ']')
    except:
      # Tries to fix keys/ strings being wrapped in single quotes, then tries to decode as above
        pred= pred.replace('\'','\"')
        try:
          return json.loads(pred[pred.find('['):-1*("".join(reversed(pred)).find(']')+1)] + ']')
          # Tries to check for instances of boolean values true and false, that might have been misspelt as True and False
        except:
          pred = pred.replace("True", "true").replace("False", "false")
          return json.loads(pred[pred.find('['):-1*("".join(reversed(pred)).find(']')+1)] + ']')

def dict_unwrap(json_pred):
  # Unwraps a dictionary if GPT-4 outputs one instead of a list
  if type(json_pred)==dict:
    for key in json_pred.keys():
      if type(json_pred[key])==list:
        print(json_pred[key])
        return json_pred[key]
    return []
  return json_pred

def list_in_str_handler(json_pred):
  '''
  parameters
  -----------------
  json_pred: list
    List of dictionaries that represents the tool call sequence

  -----------------
  json_pred: list
    Tool call sequence with string arguments like '[arg_val]' turned into 'arg_val'
  '''
  # Iterate over tool call sequence and get argument values
  for i, tool in enumerate(json_pred):

    for j, arg in enumerate(tool["arguments"]):
      arg_val= arg["argument_value"]
      # If value is a string and it starts with [ and ends with ], remove them
      if (type(arg_val)==str):
        if arg_val.startswith('[') and arg_val.endswith(']'):
          arg["argument_value"]=arg_val[1:-1]
      # If value is a list, then iterate over it and perform similar operations as above
      elif (type(arg_val)==list):
        for num_item,arg_val_item in enumerate(arg_val):
          if (type(arg_val_item)==str):
            if arg_val_item.startswith('[') and arg_val_item.endswith(']'):
              arg["argument_value"][num_item]=arg_val_item[1:-1]
  return json_pred

def func_name_handler(json_pred,tools,no_arg_tool_list):
  '''
  parameters
  -----------------
  json_pred: list
    List of dictionaries that represents the tool call sequence

  tools: dict
    Dict representing tools

  no_arg_tool_list: list
    List of tool names that do not have arguments
  returns
  -----------------
  json_pred: list
    Tool call sequence with arguments with $${function_name} errors removed
  '''
  # Iteration variable
  i=0
  # WHILE loop is required, len(range()) does not work as the loop condition is kept static while items are inserted into the loop.
  while(i<len(json_pred)):
    tool=json_pred[i]
    for j, arg in enumerate(tool["arguments"]):
      if arg["argument_name"] in tools[tool["tool_name"]]["args"] :
          # Check argument value
          temp = arg["argument_value"]
          if type(temp)==str:
            # If argument value starts with $$ but is not $$PREV[i]
            if (temp.startswith('$$')) and not temp.startswith('$$PREV['):
              # Remove the $$
              temp_lowercase_call=temp.lower()[2:]
              # Iterate over tools with no arguments to figure the appropriate tool to call
              for no_arg_tool in no_arg_tool_list:
                if temp_lowercase_call.startswith(no_arg_tool):
                  # Create tool call
                  tool_ins = {}
                  tool_ins['arguments']=[]
                  tool_ins['tool_name']=no_arg_tool
                  # Insert tool into the list
                  json_pred.insert(i,tool_ins)
                  i+=1
                  # Iterate over the tools, starting from the tool under consideration
                  for i_n, tool_n in enumerate(json_pred[i:]):
                    for j_n, arg_n in enumerate(tool_n["arguments"]):
                      # Check the argument value
                      prevset = arg_n["argument_value"]
                      # If the argument value is
                      if (type(prevset)==str):
                        # If the argument is of $$PREV[i] type, and it referenced the returned value of the tool that
                        # came at the same position as, or after the inserted tool, increment i for it,
                        if prevset.startswith("$$PREV["):
                          try:
                            n=int(prevset[7:-1])
                          except:
                            indexing_pos=prevset.find('][')
                            try:
                              n=int(prevset[7:indexing_pos])
                            except:
                              pass
                        if n>=i-1:
                          arg_n["argument_value"]=f"$$PREV[{n}]"
                      # If the argument value is a list, iterate over the values and perform the same process as above.
                      elif type(prevset)==list:
                        for list_num,prev_val in enumerate(prevset):
                          if prev_val.startswith("$$PREV["):
                            n=0
                            try:
                              n=int(prevset[7:-1])
                            except:
                              indexing_pos=prevset.find('][')
                              try:
                                n=int(prevset[7:indexing_pos])
                              except:
                                pass
                          if n>=i-1:
                            arg_n["argument_value"][list_num]=f"$$PREV[{n}]"
                  arg["argument_value"]=f"$$PREV[{i-1}]"
          # otherwise the argument value is a list, iterate over this list, and perform the same operations as above.
          elif (type(temp)==list):
            for num_arg,temp_el in enumerate(temp):

              if type(temp_el)==str:
                if (temp_el.startswith('$$')) and not temp_el.startswith('$$PREV'):
                  temp_lowercase_call=temp_el.lower()[2:]

                  for no_arg_tool in no_arg_tool_list:
                    if temp_lowercase_call.startswith(no_arg_tool):
                      tool_ins = {}
                      tool_ins['arguments']=[]
                      tool_ins['tool_name']=no_arg_tool
                      json_pred.insert(i,tool_ins)
                      i+=1

                      for i_n, tool_n in enumerate(json_pred[i:]):
                        for j_n, arg_n in enumerate(tool_n["arguments"]):

                          prevset = arg_n["argument_value"]
                          if (type(prevset) not in [list,bool,float]):
                            if prevset.startswith("$$PREV[") and int(prevset[7:-1])>=i-1:
                              n=int(prevset[7:-1])+1
                              arg_n["argument_value"]=f"$$PREV[{n}]"

                          elif type(prevset) not in [bool,float]:
                            for list_num,prev_val in enumerate(prevset):
                              try:
                                if prev_val.startswith("$$PREV[") and int(prev_val[7:-1])>=i-1:
                                  n=int(prevset[7:-1])+1
                                  arg_n["argument_value"][list_num]=f"$$PREV[{n}]"

                              except:
                                pass
                      arg["argument_value"][num_arg]=f"$$PREV[{i-1}]"
    i+=1
  return json_pred



def unknown_tool_remover(json_pred,tools):
    for i,tool_call in enumerate(json_pred):
        if tool_call["tool_name"] not in tools.keys():
            return True


def type_handler(json_pred,tools,array_check,string_check,num_check,bool_check,string_to_boolean):
  '''
  parameters
  -----------------
  json_pred: list
    List of dictionaries that represents the tool call sequence

  tools: dict
    Dict representing tools

  array_check: list
    List of keywords to check for array return types

  string_check: list
    List of keywords to check for string return types

  num_check: list
    List of keywords to check for numeral return types

  bool_check: list
    List of keywords to check for boolean return types

  string_to_boolean: dict
    Dictionary mapping common representations of True/False values to booleans

  returns
  -----------------
  json_pred: list
    Tool call sequence with tool inputs respecting tool input type requirements
  '''
  for i, tool in enumerate(json_pred):

    for j, arg in enumerate(tool["arguments"]):

      if arg["argument_name"] in tools[tool["tool_name"]]["args"] :

          arg_type = tools[tool["tool_name"]]["args"][arg["argument_name"]]["argument_type"]

          temp = json_pred[i]["arguments"][j]["argument_value"]

          # Split the argument type by spaces for easier checks later
          typcheck = set(arg_type.lower().split(' '))
          # To check if arg_type is supposed to be a list, but it is not
          if typcheck.intersection(array_check) and type(temp)!=list:
            # If argument is not $$PREV type and is a string, convert to array of strings

            if not temp.startswith("$$") and typcheck.intersection(string_check) :
              temp = [str(temp)]
            # If argument type has integer, convert to list of integers
            elif typcheck.intersection(num_check):
              if type(temp) in [int,float]:
                temp = [temp]
              else:
                try:
                  temp = [int(temp)]
                except:
                  pass
            # If argument type has boolean, convert to list of booleans
            elif typcheck.intersection(bool_check):

              try:
                if type(temp)==str:
                  temp = [string_to_boolean.get(temp,temp)]
                else:
                  temp=[bool(temp)]
              except:
                pass
          # If argument type is string, convert to string
          elif typcheck.intersection(string_check) and type(temp) != str:
            # If the argument type is currently a list, convert to string. Only the first argument is going to be considered.
            if (type(temp)==list and not temp):
              try:
                if (not temp[0].startswith("$$")):
                    temp = str(temp[0])
              except:
                pass
            if (type(temp) in [int,float]):
              temp=str(temp)
          # If argument type is boolean, convert to boolean
          elif typcheck.intersection(bool_check) and type(temp)!= bool :
            # If the argument type is currently a list, convert to boolean. Only the first argument is going to be considered.
            if (type(temp)==list and temp):
              if (type(temp[0])==str):
                if not temp[0].startswith("$$"):
                  try:
                    temp = string_to_boolean.get(temp[0], temp)
                  except:
                    pass
              elif (type(temp[0])==bool):
                temp=temp[0]
              elif (type(temp[0]) in [int,float]):
                temp=bool(temp[0])
            # If the argument type is currently a string, convert to boolean
            elif type(temp)==str:
              temp=string_to_boolean.get(temp,temp)
          json_pred[i]["arguments"][j]["argument_value"] = temp

      else:
        print("arg name not match for",arg)
  return json_pred


def prev_ret_type_handler(json_pred,tools,array_check,string_check,num_check,bool_check):
  '''
  parameters
  -----------------
  json_pred: list
    List of dictionaries that represents the tool call sequence

  tools: dict
    Dict representing tools

  array_check: list
    List of keywords to check for array return types

  string_check: list
    List of keywords to check for string return types

  num_check: list
    List of keywords to check for numeral return types

  bool_check: list
    List of keywords to check for boolean return types

  returns
  -----------------
  json_pred: list
    Tool call sequence with $$PREV[i] type arguments respecting tool input type requirements
  '''
  # Iterate over list of dictionaries
  for i, tool in enumerate(json_pred):
    for j, arg in enumerate(tool["arguments"]):

      if arg["argument_name"] in tools[tool["tool_name"]]["args"] :
          # Fetch argument type, and the current argument value
          arg_type = set(tools[tool["tool_name"]]["args"][arg["argument_name"]]["argument_type"].split(' '))
          temp = json_pred[i]["arguments"][j]["argument_value"]
          # If current argument is a string
          if type(temp)== str:
            # If it is a $$PREV type argument
            if temp.startswith('$$PREV'):
              # Identify the return type of the tool call referenced by $$PREV[i],
              refer=temp[7:-1]
              ref_type= set(tools[json_pred[int(refer)]["tool_name"]]["return_type"].split(' '))
              # If the referenced tool does not return an array, but the argument expects one
              if not ref_type.intersection(array_check) and arg_type.intersection(array_check):
                json_pred[i]["arguments"][j]["argument_value"]=[temp]
          # Otherwise, if the argument value is a list, iterate over it
          elif (type(temp)==list):
            for temp_el in temp:
              # Perform similar actions as above
              if type(temp_el)==str:
                if temp_el.startswith('$$PREV'):
                  refer=temp_el.lower()[7:-1]
                  ref_type= set(tools[json_pred[int(refer)]["tool_name"]]["return_type"].split(' '))
                  if (ref_type.intersection(array_check) and arg_type.intersection(array_check)) or (not ref_type.intersection(array_check) and not arg_type.intersection(array_check)):
                    json_pred[i]["arguments"][j]["argument_value"]=temp_el

  return json_pred

def postprocess(json_pred, tool_data):
  '''
  parameters
  -----------------
  json_pred: str
    Answer predicted by the LLM as a string

  tool_data: dict
    dictionary representing the given tools

  returns
  -----------------
  json_pred: list
    List of dictionaries that represents the final answer
  '''
  # turn json in string format into list of dicts
  # Get dictionary of tools to simplify work
  json_pred = dict_unwrap(json_pred)
  tools = {}
  for i,tool in enumerate(tool_data):
    tools[tool["tool_name"]] = tool
    tools[tool["tool_name"]]["args"] = {}
    tools[tool["tool_name"]]["return_type"] = tool["return_type"]
    for arg in tool["argument_list"]:
      tools[tool["tool_name"]]["args"][arg["argument_name"]] = arg
  for tool in json_pred:
    if not tool.get('arguments',None):
      tool["arguments"] = []
  # Lists holding keywords to search for in the argument types/ return types
  array_check = ["array","list","arrays","lists"]
  string_check=["string","str","strings"]
  num_check=["integer","int32","number","float","double","float32"]
  bool_check=["bool","boolean","true","false"]
  # Dictionary mapping common strings to boolean values
  string_to_boolean = {"True": True, "False": False, "1": True, "0": False, "yes": True, "no": False, "true" : True, "false": False, 'True':True, 'False':False}

  # Get tools for which no argument is required, to fix $${function_name} errors encountered
  no_arg_tool_list = []
  for i,tool in enumerate(tool_data):
    if not tool["argument_list"]:
      no_arg_tool_list.append(tool["tool_name"])

  try:
    if unknown_tool_remover(json_pred, tools):
      return []
  except:
    pass
  try:
    json_pred = list_in_str_handler(json_pred)
  except:
    pass

  try:
  # Fix type errors
    json_pred = type_handler(json_pred,tools,array_check,string_check,num_check,bool_check,string_to_boolean)
  except:
    pass
  try:
  # Fix $${function_name} errors
    json_pred = func_name_handler(json_pred,tools,no_arg_tool_list)
  except:
    pass
  try:
  # Fix return types for $$PREV[i] type arguments
    json_pred = prev_ret_type_handler(json_pred,tools,array_check,string_check,num_check,bool_check)
  except:
    pass

  return json_pred

tools = read_file(tool_list_path)
examples = read_file(example_path)
tool_dict = create_tool_dict(tools["tools"])
tool_obj = Tools(tool_dict, examples)

def main(query):
  if len(tool_obj.examples)<5:
    message = create_prompt_zero_shot(query,tool_obj.tools)
    res = client_conn(message, model_name_tr)
    answer = []
    try:
      answer = json.loads(res.choices[0].message.content)
      answer = postprocess(answer, list(tool_obj.tools.values())) # postprocessing
    except:
      pass
    return answer

  topk_examples = get_topk_given_query(query, tool_obj.queries, tool_obj.search_index, tool_obj.examples)
  reduced_tools = tool_retriever(list(tool_obj.tools.values()), query)
  if not reduced_tools:
    return []
  tool_list = final_tools(tool_obj.tools, list(tool_obj.tools.keys()))
  if 'lambda' in reduced_tools:
    message = create_prompt_for_query_bonus(query, topk_examples, tool_list)
  else:
    message = create_prompt_for_query(query, topk_examples, tool_list)
  res = client_conn(message, model_name_tr)
  print("GPT4 Response: ",res.choices[0].message.content)
  answer = []
  try:
    answer = json.loads(res.choices[0].message.content)
    answer = postprocess(answer, list(tool_obj.tools.values())) # postprocessing
  except:
    pass
  return answer