Transforming Data Pipelines with OpenAI’s Function Calling Feature: Implementing an Email Sending Workflow Using PostgreSQL and FastAPI

The exciting world of AI has taken one other breakthrough by the introduction of function calling capabilities in OpenAI’s latest Large Language Models (LLMs). This recent feature enhances the interaction between humans and AI, transforming it from an easy question-and-answer format to a more dynamic and lively dialogue.

But what exactly are these function calling capabilities? At their core, they permit the LLM to call predefined functions during a conversation based on the input instructions. This might be anything from sending an email to fetching data from a database based on the context of the conversation.
The advantages and applications of using function calling capabilities are vast. It significantly increases the dynamic utility of AI in various applications, from customer support to how we construct data pipelines.

Imagine a customer support AI that may answer questions and perform actions equivalent to booking appointments, sending information to an email address, or updating customer details in real time. Or consider a knowledge pipeline where the LLM agent can fetch, update, and manipulate data on command.

Within the upcoming sections, we’ll explore these applications further and supply a step-by-step guide on leveraging this recent capability by constructing an email-sending pipeline.

As at all times, the code is out there on my Github.

We wish to know what OpenAI’s newly introduced function calling feature brings to the AI race. For that, let’s understand what differentiates it from other tools and frameworks out there, like LangChain. We already introduced LangChain in the primary article of this series. It’s a preferred framework for developing AI-powered applications. The function calling feature and LangChain bring unique benefits and capabilities to the table and are built around making AI more usable, versatile, and dynamic.

We already know that the function calling feature adds an additional layer of interactivity to AI applications. It enables the model to call predefined functions inside a conversation, enhancing the LLM’s dynamism and reactivity. This recent feature can potentially simplify the technique of adding functionality to AI applications. Developers would wish to define these functions, and the model could then execute them as a part of the conversation, depending on the context. The first advantage here is its direct integration with the OpenAI models, which facilitates ease of use, quick setup, and a lower learning curve for developers conversant in the OpenAI ecosystem.

Then again, LangChain offers a comprehensive and versatile framework designed for developing more complex, data-aware, and agentic AI applications. It allows a language model to interact with its environment and make decisions based on high-level directives. Its modules provide abstractions and standard interfaces for constructing applications, including models, prompts, memory, indexes, chains, agents, and callbacks.

LangChain’s approach facilitates constructing applications where an LLM can persist its state across different interactions, sequence and chain calls to different utilities, and even interact with external data sources. We will see these as function calling capabilities on steroids. Thus, it is especially useful for developers constructing complex, multi-step applications that leverage language models. The downside of the added complexity is that it creates a steeper learning curve to make use of it fully.

In my opinion, data pipelines are amongst probably the most exciting application areas for the brand new function calling capabilities in LLMs. Data pipelines are a critical component of any data-driven organization that collects, processes, and distributes data. Typically, these processes are static, predefined, and require manual intervention for any changes or updates. That is where the dynamic behavior of an LLM that we discussed above creates a possibility.

Traditionally, database querying requires specific knowledge of query languages like SQL. With LLMs’ ability to call functions, services, and databases directly, users can retrieve data conversationally without the necessity for explicit query formulation. An LLM could translate a user’s request right into a database query, fetch the information, and return it in a user-friendly format, all in real time. This feature could democratize data access across different roles inside a company.

One other aspect that would change is data transformation. It often requires separate data cleansing and processing steps before evaluation. An LLM could streamline this process by performing data cleansing and manipulation tasks interactively based on the user’s instructions. Furthermore, manipulating real-time data during a conversation allows for more exploratory and iterative data evaluation.

A 3rd use case is data monitoring. It involves regular checks to make sure the accuracy and consistency of knowledge in a knowledge pipeline. With LLMs, monitoring tasks can turn out to be more interactive and efficient. As an illustration, an LLM can alert users to data inconsistencies during conversations and take corrective actions immediately.

Finally, the LLMs may automate the creation and distribution of knowledge reports. Users can instruct the LLM to generate a report based on specific criteria, and the LLM can fetch the information, create the report, and even send it to the relevant recipients.

We aim to create a pipeline that sends an email to a user. While this may increasingly sound straightforward, the great thing about this process lies within the interplay of different components controlled by the LLM. The AI model doesn’t merely generate an email body; it dynamically interacts with a database to retrieve user information, composes a contextually appropriate email, after which instructs a service to send it.

Our pipeline consists of three most important components: PostgreSQL, FastAPI, and the OpenAI LLM. We use PostgreSQL to store our user data. This data includes user names and their associated email addresses. It serves as our source of truth for user information. FastAPI is a contemporary, high-performance web framework for constructing APIs with Python. We use a FastAPI service to simulate the technique of sending an email. When the service receives a request to send an email, it returns a response confirming the e-mail has been sent. The LLM serves because the orchestrator of your entire process. It controls the conversation, determines the crucial actions based on the context, interacts with the PostgreSQL database to fetch user information, crafts an email message, and instructs the FastAPI service to send the e-mail.

Implementing PostgreSQL Database

The primary component of our pipeline is the PostgreSQL database where we store our user data. Establishing a PostgreSQL instance is made easy and reproducible with Docker, a platform that enables us to containerize and isolate our database environment.

You first have to install Docker to establish a PostgreSQL Docker container. Once installed, you’ll be able to pull the PostgreSQL image and run it as a container. We map the container’s port 5432 to the host’s port 5432 to access the database. In a production environment, please set your password as an environment variable and don’t set them directly in a command as below. We’re doing this technique to speed up our process.

docker run --name user_db -e POSTGRES_PASSWORD=testpass -p 5432:5432 -d postgres

With our PostgreSQL instance running, we will now create a database and a table to store our user data. We’ll use an initialization script to create a userstable with username and emailcolumns and populate it with some dummy data. This script is placed in a directory which is then mapped to the /docker-entrypoint-initdb.d directory within the container. PostgreSQL executes any scripts present in this directory upon startup. Here’s what the script (user_init.sql) looks like:

CREATE DATABASE user_database;
c user_database;
CREATE TABLE users (
username VARCHAR(50),
email VARCHAR(50)
);
INSERT INTO users (username, email) VALUES
('user1', 'user1@example.com'),
('user2', 'user2@example.com'),
('user3', 'user3@example.com'),
...
('user10', 'user10@example.com');

The LLM is able to understanding SQL commands and will be used to question the PostgreSQL database. When the LLM receives a request that involves fetching user data, it may possibly formulate a SQL query to fetch the crucial data from the database.

As an illustration, if you happen to ask the LLM to send an email to user10, the LLM can formulate the query :

SELECT  email FROM users WHERE username=’user10'; 

This permits it to fetch user10 email address from the users table. The LLM can then use this email address to instruct the FastAPI service to send the e-mail.

In the following section, we’ll guide you thru the implementation of the FastAPI service that sends the emails.

Creating the FastAPI Email Service

Our second component is a FastAPI service. This service will simulate the technique of sending emails. It’s a simple API that receives a POST request containing the recipient’s name, email, and the e-mail body. It would return a response confirming the e-mail was sent. We are going to again use Docker to make sure our service is isolated and reproducible.

First, it is advisable install Docker (if not already installed). Then, create a recent directory on your FastAPI service and move into it. Here, create a recent Python file (e.g., most important.py) and add the next code:

from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
class User(BaseModel):
name: str
email: str
body: str
@app.post("/send_email")
async def send_email(user: User):
return {
"message": f"Email successfully sent to {user.name} with email {user.email}. Email body:nn{user.body}"
}

This code defines a FastAPI application with a single endpoint /send_email/. This endpoint accepts POST requests and expects a JSON body containing the recipient’s name, email, and the e-mail body.

Next, create a Dockerfile in the identical directory with the next content:

FROM python:3.9-slim-buster
WORKDIR /app
ADD . /app
RUN pip install --no-cache-dir fastapi uvicorn
EXPOSE 1000
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "1000"]

This Dockerfile instructs Docker to create a picture based on thepython:3.9-slim-buster image, a light-weight image ideal for efficiently running python applications. It then copies our most important.py file into the /app/ directory within the image.

You possibly can construct the Docker image using the command:

docker construct -t fastapi_email_service .

After which run it with:

docker run -d -p 1000:1000 fastapi_email_service

The LLM interacts with the FastAPI service using a POST request. When the LLM decides to send an email, it generates a function call to the send_email function. The arguments of this function call contain the name, email, and the e-mail body.

The function call is processed by our Python script, which extracts the function arguments and uses them to send a POST request to our FastAPI service. The FastAPI service responds with a message indicating the e-mail was successfully sent.

Now, now we have all components of our pipeline. The subsequent section will tie the whole lot together, explaining how the LLM orchestrates the interaction between the PostgreSQL database and the FastAPI service to send an email.

Integrating with OpenAI LLM

The last piece of our pipeline is the OpenAI LLM integration. The LLM serves because the orchestrator, interpreting our commands, querying the database for user information, and instructing the FastAPI service to send emails.

Our script uses OpenAI’s API to make chat-based completions with the LLM. Each completion request consists of a series of messages and optionally a listing of function specifications that the model could call. We start the conversation with a user message, which provides a prompt to the assistant.

Here’s the chat_completion_request function we use to send a request to the API:

@retry(wait=wait_random_exponential(min=1, max=40), stop=stop_after_attempt(3))
def chat_completion_request(messages, functions=None, model=GPT_MODEL):
headers = {
"Content-Type": "application/json",
"Authorization": "Bearer " + openai.api_key,
}
json_data = {"model": model, "messages": messages}
if functions isn't None:
json_data.update({"functions": functions})
response = requests.post(
"https://api.openai.com/v1/chat/completions",
headers=headers,
json=json_data,
)
return response

We use the Chat class to administer the conversation history. It has methods so as to add a recent message to the history and display your entire conversation:

class Chat:
def __init__(self):
self.conversation_history = []
def add_prompt(self, role, content):
message = {"role": role, "content": content}
self.conversation_history.append(message)
def display_conversation(self):
for message in self.conversation_history:
print(f"{message['role']}: {message['content']}")

In our use case, the LLM must interact with our PostgreSQL database and FastAPI service. We define these functions and include them in our completion request. Here’s how we define our sql_query_email and send_emailfunctions:

functions = [
{
"name": "send_email",
"description": "Send a new email",
"parameters": {
"type": "object",
"properties": {
"to": {
"type": "string",
"description": "The destination email.",
},
"name": {
"type": "string",
"description": "The name of the person that will receive the email.",
},
"body": {
"type": "string",
"description": "The body of the email.",
},
},
"required": ["to", "name", "body"],
},
},
{
"name": "sql_query_email",
"description": "SQL query to get user emails",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "The query to get users emails.",
},
},
"required": ["query"],
},
},
]

After we make a completion request, the LLM responds with its intended actions. If the response features a function call, our script executes that function. For instance, if the LLM decides to call the sql_query_emailfunction, our script retrieves the user’s email from the database after which adds the result to the conversation history. When the send_emailfunction is named, our script sends an email using the FastAPI service.

The most important loop of our script checks for function calls within the LLM’s response and acts accordingly:

chat = Chat()
chat.add_prompt("user", "Send an email to user10 saying that he must pay the monthly subscription fee.")
result_query = ''
for i in range(2):
chat_response = chat_completion_request(
chat.conversation_history,
functions=functions
)
response_content = chat_response.json()['choices'][0]['message']
if 'function_call' in response_content:
if response_content['function_call']['name'] == 'send_email':
res = json.loads(response_content['function_call']['arguments'])
send_email(res['name'], res['to'], res['body'])
break
elif response_content['function_call']['name'] == 'sql_query_email':
result_query = query_db(json.loads(response_content['function_call']['arguments'])['query'])
chat.add_prompt('user', str(result_query))
else:
chat.add_prompt('assistant', response_content['content'])

After we run the script, we get the next output:

{
"message": "Email successfully sent to User 10 with email user10@example.com.",
"Email body": "nnDear User 10, nnThis is a reminder that your monthly subscription fee is due. Please make the payment as soon as possible to make sure uninterrupted service. Thanks on your cooperation. nnBest regards, nYour Subscription Service Team"
}

Let’s break down what happened for us to get this output. Our prompt was “Send an email to user10 saying that he must pay the monthly subscription fee.”. Note that there isn’t any information concerning the user10email in our message. The LLM identified the missing information and understood that our function query_emailwould allow it to question a database to get the e-mail from that user. After getting the e-mail, it got two things right once more: first, it should generate the body of the e-mail, and second, it should call the send_emailfunction to trigger the e-mail using the FastAPI email service.

This text explored the function calling feature by implementing a case study where the LLM coordinates a pipeline involving a PostgreSQL database and a FastAPI email service. The LLM successfully navigated the duty of retrieving a user’s email from the database and instructing the e-mail service to send a personalised message, all in response to a single prompt.

The implications of function calling in AI models might be enormous, opening up recent possibilities for automating and streamlining processes. Data pipelines could change from static and engineering heavy to dynamic entities, allowing non-technical users to quickly get their hands on the most recent data using natural language.

This text belongs to “Large Language Models Chronicles: Navigating the NLP Frontier”, a recent weekly series of articles that may explore methods to leverage the facility of enormous models for various NLP tasks. By diving into these cutting-edge technologies, we aim to empower developers, researchers, and enthusiasts to harness the potential of NLP and unlock recent possibilities.

Articles published to date:

1. Summarizing the most recent Spotify releases with ChatGPT
2. Master Semantic Search at Scale: Index Thousands and thousands of Documents with Lightning-Fast Inference Times using FAISS and Sentence Transformers
3. Unlock the Power of Audio Data: Advanced Transcription and Diarization with Whisper, WhisperX, and PyAnnotate
4. Whisper JAX vs PyTorch: Uncovering the Truth about ASR Performance on GPUs
5. Vosk for Efficient Enterprise-Grade Speech Recognition: An Evaluation and Implementation Guide
6. Testing the Massively Multilingual Speech (MMS) Model that Supports 1162 Languages
7. Harnessing the Falcon 40B Model, the Most Powerful Open-Source LLM

Keep up a correspondence: LinkedIn