I am facing the above mentioned error when I am using the assistant to call a function having parameter of type dict. Following is my code:
import ast
import json
import openai
import time
import requests
def createproductSF(input: dict) -> str:
url = "https://awaistest.dev-techloyce.com/api/v1/products"
bearerToken = 'auth-token';
headers = {
'Content-Type':'application/json',
'Authorization':bearerToken
}
new_input=input.replace('data = ','')
data=ast.literal_eval(new_input)
res = requests.post(url,headers=headers,json=data)
print(res)
return str(res)
tools_list = [{
"type": "function",
"function": {
"name": "createproductSF",
"description": "This will create a product in SubscriptionFlow by getting a dict object if it is present that is given as input to this function and will send a request to subscriptionflow to create a product with the data",
"parameters": {
"type": "object",
"properties": {
"input": {
"type": "dict",
"description": "This is the JSON product data which will be used by the function to send a post request for product creation purpose"
}
},
"required": ["input"]
}
}
}]
# Initialize the client
client = openai.OpenAI(api_key='your api-key')
# file_id=client.files.create(file=open('demo.csv'), purpose='assistants').id
# Step 1: Create an Assistant
assistant = client.beta.assistants.create(
name="SubscriptionFlow Assistant",
instructions="You are a professional SubscriptionFlow assistant. Write and run user's queries and answer them with relevance to SubscriptionFlow.",
model="gpt-4-1106-preview",
tools=tools_list,
)
# Step 2: Create a Thread
thread = client.beta.threads.create()
# Step 3: Add a Message to a Thread
message = client.beta.threads.messages.create(
thread_id=thread.id,
role="user",
content='data={"name": "Test Product","description": "This is a test product"}'
)
# Step 4: Run the Assistant
run = client.beta.threads.runs.submit_tool_outputs(
thread_id=thread.id,
assistant_id=assistant.id,
instructions="Please address the user as Shaff Khan. The user has a premium account.",
)
print(run.model_dump_json(indent=4))
while True:
# Wait for 5 seconds
time.sleep(5)
# Retrieve the run status
run_status = client.beta.threads.runs.retrieve(
thread_id=thread.id,
run_id=run.id
)
print(run_status.model_dump_json(indent=4))
# If run is completed, get messages
if run_status.status == 'completed':
messages = 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}")
break```