contact_utils.py

#!/usr/bin/env python3

# Author: Mike Tremaine 
# Date: 2024-07-19 
# Version: 1.0
# License: MIT

# Script: contact_utils.py
# Description: This script contains utility functions for the contact_helpdesk and related rag lookup scripts.

import os, json, markdown
import mysql.connector
from sendgrid import SendGridAPIClient
from sendgrid.helpers.mail import Mail
from baserun import ApiClient, log, feedback, OpenAI
from baserun.wrappers.generic import (
    GenericChoice,
    GenericClient,
    GenericCompletion,
    GenericCompletionMessage,
    GenericInputMessage,
)

#Import json from config_path
def load_json_config(config_path):
    with open(config_path, 'r') as file:
        return json.load(file)

# Function to connect to MySQL database
def connect_to_mysql(config):
    # see config.json for values
    my_conn = mysql.connector.connect(
        host=config["servername"],
        user=config["username"],
        password=config["password"],
        database=config["dbname"],
    )
    return my_conn

#We export the keys to the environment used for none generic wrappers
def initialize_global_baserun(config):
    if "BASERUN_API_KEY" not in os.environ:
        os.environ["BASERUN_API_KEY"] = config["baserun_key"]
    if "OPENAI_API_KEY" not in os.environ:
        os.environ["OPENAI_API_KEY"] = config["openai_key"]

#This function initializes the baserun client directly
def initialize_baserun(config):
    # Initialize the baserun client, passing in the User Contact ID
    baserun_id = config["baserun_id"]
    baserun_key = config["baserun_key"]
    baserun_client = GenericClient(
        name="Vision Benefits",
        user_id=baserun_id,
        # If you have the environment variable BASERUN_API_KEY set you can omit this next line
        api_client=ApiClient(api_key=baserun_key),
    )
    return baserun_client

def send_baserun_openai_query(config, prompt):
    """
    Sends a query to OpenAI's chat completions API using the provided configuration and prompt.

    Args:
        config (dict): The configuration settings for the query.
        prompt (str): The prompt for the query.

    Returns:
        tuple: A tuple containing the response message content and the trace ID.
    """
    #Check if the API key is set in the environment
    if "OPENAI_API_KEY" not in os.environ:
        initialize_global_baserun(config)

    client = OpenAI()
    completion = client.chat.completions.create(
        name="Member Benefits",
        model="gpt-4o",
        temperature=0.7,
        messages=[
            {
                "role": "user",
                "content": prompt, 
            }
        ],
    )
    print(completion) # Print the completion object to the console
    trace_id = completion.trace_id
    return completion.choices[0].message.content, trace_id

def send_baserun_tag(config, evaluation, payload):
    """
    Sends a baserun tag to OpenAI.

    Args:
        config (dict): Configuration settings.
        evaluation (bool): Indicates if the evaluation is being performed.
        payload (dict): Payload data.

    Returns:
        None
    """
    #Check if the API key is set in the environment
    if "OPENAI_API_KEY" not in os.environ:
        initialize_global_baserun(config)

    client = OpenAI()
    if evaluation:
        trace_id = payload['trace_id']
        log("Tagging resumed", trace_id=trace_id, )
        feedback("User satisfaction", str(evaluation), trace_id=trace_id)

def send_generic_baserun_message(baserun_client, model_name, prompt, response, evaluation=None):
    """
    Sends a generic baserun message using the provided baserun client.

    Parameters:
    - baserun_client: The baserun client object used to send the message.
    - model_name: The name of the model to use for the completion.
    - prompt: The prompt message to send to the model.
    - response: The response message generated by the model.
    - evaluation: Optional. The evaluation score for the response.

    Returns:
    None
    """
    # Make a "Generic" Completion object and map the prompt and answer to input/output messages
    completion = GenericCompletion(
        client=baserun_client,
        trace_id = baserun_client.trace_id,
        name=model_name,
        # To accurately capture duration we need to set the start timestamp
        start_timestamp=baserun_client.start_timestamp,
        input_messages=[GenericInputMessage(role="system", content=prompt)],
        choices=[
            GenericChoice(
                message=GenericCompletionMessage(role="assistant", content=response)
            )
        ],
    )
    completion.submit_to_baserun()
    #Traces

    print("Prompt:", prompt)
    print("Response:", response)
    print("Evaluation:", evaluation)
    if evaluation:
        trace_id = baserun_client.trace_id
        log("Tagging resumed", trace_id=trace_id, )
        feedback("User satisfaction", str(evaluation), trace_id=trace_id)


    
def send_email_via_sendgrid(api_key, from_email, to_email, subject, content):
    """
    Sends an email using the SendGrid API.

    Parameters:
    - api_key (str): The SendGrid API key.
    - from_email (str): The email address of the sender.
    - to_email (str): The email address of the recipient.
    - subject (str): The subject of the email.
    - content (str): The content of the email in markdown format.

    Returns:
    None

    Raises:
    - Exception: If an error occurs while sending the email.
    """
    #strip ``` from content
    content = content.replace("```", "")
    #strip leading word markdown from content
    content = content.replace("markdown", "")
    #convert content from markdown to html
    content = markdown.markdown(content)
    message = Mail(
        from_email=from_email,
        to_emails=to_email,
        subject=subject,
        html_content=content)
    try:
        sg = SendGridAPIClient(api_key)
        response = sg.send(message)
        print(f"Email sent! Status code: {response.status_code}")
    except Exception as e:
        print(f"An error occurred: {e}")