OpenAI assistants step-by-step
Why assistants?
Maybe we should ask before, what are assistants? From a simple view assistants are an easier approach from OpenAI to simplify Generative AI/ RAG tools, -> tools you use to build applications using an LLM to find answers based on specific data (e.g. your company data).
OpenaAI assistants are adding a lot of power for building apps over LLMs. It goes a step beyond the basic approach of using function calls, like OpenAI API or the Langchain framework, mainly by adding some facilities:
- simpler functions definition. Just add your code and a JSON schema explaining what your function does and your function is ready to work!
- unlimited prompt length. There are no more problems with large prompts (input or outputs), assistants take care of the dirty work
- prompt memory. You don’t need to recall details of your conversation to the assistant, it manages the old messages to keep track of your conversation
- conversations history. Conversations are automatically stored, just like ChatGPT does.
- Extra tools: Retrieval (including doc files) and Code interpreter (will will talk about these two later).
How OpenAI assistants work?
Note: to follow the process in this article you need an OpenAI account with an API key.
There are two ways to make an assistant:
- you can create your own assistant from the OpenAI site and then use it to check its abilities on the Playground tool (check the image below)
2. code it with Python (see next)
OpenAI assistant code for building an expert stock data analyst
This sample code uses the Yahoo Finance API yfinance to answer user questions about market behavior (you can find the whole code in my repository: https://github.com/lalo-fuentes/openai-assistant.git)
Before diving into the code let me try to explain how the assistant objects structure works:
- assistant: this object contains information about the model (gpt-4, gpt-3.5-turbo), the tools (functions, retrieve, code_interpreter), instructions (what the assistant does), …
- thread: contains messages (keeps the messages history)
- messages: user messages (queries), system messages (AI responses)
- run: executes threads
- steps: steps done by runs (like tool calls)
The code
We are going to divide our code into two modules: the functions and the app.
Functions to interact with the external world, like fetching data from an API, will be in the functions module (will see the contents later)
So let’s dive into the main module assistant.py
First of all, let’s declare the libraries and get the OpenAI API key
from openai import OpenAI
import os
import time
from def_tools import *
import json
# Import API keys
import a_env_vars
os.environ["OPENAI_API_KEY"] = a_env_vars.OPENAI_API_KEYThen, we initiate or create the assistant:
# Create assistant
client = OpenAI()
assist_name ='Market.00'
# --------------------------------------------------------------------
def initiate_assistant():
system_msg = '''
- You are an expert stock data market analyst
- You can fetch reliable data through the functions available
- Always use get_todays_date function when you need to know today's date or a relative date
'''
# check for existing assistants
my_assistants = client.beta.assistants.list(
order="desc",
limit="5",
)
result = fetch_assistant_by_name(my_assistants.data, assist_name)
if result is None:
my_assistant = client.beta.assistants.create(
instructions=system_msg,
name=assist_name,
tools=ftools,
model="gpt-3.5-turbo",
)
else:
my_assistant = client.beta.assistants.retrieve(result.id)
return my_assistant
# -------------------------------------------
def fetch_assistant_by_name(data, assist_name):
for assistant in data:
if assistant.name == assist_name:
return assistant
return None
In this app, we always use the same assistant. If it doesn’t exist we create it, if it was already created we fetch it by its name. Of course, you can use the assistants as you want, you can create a new one each time you start your application or even delete the assistants before closing your app.
Note: OpenAI offers you the ability to manage your assistant from your OpeanAI dashboard (check Assistants in your menu). You can create, modify, list, retrieve, or delete, your assistants from the code.
Then we create the thread:
# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
def open_thread():
new_thread = client.beta.threads.create()
return new_threadWe have a very simple approach, we just create a new thread each time we enter the app. The OpenAI library allows you to create, modify, retrieve, or delete your threads.
Note: As in the assistant, you can manage your threads from the OpenAI dashboard. Go to Settings / Organization / Features and capabilities, to make them visible in your dashboard.
Now we add a user message to the thread
# --------------------------------------------------------------------
# add user message to current thread
def add_query(thread, query):
thread_message = client.beta.threads.messages.create(
thread.id,
role="user",
content=query,
)
return thread_messageFinally, we run the thread, i.e. we ask for a reply to the model
# ---------------------------------------------------------------------------------------
def runOpenai(thread_id,assistant_id):
run = client.beta.threads.runs.create(
thread_id=thread_id,
assistant_id= assistant_id,
)
while True:
run = client.beta.threads.runs.retrieve(thread_id=thread_id, run_id=run.id)
print(run.status)
if run.status == 'completed':
break
elif run.status == 'requires_action':
tool_calls = run.required_action.submit_tool_outputs.tool_calls
tool_outputs = callTools(tool_calls)
run = client.beta.threads.runs.submit_tool_outputs(
thread_id=thread_id,
run_id=run.id,
tool_outputs=tool_outputs
)
time.sleep(.5)
return runCreate the run, then, retrieve the run until it is ‘completed’.
If your user query involves the use of a function the run.status will ask about ‘requires_action’. tool_calls will contain the function name the LLM is asking to call.
We are using the function callTools() to call the functions the model is proposing. Check here:
# ------------------------------------------------------------------------------------------------------------
def callTools(tool_calls):
tool_outputs = []
for t in tool_calls:
functionName = t.function.name
attributes = json.loads(t.function.arguments)
args = list(attributes.values())
# Get the real function from its name
function = globals().get(functionName)
try:
if args:
functionResponse = function(*args)
else:
functionResponse = function()
except Exception as e:
functionResponse = { "status" : 'Error in function call ' + functionName + '(' + t.function.arguments + ')', "error": str(e) }
tool_outputs.append({ "tool_call_id": t.id , "output": json.dumps(functionResponse) })
return tool_outputsAs you may noticed, we are dumping the response into a JSON schema (preferred by LLMs)
And finally, this is the main logic:
# ---------------- main logic -----------------------------------------
my_assistant = initiate_assistant()
thread = open_thread()
while True:
query = input("Enter your query: ")
message = add_query(thread, query)
run = runOpenai(thread.id, my_assistant.id)
messages = client.beta.threads.messages.list(thread_id=thread.id)
response = messages.data[0].content[0].text.value
print(response)The functions module
Wait, wait, of course, there is still the functions module missing, here you go:
import json
import yfinance as yf
from datetime import date
# functions schemas
ftools = [{
"type": "function",
"function": {
"name": "get_stock_price",
"description": "Useful when you want to get the closing price of a stock between two dates.",
"parameters": {
"type": "object",
"properties": {
"ticker": {"type": "string", "description": "The stock ticker symbol"},
"start_date": {"type": "string", "description": "The first date to fetch data (YYYY-MM-DD)"},
"end_date": {"type": "string", "description": "The last date to fetch data + 1 day (YYYY-MM-DD)"}
},
"required": ["ticker", "start_date", "end_date"]
}
}
},
{
"type": "function",
"function": {
"name": "get_todays_date",
"description": "Useful when you want to know today's date. Useful when the user asks for data from a relative date, like: last price, current price, last week price, last month price, etc.",
"parameters": {
"type": "object",
"properties": {}
}
}
}]
def get_stock_price(ticker, start_date, end_date):
ticker_data = yf.Ticker(ticker)
data = ticker_data.history(period="1d", start=start_date, end=end_date)
data_list = []
for i in range(len(data)):
data_list.append({"date": data.index[i].strftime("%Y-%m-%d"), "price": data.iloc[i]["Close"], "currency": ticker_data.info["currency"]})
return json.dumps(data_list)
def get_todays_date():
# Get todays date
today = date.today()
# Convert date to ISO 8601 chain format
today_iso = today.isoformat()
# return a JSON object date
return json.dumps({"today": today_iso})
# ---------------------------------------------------------------------------As you can see you need to declare first your functions in a JSON mode so the LLM can understand what they do and what arguments they need.
Then, just write your code to define your functions. These functions are returning JSON data, this is not mandatory but is a good practice to interact with LLMs.
Results
This is a sample of an iteration with the app. I’m printing the run.status just for better comprehension. The first action the LLM takes is asking for today’s date (models don’t know what day it is), then, it asks for yesterday’s closing price passing the right arguments: ticker, yesterday_date, and yesterday_date+1. Finally, it prints the answer.
Enter your query: give me the closing price of tesla from yesterday
in_progress
in_progress
requires_action
get_todays_date
[]
in_progress
in_progress
requires_action
get_stock_price
['TSLA', '2024-02-26', '2024-02-27']
in_progress
completed
The closing price of Tesla from yesterday (February 26, 2024) was $199.40 USD.Extra tools
As stated at the beginning of this article assistants offer you the capability, besides the functions tools already seen, to use two powerful built-in tools: Retrieval (including files) and Code interpreter
- The retrieval will allow you to add files to your account (.pdf, .md, .docx. ,csv, and many more) and then retrieve data for your app. e.g. a chatbot that finds responses from company documentation.
- Code interpreter. This can be a very powerful tool if it works as promised. “Code Interpreter allows the Assistants API to write and run Python code in a sandboxed execution environment. This tool can process files with diverse data and formatting, and generate files with data and images of graphs. Code Interpreter allows your Assistant to run code iteratively to solve challenging code and math problems. When your Assistant writes code that fails to run, it can iterate on this code by attempting to run different code until the code execution succeeds.” https://platform.openai.com/docs/assistants/tools/code-interpreter
How you use them:
assistant = client.beta.assistants.create(
instructions="You are a personal math tutor. When asked a math question, write and run code to answer the question.",
model="gpt-4-turbo-preview",
tools=[{"type": "code_interpreter"}]
)Final considerations
- Assistants are still on Beta, so expect the unexpected
- Working with your data gives you some stability, but you are never safe from hallucinations
- Assistants are big token eaters -> $$$. If you are using lots of data to retrieve and process prepare your wallet