|
from openai import OpenAI
|
|
from tools import *
|
|
|
|
|
|
OPENAI_CLIENT = OpenAI()
|
|
|
|
|
|
|
|
App_state = {
|
|
"messages":[],
|
|
"sys_msg":[{
|
|
"role": "system",
|
|
"content": "Hi, I'm a Factory supply chain assistant! Ask me a question."
|
|
}],
|
|
"language":"en",
|
|
"voice":"nova",
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def send_chat(text,messages=[]):
|
|
'''
|
|
send_chat function takes two arguments: text and messages
|
|
|
|
text: the text that the user inputs
|
|
messages: a list of tuples, each tuple contains two strings, the first string is the user input, the second string is the assistant output
|
|
|
|
return: a tuple, the first element is an empty string, the second element is the updated messages list
|
|
|
|
'''
|
|
App_state["sys_msg"].extend([{ "role": "user", "content": text }])
|
|
|
|
run(text)
|
|
|
|
return "",App_state["messages"]
|
|
|
|
def run(text,depth=0,max_depth=5):
|
|
if depth >= max_depth:
|
|
return "I'm sorry, the inquiry is too complex for me to handle. Please try again with a simpler question."
|
|
res = OPENAI_CLIENT.chat.completions.create(
|
|
model="gpt-3.5-turbo",
|
|
messages=App_state["sys_msg"],
|
|
tools=TOOL_SCHEMA,
|
|
tool_choice="auto"
|
|
)
|
|
|
|
App_state["sys_msg"].append(res.choices[0].message.model_dump())
|
|
|
|
if res.choices[0].message.tool_calls:
|
|
tool_calls = res.choices[0].message.tool_calls
|
|
tool_response = run_tool(tool_calls[0])
|
|
tool_message = {
|
|
"role":"tool",
|
|
"tool_call_id":tool_calls[0].id,
|
|
"name" : tool_calls[0].function.name,
|
|
"content":str(tool_response)
|
|
}
|
|
App_state["sys_msg"].append(tool_message)
|
|
run(text,depth+1,max_depth)
|
|
else:
|
|
App_state["messages"].append((text,res.choices[0].message.content))
|
|
|
|
def run_tool(tool_call):
|
|
tool_call_schema = tool_call.function.model_dump()
|
|
tool_name = tool_call_schema.get("name")
|
|
tool_args = json.loads(tool_call_schema.get("arguments"))
|
|
if tool_args:
|
|
result = TOOLS[tool_name](tool_args)
|
|
else:
|
|
result = TOOLS[tool_name]()
|
|
|
|
return result
|
|
|
|
def translate(file_path):
|
|
if file_path:
|
|
f = open(file_path,"rb")
|
|
res = OPENAI_CLIENT.audio.translations.create(
|
|
file=f,
|
|
model="whisper-1")
|
|
return res.text
|
|
else:
|
|
return ""
|
|
|
|
def text_to_audio(chat_messages):
|
|
text = chat_messages[-1][-1]
|
|
response = OPENAI_CLIENT.audio.speech.create(
|
|
model="tts-1",
|
|
voice=App_state["voice"],
|
|
input=text,
|
|
|
|
)
|
|
|
|
return response.content |