File size: 8,059 Bytes
225ce96 |
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 |
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():
import requests
response = requests.get(
f'https://newapi.didx.net/DidxApis/api/getAvailableDIDS.php?UserID=700290&Password={password}&CountryCode=1&AreaCode=-1')
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 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"]
}
}
}
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_country, function_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()
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 |