OpenAI Assistant With Flask — A Simple Example With Code
Introduction
In the digital realm, where innovation moves at the speed of light, OpenAI’s ChatGPT has been a game-changer, becoming a household name and introducing the masses to the capabilities of Large Language Models (LLMs). By the end of 2023, OpenAI took another leap forward by unveiling its Assistant API, a development poised to revolutionize our interaction with software as profoundly as ChatGPT has transformed our online work habits. This new API is not just an iteration; it’s a paradigm shift designed to democratize the power of Large Action Models (LAMs) for developers worldwide. The Assistant API’s mission is to bridge the gap between user requests and actionable responses, enabling developers to craft functions that are dynamically triggered by LLMs, enriching user interactions with real-time, context-aware actions. Imagine a scenario where a simple user query like “What’s the weather in Paris?” could seamlessly activate a developer-defined function getWeather(City), integrating its output directly into the LLM's response. This melding of language understanding and actionable intelligence heralds a new era of interactive technology.
The focus of this article is to demystify the Assistant API through a tangible example: managing user permissions. This implementation serves as a practical assistant for IT teams, allowing them to interact with back-end systems and databases through everyday language. Accompanying this article, you’ll find the source code and a demonstration video that together offer a comprehensive guide to harnessing the Assistant API’s potential. Our journey through this example will not only illustrate the API’s capabilities but also inspire you to explore its limitless applications in your development endeavors.
Code can be found here: Github repo.
How OpenAI Assistant API works ?
The core idea behind the Assistant API is to enable developers to effortlessly develop and link actions to the execution of an LLM. More concretely, when a user sends a message to an assistant, the following three stages unfold:
- (1) The LLM analyzes the user’s query within the context of the ongoing conversation. Depending on the question, the LLM decides whether to trigger functions previously defined by the developer.
- (2) The LLM then pauses to wait for function executions (by the server).
- (3) The results from the executed functions are incorporated to generate a quality response.
On a more technical level, the Assistant API facilitates several key functionalities:
- Creation and definition of an assistant (its purpose, which LLM to use, etc.),
- Management of conversations (threads),
- Management of messages (both from the user and the assistant),
- Creation and management of the functions that the LLM will request the system to execute,
- And finally, running the assistant based on new messages from the user.
In this article, I focus on the ability to write one’s own functions (referred to as “function calling”), but it’s worth noting that OpenAI Assistant also provides two highly useful functions that allow for code execution (Code Interpreter) and Knowledge Retrieval (by requesting users to upload files, for example).
In the following section, I delve more technically (with code examples) into the workings of the Assistant API.
Deep Dive into the Code
In order to make it simple, Flask project includes everything: frontend and backend. The purpose of this section is to explain the assistant service that can be found in app/services/assistant.py. Permission management is defined in app/services/permissions.py; as you can see this service is used by classic permission endpoints and assistant service.
Initialization of the service
First of all, let’s analyse the initialization of the service:
def __init__(self):
self.openAI = OpenAI(api_key=Config.OPENAI_KEY)
self.assistant_name = 'IT Administrator Assistant'
self.model_id= 'gpt-4-turbo-preview'
self.instruction= 'You are an IT administrator. You are responsible for managing user permissions. You have a list of users and their permissions. You can get the permissions of a user by their username, update the permissions of a user by their username, and get the user ID based on the username. You can use the following functions: getPermissionsByUsername, updatePermissionsByUsername, getUserIdByUsername.'
self.assistant = self.get_or_create_assistant()
self.thread = self.get_or_create_thread()
def get_or_create_assistant(self):
if 'assistant_id' in session:
current_app.logger.info('Getting assistant')
return self.openAI.beta.assistants.retrieve(session['assistant_id'])
else:
current_app.logger.info('Creating assistant')
assistant = self.create_assistant()
session['assistant_id'] = assistant.id
return assistant
def get_or_create_thread(self):
if 'thread_id' in session:
current_app.logger.info('Getting thread')
return self.openAI.beta.threads.retrieve(session['thread_id'])
else:
current_app.logger.info('Creating thread')
thread = self.create_thread()
session['thread_id'] = thread.id
return thread
def create_assistant(self):
current_app.logger.info('Creating assistant')
return self.openAI.beta.assistants.create(
name=self.assistant_name,
instructions=self.instruction,
model=self.model_id,
tools=[
self.define_function__get_permissions_by_username(),
self.define_function__get_user_id_by_username(),
self.define_function__update_user_permission()
]
)
def create_thread(self):
current_app.logger.info('Creating thread')
return self.openAI.beta.threads.create()During initialization I define the name of the assistant, its purpose and the LLM model used to generate responses. Here, self.instructions is important since the quality of your assistant will depends on them.
In order to keep the conversation, I implemented a way to create a new assistant and thread only when there is no existing ones. Also, using OpenAI Assistant API, you pay per thread, so be careful and try not to create a new thread or assistant every time.
Assistant functions
In order to detect if a function should be called, developers need to explicitly define these functions to the assistant. Here are my function definitions.
def define_function__get_user_id_by_username(self):
function = {
"type": "function",
"function": {
"name": "getUserIdByUsername",
"description": "Get the user ID based on the username.",
"parameters": {
"type": "object",
"properties": {
"username": {"type": "string", "description": "The username of the user."}
},
"required": ["username"]
}
}
}
return function
def define_function__get_permissions_by_username(self):
function = {
"type": "function",
"function": {
"name": "getPermissionsByUsername",
"description": "Get the permissions of a user by their username.",
"parameters": {
"type": "object",
"properties": {
"username": {"type": "string", "description": "The username of the user."}
},
"required": ["username"]
}
}
}
return function
def define_function__update_user_permission(self):
function = {
"type": "function",
"function": {
"name": "updateUserPermission",
"description": "Update the value of a permission for a user by their username.",
"parameters": {
"type": "object",
"properties": {
"username": {"type": "string", "description": "The username of the user."},
"permission": {"type": "string", "description": "The permission to update."},
"value": {"type": "boolean", "description": "The new value of the permission."}
},
"required": ["username"]
}
}
}
return functionNote that I explicitly described the function. As you can understand now, the magic happens here.
Function code
So let’s consider that the LLM is waiting for some function results. Developers need to write the code of these functions. In my case, I took care to write less and call other services so the code is generic enough.
def getUserIdByUsername(self, username):
current_app.logger.info(f'getUserIdByUsername: {username}')
user_id = current_app.permissions_service.get_user_id_by_username(username)
if user_id:
current_app.logger.info(f'User found with id: {user_id}')
return user_id
else:
current_app.logger.info('User not found')
return "No user found"
def getPermissionsByUsername(self, username):
current_app.logger.info(f'getPermissionsByUsername: {username}')
user_id = current_app.permissions_service.get_user_id_by_username(username)
if user_id:
current_app.logger.info(f'User found with id: {user_id}')
permissions = current_app.permissions_service.get_permissions_by_user_id(user_id)
current_app.logger.info(permissions)
return permissions
current_app.logger.info('User not found')
return "No user found"
def updateUserPermission(self, username, permission, value):
current_app.logger.info(f'updateUserPermission: {username}, {permission}, {value}')
user_id = current_app.permissions_service.get_user_id_by_username(username)
if user_id:
current_app.logger.info(f'User found with id: {user_id}')
updated_permission = current_app.permissions_service.update_user_permission(user_id, permission, value)
current_app.logger.info(updated_permission)
return updated_permission
else:
current_app.logger.info('User not found')
return "No user found"Run the assistant when a new user message is received
If you understand the following code block, then you’re ready to develop your own assistant.
def add_message_to_thread(self, role, message):
current_app.logger.info(f'Adding message to thread: {role}, {message}')
return self.openAI.beta.threads.messages.create(
thread_id=self.thread.id,
role=role,
content=message,
)
def run_assistant(self, message):
current_app.logger.info(f'Running assistant: {message}')
message = self.add_message_to_thread("user", message)
action_response = None
run = self.openAI.beta.threads.runs.create(
thread_id=self.thread.id,
assistant_id=self.assistant.id,
)
run = self.wait_for_update(run)
if run.status == "failed":
current_app.logger.info('Run failed')
return None
elif run.status == "requires_action":
current_app.logger.info(f'Run requires action: {run}')
action_response = self.handle_require_action(run)
else:
current_app.logger.info('Run completed')
action_response = self.get_last_assistant_message()
return action_response
def handle_require_action(self, run):
current_app.logger.info('Handling required action')
# Get the tool outputs by executing the required functions
tool_calls = run.required_action.submit_tool_outputs.tool_calls
current_app.logger.info(tool_calls)
tool_outputs = self.generate_tool_outputs(tool_calls)
# Submit the tool outputs back to the Assistant
run = self.openAI.beta.threads.runs.submit_tool_outputs(
thread_id=self.thread.id,
run_id=run.id,
tool_outputs=tool_outputs
)
run = self.wait_for_update(run)
if run.status == "failed":
current_app.logger.info('Run failed')
return None
elif run.status == "completed":
return self.get_last_assistant_message()
def wait_for_update(self, run):
while run.status == "queued" or run.status == "in_progress":
run = self.openAI.beta.threads.runs.retrieve(
thread_id=self.thread.id,
run_id=run.id,
)
time.sleep(1)
current_app.logger.info(run.status)
return run
def get_last_assistant_message(self):
current_app.logger.info('Getting last assistant message')
messages = self.openAI.beta.threads.messages.list(thread_id=self.thread.id)
if messages.data[0].role == 'assistant':
message = messages.data[0]
for content_block in message.content:
if content_block.type == 'text':
return content_block.text.value
else:
return None
def generate_tool_outputs(self, tool_calls):
current_app.logger.info('Generating tool outputs')
tool_outputs = []
for tool_call in tool_calls:
function_name = tool_call.function.name
arguments = tool_call.function.arguments
tool_call_id = tool_call.id
args_dict = json.loads(arguments)
if hasattr(self, function_name):
function_to_call = getattr(self, function_name)
output = function_to_call(**args_dict)
tool_outputs.append({
"tool_call_id": tool_call_id,
"output": output,
})
return tool_outputsEverytime you run the assistant, you have to check the status of the run. If the LLM detected the need to execution a function, then the status will be set to “requires_action”. Using the run response, you are able to extract function names that the assistant is waiting for and its parameters.
And that’s it ! You are ready to re-code Google Home in 1 day :) !
Play with it
The code can be found here in this Github page. Instructions to run the flask project can be found in the README.MD file.
The following video is a demo of what can be done without any code modification.
Conclusion
In this article, we’ve delved into the workings of Large Action Models (LAMs) by exploring OpenAI’s Assistant API. By providing the code, my aim is to empower each reader to not only benefit from these insights but also to share their own innovations and experiences. I believe that LAMs hold the potential to overcome many of the limitations currently faced by Large Language Models (LLMs), positioning OpenAI’s Assistant API as a potential game-changer in the realm of technological advancements. The integration of actionable intelligence with the nuanced understanding of language models opens up a myriad of possibilities for developers and businesses alike. As we stand on the brink of this new frontier, I am filled with anticipation for the evolution of these technologies and their applications. The future, enriched by the capabilities of LAMs and the continuous innovations in artificial intelligence, looks promising, and I am thrilled to be part of this journey toward a more interactive and intuitive digital world.
If you have any question about this post, contact me in LinkedIn.
