Overview
In this article, I’ll be covering some of the capabilities of the Assistants API capability introduced by OpenAI in November 2023 during their first DevDay.
Standing to OpenAI’s description of it: “that makes it easier for developers to build their own assistive AI apps that have goals and can call models and tools”.
Basically, an assistant is a purpose-built AI that has specific instructions, leverages extra knowledge, and can call models and tools to perform tasks. Assistants API provides new capabilities such as Code Interpreter and Retrieval as well as Function Calling to handle a lot of the heavy lifting that you previously had to do yourself and enable you to build high-quality AI apps.
In this article, I’ll focus on Function Calling capability but I’d recommend you to read the Azure OpenAI Assistants documentation to know more about the other capabilities.
What is Function Calling?
Function Calling allows you to connect Large Language Models (e.g. GPT-4 Turbo from OpenAI) to external tools or APIs. In an API call, you can describe functions and have the model itself determined to output a JSON object containing arguments for one or more functions. It doesn’t directly call the function, instead it generates a JSON that you can use to call the function.
Typical sequence of steps
- Call the model with the user query and a set of functions defined in the functions parameter.
- The model may choose to call one or more functions, and the content will be a stringified JSON object adhering to your custom schema.
- Parse the string into JSON in your code and call your function with the provided arguments (if they exist).
- Call the model again by appending the function response as a new message, allowing the model to summarize the results back to the user.
Example
In today’s example, I’ll be showcasing an example where I’ll create a “Weather assistant” that is able to get current weather on any location. To do so, I’ll have to:
- Create an Assistant and personalize it through instructions and attach a specific model.
- Add the get_weather signature function to the assistant.
- Write the get_weather function in Python.
- Run a conversation (called a “thread”) with the assistant and start asking question about the weather.
1. Create an Assistant
First, I go on the Azure OpenAI Studio interface to create the assistant (I can do it programmatically as well, but I’ll stay on the interface for that step).
On the “Assistants” tab I can access the dedicated interface where I click on “New” and I specify:
- Assistant name (that’s the name of the assistant).
- Instructions (like the system prompt which guides the personality of the assistant and defines its goals.
- Deployment (the LLM that the assistant will be attached to)
I can save my assistant and it will return to me an assistant ID specific to my Weather assistant.
2. Add the get_weather signature function
For now, I have only attached “guidelines” (though the instructions) to my Weather assistant, but my assistant can’t answer me the weather in Paris right now as it would need access to real-time data on internet through an external API for instance. In my example I want the assistant to be able to return me current weather in a specific location when I ask a question like “what’s the current weather in Paris?”.
To make it happen, I’ll “Add function” directly through the interface on below the “Assistant tools” section.
In this section, I specify a custom function for the assistant to call. The assistant would output the function arguments but would never run the function itself.
Here is the signature of the get_weather function I’ll write after in Python:
{
"name": "get_weather",
"description": "Determine weather in a certain location",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "The city and country e.g. Paris (France)"
}
},
"required": [
"location"
]
}
}
The description field is important as it will be used by the assistant to determine which function is needed to resolve a task.
On this function signature I specify:
- the name of the function (get_weather).
- the parameters, their names (location), their types (string) and their description.
- the required parameters for the function to work properly.
Then I’ve saved the function and save this modification to my assistant. I can see that it has updated the assistant directly through a POST API call (see the screenshot on the Logs tab on the right).
3. Write the get_weather function
Now I need to write the get_weather function to call this function and get the data I want to pass to my assistant.
I’m writing this function in Python and below is the code:
def get_weather(location):
base_url = "https://api.openweathermap.org/data/2.5/weather?"
complete_url = base_url + "q=" + location + "&appid=" + api_key + "&units=metric"
response = requests.get(complete_url)
weather_condition = response.json()["weather"][0]["description"]
temperature = response.json()["main"]["temp"]
return f"""Here is some information about the weather in {location}:\n
- The weather is: {weather_condition}.\n
- The temperature is: {temperature} degrees Celsius.\n
"""
This function is taking as an argument the location that is deducted by the assistant in the initial prompt (e.g. “What’s the current weather in Paris” will isolate “Paris” as the value of the location parameter).
Then it sends an API call to OpenWeatherMap which is an online service that provides global weather data via API (I had to create a free account before to get my API credentials that I pass into the api_key variable).
Through tha OpenWeatherMap endpoint that I call, I pass the location parameter to get current weather information on the specific location such as weather description and temperature (in Celsius degrees).
Then it returns a phrase in English which gives me some information about the weather. That return is key for the assistant as it will generate a response to the user based on this function’s return.
4. Run a conversation with the assistant
I won’t go into many details, but the primitives of the Assistants API are:
- Assistants, which encapsulate a base model (e.g. GPT-4 Turbo here), instructions, tools, and (context) documents.
- Threads, which represent the state of a conversation.
- Runs, which power the execution of an Assistant on a Thread, including textual responses and multi-step tool use.
Below is the code I use to :
- Declare all my environment variables (e.g. Azure OpenAI and OpenWeatherMap).
- Define the get_weather Python function.
- Python functions that I use to streamline the process of sending messages to the assistant thread.
- Print responses from my Weather assistant.
import openai
import os
import json
import time
import requests
from dotenv import load_dotenv
from pathlib import Path
from openai import AzureOpenAI
from typing import Optional
load_dotenv("./azure.env")
openai.api_type: str = "azure"
AZURE_OPENAI_API_KEY = os.getenv("AZURE_OPENAI_API_KEY")
AZURE_OPENAI_API_ENDPOINT = os.getenv("AZURE_OPENAI_API_ENDPOINT")
AZURE_OPENAI_API_VERSION = os.getenv("AZURE_OPENAI_API_VERSION")
client = AzureOpenAI(
api_key=AZURE_OPENAI_API_KEY,
api_version= AZURE_OPENAI_API_VERSION,
azure_endpoint=AZURE_OPENAI_API_ENDPOINT
)
api_key = os.getenv("OPENWEATHER_API")
def get_weather(location):
base_url = "https://api.openweathermap.org/data/2.5/weather?"
complete_url = base_url + "q=" + location + "&appid=" + api_key + "&units=metric"
response = requests.get(complete_url)
weather_condition = response.json()["weather"][0]["description"]
temperature = response.json()["main"]["temp"]
return f"""Here is some information about the weather in {location}:\n
- The weather is: {weather_condition}.\n
- The temperature is: {temperature} degrees Celsius.\n
"""
def poll_run_till_completion(
client: AzureOpenAI,
thread_id: str,
run_id: str,
available_functions: dict,
verbose: bool,
max_steps: int = 10,
wait: int = 3,
) -> None:
"""
Poll a run until it is completed or failed or exceeds a certain number of iterations (MAX_STEPS)
with a preset wait in between polls
client: Azure OpenAI client
thread_id: Thread ID
run_id: Run ID
assistant_id: Assistant ID
verbose: Print verbose output
max_steps: Maximum number of steps to poll
wait: Wait time in seconds between polls
"""
if (client is None and thread_id is None) or run_id is None:
print("Client, Thread ID and Run ID are required.")
return
try:
cnt = 0
while cnt < max_steps:
run = client.beta.threads.runs.retrieve(thread_id=thread_id, run_id=run_id)
if verbose:
print("Poll {}: {}".format(cnt, run.status))
cnt += 1
if run.status == "requires_action":
tool_responses = []
if (
run.required_action.type == "submit_tool_outputs"
and run.required_action.submit_tool_outputs.tool_calls is not None
):
tool_calls = run.required_action.submit_tool_outputs.tool_calls
for call in tool_calls:
if call.type == "function":
if call.function.name not in available_functions:
raise Exception("Function requested by the model does not exist")
function_to_call = available_functions[call.function.name]
tool_response = function_to_call(**json.loads(call.function.arguments))
tool_responses.append({"tool_call_id": call.id, "output": tool_response})
run = client.beta.threads.runs.submit_tool_outputs(
thread_id=thread_id, run_id=run.id, tool_outputs=tool_responses
)
if run.status == "failed":
print("Run failed.")
break
if run.status == "completed":
break
time.sleep(wait)
except Exception as e:
print(e)
def create_message(
client: AzureOpenAI,
thread_id: str,
role: str = "",
content: str = "",
file_ids: Optional[list] = None,
metadata: Optional[dict] = None,
message_id: Optional[str] = None,
) -> any:
"""
Create a message in a thread using the client.
client: OpenAI client
thread_id: Thread ID
role: Message role (user or assistant)
content: Message content
file_ids: Message file IDs
metadata: Message metadata
message_id: Message ID
@return: Message object
"""
if metadata is None:
metadata = {}
if file_ids is None:
file_ids = []
if client is None:
print("Client parameter is required.")
return None
if thread_id is None:
print("Thread ID is required.")
return None
try:
if message_id is not None:
return client.beta.threads.messages.retrieve(thread_id=thread_id, message_id=message_id)
if file_ids is not None and len(file_ids) > 0 and metadata is not None and len(metadata) > 0:
return client.beta.threads.messages.create(
thread_id=thread_id, role=role, content=content, file_ids=file_ids, metadata=metadata
)
if file_ids is not None and len(file_ids) > 0:
return client.beta.threads.messages.create(
thread_id=thread_id, role=role, content=content, file_ids=file_ids
)
if metadata is not None and len(metadata) > 0:
return client.beta.threads.messages.create(
thread_id=thread_id, role=role, content=content, metadata=metadata
)
return client.beta.threads.messages.create(thread_id=thread_id, role=role, content=content)
except Exception as e:
print(e)
return None
def retrieve_and_print_messages(
client: AzureOpenAI, thread_id: str, verbose: bool, out_dir: Optional[str] = None
) -> any:
"""
Retrieve a list of messages in a thread and print it out with the query and response
client: OpenAI client
thread_id: Thread ID
verbose: Print verbose output
out_dir: Output directory to save images
@return: Messages object
"""
if client is None and thread_id is None:
print("Client and Thread ID are required.")
return None
try:
messages = client.beta.threads.messages.list(thread_id=thread_id)
display_role = {"user": "User query", "assistant": "Assistant response"}
prev_role = None
if verbose:
print("\n\nCONVERSATION:")
for md in reversed(messages.data):
if prev_role == "assistant" and md.role == "user" and verbose:
print("------ \n")
for mc in md.content:
# Check if valid text field is present in the mc object
if mc.type == "text":
txt_val = mc.text.value
# Check if valid image field is present in the mc object
elif mc.type == "image_file":
image_data = client.files.content(mc.image_file.file_id)
if out_dir is not None:
out_dir_path = Path(out_dir)
if out_dir_path.exists():
image_path = out_dir_path / (mc.image_file.file_id + ".png")
with image_path.open("wb") as f:
f.write(image_data.read())
if verbose:
if prev_role == md.role:
print(txt_val)
else:
print("{}:\n{}".format(display_role[md.role], txt_val))
prev_role = md.role
return messages
except Exception as e:
print(e)
return None
available_functions = {"get_weather": get_weather}
verbose_output = True
# Declare the Assistant's ID
assistant_id = "asst_xxxxyyyyzzzz"
# Fetch the assistant
assistant = client.beta.assistants.retrieve(
assistant_id=assistant_id
)
assistant
# Create a thread
thread = client.beta.threads.create()
thread
# Prompt the model to tell us all about the data provided
run = client.beta.threads.runs.create(
thread_id=thread.id,
assistant_id=assistant.id,
instructions="What's the current weather in Paris?",
)
poll_run_till_completion(
client=client, thread_id=thread.id, run_id=run.id, available_functions=available_functions, verbose=verbose_output
)
messages = retrieve_and_print_messages(client=client, thread_id=thread.id, verbose=verbose_output)
Now, I’ll explain a bit more about what happens behind the scenes. Let’s have a look at the JSON that has been returned by the assistant.
{'id': 'run_k5j1iqvvGyOQIKO35h8UsnSi',
'assistant_id': 'asst_xxxxyyyyzzzz',
'cancelled_at': None,
'completed_at': None,
'created_at': 1711094224,
'expires_at': 1711094824,
'failed_at': None,
'file_ids': [],
'instructions': "What's the current weather in Paris?",
'last_error': None,
'metadata': {},
'model': 'gpt-4-turbo',
'object': 'thread.run',
'required_action': {'submit_tool_outputs': {'tool_calls': [{'id': 'call_NmzIJAv4gq6chMziwzYOlWIt',
'function': {'arguments': '{"location":"Paris, France"}',
'name': 'get_weather'},
'type': 'function'}]},
'type': 'submit_tool_outputs'},
'started_at': 1711094225,
'status': 'requires_action',
'thread_id': 'thread_0QbGfJ9G6fAzMDihRuvH679F',
'tools': [{'type': 'code_interpreter'},
{'function': {'name': 'get_weather',
'description': 'Determine weather in a certain location',
'parameters': {'type': 'object',
'properties': {'location': {'type': 'string',
'description': 'The city and country e.g. Paris (France)'}},
'required': ['location']}},
'type': 'function'}],
'usage': None}
When I sent the message “What the current weather in Paris?” to my assistant, it has determines my intent and if a function that it has can accomplish this task.
Effectively, it has been determined that its next required_action is to call the get_weather function with the location argument with the value of "Paris, France".
Then I need to use the get_weather Python function and pass the argument that has been found by the assistant, which returns this JSON:
[{'tool_call_id': 'call_NmzIJAv4gq6chMziwzYOlWIt',
'output': 'Here is some information about the weather in Paris, France:\n\n - The weather is: clear sky.\n\n - The temperature is: 19.55 degrees Celsius.\n\n '}]
Finally, I send this JSON response to the assistant to enable it to interpret this response and get back to me with a generated answer, based on the Python function JSON response.
When I run this in a notebook it gives me the following output (we’re pretty lucky today in Paris!):
Now, if I want to continue this conversation with a follow up question (e.g. “convert into Fahrenheit degrees”), I need to stay within the same thread ID so I run the code below in a separate notebook block:
run = client.beta.threads.runs.create(
thread_id=thread.id,
assistant_id=assistant.id,
instructions="convert into Fahrenheit degrees",
)
poll_run_till_completion(
client=client, thread_id=thread.id, run_id=run.id, available_functions=available_functions, verbose=verbose_output
)
messages = retrieve_and_print_messages(client=client, thread_id=thread.id, verbose=verbose_output)
It keeps the history of previous messages, and it responds to me with:
Conclusion
Today I’ve covered an example where I leverage the Assistants API from Azure OpenAI to create a Weather assistant with instructions and a function that the assistant determines when to call it or not and deducts the argument needed in the user’s prompt to then pass it to the Python function that gets information regarding the current weather in a specific location.
There are more use-cases where Assistants and Function calling are useful when it comes to integrating the power of Large Language Models and external APIs sources and I strongly recommend you to spend time to investigate more to simplify, enhance, or automate some of your business challenges.