Spaces:
Sleeping
Sleeping
File size: 10,178 Bytes
a18f1a9 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 |
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 |