Function Calling in OpenAI Assistants API
This weekend, I decided it was time to make a simple function with assistant api that should have been pretty straightforward. This is not what happened.
I found a github repo called awesome-assistant-api and there was a notebook that described function calling, mostly. I am not a fan of notebooks in general but this was about the only thing I could find with enough information to understand what to do myself.
My application is in the base stages and to extend it, I need the assistant to be capable of making decisions and acting on them. I chose to first try allowing the assistant to create google calendar entries. I have already done the hard part of setting up the dev account, O-Auth and handling the tokens needed. I also wrote out all of the basic methods of interfacing with docs, calendar, tasks and gmail. I also already have a decent setup for finishing out the day with a wrapup function that creates a summary of the days events, converts them from Json entries with the assistant having instructions to create headings and other formatting in markdown. Then I convert to HTML and finally run a command that uploads the HTML to docs and converts that into a google doc.
With that figured out, it must be easy to do a calendar with an assistant function, right? All said and done, I guess. With my function set up to create an entry in the calendar, I should only need to send the correct variable data to it from the assistant. This is true but you also need a JSON schema that you add to the function tool.
When looking at the code provided on the repo you need to build out this schema in the definition of your assistant. Like this:
def setup_assistant(client, script_code):
# create a new agent
assistant = client.beta.assistants.create(
name="Code Generator",
instructions=INSTRUCTIONS,
tools=[
{
"type": "code_interpreter"
},
{
"type": "function",
"function": {
"name": "execute_python_code",
"description": "Use this function to execute the generated code which requires internet access or external API access",
"parameters": {
"type": "object",
"properties": {
"code": {
"type": "string",
"description": "The python code generated by the code interpretor",
}
},
"required": ["code"],
},
},
},
{
"type": "function",
"function": {
"name": "generate_image",
"description": "generate image by Dall-e 3",
"parameters": {
"type": "object",
"properties": {
"prompt": {"type": "string", "description": "The prompt to generate image"},
"size": {"type": "string", "enum": ["c", "f"]}
},
"required": ["prompt"]
}
}
}
],
model="gpt-4-1106-preview",
)
However, I store my assistant setup in a JSON file for an interface. It allows you to choose things and keep track of threads daily. I start a new thread each day to keep the costs down. They compound on message length and quickly grow to a high cost. So, i set out to figure out how to pull in data from the function schema on the playground.
I have not found a method of pulling this data down. I can ask for the assistant setup and get the list of tools from the assistant and if a function is enabled, then function shows in that list. It does not pull the schema down though. You can use this to list the tools assigned to the current assistant ID.
current_tools = [getattr(tool, "type", None) for tool in assistant.tools]
print("Current tools:", current_tools)
Since I cannot get the function schema to pull down from the playground, i found that the best way to do it for me was to upload the function schema on startup. Since I cannot verify it exists, I will assume it does not and push it to the settup each time I start the program. It seems that as long as you keep the name the same, it replaces everything in the schema and acts as an update. So, when you run into an issue during testing you can change things in the json schema and then start the program to test.
The Code
I kept the playground open the entire time I was testing in the code to see how it reacted. So, here is what I figured out after 2 days of playing with this. Here is the google calendar function:
"""
This is a working example for calendar event. Assistant function schema matches this and
can create events in calndar when asked or decides to do so.
"""
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
from datetime import datetime, timedelta
from auth_all import get_auth_creds
def create_calendar_event(calendar_id, summary, start_time, end_time, description=None, reminders=None):
"""
Create a calendar event on a specific calendar.
:param calendar_id: ID of the calendar to create the event in.
:param summary: Summary or title of the event.
:param start_time: Start time of the event (datetime object).
:param end_time: End time of the event (datetime object).
:param description: Optional description of the event.
:param reminders: Optional reminder minutes before event.
:return: The created event.
"""
service = get_auth_creds("calendar")
event = {
'summary': summary,
'description': description,
'start': {
'dateTime': start_time.isoformat(),
'timeZone': 'America/New_York' # Replace with a valid time zone
},
'end': {
'dateTime': end_time.isoformat(),
'timeZone': 'America/New_York' # Replace with a valid time zone
},
'reminders': {
'useDefault': False,
'overrides': [
{'method': 'email', 'minutes': reminders},
{'method': 'popup', 'minutes': reminders},
],
},
}
try:
return service.events().insert(calendarId=calendar_id, body=event).execute()
except HttpError as err:
print(f'An error occurred: {err}')
# Example usage
if __name__ == "__main__":
calendar_id = '' # Replace with your specific calendar ID
start_time = datetime.now() + timedelta(days=1) # Set to one day from now
end_time = start_time + timedelta(hours=1) # One hour long event
new_event = create_calendar_event(calendar_id, 'New Event', start_time, end_time, description='Meeting with team.', reminders=30)
print('Created event:', new_event['htmlLink'])
And here is the schema I used for the function call:
"tools": [
{
"type": "code_interpreter"
},
{
"type": "retrieval"
},
{
"type": "function",
"function": {
"name": "manage_calendar",
"description": "Create an event in Google Calendar",
"parameters": {
"type": "object",
"properties": {
"action": {
"type": "string",
"enum": [
"create"
],
"description": "Action to perform on the calendar"
},
"event_details": {
"type": "object",
"properties": {
"calendar_id": {
"type": "string",
"description": "ID of the calendar to create the event in"
},
"summary": {
"type": "string",
"description": "Summary or title of the event"
},
"start_time": {
"type": "string",
"description": "Start time of the event in ISO 8601 format"
},
"end_time": {
"type": "string",
"description": "End time of the event in ISO 8601 format"
},
"description": {
"type": "string",
"description": "Optional description of the event"
},
"reminders": {
"type": "integer",
"description": "Optional reminder minutes before the event"
}
},
"required": [
"calendar_id",
"summary",
"start_time",
"end_time",
"reminders"
]
}
},
"required": [
"action",
"event_details"
]
}
}
}
]
Function Name and Description:
JSON: “name”: “manage_calendar”, “description”: “Create an event in Google Calendar”.
Python: The function create_calendar_event in Python does exactly what the JSON description states: it creates an event in Google Calendar.
Parameters Mapping:
JSON: Under “parameters”, there are two main properties - “action” and “event_details”.
Python: Each property in “event_details” corresponds to a parameter in the create_calendar_event function.
Parameter Details:
“action”: Although it’s a required field in JSON, the Python function is specifically designed for the “create” action, hence it’s implicit in its purpose.
“event_details”: This is where the main mapping occurs. Each sub-property here maps to a function argument.
“calendar_id” → calendar_id: The ID of the calendar.
“summary” → summary: The title or summary of the event.
“start_time” → start_time: The start time of the event, noted in the Python function as a datetime object.
“end_time” → end_time: The end time of the event, also as a datetime object in Python.
“description” → description: An optional description of the event.
“reminders” → reminders: Optional reminder minutes before the event.
Required Fields:
JSON: “required” lists the fields that must be provided for the function to work.
Python: The function parameters without default values (calendar_id, summary, start_time, end_time) are required, corresponding to the JSON schema.
Data Types and Formats:
The JSON schema specifies the type of each parameter (e.g., “type”: “string” for calendar_id).
In the Python function, these types are implicitly understood (e.g., calendar_id is expected to be a string).
Functionality:
The JSON schema defines what the function does and what it needs, acting like a blueprint.
The Python function is the actual implementation, using the parameters defined in the schema to perform the action.
Here is the function I use on every message. It pays attention to the status of the call and if the assistant determines that a function should be called, the requires_action status flags and then the function checks what function should be called and acts on it based off of the schema and then the assistant will generate the output needed to fulfill the schema required and if it chooses, the optional variable values.The assistant waits at certain points in the run with the status of requires_action and is looking for a response from the function or it could be presented to an end user for input I suppose. However, you cannot see this with the code I have here. If you were to watch what happens in the playground when waiting for the response, it prompts you to enter something to simulate the response. For testing, this was crucial. Otherwise, the run would sit for some amount of time waiting and you could not do anything until it expired. I could not figure out the exact time it took but it was a several minutes waiting. I will eventually write another portion of the function to look for this and set a timer that will cancel the run if certain events occur that would mean it needed to fail gracefully.
I have a lot of print statements for testing purposes because I had issues passing the iso formatted date time to the calendar function.
def wait_on_run(run, thread):
tool_outputs = [] # Initialize tool_outputs list
response_messages = [] # Initialize response_messages list
while run.status == "queued" or run.status == "in_progress":
time.sleep(sleepTime)
run = client.beta.threads.runs.retrieve(
thread_id=thread.id,
run_id=run.id)
if run.status == "completed":
get_response(thread)
if run.status == "requires_action":
tool_call = run.required_action.submit_tool_outputs.tool_calls[0]
tool_outputs = []
# Modify setup_assistant to include manage_calendar function tool in tools
# Modify run_assistant to handle manage_calendar tool call
if tool_call.function.name == "manage_calendar":
# Extract calendar data from tool call
calendar_data = json.loads(tool_call.function.arguments)
print("Calendar Data:", calendar_data) # Add this line to check the content
# Check if 'event_details' is present and a dictionary
event_details = calendar_data.get('event_details')
if event_details is not None and isinstance(event_details, dict):
# Extract individual fields
summary = event_details.get("summary")
start_time_str = event_details.get("start_time")
end_time_str = event_details.get("end_time")
description = event_details.get("description")
reminders = event_details.get("reminders")
# Convert start_time and end_time strings to datetime objects
start_time = datetime.datetime.fromisoformat(start_time_str)
end_time = datetime.datetime.fromisoformat(end_time_str)
# Print individual fields for debugging
print("Summary:", summary)
print("Start Time:", start_time)
print("End Time:", end_time)
print("Description:", description)
print("Reminders:", reminders)
# Call your create_calendar_event function with extracted data
result = create_calendar_event(
selected_calendar_id,
summary,
start_time.isoformat(), # Convert to ISO 8601 format
end_time.isoformat(), # Convert to ISO 8601 format
description=description,
reminders=reminders,
)
# Check if the event creation was successful
if result== "success":
# Append the success result to tool_outputs
tool_outputs.append(
{
"tool_call_id": tool_call.id,
"output": result,
},
)
if result== "failure":
tool_outputs.append(
{
"tool_call_id": tool_call.id,
"output": result,
},
)
# Handle the error, print error message or take appropriate action
print(f"Event creation failed: {result['error_message']}")
if tool_outputs:
print("Submitting tool outputs...")
run = client.beta.threads.runs.submit_tool_outputs(
thread_id=thread.id,
run_id=run.id,
tool_outputs=tool_outputs
)
if run.status == "cancelled":
print("Run cancelled.")
break
if run.status == "cancelling":
print("Run cancelling.")
if run.status == "failed":
print("Run failed.")
break
if run.status == "expired":
print("Run expired.")
break
return run, response_messages # Return the run and response messages
Function Overview:
wait_on_run monitors the status of a ‘run’ (an instance of a process or task) and executes actions based on its status. The function is designed to wait for a ‘run’ to reach a certain status and then perform specific actions, including handling calendar event creation.
Monitoring the Run Status:
The function enters a while loop, continuously checking the status of the ‘run’ (“queued”, “in_progress”, etc.).
time.sleep(sleepTime) is used to pause the execution for a specified time to prevent continuous, rapid querying.
Responding to Run Status Changes:
If the status changes to “completed”, the function calls get_response(thread), which presumably handles the completion of the run.
If the status is “requires_action”, it indicates that user intervention or further processing is needed. Here, it checks for a specific tool call related to the calendar event creation (“manage_calendar”).
Handling the Calendar Event Creation:
The function extracts the calendar event details from tool_call.function.arguments, which is structured based on the JSON schema you provided earlier.
The extracted details include summary, start_time, end_time, description, and reminders.
These details are converted into the appropriate format, particularly converting start_time and end_time from ISO 8601 string format to datetime objects.
The create_calendar_event function is then called with these parameters to create the calendar event.
Processing the Result of Event Creation:
The function checks if the event creation was successful (result == “success”) or failed (result == “failure”).
Based on the result, it updates tool_outputs with the outcome of the calendar event creation attempt.
If there are any outputs to submit (i.e., tool_outputs is not empty), the function calls client.beta.threads.runs.submit_tool_outputs to submit these outputs.
Handling Other Run Statuses:
The function also handles other statuses like “cancelled”, “cancelling”, “failed”, and “expired”, mostly by printing messages and breaking out of the loop.
Return Value:
Finally, the function returns the updated run object and any response_messages collected during its execution.
In summary, the wait_on_run function is a monitoring and action-execution tool that waits for a specific task (‘run’) to require action, particularly for creating a calendar event. When the run requires action for a calendar event, it processes the necessary data based on the JSON schema, calls the create_calendar_event function to create the event, and handles the result of this operation. The function is a crucial part of a larger system where runs are managed and processed as per their status and requirements.
The End
This is the beginning of what is possible. You can asynchronously call runs and functions and then pay attention to them over time to submit responses. This way, it could launch a thousand things at one time if wanted. Multiple assistants with instructions for certain functions themselves. This could alow the main assistant to e a manager for the others and decide when they should go on and do their tasks. This should allow the conversation to be faster. Right now it is pretty slow. Some responses take upwards of 30 seconds and it feels longer when waiting. I am sure this will improve as assistants move out of beta though.
This can be an extremely powerful capability. Adding external api calls, code writing and execution within your actual environment. Infinite capabilities.
All said and done, the time I spent on this one call allowed me to understand the intricate testing needed to figure out function calling. Especially when using other api calls in them. Maybe I should have started simpler.
I can now talk to CaptAInsLog and when it decides that I need to schedule out something we talked about, without me asking, it will. I have a calendar in google calendar setup specifically for this. That way, it is contained. My assistant is currently set up with instructions that make it curious about your conversation and tries to dig deeper for more details to add to a journal. It decides what is important about the conversation that may need actions. If actions are needed, it comes up witha timeline. It then schedules this timeline for you. That way, you get notifications about this outside fo the system and tied to your day to day interactions with your phone.
I will be adding task scheduler, email and query capabilities. This will allow my assistant to get relevant information about upcoming events and things it scheduled for me. Then it can bring them up in conversation throughout the day as the scheduled tasks come closer in time. I am already having unreal conversations with this and it really ahs helped me gather my thoughts and plan. Now with the ability to hold me to a timeline, i get prodded to keep up. It is a powerful tool for organization and record keeping.
Leave a comment