Spaces:
Sleeping
Sleeping
File size: 4,671 Bytes
879a8ae f18440e 879a8ae f18440e 879a8ae f18440e 879a8ae f18440e 879a8ae f18440e 879a8ae f18440e 879a8ae f18440e 879a8ae f18440e 879a8ae |
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 |
class supertec_bot:
def __init__(self):
import openai
def get_data():
import requests
headers = {
'User-Agent': 'Mozilla/5.0', # You can try different User-Agent strings if needed
}
payload = {'company_token': 'II@tNfQ70O'}
response = requests.post("https://superteclabs.com/apis2/retrieveallusers.php", data=payload,headers=headers)
data = response.json()
return response.text
self.get_data = get_data
function_balance = {
"type": "function",
"function": {
"name": "get_data",
"description": "get all data from database",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "The query to run when getting data"
},
},
"required": ["query"]
}
}
}
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="SuperTech Support Chatbot",
# instructions="You are a a personal supertec chatbot that's highly tech-oriented and a part of the Rehan Foundation. Your primary job is to respond to user queries.",
# tools=[function_balance],
# model="gpt-3.5-turbo",
# )
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="asst_IBlxaXvYi7O0mUYvhxZMzw3I",
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_data":
output = self.get_data()
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 |