temp_chatbot.py

class DIDx_Chatbot:
  def __init__(self):
    import openai

    def get_balance():
      import requests
      response = requests.get(f'https://newapi.didx.net/DidxApis/api/WebGetAccountBalance.php?UserID={user_id}&Password={password}')
      return response.text

    def get_available_dids_country():
      import requests
      response = requests.get(f'https://newapi.didx.net/DidxApis/api/getDIDCountry.php?UserID=700290&Password={password}')
      return response.text
        
    def get_available_dids(country_code : str, area_code : str):
      import requests
      response = requests.get(f'http://newapi.didx.net/DidxApis/api/getAvailableDIDS.php?UserID={user_id}&Password={password}&CountryCode={country_code}&AreaCode={area_code}')
      return response.text


    self.get_balance = get_balance
    self.get_available_dids_country = get_available_dids_country
    self.get_available_dids = get_available_dids


    function_balance = {
      "type": "function",
      "function": {
        "name": "get_balance",
        "description": "Retrieve the available balance present in account",
        "parameters": {
          "type": "object",
          "properties": {
            "Balance": {
              "type": "string",
              "description": "A balance in the amount of dollars"
            },
            "Status" : {
                "type" : "string",
                "description" : "A balance status due or something else"

            }
          },
          "required": ["account balance", "Status"]
        }
      }
    }
    function_dids = {
      "type": "function",
      "function": {
        "name": "get_available_dids_country",
        "description": "Retrieve the available DIDs country information",
        "parameters": {
          "type": "object",
          "properties": {
            "country_code": {
              "type": "string",
              "description": "A balance in the amount of dollars"
            },
            "description" : {
                "type" : "string",
                "description" : "A balance status due or something else"

            },
            "country_id": {
              "type": "string",
              "description": "A balance in the amount of dollars"
            }

          },
          "required": ["country_code","description","country_id" ]
        }
      }
    }

    function_dids_two = {
      "type": "function",
      "function": {
        "name": "get_available_dids",
        "description": "Retrieve the available DIDs list",
        "parameters": {
          "type": "object",
          "properties": {
            "country_code": {
              "type": "string",
              "description": "A country code"
            },
          "area_code": {
              "type": "string",
              "description": "An area code of a country"
            }
          },
          "required": ["country_code","area_code"]
        }
      }
    }
      
    import os
    self.client = openai.OpenAI(api_key = os.environ['openai_api_key'])

    # Step 1: Create an Assistant
    self.assistant = self.client.beta.assistants.create(
        name="DIDx Customer Support Chatbot",
        instructions="You are a personal DIDx customer support chatbot.",
        tools=[function_balance,function_dids, function_dids_two],
        model="gpt-4-1106-preview",
    )
    
  def user_auth(self, user, passw):
    global user_id
    global password

    user_id = user
    password = passw

  def user_chat(self,query):
    import time
    # Step 2: Create a Thread
    thread = self.client.beta.threads.create()

    # Step 3: Add a Message to a Thread
    message = self.client.beta.threads.messages.create(
        thread_id=thread.id,
        role="user",
        content= query
    )

    # Step 4: Run the Assistant
    run = self.client.beta.threads.runs.create(
        thread_id=thread.id,
        assistant_id=self.assistant.id,
        instructions="Please address the user as Aqdas"
    )
    answer = None
    while True:
      # Retrieve the run status
      run_status = self.client.beta.threads.runs.retrieve(
          thread_id=thread.id,
          run_id=run.id
      )
      # print(run_status.model_dump_json(indent=4))
      run_status.model_dump_json(indent=4)

      # If run is completed, get messages
      if run_status.status == 'completed':
          messages = self.client.beta.threads.messages.list(
              thread_id=thread.id
          )
          # Loop through messages and print content based on role
          for msg in messages.data:
              role = msg.role
              content = msg.content[0].text.value
              print(f"{role.capitalize()}: {content}")
              answer = f"{role.capitalize()}: {content}"
              break
          break
      elif run_status.status == 'requires_action':
          # print("Function Calling")
          required_actions = run_status.required_action.submit_tool_outputs.model_dump()
          # print('required action test: ',required_actions)
          tool_outputs = []
          import json
          for action in required_actions["tool_calls"]:
              func_name = action['function']['name']
              arguments = json.loads(action['function']['arguments'])

              if func_name == "get_balance":
                  output = self.get_balance()
                  tool_outputs.append({ 
                      "tool_call_id": action['id'],
                      "output": output
                  })
    
              elif func_name == 'get_available_dids_country':
                    output = self.get_available_dids_country()
                    tool_outputs.append({
                      "tool_call_id": action['id'],
                      "output": output
                  })
              elif func_name == 'get_available_dids':
                      output = self.get_available_dids(country_code=arguments['country_code'], area_code=arguments['area_code'])
                      tool_outputs.append({
                        "tool_call_id": action['id'],
                        "output": output
                    })
    
              else:
                  raise ValueError(f"Unknown function: {func_name}")

          print("Submitting outputs back to the Assistant...")
          self.client.beta.threads.runs.submit_tool_outputs(
              thread_id=thread.id,
              run_id=run.id,
              tool_outputs=tool_outputs
          )
      else:
          print("Waiting for the Assistant to process...")
          time.sleep(5)
      
    if answer is not None:
      print(f'this is my answer : ', answer)
      return answer