Spaces:
Sleeping
Sleeping
| 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={user_id}&Password={password}') | |
| return response.text | |
| def get_available_dids(CountryCode, AreaCode): | |
| import requests | |
| response = requests.get( | |
| f'https://newapi.didx.net/DidxApis/api/getAvailableDIDS.php?UserID={user_id}&Password={password}&CountryCode={CountryCode}&AreaCode={AreaCode}') | |
| return response.text | |
| def get_purchased_dids(): | |
| import requests | |
| response = requests.get( | |
| f'https://newapi.didx.net/DidxApis/api/getMyPurchasedNumbers.php?UserID={user_id}&Password={password}') | |
| return response.text | |
| self.get_balance = get_balance | |
| self.get_available_dids_country = get_available_dids_country | |
| self.get_available_dids = get_available_dids | |
| self.get_purchased_dids = get_purchased_dids | |
| function_balance = { | |
| "type": "function", | |
| "function": { | |
| "name": "get_balance", | |
| "description": "Retrieve the available balance present in the 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_country = { | |
| "type": "function", | |
| "function": { | |
| "name": "get_available_dids_country", | |
| "description": "Retrieve the available DIDs country information", | |
| "parameters": { | |
| "type": "object", | |
| "properties": { | |
| "country_code": { | |
| "type": "string", | |
| "description": "Country code for available DIDs" | |
| }, | |
| "description": { | |
| "type": "string", | |
| "description": "Country/City/County code for available DIDs" | |
| }, | |
| "country_id": { | |
| "type": "string", | |
| "description": "Country/City/County code for available DIDs" | |
| } | |
| }, | |
| "required": ["country_code", "description", "country_id"] | |
| } | |
| } | |
| } | |
| function_dids = { | |
| "type": "function", | |
| "function": { | |
| "name": "get_available_dids", | |
| "description": "Retrieve all available DIDs list", | |
| "parameters": { | |
| "type": "object", | |
| "properties": { | |
| "CountryCode": { | |
| "type": "string", | |
| "description": "Country code for available DIDs. Default CountryCode is 1 for US" | |
| }, | |
| "description": { | |
| "type": "string", | |
| "description": "A list of available DIDs" | |
| }, | |
| "AreaCode": { | |
| "type": "string", | |
| "description": "Area Code for the available DIDs. Default AreaCode is -1 that will show all AreaCode" | |
| } | |
| }, | |
| "required": ["CountryCode", "description", "AreaCode"] | |
| } | |
| } | |
| } | |
| function_purchased_dids = { | |
| "type": "function", | |
| "function": { | |
| "name": "get_purchased_dids", | |
| "description": "Retrieve all purchased DIDs", | |
| "parameters": { | |
| "type": "object", | |
| "properties": { | |
| "DIDNumber": { | |
| "type": "string", | |
| "description": "Purchased DIDNumber" | |
| }, | |
| "OurSetupCost": { | |
| "type": "string", | |
| "description": "Our Setup Cost" | |
| }, | |
| "OurMonthlyCharges": { | |
| "type": "string", | |
| "description": "Our Monthly Charges" | |
| }, | |
| "OurPerMinuteCharges": { | |
| "type": "string", | |
| "description": "Our Per-Minute Charges" | |
| } | |
| }, | |
| "required": ["CountryCode", "description", "AreaCode"] | |
| } | |
| } | |
| } | |
| import os | |
| from dotenv import load_dotenv | |
| # Load environment variables from .env file | |
| load_dotenv() | |
| self.client = openai.OpenAI(api_key=os.environ['openai_api_key']) | |
| print(self.client) | |
| # 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_country, function_dids, function_purchased_dids], | |
| 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="" | |
| ) | |
| 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(CountryCode=arguments['CountryCode'], | |
| AreaCode=arguments['AreaCode']) | |
| tool_outputs.append({ | |
| "tool_call_id": action['id'], | |
| "output": output | |
| }) | |
| elif func_name == 'get_purchased_dids': | |
| output = self.get_purchased_dids() | |
| 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 |