How to Make a Gmail Bot with a persona using OpenAI GPT and MindsDB
My step by step process of how I went through MindsDB codebase and eventually implemented an application handler
Introduction
The AI hype refuses to die, more so after the release of Chat GPT and more recently the GPT4. I've been missing the AI action so far, so when MindsDB & Hashnode announced this hackathon and I saw the Twitter Bot implementation using the OpenAI APIs I knew that I wanted to build a bot for the hackathon.
But then the question arose, a bot for what purpose and for which application? Twitter was already done and dusted. I wanted my bot to be a wise and witty companion who is always available, so that helped me zero down on a Gmail bot. But the Gmail Integration was not yet implemented, and that was the second reason for choosing to build it. Practicing my Python skills, and the idea of contributing to a good project were the other important reasons.
Tl;dr this article is about creating a new application (Gmail) handler for MindsDB and then using that handler to create an email bot that replies to incoming emails interestingly and poetically.
Setting up the environment
Since the first task is to develop the Gmail handler, we must set up the environment for development. But before that, I needed to know how to contribute to this project, and how to install MindsDb for development. During the installation I faced only one issue related to libmagic which was not installed on my Mac by default, so had to install it using brew install libmagic.
The next step was to learn the basics of creating an app handler. This gave me a good overview of what I'm supposed to do for creating the Gmail handler.
Going through the relevant docs and following the mentioned steps is crucial if you want to contribute to any existing project.
Running the existing installation
Now it was time to get my hands dirty, and the first step in that direction was to use an existing app handler. But before that, I followed the "Predict Home Rental Prices" tutorial from the "Learning Hub" in the local MindsDB web console, and everything worked fine.
Next was the turn of the Twitter handler. I tried creating a tweets database using the below command in the local MindsDB browser console.
CREATE DATABASE my_twitter
WITH
ENGINE = 'twitter',
PARAMETERS = {
"bearer_token": "twitter bearer token",
"consumer_key": "twitter consumer key",
"consumer_secret": "twitter consumer key secret",
"access_token": "twitter access token",
"access_token_secret": "twitter access token secret"
};
At least it should error out saying invalid credentials, but instead, I got the below error
Can't connect to db: Handler 'twitter' can not be used
Well, that's a bummer. Why it is not working? This is where you start debugging and find out what is happening in the codebase. And how do we do that? I simply searched for the error string "Can't connect to db" in the codebase and found the issue to be related to the handler not getting imported. After some more investigation saw that the zsh console has this info message
Dependencies for the handler 'twitter' are not installed by default. If you want to use "twitter" please install "['tweepy']"
And there we have it. This gives us a clue that if we want to use any of the other handlers (except the basic ones) we need to install their dependencies manually.
pip install tweepy and restarting MindsDB was enough to get the error I was hoping for in the first place
Can't connect to db: Error connecting to Twitter api: 401 Unauthorized Unauthorized. Check bearer_token
Now we're all set for development. As the docs said to study the Twitter handler, I simply created a copy of the twitter_handler folder and renamed it to gmail_handler. Then replaced "Twitter" with "Gmail" in __init__.py and __about__.py files along with related method name changes in the gmail_handler file. Verified it by executing the same Twitter create database command but replacing the engine with "gmail", and it seemed to call our gmail_handler.
Implementing the Gmail Handler
Going by the steps mentioned in how to create an application handler we need to modify the below methods. Before we can read/write emails we need to authenticate the user, so the first targets were the connect and the check_connection methods.
Setting up a Google project for Gmail APIs
To use the Gmail APIs we need to set up a Google Cloud Project and a Google Account with Gmail enabled. We will also need to enable the Gmail API from the Google Cloud Console.
Then we need to create OAuth Client Ids for authenticating users, and possibly an Auth Consent Screen (if this is the first time we're setting up OAuth)
Setting up OAuth Client Id will give us a credentials file which we will need in our gmail_handler for connection. You can find more information on how to set up a Google project for the Gmail APIs here.
Initing the GmailHandler class
We take the connection arguments (which are passed with the CREATE DATABASE command) and store them for future use. We also register an "emails" table where we will store our data.
class GmailHandler(APIHandler):
"""A class for handling connections and interactions with the Gmail API.
Attributes:
credentials_file (str): The path to the Google Auth Credentials file for authentication
and interacting with the Gmail API on behalf of the uesr.
scopes (List[str], Optional): The scopes to use when authenticating with the Gmail API.
"""
def __init__(self, name=None, **kwargs):
super().__init__(name)
self.connection_args = kwargs.get('connection_data', {})
self.credentials_file = self.connection_args['credentials_file']
self.scopes = self.connection_args.get('scopes', DEFAULT_SCOPES)
self.token_file = None
self.max_page_size = 500
self.max_batch_size = 100
self.service = None
self.is_connected = False
emails = EmailsTable(self)
self._register_table('emails', emails)
Handling Google Authentication
Following the link in the previous section, and the MindsDB code requirements, we need to do the following
Replace the content of requirements.txt inside the gmail_handler folder with the following. Do remember to install these modules using the pip install command
google-api-python-client google-auth-httplib2 google-auth-oauthlibThe
connectmethod. Here we use the credentials files created in the previous section for authenticating the user.def connect(self): """Authenticate with the Gmail API using the credentials file. Returns ------- service: object The authenticated Gmail API service object. """ if self.is_connected is True: return self.service self.service = self.create_connection() self.is_connected = True return self.service def create_connection(self): creds = None token_file = os.path.join(os.path.dirname(self.credentials_file), 'token.json') if os.path.isfile(token_file): creds = Credentials.from_authorized_user_file(token_file, self.scopes) if not creds or not creds.valid: if creds and creds.expired and creds.refresh_token: creds.refresh(Request()) elif not os.path.isfile(self.credentials_file): raise Exception('Credentials must be a file path') else: flow = InstalledAppFlow.from_client_secrets_file(self.credentials_file, self.scopes) creds = flow.run_local_server(port=0, timeout_seconds=120) # Save the credentials for the next run with open(token_file, 'w') as token: token.write(creds.to_json()) return build('gmail', 'v1', credentials=creds)The
check_connectionmethoddef check_connection(self) -> StatusResponse: """Check connection to the handler. Returns ------- StatusResponse Status confirmation """ response = StatusResponse(False) try: # Call the Gmail API service = self.connect() result = service.users().getProfile(userId='me').execute() if result and result.get('emailAddress', None) is not None: response.success = True except HttpError as error: response.error_message = f'Error connecting to Gmail api: {error}.' log.logger.error(response.error_message) if response.success is False and self.is_connected is True: self.is_connected = False return response
Now we're ready to run the create database command and authenticate the user (of course we've not decided on the columns of our table but we will come to that). Simply run
CREATE DATABASE mindsdb_gmail
WITH ENGINE = 'gmail',
PARAMETERS = {
"credentials_file": "mindsdb/integrations/handlers/gmail_handler/credentials.json"
};
Fetching Emails from the Gmail API
The flow for fetching emails using MindsDB is like this: you execute an SQL SELECT query, and the select method of the APITable class gets called. There you parse the query params and finally call the Gmail API accordingly.
- The
selectmethod of the EmailsTable
class EmailsTable(APITable):
"""Implementation for the emails table for Gmail"""
def select(self, query: ast.Select) -> Response:
"""Pulls emails from Gmail "users.messages.list" API
Parameters
----------
query : ast.Select
Given SQL SELECT query
Returns
-------
pd.DataFrame
Email matching the query
Raises
------
NotImplementedError
If the query contains an unsupported operation or condition
"""
conditions = extract_comparison_conditions(query.where)
params = {}
for op, arg1, arg2 in conditions:
if op == 'or':
raise NotImplementedError(f'OR is not supported')
if arg1 in ['query', 'label_ids', 'include_spam_trash']:
if op == '=':
if arg1 == 'query':
params['q'] = arg2
elif arg1 == 'label_ids':
params['labelIds'] = arg2.split(',')
else:
params['includeSpamTrash'] = arg2
else:
raise NotImplementedError(f'Unknown op: {op}')
else:
raise NotImplementedError(f'Unknown clause: {arg1}')
if query.limit is not None:
params['maxResults'] = query.limit.value
result = self.handler.call_gmail_api(
method_name='list_messages',
params=params
)
# filter targets
columns = []
for target in query.targets:
if isinstance(target, ast.Star):
columns = []
break
elif isinstance(target, ast.Identifier):
columns.append(target.parts[-1])
else:
raise NotImplementedError(f"Unknown query target {type(target)}")
if len(columns) == 0:
columns = self.get_columns()
# columns to lower case
columns = [name.lower() for name in columns]
if len(result) == 0:
result = pd.DataFrame([], columns=columns)
else:
# add absent columns
for col in set(columns) & set(result.columns) ^ set(columns):
result[col] = None
# filter by columns
result = result[columns]
return result
- The
get_columnsmethod. These are the columns that our"EmailsTable"will have.
def get_columns(self):
"""Gets all columns to be returned in pandas DataFrame responses
Returns
-------
List[str]
List of columns
"""
return [
'id',
'message_id',
'thread_id',
'label_ids',
'from',
'to',
'date',
'subject',
'snippet',
'body',
]
- The
call_gmail_apimethod of theGmailHandlerclass. The way the Gmail messages API works is, first it returns a list of the messages that match your query criteria. These messages contain just the threadId & the messageId etc and not the full email. Then using the"messageIds"you fetch the full messages separately.
def call_gmail_api(self, method_name: str = None, params: dict = None):
"""Call Gmail API and map the data to pandas DataFrame
Args:
method_name (str): method name
params (dict): query parameters
Returns:
DataFrame
"""
service = self.connect()
if method_name == 'list_messages':
method = service.users().messages().list
elif method_name == 'send_message':
method = service.users().messages().send
else:
raise NotImplementedError(f'Unknown method_name: {method_name}')
left = None
count_results = None
if 'maxResults' in params:
count_results = params['maxResults']
params['userId'] = 'me'
data = []
limit_exec_time = time.time() + 60
while True:
if time.time() > limit_exec_time:
raise RuntimeError('Handler request timeout error')
if count_results is not None:
left = count_results - len(data)
if left == 0:
break
elif left < 0:
# got more results that we need
data = data[:left]
break
if left > self.max_page_size:
params['maxResults'] = self.max_page_size
else:
params['maxResults'] = left
log.logger.debug(f'Calling Gmail API: {method_name} with params ({params})')
resp = method(**params).execute()
if 'messages' in resp:
self._handle_list_messages_response(data, resp['messages'])
elif isinstance(resp, dict):
data.append(resp)
if count_results is not None and 'nextPageToken' in resp:
params['pageToken'] = resp['nextPageToken']
else:
break
df = pd.DataFrame(data)
return df
- Inner method
_handle_list_messages_responseand other related methods
# Handle the API response by downloading the full messages
# using a Batch Request.
def _handle_list_messages_response(self, data, messages):
total_pages = len(messages) // self.max_batch_size
for page in range(total_pages):
self._get_messages(data, messages[page * self.max_batch_size:(page + 1) * self.max_batch_size])
# Get the remaining messsages, if any
if len(messages) % self.max_batch_size > 0:
self._get_messages(data, messages[total_pages * self.max_batch_size:])
def _get_messages(self, data, messages):
batch_req = self.service.new_batch_http_request(lambda id, response, exception: self._parse_message(data, response, exception))
for message in messages:
batch_req.add(self.service.users().messages().get(userId='me', id=message['id']))
batch_req.execute()
# This method shows how to parse the full email returned
# by the Gmail API
def _parse_message(self, data, message, exception):
if exception:
log.logger.error(f'Exception in getting full email: {exception}')
return
payload = message['payload']
headers = payload.get("headers")
parts = payload.get("parts")
row = {
'id': message['id'],
'thread_id': message['threadId'],
'label_ids': message.get('labelIds', []),
'snippet': message.get('snippet', ''),
}
if headers:
for header in headers:
key = header['name'].lower()
value = header['value']
if key in ['from', 'to', 'subject', 'date']:
row[key] = value
elif key == 'message-id':
row['message_id'] = value
row['body'] = self._parse_parts(parts)
data.append(row)
def _parse_parts(self, parts):
if not parts:
return
body = ''
for part in parts:
if part['mimeType'] == 'text/plain':
part_body = part.get('body', {}).get('data', '')
body += urlsafe_b64decode(part_body).decode('utf-8')
elif part['mimeType'] == 'multipart/alternative' or 'parts' in part:
# Recursively iterate over nested parts to find the plain text body
body += self._parse_parts(part['parts'])
else:
log.logger.debug(f"Unhandled mimeType: {part['mimeType']}")
return body
The above is sufficient to fetch and store the emails of the authenticated user in the database. We can run an SQSL SELECT query like below to fetch the emails. The query parameter supports all the filter options that are available in the Gmail API
SELECT *
FROM mindsdb_gmail.emails
WHERE query = 'from:[email protected] OR search_text OR from:[email protected]'
AND label_ids = "INBOX,UNREAD"
LIMIT 20;
Sending Emails using the Gmail API
For sending emails through the Gmail API and MindsDB we need to use the SQL INSERT query. This in turn calls the insert method of the EmailsTable class, we created earlier.
def insert(self, query: ast.Insert):
"""Sends emails using the Gmail "users.messages.send" API
Parameters
----------
query : ast.Insert
Given SQL INSERT query
Raises
------
ValueError
If the query contains an unsupported condition
"""
columns = [col.name for col in query.columns]
if self.handler.connection_args.get('credentials_file', None) is None:
raise ValueError(
"Need the Google Auth Credentials file in order to write an email"
)
supported_columns = {"message_id", "thread_id", "to_email", "subject", "body"}
if not set(columns).issubset(supported_columns):
unsupported_columns = set(columns).difference(supported_columns)
raise ValueError(
"Unsupported columns for create email: "
+ ", ".join(unsupported_columns)
)
for row in query.values:
params = dict(zip(columns, row))
if not 'to_email' in params:
raise ValueError('"to_email" parameter is required to send an email')
message = EmailMessage()
message['To'] = params['to_email']
message['Subject'] = params['subject'] if 'subject' in params else ''
content = params['body'] if 'body' in params else ''
message.set_content(content)
# If threadId is present then add References and In-Reply-To headers
# so that proper threading can happen
if 'thread_id' in params and 'message_id' in params:
message['In-Reply-To'] = params['message_id']
message['References'] = params['message_id']
encoded_message = urlsafe_b64encode(message.as_bytes()).decode()
message = {
'raw': encoded_message
}
if 'thread_id' in params:
message['threadId'] = params['thread_id']
self.handler.call_gmail_api('send_message', {'body': message})
This method calls the same call_gmail_api we saw earlier to send an email. We can use an SQL INSERT query to send an email. The thread_id and message_id parameter values are only required if we're replying to an incoming email and want that our reply should form a thread with the original email. (The "subject" should exactly match the original subject line for it to work)
INSERT INTO mindsdb_gmail.emails (thread_id, message_id, to_email, subject, body)
VALUES ('187cbdd861350934d', '8e54ccfd-abd0-756b-a12e-f7bc95ebc75b@Spark', '[email protected]', 'Trying out MindsDB',
'This seems awesome. You must try it out whenever you can.')
Creating the Gmail Bot
Now that we're unblocked and can fetch/send emails easily using our shiny new GmailHandler, we're ready to work on our Gmail bot.
Obtaining an OpenAI API key
Since we're developing this locally we do not have the luxury of using the inbuilt API key provided by MindsDB Cloud. We need to create an account on OpenAI and create an API key.
Training the Model using MindsDB
I do not have access to GPT4 APIs, so I'm using the gpt-3.5-turbo model itself. We're just telling GPT to respond to the email with a proper salutation and signature and to keep the email tone casual. This is done using the prompt_template parameter.
CREATE MODEL mindsdb.gpt_model
PREDICT response
USING
engine = 'openai',
max_tokens = 500,
api_key = '<your_api_key>',
model_name = 'gpt-3.5-turbo',
prompt_template = 'From input message: {{input_text}}\
by from_user: {{from_email}}\
In less than 500 characters, write an email response to {{from_email}} in the following format:\
Start with proper salutation and respond with a short message in a casual tone, and sign the email with my name mindsdb';
Once the training is complete we're ready to see our bot in action. Run the following command
SELECT response
FROM mindsdb.gpt_model_email
WHERE from_email = "[email protected]"
AND input_text = "Hi there, I'm bored. Give me a puzzle to solve";
And we get the following response, seems quite all right, isn't it?
On asking for a new puzzle it says the following
Giving the bot a persona
Since our initial experiments seem to work fine, we can get a bit adventurous now. Let's give our bot a combined persona of Master Yoda from the Star Wars movies, and Edgar Allan Poe, the famous poet. We do this by changing the prompt_template in our earlier command
CREATE MODEL mindsdb.gpt_model_yodapoe
PREDICT response
USING
engine = 'openai',
max_tokens = 800,
api_key = '<your_api_key>',
model_name = 'gpt-3.5-turbo', -- you can also use 'text-davinci-003' or 'gpt-3.5-turbo'
prompt_template = 'From input message: {{input_text}}\
by from_user: {{from_email}}\
In less than 500 characters, write an email response to {{from_email}} in the following format:\
<respond with a 4 line poem as if you were Edgar Allan Poe but you are also a wise elder like Master Yoda from the Star Wars movies. The wordings should be like Master Yoda but the format should be like Poe. Do not mention that you are Master Yoda or Edgar Allan Poe. Sign it with a made up quote similar to what Voltaire, Nietzsche etc would say. Do not explain or say anything else about the quote.>';
Train the model, and then run the same SELECT queries by changing the model name
SELECT response
FROM mindsdb.gpt_model_yodapoe
WHERE from_email = "[email protected]"
AND input_text = "Hi there, I'm bored. Give me a puzzle to solve";
This is what I get
On changing the input text to the following
SELECT response
FROM mindsdb.gpt_model_yodapoe
WHERE from_email = "[email protected]"
AND input_text = "Hi there, What's in a hackathon?";
we get the following result
Overall this seems to be working fine. We can always fine-tune the persona based on our taste. Now we can connect this response generation with the actual email sending and we may as well create a scheduled job to read emails at regular intervals and reply to them based on some predefined criteria.
Outcomes
When I started working on this feature, I had the following goals and their respective outcomes at the end
Create the Gmail handler: This in my opinion is done. There may be some bugs that I'll need to solve in due course
Contribute to the MindsDB project: I've already opened a PR for my changes and now I'm hoping that it gets merged into the codebase.
Create a Gmail Bot: We've all the ingredients in place, we just need to deploy it somewhere so that it is always available. I did try deploying the source on a droplet but I'm stuck with an error for which I've raised a GitHub issue.
Practice my Python skills: I think I've made good progress on this while working on the feature. Python is not my area of expertise, so I'm quite pumped that I was able to create a working integration in a short span of 7-8 days.
Conclusion
Overall it was quite fun and interesting to create a Gmail Bot and the needed Gmail Handler for MindsDB. During the process, I got to see the inner workings of the MindsDB codebase. I learnt to use the Gmail APIs and how emails are structured behind the scenes and how to parse it. It also allowed me to use the MindsDB integration with OpenAI, and now I too can say AI is eating the world :-).
Hope you liked reading the article. If you've noticed any error or issue anywhere please do let me know in the comments.
-- Keep adding the bits, soon you'll have more bytes than you may need.