Spaces:
Sleeping
Sleeping
| class Assistant_DIDx: | |
| def __init__(self,client): | |
| import requests | |
| import os | |
| import json | |
| def get_balance(): | |
| """Get Account Balance""" | |
| 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(): | |
| """Get list of countries of available dids """ | |
| response = requests.get(f'https://newapi.didx.net/DidxApis/api/getDIDCountry.php?UserID={user_id}&Password={password}') | |
| return response.text | |
| def get_available_dids_number(country_code : str, area_code: str = '-1'): | |
| '''Get available dids based on area and country code''' | |
| base_url = f'http://newapi.didx.net/DidxApis/api/getAvailableDIDS.php?UserID={user_id}&Password={password}&CountryCode={country_code}&AreaCode={area_code}' | |
| response = requests.get(base_url) | |
| return response.text | |
| def purchase_did(did_number: str): | |
| '''Purchase a DID''' | |
| response = requests.get(f'http://newapi.didx.net/DidxApis/api/BuyDIDByNumber.php?UserID={user_id}&Password={password}&DIDNumber={did_number}') | |
| return response.text | |
| def get_did_cost(did_number: str): | |
| '''Get Cost of a DID number''' | |
| response = requests.get(f'http://newapi.didx.net/DidxApis/api/GetCostOfDIDByNumber.php?UserID={user_id}&Password={password}&DIDNumber={did_number}') | |
| return response.text | |
| def get_did_general_info(did_number: str): | |
| '''Get general info about a signle did number''' | |
| response = requests.get(f'http://newapi.didx.net/DidxApis/api/getDIDInfo.php?UserID={user_id}&Password={password}&DIDNumber={did_number}') | |
| return response.text | |
| def get_did_calllogs(did_number: str): | |
| '''Get call log info of a single did''' | |
| response = requests.get(f'http://newapi.didx.net/DidxApis/api/getDIDCallLog.php?UserID={user_id}&Password={password}&DIDNumber={did_number}') | |
| return response.text | |
| def get_list_purchase_did(): | |
| '''get a list of purchased dids''' | |
| response = requests.get(f'https://newapi.didx.net/DidxApis/api/getMyPurchasedNumbers.php?UserID={user_id}&Password={password}') | |
| return response.text | |
| function_balance = { | |
| "type": "function", | |
| "function": { | |
| "name": "get_balance", | |
| "description": "Retrieve the available balance present in account", | |
| "parameters": { | |
| "type": "object", | |
| "properties": {}, | |
| "required": [] | |
| } | |
| } | |
| } | |
| function_dids_country = { | |
| "type": "function", | |
| "function": { | |
| "name": "get_available_dids_country", | |
| "description": "RETRIEVE A LIST OF COUNTRIES OF AVAILABLE DIDS", | |
| "parameters": { | |
| "type": "object", | |
| "properties": {}, | |
| "required": [] | |
| } | |
| } | |
| } | |
| function_dids_number= { | |
| "type": "function", | |
| "function": { | |
| "name": "get_available_dids_number", | |
| "description": "GET A LIST OF AVAILABLE DIDs", | |
| "parameters": { | |
| "type": "object", | |
| "properties": { | |
| 'country_code' : { | |
| 'type' : 'string', | |
| 'description' : 'The country code' | |
| }, | |
| 'area_code': { | |
| 'type' : 'string', | |
| 'description' : 'The area code of the country' | |
| }, | |
| }, | |
| "required": ['country_code'] | |
| } | |
| } | |
| } | |
| function_purchase_did= { | |
| "type": "function", | |
| "function": { | |
| "name": "purchase_did", | |
| "description": "TO PURCHASE A DID", | |
| "parameters": { | |
| "type": "object", | |
| "properties": { | |
| 'did_number' : { | |
| 'type' : 'string', | |
| 'description' : 'A did number' | |
| }, | |
| }, | |
| "required": ['did_number'] | |
| } | |
| } | |
| } | |
| function_get_did_cost= { | |
| "type": "function", | |
| "function": { | |
| "name": "get_did_cost", | |
| "description": "GET COST OF A DID NUMBER", | |
| "parameters": { | |
| "type": "object", | |
| "properties": { | |
| 'did_number' : { | |
| 'type' : 'string', | |
| 'description' : 'A did number' | |
| }, | |
| }, | |
| "required": ['did_number'] | |
| } | |
| } | |
| } | |
| function_get_did_general_info= { | |
| "type": "function", | |
| "function": { | |
| "name": "get_did_general_info", | |
| "description": "This method retrieve general DID Information", | |
| "parameters": { | |
| "type": "object", | |
| "properties": { | |
| 'did_number' : { | |
| 'type' : 'string', | |
| 'description' : 'A did number' | |
| }, | |
| }, | |
| "required": ['did_number'] | |
| } | |
| } | |
| } | |
| function_get_did_calllogs= { | |
| "type": "function", | |
| "function": { | |
| "name": "get_did_calllogs", | |
| "description": "This method retrieves DID call logs.", | |
| "parameters": { | |
| "type": "object", | |
| "properties": { | |
| 'did_number' : { | |
| 'type' : 'string', | |
| 'description' : 'A did number' | |
| }, | |
| }, | |
| "required": ['did_number'] | |
| } | |
| } | |
| } | |
| function_get_list_purchase_did= { | |
| "type": "function", | |
| "function": { | |
| "name": "get_list_purchase_did", | |
| "description": "Method for Getting Purchased Numbers. This method returns the list of available DIDs which you have purchased in your account.", | |
| "parameters": { | |
| "type": "object", | |
| "properties": {}, | |
| }, | |
| "required": [] | |
| } | |
| } | |
| import json | |
| assistant_file_path = 'didx_assistant.json' | |
| if os.path.exists(assistant_file_path): | |
| with open(assistant_file_path) as file: | |
| assistant_data = json.load(file) | |
| assistant_id = assistant_data['assistant_id'] | |
| print("Loaded existing assistant ID.") | |
| else: | |
| print('creating new assistant') | |
| assistant = client.beta.assistants.create( | |
| name="DIDx Customer Support Chatbot", | |
| instructions="You are a personal DIDx customer support chatbot.", | |
| tools=[function_balance, function_dids_country, function_dids_number, function_purchase_did, function_get_did_cost,function_get_did_general_info,function_get_did_calllogs,function_get_list_purchase_did], | |
| model="gpt-4-1106-preview", | |
| ) | |
| assistant_id = assistant.id | |
| self.assistant_id = assistant.id | |
| self.get_balance = get_balance | |
| self.get_available_dids_country = get_available_dids_country | |
| self.get_available_dids_number = get_available_dids_number | |
| self.purchase_did = purchase_did | |
| self.get_did_cost = get_did_cost | |
| self.get_did_general_info = get_did_general_info | |
| self.get_did_calllogs = get_did_calllogs | |
| self.get_list_purchase_did = get_list_purchase_did | |
| def user_auth(self, user, passw): | |
| global user_id | |
| global password | |
| user_id = user | |
| password = passw | |
| def chat_user(self, client, user_query): | |
| import time | |
| import json | |
| thread = client.beta.threads.create() | |
| message = client.beta.threads.messages.create( | |
| thread_id=thread.id, | |
| role="user", | |
| content= user_query | |
| ) | |
| run = client.beta.threads.runs.create( | |
| thread_id=thread.id, | |
| assistant_id=self.assistant_id, | |
| instructions="Please address the user as Aqdas" | |
| ) | |
| while True: | |
| run_status = client.beta.threads.runs.retrieve( | |
| thread_id = thread.id, | |
| run_id = run.id | |
| ) | |
| if run_status.status == 'completed': | |
| message = client.beta.threads.messages.list( | |
| thread_id = thread.id | |
| ) | |
| for msg in message.data: | |
| role = msg.role | |
| content = msg.content[0].text.value | |
| answer = f'{role.capitalize()} : {content}' | |
| print(answer) | |
| return answer | |
| break | |
| break | |
| elif run_status.status == 'requires_action': | |
| print('Function Calling') | |
| required_actions = run_status.required_action.submit_tool_outputs.model_dump() | |
| tool_outputs = [] | |
| 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_number': | |
| output = self.get_available_dids_number(country_code=arguments['country_code']) | |
| tool_outputs.append({'tool_call_id':action['id'], | |
| 'output': output}) | |
| elif func_name == 'purchase_did': | |
| output = self.purchase_did(did_number=arguments['did_number']) | |
| tool_outputs.append({'tool_call_id':action['id'], | |
| 'output': output}) | |
| elif func_name == 'get_did_cost': | |
| output = self.get_did_cost(did_number=arguments['did_number']) | |
| tool_outputs.append({'tool_call_id':action['id'], | |
| 'output': output}) | |
| elif func_name == 'get_did_general_info': | |
| output = self.get_did_general_info(did_number=arguments['did_number']) | |
| tool_outputs.append({'tool_call_id':action['id'], | |
| 'output': output}) | |
| elif func_name == 'get_did_calllogs': | |
| output = self.get_did_calllogs(did_number=arguments['did_number']) | |
| tool_outputs.append({'tool_call_id':action['id'], | |
| 'output': output}) | |
| elif func_name == 'get_list_purchase_did': | |
| output = self.get_list_purchase_did(did_number=arguments['did_number']) | |
| tool_outputs.append({'tool_call_id':action['id'], | |
| 'output': output}) | |
| else: | |
| raise ValueError(f'Unknown function:{func_name}') | |
| 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 proces') | |
| time.sleep(5) |